import { Injectable, Logger, OnModuleInit } from '@nestjs/common';
import { and, eq, inArray, lte, sql } from 'drizzle-orm';
import { createHmac } from 'crypto';
import { DbService } from '@/common/db/db.service';
import { assertPublicUrl, fetchWithTimeout } from '@/common/util/net';
import { webhookDeliveries, webhookSubscriptions, emailOutbox, documents } from '@/db/schema';
import { EncryptionService } from '@/common/crypto/encryption.service';
import { MailService } from '@/common/mail/mail.service';
import { StorageService } from '@/common/storage/storage.service';
import { loadEnv } from '@/config/env';

const BACKOFF_MINUTES = [1, 5, 30, 120, 720];

/** Delivers webhooks (HMAC-SHA256 signed) and queued emails with retry/backoff. */
@Injectable()
export class SendersJob implements OnModuleInit {
  private readonly logger = new Logger(SendersJob.name);
  private busyWebhooks = false;
  private busyEmails = false;

  constructor(
    private readonly db: DbService,
    private readonly crypto: EncryptionService,
    private readonly mail: MailService,
    private readonly storage: StorageService,
  ) {}

  onModuleInit() {
    if (!loadEnv().JOBS_ENABLED) return;
    setInterval(() => void this.sendWebhooks(), 5000).unref();
    setInterval(() => void this.sendEmails(), 10000).unref();
  }

  async sendWebhooks(): Promise<number> {
    if (this.busyWebhooks) return 0;
    this.busyWebhooks = true;
    let sent = 0;
    try {
      const batch = await this.db.runBypass(async (tx) => {
        const rows = await tx
          .select({ d: webhookDeliveries, s: webhookSubscriptions })
          .from(webhookDeliveries)
          .innerJoin(webhookSubscriptions, eq(webhookSubscriptions.id, webhookDeliveries.subscriptionId))
          .where(and(eq(webhookDeliveries.status, 'PENDING'), lte(webhookDeliveries.nextAttemptAt, new Date())))
          .orderBy(webhookDeliveries.createdAt)
          .limit(25)
          .for('update', { of: webhookDeliveries, skipLocked: true });
        if (rows.length) await tx.update(webhookDeliveries).set({ status: 'SENDING' }).where(inArray(webhookDeliveries.id, rows.map((r) => r.d.id)));
        return rows;
      });
      for (const { d, s } of batch) {
        const body = JSON.stringify(d.payload);
        const secret = this.crypto.decrypt(s.secretEncrypted);
        const ts = Math.floor(Date.now() / 1000).toString();
        const signature = createHmac('sha256', secret).update(`${ts}.${body}`).digest('hex');
        let code: number | null = null;
        let text = '';
        try {
          await assertPublicUrl(s.url, 'Webhook URL'); // re-checked at send time (DNS rebinding)
          const res = await fetchWithTimeout(s.url, { method: 'POST', headers: { 'content-type': 'application/json', 'x-wms-event': d.eventType, 'x-wms-delivery': d.id, 'x-wms-timestamp': ts, 'x-wms-signature': `v1=${signature}` }, body }, 10000);
          code = res.status;
          text = (await res.text()).slice(0, 300);
        } catch (e) {
          text = (e as Error).message;
        }
        const ok = code !== null && code >= 200 && code < 300;
        const attempts = d.attempts + 1;
        const next = ok ? null : attempts >= BACKOFF_MINUTES.length ? null : new Date(Date.now() + BACKOFF_MINUTES[attempts] * 60000);
        await this.db.runBypass((tx) => tx.update(webhookDeliveries).set({ status: ok ? 'DELIVERED' : next ? 'PENDING' : 'DEAD', attempts, responseCode: code, responseBody: text, nextAttemptAt: next, deliveredAt: ok ? new Date() : null }).where(eq(webhookDeliveries.id, d.id)));
        if (ok) sent++;
        else this.logger.warn(`Webhook ${d.id} to ${s.url} failed (${code ?? 'network'}), attempt ${attempts}`);
      }
    } catch (e) {
      this.logger.error(`Webhook sender failed: ${(e as Error).message}`);
    } finally {
      this.busyWebhooks = false;
    }
    return sent;
  }

  async sendEmails(): Promise<number> {
    if (this.busyEmails) return 0;
    this.busyEmails = true;
    let sent = 0;
    try {
      const batch = await this.db.runBypass(async (tx) => {
        const rows = await tx.select().from(emailOutbox).where(and(eq(emailOutbox.status, 'PENDING'), sql`${emailOutbox.attempts} < 5`)).orderBy(emailOutbox.createdAt).limit(20).for('update', { skipLocked: true });
        if (rows.length) await tx.update(emailOutbox).set({ status: 'SENDING' }).where(inArray(emailOutbox.id, rows.map((r) => r.id)));
        return rows;
      });
      for (const m of batch) {
        try {
          const attachments: { filename: string; content: Buffer }[] = [];
          for (const a of m.attachments) {
            if (a.documentId) {
              const [doc] = await this.db.runBypass((tx) => tx.select().from(documents).where(eq(documents.id, a.documentId!)).limit(1));
              if (doc) attachments.push({ filename: `${doc.number}.pdf`, content: await this.storage.get(doc.storageKey) });
            } else if (a.storageKey) {
              attachments.push({ filename: a.fileName ?? 'attachment', content: await this.storage.get(a.storageKey) });
            }
          }
          await this.mail.deliver({ to: m.to, cc: m.cc, subject: m.subject, html: m.bodyHtml, attachments });
          await this.db.runBypass((tx) => tx.update(emailOutbox).set({ status: 'SENT', sentAt: new Date(), attempts: m.attempts + 1 }).where(eq(emailOutbox.id, m.id)));
          sent++;
        } catch (e) {
          this.logger.error(`Email ${m.id} failed: ${(e as Error).message}`);
          await this.db.runBypass((tx) => tx.update(emailOutbox).set({ status: m.attempts + 1 >= 5 ? 'FAILED' : 'PENDING', attempts: m.attempts + 1, lastError: (e as Error).message.slice(0, 1000) }).where(eq(emailOutbox.id, m.id)));
        }
      }
    } finally {
      this.busyEmails = false;
    }
    return sent;
  }
}