import { Injectable, Logger, OnModuleInit } from '@nestjs/common';
import { and, eq, inArray, sql } from 'drizzle-orm';
import { assertPublicUrl, fetchWithTimeout } from '@/common/util/net';
import { DbService } from '@/common/db/db.service';
import { integrationConnectors, integrationMessages, outboundOrderLines, skus, storers, warehouses } from '@/db/schema';
import { EventBus, DomainEvent } from '@/common/events/event-bus.service';
import { requireRequestContext, runWithContext, systemContext } from '@/common/context/request-context';
import { NotFoundError, ValidationError } from '@/common/errors/domain-error';
import { serialize, resolveStorerScope } from '@/common/util';
import { ApiUpsertItemsSchema, ApiAsnSchema, ApiOrderSchema } from '@wms/shared';
import { PublicApiService } from '../public-api.service';
import { WebhooksService } from '../webhooks.service';
import { ErpAdapter, NormalizedInbound } from './erp-adapter';
import { SapIdocAdapter } from './sap-idoc.adapter';

/**
 * Routes inbound ERP documents to the right adapter and produces outbound ERP
 * messages (GRN / PGI confirmations) into integration_messages. Outbound
 * messages are either pushed to the connector's endpoint (config.outboundUrl)
 * or pulled by the ERP middleware from GET /v1/erp/outbox and acknowledged.
 */
@Injectable()
export class ErpService implements OnModuleInit {
  private readonly logger = new Logger(ErpService.name);
  private readonly adapters = new Map<string, ErpAdapter>();

  constructor(
    private readonly db: DbService,
    private readonly events: EventBus,
    private readonly publicApi: PublicApiService,
    private readonly webhooks: WebhooksService,
  ) {
    this.register(new SapIdocAdapter());
  }

  register(adapter: ErpAdapter) {
    this.adapters.set(adapter.type, adapter);
  }

  onModuleInit() {
    this.events.subscribe('asn.closed', (e) => this.produceOutbound(e));
    this.events.subscribe('order.shipped', (e) => this.produceOutbound(e));
    this.events.subscribe('order.short_closed', (e) => this.produceOutbound(e));
  }

  /** Inbound document from the ERP (raw text). */
  async ingest(connectorId: string, raw: string, idempotencyKey?: string) {
    const ctx = requireRequestContext();
    // 1. parse + log (own transaction so the log survives a failing apply)
    const { conn, messages } = await this.db.run(async (tx) => {
      const [c] = await tx.select().from(integrationConnectors).where(and(eq(integrationConnectors.id, connectorId), eq(integrationConnectors.active, true))).limit(1);
      if (!c) throw new NotFoundError('Connector', connectorId);
      if (ctx.storerId && c.storerId && c.storerId !== ctx.storerId) throw new NotFoundError('Connector', connectorId);
      const adapter = this.adapters.get(c.type);
      if (!adapter) throw new ValidationError(`No adapter for connector type ${c.type}`);
      const config = this.webhooks.decryptConfig(c.config);
      const storerCode = (config.storerCode as string | undefined) ?? (c.storerId ? (await tx.select({ code: storers.code }).from(storers).where(eq(storers.id, c.storerId)).limit(1))[0]?.code : undefined);
      return { conn: c, messages: adapter.parseInbound(raw, { ...config, storerCode }) };
    });
    const results: unknown[] = [];
    for (const [i, m] of messages.entries()) {
      const key = idempotencyKey ? `${idempotencyKey}:${m.externalId ?? i}` : m.externalId ? `${conn.id}:${m.kind}:${m.externalId}` : undefined;
      const logRow = await this.db.run(async (tx) => {
        const [row] = await tx
          .insert(integrationMessages)
          .values({ tenantId: ctx.tenantId!, storerId: conn.storerId, connectorId: conn.id, direction: 'INBOUND', messageType: `${conn.type}_${m.kind}`, externalId: m.externalId ?? null, idempotencyKey: key ?? null, payload: m as unknown as Record<string, unknown>, rawPayload: raw.length > 200000 ? raw.slice(0, 200000) : raw, status: 'PROCESSING', attempts: 1 })
          .onConflictDoNothing({ target: [integrationMessages.tenantId, integrationMessages.idempotencyKey] })
          .returning();
        return row ?? null;
      });
      if (!logRow) {
        results.push({ kind: m.kind, externalId: m.externalId, idempotent: true });
        continue;
      }
      // 2. apply in its own transaction; 3. record the outcome
      try {
        const r = await this.apply(m);
        await this.db.run((tx) => tx.update(integrationMessages).set({ status: 'PROCESSED', processedAt: new Date(), resultRefType: m.kind, resultRefId: (r as { id?: string }).id ?? null }).where(eq(integrationMessages.id, logRow.id)));
        results.push({ kind: m.kind, externalId: m.externalId, ...r });
      } catch (e) {
        await this.db.run((tx) => tx.update(integrationMessages).set({ status: 'FAILED', lastError: (e as Error).message }).where(eq(integrationMessages.id, logRow.id)));
        results.push({ kind: m.kind, externalId: m.externalId, error: (e as Error).message });
      }
    }
    return results;
  }

