import { env } from "cloudflare:workers";
import { all } from "../db";
import { accountDb, repoDb } from "../scoped-db";
import { ownerSiteScopes } from "./store";
import { counterTotals } from "../counters";
import { periodEnd, periodOf } from "../metering/record";
import { billingSettings, hardCap, siteCapKey } from "../metering/quota";
import { notifyUsers } from "../notify/deliver";
import { publicUrl } from "../events/emit";
import type { MeterId } from "~/lib/meters";
import { HOSTING_METERS, ROLLUP_DAYS } from "./metering";
import { usageDay } from "./usage";

const DAY_MS = 24 * 60 * 60 * 1000;
/** Per isolate: how often one account's caps or the global ceiling are re-read after counter flushes. */
const ACCOUNT_CHECK_MS = 30_000;
const GLOBAL_CHECK_MS = 60_000;
const FLAG_CACHE_SECONDS = 60;
const GLOBAL_FLAG_TTL_SECONDS = 2 * 24 * 60 * 60;
const MIN_TTL_SECONDS = 60;
const MAX_TRACKED_ACCOUNTS = 1000;
const lastChecked = new Map<string, number>();
let lastGlobalCheck = 0;

type Totals = { requests: number; cpu_ms: number; bytes_out: number };

/** Days of this period a scope may not have folded into usage_periods yet: today and the few before it. */
function recentDays(now: number): string[] {
  const period = periodOf(now);
  const days = [usageDay(now)];
  for (let back = 1; back <= ROLLUP_DAYS; back += 1) {
    const day = usageDay(now - back * DAY_MS);
    if (!day.startsWith(period)) break;
    days.push(day);
  }
  return days;
}

/** Counters each scope has but not yet folded (its `site_rollups` says which days it folded). */
async function unrolledTotals(accountId: string, days: string[]) {
  const list = days.map(() => "?").join(", ");
  const sql = `SELECT COALESCE(SUM(requests), 0) AS requests, COALESCE(SUM(cpu_ms), 0) AS cpu_ms, COALESCE(SUM(bytes_out), 0) AS bytes_out
         FROM site_usage WHERE account_id = ? AND day IN (${list}) AND day NOT IN (SELECT day FROM site_rollups)`;
  const totals: Totals = { requests: 0, cpu_ms: 0, bytes_out: 0 };
  for (const scope of await ownerSiteScopes(accountId)) {
    const row = await repoDb(scope).first<Totals>(sql, accountId, ...days);
    totals.requests += row?.requests ?? 0;
    totals.cpu_ms += row?.cpu_ms ?? 0;
    totals.bytes_out += row?.bytes_out ?? 0;
  }
  return totals;
}

/** This period's hosting usage of an account: rolled-up meters plus counters D1 has but the rollup has not folded. */
export async function hostingUsage(accountId: string, now = Date.now()): Promise<Record<string, number>> {
  const [rolled, siteDays] = await Promise.all([
    accountDb(accountId).all<{ meter: MeterId; quantity: number }>(
      `SELECT meter, quantity FROM usage_periods WHERE account_id = ? AND period = ? AND meter IN (${HOSTING_METERS.map(() => "?").join(", ")})`,
      accountId,
      periodOf(now),
      ...HOSTING_METERS,
    ),
    Promise.resolve(recentDays(now)),
  ]);
  const sites = await unrolledTotals(accountId, siteDays);
  const usage: Record<string, number> = {
    "compute.site_requests": sites.requests,
    "compute.site_cpu": sites.cpu_ms,
    "network.sites": sites.bytes_out,
  };
  for (const row of rolled) usage[row.meter] = (usage[row.meter] ?? 0) + row.quantity;
  return usage;
}

/**
 * Same-day stop: sets the owner's serving stop as soon as flushed counters reach a hosting hard cap, instead of
 * waiting for the daily rollup. Reads D1 and writes only the KV flag, so request traffic adds no D1 writes.
 */
