import { Injectable, Logger, OnModuleInit } from '@nestjs/common'; import { and, eq, sql, isNull, inArray } from 'drizzle-orm'; import Decimal from 'decimal.js'; import { ChargeUnit, CreateManualChargeDto, TariffConditions } from '@wms/shared'; import { DbService, Tx } from '@/common/db/db.service'; import { chargeEvents, asns, asnLines, receiptLines, inventoryMovements, outboundOrders, outboundOrderLines, shipments, skuPacks, warehouses } from '@/db/schema'; import { EventBus, DomainEvent } from '@/common/events/event-bus.service'; import { AuditService } from '@/common/audit/audit.service'; import { requireRequestContext, runWithContext, systemContext } from '@/common/context/request-context'; import { DomainError, NotFoundError, ErrorCodes } from '@/common/errors/domain-error'; import { D, ZERO, monthKey, weekKey, serialize, todayIso } from '@/common/util'; import { ContractsService, TariffRow } from './contracts.service'; /** * Captures billable events at source. * - asn.closed -> HANDLING_IN tariffs with trigger ASN_CLOSED * - order.shipped / short_closed -> HANDLING_OUT tariffs with trigger ORDER_SHIPPED * - manual VAS charges from the UI * Each charge is idempotent on (tariff, ref) so replays never double bill. */ @Injectable() export class ChargeCaptureService implements OnModuleInit { private readonly logger = new Logger(ChargeCaptureService.name); constructor( private readonly db: DbService, private readonly events: EventBus, private readonly contracts: ContractsService, private readonly audit: AuditService, ) {} onModuleInit() { this.events.subscribe('asn.closed', (e) => this.onAsnClosed(e)); this.events.subscribe('order.shipped', (e) => this.onOrderShipped(e)); this.events.subscribe('order.short_closed', (e) => this.onOrderShipped(e)); } // ---------------- event handlers (run by the outbox dispatcher) ---------------- private async onAsnClosed(e: DomainEvent) { await runWithContext(systemContext(e.tenantId), () => this.db.run(async (tx) => { const asnId = e.payload.asnId as string; const [asn] = await tx.select().from(asns).where(eq(asns.id, asnId)).limit(1); if (!asn) return; const date = todayIso(await this.warehouseTz(tx, asn.warehouseId), asn.closedAt ?? new Date()); const contract = await this.contracts.activeContract(tx, asn.storerId, date); if (!contract) { this.logger.warn(`No active contract for storer ${asn.storerId}; ASN ${asn.asnNumber} not billed`); return; } const tariffsRows = (await this.contracts.activeTariffs(tx, contract.id, date)).filter((t) => t.trigger === 'ASN_CLOSED'); if (!tariffsRows.length) return; const metrics = await this.asnMetrics(tx, asnId); for (const t of tariffsRows) { if (!this.conditionsMatch(t.conditions as TariffConditions, { containerSize: asn.containerSize ?? undefined, asnType: asn.type, warehouseId: asn.warehouseId })) continue; const qty = this.quantityFor(t.chargeUnit, metrics); if (qty.lte(0)) continue; await this.capture(tx, { contract, tariff: t, warehouseId: asn.warehouseId, storerId: asn.storerId, refType: 'ASN', refId: asn.id, refNumber: asn.asnNumber, quantity: qty, chargeDate: date, detail: { externalRef: asn.externalRef, containerSize: asn.containerSize, ...metrics.asNumbers }, idempotencyKey: `${t.id}:ASN:${asn.id}` }); } }), ); } private async onOrderShipped(e: DomainEvent) { await runWithContext(systemContext(e.tenantId), () => this.db.run(async (tx) => { const orderId = e.payload.orderId as string; const [o] = await tx.select().from(outboundOrders).where(eq(outboundOrders.id, orderId)).limit(1); if (!o) return; const date = todayIso(await this.warehouseTz(tx, o.warehouseId), o.shippedAt ?? new Date()); const contract = await this.contracts.activeContract(tx, o.storerId, date); if (!contract) { this.logger.warn(`No active contract for storer ${o.storerId}; order ${o.orderNumber} not billed`); return; } const tariffsRows = (await this.contracts.activeTariffs(tx, contract.id, date)).filter((t) => t.trigger === 'ORDER_SHIPPED'); if (!tariffsRows.length) return; const metrics = await this.orderMetrics(tx, orderId); for (const t of tariffsRows) { if (!this.conditionsMatch(t.conditions as TariffConditions, { orderType: o.type, warehouseId: o.warehouseId, containerSize: (o.attributes as { containerSize?: string }).containerSize })) continue; const qty = this.quantityFor(t.chargeUnit, metrics); if (qty.lte(0)) continue; await this.capture(tx, { contract, tariff: t, warehouseId: o.warehouseId, storerId: o.storerId, refType: 'ORDER', refId: o.id, refNumber: o.orderNumber, quantity: qty, chargeDate: date, detail: { externalRef: o.externalRef, ...metrics.asNumbers }, idempotencyKey: `${t.id}:ORDER:${o.id}` }); } }), ); } // ---------------- manual ---------------- manual(dto: CreateManualChargeDto) { const ctx = requireRequestContext(); return this.db.run(async (tx) => { const contract = await this.contracts.activeContract(tx, dto.storerId, dto.chargeDate); if (!contract) throw new DomainError(ErrorCodes.CONTRACT_MISSING, 'No active billing contract for this storer on that date'); const [t] = await this.contracts.activeTariffs(tx, contract.id, dto.chargeDate).then((rows) => rows.filter((r) => r.id === dto.tariffId)); if (!t) throw new NotFoundError('Tariff', dto.tariffId); const row = await this.capture(tx, { contract, tariff: t, warehouseId: dto.warehouseId, storerId: dto.storerId, refType: dto.refType, refId: dto.refId ?? null, refNumber: dto.refNumber ?? null, quantity: D(dto.quantity), chargeDate: dto.chargeDate, detail: dto.detail, rateOverride: dto.rate ? D(dto.rate) : undefined, description: dto.description, source: 'MANUAL', createdById: ctx.userId }); await this.audit.log({ action: 'charge.manual', entityType: 'charge_event', entityId: row.id, after: dto }, tx); return serialize(row); }); } waive(id: string, reason: string) { return this.db.run(async (tx) => { const [c] = await tx.select().from(chargeEvents).where(eq(chargeEvents.id, id)).limit(1).for('update'); if (!c) throw new NotFoundError('Charge', id); if (c.status !== 'PENDING') throw new DomainError(ErrorCodes.INVALID_STATE, `Charge is ${c.status}`); const [row] = await tx.update(chargeEvents).set({ status: 'WAIVED', waivedReason: reason }).where(eq(chargeEvents.id, id)).returning(); await this.audit.log({ action: 'charge.waive', entityType: 'charge_event', entityId: id, after: { reason } }, tx); return serialize(row); }); } // ---------------- core ---------------- async capture( tx: Tx, input: { contract: { id: string; currency: string; weekStartsOn: number; billingCycle: string }; tariff: TariffRow; warehouseId: string; storerId: string; refType: string | null; refId: string | null; refNumber: string | null; quantity: Decimal; chargeDate: string; detail?: Record; idempotencyKey?: string; rateOverride?: Decimal; description?: string; source?: 'SYSTEM' | 'MANUAL' | 'AI'; createdById?: string | null }, ) { const ctx = requireRequestContext(); const rate = input.rateOverride ?? this.rateFor(input.tariff, input.quantity); let amount = input.quantity.mul(rate); let minimumApplied = false; if (input.tariff.minimumCharge && amount.lt(input.tariff.minimumCharge)) { amount = D(input.tariff.minimumCharge); minimumApplied = true; } const periodKey = input.contract.billingCycle === 'WEEKLY' ? weekKey(input.chargeDate, input.contract.weekStartsOn).key : monthKey(input.chargeDate); const [row] = await tx .insert(chargeEvents) .values({ tenantId: ctx.tenantId!, warehouseId: input.warehouseId, storerId: input.storerId, contractId: input.contract.id, tariffId: input.tariff.id, refType: input.refType, refId: input.refId, refNumber: input.refNumber, description: input.description ?? input.tariff.description, quantity: input.quantity, unit: input.tariff.chargeUnit, rate, amount: amount.toDecimalPlaces(4), currency: input.contract.currency, chargeDate: input.chargeDate, periodKey, source: input.source ?? 'SYSTEM', detail: { ...(input.detail ?? {}), ...(minimumApplied ? { minimumApplied: true, listRate: rate.toString() } : {}) }, idempotencyKey: input.idempotencyKey ?? null, createdById: input.createdById ?? null, }) .onConflictDoNothing({ target: [chargeEvents.tenantId, chargeEvents.idempotencyKey] }) .returning(); if (row) await this.events.publish('charge.captured', { chargeId: row.id, tariffCode: input.tariff.code, amount: row.amount.toString(), refType: input.refType, refNumber: input.refNumber }, tx, { warehouseId: input.warehouseId, storerId: input.storerId }); return row ?? (await tx.select().from(chargeEvents).where(and(eq(chargeEvents.idempotencyKey, input.idempotencyKey!))).limit(1))[0]; } private async warehouseTz(tx: Tx, warehouseId: string): Promise { const [wh] = await tx.select({ timezone: warehouses.timezone }).from(warehouses).where(eq(warehouses.id, warehouseId)).limit(1); return wh?.timezone ?? 'Asia/Kuwait'; } /** Tiered rates: the tier whose upTo covers the quantity (volume-based rate card). */ rateFor(t: TariffRow, qty: Decimal): Decimal { if (t.tiers?.length) { const sorted = [...t.tiers].sort((a, b) => (a.upTo ?? Infinity) - (b.upTo ?? Infinity)); const tier = sorted.find((x) => x.upTo === null || qty.lte(x.upTo)) ?? sorted[sorted.length - 1]; return D(tier.rate); } return D(t.rate); } conditionsMatch(c: TariffConditions, facts: { containerSize?: string; asnType?: string; orderType?: string; warehouseId?: string; taskType?: string }): boolean { if (c.containerSize && c.containerSize !== facts.containerSize) return false; if (c.asnType && c.asnType !== facts.asnType) return false; if (c.orderType && c.orderType !== facts.orderType) return false; if (c.warehouseId && c.warehouseId !== facts.warehouseId) return false; if (c.taskType && c.taskType !== facts.taskType) return false; return true; } private quantityFor(unit: ChargeUnit, m: Metrics): Decimal { switch (unit) { case 'PER_CBM': return m.cbm; case 'PER_KG': return m.weightKg; case 'PER_PALLET': return m.pallets; case 'PER_CASE': return m.cases; case 'PER_PIECE': return m.pieces; case 'PER_LINE': return m.lines; case 'PER_ORDER': case 'PER_SHIPMENT': case 'PER_DOCUMENT': return D(1); case 'PER_CONTAINER_20': return m.containerSize === '20' ? D(1) : ZERO; case 'PER_CONTAINER_40': return ['40', '40HC', '45'].includes(m.containerSize ?? '') ? D(1) : ZERO; case 'PER_TRUCK': return ['TRUCK', 'VAN'].includes(m.containerSize ?? '') ? D(1) : ZERO; case 'FLAT': return D(1); default: return ZERO; // storage units are computed by the storage billing run, never per event } } async asnMetrics(tx: Tx, asnId: string): Promise { const [asn] = await tx.select().from(asns).where(eq(asns.id, asnId)).limit(1); const movs = await tx.select({ cbm: sql`coalesce(sum(${inventoryMovements.cbm}),0)`, kg: sql`coalesce(sum(${inventoryMovements.weightKg}),0)`, qty: sql`coalesce(sum(${inventoryMovements.qty}),0)` }).from(inventoryMovements).where(and(eq(inventoryMovements.refType, 'ASN'), eq(inventoryMovements.refId, asnId), inArray(inventoryMovements.type, ['RECEIPT', 'RETURN_RECEIPT']))); const rev = await tx.select({ qty: sql`coalesce(sum(${inventoryMovements.qty}),0)`, cbm: sql`coalesce(sum(${inventoryMovements.cbm}),0)` }).from(inventoryMovements).where(and(eq(inventoryMovements.refType, 'ASN'), eq(inventoryMovements.refId, asnId), eq(inventoryMovements.type, 'RECEIPT_REVERSAL'))); const [{ pallets }] = await tx.select({ pallets: sql`count(distinct ${receiptLines.lpnId})`.mapWith(Number) }).from(receiptLines).where(and(eq(receiptLines.asnId, asnId), isNull(receiptLines.reversedAt))); const [{ lines }] = await tx.select({ lines: sql`count(*)`.mapWith(Number) }).from(asnLines).where(eq(asnLines.asnId, asnId)); const cases = await this.caseEquivalent(tx, (await tx.select({ skuId: receiptLines.skuId, qty: sql`sum(${receiptLines.qty})` }).from(receiptLines).where(and(eq(receiptLines.asnId, asnId), isNull(receiptLines.reversedAt))).groupBy(receiptLines.skuId))); const pieces = D(movs[0].qty).minus(rev[0].qty); const cbm = D(movs[0].cbm).minus(rev[0].cbm); return { cbm, weightKg: D(movs[0].kg), pallets: D(pallets), cases, pieces, lines: D(lines), containerSize: asn?.containerSize ?? undefined, asNumbers: { cbm: cbm.toString(), pallets, pieces: pieces.toString(), cases: cases.toString(), lines } }; } async orderMetrics(tx: Tx, orderId: string): Promise { const [o] = await tx.select().from(outboundOrders).where(eq(outboundOrders.id, orderId)).limit(1); const ship = await tx.select({ cartons: sql`coalesce(sum(${shipments.cartonCount}),0)`.mapWith(Number), pallets: sql`coalesce(sum(${shipments.palletCount}),0)`.mapWith(Number), cbm: sql`coalesce(sum(${shipments.totalCbm}),0)`, kg: sql`coalesce(sum(${shipments.totalWeightKg}),0)` }).from(shipments).where(and(eq(shipments.orderId, orderId), eq(shipments.status, 'SHIPPED'))); const lineRows = await tx.select({ skuId: outboundOrderLines.skuId, qty: outboundOrderLines.shippedQty }).from(outboundOrderLines).where(and(eq(outboundOrderLines.orderId, orderId), sql`${outboundOrderLines.shippedQty} > 0`)); const cases = await this.caseEquivalent(tx, lineRows.map((l) => ({ skuId: l.skuId, qty: l.qty.toString() }))); const pieces = lineRows.reduce((a, l) => a.plus(l.qty), ZERO); const movCbm = await tx.select({ cbm: sql`coalesce(sum(${inventoryMovements.cbm}),0)`, kg: sql`coalesce(sum(${inventoryMovements.weightKg}),0)` }).from(inventoryMovements).where(and(eq(inventoryMovements.refType, 'ORDER'), eq(inventoryMovements.refId, orderId), eq(inventoryMovements.type, 'SHIP'))); const cbm = D(ship[0].cbm).gt(0) ? D(ship[0].cbm) : D(movCbm[0].cbm); const kg = D(ship[0].kg).gt(0) ? D(ship[0].kg) : D(movCbm[0].kg); return { cbm, weightKg: kg, pallets: D(ship[0].pallets), cases, pieces, lines: D(lineRows.length), containerSize: (o?.attributes as { containerSize?: string } | undefined)?.containerSize, asNumbers: { cbm: cbm.toString(), pallets: ship[0].pallets, cartons: ship[0].cartons, pieces: pieces.toString(), lines: lineRows.length } }; } /** Converts base quantities to case equivalents using each SKU's CASE pack (falls back to pieces). */ private async caseEquivalent(tx: Tx, rows: { skuId: string; qty: string }[]): Promise { if (!rows.length) return ZERO; const packs = await tx.select().from(skuPacks).where(and(inArray(skuPacks.skuId, rows.map((r) => r.skuId)), eq(skuPacks.level, 'CASE'))); return rows.reduce((acc, r) => { const p = packs.find((x) => x.skuId === r.skuId); return acc.plus(p ? D(r.qty).div(p.qtyInBase).toDecimalPlaces(4, Decimal.ROUND_UP) : D(r.qty)); }, ZERO); } } interface Metrics { cbm: Decimal; weightKg: Decimal; pallets: Decimal; cases: Decimal; pieces: Decimal; lines: Decimal; containerSize?: string; asNumbers: Record; }