import { pruneIntegrationLeftovers } from "./retention";
import { env } from "cloudflare:workers";
import { all, chunks, first, placeholders, run } from "../db";
import { repoDb } from "../scoped-db";
import { sha256Hex } from "../crypto";
import { loadRepoById } from "../repos";
import { badRequest, notFoundError } from "../errors";
import { publicUrl } from "../events/emit";
import { deliverToIntegration } from "./deliver";
import type { DeliverInput } from "./deliver";
import { isPermanentFailure, isSuccess } from "./outbound";
import { integrationOwner, pruneUsageKeys, recordOutboundDelivery } from "./metering";
import { failImportJob, runImportPage } from "./github-import";
import type { ImportJobMessage } from "./github-import";
import { integrationsForRepo, loadIntegration, parseConfig, parseEvents, recordDelivery } from "./store";
import type { IntegrationRow } from "./store";
import { passesFilters, summarizeEvent } from "./summarize";
import type { EventSummary, RepoContext } from "./summarize";
import { notifyIntegrationDead } from "../notify/integrations";
import type { RepoRow } from "~/lib/types";

export const INTEGRATIONS_QUEUE = "gitfor-integrations";
export const INTEGRATIONS_DLQ = "gitfor-integrations-dlq";

const BASE_RETRY_SECONDS = 15;
const MAX_RETRY_SECONDS = 3600;
const MAX_QUEUED_PAYLOAD = 96 * 1024;
const MAX_PAYLOAD_COMMITS = 20;
const DIGEST_BATCH = 25;
const DIGEST_ITEMS = 80;
const MAX_DIGEST_BACKLOG = 500;
const MIRROR_BATCH = 10;
const ID_LENGTH = 20;
const DELIVERY_LEASE_MS = 2 * 60_000;
const MS_PER_SECOND = 1000;
/** wrangler.jsonc `max_retries: 6` for gitfor-integrations, i.e. 7 attempts. */
const MAX_ATTEMPTS = 7;
const MIRROR_LEASE_MS = 20 * 60_000;
const MAX_BATCH_BYTES = 200 * 1024;
const MAX_BATCH_MESSAGES = 50;
const MIRROR_INTERVAL_MS = 15 * 60_000;

export type IntegrationMessage =
  | {
      kind: "integration.dispatch";
      repoId: string;
      event: string;
      payload: Record<string, unknown>;
      /** Stable across retries of the source event, so each (event, integration) pair delivers once. */
      eventId?: string;
    }
  | ({ kind: "integration.deliver"; integrationId: string } & DeliverInput)
  | ImportJobMessage;

/** Fans a forge event out to integrations. Called from the events consumer for every dispatched webhook. */
export async function enqueueIntegrationDispatch(
  repoId: string,
  event: string,
  payload: Record<string, unknown>,
  eventId?: string,
): Promise<void> {
  const any = await first<{ ok: number }>(
    `SELECT 1 AS ok FROM repos r JOIN integrations i ON i.active = 1
       AND ((i.scope_type = 'repo' AND i.scope_id = r.id) OR (i.scope_type = 'org' AND i.scope_id = r.owner_id))
     WHERE r.id = ? LIMIT 1`,
    repoId,
  );
  if (!any) return;
  const message: IntegrationMessage = {
    kind: "integration.dispatch",
    repoId,
    event,
    payload: boundPayload(payload),
    eventId,
  };
  await env.INTEGRATIONS.send(message);
}

/** Queue messages are capped at 128 KB, so long push payloads lose commits past the first few. */
function boundPayload(payload: Record<string, unknown>): Record<string, unknown> {
  if (JSON.stringify(payload).length <= MAX_QUEUED_PAYLOAD) return payload;
  const commits = Array.isArray(payload.commits) ? payload.commits.slice(0, MAX_PAYLOAD_COMMITS) : undefined;
  const trimmed = { ...payload, ...(commits ? { commits } : {}) };
  if (JSON.stringify(trimmed).length <= MAX_QUEUED_PAYLOAD) return trimmed;
  return { action: payload.action, ref: payload.ref, sender: payload.sender, truncated: true };
}

