"""POS 退货编排：查单、校验、拆分退款、结算资金、回补库存、发放 Store Credit。

资金安全的两条铁律（贯穿本模块）：
1. 资金未确认成功前，绝不回补库存、绝不发放 Store Credit。
2. 所有对外副作用都以 PosReturn.idempotency_key 为幂等边界，网络重试不重复发钱。
"""
from __future__ import annotations

from datetime import datetime, timedelta
from decimal import Decimal

from sqlalchemy import and_, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.models.customer import Customer
from app.core.models.order import Order, OrderItem
from app.core.models.payment import Payment
from app.core.models.product import Product, ProductVariant
from app.core.models.wallet import CustomerWallet, WalletTransaction
from app.plugins.pos_operations.models import PosReturn
from app.plugins.pos_operations.returns import (
    ReturnInvalid, TenderAvailable, split_refund, validate_return_quantity, within_return_window,
)
from app.plugins.pos_operations.services import get_rules


def _cents(value: Decimal) -> int:
    return int((Decimal(value) * 100).quantize(Decimal("1")))


def _dollars(cents: int) -> Decimal:
    return (Decimal(cents) / Decimal("100")).quantize(Decimal("0.01"))


def sold_quantity(item: OrderItem) -> Decimal:
    """称重商品以快照里的精确重量为准。

    order_items.quantity 是 DECIMAL(12,2)，0.125kg 落库会变成 0.13；下单时的精确重量
    保存在 product_snapshot.weight_kg 里，退货必须用它，否则可退量会凭空多出来。
    """
    snap = item.product_snapshot or {}
    if snap.get("sold_by_weight") and snap.get("weight_kg"):
        try:
            return Decimal(str(snap["weight_kg"]))
        except (ValueError, ArithmeticError):
            pass
    return Decimal(str(item.quantity))


async def returned_quantities(db: AsyncSession, tenant_id: int, order_id: int) -> dict[str, Decimal]:
    """该订单每个明细已退/在退数量，键为 order_item_id 的字符串形式。

    completed 与 pending 都占用额度：pending 表示资金可能已经在终端上发生，如果不占
    额度，两条 Lane 可以同时把同一件商品各退一次。failed 才释放额度。
    """
    rows = (await db.execute(
        select(PosReturn).where(
            PosReturn.tenant_id == tenant_id,
            PosReturn.source_order_id == order_id,
            PosReturn.fund_status.in_(("completed", "pending")),
        )
    )).scalars().all()
    totals: dict[str, Decimal] = {}
    for record in rows:
        if record.stock_disposition == "none":
            continue
        for line in (record.items or []):
            key = str(line.get("orderItemId"))
            totals[key] = totals.get(key, Decimal("0")) + Decimal(str(line.get("quantity", "0")))
    return totals


async def lookup_order_for_return(
    db: AsyncSession, *, tenant_id: int, store_id: int, order_no: str, now: datetime | None = None,
) -> dict:
    """按订单号查可退明细。不抛异常，用 ok/reason 表达业务拒绝，便于 POS 直接展示。

    收银台手里只有本地单号，而 POS 同步上来的订单在云端叫 "POS-" + 本地单号
    （见 pos_sync.services.sync_order，同样截断到 50 字符）。两种写法都要能查到，
    否则小票上印的号、最近订单列表点的号，全都查不到自己那张单。
    """
    now = now or datetime.utcnow()

    async def _find(no: str):
        return (await db.execute(
            select(Order).where(Order.tenant_id == tenant_id, Order.order_no == no)
        )).scalar_one_or_none()

    # 先按原样查：真正的云端单号优先，不会被下面的推导形式抢走。
    order = await _find(order_no)
    if order is None and not order_no.startswith("POS-"):
        # 截断规则必须和 sync_order 写入时一致，超长单号才能精确对上。
        order = await _find(f"POS-{order_no}"[:50])
    if order is None:
        return {"ok": False, "reason": "order_not_found"}
    if order.status not in ("paid", "completed", "shipped", "delivered"):
        return {"ok": False, "reason": f"order_status_{order.status}"}

    rules = await get_rules(db, tenant_id=tenant_id, store_id=store_id)
    paid_at = order.paid_at or order.created_at
    if not within_return_window(paid_at, now, rules.return_window_days):
        return {"ok": False, "reason": "outside_return_window",
                "returnWindowDays": rules.return_window_days}

    items = (await db.execute(
        select(OrderItem).where(OrderItem.order_id == order.id)
    )).scalars().all()
    already = await returned_quantities(db, tenant_id, order.id)

    lines = []
    for item in items:
        sold = sold_quantity(item)
        done = already.get(str(item.id), Decimal("0"))
        remaining = max(sold - done, Decimal("0"))
        snap = item.product_snapshot or {}
        lines.append({
            "orderItemId": item.id,
            "productId": item.product_id,
            "variantId": item.variant_id,
            "name": snap.get("name") or "",
            "unit": snap.get("unit") or "ea",
            "soldByWeight": bool(snap.get("sold_by_weight")),
            "soldQuantity": str(sold),
            "returnedQuantity": str(done),
            "remainingQuantity": str(remaining),
            "unitPriceCents": _cents(item.unit_price),
        })

    return {
        "ok": True,
        "orderId": order.id,
        "orderNo": order.order_no,
        "paidAt": paid_at.isoformat() if paid_at else None,
        "customerId": order.customer_id,
        "currency": order.currency,
        "rulesVersion": rules.rules_version,
        "lines": lines,
        "tenders": await _tender_availability(db, tenant_id, order.id),
    }


