import { env } from "cloudflare:workers";
import { invalidateUnread } from "../user-inbox";
import { resolveMentions } from "../mentions";
import { isArtifactsError } from "../git/artifacts";
import { loadRepoById, repoReadableSql } from "../repos";
import { all, chunks, MAX_IN_LIST, placeholders } from "../db";
import { accountDb, repoDb } from "../scoped-db";
import { locate } from "../scoped-db/locators";
import { randomId } from "../crypto";
import { issueParticipants } from "../issues";
import { CAPS } from "../limits";
import { indexRepoCode } from "../code/search-index";
import { deliverEmail, emailTemplates, queueEmail } from "../email";
import { replyAddressFor } from "../email-reply";
import { expandRecipients, planDeliveries } from "../notification-prefs";
import { deliverDirect, deliverOnce, sendBudgetedEmail } from "../notify/deliver";
import { emailFooter } from "../notify/prefs";
import { publicUrl } from "./emit";
import type { NotifyMessage, QueueMessage } from "./emit";
import { triggerWorkflowsForEvent } from "../ci/event-triggers";
import { enqueueIntegrationDispatch } from "../integrations/queue";
import { publishRepoEvent } from "../mcp/events";
import { deliverWebhook } from "../webhooks/legacy";

const BASE_RETRY_SECONDS = 15;
const MAX_RETRY_SECONDS = 3600;

function hostOf(url: string): string | null {
  try {
    return url ? new URL(url).hostname.toLowerCase() : null;
  } catch {
    return null;
  }
}

/** Webhooks may not target the forge's own hostnames (or subdomains), so deliveries cannot reflect internal pages. */
const IPV4_LITERAL = /^\d{1,3}(\.\d{1,3}){3}$/;
const BLOCKED_SUFFIXES = [".localhost", ".local", ".internal", ".workers.dev"];

/**
 * Webhook targets must be public hosts: not this forge or its Sites domain, not IP literals
 * (loopback, private, link-local or metadata addresses), not single-label or internal names.
 */
/**
 * Local dev only: with INTEGRATIONS_ALLOW_LOOPBACK on a loopback forge, a loopback sink on a different port
 * than the forge is allowed so webhook delivery can be tested. Production never sets the flag.
 */
function devLoopbackSink(url: string): boolean {
  if (env.INTEGRATIONS_ALLOW_LOOPBACK !== "true") return false;
  try {
    const target = new URL(url);
    const forge = new URL(env.PUBLIC_URL);
    const loopback = (host: string) => host === "localhost" || host === "127.0.0.1" || host === "[::1]";
    return loopback(forge.hostname) && loopback(target.hostname) && target.port !== forge.port;
  } catch {
    return false;
  }
}

export function isForgeWebhookUrl(url: string): boolean {
  if (devLoopbackSink(url)) return false;
  const host = hostOf(url)?.replace(/\.$/, "");
  if (!host) return true;
  if (host.startsWith("[") || host.includes(":") || IPV4_LITERAL.test(host) || !host.includes(".")) return true;
  if (BLOCKED_SUFFIXES.some((suffix) => host.endsWith(suffix))) return true;
  const forgeHosts = [hostOf(env.PUBLIC_URL), env.SITES_DOMAIN ? env.SITES_DOMAIN.toLowerCase() : null, "localhost"];
  return forgeHosts.some((forge) => forge && (host === forge || host.endsWith(`.${forge}`)));
}

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" }>, eventId: string): Promise<void> {
  await triggerWorkflowsForEvent(message.repoId, message.event, message.payload).catch((error) =>
    console.error("Failed to trigger workflows for event", message.event, error),
  );
  await enqueueIntegrationDispatch(message.repoId, message.event, message.payload, eventId).catch((error) =>
    console.error("Failed to dispatch integrations for event", message.event, error),
  );
  const hooks = await repoDb(message.repoId).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 })));
}

/** Legacy repo webhooks are delivered through the guarded client in `webhooks/legacy.ts`. */
export { deliverWebhook };

async function deliver(message: Extract<QueueMessage, { kind: "webhook.deliver" }>, attempt: number): Promise<boolean> {
  const repoId = await locate("webhook", message.webhookId);
  if (!repoId) return true;
  const hook = await repoDb(repoId).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);
}

const MAX_NOTIFIED_MENTIONS = 50;

