import { Injectable, Logger } from '@nestjs/common';
import { and, desc, eq, inArray, isNull, lte, sql } from 'drizzle-orm';
import { CreateWaveDto, PaginationQueryDto, WaveStrategy } from '@wms/shared';
import { DbService, Tx } from '@/common/db/db.service';
import { waves, outboundOrders, outboundOrderLines, allocations, locations, zones, tasks, skus, lpns } from '@/db/schema';
import { AuditService } from '@/common/audit/audit.service';
import { NumberingService } from '@/common/numbering/numbering.service';
import { EventBus } from '@/common/events/event-bus.service';
import { requireRequestContext } from '@/common/context/request-context';
import { DomainError, NotFoundError, ValidationError, ErrorCodes } from '@/common/errors/domain-error';
import { D, sum, paginate, offsetOf, assertWarehouseAccess, serialize } from '@/common/util';
import { LookupService } from '@/modules/master-data/lookup.service';
import { TasksService } from '@/modules/tasks/tasks.service';
import { AllocationService } from './allocation.service';
import { OrdersService } from './orders.service';
import { DocumentsService } from '@/modules/documents/documents.service';

/**
 * Waves group orders for allocation and release. Release turns allocations into
 * PICK tasks sequenced along the pick path (zone sequence, then location sequence).
 *
 * BY_ORDER: one task per allocation, grouped per order (picker completes an order at a time).
 * BY_ZONE:  tasks grouped per zone across orders (zone pickers), consolidation at packing.
 * BATCH:    allocations of the same bucket across orders merged into one task with a
 *           per-order breakdown; sorted at the packing station.
 */
@Injectable()
export class WavesService {
  private readonly logger = new Logger(WavesService.name);

  constructor(
    private readonly db: DbService,
    private readonly lookup: LookupService,
    private readonly numbering: NumberingService,
    private readonly audit: AuditService,
    private readonly events: EventBus,
    private readonly tasksService: TasksService,
    private readonly allocation: AllocationService,
    private readonly orders: OrdersService,
    private readonly documents: DocumentsService,
  ) {}

  list(q: PaginationQueryDto & { warehouseId?: string; status?: string }) {
    return this.db.run(async (tx) => {
      const where = and(q.warehouseId ? eq(waves.warehouseId, q.warehouseId) : undefined, q.status ? inArray(waves.status, q.status.split(',') as never) : undefined);
      const [{ count }] = await tx.select({ count: sql<number>`count(*)`.mapWith(Number) }).from(waves).where(where);
      const rows = await tx.select().from(waves).where(where).orderBy(desc(waves.createdAt)).limit(q.pageSize).offset(offsetOf(q));
      return paginate(rows, count, q);
    });
  }

  get(id: string) {
    return this.db.run(async (tx) => {
      const [w] = await tx.select().from(waves).where(eq(waves.id, id)).limit(1);
      if (!w) throw new NotFoundError('Wave', id);
      assertWarehouseAccess(w.warehouseId);
      const orderRows = await tx.select({ id: outboundOrders.id, orderNumber: outboundOrders.orderNumber, externalRef: outboundOrders.externalRef, status: outboundOrders.status, storerId: outboundOrders.storerId, shipTo: outboundOrders.shipTo, totalOrderedQty: outboundOrders.totalOrderedQty }).from(outboundOrders).where(eq(outboundOrders.waveId, id));
      const taskRows = await tx.select({ status: tasks.status, count: sql<number>`count(*)`.mapWith(Number) }).from(tasks).where(and(eq(tasks.groupKey, id), eq(tasks.type, 'PICK'))).groupBy(tasks.status);
      return serialize({ ...w, orders: orderRows, taskSummary: taskRows });
    });
  }