  /** Adapter output is validated against the public API contracts before it touches the database (NaN quantities, missing codes ...). */
  private async apply(m: NormalizedInbound): Promise<Record<string, unknown>> {
    switch (m.kind) {
      case 'ITEMS': {
        const parsed = ApiUpsertItemsSchema.safeParse({ items: m.items });
        if (!parsed.success) throw new ValidationError('ERP item document is not valid', parsed.error.flatten());
        return this.publicApi.upsertItems(m.storerCode, parsed.data.items as never) as Promise<Record<string, unknown>>;
      }
      case 'ASN': {
        const parsed = ApiAsnSchema.safeParse(m.asn);
        if (!parsed.success) throw new ValidationError('ERP purchase order document is not valid', parsed.error.flatten());
        return this.publicApi.createAsn(parsed.data as never) as Promise<Record<string, unknown>>;
      }
      case 'ORDER': {
        const parsed = ApiOrderSchema.safeParse(m.order);
        if (!parsed.success) throw new ValidationError('ERP order document is not valid', parsed.error.flatten());
        return this.publicApi.createOrder(parsed.data as never) as Promise<Record<string, unknown>>;
      }
    }
  }

  /** On relevant events, every active ERP connector of the storer gets an outbound message. */
  private async produceOutbound(e: DomainEvent) {
    if (!e.storerId) return;
    const pushes: { id: string; url: string; body: string; config: Record<string, unknown> }[] = [];
    await runWithContext(systemContext(e.tenantId), () =>
      this.db.run(async (tx) => {
        const conns = await tx.select().from(integrationConnectors).where(and(eq(integrationConnectors.active, true), inArray(integrationConnectors.type, Array.from(this.adapters.keys()) as never), sql`(${integrationConnectors.storerId} is null or ${integrationConnectors.storerId} = ${e.storerId})`));
        if (!conns.length) return;
        const [storer] = await tx.select({ code: storers.code }).from(storers).where(eq(storers.id, e.storerId!)).limit(1);
        const [wh] = e.warehouseId ? await tx.select({ code: warehouses.code }).from(warehouses).where(eq(warehouses.id, e.warehouseId)).limit(1) : [null];
        let document: Record<string, unknown> | undefined;
        if (e.type === 'order.shipped' || e.type === 'order.short_closed') {
          const lines = await tx.select({ l: outboundOrderLines, skuCode: skus.code }).from(outboundOrderLines).innerJoin(skus, eq(skus.id, outboundOrderLines.skuId)).where(eq(outboundOrderLines.orderId, e.payload.orderId as string));
          document = { lines: lines.filter((x) => Number(x.l.shippedQty) > 0).map((x) => ({ skuCode: x.skuCode, shippedQty: x.l.shippedQty.toString(), externalLineRef: x.l.externalLineRef ?? undefined, lotNumber: x.l.requestedLot ?? undefined })) };
        }
        for (const conn of conns) {
          const adapter = this.adapters.get(conn.type)!;
          const config = this.webhooks.decryptConfig(conn.config);
          const msg = adapter.formatOutbound(e, { storerCode: storer?.code, warehouseCode: wh?.code, document }, config);
          if (!msg) continue;
          const [row] = await tx
            .insert(integrationMessages)
            .values({ tenantId: e.tenantId, storerId: e.storerId, connectorId: conn.id, direction: 'OUTBOUND', messageType: msg.messageType, externalId: msg.externalId ?? null, idempotencyKey: `${conn.id}:${e.id}`, payload: msg.payload, rawPayload: msg.rawPayload ?? null, status: 'RECEIVED' })
            .onConflictDoNothing({ target: [integrationMessages.tenantId, integrationMessages.idempotencyKey] })
            .returning();
          if (row && config.outboundUrl) pushes.push({ id: row.id, url: String(config.outboundUrl), body: msg.rawPayload ?? JSON.stringify(msg.payload), config });
        }
      }),
    );
    // Deliver after the transaction committed so a slow ERP never holds a database connection or the dispatcher.
    for (const p of pushes) await this.push(p.id, p.url, p.body, p.config);
  }