async def _tender_availability(db: AsyncSession, tenant_id: int, order_id: int) -> list[dict]:
    """原订单各支付方式的可退余额。"""
    payments = (await db.execute(
        select(Payment).where(
            Payment.tenant_id == tenant_id,
            Payment.order_id == order_id,
            Payment.status == "completed",
        )
    )).scalars().all()

    refunded = (await db.execute(
        select(PosReturn.refund_splits).where(
            PosReturn.tenant_id == tenant_id,
            PosReturn.source_order_id == order_id,
            PosReturn.fund_status.in_(("pending", "completed")),
        )
    )).scalars().all()
    done: dict[int, int] = {}
    for splits in refunded:
        for s in (splits or []):
            pid = s.get("paymentId")
            if pid is not None:
                done[pid] = done.get(pid, 0) + int(s.get("amountCents", 0))

    out = []
    for p in payments:
        method = (p.extra_data or {}).get("payment_method") or p.gateway
        paid = _cents(p.amount)
        out.append({
            "paymentId": p.id,
            "paymentMethod": "eftpos" if method == "eftpos" else "cash" if method == "cash" else method,
            "paidCents": paid,
            "refundedCents": done.get(p.id, 0),
            "availableCents": max(paid - done.get(p.id, 0), 0),
            # 原始终端交易号：Windcave 退款必须引用它，否则实机大概率拒绝。
            "originalTxnRef": p.gateway_ref,
        })
    return out


async def plan_receipt_return(
    db: AsyncSession, *, tenant_id: int, store_id: int, order_no: str,
    requested: list[dict], refund_method: str = "original",
    refund_amount_cents: int | None = None, money_only: bool = False,
    now: datetime | None = None,
) -> dict:
    """校验退货请求并算出退款拆分，但不落库。POS 用它给审批人展示确切金额。"""
    view = await lookup_order_for_return(
        db, tenant_id=tenant_id, store_id=store_id, order_no=order_no, now=now)
    if not view["ok"]:
        return view

    by_id = {line["orderItemId"]: line for line in view["lines"]}

    # 同一 orderItemId 出现两次会各自对着同一份剩余额度校验，合起来就能超退。
    ids = [r.get("orderItemId") for r in requested]
    if len(set(ids)) != len(ids):
        raise ReturnInvalid("同一明细不能重复提交")

    total = 0
    resolved = []
    for req in requested:
        line = by_id.get(req.get("orderItemId"))
        if line is None:
            raise ReturnInvalid("退货明细不属于该订单")
        qty = validate_return_quantity(
            Decimal(str(req.get("quantity", "0"))),
            Decimal(line["soldQuantity"]),
            Decimal(line["returnedQuantity"]),
            sold_by_weight=line["soldByWeight"],
        )
        amount = int((Decimal(line["unitPriceCents"]) * qty).quantize(Decimal("1")))
        total += amount
        resolved.append({
            "orderItemId": line["orderItemId"],
            "productId": line["productId"],
            "variantId": line["variantId"],
            "name": line["name"],
            "quantity": str(qty),
            "unitPriceCents": line["unitPriceCents"],
            "lineTotalCents": amount,
        })

    if not resolved:
        raise ReturnInvalid("请至少选择一件退货商品")

    selected_total = total
    if refund_amount_cents is not None:
        if refund_amount_cents > selected_total:
            raise ReturnInvalid("部分退款金额超过所选商品可退金额")
        total = refund_amount_cents
    if money_only:
        for line in resolved:
            line["inventoryQuantity"] = "0"

    tenders = [
        TenderAvailable(
            paymentMethod=t["paymentMethod"], paidCents=t["paidCents"],
            refundedCents=t["refundedCents"], paymentId=t["paymentId"],
            originalTxnRef=t.get("originalTxnRef"),
        )
        for t in view["tenders"]
    ]
    splits = ([{"paymentMethod": "cash", "amountCents": total,
                "paymentId": None, "originalTxnRef": None}]
              if refund_method == "cash" else split_refund(tenders, total))

    return {
        "ok": True,
        "orderId": view["orderId"],
        "orderNo": view["orderNo"],
        "customerId": view["customerId"],
        "rulesVersion": view["rulesVersion"],
        "items": resolved,
        "refundTotalCents": total,
        "moneyOnly": money_only,
        "splits": [
            s if isinstance(s, dict) else {
                "paymentMethod": s.paymentMethod, "amountCents": s.amountCents,
                "paymentId": s.paymentId, "originalTxnRef": s.originalTxnRef,
            }
            for s in splits
        ],
    }


