import { env } from "cloudflare:workers";
import { all, first, run } from "../db";
import { hmacSha256Hex, randomId } from "../crypto";
import { issueParticipants } from "../issues";
import { CAPS } from "../limits";
import type { NotifyMessage, QueueMessage, WebhookEvent } from "./emit";

const DELIVERY_TIMEOUT_MS = 10_000;
const MAX_RESPONSE_BODY = 4096;
const BASE_RETRY_SECONDS = 15;
const MAX_RETRY_SECONDS = 3600;
const HTTP_OK_MIN = 200;
const HTTP_OK_MAX = 300;
const MENTION = /(?:^|[^\w@])@([A-Za-z0-9](?:[A-Za-z0-9-]{0,38}))/g;

type WebhookRow = { id: string; repo_id: string; url: string; secret: string | null; events: string; active: number };

async function dispatch(message: Extract<QueueMessage, { kind: "webhook.dispatch" }>): Promise<void> {
  const hooks = await all<WebhookRow>("SELECT * FROM webhooks WHERE repo_id = ? AND active = 1", message.repoId);
  const deliveries: QueueMessage[] = [];
  for (const hook of hooks) {
    const events: string[] = JSON.parse(hook.events);
    if (!events.includes("*") && !events.includes(message.event)) continue;
    deliveries.push({
      kind: "webhook.deliver",
      webhookId: hook.id,
      event: message.event,
      deliveryId: randomId(),
      payload: message.payload,
    });
  }
  if (deliveries.length > 0) await env.EVENTS.sendBatch(deliveries.map((body) => ({ body })));
}

async function readPrefix(response: Response, limit: number): Promise<string> {
  if (!response.body) return "";
  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let text = "";
  while (text.length < limit) {
    const { value, done } = await reader.read();
    if (done) break;
    text += decoder.decode(value, { stream: true });
  }
  await reader.cancel().catch(() => undefined);
  return text.slice(0, limit);
}

export async function deliverWebhook(
  hook: Pick<WebhookRow, "id" | "url" | "secret">,
  event: WebhookEvent,
  deliveryId: string,
  payload: Record<string, unknown>,
  attempt: number,
): Promise<boolean> {
  const body = JSON.stringify(payload);
  const headers: Record<string, string> = {
    "content-type": "application/json",
    "user-agent": "stig-hookshot",
    "x-stig-event": event,
    "x-stig-delivery": deliveryId,
    "x-github-event": event,
    "x-github-delivery": deliveryId,
  };
  if (hook.secret) {
    const signature = `sha256=${await hmacSha256Hex(hook.secret, body)}`;
    headers["x-stig-signature-256"] = signature;
    headers["x-hub-signature-256"] = signature;
  }
  const started = Date.now();
  let status: number | null = null;
  let responseBody: string | null = null;
  let error: string | null = null;
  try {
    const response = await fetch(hook.url, {
      method: "POST",
      headers,
      body,
      signal: AbortSignal.timeout(DELIVERY_TIMEOUT_MS),
    });
    status = response.status;
    responseBody = await readPrefix(response, MAX_RESPONSE_BODY);
  } catch (caught) {
    error = caught instanceof Error ? caught.message : String(caught);
  }
  await run(
    `INSERT INTO webhook_deliveries (id, webhook_id, event, status_code, error, request_body, response_body, duration_ms, attempt, created_at)
     VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
    randomId(),
    hook.id,
    event,
    status,
    error,
    body,
    responseBody,
    Date.now() - started,
    attempt,
    Date.now(),
  );
  return status !== null && status >= HTTP_OK_MIN && status < HTTP_OK_MAX;
}

async function deliver(message: Extract<QueueMessage, { kind: "webhook.deliver" }>, attempt: number): Promise<boolean> {
  const hook = await first<WebhookRow>("SELECT * FROM webhooks WHERE id = ? AND active = 1", message.webhookId);
  if (!hook) return true;
  return deliverWebhook(hook, message.event, message.deliveryId, message.payload, attempt);
}

async function mentionedUsers(text: string | null): Promise<string[]> {
  if (!text) return [];
  const names = [...new Set([...text.matchAll(MENTION)].map((match) => match[1].toLowerCase()))].slice(0, 50);
  if (names.length === 0) return [];
  const rows = await all<{ id: string }>(
    `SELECT a.id FROM accounts a JOIN users u ON u.account_id = a.id WHERE a.name IN (${names.map(() => "?").join(", ")})`,
    ...names,
  );
  return rows.map((row) => row.id);
}

async function notify(message: NotifyMessage): Promise<void> {
  const recipients = new Map<string, string>();
  for (const userId of message.directRecipients) recipients.set(userId, message.reason);
  if (message.issueId) {
    for (const userId of await issueParticipants(message.issueId)) {
      if (!recipients.has(userId)) recipients.set(userId, message.reason);
    }
  }
  const watchers = await all<{ user_id: string }>("SELECT user_id FROM watches WHERE repo_id = ?", message.repoId);
  for (const watcher of watchers) {
    if (!recipients.has(watcher.user_id)) recipients.set(watcher.user_id, "subscribed");
  }
  for (const userId of await mentionedUsers(message.mentionText)) recipients.set(userId, "mention");
  if (message.actorId) recipients.delete(message.actorId);
  const capped = [...recipients].slice(0, CAPS.notificationRecipients);

  const now = Date.now();
  for (const [userId, reason] of capped) {
    const canSee = await first<{ ok: number }>(
      `SELECT 1 AS ok FROM repos r WHERE r.id = ?1 AND (r.visibility = 'public' OR r.owner_id = ?2
        OR EXISTS (SELECT 1 FROM collaborators c WHERE c.repo_id = r.id AND c.user_id = ?2)
        OR EXISTS (SELECT 1 FROM org_members m WHERE m.org_id = r.owner_id AND m.user_id = ?2))`,
      message.repoId,
      userId,
    );
    if (!canSee) continue;
    await run(
      "INSERT INTO notifications (id, user_id, repo_id, reason, title, url, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
      randomId(),
      userId,
      message.repoId,
      reason,
      message.title,
      message.url,
      now,
    );
  }
}

/** Queue consumer for webhook fan-out/delivery and notification fan-out, with exponential retry. */
export async function handleQueueBatch(batchOfMessages: MessageBatch<QueueMessage>): Promise<void> {
  for (const message of batchOfMessages.messages) {
    try {
      const body = message.body;
      if (body.kind === "webhook.dispatch") await dispatch(body);
      else if (body.kind === "notify") await notify(body);
      else if (body.kind === "webhook.deliver") {
        const ok = await deliver(body, message.attempts);
        if (!ok) {
          message.retry({
            delaySeconds: Math.min(BASE_RETRY_SECONDS * 2 ** (message.attempts - 1), MAX_RETRY_SECONDS),
          });
          continue;
        }
      }
      message.ack();
    } catch (error) {
      console.error("Queue message failed", error);
      message.retry({ delaySeconds: BASE_RETRY_SECONDS * message.attempts });
    }
  }
}