  create(dto: CreateWaveDto) {
    assertWarehouseAccess(dto.warehouseId);
    const ctx = requireRequestContext();
    return this.db.run(async (tx) => {
      // select orders
      let orderIds = dto.orderIds ?? [];
      if (!orderIds.length) {
        const c = dto.criteria ?? {};
        const rows = await tx
          .select({ id: outboundOrders.id })
          .from(outboundOrders)
          .where(
            and(
              eq(outboundOrders.warehouseId, dto.warehouseId),
              inArray(outboundOrders.status, ['OPEN', 'PARTIALLY_ALLOCATED', 'ALLOCATED']),
              isNull(outboundOrders.waveId),
              c.storerId ? eq(outboundOrders.storerId, c.storerId) : undefined,
              c.carrierId ? eq(outboundOrders.carrierId, c.carrierId) : undefined,
              c.shipDateFrom ? sql`${outboundOrders.requestedShipDate} >= ${c.shipDateFrom}` : undefined,
              c.shipDateTo ? sql`${outboundOrders.requestedShipDate} <= ${c.shipDateTo}` : undefined,
              c.priorityMax ? lte(outboundOrders.priority, c.priorityMax) : undefined,
              c.deliveryLocationCode ? sql`${outboundOrders.shipTo}->>'deliveryLocationCode' = ${c.deliveryLocationCode}` : undefined,
            ),
          )
          .orderBy(outboundOrders.priority, outboundOrders.requestedShipDate, outboundOrders.createdAt)
          .limit(c.maxOrders ?? 500);
        orderIds = rows.map((r) => r.id);
      }
      if (!orderIds.length) throw new ValidationError('No eligible orders for the wave');
      const ordersRows = await tx.select().from(outboundOrders).where(inArray(outboundOrders.id, orderIds)).for('update');
      for (const o of ordersRows) {
        if (o.warehouseId !== dto.warehouseId) throw new ValidationError(`Order ${o.orderNumber} is in another warehouse`);
        if (o.waveId) throw new DomainError(ErrorCodes.INVALID_STATE, `Order ${o.orderNumber} is already in a wave`);
        if (!['OPEN', 'PARTIALLY_ALLOCATED', 'ALLOCATED'].includes(o.status)) throw new DomainError(ErrorCodes.INVALID_STATE, `Order ${o.orderNumber} is ${o.status}`);
      }
      const waveNumber = await this.numbering.next('WAVE', tx, dto.warehouseId);
      const [w] = await tx.insert(waves).values({ tenantId: ctx.tenantId!, warehouseId: dto.warehouseId, waveNumber, strategy: dto.strategy, criteria: dto.criteria ?? {}, orderCount: orderIds.length, createdById: ctx.userId! }).returning();
      await tx.update(outboundOrders).set({ waveId: w.id }).where(inArray(outboundOrders.id, orderIds));
      await this.audit.log({ action: 'wave.create', entityType: 'wave', entityId: w.id, after: { ...dto, orderIds } }, tx);
      let result = w;
      if (dto.allocateNow) result = await this.allocateInTx(tx, w.id, dto.allowPartial);
      if (dto.releaseNow) result = await this.releaseInTx(tx, w.id);
      return serialize(result);
    });
  }

  allocate(id: string, allowPartial = true) {
    return this.db.run((tx) => this.allocateInTx(tx, id, allowPartial).then(serialize));
  }

  private async allocateInTx(tx: Tx, waveId: string, allowPartial: boolean) {
    const [w] = await tx.select().from(waves).where(eq(waves.id, waveId)).limit(1).for('update');
    if (!w) throw new NotFoundError('Wave', waveId);
    if (!['PLANNED', 'ALLOCATED'].includes(w.status)) throw new DomainError(ErrorCodes.INVALID_STATE, `Wave is ${w.status}`);
    await tx.update(waves).set({ status: 'ALLOCATING' }).where(eq(waves.id, waveId));
    const orderRows = await tx.select({ id: outboundOrders.id, priority: outboundOrders.priority }).from(outboundOrders).where(eq(outboundOrders.waveId, waveId)).orderBy(outboundOrders.priority, outboundOrders.createdAt);
    let shortLines = 0;
    let lineCount = 0;
    for (const o of orderRows) {
      const res = await this.allocation.allocateOrder(tx, o.id, allowPartial);
      lineCount += res.length;
      shortLines += res.filter((r) => r.short.gt(0)).length;
      const status = await this.orders.refreshStatus(tx, o.id);
      if (status === 'ALLOCATED' || status === 'PARTIALLY_ALLOCATED') await this.events.publish('order.allocated', { orderId: o.id, status, short: res.filter((r) => r.short.gt(0)).map((r) => ({ lineId: r.lineId, short: r.short.toString() })) }, tx, { warehouseId: w.warehouseId });
    }
    const [row] = await tx.update(waves).set({ status: 'ALLOCATED', allocatedAt: new Date(), lineCount, shortLines }).where(eq(waves.id, waveId)).returning();
    await this.audit.log({ action: 'wave.allocate', entityType: 'wave', entityId: waveId, after: { lineCount, shortLines } }, tx);
    return row;
  }

  release(id: string) {
    return this.db.run((tx) => this.releaseInTx(tx, id).then(serialize));
  }