  private async push(messageId: string, url: string, body: string, config: Record<string, unknown>) {
    try {
      const headers: Record<string, string> = { 'content-type': body.trim().startsWith('<') ? 'application/xml' : 'application/json' };
      if (config.outboundAuthHeader) headers['authorization'] = String(config.outboundAuthHeader);
      await assertPublicUrl(url, 'Connector outbound URL');
      const res = await fetchWithTimeout(url, { method: 'POST', headers, body }, 15000);
      await this.db.runBypass((tx) => tx.update(integrationMessages).set({ status: res.ok ? 'PROCESSED' : 'FAILED', processedAt: res.ok ? new Date() : null, attempts: 1, lastError: res.ok ? null : `HTTP ${res.status}` }).where(eq(integrationMessages.id, messageId)));
    } catch (e) {
      await this.db.runBypass((tx) => tx.update(integrationMessages).set({ status: 'FAILED', attempts: 1, lastError: (e as Error).message }).where(eq(integrationMessages.id, messageId)));
    }
  }

  /** Pull model: unacknowledged outbound messages for the caller's storer / all. */
  outbox(limit = 50) {
    const ctx = requireRequestContext();
    return this.db.run(async (tx) => {
      const rows = await tx
        .select()
        .from(integrationMessages)
        .where(and(eq(integrationMessages.direction, 'OUTBOUND'), eq(integrationMessages.status, 'RECEIVED'), ctx.storerId ? eq(integrationMessages.storerId, ctx.storerId) : undefined))
        .orderBy(integrationMessages.createdAt)
        .limit(limit);
      return serialize(rows.map((r) => ({ id: r.id, messageType: r.messageType, externalId: r.externalId, createdAt: r.createdAt, payload: r.payload, raw: r.rawPayload })));
    });
  }

  ack(ids: string[]) {
    return this.db.run(async (tx) => {
      const scoped = resolveStorerScope();
      const rows = await tx.update(integrationMessages).set({ status: 'PROCESSED', processedAt: new Date() }).where(and(inArray(integrationMessages.id, ids), eq(integrationMessages.direction, 'OUTBOUND'), scoped ? eq(integrationMessages.storerId, scoped) : undefined)).returning({ id: integrationMessages.id });
      return { acknowledged: rows.length };
    });
  }

  async retry(messageId: string) {
    const prep = await this.db.run(async (tx) => {
      const [m] = await tx.select().from(integrationMessages).where(eq(integrationMessages.id, messageId)).limit(1);
      if (!m) throw new NotFoundError('Message', messageId);
      if (m.direction === 'OUTBOUND') {
        await tx.update(integrationMessages).set({ status: 'RECEIVED', lastError: null }).where(eq(integrationMessages.id, messageId));
        return null;
      }
      if (!m.connectorId || !m.rawPayload) throw new ValidationError('Only connector-based inbound messages with a raw payload can be retried');
      await tx.update(integrationMessages).set({ attempts: m.attempts + 1 }).where(eq(integrationMessages.id, messageId));
      return { connectorId: m.connectorId, raw: m.rawPayload, id: m.id };
    });
    if (!prep) return { ok: true };
    return { ok: true, results: await this.ingest(prep.connectorId, prep.raw, `retry:${prep.id}:${Date.now()}`) };
  }
}