async def find_return_by_key(
    db: AsyncSession, *, tenant_id: int, idempotency_key: str,
) -> PosReturn | None:
    """按幂等键取既有退货记录。

    重试必须在重新 plan 之前先走这里：第一次落库的 pending 记录本身就占用可退额度
    （见 returned_quantities），重算 plan 会撞上自己，剩余额度永远是 0。
    """
    return (await db.execute(
        select(PosReturn).where(
            PosReturn.tenant_id == tenant_id,
            PosReturn.idempotency_key == idempotency_key,
        )
    )).scalar_one_or_none()


async def create_return(
    db: AsyncSession, *, tenant_id: int, store_id: int, lane_id: str | None,
    idempotency_key: str, plan: dict, operator_user_id: int, approver_user_id: int,
    reason: str | None, stock_disposition: str,
    source_type: str = "receipt", customer_id: int | None = None,
    reference_price_cents: int | None = None, override_price_cents: int | None = None,
    override_reason: str | None = None, refund_method: str = "original",
) -> PosReturn:
    """落一条 pending 的退货记录。资金尚未发生，库存也还没动。"""
    existing = (await db.execute(
        select(PosReturn).where(
            PosReturn.tenant_id == tenant_id,
            PosReturn.idempotency_key == idempotency_key,
        )
    )).scalar_one_or_none()
    if existing:
        return existing  # 重试：返回既有记录，绝不重复退款

    # 锁住原订单，让并发的两条 Lane 串行进入下面的额度校验。plan 是在锁外算的，
    # 期间别人可能已经退掉了同一件商品，所以这里必须重算一次而不是相信入参。
    if plan.get("orderId"):
        await db.execute(
            select(Order.id).where(Order.id == plan["orderId"]).with_for_update()
        )
        validate_refund_split_availability(
            plan, await _tender_availability(db, tenant_id, plan["orderId"]),
        )
        already = await returned_quantities(db, tenant_id, plan["orderId"])
        items = (await db.execute(
            select(OrderItem).where(OrderItem.order_id == plan["orderId"])
        )).scalars().all()
        sold = {str(i.id): sold_quantity(i) for i in items}
        for line in plan["items"]:
            key = str(line.get("orderItemId"))
            remaining = sold.get(key, Decimal("0")) - already.get(key, Decimal("0"))
            if Decimal(str(line["quantity"])) > remaining:
                raise ReturnInvalid(f"退货数量超出剩余可退数量（剩余 {max(remaining, Decimal('0'))}）")

    effective_customer_id = customer_id or plan.get("customerId")
    refund_request_id = await _create_core_refund_request(
        db, tenant_id=tenant_id, plan=plan, customer_id=effective_customer_id,
        reason=reason, operator_user_id=operator_user_id,
    )

    record = PosReturn(
        tenant_id=tenant_id, store_id=store_id, lane_id=lane_id,
        idempotency_key=idempotency_key, source_type=source_type,
        source_order_id=plan.get("orderId"), refund_request_id=refund_request_id,
        customer_id=effective_customer_id,
        items=plan["items"], refund_total_cents=plan["refundTotalCents"],
        refund_splits=plan.get("splits"), refund_method=refund_method,
        reference_price_cents=reference_price_cents, override_price_cents=override_price_cents,
        override_reason=override_reason, reason=reason,
        stock_disposition=stock_disposition, stock_applied=0,
        operator_user_id=operator_user_id, approver_user_id=approver_user_id,
        fund_status="pending", rules_version=plan.get("rulesVersion", 1),
    )
    db.add(record)
    await db.flush()
    await _audit(db, record, "pos.return.created")
    await db.commit()
    await db.refresh(record)
    return record


