from __future__ import annotations

import hashlib
import hmac
import json
import secrets
from datetime import datetime, timedelta, timezone
from decimal import ROUND_HALF_UP, Decimal, InvalidOperation

from fastapi import HTTPException
from sqlalchemy import and_, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import get_settings
from app.core.models.customer import Customer
from app.core.models.discount import Discount
from app.core.models.member import MemberLevel
from app.core.models.order import Order, OrderItem
from app.core.models.payment import Payment
from app.core.models.product import Product, ProductPriceRule, ProductVariant
from app.core.models.tenant import Tenant
from app.core.models.wallet import CustomerWallet, WalletTransaction
from app.core.services.inventory import deduct_stock
from app.core.services.pricing import PricingCartItem, _resolve_rule_for_line, calculate_pricing
from app.core.signals import order_created, order_paid
from app.plugins.pos_sync.models import PosLane, PosOrderSync
from app.plugins.pos_sync.schemas import (
    PosMemberCreateIn, PosQuoteIn, PosQuoteLineOut, PosQuoteOut, PosSyncOrderIn, PosSyncOrderOut,
    PosSyncPaymentIn, PosWalletChargeIn, PosWalletBalanceOut,
)

# A counter sale has no fulfilment/shipping step: it is complete once the POS
# records its approved payment.  Keep this separate from ordinary web orders.
POS_SYNC_ORDER_STATUS = "completed"


_B32_ALPHABET = "ABCDEFGHIJKLMNOPQRSTUVWXYZ234567"  # RFC 4648 Base32, no padding


def generate_pairing_code() -> str:
    # 26 chars * 5 bits = 130 bits of entropy, human-enterable.
    return "".join(secrets.choice(_B32_ALPHABET) for _ in range(26))


def generate_sync_token() -> str:
    return secrets.token_urlsafe(32)


def hash_sync_token(token: str) -> str:
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def verify_sync_token(token: str, token_hash: str) -> bool:
    if not token or not token_hash:
        return False
    return hmac.compare_digest(hash_sync_token(token), token_hash)


PAIRING_TTL_MINUTES = 10
AGE_APPROVAL_TTL_MS = 60_000
PRICE_OVERRIDE_APPROVAL_TTL_MS = 180_000


async def list_lanes(db: AsyncSession, tenant_id: int | None = None) -> list[dict]:
    stmt = select(PosLane, Tenant.name).join(Tenant, Tenant.id == PosLane.tenant_id).order_by(PosLane.id.asc())
    if tenant_id:
        stmt = stmt.where(PosLane.tenant_id == tenant_id)
    rows = (await db.execute(stmt)).all()
    now = datetime.utcnow()
    return [
        {
            "id": lane.id,
            "tenant_id": lane.tenant_id,
            "tenant_name": tname,
            "store_id": lane.store_id,
            "store_code": lane.store_code,
            "store_timezone": lane.store_timezone,
            "lane_id": lane.lane_id,
            "lane_name": lane.lane_name,
            "is_active": bool(lane.is_active),
            "last_seen_at": lane.last_seen_at.isoformat() if lane.last_seen_at else None,
            "pairing_available": bool(
                lane.pairing_code_hash
                and lane.pairing_consumed_at is None
                and lane.pairing_expires_at is not None
                and lane.pairing_expires_at > now
            ),
        }
        for lane, tname in rows
    ]


def _issue_code(lane: PosLane) -> str:
    raw = generate_pairing_code()
    lane.pairing_code_hash = hash_sync_token(raw)
    lane.pairing_expires_at = datetime.utcnow() + timedelta(minutes=PAIRING_TTL_MINUTES)
    lane.pairing_consumed_at = None
    return raw


async def create_lane(
    db: AsyncSession, *, tenant_id: int, store_id: int, store_code: str,
    store_timezone: str, lane_id: str, lane_name: str | None,
    store_country: str, store_province: str,
) -> tuple[PosLane, str]:
    dup = (await db.execute(
        select(PosLane).where(PosLane.tenant_id == tenant_id, PosLane.lane_id == lane_id)
    )).scalar_one_or_none()
    if dup:
        raise HTTPException(status_code=409, detail="lane_id already exists for this tenant")
    # No cloud sync token exists until the first pairing exchange. Store the hash of a
    # discarded random token so token_hash is always a real 64-char hash, never a magic
    # empty sentinel; the first exchange rotates it to the token actually sent to the Agent.
    lane = PosLane(
        tenant_id=tenant_id, store_id=store_id, store_code=store_code,
        store_timezone=store_timezone, lane_id=lane_id, lane_name=lane_name,
        store_country=store_country, store_province=store_province,
        token_hash=hash_sync_token(generate_sync_token()), is_active=1,
    )
    raw = _issue_code(lane)
    db.add(lane)
    await db.commit()
    await db.refresh(lane)
    return lane, raw


async def issue_pairing_code(db: AsyncSession, lane_pk: int) -> str:
    lane = (await db.execute(select(PosLane).where(PosLane.id == lane_pk))).scalar_one_or_none()
    if not lane:
        raise HTTPException(status_code=404, detail="lane not found")
    if not lane.store_code or not lane.store_timezone:
        raise HTTPException(status_code=422, detail="lane needs store_code and store_timezone before pairing")
    if not lane.store_country:
        raise HTTPException(status_code=422, detail="lane needs store_country before pairing")
    raw = _issue_code(lane)  # does NOT rotate token_hash
    await db.commit()
    return raw


async def update_lane(
    db: AsyncSession, lane_pk: int, *, lane_name: str | None, is_active: bool | None,
    store_country: str | None = None, store_province: str | None = None,
) -> dict:
    lane = (await db.execute(select(PosLane).where(PosLane.id == lane_pk))).scalar_one_or_none()
    if not lane:
        raise HTTPException(status_code=404, detail="lane not found")
    if lane_name is not None:
        lane.lane_name = lane_name
    if is_active is not None:
        lane.is_active = 1 if is_active else 0
    if store_country is not None:
        lane.store_country = store_country
    if store_province is not None:
        lane.store_province = store_province
    await db.commit()
    return {"id": lane.id, "lane_name": lane.lane_name, "is_active": bool(lane.is_active)}


async def unpair_lane(db: AsyncSession, lane_pk: int) -> dict:
    """Revoke the current Agent without disabling the lane for a future pairing."""
    lane = (await db.execute(select(PosLane).where(PosLane.id == lane_pk))).scalar_one_or_none()
    if not lane:
        raise HTTPException(status_code=404, detail="lane not found")
    lane.token_hash = hash_sync_token(generate_sync_token())
    lane.pairing_code_hash = None
    lane.pairing_expires_at = None
    lane.pairing_consumed_at = None
    await db.commit()
    return {"id": lane.id, "is_active": bool(lane.is_active), "unpaired": True}


