import { env } from "cloudflare:workers";
import { all, chunks, first, placeholders, run } from "../db";
import { randomId } from "../crypto";
import { loadRepoById } from "../repos";
import { publicUrl } from "../events/emit";
import { deliverToIntegration } from "./deliver";
import type { DeliverInput } from "./deliver";
import { isSuccess } from "./outbound";
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 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 MIRROR_INTERVAL_MS = 15 * 60_000;

export type IntegrationMessage =
  | { kind: "integration.dispatch"; repoId: string; event: string; payload: Record<string, unknown> }
  | ({ 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>,
): 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) };
  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 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;
}

async function dispatch(message: Extract<IntegrationMessage, { kind: "integration.dispatch" }>): 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 deliveries: IntegrationMessage[] = [];
  for (const row of rows) {
    if (row.kind === "linear" || row.kind === "github") {
      if (!wantsRawEvent(row, message.event)) continue;
    } else {
      if (!summary) continue;
      if (!passesFilters(summary, parseEvents(row), parseConfig(row).filters, repo.default_branch)) continue;
      if (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) continue;
        await run(
          "INSERT INTO integration_digest_items (id, integration_id, repo_id, event, payload, created_at) VALUES (?, ?, ?, ?, ?, ?)",
          randomId(),
          row.id,
          repo.id,
          message.event,
          JSON.stringify(summary),
          Date.now(),
        );
        continue;
      }
    }
    const raw = row.kind === "webhook" || row.kind === "linear" || row.kind === "github";
    const deliveryId = await recordDelivery({
      integrationId: row.id,
      event: message.event,
      repoId: repo.id,
      status: "pending",
      payload: { event: message.event, payload, summary },
    });
    deliveries.push({
      kind: "integration.deliver",
      integrationId: row.id,
      deliveryId,
      repoId: repo.id,
      event: message.event,
      payload: raw ? payload : {},
      summary,
    });
  }
  if (deliveries.length > 0) await env.INTEGRATIONS.sendBatch(deliveries.map((body) => ({ body })));
}

async function deliver(
  message: Extract<IntegrationMessage, { kind: "integration.deliver" }>,
  attempt: number,
): Promise<boolean> {
  const row = await loadIntegration(message.integrationId);
  if (!row || row.active !== 1) return true;
  const result = await deliverToIntegration(row, message);
  const ok = isSuccess(result);
  await recordDelivery({
    id: message.deliveryId,
    integrationId: row.id,
    event: message.event,
    repoId: message.repoId,
    status: ok ? "success" : "failed",
    statusCode: result.status,
    error: result.error,
    payload: { event: message.event, payload: message.payload, summary: message.summary, digest: message.digest },
    responseBody: result.body,
    durationMs: result.durationMs,
    attempt,
  });
  return ok;
}

/** 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);
      else if (body.kind === "integration.job") await runImportPage(body);
      else if (body.kind === "integration.deliver" && !(await deliver(body, message.attempts))) {
        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', updated_at = ? WHERE id = ?",
        Date.now(),
        body.deliveryId,
      );
    } else if (body.kind === "integration.job") {
      await failImportJob(body.jobId, "Gave up after repeated failures");
    } 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 = ?",
    deliveryId,
    row.id,
  );
  if (!stored) return;
  const parsed = JSON.parse(stored.payload) as {
    payload?: Record<string, unknown>;
    summary?: EventSummary | null;
    digest?: EventSummary[];
  };
  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,
  };
  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 },
  });
  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> {
  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 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);
    }
    await run("UPDATE integrations SET last_digest_at = ? WHERE id = ?", now, row.id);
  }
  const mirrors = await all<IntegrationRow>(
    `SELECT * FROM integrations WHERE active = 1 AND kind = 'github' AND scope_type = 'repo'
     AND json_extract(config, '$.githubMirror') = 'pull' AND 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 });
  }
}