import { env } from "cloudflare:workers";
import { all, chunks } from "../db";
import { stmt } from "../scoped-db";
import { allSites, claimHost, forgetSite, loadSiteRow, releaseHost, siteDb, siteScopes } from "./store";
import { HttpError } from "../errors";
import { loadRepoById } from "../repos";
import { customDomainsEnabled, deleteCustomHostname } from "./domains";
import { customHostKey, labelUrl, metaKey, publishMeta, sitesDomain, slotKey } from "./config";
import { assertLabelFree, loadSiteById, loadSiteForRepo } from "./deploy";
import { siteLabel } from "./naming";
import { refreshSiteDomain, releaseSiteDomain } from "./site";
import { aliasLabels, dropSiteAliases, sweepClosedPullAliases } from "./pr-aliases";
import { customDomainCount, notifyDomainOwners, syncDomainAddonQuantity } from "../billing/domain-addon";
import { CUSTOM_DOMAIN_LIMIT } from "~/lib/pricing";
import type { SiteMeta, SiteRow } from "./config";

const OWNER_BATCH = 200;
const RELABEL_BLOCKED = "Site host";
const RECONCILE_BATCH = 100;
const DOMAIN_BATCH = 10;
const META_PAGE = 500;
const CURSOR_KEY = "sites:reconcile:cursor";
const META_CURSOR_KEY = "sites:reconcile:meta-cursor";

/** Unpublishes everything a site serves (meta, every label, custom host, SaaS hostname) from KV-only state. */
async function teardown(
  meta: Pick<SiteMeta, "id" | "customDomain" | "customHostnameId">,
  labels: string[],
): Promise<void> {
  await Promise.all(labels.map((label) => env.KV.delete(slotKey(label))));
  if (meta.customDomain) await env.KV.delete(customHostKey(meta.customDomain));
  if (meta.customHostnameId && customDomainsEnabled()) {
    await deleteCustomHostname(meta.customHostnameId).catch((error: unknown) =>
      console.warn("Custom hostname cleanup failed", meta.id, error),
    );
  }
  await env.KV.delete(metaKey(meta.id));
}

async function servedLabels(site: SiteRow): Promise<string[]> {
  const rows = await siteDb(site).all<{ label: string }>(
    "SELECT DISTINCT label FROM site_deployments WHERE site_id = ? AND status = 'active' LIMIT 500",
    site.id,
  );
  return [site.label, ...rows.map((row) => row.label), ...(await aliasLabels(site.id))];
}

/**
 * Captures what a repo's site serves before the repository row is deleted (D1 cascades the site away) and returns
 * the teardown to run once the delete has committed.
 */
export async function siteTeardownFor(repoId: string): Promise<() => Promise<void>> {
  const site = await loadSiteForRepo(repoId);
  const labels = site ? await servedLabels(site) : [];
  const meta = site && { id: site.id, customDomain: site.custom_domain, customHostnameId: site.custom_hostname_id };
  return async () => {
    if (meta) await teardown(meta, labels);
    if (site) await forgetSite(site);
    if (site?.custom_domain) await syncDomainAddonQuantity(site.owner_id);
  };
}

/** Moves a site to the label its current owner and name imply; previews are dropped and rebuild on the next push. */
async function relabel(site: SiteRow, label: string): Promise<SiteRow> {
  const labels = await servedLabels(site);
  await claimHost(label, site.id);
  await siteDb(site).batch([
    stmt(
      "UPDATE sites SET label = ?, disabled_reason = CASE WHEN disabled_reason LIKE ? THEN NULL ELSE disabled_reason END WHERE id = ?",
      label,
      `${RELABEL_BLOCKED}%`,
      site.id,
    ),
    stmt("UPDATE site_deployments SET label = ? WHERE site_id = ? AND label = ?", label, site.id, site.label),
    stmt(
      "UPDATE site_deployments SET status = 'superseded' WHERE site_id = ? AND environment = 'preview' AND status = 'active'",
      site.id,
    ),
  ]);
  await Promise.all(labels.map((old) => env.KV.delete(slotKey(old))));
  for (const old of labels) if (old !== label) await releaseHost(old, site.id);
  await dropSiteAliases(site.id);
  if (site.active_deployment_id) {
    await env.KV.put(slotKey(label), JSON.stringify({ siteId: site.id, deploymentId: site.active_deployment_id }));
  }
  if (site.custom_domain) {
    const custom = await env.KV.get(customHostKey(site.custom_domain));
    if (custom) await env.KV.put(customHostKey(site.custom_domain), JSON.stringify({ siteId: site.id, label }));
  }
  return (await loadSiteById(site.id)) ?? site;
}

/**
 * Re-derives a repo site's state from its repository: tears it down when the repo is gone, follows renames and
 * transfers, and gates a public site whose repository became private.
 */
