"""对外库存服务。core/services/inventory.py 通过本模块委托。"""
import hashlib
from decimal import Decimal

from fastapi import HTTPException
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession


def _ledger_key(doc_type: str, doc_id, pid: int, vid: int, bkey: int, suffix: str) -> str:
    """确定性幂等键：由「业务操作身份」派生，不是随机 UUID。

    - 同一业务操作的网络重试/重放 → 同键 → 唯一约束挡住重复库存事实
    - 不同业务操作（部分退款、改单重锁、不同状态事件）→ suffix 不同 → 不误撞
    suffix 由调用方按业务操作 ID 传入（退款单 ID、订单版本、状态事件），
    缺省用状态事件；POS 离线单另有确定键（见 _write_offline_leg）。
    """
    raw = f"{doc_type}:{doc_id}:{pid}:{vid}:{bkey}:{suffix}"
    return hashlib.sha1(raw.encode()).hexdigest()


def _operation_key(doc_type: str, doc_id, suffix: str = "") -> str:
    """一个库存业务操作的确定性认领键（不是某一条批次/余额腿的键）。"""
    raw = f"{doc_type}:{doc_id}:{suffix}"
    return hashlib.sha256(raw.encode()).hexdigest()


async def _claim_operation(db: AsyncSession, tenant_id: int, doc_type: str, doc_id,
                           suffix: str = "") -> bool:
    """原子认领业务操作；False 表示同一操作已成功提交或正在由另一请求处理。

    认领和后续库存写入必须处在同一个外层事务：后续校验/写入失败时，认领随之回滚，
    客户端可安全重试。该短路必须发生在 FEFO、成本层和可用量校验之前。
    """
    r = await db.execute(
        text("INSERT IGNORE INTO inventory_idempotency (tenant_id, op_key, created_at) "
             "VALUES (:tid,:key,NOW(3))"),
        {"tid": tenant_id, "key": _operation_key(doc_type, doc_id, suffix)},
    )
    return bool(r.rowcount)


async def _txn_exists(db: AsyncSession, tenant_id: int, key: str) -> bool:
    """该幂等键是否已写过流水（已提交的重复请求）。用于优雅去重：
    命中则整条腿跳过（余额已在首次应用过），正常返回而非撞唯一约束报错。"""
    r = await db.execute(
        text("SELECT 1 FROM inventory_transactions WHERE tenant_id=:tid AND idempotency_key=:k LIMIT 1"),
        {"tid": tenant_id, "k": key},
    )
    return r.fetchone() is not None

from app.plugins.inventory.lifecycle import LifecycleState
from app.plugins.inventory.ledger import TxnRequest, build_txn
from app.plugins.inventory.policy import InventoryPolicy
from app.plugins.inventory.projection import project

Key = tuple[int, int]  # (product_id, variant_id)，0 表示无规格


def state_from_settings_row(raw: str | None) -> LifecycleState | None:
    """无记录或值非法 → None（未接管，走旧逻辑）。"""
    try:
        return LifecycleState(raw)
    except ValueError:
        return None


async def get_state(db: AsyncSession, tenant_id: int) -> LifecycleState | None:
    r = await db.execute(
        text("SELECT lifecycle_state FROM inventory_settings WHERE tenant_id=:tid LIMIT 1"),
        {"tid": tenant_id},
    )
    row = r.fetchone()
    return state_from_settings_row(row[0] if row else None)


def check_availability(balances: dict[Key, Decimal],
                       wants: list[tuple[int, int, Decimal]],
                       *, allow_negative: bool) -> None:
    if allow_negative:
        return
    errors = []
    for pid, vid, qty in wants:
        avail = balances.get((pid, vid), Decimal("0"))
        if avail < qty:
            errors.append({"product_id": pid, "variant_id": vid or None,
                           "available": str(avail), "required": str(qty),
                           "msg": f"商品 {pid} 库存不足（剩余 {avail}，需 {qty}）"})
    if errors:
        raise HTTPException(status_code=400, detail=errors)


def plan_deduction(wants: list[tuple[int, int, Decimal]]) -> list[tuple[int, int, Decimal]]:
    return [(pid, vid, -qty) for pid, vid, qty in wants]


async def _load_balances(db: AsyncSession, tenant_id: int, keys: list[Key],
                         warehouse_id: int | None = None) -> dict[Key, Decimal]:
    if not keys:
        return {}
    pids = sorted({k[0] for k in keys})
    where_warehouse = " AND warehouse_id=:wh" if warehouse_id is not None else ""
    params = {"tid": tenant_id}
    if warehouse_id is not None:
        params["wh"] = warehouse_id
    r = await db.execute(
        text("SELECT product_id, variant_id, SUM(qty) FROM inventory_balances "
             "WHERE tenant_id=:tid AND stock_state='sellable' "
             f"AND product_id IN ({','.join(str(int(p)) for p in pids)}) {where_warehouse} "
             "GROUP BY product_id, variant_id FOR UPDATE"),
        params,
    )
    return {(row[0], row[1]): Decimal(str(row[2] or 0)) for row in r.fetchall()}


async def _default_warehouse_id(db: AsyncSession, tenant_id: int) -> int:
    r = await db.execute(
        text("SELECT default_warehouse_id FROM inventory_settings WHERE tenant_id=:tid LIMIT 1"),
        {"tid": tenant_id},
    )
    row = r.fetchone()
    if not row or not row[0]:
        raise HTTPException(status_code=409, detail="未配置默认仓库，请先在进销存设置中指定")
    warehouse_id = int(row[0])
    exists = await db.execute(
        text("SELECT id FROM inventory_warehouses WHERE id=:id AND tenant_id=:tid AND is_active=1"),
        {"id": warehouse_id, "tid": tenant_id},
    )
    if not exists.fetchone():
        raise HTTPException(status_code=409, detail="默认仓库无效、已停用或不属于当前租户")
    return warehouse_id


async def _allow_negative(db: AsyncSession, tenant_id: int) -> bool:
    r = await db.execute(
        text("SELECT allow_negative FROM inventory_settings WHERE tenant_id=:tid LIMIT 1"),
        {"tid": tenant_id},
    )
    row = r.fetchone()
    return bool(row[0]) if row else False