async def _create_core_refund_request(
    db, *, tenant_id: int, plan: dict, customer_id: int | None,
    reason: str | None, operator_user_id: int,
) -> int | None:
    """Expose member receipt returns in the existing refund centre.

    Receiptless and guest sales cannot use the legacy core table because it requires
    both an order and a customer; PosReturn remains their complete audit record.
    """
    if not plan.get("orderId") or not customer_id:
        return None
    from app.core.models.refund import RefundRequest

    request = RefundRequest(
        tenant_id=tenant_id, order_id=plan["orderId"],
        order_no=plan.get("orderNo") or str(plan["orderId"]),
        customer_id=customer_id, type="return", reason=(reason or "POS return").strip(),
        status="processing", refund_amount=Decimal(plan["refundTotalCents"]) / Decimal(100),
        return_items=[{
            "product_id": item.get("productId"), "variant_id": item.get("variantId"),
            "qty": item.get("quantity"),
        } for item in plan.get("items") or []],
        admin_id=operator_user_id,
    )
    db.add(request)
    await db.flush()
    return request.id


def validate_refund_split_availability(plan: dict, current: list[dict]) -> None:
    available = {row.get("paymentId"): int(row.get("availableCents", 0)) for row in current}
    for split in plan.get("splits") or []:
        payment_id = split.get("paymentId")
        if payment_id is None:
            continue
        if int(split.get("amountCents", 0)) > available.get(payment_id, 0):
            raise ReturnInvalid("退款金额超过原支付方式的剩余可退余额")


async def settle_return(
    db: AsyncSession, *, tenant_id: int, return_id: int, splits_result: list[dict],
    store_id: int | None = None, lane_id: str | None = None,
) -> dict:
    """登记资金结果并在全部成功时完成退货。

    任何一笔未 approved（含结果未知）都保持 pending：不回补库存、不发 Store Credit、
    不标记完成，也不自动重试——由人工在 Admin 处理。
    """
    # 锁住退货记录：两个并发 settle 都会读到 pending，各自回补一次库存。
    conditions = [PosReturn.tenant_id == tenant_id, PosReturn.id == return_id]
    if store_id is not None:
        conditions.append(PosReturn.store_id == store_id)
    if lane_id is not None:
        conditions.append(PosReturn.lane_id == lane_id)
    record = (await db.execute(
        select(PosReturn).where(*conditions).with_for_update()
    )).scalar_one_or_none()
    if record is None:
        raise ReturnInvalid("退货记录不存在")
    if record.fund_status == "completed":
        return {"ok": True, "fundStatus": "completed", "returnId": record.id}  # 幂等

    # 结果必须与创建时的退款计划逐笔对上。只校验「总额相等」的话，
    # 本应退 EFTPOS 的订单可以提交一笔伪造的 CASH approved 蒙混过关。
    planned = _split_key_set(record.refund_splits or [])
    reported = _split_key_set(splits_result)
    if planned and planned != reported:
        record.fund_status = "pending"
        await db.commit()
        return {"ok": False, "fundStatus": "pending", "returnId": record.id,
                "reason": "splits_do_not_match_plan"}

    record.refund_splits = splits_result

    # 资金确认必须是「肯定的证据」，不能靠「没有反证」。空列表 any([]) 为 False，
    # 若只判断 any(...) 就会把「什么都没退」当成全部成功。
    settled = sum(int(s.get("amountCents", 0)) for s in splits_result
                  if s.get("status") == "approved")
    all_approved = bool(splits_result) and all(
        s.get("status") == "approved"
        and (s.get("paymentMethod") != "eftpos" or bool(str(s.get("providerTxnRef") or "").strip()))
        for s in splits_result)

    if not all_approved or settled != record.refund_total_cents:
        record.fund_status = "pending"
        await db.commit()
        return {"ok": False, "fundStatus": "pending", "returnId": record.id,
                "reason": "funds_not_confirmed"}

    # 完成、回补库存、发放 Store Credit 必须在同一事务里：先提交「已完成」再发钱，
    # 中途失败就会留下「已完成但顾客没拿到钱」的记录。
    record.fund_status = "completed"
    await _apply_stock(db, record)
    credit = None
    if record.refund_method == "store_credit":
        credit = await issue_store_credit(db, tenant_id=tenant_id, record=record)
    await _complete_core_refund_request(db, tenant_id=tenant_id, record=record)
    await _audit(db, record, "pos.return.completed")
    await db.commit()
    out = {"ok": True, "fundStatus": "completed", "returnId": record.id}
    if credit is not None:
        out["storeCredit"] = credit
    return out