export function repoContext(repo: RepoRow): RepoContext {
  return {
    fullName: `${repo.owner_name}/${repo.name}`,
    url: publicUrl(`/${repo.owner_name}/${repo.name}`),
    defaultBranch: repo.default_branch,
  };
}

type PullDetails = { title: string; body: string; head_ref: string; base_ref: string; draft: number };

/** Merge events only carry a number; fill in title and branches so chat messages and Linear keys work. */
async function enrichPull(repo: RepoRow, payload: Record<string, unknown>): Promise<Record<string, unknown>> {
  const pull = (payload.pull_request ?? {}) as Record<string, unknown>;
  if (typeof pull.title === "string") return payload;
  const number = Number(pull.number ?? payload.number);
  if (!Number.isInteger(number)) return payload;
  const details = await repoDb(repo.id).first<PullDetails>(
    `SELECT i.title, i.body, p.head_ref, p.base_ref, p.draft FROM issues i JOIN pull_requests p ON p.issue_id = i.id
     WHERE i.repo_id = ? AND i.number = ?`,
    repo.id,
    number,
  );
  if (!details) return payload;
  return {
    ...payload,
    pull_request: {
      ...pull,
      title: details.title,
      body: details.body,
      head: { ref: details.head_ref },
      base: { ref: details.base_ref },
      draft: details.draft === 1,
    },
  };
}

function wantsRawEvent(row: IntegrationRow, event: string): boolean {
  if (row.kind === "linear") return event === "pull_request" || event === "push";
  if (row.kind === "github") return event === "push";
  return false;
}

/** Stable per (queue message, integration) so a retried dispatch reuses rows instead of duplicating them. */
export async function stableId(messageId: string, integrationId: string): Promise<string> {
  return (await sha256Hex(`${messageId}:${integrationId}`)).slice(0, ID_LENGTH);
}

/** sendBatch is capped by count and total bytes, so deliveries go out in size-bounded chunks. */
async function sendChunked(messages: IntegrationMessage[]): Promise<void> {
  let chunk: IntegrationMessage[] = [];
  let bytes = 0;
  for (const message of messages) {
    const size = JSON.stringify(message).length;
    if (chunk.length > 0 && (bytes + size > MAX_BATCH_BYTES || chunk.length >= MAX_BATCH_MESSAGES)) {
      await env.INTEGRATIONS.sendBatch(chunk.map((body) => ({ body })));
      chunk = [];
      bytes = 0;
    }
    chunk.push(message);
    bytes += size;
  }
  if (chunk.length > 0) await env.INTEGRATIONS.sendBatch(chunk.map((body) => ({ body })));
}

type DispatchContext = {
  repo: RepoRow;
  event: string;
  payload: Record<string, unknown>;
  summary: EventSummary | null;
  messageId: string;
};

/** Plans one integration's share of an event. Re-running it for a retried dispatch never resends finished work. */
async function planRow(row: IntegrationRow, context: DispatchContext): Promise<IntegrationMessage | null> {
  const { repo, event, payload, summary } = context;
  if (row.kind === "linear" || row.kind === "github") {
    if (!wantsRawEvent(row, event)) return null;
  } else {
    if (!summary) return null;
    if (!passesFilters(summary, parseEvents(row), parseConfig(row).filters, repo.default_branch)) return null;
  }
  const id = await stableId(context.messageId, row.id);
  if (summary && row.kind !== "linear" && row.kind !== "github" && row.delivery_mode === "digest") {
    const backlog = await first<{ count: number }>(
      "SELECT COUNT(*) AS count FROM integration_digest_items WHERE integration_id = ?",
      row.id,
    );
    if ((backlog?.count ?? 0) >= MAX_DIGEST_BACKLOG) return null;
    await run(
      "INSERT OR IGNORE INTO integration_digest_items (id, integration_id, repo_id, event, payload, created_at) VALUES (?, ?, ?, ?, ?, ?)",
      id,
      row.id,
      repo.id,
      event,
      JSON.stringify(summary),
      Date.now(),
    );
    return null;
  }
  const existing = await first<{ status: string }>("SELECT status FROM integration_deliveries WHERE id = ?", id);
  if (existing && existing.status !== "pending") return null;
  if (!existing) {
    await recordDelivery({
      id,
      integrationId: row.id,
      event,
      repoId: repo.id,
      status: "pending",
      payload: { event, payload, summary },
    });
  }
  const raw = row.kind === "webhook" || row.kind === "linear" || row.kind === "github";
  return {
    kind: "integration.deliver",
    integrationId: row.id,
    deliveryId: id,
    repoId: repo.id,
    event,
    payload: raw ? payload : {},
    summary,
  };
}

