import { Injectable, Logger, OnModuleInit } from '@nestjs/common';
import { and, eq, inArray, sql, or, isNull } from 'drizzle-orm';
import { DbService } from '@/common/db/db.service';
import { outboxEvents, webhookSubscriptions, webhookDeliveries } from '@/db/schema';
import { EventBus, DomainEvent } from '@/common/events/event-bus.service';
import { loadEnv } from '@/config/env';
import { runWithContext, systemContext } from '@/common/context/request-context';

/** A claimed event not finished within this time is considered abandoned and re-dispatched. */
const LEASE_MINUTES = 5;
const MAX_ATTEMPTS = 5;

/**
 * Polls outbox_events and fans each event out to in-process handlers and to
 * matching webhook subscriptions. Safe to run on several API instances:
 * rows are claimed with FOR UPDATE SKIP LOCKED.
 */
@Injectable()
export class OutboxDispatcherJob implements OnModuleInit {
  private readonly logger = new Logger(OutboxDispatcherJob.name);
  private running = false;
  private timer?: NodeJS.Timeout;

  constructor(
    private readonly db: DbService,
    private readonly bus: EventBus,
  ) {}

  onModuleInit() {
    if (!loadEnv().JOBS_ENABLED) return;
    this.timer = setInterval(() => void this.tick(), 2000);
    this.timer.unref();
  }

  async tick(): Promise<number> {
    if (this.running) return 0;
    this.running = true;
    let processed = 0;
    try {
      // claim a batch
      const batch = await this.db.runBypass(async (tx) => {
        // PENDING rows, plus PROCESSING rows whose lease expired (worker crashed mid-batch).
        const rows = await tx
          .select()
          .from(outboxEvents)
          .where(or(eq(outboxEvents.status, 'PENDING'), and(eq(outboxEvents.status, 'PROCESSING'), sql`${outboxEvents.claimedAt} < now() - interval '${sql.raw(String(LEASE_MINUTES))} minutes'`)))
          .orderBy(outboxEvents.createdAt)
          .limit(50)
          .for('update', { skipLocked: true });
        if (rows.length) await tx.update(outboxEvents).set({ status: 'PROCESSING', claimedAt: new Date() }).where(inArray(outboxEvents.id, rows.map((r) => r.id)));
        return rows;
      });
      for (const row of batch) {
        const event: DomainEvent = { id: row.id, type: row.type as DomainEvent['type'], tenantId: row.tenantId, warehouseId: row.warehouseId ?? undefined, storerId: row.storerId ?? undefined, occurredAt: row.createdAt.toISOString(), actorId: row.actorId ?? undefined, attempt: row.attempts, payload: row.payload };
        try {
          const { errors } = await runWithContext(systemContext(row.tenantId), () => this.bus.dispatch(event));
          await runWithContext(systemContext(row.tenantId), () => this.enqueueWebhooks(event));
          if (errors.length) throw new Error(errors.map((e) => e.message).join('; '));
          await this.db.runBypass((tx) => tx.update(outboxEvents).set({ status: 'PROCESSED', processedAt: new Date(), attempts: row.attempts + 1 }).where(eq(outboxEvents.id, row.id)));
          processed++;
        } catch (e) {
          const attempts = row.attempts + 1;
          const dead = attempts >= MAX_ATTEMPTS;
          this.logger.error(`Event ${row.type} ${row.id} failed (attempt ${attempts}): ${(e as Error).message}`);
          await this.db.runBypass((tx) => tx.update(outboxEvents).set({ status: dead ? 'DEAD' : 'PENDING', attempts, lastError: (e as Error).message.slice(0, 2000) }).where(eq(outboxEvents.id, row.id)));
        }
      }
    } catch (e) {
      this.logger.error(`Outbox tick failed: ${(e as Error).message}`);
    } finally {
      this.running = false;
    }
    return processed;
  }

  private async enqueueWebhooks(event: DomainEvent) {
    await this.db.runAsTenant(event.tenantId, async (tx) => {
      const subs = await tx
        .select()
        .from(webhookSubscriptions)
        .where(and(eq(webhookSubscriptions.active, true), sql`${event.type} = any(${webhookSubscriptions.events})`, or(isNull(webhookSubscriptions.storerId), event.storerId ? eq(webhookSubscriptions.storerId, event.storerId) : sql`false`)));
      if (!subs.length) return;
      const { _actorName, ...payload } = event.payload as Record<string, unknown> & { _actorName?: string };
      void _actorName;
      await tx.insert(webhookDeliveries).values(
        subs.map((s) => ({ tenantId: event.tenantId, subscriptionId: s.id, eventType: event.type, eventId: event.id, payload: { id: event.id, type: event.type, occurredAt: event.occurredAt, warehouseId: event.warehouseId, storerId: event.storerId, data: payload }, nextAttemptAt: new Date() })),
      );
    });
  }
}