async def exchange_pairing_code(db: AsyncSession, raw_code: str) -> dict:
    code_hash = hash_sync_token(raw_code)
    now = datetime.utcnow()
    lane = (await db.execute(
        select(PosLane).where(PosLane.pairing_code_hash == code_hash).with_for_update()
    )).scalar_one_or_none()
    if (
        not lane or not lane.is_active
        or lane.pairing_consumed_at is not None
        or lane.pairing_expires_at is None or lane.pairing_expires_at < now
        or not lane.store_code or not lane.store_timezone or not lane.store_country
    ):
        raise HTTPException(status_code=401, detail="invalid pairing code")
    raw_token = generate_sync_token()
    lane.token_hash = hash_sync_token(raw_token)
    lane.pairing_consumed_at = now
    lane.pairing_code_hash = None
    lane.last_seen_at = now
    await db.commit()
    settings = get_settings()
    return {
        "tenantId": lane.tenant_id,
        "storeId": lane.store_id,
        "storeCode": lane.store_code,
        "storeTimezone": lane.store_timezone,
        "laneId": lane.lane_id,
        "laneName": lane.lane_name or lane.lane_id,
        "cloudBaseUrl": settings.POS_PUBLIC_API_BASE_URL,
        "cloudSyncToken": raw_token,
        "allowedOrigin": settings.POS_AGENT_ALLOWED_ORIGIN,
    }