async def assert_period_open(db: AsyncSession, tenant_id: int) -> None:
    """关账守卫：启用关账且当前期间已关闭时，拒绝手工库存变动。

    仅约束手工后台操作（纠正/调拨/盘点/组装）；销售出入库属当前经营，不受关账冻结。
    """
    from datetime import datetime
    from app.plugins.inventory.period import period_of, assert_open
    r = await db.execute(
        text("SELECT period_close_enabled, closed_through FROM inventory_settings WHERE tenant_id=:tid LIMIT 1"),
        {"tid": tenant_id},
    )
    row = r.fetchone()
    if not row or not row[0]:
        return
    assert_open(period_of(datetime.utcnow()), row[1])


def _keys(items) -> list[tuple[int, int, Decimal]]:
    """把 PricingCartItem 或 restore 的 dict 统一成 (pid, vid, qty)。"""
    # 订单明细允许同一 SKU 出现多行；账本必须合并为一条业务腿，
    # 否则会撞上同一订单/SKU 的幂等键，且库存校验会被逐行绕过。
    totals: dict[tuple[int, int], Decimal] = {}
    order: list[tuple[int, int]] = []
    for it in items:
        if isinstance(it, dict):
            pid, vid, qty = it["product_id"], it.get("variant_id") or 0, Decimal(str(it["qty"]))
        else:
            qty = it.stock_qty_override if getattr(it, "stock_qty_override", None) is not None else it.qty
            pid, vid = it.product_id, it.variant_id or 0
        key = (int(pid), int(vid))
        if key not in totals:
            totals[key] = Decimal("0")
            order.append(key)
        totals[key] += Decimal(str(qty))
    return [(pid, vid, totals[(pid, vid)]) for pid, vid in order]


async def _apply(db: AsyncSession, tenant_id: int, wants, doc_type: str,
                 doc_id: int, sign: int, suffix: str = "") -> None:
    """写流水 + 更新余额 + 回写投影，全部在调用方事务内。

    sign=-1 扣减、sign=+1 恢复。wants 元素为 (pid, vid, qty)，qty 恒正。
    """
    wh = await _default_warehouse_id(db, tenant_id)
    balances = await _load_balances(db, tenant_id, [(p, v) for p, v, _ in wants], wh)
    # 聚合父商品增量：variant 出入库时父商品 stock_qty 同步变动，与旧逻辑展示口径一致
    product_delta: dict[int, Decimal] = {}
    for pid, vid, qty in wants:
        key = _ledger_key(doc_type, doc_id, pid, vid, 0, suffix)
        if await _txn_exists(db, tenant_id, key):
            continue  # 已应用过的重复请求：跳过该商品腿，优雅幂等
        before = balances.get((pid, vid), Decimal("0"))
        delta = qty * sign
        t = build_txn(TxnRequest(tenant_id=tenant_id, product_id=pid,
                                 variant_id=vid or None, warehouse_id=wh,
                                 qty_delta=delta, doc_type=doc_type, doc_id=doc_id),
                      qty_before=before)
        t.idempotency_key = key
        db.add(t)
        await db.execute(
            text("INSERT INTO inventory_balances "
                 "(tenant_id, product_id, variant_id, warehouse_id, location_id, batch_id, "
                 " stock_state, qty, created_at, updated_at) "
                 "VALUES (:tid,:pid,:vid,:wh,0,0,'sellable',:q,NOW(3),NOW(3)) "
                 "ON DUPLICATE KEY UPDATE qty = qty + :q, updated_at = NOW(3)"),
            {"tid": tenant_id, "pid": pid, "vid": vid, "wh": wh, "q": delta},
        )
        after = before + delta
        p = project(sellable_on_hand=after, reserved=Decimal("0"))
        if vid:
            await db.execute(
                text("UPDATE product_variants SET stock_qty=:q WHERE id=:vid AND tenant_id=:tid"),
                {"q": p.stock_qty, "vid": vid, "tid": tenant_id},
            )
        product_delta[pid] = product_delta.get(pid, Decimal("0")) + delta

    # 无规格商品直接落 products.stock_qty；有规格商品按增量累加到父商品
    for pid, delta in product_delta.items():
        await db.execute(
            text("UPDATE products SET stock_qty = stock_qty + :d WHERE id=:pid AND tenant_id=:tid"),
            {"d": delta, "pid": pid, "tid": tenant_id},
        )


# ── 两段式锁库：下单锁 → 发货出 → 取消释放 ──────────────────────

async def _reserved_leg(db: AsyncSession, tenant_id: int, pid: int, vid: int, wh: int,
                        delta: Decimal, doc_type: str, doc_id: int, suffix: str = "") -> None:
    """写一条 reserved 状态的流水 + 余额，并回写 reserved_qty 投影。"""
    key = _ledger_key(doc_type, doc_id, pid, vid, -1, suffix)
    if await _txn_exists(db, tenant_id, key):
        return  # 已应用过的重复请求：跳过整条腿，优雅幂等
    before = (await db.execute(
        text("SELECT COALESCE(SUM(qty),0) FROM inventory_balances WHERE tenant_id=:tid "
             "AND product_id=:pid AND variant_id=:vid AND warehouse_id=:wh "
             "AND stock_state='reserved' FOR UPDATE"),
        {"tid": tenant_id, "pid": pid, "vid": vid, "wh": wh},
    )).scalar()
    before = Decimal(str(before or 0))
    t = build_txn(TxnRequest(tenant_id=tenant_id, product_id=pid, variant_id=vid or None,
                             warehouse_id=wh, qty_delta=delta, doc_type=doc_type, doc_id=doc_id,
                             stock_state="reserved"), qty_before=before)
    t.idempotency_key = key
    db.add(t)
    await db.execute(
        text("INSERT INTO inventory_balances "
             "(tenant_id, product_id, variant_id, warehouse_id, location_id, batch_id, stock_state, qty, created_at, updated_at) "
             "VALUES (:tid,:pid,:vid,:wh,0,0,'reserved',:q,NOW(3),NOW(3)) "
             "ON DUPLICATE KEY UPDATE qty = qty + :q, updated_at = NOW(3)"),
        {"tid": tenant_id, "pid": pid, "vid": vid, "wh": wh, "q": delta},
    )
    # reserved_qty 投影：无规格落 products，有规格落 variant + 汇总到 products
    await db.execute(
        text("UPDATE products SET reserved_qty = GREATEST(0, reserved_qty + :d) WHERE id=:pid AND tenant_id=:tid"),
        {"d": delta, "pid": pid, "tid": tenant_id},
    )
    if vid:
        await db.execute(
            text("UPDATE product_variants SET reserved_qty = GREATEST(0, reserved_qty + :d) WHERE id=:vid AND tenant_id=:tid"),
            {"d": delta, "vid": vid, "tid": tenant_id},
        )


