import { env } from "cloudflare:workers";
import { repoDb, type ScopedDb } from "../scoped-db";
import { locate } from "../scoped-db/locators";
import { claimHost, siteDb, siteDbById } from "./store";
import { SITE_LIMITS, slotKey } from "./config";
import { commitLabel, pullLabel } from "./naming";
import { isOrphanSite } from "./sync";
import type { SiteDeploymentRow, SiteRow, SlotPointer } from "./config";

type AliasSite = Pick<SiteRow, "id" | "repo_id" | "label" | "commit_previews">;
type AliasDeployment = Pick<SiteDeploymentRow, "id" | "environment" | "branch" | "sha">;

type OpenPull = { number: number };

/** Open pull requests of the site's own repository whose head is `branch`; fork pulls never get an alias. */
async function openPullsFor(site: AliasSite, branch: string): Promise<OpenPull[]> {
  if (!site.repo_id) return [];
  return repoDb(site.repo_id).all<OpenPull>(
    `SELECT i.number FROM pull_requests p JOIN issues i ON i.id = p.issue_id
     WHERE p.repo_id = ? AND p.head_repo_id = ? AND p.head_ref = ? AND i.state = 'open' LIMIT ?`,
    site.repo_id,
    site.repo_id,
    branch,
    SITE_LIMITS.previewsPerPush,
  );
}

/** Writes an alias pointer unless another live site already owns that host. */
async function point(site: AliasSite, label: string, deploymentId: string): Promise<boolean> {
  const existing = await env.KV.get<SlotPointer>(slotKey(label), "json");
  if (existing && existing.siteId !== site.id && !(await isOrphanSite(existing.siteId))) return false;
  const pointer: SlotPointer = { siteId: site.id, deploymentId };
  await env.KV.put(slotKey(label), JSON.stringify(pointer));
  await claimHost(label, site.id);
  return true;
}

async function record(
  site: AliasSite,
  label: string,
  number: number,
  kind: "pr" | "commit",
  deploymentId: string,
): Promise<void> {
  await siteDb(site).run(
    `INSERT INTO site_pr_aliases (site_id, label, pr_number, kind, deployment_id, created_at) VALUES (?, ?, ?, ?, ?, ?)
     ON CONFLICT (site_id, label, pr_number) DO UPDATE SET deployment_id = excluded.deployment_id,
       created_at = excluded.created_at`,
    site.id,
    label,
    number,
    kind,
    deploymentId,
    Date.now(),
  );
}

/** Keeps the newest commit previews of a pull request and drops the hosts of the rest. */
async function pruneCommitPreviews(site: AliasSite, number: number): Promise<void> {
  const db = siteDb(site);
  const stale = await db.all<{ label: string }>(
    `SELECT label FROM site_pr_aliases WHERE site_id = ? AND pr_number = ? AND kind = 'commit'
     ORDER BY created_at DESC, rowid DESC LIMIT -1 OFFSET ?`,
    site.id,
    number,
    SITE_LIMITS.commitPreviewsPerPull,
  );
  for (const row of stale) {
    await db.run(
      "DELETE FROM site_pr_aliases WHERE site_id = ? AND label = ? AND pr_number = ?",
      site.id,
      row.label,
      number,
    );
  }
  await releaseLabels(
    db,
    site.id,
    stale.map((row) => row.label),
  );
}

/** Deletes the hosts of labels no alias row of the site still claims (another pull request may share a commit). */
async function releaseLabels(db: ScopedDb, siteId: string, labels: string[]): Promise<void> {
  for (const label of new Set(labels)) {
    const held = await db.first<{ one: number }>(
      "SELECT 1 AS one FROM site_pr_aliases WHERE site_id = ? AND label = ? LIMIT 1",
      siteId,
      label,
    );
    if (!held) await env.KV.delete(slotKey(label));
  }
}