export async function syncSiteForRepo(repoId: string): Promise<void> {
  let site = await loadSiteForRepo(repoId);
  if (!site) return;
  const repo = await loadRepoById(repoId);
  if (!repo) {
    await (
      await siteTeardownFor(repoId)
    )();
    return;
  }
  if (site.scratch !== 1 && site.owner_id !== repo.owner_id) {
    // A transfer moves the site, its usage and its custom domain to the new owner (syncHostingForRepo bills it).
    await siteDb(site).run(
      "UPDATE sites SET owner_id = ?, updated_at = ? WHERE id = ?",
      repo.owner_id,
      Date.now(),
      site.id,
    );
    site = (await loadSiteById(site.id)) ?? site;
  }
  const expected = siteLabel(repo.owner_name, repo.name, sitesDomain());
  if (expected !== site.label) {
    const taken = await assertLabelFree(site.id, expected).then(
      () => false,
      (error: unknown) => {
        if (error instanceof HttpError) return true;
        throw error;
      },
    );
    if (taken) {
      console.warn("Site relabel blocked", site.id, labelUrl(expected));
      await siteDb(site).run(
        "UPDATE sites SET disabled_reason = ? WHERE id = ? AND disabled_reason IS NULL",
        `${RELABEL_BLOCKED} ${expected} is taken`,
        site.id,
      );
      site = (await loadSiteById(site.id)) ?? site;
    } else {
      site = await relabel(site, expected);
    }
  }
  if (repo.visibility !== "public" && site.access === "public" && site.scratch !== 1) {
    await siteDb(site).run(
      "UPDATE sites SET access = 'gitfor', access_version = access_version + 1, updated_at = ? WHERE id = ?",
      Date.now(),
      site.id,
    );
    site = (await loadSiteById(site.id)) ?? site;
  }
  await publishMeta(site);
}

/**
 * A transfer can't push the new owner past CUSTOM_DOMAIN_LIMIT: the moved repository's custom domains are unmapped
 * (and their Cloudflare hostnames deleted) instead, and the new owner is told.
 */
async function enforceDomainCapAfterTransfer(repoId: string, ownerId: string): Promise<void> {
  if ((await customDomainCount(ownerId)) <= CUSTOM_DOMAIN_LIMIT) return;
  const site = await loadSiteForRepo(repoId);
  if (!site?.custom_domain) return;
  const dropped = [site.custom_domain];
  await releaseSiteDomain(site);
  await notifyDomainOwners(
    ownerId,
    `Custom domains removed from a transferred repository: the account already holds ${CUSTOM_DOMAIN_LIMIT}`,
    `Removed: ${dropped.join(", ")}. Free up a domain, then add it again.`,
    "domain-cap-transfer",
  );
}

/** After any visibility, name or owner change: brings Sites in line with the repository. */
export async function syncHostingForRepo(repoId: string, previousOwnerId: string | null = null): Promise<void> {
  await syncSiteForRepo(repoId);
  if (previousOwnerId) {
    const repo = await loadRepoById(repoId);
    if (repo && repo.owner_id !== previousOwnerId) {
      await enforceDomainCapAfterTransfer(repoId, repo.owner_id);
      await syncDomainAddonQuantity(previousOwnerId);
      await syncDomainAddonQuantity(repo.owner_id);
    }
  }
}

export async function syncSitesForOwner(ownerId: string): Promise<void> {
  const rows = await all<{ id: string }>("SELECT id FROM repos WHERE owner_id = ? LIMIT ?", ownerId, OWNER_BATCH);
  for (const row of rows) await syncSiteForRepo(row.id);
}

async function reconcileRepoSites(): Promise<void> {
  const cursor = (await env.KV.get(CURSOR_KEY)) ?? "";
  const scopes = await siteScopes();
  const repoIds = scopes
    .filter((scope) => !scope.startsWith("site:") && scope > cursor)
    .sort()
    .slice(0, RECONCILE_BATCH);
  for (const repoId of repoIds) await syncSiteForRepo(repoId);
  await env.KV.put(CURSOR_KEY, repoIds.length === RECONCILE_BATCH ? repoIds[repoIds.length - 1] : "");
  const domains = (await allSites("custom_hostname_id IS NOT NULL"))
    .sort((left, right) => (left.domain_checked_at ?? 0) - (right.domain_checked_at ?? 0))
    .slice(0, DOMAIN_BATCH);
  for (const site of domains) {
    await refreshSiteDomain(site).catch((error: unknown) => console.warn("Domain refresh failed", site.id, error));
    await siteDb(site).run("UPDATE sites SET domain_checked_at = ? WHERE id = ?", Date.now(), site.id);
  }
}

/** Deletes KV metadata whose D1 site row is gone (for example after a cascade), so the host stops serving. */
async function reconcileOrphanMeta(): Promise<void> {
  const cursor = (await env.KV.get(META_CURSOR_KEY)) ?? undefined;
  const page = await env.KV.list({ prefix: "sites:meta:", cursor, limit: META_PAGE });
  const ids = page.keys.map((key) => key.name.slice("sites:meta:".length));
  for (const group of chunks(ids)) {
    for (const id of group) {
      if (!(await isOrphanSite(id))) continue;
      const meta = await env.KV.get<SiteMeta>(metaKey(id), "json");
      if (meta) await teardown(meta, [meta.label]);
    }
  }
  if (page.list_complete) await env.KV.delete(META_CURSOR_KEY);
  else await env.KV.put(META_CURSOR_KEY, page.cursor);
}

export async function reconcileSites(): Promise<void> {
  if (!sitesDomain()) return;
  await reconcileRepoSites();
  await reconcileOrphanMeta();
  await sweepClosedPullAliases();
}

/** True when the slot's owner no longer exists, so a new site may take the label over. */
export async function isOrphanSite(siteId: string): Promise<boolean> {
  return (await loadSiteRow(siteId)) === null;
}