async def lock_via_ledger(db: AsyncSession, tenant_id: int, items, order_id: int,
                          suffix: str = "") -> None:
    """下单锁库：sellable 不动，reserved 增加，available 相应下降。"""
    if not await _claim_operation(db, tenant_id, "order_lock", order_id, suffix):
        return
    wh = await _default_warehouse_id(db, tenant_id)
    wants = _keys(items)
    # 先锁定同一默认仓内的余额行，再读可售量并写 reserved；所有操作在订单事务内，
    # 使并发下单串行化，避免两个请求都读到同一份可售库存。
    await _lock_balance_rows(db, tenant_id, [(pid, vid) for pid, vid, _ in wants], wh)
    available = await _load_available(db, tenant_id, [(pid, vid) for pid, vid, _ in wants], wh)
    check_availability(available, wants, allow_negative=await _allow_negative(db, tenant_id))
    for pid, vid, qty in wants:
        await _reserved_leg(db, tenant_id, pid, vid, wh, qty, "order_lock", order_id, suffix)


async def release_via_ledger(db: AsyncSession, tenant_id: int, items, order_id: int,
                             suffix: str = "") -> None:
    """取消/超时释放锁定：reserved 减少，sellable 不动。"""
    if not await _claim_operation(db, tenant_id, "order_release", order_id, suffix):
        return
    wh = await _default_warehouse_id(db, tenant_id)
    for pid, vid, qty in _keys(items):
        await _reserved_leg(db, tenant_id, pid, vid, wh, -qty, "order_release", order_id, suffix)


async def _cost_method(db: AsyncSession, tenant_id: int) -> str:
    r = await db.execute(
        text("SELECT cost_method FROM inventory_settings WHERE tenant_id=:tid LIMIT 1"),
        {"tid": tenant_id},
    )
    row = r.fetchone()
    return row[0] if row else "moving_avg"


async def _consume_cost_pool_legacy(db: AsyncSession, tenant_id: int, pid: int, vid: int,
                             out_qty: Decimal, method: str) -> Decimal:
    """池化成本消耗，返回 out_qty 总成本。batch_actual 不走这里（见 _consume_batch_layer）。

    - moving_avg：按现存层加权平均单价计价，并**按比例**消耗各层，保持剩余均价不变
    - fifo：最老层优先计价并消耗
    层不足的部分（负库存）成本记 0。
    """
    rows = (await db.execute(
        text("SELECT id, qty_in - qty_consumed AS rem, unit_cost FROM inventory_cost_layers "
             "WHERE tenant_id=:tid AND product_id=:pid AND variant_id=:vid AND qty_in > qty_consumed "
             "ORDER BY id FOR UPDATE"),
        {"tid": tenant_id, "pid": pid, "vid": vid},
    )).fetchall()
    if not rows:
        return Decimal("0")
    layers = [(r[0], Decimal(str(r[1])), Decimal(str(r[2]))) for r in rows]
    if method == "moving_avg":
        tot_qty = sum(l[1] for l in layers)
        avg = (sum(l[1] * l[2] for l in layers) / tot_qty) if tot_qty > 0 else layers[-1][2]
        cost = (avg * out_qty).quantize(Decimal("0.0001"))
        # 层消耗封顶在现存总量内（负库存部分按均价计成本，但不把 qty_consumed 顶穿 qty_in）
        cap = min(out_qty, tot_qty)
        consumed = Decimal("0")
        last = len(layers) - 1
        for i, (lid, rem, _c) in enumerate(layers):
            if tot_qty <= 0 or consumed >= cap:
                break
            if i < last:
                take = (rem / tot_qty * cap).quantize(Decimal("0.0001"))
                take = min(take, rem, cap - consumed)
            else:
                take = min(cap - consumed, rem)  # 末层吸收残差，仍不超过本层剩余
            consumed += take
            await db.execute(
                text("UPDATE inventory_cost_layers SET qty_consumed = qty_consumed + :t, updated_at=NOW(3) WHERE id=:id"),
                {"t": take, "id": lid},
            )
        return cost
    from app.plugins.inventory.costing import consume_fifo, Layer
    cost, _ = consume_fifo([Layer(l[1], l[2]) for l in layers], out_qty)
    remaining = out_qty
    for lid, rem, _c in layers:
        if remaining <= 0:
            break
        take = rem if rem < remaining else remaining
        await db.execute(
            text("UPDATE inventory_cost_layers SET qty_consumed = qty_consumed + :t, updated_at=NOW(3) WHERE id=:id"),
            {"t": take, "id": lid},
        )
        remaining -= take
    return cost


async def _consume_batch_layer_legacy(db: AsyncSession, tenant_id: int, pid: int, vid: int,
                               batch_id: int, qty: Decimal) -> Decimal:
    """batch_actual：跨该批次的多条成本层逐层消耗，返回加权单位成本。

    同一批次多次入库(不同进价)会有多条成本层；必须逐层消耗，不能只顶第一层，
    否则 qty_consumed 会顶穿 qty_in 且成本只取第一层单价。
    """
    rows = (await db.execute(
        text("SELECT id, qty_in - qty_consumed AS rem, unit_cost FROM inventory_cost_layers "
             "WHERE tenant_id=:tid AND product_id=:pid AND variant_id=:vid AND batch_id=:bid "
             "AND qty_in > qty_consumed ORDER BY id FOR UPDATE"),
        {"tid": tenant_id, "pid": pid, "vid": vid, "bid": batch_id},
    )).fetchall()
    if not rows:
        return Decimal("0")
    remaining = qty
    total_cost = Decimal("0")
    for lid, rem, uc in rows:
        if remaining <= 0:
            break
        rem = Decimal(str(rem)); uc = Decimal(str(uc))
        take = rem if rem < remaining else remaining
        total_cost += take * uc
        await db.execute(
            text("UPDATE inventory_cost_layers SET qty_consumed = qty_consumed + :t, updated_at=NOW(3) WHERE id=:id"),
            {"t": take, "id": lid},
        )
        remaining -= take
    consumed = qty - remaining
    return (total_cost / consumed).quantize(Decimal("0.0001")) if consumed > 0 else Decimal(str(rows[0][2]))


def _additional_take(additional_remaining: Decimal, take: Decimal, layer_remaining: Decimal) -> Decimal:
    if take == layer_remaining:
        return additional_remaining
    return (additional_remaining * take / layer_remaining).quantize(Decimal("0.0001"))


