/**
 * apps/worker — outbound webhooks (dispatcher + event fan-out), TASK-0130.
 *
 * Inbound webhooks (the API exposing /webhooks/incoming/…) live in apps/api —
 * this processor is OUTBOUND only. The delivery CONTRACT (envelope,
 * signature, transport, SSRF, backoff, 4xx/5xx split) is implemented ONCE in
 * `@factory/shared` (`webhook-delivery.ts`) and shared with the API's
 * synchronous test-send endpoint; this file is the BullMQ/Postgres
 * orchestration around it.
 *
 * Entry points:
 *  1. FAN-OUT (stream) — StreamConsumer on `factory:events`
 *     (group worker-webhooks): for each catalog event, look up active
 *     Webhook rows subscribed to it (exact name or `*`) and enqueue one
 *     dispatch job per webhook. The stable `evt_…` id is minted ONCE here
 *     (dedupe key — `X-Factory-Event-Id`).
 *  2. DISPATCH (queue) — BullMQ `webhooks`, row jobs:
 *        { kind:"row", webhookId, event, payload, deliveryId?, eventId?,
 *          test?, resubmit? (deliveryId + originDeliveryId of the retry) }
 *     The API enqueues the same shape for test sends + resubmits.
 *
 * Dispatch (row):
 *   - load Webhook (must exist + active + subscribed; `*` = all),
 *   - resolve the signing secret (shared `resolveSigningSecret`:
 *     `secretEncrypted` → legacy ciphertext in `secretHash` → bare-hash),
 *   - build the docs §3 envelope `{id: evt_…, type, workspaceId, occurredAt,
 *     workspace:{id,name,slug}, data, dataDelta?}` (occurredAt frozen at
 *     fan-out so retries don't mutate the payload),
 *   - POST via the shared transport (10 s timeout, ≤3 redirects re-SSRF'd),
 *   - 2xx → DELIVERED {statusCode, deliveredAt};
 *     4xx → REJECTED (terminal — the endpoint's contract bug, no retry);
 *     5xx/timeout/transport → FAILED + RE-ENQUEUE after the backoff delay
 *     (worker-owned retry, docs §4.3 / task `min(30s·2^attempt,15min)` ±20%
 *     jitter, env-overridable); after 8 attempts → DEAD_LETTER (terminal)
 *     + admin notification (throttled 24 h per webhook, Redis key).
 *
 * BullMQ attempts are set to 1 (no built-in backoff): every failed attempt
 * is scheduled as a DELAYED job with the exact backoff delay, so
 * `webhook_deliveries.nextRetryAt` is always the real scheduled time and the
 * 8-attempt cap is deterministic (C-13.8d).
 */
import { asJson } from "../lib/prisma.js";
import { QUEUES, enqueue } from "../queues.js";
import { getLogger } from "../logger.js";
import type { WorkerEnv } from "../config.js";
import { StreamConsumer, type StreamEntry } from "../lib/redis-stream.js";
import { notifyWorkspaceAdmins } from "../lib/notify.js";
import { tryParseEvent, WEBHOOK_EVENT_CATALOG } from "../events.js";
import {
  buildDeliveryEnvelope,
  classifyDeliveryResult,
  generateId,
  newDeliveryId,
  postWebhookDelivery,
  resolveSigningSecret,
  webhookBackoffMs,
  type DeliveryEnvelopeInput
} from "@factory/shared";
import { Worker, type Job } from "bullmq";
import { withJobOtel } from "../lib/job-otel.js";
import type { Redis } from "ioredis";
import { z } from "zod";
import type { PrismaClient } from "@prisma/client";

const log = getLogger("webhooks");

/** 8 total attempts (1 initial + 7 retries) — docs §4.3. */
export const WEBHOOK_MAX_ATTEMPTS = 8;
/** 24 h admin-notification throttle per dead-lettered webhook (Redis key). */
const NOTIFY_THROTTLE_KEY = (webhookId: string) => `webhook:dlq-notified:${webhookId}`;
const NOTIFY_THROTTLE_SECONDS = 24 * 3600;