export async function enforceSameDayCap(accountId: string, now = Date.now()): Promise<boolean> {
  if (accountId.startsWith("scratch:")) return false;
  const period = periodOf(now);
  if (await env.KV.get(siteCapKey(accountId, period))) return true;
  const settings = await billingSettings(accountId);
  const caps = HOSTING_METERS.map((meter) => [meter, hardCap(settings, meter)] as const).filter(
    (entry): entry is readonly [MeterId, number] => entry[1] !== null,
  );
  if (caps.length === 0) return false;
  const usage = await hostingUsage(accountId, now);
  const reached = caps.find(([meter, cap]) => (usage[meter] ?? 0) >= cap);
  if (!reached) return false;
  const ttl = Math.max(MIN_TTL_SECONDS, Math.ceil((periodEnd(period) - now) / 1000));
  await env.KV.put(siteCapKey(accountId, period), reached[0], { expirationTtl: ttl });
  return true;
}

function limit(raw: string | undefined): number {
  const value = Number(raw ?? 0);
  return Number.isFinite(value) && value > 0 ? value : 0;
}

function globalKey(day: string): string {
  return `sites:global-ceiling:${day}`;
}

/** Whether today's platform-wide dynamic site ceiling was crossed; dynamic sites answer 503 until the next UTC day. */
export async function globalCeilingHit(now = Date.now()): Promise<boolean> {
  if (!limit(env.SITES_DAILY_CPU_MS_LIMIT) && !limit(env.SITES_DAILY_REQUEST_LIMIT)) return false;
  return (await env.KV.get(globalKey(usageDay(now)), { cacheTtl: FLAG_CACHE_SECONDS })) !== null;
}

/** Sums today's site counters against SITES_DAILY_CPU_MS_LIMIT / SITES_DAILY_REQUEST_LIMIT (0 = unlimited). */
export async function checkGlobalCeiling(now = Date.now()): Promise<boolean> {
  const cpuLimit = limit(env.SITES_DAILY_CPU_MS_LIMIT);
  const requestLimit = limit(env.SITES_DAILY_REQUEST_LIMIT);
  if (!cpuLimit && !requestLimit) return false;
  const day = usageDay(now);
  if (await env.KV.get(globalKey(day))) return true;
  // Every site's counter forwards its flushed deltas to one platform total, so this reads one object.
  const [requests, cpuMs, bytesOut] = await counterTotals("platform-sites", "global", day);
  const totals: Totals = { requests, cpu_ms: cpuMs, bytes_out: bytesOut };
  const over = (cpuLimit > 0 && totals.cpu_ms >= cpuLimit) || (requestLimit > 0 && totals.requests >= requestLimit);
  if (!over) return false;
  await env.KV.put(globalKey(day), JSON.stringify(totals), { expirationTtl: GLOBAL_FLAG_TTL_SECONDS });
  await notifyAdminsOfCeiling(day, totals);
  return true;
}

async function notifyAdminsOfCeiling(day: string, totals: Totals): Promise<void> {
  const admins = await all<{ account_id: string }>("SELECT account_id FROM users WHERE is_admin = 1 LIMIT 50");
  await notifyUsers({
    category: "quota",
    repoId: null,
    accountId: null,
    access: "account",
    recipients: admins.map((row) => row.account_id),
    title: `Dynamic sites hit the daily platform ceiling (${day}) and answer 503 until midnight UTC`,
    details: [`Requests today: ${totals.requests}`, `CPU ms today: ${totals.cpu_ms}`],
    url: publicUrl("/admin"),
    dedupeKey: `sites-global-ceiling:${day}`,
  }).catch((error: unknown) => console.error("Global ceiling notification failed", error));
}

/** Counters hook: after site counters reach D1, re-check the touched owners' caps and the global ceiling. */
export async function afterHostingFlush(accountIds: string[], now = Date.now()): Promise<void> {
  for (const accountId of accountIds) {
    if (now - (lastChecked.get(accountId) ?? 0) < ACCOUNT_CHECK_MS) continue;
    // Insertion order makes the map an LRU: re-inserting moves the account to the end, the oldest goes first.
    lastChecked.delete(accountId);
    lastChecked.set(accountId, now);
    if (lastChecked.size > MAX_TRACKED_ACCOUNTS) {
      const oldest = lastChecked.keys().next().value;
      if (oldest !== undefined) lastChecked.delete(oldest);
    }
    await enforceSameDayCap(accountId, now);
  }
  if (now - lastGlobalCheck >= GLOBAL_CHECK_MS) {
    lastGlobalCheck = now;
    await checkGlobalCeiling(now);
  }
}