async def _consume_cost_pool(db: AsyncSession, tenant_id: int, pid: int, vid: int,
                             out_qty: Decimal, method: str) -> Decimal:
    rows = (await db.execute(
        text("SELECT id,qty_in-qty_consumed,unit_cost,additional_cost-additional_cost_consumed "
             "FROM inventory_cost_layers WHERE tenant_id=:tid AND product_id=:pid AND variant_id=:vid "
             "AND qty_in>qty_consumed ORDER BY id FOR UPDATE"),
        {"tid": tenant_id, "pid": pid, "vid": vid},
    )).fetchall()
    if not rows:
        return Decimal("0")
    layers = [(int(row[0]), Decimal(str(row[1])), Decimal(str(row[2])), Decimal(str(row[3] or 0)))
              for row in rows]
    total_qty = sum((row[1] for row in layers), Decimal("0"))
    if method == "moving_avg":
        average = sum((row[1] * row[2] + row[3] for row in layers), Decimal("0")) / total_qty
        total_cost = average * out_qty
        cap = min(out_qty, total_qty)
        consumed = Decimal("0")
        for index, (layer_id, remaining, _unit_cost, additional_remaining) in enumerate(layers):
            if consumed >= cap:
                break
            take = (remaining / total_qty * cap).quantize(Decimal("0.0001")) if index < len(layers) - 1 \
                else cap - consumed
            take = min(take, remaining, cap - consumed)
            additional = _additional_take(additional_remaining, take, remaining)
            await db.execute(
                text("UPDATE inventory_cost_layers SET qty_consumed=qty_consumed+:qty,"
                     "additional_cost_consumed=additional_cost_consumed+:additional,updated_at=NOW(3) "
                     "WHERE id=:id AND tenant_id=:tid"),
                {"qty": take, "additional": additional, "id": layer_id, "tid": tenant_id},
            )
            consumed += take
        return total_cost.quantize(Decimal("0.0001"))
    remaining_out = out_qty
    total_cost = Decimal("0")
    for layer_id, remaining, unit_cost, additional_remaining in layers:
        if remaining_out <= 0:
            break
        take = min(remaining_out, remaining)
        additional = _additional_take(additional_remaining, take, remaining)
        total_cost += take * unit_cost + additional
        await db.execute(
            text("UPDATE inventory_cost_layers SET qty_consumed=qty_consumed+:qty,"
                 "additional_cost_consumed=additional_cost_consumed+:additional,updated_at=NOW(3) "
                 "WHERE id=:id AND tenant_id=:tid"),
            {"qty": take, "additional": additional, "id": layer_id, "tid": tenant_id},
        )
        remaining_out -= take
    if remaining_out > 0:
        last = layers[-1]
        fallback = (last[1] * last[2] + last[3]) / last[1]
        total_cost += remaining_out * fallback
    return total_cost.quantize(Decimal("0.0001"))


async def _consume_batch_layer(db: AsyncSession, tenant_id: int, pid: int, vid: int,
                               batch_id: int, qty: Decimal) -> Decimal:
    rows = (await db.execute(
        text("SELECT id,qty_in-qty_consumed,unit_cost,additional_cost-additional_cost_consumed "
             "FROM inventory_cost_layers WHERE tenant_id=:tid AND product_id=:pid AND variant_id=:vid "
             "AND batch_id=:batch AND qty_in>qty_consumed ORDER BY id FOR UPDATE"),
        {"tid": tenant_id, "pid": pid, "vid": vid, "batch": batch_id},
    )).fetchall()
    if not rows:
        return Decimal("0")
    remaining_out = qty
    total_cost = Decimal("0")
    for layer_id, remaining, unit_cost, additional_remaining in rows:
        if remaining_out <= 0:
            break
        remaining = Decimal(str(remaining)); unit_cost = Decimal(str(unit_cost))
        additional_remaining = Decimal(str(additional_remaining or 0))
        take = min(remaining_out, remaining)
        additional = _additional_take(additional_remaining, take, remaining)
        total_cost += take * unit_cost + additional
        await db.execute(
            text("UPDATE inventory_cost_layers SET qty_consumed=qty_consumed+:qty,"
                 "additional_cost_consumed=additional_cost_consumed+:additional,updated_at=NOW(3) "
                 "WHERE id=:id AND tenant_id=:tid"),
            {"qty": take, "additional": additional, "id": layer_id, "tid": tenant_id},
        )
        remaining_out -= take
    consumed = qty - remaining_out
    return (total_cost / consumed).quantize(Decimal("0.0001")) if consumed > 0 else Decimal("0")


async def _sellable_leg(db: AsyncSession, tenant_id: int, pid: int, vid: int, wh: int,
                        delta: Decimal, batch_id: int | None, doc_type: str, doc_id: int,
                        unit_cost: Decimal | None, suffix: str = "") -> None:
    """写一条 sellable 余额腿 + 流水（可带 batch 与单位成本）。幂等键含 batch 与业务后缀。"""
    bkey = batch_id or 0
    key = _ledger_key(doc_type, doc_id, pid, vid or 0, bkey, suffix)
    if await _txn_exists(db, tenant_id, key):
        return  # 已应用过的重复请求：跳过整条腿，优雅幂等
    before = (await db.execute(
        text("SELECT COALESCE(SUM(qty),0) FROM inventory_balances WHERE tenant_id=:tid "
             "AND product_id=:pid AND variant_id=:vid AND warehouse_id=:wh AND batch_id=:bk "
             "AND stock_state='sellable' FOR UPDATE"),
        {"tid": tenant_id, "pid": pid, "vid": vid, "wh": wh, "bk": bkey},
    )).scalar()
    before = Decimal(str(before or 0))
    t = build_txn(TxnRequest(tenant_id=tenant_id, product_id=pid, variant_id=vid or None,
                             warehouse_id=wh, qty_delta=delta, doc_type=doc_type, doc_id=doc_id,
                             batch_id=batch_id, unit_cost=unit_cost), qty_before=before)
    t.idempotency_key = key
    db.add(t)
    await db.execute(
        text("INSERT INTO inventory_balances "
             "(tenant_id, product_id, variant_id, warehouse_id, location_id, batch_id, stock_state, qty, created_at, updated_at) "
             "VALUES (:tid,:pid,:vid,:wh,0,:bk,'sellable',:q,NOW(3),NOW(3)) "
             "ON DUPLICATE KEY UPDATE qty = qty + :q, updated_at = NOW(3)"),
        {"tid": tenant_id, "pid": pid, "vid": vid, "wh": wh, "bk": bkey, "q": delta},
    )