// ---------------------------------------------------------------------------
// Job schema (row-based only — the send_webhook automation action does its
// own inline fetch in the automations executor and never touches this queue)
// ---------------------------------------------------------------------------

export const webhookJobSchema = z.object({
  kind: z.literal("row"),
  webhookId: z.string().min(1),
  event: z.string().min(1),
  payload: z.record(z.string(), z.unknown()).default({}),
  deliveryId: z.string().min(1).optional(),
  /** Stable envelope id (`evt_…`) — minted once at fan-out; reused on retry. */
  eventId: z.string().min(1).optional(),
  /** Frozen occurredAt (ISO) — set at fan-out; retries keep it. */
  occurredAt: z.string().min(1).optional(),
  test: z.literal(true).optional(),
  /** Resubmit marker: the ORIGINAL delivery id this row re-delivers. */
  originDeliveryId: z.string().min(1).optional(),
  /** 1-based attempt counter (carried across worker-owned retries). */
  attempt: z.number().int().min(1).optional()
});
export type WebhookJobData = z.infer<typeof webhookJobSchema>;

// ---------------------------------------------------------------------------
// Subscription matching (exact name or the `*` wildcard)
// ---------------------------------------------------------------------------

export function isWebhookSubscribed(events: string[], event: string): boolean {
  return events.includes("*") || events.includes(event);
}

/**
 * Backoff for the Nth attempt (1-based) using the deployment's configured
 * base/cap (task default 30 s / 15 min, ±20 % jitter — C-13.8d).
 */
function backoffFor(env: WorkerEnv, attempt: number): number {
  return webhookBackoffMs(attempt, {
    baseMs: env.WEBHOOK_BACKOFF_BASE_MS,
    maxMs: env.WEBHOOK_BACKOFF_MAX_MS
  });
}

// ---------------------------------------------------------------------------
// Worker
// ---------------------------------------------------------------------------

export function createWebhooksWorker(opts: {
  connection: Redis;
  prisma: PrismaClient;
  env: WorkerEnv;
  concurrency?: number;
  fetchImpl?: typeof fetch;
}): Worker {
  const { connection, prisma, env } = opts;
  const doFetch = opts.fetchImpl ?? fetch;

  return new Worker(
    QUEUES.webhooks,
    withJobOtel(QUEUES.webhooks, async (job: Job) => {
      let data: WebhookJobData;
      try {
        data = webhookJobSchema.parse(job.data);
      } catch (err) {
        log.error({ jobId: job.id, err: (err as Error).message }, "webhook job unparseable — discarding");
        job.discard();
        return;
      }
      try {
        await dispatchRow(doFetch, prisma, connection, env, job, data);
      } catch (err) {
        // Unexpected crash mid-dispatch (DB outage, …): record it, dead-letter
        // the row if it exists, and DISCARD — BullMQ's default 8-attempt
        // backoff would re-run it with a schedule the delivery log never
        // shows (C-13.8d: the log must reflect the real schedule).
        log.error(
          { jobId: job.id, webhookId: data.webhookId, event: data.event, err: (err as Error).message },
          "webhook dispatch crashed — dead-lettering"
        );
        if (data.deliveryId) {
          await prisma.webhookDelivery
            .update({
              where: { id: data.deliveryId },
              data: { status: "DEAD_LETTER", error: `crashed: ${(err as Error).message}`, nextRetryAt: null }
            })
            .catch(() => undefined);
        }
        job.discard();
      }
    }),
    {
      connection,
      concurrency: opts.concurrency ?? env.WORKER_CONCURRENCY
      // Worker-owned retry scheduling: no BullMQ attempts/backoff — every
      // FAILED attempt re-enqueues itself as a delayed job (see dispatchRow).
    }
  );
}

/** One dispatch attempt for a row-based job. Never throws on 4xx/5xx —
 *  terminal states are recorded, retries are re-enqueued. */