async def _complete_core_refund_request(db, *, tenant_id: int, record) -> None:
    refund_request_id = getattr(record, "refund_request_id", None)
    if not refund_request_id:
        return
    from datetime import datetime
    from app.core.models.refund import RefundRequest

    request = (await db.execute(select(RefundRequest).where(
        RefundRequest.tenant_id == tenant_id, RefundRequest.id == refund_request_id,
    ).with_for_update())).scalar_one_or_none()
    if request is not None:
        request.status = "completed"
        request.processed_at = request.processed_at or datetime.utcnow()
        request.completed_at = datetime.utcnow()


def _split_key_set(splits: list[dict]) -> set:
    """退款拆分的可比较指纹：支付方式 + 原支付 ID + 金额。"""
    return {
        (s.get("paymentMethod"), s.get("paymentId"), int(s.get("amountCents", 0)))
        for s in splits
    }


async def _audit(db: AsyncSession, record: PosReturn, action: str) -> None:
    """写入现有 audit_logs，Admin 审计页按 action 前缀 `pos.` 即可筛出 POS 记录。"""
    from app.services.audit import log_audit
    await log_audit(
        db, tenant_id=record.tenant_id, action=action,
        actor_type="pos", actor_id=record.operator_user_id,
        target_type="pos_return", target_id=record.id,
        target_name=record.idempotency_key,
        changes={
            "sourceType": record.source_type,
            "refundTotalCents": record.refund_total_cents,
            "refundMethod": record.refund_method,
            "approverUserId": record.approver_user_id,
            "stockDisposition": record.stock_disposition,
            "referencePriceCents": record.reference_price_cents,
            "overridePriceCents": record.override_price_cents,
            "overrideReason": record.override_reason,
        },
    )


async def _apply_stock(db: AsyncSession, record: PosReturn) -> None:
    """仅 resellable 回补可售库存；damaged/inspection 不回补。幂等。

    统一走 core 库存服务：进销存已接管的租户自动落账本流水，未接管的走旧的
    stock_qty 直接回补。仍按租户 + 商品双重限定，防止构造别租户 ID 改到对方库存。
    """
    from app.core.services.inventory import restore_stock

    if record.stock_applied:
        return
    if record.stock_disposition == "none":
        record.stock_applied = 1
        return
    restore_items: list[dict] = []
    for line in (record.items or []):
        qty = Decimal(str(line.get("inventoryQuantity", line.get("quantity", "0"))))
        if qty <= 0:
            continue
        if line.get("variantId"):
            variant = (await db.execute(
                select(ProductVariant)
                .join(Product, Product.id == ProductVariant.product_id)
                .where(
                    ProductVariant.id == line["variantId"],
                    ProductVariant.product_id == line.get("productId"),
                    Product.tenant_id == record.tenant_id,
                )
            )).scalar_one_or_none()
            if variant is not None:
                restore_items.append({"product_id": variant.product_id,
                                      "variant_id": variant.id, "qty": qty})
        elif line.get("productId"):
            product = (await db.execute(
                select(Product).where(
                    Product.id == line["productId"],
                    Product.tenant_id == record.tenant_id,
                )
            )).scalar_one_or_none()
            if product is not None:
                restore_items.append({"product_id": product.id,
                                      "variant_id": None, "qty": qty})
    if restore_items:
        source_order_id = getattr(record, "source_order_id", None)
        inventory_order_id = source_order_id or record.id
        if source_order_id and record.stock_disposition != "resellable":
            from app.plugins.inventory.services import get_state
            if await get_state(db, record.tenant_id) is not None:
                from app.plugins.inventory.models import InventoryAllocation
                has_allocation = (await db.execute(
                    select(InventoryAllocation.id).where(
                        InventoryAllocation.tenant_id == record.tenant_id,
                        InventoryAllocation.order_id == source_order_id,
                    ).limit(1)
                )).scalar()
                if has_allocation:
                    from app.plugins.inventory.returns import restore_allocated_by_sku
                    disposition = "inspect" if record.stock_disposition == "inspection" else "scrap"
                    await restore_allocated_by_sku(
                        db, record.tenant_id, source_order_id, restore_items,
                        suffix=f"pr{record.id}", disposition=disposition,
                        operator_id=record.operator_user_id,
                    )
            # Legacy inventory intentionally keeps non-resellable returns out of sellable stock.
        elif record.stock_disposition == "resellable":
            await restore_stock(db, record.tenant_id, restore_items, order_id=inventory_order_id,
                                restock=True, idem_suffix=f"pr{record.id}")
    record.stock_applied = 1