async def _sellable_out(db: AsyncSession, tenant_id: int, pid: int, vid: int, wh: int,
                        qty: Decimal, doc_type: str, doc_id: int, suffix: str = "") -> None:
    """出库 qty：批次租户按 FEFO 消耗批次余额（排除过期/QC 不合格），非批次或缺口落
    batch_id=0；按成本方法计价（batch_actual 逐批、其余池化）；回写 stock_qty 投影。"""
    from app.plugins.inventory.allocation import allocate_fefo
    method = await _cost_method(db, tenant_id)
    # 1) 决定出库腿 (batch_id, qty)
    legs: list[tuple[int | None, Decimal]] = []
    remaining = qty
    stocks = await _load_batch_stocks(db, tenant_id, pid, vid, warehouse_id=wh)
    allocs, gap = allocate_fefo(stocks, qty) if qty > 0 else ([], Decimal("0"))
    legs = [(a.batch_id, a.qty) for a in allocs]
    remaining = gap
    if remaining > 0:
        legs.append((None, remaining))
    # 2) 计成本并写腿
    if method == "batch_actual":
        for batch_id, lqty in legs:
            uc = await _consume_batch_layer(db, tenant_id, pid, vid, batch_id or 0, lqty)
            await _sellable_leg(db, tenant_id, pid, vid, wh, -lqty, batch_id, doc_type, doc_id, uc, suffix)
    else:
        total_cost = await _consume_cost_pool(db, tenant_id, pid, vid, qty, method)
        uc = (total_cost / qty).quantize(Decimal("0.0001")) if qty > 0 else None
        for batch_id, lqty in legs:
            await _sellable_leg(db, tenant_id, pid, vid, wh, -lqty, batch_id, doc_type, doc_id, uc, suffix)
    # 3) 投影
    await db.execute(
        text("UPDATE products SET stock_qty = stock_qty - :q WHERE id=:pid AND tenant_id=:tid"),
        {"q": qty, "pid": pid, "tid": tenant_id},
    )
    if vid:
        await db.execute(
            text("UPDATE product_variants SET stock_qty = stock_qty - :q WHERE id=:vid AND tenant_id=:tid"),
            {"q": qty, "vid": vid, "tid": tenant_id},
        )


async def ship_out_via_ledger(db: AsyncSession, tenant_id: int, items, order_id: int,
                              suffix: str = "") -> None:
    """发货出库：reserved 减少 + sellable 实扣（批次 FEFO + 成本层消耗）。"""
    if not await _claim_operation(db, tenant_id, "order_ship", order_id, suffix):
        return
    wh = await _default_warehouse_id(db, tenant_id)
    explicitly_shipped = (await db.execute(
        text("SELECT product_id,variant_id,SUM(shipped_qty) FROM inventory_allocations "
             "WHERE tenant_id=:tid AND order_id=:order GROUP BY product_id,variant_id"),
        {"tid": tenant_id, "order": order_id},
    )).fetchall()
    shipped_by_key = {(int(row[0]), int(row[1] or 0)): Decimal(str(row[2] or 0))
                      for row in explicitly_shipped}
    wants = _keys(items)
    if explicitly_shipped:
        incomplete = [
            {"product_id": pid, "variant_id": vid or None,
             "ordered_qty": str(qty),
             "shipped_qty": str(shipped_by_key.get((int(pid), int(vid or 0)), Decimal("0")))}
            for pid, vid, qty in wants
            if shipped_by_key.get((int(pid), int(vid or 0)), Decimal("0")) != qty
        ]
        if incomplete:
            raise HTTPException(
                status_code=409,
                detail={"message": "订单已进入分配履约流程，请先完成拣货和发货", "lines": incomplete},
            )
        # Explicit fulfillment already wrote the exact warehouse/location/lot/serial legs.
        return
    for pid, vid, qty in wants:
        await _reserved_leg(db, tenant_id, pid, vid, wh, -qty, "order_ship_res", order_id, suffix)
        await _sellable_out(db, tenant_id, pid, vid, wh, qty, "order_ship", order_id, suffix)


async def _lock_balance_rows(db: AsyncSession, tenant_id: int, keys: list[Key], warehouse_id: int) -> None:
    """锁定已存在余额行；调用者随后才可读/改可售与预留数量。"""
    if not keys:
        return
    pids = sorted({k[0] for k in keys})
    await db.execute(
        text("SELECT id FROM inventory_balances WHERE tenant_id=:tid AND warehouse_id=:wh "
             f"AND product_id IN ({','.join(str(int(p)) for p in pids)}) FOR UPDATE"),
        {"tid": tenant_id, "wh": warehouse_id},
    )


async def _load_available(db: AsyncSession, tenant_id: int, keys: list[Key],
                          warehouse_id: int | None = None) -> dict[Key, Decimal]:
    """可下单量 = 可分配 sellable - reserved。

    可分配 sellable 排除**已过期**与 **QC 非 passed** 的批次余额（这些出库时会被
    FEFO 跳过），否则订单能锁库、发货却无货可出、缺口落 batch_id=0 负库存绕过限制。
    非批次余额(batch_id=0)与无批次租户不受影响。
    """
    if not keys:
        return {}
    pids = sorted({k[0] for k in keys})
    where_warehouse = " AND bal.warehouse_id=:wh" if warehouse_id is not None else ""
    params = {"tid": tenant_id}
    if warehouse_id is not None:
        params["wh"] = warehouse_id
    r = await db.execute(
        text("SELECT bal.product_id, bal.variant_id, "
             "  SUM(CASE WHEN bal.stock_state='sellable' AND ("
             "      bal.batch_id=0 OR (b.qc_state='passed' AND (b.expires_on IS NULL OR b.expires_on>=CURDATE()))"
             "    ) THEN bal.qty ELSE 0 END) "
             "  - SUM(CASE WHEN bal.stock_state='reserved' THEN bal.qty ELSE 0 END) "
             "FROM inventory_balances bal LEFT JOIN inventory_batches b ON b.id = bal.batch_id "
             "WHERE bal.tenant_id=:tid "
             f"AND bal.product_id IN ({','.join(str(int(p)) for p in pids)}) {where_warehouse} "
             "GROUP BY bal.product_id, bal.variant_id"),
        params,
    )
    return {(row[0], row[1]): Decimal(str(row[2] or 0)) for row in r.fetchall()}