async function dispatchRow(
  doFetch: typeof fetch,
  prisma: PrismaClient,
  connection: Redis,
  env: WorkerEnv,
  job: Job,
  data: WebhookJobData
): Promise<void> {
  const { webhookId, event } = data;
  const isTest = Boolean(data.test);
  const isResubmit = Boolean(data.originDeliveryId);
  const attempt = data.attempt ?? 1;
  const started = Date.now();
  log.info(
    { jobId: job.id, webhookId, event, test: isTest, resubmit: isResubmit, attempt },
    "dispatching webhook"
  );

  const webhook = await prisma.webhook.findUnique({ where: { id: webhookId } });
  if (!webhook) {
    log.warn({ jobId: job.id, webhookId }, "webhook row missing — discarding");
    await touchDelivery(prisma, data.deliveryId, { status: "REJECTED", error: "webhook deleted" });
    job.discard();
    return;
  }
  if (!webhook.active && !isTest) {
    log.info({ jobId: job.id, webhookId }, "webhook inactive — discarding");
    await touchDelivery(prisma, data.deliveryId, { status: "REJECTED", error: "webhook inactive" });
    job.discard();
    return;
  }
  if (!isTest && !isWebhookSubscribed(webhook.events, event)) {
    log.info({ jobId: job.id, webhookId, event }, "not subscribed — discarding");
    await touchDelivery(prisma, data.deliveryId, { status: "REJECTED", error: "not subscribed" });
    job.discard();
    return;
  }

  const resolved = resolveSigningSecret(env.MASTER_ENCRYPTION_KEY, webhook.secretEncrypted, webhook.secretHash);
  if (!resolved) {
    log.error({ jobId: job.id, webhookId }, "webhook has no resolvable signing secret — dead-lettering");
    await touchDelivery(prisma, data.deliveryId, { status: "DEAD_LETTER", error: "signing secret unavailable" });
    job.discard();
    return;
  }

  const deliveryId = data.deliveryId ?? newDeliveryId();
  const wireEvent = isTest ? "webhook.test" : event;

  // The docs §3 envelope — `id` + `occurredAt` are frozen across retries.
  const ws = await prisma.workspace.findUnique({
    where: { id: webhook.workspaceId },
    select: { id: true, name: true, slug: true }
  });
  const input: DeliveryEnvelopeInput = {
    type: wireEvent,
    workspaceId: webhook.workspaceId,
    data: data.payload
  };
  if (data.eventId) input.id = data.eventId;
  if (ws) input.workspace = ws;
  if (data.occurredAt) input.occurredAt = data.occurredAt;
  if (isTest) input.test = true;
  const envelope = buildDeliveryEnvelope(input);

  // PENDING upsert: the delivery row exists across every retry; `attempts`
  // is the attempt THIS dispatch is making (1-based).
  await prisma.webhookDelivery.upsert({
    where: { id: deliveryId },
    create: {
      id: deliveryId,
      webhookId,
      event: wireEvent,
      payload: asJson(envelope),
      status: "PENDING",
      attempts: attempt,
      originDeliveryId: data.originDeliveryId ?? null
    },
    update: { attempts: attempt, nextRetryAt: null, status: "PENDING" }
  });

  const result = await postWebhookDelivery(
    {
      url: webhook.url,
      secret: resolved.secret,
      event: wireEvent,
      envelope,
      deliveryId,
      attempt
    },
    { allowPrivate: env.WEBHOOK_ALLOW_PRIVATE, fetchImpl: doFetch }
  );
  const outcome = classifyDeliveryResult(result);

  if (outcome === "DELIVERED") {
    await prisma.webhookDelivery.update({
      where: { id: deliveryId },
      data: {
        status: "DELIVERED",
        statusCode: result.status,
        error: null,
        deliveredAt: new Date(),
        latencyMs: result.latencyMs
      }
    });
    log.info(
      {
        jobId: job.id,
        webhookId,
        deliveryId,
        status: result.status,
        latencyMs: result.latencyMs,
        attempt,
        secretSource: resolved.source
      },
      "webhook delivered"
    );
    return;
  }

  if (outcome === "REJECTED") {
    // 4xx = the endpoint's contract bug: terminal, NO retry (docs §4.1).
    await prisma.webhookDelivery.update({
      where: { id: deliveryId },
      data: {
        status: "REJECTED",
        statusCode: result.status,
        error: `rejected: HTTP ${result.status}`,
        nextRetryAt: null,
        latencyMs: result.latencyMs
      }
    });
    log.warn({ jobId: job.id, webhookId, deliveryId, status: result.status, attempt }, "webhook 4xx — rejected (no retry)");
    return;
  }

  // FAILED (5xx / timeout / transport): retry, or dead-letter at the cap.
  if (attempt >= WEBHOOK_MAX_ATTEMPTS) {
    await markDeadLetter(prisma, connection, env, webhook, deliveryId, result, attempt);
    return;
  }
  const delay = backoffFor(env, attempt);
  const nextRetryAt = new Date(Date.now() + delay);
  await prisma.webhookDelivery.update({
    where: { id: deliveryId },
    data: {
      status: "FAILED",
      statusCode: result.status || null,
      error: result.transportError ?? `non-2xx: HTTP ${result.status}`,
      nextRetryAt,
      latencyMs: result.latencyMs
    }
  });
  // Re-enqueue the SAME logical delivery as a delayed job (stable jobId —
  // BullMQ coalesces duplicates; the attempt counter rides in the payload).
  await enqueue(connection, QUEUES.webhooks, {
    kind: "row",
    webhookId,
    event,
    payload: data.payload,
    deliveryId,
    eventId: data.eventId,
    occurredAt: data.occurredAt,
    attempt: attempt + 1,
    originDeliveryId: data.originDeliveryId
  // BullMQ job ids may not contain ":" ("Custom Id cannot contain :") — dash it.
  }, { delay, jobId: `${deliveryId}-a${attempt + 1}` });
  log.warn(
    { jobId: job.id, webhookId, deliveryId, status: result.status, attempt, nextRetryMs: delay },
    "webhook failed — retry scheduled"
  );
}