async def issue_store_credit(db: AsyncSession, *, tenant_id: int, record: PosReturn) -> dict:
    """把退款金额发放为 Store Credit。以 PosReturn.id 为幂等键，重试不重复入账。"""
    if record.customer_id is None:
        raise ReturnInvalid("无小票退货必须绑定顾客")

    dup = (await db.execute(
        select(WalletTransaction).where(
            WalletTransaction.tenant_id == tenant_id,
            WalletTransaction.source_type == "pos_return",
            WalletTransaction.source_id == record.id,
        )
    )).scalar_one_or_none()
    if dup:
        return {"ok": True, "walletTransactionId": dup.id, "duplicate": True}

    wallet = (await db.execute(
        select(CustomerWallet).where(
            CustomerWallet.tenant_id == tenant_id,
            CustomerWallet.customer_id == record.customer_id,
        ).with_for_update()
    )).scalar_one_or_none()
    if wallet is None:
        wallet = CustomerWallet(
            tenant_id=tenant_id, customer_id=record.customer_id, balance=Decimal("0.00"))
        db.add(wallet)
        await db.flush()

    amount = _dollars(record.refund_total_cents)
    wallet.balance = (wallet.balance or Decimal("0.00")) + amount
    txn = WalletTransaction(
        tenant_id=tenant_id, customer_id=record.customer_id, amount=amount,
        balance_after=wallet.balance, type="order_refund",
        source_type="pos_return", source_id=record.id,
        note=f"POS return {record.idempotency_key}",
    )
    db.add(txn)
    await db.flush()
    return {"ok": True, "walletTransactionId": txn.id, "duplicate": False}


async def reference_price_cents(
    db: AsyncSession, *, tenant_id: int, store_id: int, product_id: int,
    variant_id: int | None, window_days: int, now: datetime | None = None,
) -> dict:
    """无小票退货的参考价 = min(当前售价, 退货期限内最低实际成交价)。

    没有历史成交价时用当前售价。实际成交价取 order_items.unit_price（已含会员价与
    行内折扣），因此不需要再单独摊分订单级折扣。
    """
    now = now or datetime.utcnow()
    product = (await db.execute(
        select(Product).where(Product.tenant_id == tenant_id, Product.id == product_id)
    )).scalar_one_or_none()
    if product is None:
        raise ReturnInvalid("商品不存在")

    current = product.base_price or Decimal("0")
    if variant_id:
        # 同样要按租户 + 商品限定：否则别的租户的 variant 价格修正会影响本店的
        # Store Credit 金额。
        variant = (await db.execute(
            select(ProductVariant)
            .join(Product, Product.id == ProductVariant.product_id)
            .where(
                ProductVariant.id == variant_id,
                ProductVariant.product_id == product_id,
                Product.tenant_id == tenant_id,
            )
        )).scalar_one_or_none()
        if variant is not None:
            current = current + (variant.price_modifier or Decimal("0"))
    current_cents = _cents(current)

    conditions = [
        OrderItem.tenant_id == tenant_id,
        OrderItem.product_id == product_id,
        Order.status.in_(("paid", "completed", "shipped", "delivered")),
    ]
    if variant_id:
        conditions.append(OrderItem.variant_id == variant_id)
    if window_days > 0:
        cutoff = now - timedelta(days=window_days)
        conditions.append(Order.created_at >= cutoff)

    lowest = (await db.execute(
        select(func.min(OrderItem.unit_price))
        .join(Order, Order.id == OrderItem.order_id)
        .where(and_(*conditions))
    )).scalar()

    lowest_cents = _cents(lowest) if lowest is not None else None
    reference = min(current_cents, lowest_cents) if lowest_cents is not None else current_cents
    return {
        "currentPriceCents": current_cents,
        "lowestSoldPriceCents": lowest_cents,
        "referencePriceCents": reference,
    }