  private async releaseInTx(tx: Tx, waveId: string) {
    const [w] = await tx.select().from(waves).where(eq(waves.id, waveId)).limit(1).for('update');
    if (!w) throw new NotFoundError('Wave', waveId);
    if (w.status !== 'ALLOCATED') throw new DomainError(ErrorCodes.INVALID_STATE, `Wave must be ALLOCATED to release (is ${w.status})`);
    const wh = await this.lookup.warehouse(tx, w.warehouseId);
    const stagingId = (wh.settings as Record<string, string>).defaultStagingLocationId;
    if (!stagingId) throw new ValidationError('Warehouse has no default staging location configured');

    const rows = await tx
      .select({ a: allocations, orderNumber: outboundOrders.orderNumber, orderId: outboundOrders.id, storerId: outboundOrders.storerId, priority: outboundOrders.priority, lineNo: outboundOrderLines.lineNo, packLevel: outboundOrderLines.packLevel, zoneId: locations.zoneId, zoneCode: zones.code, zoneSeq: zones.pickSequence, locSeq: locations.pickSequence, locCode: locations.code, skuCode: skus.code })
      .from(allocations)
      .innerJoin(outboundOrders, eq(outboundOrders.id, allocations.orderId))
      .innerJoin(outboundOrderLines, eq(outboundOrderLines.id, allocations.orderLineId))
      .innerJoin(locations, eq(locations.id, allocations.locationId))
      .innerJoin(zones, eq(zones.id, locations.zoneId))
      .innerJoin(skus, eq(skus.id, allocations.skuId))
      .where(and(eq(outboundOrders.waveId, waveId), eq(allocations.status, 'ALLOCATED'), isNull(allocations.pickTaskId)));

    // pick path order
    rows.sort((x, y) => x.zoneSeq - y.zoneSeq || x.locSeq - y.locSeq || x.locCode.localeCompare(y.locCode));

    let taskCount = 0;
    let seq = 0;
    if (w.strategy === 'BATCH') {
      // merge identical buckets across orders
      const groups = new Map<string, typeof rows>();
      for (const r of rows) {
        const k = `${r.a.balanceId}`;
        groups.set(k, [...(groups.get(k) ?? []), r]);
      }
      for (const group of groups.values()) {
        seq += 10;
        const first = group[0];
        const task = await this.tasksService.create(tx, {
          warehouseId: w.warehouseId,
          type: 'PICK',
          priority: Math.min(...group.map((g) => g.priority)),
          sequence: seq,
          groupKey: waveId,
          zoneId: first.zoneId,
          storerId: first.storerId,
          skuId: first.a.skuId,
          qty: sum(group.map((g) => g.a.qty)),
          packLevel: first.packLevel,
          fromLocationId: first.a.locationId,
          toLocationId: stagingId,
          lpnId: first.a.lpnId,
          lotNumber: first.a.lotNumber,
          expiryDate: first.a.expiryDate,
          refType: 'WAVE',
          refId: waveId,
          data: { waveNumber: w.waveNumber, strategy: 'BATCH', allocations: group.map((g) => ({ allocationId: g.a.id, orderId: g.orderId, orderNumber: g.orderNumber, qty: g.a.qty.toString() })) },
        });
        await tx.update(allocations).set({ pickTaskId: task.id }).where(inArray(allocations.id, group.map((g) => g.a.id)));
        taskCount++;
      }
    } else {
      const ordered = w.strategy === 'BY_ORDER' ? [...rows].sort((x, y) => x.priority - y.priority || x.orderNumber.localeCompare(y.orderNumber) || x.zoneSeq - y.zoneSeq || x.locSeq - y.locSeq) : rows;
      for (const r of ordered) {
        seq += 10;
        const task = await this.tasksService.create(tx, {
          warehouseId: w.warehouseId,
          type: 'PICK',
          priority: r.priority,
          sequence: seq,
          groupKey: w.strategy === 'BY_ZONE' ? `${waveId}:${r.zoneCode}` : waveId,
          zoneId: r.zoneId,
          storerId: r.storerId,
          skuId: r.a.skuId,
          qty: r.a.qty,
          packLevel: r.packLevel,
          fromLocationId: r.a.locationId,
          toLocationId: stagingId,
          lpnId: r.a.lpnId,
          lotNumber: r.a.lotNumber,
          expiryDate: r.a.expiryDate,
          refType: 'ORDER',
          refId: r.orderId,
          data: { waveId, waveNumber: w.waveNumber, allocationId: r.a.id, orderNumber: r.orderNumber, lineNo: r.lineNo, serials: r.a.serials },
        });
        await tx.update(allocations).set({ pickTaskId: task.id }).where(eq(allocations.id, r.a.id));
        taskCount++;
      }
    }
    await tx.update(outboundOrders).set({ status: 'RELEASED', releasedAt: new Date() }).where(and(eq(outboundOrders.waveId, waveId), inArray(outboundOrders.status, ['ALLOCATED', 'PARTIALLY_ALLOCATED'])));
    const [row] = await tx.update(waves).set({ status: 'RELEASED', releasedAt: new Date(), taskCount }).where(eq(waves.id, waveId)).returning();
    await this.audit.log({ action: 'wave.release', entityType: 'wave', entityId: waveId, after: { taskCount } }, tx);
    await this.events.publish('wave.released', { waveId, waveNumber: w.waveNumber, taskCount }, tx, { warehouseId: w.warehouseId });
    for (const o of await tx.select({ id: outboundOrders.id, orderNumber: outboundOrders.orderNumber, storerId: outboundOrders.storerId }).from(outboundOrders).where(eq(outboundOrders.waveId, waveId))) {
      await this.events.publish('order.released', { orderId: o.id, orderNumber: o.orderNumber, waveNumber: w.waveNumber }, tx, { warehouseId: w.warehouseId, storerId: o.storerId });
    }
    return row;
  }