/** DEAD_LETTER terminal status + throttled (24 h) admin notification. */
async function markDeadLetter(
  prisma: PrismaClient,
  connection: Redis,
  env: WorkerEnv,
  webhook: { id: string; workspaceId: string; url?: string },
  deliveryId: string,
  result: { status: number; transportError?: string; latencyMs: number },
  attempt: number
): Promise<void> {
  const error = result.transportError ?? `non-2xx: HTTP ${result.status}`;
  log.error(
    { webhookId: webhook.id, deliveryId, status: result.status, attempt, err: error },
    "webhook dispatch failed after all retries → DEAD_LETTER"
  );
  await prisma.webhookDelivery.update({
    where: { id: deliveryId },
    data: {
      status: "DEAD_LETTER",
      statusCode: result.status || null,
      error,
      nextRetryAt: null,
      latencyMs: result.latencyMs
    }
  });
  // Alert (in-app + email) — throttled 24 h per webhook so a persistently
  // dead endpoint does not spam the admin inbox on every event.
  try {
    const key = NOTIFY_THROTTLE_KEY(webhook.id);
    const notified = await connection.set(key, "1", "EX", NOTIFY_THROTTLE_SECONDS, "NX");
    if (notified !== "OK") return;
    await notifyWorkspaceAdmins(
      prisma,
      connection,
      env.SMTP_HOST,
      webhook.workspaceId,
      "Webhook delivery failed",
      `Webhook ${webhook.id} (${webhook.url}) could not be completed after ${attempt} attempts (${error}) and was moved to the dead letter queue. Resubmit it from the delivery log when the endpoint recovers.`,
      { type: "webhook", id: webhook.id },
      "INTEGRATION"
    );
  } catch (err) {
    log.warn({ err: (err as Error).message }, "dead-letter notification failed (non-fatal)");
  }
}

async function touchDelivery(
  prisma: PrismaClient,
  deliveryId: string | undefined,
  patch: Record<string, unknown>
): Promise<void> {
  if (!deliveryId) return;
  await prisma.webhookDelivery
    .update({ where: { id: deliveryId }, data: patch })
    .catch(() => undefined);
}

// ---------------------------------------------------------------------------
// Fan-out consumer (domain event → per-webhook dispatch jobs)
// ---------------------------------------------------------------------------