async function dispatch(
  message: Extract<IntegrationMessage, { kind: "integration.dispatch" }>,
  messageId: string,
): Promise<void> {
  const repo = await loadRepoById(message.repoId);
  if (!repo) return;
  const rows = await integrationsForRepo(repo);
  if (rows.length === 0) return;
  const payload = message.event === "pull_request" ? await enrichPull(repo, message.payload) : message.payload;
  const summary = summarizeEvent(message.event, payload, repoContext(repo));
  const context: DispatchContext = { repo, event: message.event, payload, summary, messageId };
  const deliveries: IntegrationMessage[] = [];
  let failed: unknown = null;
  for (const row of rows) {
    try {
      const planned = await planRow(row, context);
      if (planned) deliveries.push(planned);
    } catch (error) {
      console.error("Integration dispatch failed for", row.id, error);
      failed = error;
    }
  }
  await sendChunked(deliveries);
  if (failed) throw failed;
}

/**
 * Claims the delivery for this queue message (Queues keep `message.id` stable across retries). The owner
 * keeps its claim through retry backoff, so a duplicate only waits and retries until the delivery is terminal
 * or the owner's lease lapses (e.g. after a crash), and then takes over.
 * Terminal deliveries (success, permanent failure, paused) are never sent again.
 */
async function deliver(
  message: Extract<IntegrationMessage, { kind: "integration.deliver" }>,
  attempt: number,
  messageId: string,
  retryHoldMs: number,
): Promise<"done" | "retry"> {
  const row = await loadIntegration(message.integrationId);
  if (!row) return "done";
  const now = Date.now();
  const leaseMs = row.kind === "github" ? MIRROR_LEASE_MS : DELIVERY_LEASE_MS;
  const leaseUntil = now + leaseMs;
  const lease = await run(
    `UPDATE integration_deliveries SET lease_until = ?, lease_owner = ? WHERE id = ? AND terminal = 0
     AND status IN ('pending', 'failed') AND (lease_until IS NULL OR lease_until < ? OR lease_owner = ?)`,
    leaseUntil,
    messageId,
    message.deliveryId,
    now,
    messageId,
  );
  if (!lease.meta.changes) {
    const current = await first<{ terminal: number; status: string }>(
      "SELECT terminal, status FROM integration_deliveries WHERE id = ?",
      message.deliveryId,
    );
    return !current || current.terminal === 1 || current.status === "success" ? "done" : "retry";
  }
  let outcome: "done" | "retry" = "retry";
  try {
    outcome = await deliverLeased(row, message, attempt);
    return outcome;
  } finally {
    if (outcome === "done") {
      await run(
        "UPDATE integration_deliveries SET lease_until = NULL, lease_owner = NULL WHERE id = ? AND lease_owner = ?",
        message.deliveryId,
        messageId,
      );
    } else {
      await run(
        "UPDATE integration_deliveries SET lease_until = ? WHERE id = ? AND lease_owner = ? AND lease_until = ?",
        Date.now() + retryHoldMs + leaseMs,
        message.deliveryId,
        messageId,
        leaseUntil,
      );
    }
  }
}