  /** Printable pick list for a wave (all open pick tasks in path order). */
  pickList(waveId: string) {
    return this.db.run(async (tx) => {
      const [w] = await tx.select().from(waves).where(eq(waves.id, waveId)).limit(1);
      if (!w) throw new NotFoundError('Wave', waveId);
      const rows = await tx
        .select({ t: tasks, fromLocationCode: locations.code, skuCode: skus.code, description: skus.description, lpnNumber: lpns.lpnNumber })
        .from(tasks)
        .leftJoin(locations, eq(locations.id, tasks.fromLocationId))
        .leftJoin(skus, eq(skus.id, tasks.skuId))
        .leftJoin(lpns, eq(lpns.id, tasks.lpnId))
        .where(and(eq(tasks.type, 'PICK'), sql`${tasks.groupKey} like ${waveId + '%'}`))
        .orderBy(tasks.sequence);
      const orderRows = await tx.select({ orderNumber: outboundOrders.orderNumber }).from(outboundOrders).where(eq(outboundOrders.waveId, waveId));
      const doc = await this.documents.generate(tx, {
        type: 'PICK_LIST',
        number: `PL-${w.waveNumber}`,
        refType: 'WAVE',
        refId: waveId,
        data: { wave: w, orders: orderRows, tasks: rows.map((r) => ({ sequence: r.t.sequence, fromLocationCode: r.fromLocationCode, lpnNumber: r.lpnNumber, skuCode: r.skuCode, description: r.description, lotNumber: r.t.lotNumber, qty: r.t.qty?.toString(), orderNumber: (r.t.data as { orderNumber?: string }).orderNumber ?? 'BATCH' })) },
      });
      return doc;
    });
  }

  cancel(id: string) {
    return this.db.run(async (tx) => {
      const [w] = await tx.select().from(waves).where(eq(waves.id, id)).limit(1).for('update');
      if (!w) throw new NotFoundError('Wave', id);
      if (['COMPLETED', 'CANCELLED'].includes(w.status)) throw new DomainError(ErrorCodes.INVALID_STATE, `Wave is ${w.status}`);
      const picked = await tx.select({ q: sql<string>`coalesce(sum(${allocations.pickedQty}),0)` }).from(allocations).innerJoin(outboundOrders, eq(outboundOrders.id, allocations.orderId)).where(eq(outboundOrders.waveId, id));
      if (D(picked[0].q).gt(0)) throw new DomainError(ErrorCodes.INVALID_STATE, 'Wave has picked stock; complete or unpick before cancelling');
      await tx.update(tasks).set({ status: 'CANCELLED', exceptionNotes: 'Wave cancelled' }).where(and(sql`${tasks.groupKey} like ${id + '%'}`, inArray(tasks.status, ['PENDING', 'ASSIGNED', 'IN_PROGRESS'])));
      const orderRows = await tx.select({ id: outboundOrders.id }).from(outboundOrders).where(eq(outboundOrders.waveId, id));
      for (const o of orderRows) await this.allocation.deallocateOrder(tx, o.id);
      const [row] = await tx.update(waves).set({ status: 'CANCELLED' }).where(eq(waves.id, id)).returning();
      await this.audit.log({ action: 'wave.cancel', entityType: 'wave', entityId: id }, tx);
      return serialize(row);
    });
  }

  /** Called by the pick handler after each task: closes the wave when nothing remains. */
  async checkCompletion(tx: Tx, waveId: string) {
    const [{ open }] = await tx.select({ open: sql<number>`count(*)`.mapWith(Number) }).from(tasks).where(and(sql`${tasks.groupKey} like ${waveId + '%'}`, eq(tasks.type, 'PICK'), inArray(tasks.status, ['PENDING', 'ASSIGNED', 'IN_PROGRESS', 'EXCEPTION'])));
    if (open === 0) {
      const [w] = await tx.update(waves).set({ status: 'COMPLETED', completedAt: new Date() }).where(and(eq(waves.id, waveId), inArray(waves.status, ['RELEASED', 'IN_PROGRESS']))).returning();
      if (w) await this.events.publish('wave.completed', { waveId, waveNumber: w.waveNumber }, tx, { warehouseId: w.warehouseId });
    } else {
      await tx.update(waves).set({ status: 'IN_PROGRESS' }).where(and(eq(waves.id, waveId), eq(waves.status, 'RELEASED')));
    }
  }


}