/** After a branch preview goes live: moves each open same-repo pull request's alias (and commit preview) onto it. */
export async function syncPullAliases(site: AliasSite, deployment: AliasDeployment): Promise<void> {
  if (deployment.environment !== "preview" || !deployment.branch) return;
  for (const pull of await openPullsFor(site, deployment.branch)) {
    const label = pullLabel(site.label, pull.number);
    if (await point(site, label, deployment.id)) await record(site, label, pull.number, "pr", deployment.id);
    if (site.commit_previews !== 1) continue;
    const commit = commitLabel(site.label, deployment.sha);
    if (await point(site, commit, deployment.id)) await record(site, commit, pull.number, "commit", deployment.id);
    await pruneCommitPreviews(site, pull.number);
  }
}

/** Every host a site's pull request aliases serve, for teardown. */
export async function aliasLabels(siteId: string): Promise<string[]> {
  const db = await siteDbById(siteId);
  if (!db) return [];
  const rows = await db.all<{ label: string }>(
    "SELECT DISTINCT label FROM site_pr_aliases WHERE site_id = ? LIMIT 1000",
    siteId,
  );
  return rows.map((row) => row.label);
}

/** Drops every alias host of a site (relabel or unpublish); the rows go too. */
export async function dropSiteAliases(siteId: string): Promise<void> {
  const labels = await aliasLabels(siteId);
  await Promise.all(labels.map((label) => env.KV.delete(slotKey(label))));
  await (await siteDbById(siteId))?.run("DELETE FROM site_pr_aliases WHERE site_id = ?", siteId);
}

async function dropPull(db: ScopedDb, siteId: string, number: number): Promise<void> {
  const rows = await db.all<{ label: string }>(
    "SELECT label FROM site_pr_aliases WHERE site_id = ? AND pr_number = ?",
    siteId,
    number,
  );
  await db.run("DELETE FROM site_pr_aliases WHERE site_id = ? AND pr_number = ?", siteId, number);
  await releaseLabels(
    db,
    siteId,
    rows.map((row) => row.label),
  );
}

/** When a branch preview host is removed: the pull request and commit aliases serving its deployments go too. */
export async function dropAliasesForPreview(siteId: string, previewLabel: string): Promise<void> {
  const db = await siteDbById(siteId);
  if (!db) return;
  const rows = await db.all<{ label: string }>(
    `SELECT label FROM site_pr_aliases WHERE site_id = ?1
     AND deployment_id IN (SELECT id FROM site_deployments WHERE site_id = ?1 AND label = ?2)`,
    siteId,
    previewLabel,
  );
  await db.run(
    `DELETE FROM site_pr_aliases WHERE site_id = ?1
     AND deployment_id IN (SELECT id FROM site_deployments WHERE site_id = ?1 AND label = ?2)`,
    siteId,
    previewLabel,
  );
  await releaseLabels(
    db,
    siteId,
    rows.map((row) => row.label),
  );
}

/** When a pull request closes or merges: its `pr-N` and commit hosts stop serving. */
export async function dropPullAliasesForIssue(issueId: string): Promise<void> {
  const repoId = await locate("issue", issueId);
  if (!repoId) return;
  const db = repoDb(repoId);
  const row = await db.first<{ site_id: string; number: number }>(
    `SELECT s.id AS site_id, i.number FROM issues i JOIN sites s ON s.repo_id = i.repo_id WHERE i.id = ?`,
    issueId,
  );
  if (row) await dropPull(db, row.site_id, row.number);
}

const SWEEP_BATCH = 50;

/** Aliases whose pull request is no longer open: the close hook drops them, the scope's alarm is the safety net. */
export const STALE_PULL_ALIASES = `site_pr_aliases a JOIN sites s ON s.id = a.site_id
  LEFT JOIN issues i ON i.repo_id = s.repo_id AND i.number = a.pr_number
  WHERE i.id IS NULL OR i.state != 'open'`;

/** RepoState alarm: drops a batch of the scope's stale pull aliases; true when there may be more. */
export async function sweepClosedPullAliasesForScope(scopeId: string): Promise<boolean> {
  const db = repoDb(scopeId);
  const rows = await db.all<{ site_id: string; pr_number: number }>(
    `SELECT DISTINCT a.site_id, a.pr_number FROM ${STALE_PULL_ALIASES} LIMIT ?`,
    SWEEP_BATCH,
  );
  for (const row of rows) await dropPull(db, row.site_id, row.pr_number);
  return rows.length === SWEEP_BATCH;
}