async function deliverLeased(
  row: IntegrationRow,
  message: Extract<IntegrationMessage, { kind: "integration.deliver" }>,
  attempt: number,
): Promise<"done" | "retry"> {
  if (row.active !== 1) {
    await run(
      "UPDATE integration_deliveries SET status = 'failed', error = 'Integration is paused', terminal = 1, updated_at = ? WHERE id = ?",
      Date.now(),
      message.deliveryId,
    );
    return "done";
  }
  const result = await deliverToIntegration(row, message);
  const ok = isSuccess(result);
  const lastAttempt = attempt >= MAX_ATTEMPTS;
  await recordDelivery({
    id: message.deliveryId,
    integrationId: row.id,
    event: message.event,
    repoId: message.repoId,
    status: ok ? "success" : lastAttempt && !isPermanentFailure(result) ? "dead" : "failed",
    terminal: ok || isPermanentFailure(result) || lastAttempt,
    statusCode: result.status,
    error: result.error,
    payload: {
      event: message.event,
      payload: message.payload,
      summary: message.summary,
      digest: message.digest,
      digestLabel: message.digestLabel,
    },
    responseBody: result.body,
    durationMs: result.durationMs,
    attempt,
  });
  const ownerId = await integrationOwner(row.scope_type, row.scope_id);
  const final = ok || isPermanentFailure(result) || lastAttempt;
  if (ownerId && final) {
    const sent = JSON.stringify(message.summary ?? message.digest ?? message.payload ?? null);
    const bytes = new TextEncoder().encode(sent).byteLength;
    await recordOutboundDelivery({
      ownerId,
      repoId: message.repoId,
      kind: row.kind,
      ok,
      bytes,
      deliveryId: message.deliveryId,
    });
  }
  return ok || isPermanentFailure(result) || lastAttempt ? "done" : "retry";
}

/** Retries with exponential backoff; after `max_retries` Cloudflare moves the message to the dead-letter queue. */
export async function handleIntegrationBatch(batch: MessageBatch<IntegrationMessage>): Promise<void> {
  if (batch.queue === INTEGRATIONS_DLQ) return handleDeadLetters(batch);
  for (const message of batch.messages) {
    const delaySeconds = Math.min(BASE_RETRY_SECONDS * 2 ** (message.attempts - 1), MAX_RETRY_SECONDS);
    try {
      const body = message.body;
      if (body.kind === "integration.dispatch") await dispatch(body, body.eventId ?? message.id);
      else if (body.kind === "integration.job") await runImportPage(body);
      else if (body.kind === "integration.deliver") {
        if ((await deliver(body, message.attempts, message.id, delaySeconds * MS_PER_SECOND)) === "retry") {
          message.retry({ delaySeconds });
          continue;
        }
      }
      message.ack();
    } catch (error) {
      console.error("Integration message failed", error);
      message.retry({ delaySeconds });
    }
  }
}

async function handleDeadLetters(batch: MessageBatch<IntegrationMessage>): Promise<void> {
  for (const message of batch.messages) {
    const body = message.body;
    if (body.kind === "integration.deliver") {
      await run(
        `UPDATE integration_deliveries SET status = 'dead', terminal = 1, updated_at = ?1 WHERE id = ?2
         AND status != 'success' AND (lease_until IS NULL OR lease_until < ?1)`,
        Date.now(),
        body.deliveryId,
      );
      await notifyIntegrationDead({ deliveryId: body.deliveryId }, "A delivery");
    } else if (body.kind === "integration.job") {
      await failImportJob(body.jobId, "Gave up after repeated failures");
      await notifyIntegrationDead({ jobId: body.jobId }, "An import job");
    } else {
      console.error("Dropped integration dispatch after retries", body.event, body.repoId);
    }
    message.ack();
  }
}