async def validate_via_ledger(db: AsyncSession, tenant_id: int, items,
                              *, oversell_product_ids: set[int] | None = None) -> None:
    wants = [row for row in _keys(items) if row[0] not in (oversell_product_ids or set())]
    wh = await _default_warehouse_id(db, tenant_id)
    avail = await _load_available(db, tenant_id, [(p, v) for p, v, _ in wants], wh)
    check_availability(avail, wants, allow_negative=await _allow_negative(db, tenant_id))


# 订单状态 → 库存阶段：out=已实扣，reserved=已锁未扣
_OUT_STATUSES = frozenset({"shipped", "completed"})
_RESERVED_STATUSES = frozenset({"pending", "paid"})


async def load_policy(db: AsyncSession, tenant_id: int) -> InventoryPolicy:
    row = (await db.execute(
        text("SELECT sales_policy FROM inventory_settings WHERE tenant_id=:tid LIMIT 1"),
        {"tid": tenant_id},
    )).fetchone()
    return InventoryPolicy.from_json(row[0] if row else None)


async def apply_transition_via_ledger(db: AsyncSession, tenant_id: int, items,
                                      old_status: str | None, new_status: str, order_id: int,
                                      idem_suffix: str = "") -> None:
    """两段式状态机，按订单状态阶段(reserved=已锁未扣 / out=已实扣)驱动库存。

    - 进入 reserved 且此前未锁(None/未接管前)：首次锁库（含改单重扣、状态推进入锁定）
    - 进入 out(shipped/completed)：reserved → 实扣出库
    - out → reserved(撤销发货)：回补 sellable 并重新锁定
    - 进入 cancelled 且此前在 reserved 阶段：释放锁定
    - out → cancelled：不自动回补；物理退货由退款/退货流程(restock=True)负责，
      避免与退款重复回补

    幂等键后缀：调用方给了 idem_suffix(如改单版本)就用它；否则用状态事件
    (old>new)，使同一状态变更的重试去重、不同事件不误撞。
    """
    suffix = idem_suffix or f"{old_status}>{new_status}"
    action = (await load_policy(db, tenant_id)).action(old_status, new_status)
    if action == "reserve":
        await lock_via_ledger(db, tenant_id, items, order_id, suffix)
    elif action == "ship":
        await ship_out_via_ledger(db, tenant_id, items, order_id, suffix)
    elif action == "restore_and_reserve":
        # 撤销发货：sellable 回补并重新锁定
        await restore_via_ledger(db, tenant_id, items, order_id, suffix)
        await lock_via_ledger(db, tenant_id, items, order_id, suffix)
    elif action == "release":
        await release_via_ledger(db, tenant_id, items, order_id, suffix)


async def deduct_via_ledger(db: AsyncSession, tenant_id: int, items, order_id: int,
                            suffix: str = "") -> None:
    if not await _claim_operation(db, tenant_id, "order", order_id, suffix):
        return
    await _apply(db, tenant_id, _keys(items), "order", order_id, sign=-1, suffix=suffix)


async def _batch_enabled(db: AsyncSession, tenant_id: int) -> bool:
    r = await db.execute(
        text("SELECT batch_enabled FROM inventory_settings WHERE tenant_id=:tid LIMIT 1"),
        {"tid": tenant_id},
    )
    row = r.fetchone()
    return bool(row[0]) if row else False


async def _serial_enabled(db: AsyncSession, tenant_id: int) -> bool:
    r = await db.execute(
        text("SELECT serial_enabled FROM inventory_settings WHERE tenant_id=:tid LIMIT 1"),
        {"tid": tenant_id},
    )
    row = r.fetchone()
    return bool(row[0]) if row else False


async def _load_batch_stocks(db: AsyncSession, tenant_id: int, product_id: int, variant_id: int,
                             warehouse_id: int | None = None):
    """可分配批次余额：排除已过期批次与 QC 非可售（passed 以外）批次。"""
    from app.plugins.inventory.allocation import BatchStock
    r = await db.execute(
        text("SELECT bal.batch_id, bal.qty, b.expires_on, b.id "
             "FROM inventory_balances bal JOIN inventory_batches b ON b.id = bal.batch_id "
             "WHERE bal.tenant_id=:tid AND bal.product_id=:pid AND bal.variant_id=:vid AND bal.warehouse_id=:wh "
             "AND bal.stock_state='sellable' AND bal.qty > 0 AND bal.batch_id <> 0 "
             "AND b.qc_state = 'passed' "
             "AND (b.expires_on IS NULL OR b.expires_on >= CURDATE()) FOR UPDATE"),
        {"tid": tenant_id, "pid": product_id, "vid": variant_id,
         "wh": warehouse_id or await _default_warehouse_id(db, tenant_id)},
    )
    return [BatchStock(batch_id=x[0], qty=Decimal(str(x[1])), expires_on=x[2], created_id=x[3])
            for x in r.fetchall()]


async def deduct_offline_via_ledger(db: AsyncSession, tenant_id: int, items, order_id: int) -> None:
    """POS 离线单入账：永远允许负库存（钱已收，不能拒绝入账）。

    批次租户按 FEFO 分配；FEFO 有缺口时缺口部分写入「未分配负库存」异常桶
    （unallocated=1、batch 留空），绝不虚构批次。
    """
    from app.plugins.inventory.allocation import allocate_fefo
    # 离线单没有逐件序列号采集，启用序列号管理时宁可进入待同步异常，不能伪造序列号出库。
    if await _serial_enabled(db, tenant_id):
        raise HTTPException(status_code=409, detail="已启用序列号管理，POS 离线单不可自动扣减库存")
    wh = await _default_warehouse_id(db, tenant_id)
    batch_on = await _batch_enabled(db, tenant_id)
    for pid, vid, qty in _keys(items):
        product_delta = -qty
        if batch_on:
            stocks = await _load_batch_stocks(db, tenant_id, pid, vid)
            allocs, gap = allocate_fefo(stocks, qty) if qty > 0 else ([], Decimal("0"))
            for a in allocs:
                await _write_offline_leg(db, tenant_id, pid, vid, wh, -a.qty, order_id,
                                         batch_id=a.batch_id, unallocated=False)
            if gap > 0:
                await _write_offline_leg(db, tenant_id, pid, vid, wh, -gap, order_id,
                                         batch_id=None, unallocated=True)
        else:
            await _write_offline_leg(db, tenant_id, pid, vid, wh, product_delta, order_id,
                                     batch_id=None, unallocated=False)
        # 投影：父商品与规格按净变化回写
        after_v = None
        if vid:
            r = await db.execute(
                text("SELECT COALESCE(SUM(qty),0) FROM inventory_balances "
                     "WHERE tenant_id=:tid AND product_id=:pid AND variant_id=:vid AND stock_state='sellable'"),
                {"tid": tenant_id, "pid": pid, "vid": vid},
            )
            after_v = Decimal(str(r.scalar() or 0))
            await db.execute(
                text("UPDATE product_variants SET stock_qty=:q WHERE id=:vid AND tenant_id=:tid"),
                {"q": after_v, "vid": vid, "tid": tenant_id},
            )
        await db.execute(
            text("UPDATE products SET stock_qty = stock_qty + :d WHERE id=:pid AND tenant_id=:tid"),
            {"d": product_delta, "pid": pid, "tid": tenant_id},
        )