/** Former handles still reach the renamed user; orgs are skipped since only users get notified. */
async function mentionedUsers(text: string | null): Promise<string[]> {
  if (!text) return [];
  const accounts = [...new Set([...(await resolveMentions([text])).values()].map((target) => target.id))].slice(
    0,
    MAX_NOTIFIED_MENTIONS,
  );
  if (accounts.length === 0) return [];
  const rows = await all<{ account_id: string }>(
    `SELECT account_id FROM users WHERE account_id IN (${placeholders(accounts.length)})`,
    ...accounts,
  );
  return rows.map((row) => row.account_id);
}

async function notify(message: NotifyMessage, messageId: string): 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);
    }
  }
  for (const userId of await mentionedUsers(message.mentionText)) recipients.set(userId, "mention");
  await expandRecipients(message, recipients);
  if (message.actorId) recipients.delete(message.actorId);
  const capped = [...recipients].slice(0, CAPS.notificationRecipients);

  const now = Date.now();
  const deliveries = await planDeliveries(capped);
  const readers = await usersWhoCanRead(
    message.repoId,
    deliveries.map((delivery) => delivery.userId),
  );
  const visible = deliveries.filter((delivery) => readers.has(delivery.userId));
  const web = visible.filter((delivery) => delivery.web);
  const url = publicUrl(message.url);
  await Promise.all(
    web.map(({ userId, reason }) =>
      accountDb(userId).run(
        `INSERT OR IGNORE INTO notifications (id, user_id, repo_id, reason, title, url, created_at)
         VALUES (?, ?, ?, ?, ?, ?, ?)`,
        notificationId(messageId, userId),
        userId,
        message.repoId,
        reason,
        message.title,
        url,
        now,
      ),
    ),
  );
  await invalidateUnread(web.map((delivery) => delivery.userId));
  let failed = false;
  for (const { userId, reason, email } of visible) {
    if (!email) continue;
    const target = reason === "subscribed" ? "watching" : "participating";
    try {
      await deliverOnce(emailDeliveryId(messageId), userId, () =>
        sendBudgetedEmail(
          userId,
          email,
          async () =>
            queueEmail(
              email,
              emailTemplates.notification(message.title, reason, message.url, await emailFooter(userId, target), {
                excerpt: message.mentionText,
                replyTo: message.issueId ? await replyAddressFor(userId, message.issueId) : null,
              }),
            ),
          null,
        ),
      );
    } catch (error) {
      failed = true;
      console.error("Notification email failed for one recipient", error);
    }
  }
  if (failed) throw new Error("Some notification emails failed; retrying the rest");
}

/** Dedupe marker for a fan-out's emails; prefixed so it never collides with a direct notification's id. */
export function emailDeliveryId(messageId: string): string {
  return `notify-email:${messageId}`;
}

/** Stable per (queue message, user), so a retried fan-out re-inserts nothing it already wrote. */
export function notificationId(messageId: string, userId: string): string {
  return `${messageId}:${userId}`;
}

/** One IN query per chunk instead of a visibility lookup per recipient. */
async function usersWhoCanRead(repoId: string, userIds: string[]): Promise<Set<string>> {
  const readers = new Set<string>();
  for (const part of chunks([...new Set(userIds)], MAX_IN_LIST - 1)) {
    const rows = await all<{ id: string }>(
      `SELECT u.id FROM (SELECT value AS id FROM json_each(?2)) u JOIN repos r ON r.id = ?1
       WHERE ${repoReadableSql("r", "u.id")}`,
      repoId,
      JSON.stringify(part),
    );
    for (const row of rows) readers.add(row.id);
  }
  return readers;
}

/** True only when the message names a repository whose row is gone (deleted), never on other NOT_FOUNDs. */
async function repoGone(body: QueueMessage): Promise<boolean> {
  const repoId = "repoId" in body && typeof body.repoId === "string" ? body.repoId : null;
  return repoId !== null && (await loadRepoById(repoId)) === null;
}

/** 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, message.id);
        await publishRepoEvent(body.repoId, body.event, body.payload).catch((error: unknown) =>
          console.error("MCP event publish failed", error),
        );
      } else if (body.kind === "notify") await notify(body, message.id);
      else if (body.kind === "notify.direct") await deliverDirect(body);
      else if (body.kind === "code.index") await indexRepoCode(body);
      else if (body.kind === "email.send") await deliverEmail(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) {
      if (isArtifactsError(error) && error.code === "NOT_FOUND" && (await repoGone(message.body))) {
        console.warn("Queue message dropped: repository no longer exists", message.body.kind);
        message.ack();
        continue;
      }
      console.error("Queue message failed", error);
      message.retry({ delaySeconds: BASE_RETRY_SECONDS * message.attempts });
    }
  }
}