def payload_hash(payload: PosSyncOrderIn) -> str:
    data = payload.model_dump(mode="json")
    for key, default in (
        ("priceOverrideApplied", False),
        ("priceOverrideEvidence", None),
        ("orderDiscountCents", 0),
    ):
        if data.get(key) == default:
            data.pop(key, None)
    for payment in data.get("payments", []):
        if payment.get("walletTxnId") is None:
            payment.pop("walletTxnId", None)
    for item in data["items"]:
        unit_price = item["unitPriceCents"]
        if item.get("baseUnitPriceCents") == unit_price:
            item.pop("baseUnitPriceCents", None)
        if item.get("listPriceCents") == unit_price:
            item.pop("listPriceCents", None)
        for key, default in (
            ("baseUnitPriceCents", None),
            ("listPriceCents", None),
            ("manualUnitPriceCents", None),
            ("lineDiscountCents", 0),
        ):
            if item.get(key) == default:
                item.pop(key, None)
    raw = json.dumps(data, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()


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


def to_cents(value: Decimal) -> int:
    return int((Decimal(value) * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP))


def parse_created_at(value: str) -> datetime:
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00")).replace(tzinfo=None)
    except ValueError:
        return datetime.utcnow()


async def authenticate_lane(db: AsyncSession, body: PosSyncOrderIn, bearer_token: str) -> PosLane:
    result = await db.execute(
        select(PosLane).where(
            PosLane.tenant_id == body.tenantId,
            PosLane.store_id == body.storeId,
            PosLane.lane_id == body.laneId,
            PosLane.is_active == 1,
        )
    )
    lane = result.scalar_one_or_none()
    if not lane or not verify_sync_token(bearer_token, lane.token_hash):
        raise HTTPException(status_code=401, detail="invalid POS sync token")
    lane.last_seen_at = datetime.utcnow()
    return lane


async def authenticate_lane_by_ids(
    db: AsyncSession, tenant_id: int, store_id: int, lane_id: str, bearer_token: str
) -> PosLane:
    """Like authenticate_lane but for GET requests that carry ids as query params."""
    result = await db.execute(
        select(PosLane).where(
            PosLane.tenant_id == tenant_id,
            PosLane.store_id == store_id,
            PosLane.lane_id == lane_id,
            PosLane.is_active == 1,
        )
    )
    lane = result.scalar_one_or_none()
    if not lane or not verify_sync_token(bearer_token, lane.token_hash):
        raise HTTPException(status_code=401, detail="invalid POS sync token")
    lane.last_seen_at = datetime.utcnow()
    return lane


async def authenticate_lane_by_token(db: AsyncSession, bearer_token: str) -> PosLane:
    """Resolve the lane straight from its sync token (used by tokenless GET proxies)."""
    lane = (await db.execute(
        select(PosLane).where(PosLane.token_hash == hash_sync_token(bearer_token), PosLane.is_active == 1)
    )).scalar_one_or_none()
    if not lane:
        raise HTTPException(status_code=401, detail="invalid POS sync token")
    return lane


async def search_members(db: AsyncSession, tenant_id: int, q: str, limit: int = 20) -> dict:
    like = f"%{q}%"
    rows = (await db.execute(
        select(Customer, MemberLevel.name)
        .outerjoin(MemberLevel, MemberLevel.id == Customer.member_level_id)
        .where(
            Customer.tenant_id == tenant_id,
            Customer.is_active == 1,
            or_(Customer.name.like(like), Customer.phone.like(like), Customer.email.like(like)),
        )
        .order_by(Customer.orders_count.desc())
        .limit(limit)
    )).all()
    return {"members": [
        {"id": c.id, "name": c.name, "phone": c.phone, "tierName": tier, "memberLevelId": c.member_level_id}
        for c, tier in rows
    ]}


def synthetic_member_email(phone: str, tenant_domain: str) -> str:
    """给不出邮箱的顾客生成一个占位邮箱。

    从**手机号数字**派生，不是序号 —— 这样 (tenant_id, email) 那条既有唯一约束
    就顺带成了手机号的唯一约束：另一台收银机再录同一个号，无论写成
    "021 123 4567" 还是 "0211234567"，都归一成同一个邮箱而撞键，不会静默多出
    一个会员。手机号列本身没有唯一约束，不想为此做迁移和历史数据清洗。

    ponytail: 靠派生邮箱兜手机号唯一性。真要按手机号做外键级约束，
    再加 (tenant_id, phone) 唯一索引并清洗历史数据。
    """
    digits = "".join(ch for ch in phone if ch.isdigit())
    return f"pos-{digits}@{tenant_domain}"


async def _member_out(db: AsyncSession, customer: Customer, *, existing: bool) -> dict:
    tier = None
    if customer.member_level_id:
        tier = (await db.execute(
            select(MemberLevel.name).where(MemberLevel.id == customer.member_level_id)
        )).scalar_one_or_none()
    return {
        "id": customer.id, "name": customer.name, "phone": customer.phone,
        "tierName": tier, "existing": existing,
    }


async def create_member(
    db: AsyncSession, tenant_id: int, body: PosMemberCreateIn,
) -> dict:
    """柜台建会员。已存在则返回既有会员，不报错 —— 收银员没有别的路可走。"""
    tenant = (await db.execute(select(Tenant).where(Tenant.id == tenant_id))).scalar_one_or_none()
    if tenant is None:
        raise HTTPException(status_code=404, detail="tenant not found")

    phone = body.phone.strip()
    email = (body.email or "").strip().lower() or synthetic_member_email(phone, tenant.domain)

    # 邮箱是身份键，先查它；再按手机号兜一次（线上/后台建的会员用的是真邮箱，
    # 派生邮箱查不到，但手机号能对上）。任一命中都返回既有会员。
    for criterion in (Customer.email == email, Customer.phone == phone):
        hit = (await db.execute(
            select(Customer).where(Customer.tenant_id == tenant_id, criterion).order_by(Customer.id)
        )).scalars().first()
        if hit is not None:
            return await _member_out(db, hit, existing=True)

    customer = Customer(
        tenant_id=tenant_id,
        name=body.name.strip(),
        email=email,
        phone=phone,
        is_active=1,
        # 留痕：将来接上订单邮件时，要靠这个跳过派生邮箱，否则每笔 POS 会员单
        # 都会往一个不存在的地址发信并退信。没有密码 = 暂时不能在线登录。
        extra_data={"created_via": "pos", "synthetic_email": not body.email},
    )
    db.add(customer)
    await db.commit()
    await db.refresh(customer)
    return await _member_out(db, customer, existing=False)


async def lookup_receipts(
    db: AsyncSession, *, tenant_id: int, store_id: int, q: str, limit: int = 20,
    offset: int = 0, date: str = "",
) -> dict:
    """Cloud receipt lookup scoped to the lane's own store.

    When ``q`` is empty returns recent orders (paginated) for the store so the POS
    return flow can list all recent transactions.  When ``q`` is given the existing
    order-number search applies.
    """
    q = (q or "").strip()
    limit = max(1, min(limit, 50))
    offset = max(0, offset)
    date = date if len(date) == 10 and date[4] == "-" and date[7] == "-" else ""

    # ponytail: store_id lives inside payload_json.  To paginate across all orders we
    # over-fetch and filter in Python.  For stores with <10k orders this is fine;
    # add a store_id column to pos_order_sync if throughput matters.
    if not q:
        rows = (await db.execute(
            select(PosOrderSync)
            .where(PosOrderSync.tenant_id == tenant_id)
            .order_by(PosOrderSync.id.desc())
            .offset(offset)
            .limit(limit * 10)
        )).scalars().all()
        orders = []
        for r in rows:
            p = r.payload_json or {}
            if p.get("storeId") != store_id:
                continue
            if date and not str(p.get("createdAt") or "").startswith(date):
                continue
            orders.append({
                "localOrderNo": r.local_order_no,
                "totalCents": p.get("totalCents"),
                "createdAt": p.get("createdAt"),
                "syncStatus": r.sync_status,
            })
            if len(orders) >= limit:
                break
        return {"orders": orders}

    like = f"%{q}%"
    rows = (await db.execute(
        select(PosOrderSync)
        .where(PosOrderSync.tenant_id == tenant_id, PosOrderSync.local_order_no.like(like))
        .order_by(PosOrderSync.id.desc())
        .limit(limit * 20)
    )).scalars().all()
    orders = []
    for r in rows:
        p = r.payload_json or {}
        if p.get("storeId") != store_id:
            continue
        if date and not str(p.get("createdAt") or "").startswith(date):
            continue
        orders.append({
            "localOrderNo": r.local_order_no,
            "totalCents": p.get("totalCents"),
            "createdAt": p.get("createdAt"),
            "syncStatus": r.sync_status,
            "receiptHtml": p.get("receiptHtml"),
        })
        if len(orders) >= limit:
            break
    return {"orders": orders}


async def get_receipt_detail(
    db: AsyncSession, *, tenant_id: int, store_id: int, local_order_no: str,
) -> dict | None:
    """单张小票明细，供 POS 查看跨机订单（本机 SQLite 只有本终端卖出的单）。

    数据取自 pos_order_sync.payload_json —— 同步上来的原始订单里 items/payments/
    receiptHtml 都在，不必 JOIN orders/order_items/payments。代价是它是**成交快照**，
    不反映后台事后改单；对"查小票"这个用途正是想要的语义。

    ponytail: 只读 payload_json。要反映后台改单请改为 JOIN 真表。

    查不到（或不属于本 lane 的门店）返回 None，由路由层转 404。
    """
    row = (await db.execute(
        select(PosOrderSync).where(
            PosOrderSync.tenant_id == tenant_id,
            PosOrderSync.local_order_no == local_order_no,
        ).order_by(PosOrderSync.id.desc())
    )).scalars().first()
    if row is None:
        return None
    p = row.payload_json or {}
    # 越店防护：和 lookup_receipts 一致，lane 只能读自己门店的单。
    if p.get("storeId") != store_id:
        return None
    return {
        "localOrderNo": row.local_order_no,
        "createdAt": p.get("createdAt"),
        "totalCents": p.get("totalCents"),
        "receiptHtml": p.get("receiptHtml"),
        # 模板版本/打印状态是终端本地的概念，云端不掌握。
        "receiptTemplateVersion": None,
        "printStatus": "",
        "reprintCount": 0,
        "items": [
            {
                "name": i.get("name") or "",
                "quantity": str(i.get("quantity") or "0"),
                "unitPriceCents": i.get("unitPriceCents") or 0,
                "lineTotalCents": i.get("lineTotalCents") or 0,
            }
            for i in (p.get("items") or [])
        ],
        "payments": [
            {
                "paymentMethod": pay.get("paymentMethod") or "",
                "amountCents": pay.get("amountCents") or 0,
            }
            for pay in (p.get("payments") or [])
        ],
    }


async def resolve_member(db: AsyncSession, tenant_id: int, member_id: int | None) -> Customer | None:
    """按租户校验会员归属；查不到/停用则返回 None（调用方回落 walk-in）。"""
    if not member_id:
        return None
    result = await db.execute(select(Customer).where(
        Customer.id == member_id,
        Customer.tenant_id == tenant_id,
        Customer.is_active == 1,
    ))
    return result.scalar_one_or_none()


async def get_member_balance(db: AsyncSession, tenant_id: int, customer_id: int) -> dict:
    """查询会员钱包余额。无 wallet 记录 = 0。"""
    if not await resolve_member(db, tenant_id, customer_id):
        return PosWalletBalanceOut(ok=False, customerId=customer_id, reason="member_not_found").model_dump()
    w = (await db.execute(
        select(CustomerWallet).where(
            CustomerWallet.customer_id == customer_id,
            CustomerWallet.tenant_id == tenant_id,
        )
    )).scalar_one_or_none()
    if not w:
        return PosWalletBalanceOut(ok=True, customerId=customer_id, balanceCents=0).model_dump()
    return PosWalletBalanceOut(
        ok=True, customerId=customer_id,
        balanceCents=int(Decimal(w.balance * 100).to_integral_value()),
        currency=w.currency,
    ).model_dump()


# ponytail: 幂等键走 WalletTransaction.note 字段（"idempotency:{key}"），
# 唯一性靠同事务内的 SELECT FOR UPDATE；不动 schema 也省一次迁移。
def _idempotency_note(key: str) -> str:
    return f"idempotency:{key}"


async def charge_member_balance(
    db: AsyncSession, tenant_id: int, body: PosWalletChargeIn,
) -> dict:
    """扣会员余额。幂等键：同 key 重复请求返回上次结果（不重复扣款）。"""
    if not await resolve_member(db, tenant_id, body.customerId):
        return {"ok": False, "customerId": body.customerId, "reason": "member_not_found"}

    # 必须先拿钱包行锁再查幂等：同会员的并发扣款在同一把锁上串行化，
    # 第二个请求等锁期间第一个已 commit，幂等查询因此能看到重复 key。
    # 幂等查询若在锁前执行，两个并发请求都查到"无记录"，双双扣款。
    w = (await db.execute(
        select(CustomerWallet).where(
            CustomerWallet.customer_id == body.customerId,
            CustomerWallet.tenant_id == tenant_id,
        ).with_for_update()
    )).scalar_one_or_none()
    if w is None:
        return {"ok": False, "customerId": body.customerId, "reason": "wallet_not_found"}

    note = _idempotency_note(body.idempotencyKey)
    existing = (await db.execute(
        select(WalletTransaction).where(
            WalletTransaction.customer_id == body.customerId,
            WalletTransaction.tenant_id == tenant_id,
            WalletTransaction.type == "order_pay",
            WalletTransaction.note == note,
        )
    )).scalar_one_or_none()
    if existing is not None:
        # 之前已扣过：返回当时的扣款金额与余额快照，duplicate=True 让前端知道是幂等回放。
        # 同 key 换金额是调用方 bug，不能静默返回旧值，直接报错。
        charged_cents = int((-(existing.amount)) * 100)
        if charged_cents != body.amountCents:
            return {
                "ok": False, "customerId": body.customerId,
                "reason": "idempotency_amount_mismatch",
            }
        return {
            "ok": True,
            "customerId": body.customerId,
            "amountCents": charged_cents,
            "balanceAfterCents": int(Decimal(existing.balance_after * 100).to_integral_value()),
            "walletTxnId": existing.id,
            "duplicate": True,
        }

    amount = Decimal(body.amountCents) / Decimal(100)
    if Decimal(w.balance) < amount:
        return {"ok": False, "customerId": body.customerId, "reason": "insufficient_balance"}

    w.balance = Decimal(w.balance) - amount
    txn = WalletTransaction(
        tenant_id=tenant_id,
        customer_id=body.customerId,
        amount=-amount,
        balance_after=w.balance,
        type="order_pay",
        source_type="pos_sale",
        source_id=None,
        note=note,
    )
    db.add(txn)
    await db.commit()
    await db.refresh(txn)
    return {
        "ok": True,
        "customerId": body.customerId,
        "amountCents": body.amountCents,
        "balanceAfterCents": int((w.balance * 100).to_integral_value()),
        "walletTxnId": txn.id,
        "duplicate": False,
    }


async def _validate_balance_payments(
    db: AsyncSession, body: PosSyncOrderIn, customer_id: int,
) -> tuple[str | None, list[WalletTransaction]]:
    """校验并占用余额支付的钱包流水。

    校验：流水必须真实存在（来自 /pos/wallet/charge）、归属本单客户、金额一致、
    币种一致、且尚未被其它订单占用（source_id IS NULL）。校验即 FOR UPDATE 锁行，
    与并发订单形成串行化：先到者写 source_id 后，后到者的查询条件自然落空。
    返回 (失败原因, 待占用流水)。
    """
    txns: list[WalletTransaction] = []
    seen: set[int] = set()
    for payment in body.payments:
        if payment.paymentMethod != "balance":
            continue
        # 同一笔流水在一张单里出现两次：source_id 要到落库后才回写，两次查询都还是
        # NULL、金额也各自相等，不去重就能用一笔 $5 扣款顶掉 $10 的单。
        if payment.walletTxnId in seen:
            return "wallet_txn_duplicate", []
        seen.add(payment.walletTxnId)
        txn = (await db.execute(
            select(WalletTransaction).where(
                WalletTransaction.id == payment.walletTxnId,
                WalletTransaction.tenant_id == body.tenantId,
                WalletTransaction.customer_id == customer_id,
                WalletTransaction.type == "order_pay",
                WalletTransaction.source_id.is_(None),
            ).with_for_update()
        )).scalar_one_or_none()
        if txn is None:
            return "wallet_txn_invalid", []
        if int((-(txn.amount)) * 100) != payment.amountCents:
            return "wallet_txn_amount_mismatch", []
        wallet = (await db.execute(
            select(CustomerWallet).where(
                CustomerWallet.customer_id == customer_id,
                CustomerWallet.tenant_id == body.tenantId,
            )
        )).scalar_one_or_none()
        if wallet is None or wallet.currency != body.currency:
            return "wallet_currency_mismatch", []
        txns.append(txn)
    return None, txns


async def quote_cart(
    db: AsyncSession, tenant_id: int, body: PosQuoteIn, store_id: int = 0,
) -> PosQuoteOut:
    """会员价报价：仅联网可用，必须在收款前调用。云端为定价唯一权威。"""
    member = await resolve_member(db, tenant_id, body.memberId)
    if member is None:
        return PosQuoteOut(ok=False, memberId=body.memberId, reason="member_not_found")

    keys = [(i.cloudProductId, i.cloudVariantId) for i in body.items]
    if len(set(keys)) != len(keys):
        return PosQuoteOut(ok=False, memberId=body.memberId, reason="duplicate_line")

    try:
        items = [PricingCartItem(
            product_id=i.cloudProductId,
            variant_id=i.cloudVariantId,
            qty=Decimal(str(i.quantity)),  # 原始精度，不量化：称重商品依赖三位小数
        ) for i in body.items]
        # POS 渠道报价：channel='pos' 让 ProductPriceRule 命中走 POS 规则；
        # 没规则时回落 tier/member fallback。
        result = await calculate_pricing(db, member, tenant_id, items, is_pickup=True, channel="pos")
    except HTTPException as exc:
        return PosQuoteOut(ok=False, memberId=body.memberId, reason=str(exc.detail))

    priced = {(line.product_id, line.variant_id): line for line in result.lines}
    lines: list[PosQuoteLineOut] = []
    for item in body.items:
        line = priced.get((item.cloudProductId, item.cloudVariantId))
        if line is None:
            return PosQuoteOut(ok=False, memberId=body.memberId, reason="line_not_priced")
        lines.append(PosQuoteLineOut(
            lineNo=item.lineNo,
            cloudProductId=item.cloudProductId,
            cloudVariantId=item.cloudVariantId,
            quantity=item.quantity,
            unitPriceCents=to_cents(line.unit_price),
            lineTotalCents=to_cents(line.line_total),
        ))

    tier_name = None
    if member.member_level_id:
        tier_name = (await db.execute(
            select(MemberLevel.name).where(MemberLevel.id == member.member_level_id)
        )).scalar_one_or_none()

    promo_total = getattr(result, "coupon_discount", Decimal("0"))
    promo_labels = [r.label for r in getattr(result, "promotions_applied", []) if r.applied and r.amount > 0]
    return PosQuoteOut(
        ok=True, memberId=member.id, tierName=tier_name, lines=lines,
        memberDiscountCents=to_cents(result.member_discount),
        promoDiscountCents=to_cents(promo_total),
        promoLabel="; ".join(promo_labels) if promo_labels else None,
        quoteToken=sign_quote(
            tenant_id=tenant_id, store_id=store_id, member_id=member.id, lines=lines,
        ),
    )


async def quote_promotions(
    db: AsyncSession, tenant_id: int, body, store_id: int = 0,
):
    """纯促销报价：不要求会员，只跑促销引擎算折扣。"""
    from .schemas import PosPromoQuoteOut

    member = None
    if body.memberId:
        member = await resolve_member(db, tenant_id, body.memberId)

    try:
        items = [PricingCartItem(
            product_id=i.cloudProductId,
            variant_id=i.cloudVariantId,
            qty=Decimal(str(i.quantity)),
        ) for i in body.items]
        result = await calculate_pricing(db, member, tenant_id, items, is_pickup=True, channel="pos")
    except HTTPException as exc:
        return PosPromoQuoteOut(ok=False, reason=str(exc.detail))

    promo_total = getattr(result, "coupon_discount", Decimal("0"))
    promo_labels = [r.label for r in getattr(result, "promotions_applied", []) if r.applied and r.amount > 0]
    applied_ids = [r.discount_id for r in getattr(result, "promotions_applied", []) if r.applied and r.amount > 0]
    discount_product_ids: list[int] = []
    if applied_ids:
        promos = (await db.execute(select(Discount).where(Discount.id.in_(applied_ids)))).scalars().all()
        for promo in promos:
            discount_product_ids.extend(promo.discount_product_ids or promo.condition_product_ids or [])
    priced = {(line.product_id, line.variant_id): line for line in result.lines}
    quote_lines = [PosQuoteLineOut(
        lineNo=item.lineNo, cloudProductId=item.cloudProductId,
        cloudVariantId=item.cloudVariantId, quantity=item.quantity,
        unitPriceCents=to_cents(priced[(item.cloudProductId, item.cloudVariantId)].unit_price),
        lineTotalCents=to_cents(priced[(item.cloudProductId, item.cloudVariantId)].line_total),
    ) for item in body.items]

    return PosPromoQuoteOut(
        ok=True,
        promoDiscountCents=to_cents(promo_total),
        promoLabel="; ".join(promo_labels) if promo_labels else None,
        quoteToken=sign_quote(
            tenant_id=tenant_id, store_id=store_id, member_id=member.id if member else 0,
            lines=quote_lines, promotion_discount_cents=to_cents(promo_total),
        ),
        discountProductIds=list(dict.fromkeys(discount_product_ids)),
    )


# ── 会员报价签名 ───────────────────────────────────────────────
#
# 报价必须可验证，否则 POS 可以自己编一个会员价发上来。签名覆盖租户、会员、
# 每行单价和折扣总额，并带过期时间——改任何一项，同步时都会验签失败。

QUOTE_TTL_SECONDS = 15 * 60


def quote_fingerprint(
    *, tenant_id: int, member_id: int, lines, store_id: int = 0,
    promotion_discount_cents: int = 0,
) -> str:
    """报价内容指纹。

    覆盖商品、SKU、**数量**和单价——数量必须进签名，否则拿 1 件的报价可以套用到
    任意数量。刻意不含折扣总额：折扣已经体现在每行单价里，而订单载荷里并不携带它，
    把它写进签名会让所有真实会员订单验签失败。
    行按内容排序，与购物车顺序无关。
    """
    def key(line):
        get = (lambda k: getattr(line, k, None)) if hasattr(line, "cloudProductId") else line.get
        return (
            int(get("cloudProductId")),
            int(get("cloudVariantId") or 0),
            str(get("quantity")),
            int(get("unitPriceCents")),
            int(get("lineTotalCents") if get("lineTotalCents") is not None else -1),
        )

    parts = sorted(key(line) for line in lines)
    payload = {"t": tenant_id, "s": store_id, "m": member_id, "l": parts}
    # Keep old zero-promotion member-quote tokens valid during a staged rollout.
    if promotion_discount_cents:
        payload["p"] = promotion_discount_cents
    raw = json.dumps(
        payload,
        separators=(",", ":"), sort_keys=True,
    )
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()


def sign_quote(*, tenant_id: int, member_id: int, lines, store_id: int = 0,
               promotion_discount_cents: int = 0, now: datetime | None = None) -> str:
    now = now or datetime.utcnow()
    expires = int((now + timedelta(seconds=QUOTE_TTL_SECONDS)).timestamp())
    fp = quote_fingerprint(
        tenant_id=tenant_id, store_id=store_id, member_id=member_id, lines=lines,
        promotion_discount_cents=promotion_discount_cents,
    )
    body = f"{fp}.{expires}"
    mac = hmac.new(get_settings().SECRET_KEY.encode("utf-8"),
                   body.encode("utf-8"), hashlib.sha256).hexdigest()
    return f"{body}.{mac}"


def verify_quote(token: str | None, *, tenant_id: int, member_id: int, lines,
                 store_id: int = 0, promotion_discount_cents: int = 0,
                 now: datetime | None = None) -> bool:
    if not token:
        return False
    try:
        fp, expires_raw, mac = token.split(".")
        expires = int(expires_raw)
    except (ValueError, AttributeError):
        return False

    body = f"{fp}.{expires}"
    expected = hmac.new(get_settings().SECRET_KEY.encode("utf-8"),
                        body.encode("utf-8"), hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, mac):
        return False
    if (now or datetime.utcnow()).timestamp() > expires:
        return False
    return hmac.compare_digest(fp, quote_fingerprint(
        tenant_id=tenant_id, store_id=store_id, member_id=member_id, lines=lines,
        promotion_discount_cents=promotion_discount_cents))


def member_quote_lines(items) -> list[dict]:
    """Return the signed member-price view, before a local line-level override."""
    out = []
    for item in items:
        adjusted = item.manualUnitPriceCents is not None or item.lineDiscountCents > 0
        quoted_unit = item.baseUnitPriceCents if adjusted and item.baseUnitPriceCents is not None else item.unitPriceCents
        quoted_total = (
            to_cents(Decimal(quoted_unit) * Decimal(str(item.quantity)) / 100)
            if adjusted and item.baseUnitPriceCents is not None
            else item.lineTotalCents
        )
        out.append({
            "cloudProductId": item.cloudProductId, "cloudVariantId": item.cloudVariantId,
            "quantity": item.quantity, "unitPriceCents": quoted_unit,
            "lineTotalCents": quoted_total,
        })
    return out


def existing_sync_result(existing, body: PosSyncOrderIn, phash: str) -> PosSyncOrderOut | None:
    if existing.idempotency_key != body.idempotencyKey or existing.payload_hash != phash:
        return PosSyncOrderOut(ok=False, syncStatus="conflict", reason="idempotency_conflict")
    if existing.sync_status == "synced":
        return PosSyncOrderOut(
            ok=True, syncStatus="synced", cloudOrderId=existing.cloud_order_id,
            cloudOrderNo=getattr(existing, "cloud_order_no", None),
        )
    return None


async def sync_order(db: AsyncSession, body: PosSyncOrderIn, bearer_token: str) -> PosSyncOrderOut:
    await authenticate_lane(db, body, bearer_token)
    phash = payload_hash(body)

    existing = await _find_existing_sync(db, body)
    if existing:
        result = existing_sync_result(existing, body, phash)
        if result is not None:
            return result
        # A conflict has no cloud order. Re-run the current validation rules rather than
        # reporting a false idempotent success after a rule/configuration repair.
        await db.delete(existing)
        await db.flush()

    amount_error = validate_order_amounts(body)
    if amount_error:
        return await _record_conflict(db, body, phash, amount_error)

    # 先解析会员身份，让 catalogue_prices 能按会员等级评估 POS 规则命中
    member = await resolve_member(db, body.tenantId, body.memberId)
    if body.memberId and member is None:
        return await _record_conflict(db, body, phash, "member_not_found")

    conflict, catalogue_prices = await _validate_products(db, body, member=member)
    if conflict:
        return await _record_conflict(db, body, phash, conflict)

    # 选了会员就归属会员（店内消费计入其累计/等级）。
    # 会员无效时不能静默回落 walk-in：这单是按会员价收的钱，归属悄悄变了就对不上账。
    # member 已在 catalogue_prices 解析阶段被读取过，此处直接复用避免再查一次。
    pass

    # 只要挂了会员，价格就必须有云端签名背书。
    # 触发条件刻意不是 memberBenefitApplied——那是客户端自报的，把它设成 false
    # 就能带着会员 ID 同步任意低价订单。
    member_quote_valid = False
    if member is not None:
        member_quote_valid = verify_quote(
            body.quoteToken, tenant_id=body.tenantId, store_id=body.storeId,
            member_id=member.id, lines=member_quote_lines(body.items),
        )
        if not member_quote_valid:
            return await _record_conflict(db, body, phash, "member_quote_invalid")

    promotion_quote_valid = body.promotionDiscountCents == 0 and not body.promotionApplied
    if body.promotionDiscountCents or body.promotionApplied:
        quoted_lines = [
            {"cloudProductId": i.cloudProductId, "cloudVariantId": i.cloudVariantId,
             "quantity": i.quantity, "unitPriceCents": i.unitPriceCents,
             "lineTotalCents": i.lineTotalCents}
            for i in body.items
        ]
        promotion_quote_valid = verify_quote(
            body.quoteToken, tenant_id=body.tenantId, store_id=body.storeId,
            member_id=member.id if member else 0, lines=quoted_lines,
            promotion_discount_cents=body.promotionDiscountCents,
        )
        if not promotion_quote_valid:
            return await _record_conflict(db, body, phash, "promotion_quote_invalid")

    price_adjustments = (
        price_override_adjustments(body)
        if price_override_applied(
            body,
            catalogue_prices=catalogue_prices,
            member_quote_valid=member_quote_valid or promotion_quote_valid,
        )
        else None
    )
    # 门店可关闭改价二次审批；关闭后不再校验 evidence
    if price_adjustments is not None and _price_override_approval_required(
        db, body.tenantId, body.storeId,
    ) and not verify_price_override_evidence(
        body.priceOverrideEvidence, sync_token=bearer_token, body=body,
    ):
        return await _record_conflict(db, body, phash, "price_override_invalid")

    required_age = await _required_age(db, body)
    if required_age and not verify_age_approval_evidence(
        body.ageApprovalEvidence, sync_token=bearer_token, body=body,
        required_age=required_age,
    ):
        return await _record_conflict(db, body, phash, "age_approval_invalid")

    customer = member or await _get_walk_in_customer(db, body.tenantId)

    # 余额支付校验：流水必须真实（来自 /pos/wallet/charge）、归属本单客户、金额与币种
    # 一致、且未被占用。校验即锁行，order 落库后回写 source_id 完成占用——同一笔扣款
    # 不可能再挂到其它订单上。
    wallet_error, wallet_txns = await _validate_balance_payments(db, body, customer.id)
    if wallet_error:
        return await _record_conflict(db, body, phash, wallet_error)

    created_at = parse_created_at(body.createdAt)
    order = Order(
        tenant_id=body.tenantId,
        customer_id=customer.id,
        order_no=f"POS-{body.localOrderNo}"[:50],
        status=POS_SYNC_ORDER_STATUS,
        subtotal=cents(body.subtotalCents),
        discount_total=cents(body.discountCents),
        shipping_total=Decimal("0.00"),
        tax_total=cents(body.taxCents),
        grand_total=cents(body.totalCents),
        currency=body.currency,
        shipping_address=_walk_in_address(),
        billing_address=None,
        note=None,
        paid_at=created_at,
        extra_attributes={
            "source": "pos",
            "pos_local_order_no": body.localOrderNo,
            "pos_lane_id": body.laneId,
            "pos_cash_rounding_cents": body.cashRoundingCents,
            "pos_payable_cents": body.payableCents,
            "pos_created_offline": body.createdOffline,
            "pos_member_id": body.memberId,
            "pos_member_benefit_applied": body.memberBenefitApplied,
            "pos_promotion_applied": body.promotionApplied,
            "pos_price_adjustments": price_adjustments,
            "pos_price_override_evidence": (
                body.priceOverrideEvidence if price_adjustments is not None else None
            ),
            # 小票快照留档（云端兜底重打的数据源；payload_json 也存了全量副本）。
            "pos_receipt_html": body.receiptHtml,
            "pos_receipt_hash": body.receiptHash,
            "pos_receipt_template_key": body.receiptTemplateKey,
            "pos_receipt_template_version": body.receiptTemplateVersion,
            "pos_receipt_rules_version": body.receiptRulesVersion,
            "pos_print_status": body.printStatus,
        },
    )
    db.add(order)
    await db.flush()

    cart_items: list[PricingCartItem] = []
    for item in body.items:
        quantity = Decimal(str(item.quantity)).quantize(Decimal("0.01"))
        db.add(OrderItem(
            tenant_id=body.tenantId,
            order_id=order.id,
            product_id=item.cloudProductId,
            variant_id=item.cloudVariantId,
            product_snapshot={
                "name": item.name,
                "sku": item.sku,
                "barcode": item.barcode,
                "unit": item.unit,
                "sold_by_weight": item.soldByWeight,
                "weight_kg": item.weightKg,
            },
            quantity=quantity,
            unit_price=cents(item.unitPriceCents),
            total_price=cents(item.lineTotalCents),
            tax_rate=Decimal(str(item.effectiveTaxRate)),
            tax_amount=cents(item.taxCents),
        ))
        cart_items.append(PricingCartItem(
            product_id=item.cloudProductId,
            variant_id=item.cloudVariantId,
            qty=quantity,
        ))

    for payment in body.payments:
        extra_data = {
            "source": "pos",
            "payment_method": payment.paymentMethod,
            "tendered_cents": payment.tenderedCents,
            "change_cents": payment.changeCents,
            # 脱敏存根留档（Agent 已脱敏，不含完整卡号/track2）。
            "eftpos_receipt_masked": payment.eftposReceiptMasked,
        }
        if payment.paymentMethod == "balance":
            # 记下钱包流水 id，对账可追溯到具体扣款。
            extra_data["wallet_txn_id"] = payment.walletTxnId
        db.add(Payment(
            tenant_id=body.tenantId,
            order_id=order.id,
            gateway=payment.provider or payment.paymentMethod,
            amount=cents(payment.amountCents),
            currency=body.currency,
            status="completed" if payment.status == "approved" else payment.status,
            gateway_ref=payment.providerTxnRef,
            extra_data=extra_data,
        ))

    await deduct_stock(db, body.tenantId, cart_items, order.id, force_oversell=True)
    # 占用余额流水：source_id 从 NULL → 本单 id，完成"校验 + 占用"的收尾。
    # 并发订单若已占用，前面校验的 source_id IS NULL 条件会自然落空，不会走到这里。
    for wallet_txn in wallet_txns:
        wallet_txn.source_id = order.id
    sync = PosOrderSync(
        tenant_id=body.tenantId,
        lane_id=body.laneId,
        local_order_no=body.localOrderNo,
        idempotency_key=body.idempotencyKey,
        payload_hash=phash,
        cloud_order_id=order.id,
        sync_status="synced",
        payload_json=body.model_dump(mode="json"),
        price_adjustments_json=price_adjustments,
        price_override_evidence=(
            body.priceOverrideEvidence if price_adjustments is not None else None
        ),
    )
    db.add(sync)
    await db.commit()

    order_created.send(sender=Order, order=order, tenant_id=body.tenantId)
    order_paid.send(sender=Order, order=order, tenant_id=body.tenantId)
    return PosSyncOrderOut(ok=True, syncStatus="synced", cloudOrderId=order.id, cloudOrderNo=order.order_no)


def validate_order_amounts(body: PosSyncOrderIn) -> str | None:
    line_discount_total = 0
    subtotal = 0
    line_totals: list[int] = []
    for item in body.items:
        try:
            quantity = Decimal(item.quantity)
        except (InvalidOperation, TypeError, ValueError):
            return "line_total_mismatch"
        if not quantity.is_finite() or quantity <= 0:
            return "line_total_mismatch"
        gross = int(
            (Decimal(item.unitPriceCents) * quantity).quantize(
                Decimal("1"), rounding=ROUND_HALF_UP,
            )
        )
        if gross - item.lineDiscountCents != item.lineTotalCents:
            return "line_total_mismatch"
        subtotal += gross
        line_discount_total += item.lineDiscountCents
        line_totals.append(item.lineTotalCents)

    if subtotal != body.subtotalCents:
        return "subtotal_mismatch"
    if line_discount_total + body.orderDiscountCents + body.promotionDiscountCents != body.discountCents:
        return "discount_mismatch"
    if body.orderDiscountCents + body.promotionDiscountCents > sum(line_totals):
        return "discount_mismatch"
    if body.subtotalCents - body.discountCents != body.totalCents:
        return "total_mismatch"
    if sum(payment.amountCents for payment in body.payments) != body.payableCents:
        return "payment_mismatch"

    allocations = _allocate_cents(body.orderDiscountCents + body.promotionDiscountCents, line_totals)
    expected_taxes: list[int] = []
    for item, allocation in zip(body.items, allocations, strict=True):
        if not item.pricesIncludeTax:
            expected_taxes.append(0)
            continue
        try:
            rate = Decimal(item.effectiveTaxRate)
            if not rate.is_finite() or rate < 0:
                return "tax_mismatch"
            consideration = Decimal(item.lineTotalCents - allocation)
            expected_taxes.append(int(
                (consideration * rate / (Decimal("1") + rate)).quantize(
                    Decimal("1"), rounding=ROUND_HALF_UP,
                )
            ) if rate else 0)
        except (InvalidOperation, TypeError, ValueError):
            return "tax_mismatch"
    if (
        any(item.taxCents != expected for item, expected in zip(body.items, expected_taxes, strict=True))
        or sum(expected_taxes) != body.taxCents
    ):
        return "tax_mismatch"

    has_eftpos = any(payment.paymentMethod == "eftpos" for payment in body.payments)
    if has_eftpos:
        if body.cashRoundingCents != 0 or body.payableCents != body.totalCents:
            return "rounding_mismatch"
    elif (
        body.cashRoundingCents < -4
        or body.cashRoundingCents > 5
        or body.payableCents != body.totalCents + body.cashRoundingCents
    ):
        return "rounding_mismatch"
    return None


def _allocate_cents(total: int, weights: list[int]) -> list[int]:
    if total == 0:
        return [0] * len(weights)
    denominator = sum(weights)
    if denominator == 0:
        return [0] * len(weights)
    shares = [total * weight // denominator for weight in weights]
    remainders = [total * weight % denominator for weight in weights]
    for index in sorted(range(len(weights)), key=lambda i: (-remainders[i], i))[
        : total - sum(shares)
    ]:
        shares[index] += 1
    return shares


async def _find_existing_sync(db: AsyncSession, body: PosSyncOrderIn) -> PosOrderSync | None:
    result = await db.execute(
        select(PosOrderSync).where(
            PosOrderSync.tenant_id == body.tenantId,
            or_(
                PosOrderSync.idempotency_key == body.idempotencyKey,
                PosOrderSync.local_order_no == body.localOrderNo,
            ),
        )
    )
    return result.scalars().first()


async def _validate_products(
    db: AsyncSession, body: PosSyncOrderIn, member: Customer | None = None,
) -> tuple[str | None, list[int]]:
    """POS 订单落单复核：返回每行的"云端预期单价"（整数分）。

    预期单价按 POS 渠道 + 当前时间 + 当前数量解析 ProductPriceRule 命中规则；
    没有命中走原 base+modifier，作为云端权威。后端用这个列表判断 POS 上送单价
    是否偏离预期 —— 仅偏离时才进入人工改价 evidence 校验。

    ponytail: 每条都跑规则解析器，规则数量很小（典型 < 10/商品），不是热路径。
    """
    product_ids = {item.cloudProductId for item in body.items}
    result = await db.execute(
        select(Product.id, Product.base_price).where(
            Product.tenant_id == body.tenantId, Product.id.in_(product_ids),
        )
    )
    product_prices = {row[0]: row[1] for row in result.all()}
    if product_prices.keys() != product_ids:
        return "product_not_found", []

    variant_items = [item for item in body.items if item.cloudVariantId is not None]
    variant_prices: dict[int, tuple[int, Decimal]] = {}
    if variant_items:
        pairs = [
            and_(
                ProductVariant.id == item.cloudVariantId,
                ProductVariant.product_id == item.cloudProductId,
            )
            for item in variant_items
        ]
        vr = await db.execute(
            select(
                ProductVariant.id, ProductVariant.product_id, ProductVariant.price_modifier,
            ).where(ProductVariant.tenant_id == body.tenantId, or_(*pairs))
        )
        variant_prices = {row[0]: (row[1], row[2]) for row in vr.all()}
        expected = {item.cloudVariantId for item in variant_items}
        if variant_prices.keys() != expected:
            return "variant_not_found", []

    now = datetime.utcnow()
    member_level_id = member.member_level_id if member else None
    prices: list[int] = []
    for item in body.items:
        base = Decimal(product_prices[item.cloudProductId]) + (
            Decimal(variant_prices[item.cloudVariantId][1])
            if item.cloudVariantId is not None
            else Decimal("0")
        )
        qty = Decimal(str(item.quantity))
        rule, rule_unit = await _resolve_rule_for_line(
            db,
            tenant_id=body.tenantId,
            product_id=item.cloudProductId,
            variant_id=item.cloudVariantId,
            channel="pos",
            member_level_id=member_level_id,
            qty=qty,
            now=now,
        )
        unit = rule_unit if rule_unit is not None else base
        prices.append(to_cents(unit))
    return None, prices


async def _required_age(db: AsyncSession, body: PosSyncOrderIn) -> int:
    product_ids = {item.cloudProductId for item in body.items}
    rows = (await db.execute(select(Product.id, Product.extra_attributes).where(
        Product.tenant_id == body.tenantId, Product.id.in_(product_ids),
    ))).all()
    required = 0
    for _, attributes in rows:
        attributes = attributes if isinstance(attributes, dict) else {}
        try:
            required = max(required, int(attributes.get("pos_minimum_age") or 0))
        except (TypeError, ValueError):
            continue
    return required


async def _price_override_approval_required(db: AsyncSession, tenant_id: int, store_id: int) -> bool:
    """ponytail: lazy import to avoid pos_operations ↔ pos_sync circular deps."""
    from app.plugins.pos_operations.services import get_rules
    rules = await get_rules(db, tenant_id=tenant_id, store_id=store_id)
    return bool(rules.price_override_approval_required)


def _age_cart_hash(items) -> str:
    parts = sorted(
        f"{item.cloudProductId}|{item.quantity}|{item.unitPriceCents}" for item in items
    )
    value = 0x811C9DC5
    for char in ";".join(parts):
        value ^= ord(char)
        value = (value * 0x01000193) & 0xFFFFFFFF
    return f"{value:08x}"


def price_override_content_hash(body: PosSyncOrderIn) -> str:
    raw = json.dumps({
        "lines": [[
            item.cloudProductId,
            item.cloudVariantId,
            item.quantity,
            item.baseUnitPriceCents,
            item.listPriceCents,
            item.manualUnitPriceCents,
            item.unitPriceCents,
            item.lineTotalCents,
            item.lineDiscountCents,
        ] for item in body.items],
        "orderDiscountCents": body.orderDiscountCents,
    }, separators=(",", ":"), ensure_ascii=False)
    value = 0x811C9DC5
    encoded = raw.encode("utf-16-le")
    for index in range(0, len(encoded), 2):
        value ^= encoded[index] | (encoded[index + 1] << 8)
        value = (value * 0x01000193) & 0xFFFFFFFF
    return f"{value:x}"


def price_override_adjustments(body: PosSyncOrderIn) -> dict:
    return {
        "orderDiscountCents": body.orderDiscountCents,
        "lines": [{
            "lineNo": index,
            "cloudProductId": item.cloudProductId,
            "cloudVariantId": item.cloudVariantId,
            "baseUnitPriceCents": (
                item.baseUnitPriceCents
                if item.baseUnitPriceCents is not None
                else item.unitPriceCents
            ),
            "listPriceCents": item.listPriceCents,
            "manualUnitPriceCents": item.manualUnitPriceCents,
            "lineDiscountCents": item.lineDiscountCents,
        } for index, item in enumerate(body.items, start=1)],
    }


def price_override_applied(
    body: PosSyncOrderIn, *, catalogue_prices: list[int] | None = None,
    member_quote_valid: bool = False,
) -> bool:
    if (
        body.priceOverrideApplied
        or body.priceOverrideEvidence is not None
        or body.discountCents > 0
        or body.orderDiscountCents > 0
        or any(
            item.manualUnitPriceCents is not None or item.lineDiscountCents > 0
            for item in body.items
        )
    ):
        return True
    if member_quote_valid:
        return False
    if any(
        item.unitPriceCents != (
            item.listPriceCents
            if item.listPriceCents is not None
            else item.baseUnitPriceCents
        )
        for item in body.items
        if item.listPriceCents is not None or item.baseUnitPriceCents is not None
    ):
        return True
    return catalogue_prices is not None and (
        len(catalogue_prices) != len(body.items)
        or any(
            item.unitPriceCents != catalogue_price
            for item, catalogue_price in zip(
                body.items, catalogue_prices, strict=True,
            )
        )
    )


def verify_price_override_evidence(
    evidence: dict | None, *, sync_token: str, body: PosSyncOrderIn,
) -> bool:
    if (
        not isinstance(evidence, dict)
        or not sync_token
        or not body.clientRequestId
        or body.cashierId is None
    ):
        return False
    signature = evidence.get("signature")
    if not isinstance(signature, str):
        return False
    claims = {key: value for key, value in evidence.items() if key != "signature"}
    expected = {
        "actionType": "price_override",
        "tenantId": body.tenantId,
        "storeId": body.storeId,
        "laneId": body.laneId,
        "operatorUserId": body.cashierId,
        "clientRequestId": body.clientRequestId,
        "contentHash": price_override_content_hash(body),
        "adjustments": price_override_adjustments(body),
    }
    if any(claims.get(key) != value for key, value in expected.items()):
        return False
    approver = claims.get("approverUserId")
    if not isinstance(approver, int) or approver == body.cashierId:
        return False
    created_at = claims.get("createdAt")
    expires_at = claims.get("expiresAt")
    if not isinstance(created_at, int) or not isinstance(expires_at, int):
        return False
    try:
        sale_at = datetime.fromisoformat(body.createdAt.replace("Z", "+00:00"))
        if sale_at.tzinfo is None:
            sale_at = sale_at.replace(tzinfo=timezone.utc)
        sale_ms = int(sale_at.timestamp() * 1000)
    except (TypeError, ValueError):
        return False
    if (
        not created_at <= sale_ms <= expires_at
        or expires_at - created_at > PRICE_OVERRIDE_APPROVAL_TTL_MS
    ):
        return False
    raw = json.dumps(claims, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
    digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()
    expected_signature = hmac.new(
        sync_token.encode("utf-8"), digest.encode("ascii"), hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(signature, expected_signature)


def verify_age_approval_evidence(
    evidence: dict | None, *, sync_token: str, body: PosSyncOrderIn, required_age: int,
) -> bool:
    if not isinstance(evidence, dict) or not sync_token:
        return False
    signature = evidence.get("signature")
    if not isinstance(signature, str):
        return False
    claims = {key: value for key, value in evidence.items() if key != "signature"}
    expected = {
        "actionType": "age_sale", "tenantId": body.tenantId, "storeId": body.storeId,
        "laneId": body.laneId, "operatorUserId": body.cashierId,
        "clientRequestId": body.clientRequestId, "contentHash": _age_cart_hash(body.items),
        "requiredAge": required_age,
    }
    if any(claims.get(key) != value for key, value in expected.items()):
        return False
    if claims.get("approverUserId") in (None, body.cashierId):
        return False
    created_at = claims.get("createdAt")
    expires_at = claims.get("expiresAt")
    try:
        sale_ms = int(parse_created_at(body.createdAt).timestamp() * 1000)
    except Exception:
        return False
    if not isinstance(created_at, int) or not isinstance(expires_at, int):
        return False
    if (
        not created_at <= sale_ms <= expires_at
        or expires_at - created_at > AGE_APPROVAL_TTL_MS
    ):
        return False
    raw = json.dumps(claims, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
    digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()
    expected_signature = hmac.new(
        sync_token.encode("utf-8"), digest.encode("ascii"), hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(signature, expected_signature)


async def _record_conflict(
    db: AsyncSession, body: PosSyncOrderIn, phash: str, reason: str
) -> PosSyncOrderOut:
    db.add(PosOrderSync(
        tenant_id=body.tenantId,
        lane_id=body.laneId,
        local_order_no=body.localOrderNo,
        idempotency_key=body.idempotencyKey,
        payload_hash=phash,
        sync_status="conflict",
        conflict_reason=reason,
        payload_json=body.model_dump(mode="json"),
        price_adjustments_json=(
            price_override_adjustments(body) if price_override_applied(body) else None
        ),
        price_override_evidence=(
            body.priceOverrideEvidence if price_override_applied(body) else None
        ),
    ))
    await db.commit()
    return PosSyncOrderOut(ok=False, syncStatus="conflict", reason=reason)


async def _get_walk_in_customer(db: AsyncSession, tenant_id: int) -> Customer:
    email = f"pos-walkin+{tenant_id}@local.pos"
    result = await db.execute(select(Customer).where(Customer.tenant_id == tenant_id, Customer.email == email))
    customer = result.scalar_one_or_none()
    if customer:
        return customer
    customer = Customer(
        tenant_id=tenant_id,
        email=email,
        name="POS Walk-in",
        is_active=1,
        points_balance=0,
        total_spent=Decimal("0.00"),
        orders_count=0,
        extra_data={"source": "pos_sync"},
    )
    db.add(customer)
    await db.flush()
    return customer


def _walk_in_address() -> dict:
    return {
        "name": "POS Walk-in",
        "phone": "",
        "country": "",
        "province": "",
        "city": "",
        "district": "",
        "address": "POS sale",
        "zip_code": "",
    }


async def close_conflict_record(
    db: AsyncSession, tenant_id: int, lane_id: str,
    local_order_no: str, reason: str, approver_user_id: int | None,
) -> dict:
    result = await db.execute(
        select(PosOrderSync).where(
            PosOrderSync.tenant_id == tenant_id,
            PosOrderSync.local_order_no == local_order_no,
            PosOrderSync.sync_status == "conflict",
        )
    )
    record = result.scalar_one_or_none()
    if not record:
        raise HTTPException(status_code=404, detail="conflict record not found")
    payload = dict(record.payload_json or {})
    payload["conflictClosure"] = {
        "reason": reason,
        "approverUserId": approver_user_id,
        "closedAt": datetime.now(timezone.utc).isoformat(),
    }
    record.payload_json = payload
    record.sync_status = "closed_local_only"
    await db.commit()
    return {"ok": True, "localOrderNo": local_order_no, "syncStatus": "closed_local_only"}