async def _write_offline_leg(db, tenant_id, pid, vid, wh, delta, order_id,
                             *, batch_id, unallocated) -> None:
    bkey = batch_id or 0
    # 幂等键需带 batch 区分同单多批次腿；POS 离线重放同键跳过（优雅幂等）
    key = f"pos_offline:{order_id}:{pid}:{vid or 0}:{bkey}:{1 if unallocated else 0}"[:120]
    if await _txn_exists(db, tenant_id, key):
        return
    before = (await db.execute(
        text("SELECT COALESCE(SUM(qty),0) FROM inventory_balances WHERE tenant_id=:tid "
             "AND product_id=:pid AND variant_id=:vid AND warehouse_id=:wh AND batch_id=:bk "
             "AND stock_state='sellable' FOR UPDATE"),
        {"tid": tenant_id, "pid": pid, "vid": vid, "wh": wh, "bk": bkey},
    )).scalar()
    before = Decimal(str(before or 0))
    req = TxnRequest(tenant_id=tenant_id, product_id=pid, variant_id=vid or None,
                     warehouse_id=wh, qty_delta=delta, doc_type="pos_offline", doc_id=order_id,
                     batch_id=batch_id, unallocated=unallocated,
                     reason="POS离线-未分配" if unallocated else None)
    t = build_txn(req, qty_before=before)
    t.idempotency_key = key
    db.add(t)
    await db.execute(
        text("INSERT INTO inventory_balances "
             "(tenant_id, product_id, variant_id, warehouse_id, location_id, batch_id, "
             " stock_state, qty, created_at, updated_at) "
             "VALUES (:tid,:pid,:vid,:wh,0,:bk,'sellable',:q,NOW(3),NOW(3)) "
             "ON DUPLICATE KEY UPDATE qty = qty + :q, updated_at = NOW(3)"),
        {"tid": tenant_id, "pid": pid, "vid": vid, "wh": wh, "bk": bkey, "q": delta},
    )


async def restore_via_ledger(db: AsyncSession, tenant_id: int, items, order_id: int,
                             suffix: str = "") -> None:
    allocated = (await db.execute(
        text("SELECT 1 FROM inventory_allocations WHERE tenant_id=:tid AND order_id=:order LIMIT 1"),
        {"tid": tenant_id, "order": order_id},
    )).scalar()
    if allocated:
        from app.plugins.inventory.returns import restore_allocated_by_sku
        await restore_allocated_by_sku(db, tenant_id, order_id, items, suffix=suffix)
        return
    if not await _claim_operation(db, tenant_id, "order_restore", order_id, suffix):
        return
    await _apply(db, tenant_id, _keys(items), "order_restore", order_id, sign=1, suffix=suffix)


# ── Admin 启用与管理 ────────────────────────────────────────────

def build_opening_txns(rows: list[tuple[int, int, Decimal]]
                       ) -> list[tuple[int, int, Decimal]]:
    """把 (product_id, variant_id, stock_qty) 过滤成需要写期初流水的非零行。

    stock_qty 为 0 的商品不建期初流水（余额表缺省即 0），避免无谓行。
    """
    return [(pid, vid, qty) for pid, vid, qty in rows if qty and qty != 0]


def encode_cursor(txn_id: int) -> str:
    """流水分页游标：按自增主键降序，游标即上一页最后一条 id。"""
    return str(txn_id)


def decode_cursor(cursor: str | None) -> int | None:
    if not cursor:
        return None
    try:
        return int(cursor)
    except (TypeError, ValueError):
        return None


async def enable_inventory(db: AsyncSession, tenant_id: int, *, warehouse_code: str,
                           warehouse_name: str, opening_mode: str, base_currency: str) -> int:
    """启用进销存：建默认仓库 + 默认库位 + 设置行；from_stock 时把现有库存写期初流水。

    返回默认仓库 id。重复启用（已存在设置行）抛 409。
    """
    exists = (await db.execute(
        text("SELECT id FROM inventory_settings WHERE tenant_id=:tid LIMIT 1"),
        {"tid": tenant_id},
    )).fetchone()
    if exists:
        raise HTTPException(status_code=409, detail="进销存已启用，请勿重复启用")

    await db.execute(
        text("INSERT INTO inventory_warehouses (tenant_id, code, name, is_active, created_at, updated_at) "
             "VALUES (:tid,:code,:name,1,NOW(3),NOW(3))"),
        {"tid": tenant_id, "code": warehouse_code, "name": warehouse_name},
    )
    wh_id = (await db.execute(text("SELECT LAST_INSERT_ID()"))).scalar()
    await db.execute(
        text("INSERT INTO inventory_locations "
             "(tenant_id, warehouse_id, parent_id, code, name, is_default, created_at, updated_at) "
             "VALUES (:tid,:wh,NULL,'DEFAULT','默认库位',1,NOW(3),NOW(3))"),
        {"tid": tenant_id, "wh": wh_id},
    )
    # 显式列出所有标志列（默认全关 = 基础零售预设），不依赖 DB 端默认值
    await db.execute(
        text("INSERT INTO inventory_settings "
             "(tenant_id, lifecycle_state, warehouse_mode, cost_method, "
             " batch_enabled, expiry_enabled, serial_enabled, qc_enabled, allow_negative, "
             " purchase_approval, period_close_enabled, base_currency, "
             " default_warehouse_id, created_at, updated_at) "
             "VALUES (:tid,'active','simple','moving_avg',"
             " 0,0,0,0,0,0,0,:cur,:wh,NOW(3),NOW(3))"),
        {"tid": tenant_id, "cur": base_currency, "wh": wh_id},
    )

    if opening_mode == "from_stock":
        await import_opening_from_stock(db, tenant_id, wh_id)
    return int(wh_id)