/** Enqueues an existing delivery again (new attempt counter, same stored payload). */
export async function redeliver(row: IntegrationRow, deliveryId: string): Promise<void> {
  const stored = await first<{ event: string; repo_id: string | null; payload: string }>(
    "SELECT event, repo_id, payload FROM integration_deliveries WHERE id = ? AND integration_id = ? AND direction = 'outbound'",
    deliveryId,
    row.id,
  );
  if (!stored) throw notFoundError("Delivery not found");
  const parsed = JSON.parse(stored.payload) as {
    truncated?: boolean;
    payload?: Record<string, unknown>;
    summary?: EventSummary | null;
    digest?: EventSummary[];
    digestLabel?: string;
  };
  if (parsed.truncated || parsed.payload?.truncated)
    throw badRequest("This delivery was too large to keep and cannot be redelivered");
  const id = await recordDelivery({
    integrationId: row.id,
    event: stored.event,
    repoId: stored.repo_id,
    status: "pending",
    payload: parsed,
  });
  const message: IntegrationMessage = {
    kind: "integration.deliver",
    integrationId: row.id,
    deliveryId: id,
    repoId: stored.repo_id,
    event: stored.event,
    payload: parsed.payload ?? {},
    summary: parsed.summary ?? null,
    digest: parsed.digest,
    digestLabel: parsed.digestLabel,
  };
  await env.INTEGRATIONS.send(message);
}

/** Queues one delivery directly, bypassing filters (test events, digests, mirror ticks). */
export async function enqueueDirect(row: IntegrationRow, input: Omit<DeliverInput, "deliveryId">): Promise<string> {
  const deliveryId = await recordDelivery({
    integrationId: row.id,
    event: input.event,
    repoId: input.repoId,
    status: "pending",
    payload: {
      event: input.event,
      payload: input.payload,
      summary: input.summary,
      digest: input.digest,
      digestLabel: input.digestLabel,
    },
  });
  const message: IntegrationMessage = { kind: "integration.deliver", integrationId: row.id, deliveryId, ...input };
  await env.INTEGRATIONS.send(message);
  return deliveryId;
}

/** Cron: flushes due digests into a single message per integration, and ticks GitHub pull mirrors. */
export async function runIntegrationsCron(now = Date.now()): Promise<void> {
  await pruneUsageKeys(now);
  const due = await all<IntegrationRow>(
    `SELECT * FROM integrations WHERE active = 1 AND delivery_mode = 'digest' AND last_digest_at + digest_minutes * 60000 <= ?
     AND EXISTS (SELECT 1 FROM integration_digest_items d WHERE d.integration_id = integrations.id) LIMIT ?`,
    now,
    DIGEST_BATCH,
  );
  for (const row of due) {
    const claimed = await run(
      "UPDATE integrations SET last_digest_at = ? WHERE id = ? AND last_digest_at = ?",
      now,
      row.id,
      row.last_digest_at,
    );
    if (!claimed.meta.changes) continue;
    const items = await all<{ id: string; payload: string }>(
      "SELECT id, payload FROM integration_digest_items WHERE integration_id = ? ORDER BY created_at LIMIT ?",
      row.id,
      DIGEST_ITEMS,
    );
    const digest = items.map((item) => JSON.parse(item.payload) as EventSummary);
    await enqueueDirect(row, {
      repoId: null,
      event: "digest",
      payload: {},
      summary: null,
      digest,
      digestLabel: row.name,
    });
    for (const part of chunks(items.map((item) => item.id))) {
      await run(`DELETE FROM integration_digest_items WHERE id IN (${placeholders(part.length)})`, ...part);
    }
  }
  const mirrors = await all<IntegrationRow>(
    `SELECT i.* FROM integrations i JOIN repos r ON r.id = i.scope_id
     WHERE i.active = 1 AND i.kind = 'github' AND i.scope_type = 'repo'
     AND json_extract(i.config, '$.githubMirror') = 'pull' AND i.last_digest_at <= ? LIMIT ?`,
    now - MIRROR_INTERVAL_MS,
    MIRROR_BATCH,
  );
  for (const row of mirrors) {
    await run("UPDATE integrations SET last_digest_at = ? WHERE id = ?", now, row.id);
    await enqueueDirect(row, { repoId: row.scope_id, event: "mirror.pull", payload: {}, summary: null });
  }
  await pruneIntegrationLeftovers(now);
}