export function createWebhooksConsumer(opts: {
  redis: Redis;
  prisma: PrismaClient;
  consumerId: string;
}): StreamConsumer {
  const { redis, prisma, consumerId } = opts;
  return new StreamConsumer(
    redis,
    "factory:events",
    "worker-webhooks",
    consumerId,
    async (entry: StreamEntry) => {
      const parsed = tryParseEvent(entry);
      if (!parsed) return; // poison entry: drop + XACK (logged)
      const { envelope } = parsed;
      if (
        !WEBHOOK_EVENT_CATALOG.includes(
          envelope.event as (typeof WEBHOOK_EVENT_CATALOG)[number]
        )
      ) {
        return; // XACK: not a webhook-able event
      }
      const p = envelope.payload as Record<string, unknown>;
      const workspaceId = p.workspaceId ? String(p.workspaceId) : null;
      const webhooks = await prisma.webhook.findMany({
        where: {
          active: true,
          OR: [{ events: { has: "*" } }, { events: { has: envelope.event } }],
          ...(workspaceId ? { workspaceId } : {})
        },
        select: { id: true, events: true }
      });
      for (const wh of webhooks) {
        // Defense in depth: the Prisma `has` filter is exact-match; the
        // wildcard check lives here (and again in dispatchRow).
        if (!isWebhookSubscribed(wh.events, envelope.event)) continue;
        // The delivery row is created by the dispatcher (PENDING). The
        // stable `eventId` + `occurredAt` are minted ONCE here so every
        // retry carries the same `X-Factory-Event-Id` / envelope time.
        await enqueue(redis, QUEUES.webhooks, {
          kind: "row",
          webhookId: wh.id,
          event: envelope.event,
          payload: envelope.payload,
          eventId: generateId("evt"),
          occurredAt: new Date().toISOString()
        });
      }
      if (webhooks.length) {
        log.debug({ event: envelope.event, webhooks: webhooks.length }, "fan-out enqueued");
      }
    }
  );
}

// ---------------------------------------------------------------------------
// Final-failure hook (worker index.ts `failed` event — belt & braces:
// dispatchRow dead-letters at the cap itself; this catches unexpected job
// crashes so a row never stays PENDING forever).
// ---------------------------------------------------------------------------

export function handleWebhookFinalFailure(
  prisma: PrismaClient,
  connection: Redis,
  env: WorkerEnv
): (job: Job, err: Error) => Promise<void> {
  return async (job, err) => {
    let data: WebhookJobData;
    try {
      data = webhookJobSchema.parse(job.data);
    } catch {
      log.error({ jobId: job.id, err: err.message }, "webhook job failed (unparseable data)");
      return;
    }
    log.error(
      {
        jobId: job.id,
        webhookId: data.webhookId,
        event: data.event,
        err: err.message
      },
      "webhook job crashed → dead-lettering (unexpected error)"
    );
    try {
      await prisma.webhookDelivery
        .update({
          where: { id: data.deliveryId ?? "" },
          data: {
            status: "DEAD_LETTER",
            error: `crashed: ${err.message}`,
            nextRetryAt: null
          }
        })
        .catch(() => undefined);
      const webhook = await prisma.webhook.findUnique({
        where: { id: data.webhookId },
        select: { id: true, workspaceId: true }
      });
      if (webhook) {
        const key = NOTIFY_THROTTLE_KEY(webhook.id);
        const notified = await connection.set(key, "1", "EX", NOTIFY_THROTTLE_SECONDS, "NX");
        if (notified === "OK") {
          await notifyWorkspaceAdmins(
            prisma,
            connection,
            env.SMTP_HOST,
            webhook.workspaceId,
            "Webhook delivery failed",
            `A webhook delivery crashed unexpectedly (${err.message}) and was moved to the dead letter queue.`,
            { type: "webhook", id: webhook.id },
            "INTEGRATION"
          );
        }
      }
    } catch (inner) {
      log.error({ err: (inner as Error).message }, "webhook crash bookkeeping failed");
    }
  };
}