def _product_age(product: Product) -> int | None:
    """读取商品的 POS 年龄限制；非法值或 <=0 一律视为不限制。"""
    try:
        raw = (product.extra_attributes or {}).get("pos_minimum_age")
        age = int(raw)
    except (TypeError, ValueError, AttributeError):
        return None
    return age if age > 0 else None


async def list_product_ages(
    db: AsyncSession, *, tenant_id: int, q: str = "", only_restricted: bool = False, limit: int = 50,
) -> list[dict]:
    stmt = select(Product).where(Product.tenant_id == tenant_id)
    if q:
        like = f"%{q}%"
        stmt = stmt.where((Product.name.like(like)) | (Product.sku.like(like)))
    rows = (await db.execute(stmt.order_by(Product.id.desc()).limit(limit * 4 if only_restricted else limit))).scalars().all()

    out = []
    for product in rows:
        age = _product_age(product)
        if only_restricted and age is None:
            continue
        out.append({"id": product.id, "sku": product.sku, "name": product.name, "minimumAge": age})
        if len(out) >= limit:
            break
    return out


async def set_product_age(
    db: AsyncSession, *, tenant_id: int, product_id: int, minimum_age: int | None,
) -> dict:
    product = (await db.execute(
        select(Product).where(Product.tenant_id == tenant_id, Product.id == product_id)
    )).scalar_one_or_none()
    if product is None:
        raise ReturnInvalid("商品不存在")

    attrs = dict(product.extra_attributes or {})
    if minimum_age and minimum_age > 0:
        attrs["pos_minimum_age"] = int(minimum_age)
    else:
        attrs.pop("pos_minimum_age", None)
    # 整体替换：JSON 列的原地 mutation 不会被 SQLAlchemy 侦测到。
    product.extra_attributes = attrs
    await db.commit()
    return {"ok": True, "id": product.id, "minimumAge": _product_age(product)}


async def get_return(db: AsyncSession, tenant_id: int, return_id: int) -> PosReturn | None:
    return (await db.execute(
        select(PosReturn).where(PosReturn.tenant_id == tenant_id, PosReturn.id == return_id)
    )).scalar_one_or_none()


async def list_returns(
    db: AsyncSession, *, tenant_id: int, store_id: int | None = None,
    fund_status: str | None = None, limit: int = 50,
) -> list[dict]:
    stmt = select(PosReturn).where(PosReturn.tenant_id == tenant_id)
    if store_id:
        stmt = stmt.where(PosReturn.store_id == store_id)
    if fund_status:
        stmt = stmt.where(PosReturn.fund_status == fund_status)
    rows = (await db.execute(stmt.order_by(PosReturn.id.desc()).limit(limit))).scalars().all()
    return [
        {
            "id": r.id, "storeId": r.store_id, "laneId": r.lane_id,
            "sourceType": r.source_type, "sourceOrderId": r.source_order_id,
            "customerId": r.customer_id, "items": r.items,
            "refundTotalCents": r.refund_total_cents, "refundSplits": r.refund_splits,
            "refundMethod": r.refund_method, "referencePriceCents": r.reference_price_cents,
            "overridePriceCents": r.override_price_cents, "overrideReason": r.override_reason,
            "reason": r.reason, "stockDisposition": r.stock_disposition,
            "stockApplied": bool(r.stock_applied), "operatorUserId": r.operator_user_id,
            "approverUserId": r.approver_user_id, "fundStatus": r.fund_status,
            "rulesVersion": r.rules_version,
            "createdAt": r.created_at.isoformat() if r.created_at else None,
        }
        for r in rows
    ]


async def resolve_customer(db: AsyncSession, tenant_id: int, customer_id: int) -> Customer | None:
    return (await db.execute(
        select(Customer).where(
            Customer.id == customer_id, Customer.tenant_id == tenant_id, Customer.is_active == 1)
    )).scalar_one_or_none()