async def import_opening_from_stock(db: AsyncSession, tenant_id: int, wh_id: int) -> None:
    """把现有 products / product_variants 的 stock_qty 写为期初流水与余额。

    整批在调用方事务内；有规格商品以 variant 粒度建账，无规格以 product 粒度。
    """
    variants = (await db.execute(
        text("SELECT v.id, v.product_id, v.stock_qty, p.cost_price FROM product_variants v "
             "JOIN products p ON p.id = v.product_id WHERE v.tenant_id=:tid"),
        {"tid": tenant_id},
    )).fetchall()
    variant_pids = {v[1] for v in variants}
    # rows: (pid, vid, qty, cost_price)
    rows: list[tuple[int, int, Decimal, Decimal]] = [
        (v[1], v[0], Decimal(str(v[2] or 0)), Decimal(str(v[3] or 0))) for v in variants
    ]
    products = (await db.execute(
        text("SELECT id, stock_qty, cost_price FROM products WHERE tenant_id=:tid"),
        {"tid": tenant_id},
    )).fetchall()
    # 无规格商品才按 product 粒度建期初；有规格的库存已按 variant 计
    rows += [(p[0], 0, Decimal(str(p[1] or 0)), Decimal(str(p[2] or 0)))
             for p in products if p[0] not in variant_pids]

    for pid, vid, qty, cost_price in rows:
        if not qty or qty == 0:
            continue
        db.add(build_txn(TxnRequest(tenant_id=tenant_id, product_id=pid,
                                    variant_id=vid or None, warehouse_id=wh_id,
                                    qty_delta=qty, doc_type="opening", doc_id=None,
                                    unit_cost=cost_price or None, reason="期初导入"),
                         qty_before=Decimal("0")))
        await db.execute(
            text("INSERT INTO inventory_balances "
                 "(tenant_id, product_id, variant_id, warehouse_id, location_id, batch_id, "
                 " stock_state, qty, created_at, updated_at) "
                 "VALUES (:tid,:pid,:vid,:wh,0,0,'sellable',:q,NOW(3),NOW(3)) "
                 "ON DUPLICATE KEY UPDATE qty = qty + :q, updated_at = NOW(3)"),
            {"tid": tenant_id, "pid": pid, "vid": vid, "wh": wh_id, "q": qty},
        )
        # 期初成本层：按商品成本价建账，供后续出库 COGS 计价
        await db.execute(
            text("INSERT INTO inventory_cost_layers "
                 "(tenant_id, product_id, variant_id, batch_id, qty_in, qty_consumed, unit_cost, created_at, updated_at) "
                 "VALUES (:tid,:pid,:vid,0,:q,0,:uc,NOW(3),NOW(3))"),
            {"tid": tenant_id, "pid": pid, "vid": vid, "q": qty, "uc": cost_price or Decimal("0")},
        )


async def apply_adjustment(db: AsyncSession, tenant_id: int, lines, operator_id: int | None) -> None:
    """库存纠正单：直接按 qty_delta 增减，写流水 + 余额 + 投影。suspended 由路由层拦截。"""
    await assert_period_open(db, tenant_id)
    default_wh = await _default_warehouse_id(db, tenant_id)
    allow_negative = await _allow_negative(db, tenant_id)
    batch_on = await _batch_enabled(db, tenant_id)
    for ln in lines:
        delta = Decimal(str(ln.qty_delta))
        if delta == 0:
            raise HTTPException(status_code=400, detail="纠正数量不能为 0")
        wh = int(ln.warehouse_id or default_wh)
        warehouse = await db.execute(
            text("SELECT id FROM inventory_warehouses WHERE id=:id AND tenant_id=:tid AND is_active=1"),
            {"id": wh, "tid": tenant_id},
        )
        if not warehouse.fetchone():
            raise HTTPException(status_code=400, detail="纠正仓库不存在、已停用或不属于当前租户")
        if batch_on and not ln.batch_id:
            raise HTTPException(status_code=400, detail="已启用批次管理，库存纠正必须选择具体批次")
        if not batch_on and ln.batch_id:
            raise HTTPException(status_code=400, detail="未启用批次管理，库存纠正不能指定批次")
        batch_id = int(ln.batch_id or 0)
        vid = ln.variant_id or 0
        product = await db.execute(
            text("SELECT id FROM products WHERE id=:pid AND tenant_id=:tid"),
            {"pid": ln.product_id, "tid": tenant_id},
        )
        if not product.fetchone():
            raise HTTPException(status_code=400, detail=f"商品 {ln.product_id} 不属于当前租户")
        before = (await db.execute(
            text("SELECT COALESCE(SUM(qty),0) FROM inventory_balances "
                 "WHERE tenant_id=:tid AND product_id=:pid AND variant_id=:vid "
                 "AND warehouse_id=:wh AND batch_id=:bid AND stock_state='sellable' FOR UPDATE"),
            {"tid": tenant_id, "pid": ln.product_id, "vid": vid, "wh": wh, "bid": batch_id},
        )).scalar()
        before = Decimal(str(before or 0))
        if delta < 0 and not allow_negative and before + delta < 0:
            raise HTTPException(status_code=400, detail=f"商品 {ln.product_id} 库存不足，禁止调整为负库存")
        db.add(build_txn(TxnRequest(tenant_id=tenant_id, product_id=ln.product_id,
                                    variant_id=ln.variant_id, warehouse_id=wh,
                                    qty_delta=delta, doc_type="adjustment", doc_id=None,
                                    batch_id=batch_id or None, operator_id=operator_id, reason=ln.reason),
                         qty_before=before))
        await db.execute(
            text("INSERT INTO inventory_balances "
                 "(tenant_id, product_id, variant_id, warehouse_id, location_id, batch_id, "
                 " stock_state, qty, created_at, updated_at) "
                 "VALUES (:tid,:pid,:vid,:wh,0,:bid,'sellable',:q,NOW(3),NOW(3)) "
                "ON DUPLICATE KEY UPDATE qty = qty + :q, updated_at = NOW(3)"),
            {"tid": tenant_id, "pid": ln.product_id, "vid": vid, "wh": wh, "bid": batch_id, "q": delta},
        )
        after = before + delta
        p = project(sellable_on_hand=after, reserved=Decimal("0"))
        if vid:
            await db.execute(
                text("UPDATE product_variants SET stock_qty=:q WHERE id=:vid AND tenant_id=:tid"),
                {"q": p.stock_qty, "vid": vid, "tid": tenant_id},
            )
        await db.execute(
            text("UPDATE products SET stock_qty = stock_qty + :d WHERE id=:pid AND tenant_id=:tid"),
            {"d": delta, "pid": ln.product_id, "tid": tenant_id},
        )