import { env } from "cloudflare:workers";
import { first } from "../db";
import { accountDb, stmt } from "../scoped-db";
import type { ScopedStatement } from "../scoped-db";
import { meterDef } from "~/lib/meters";
import type { MeterId } from "~/lib/meters";
import { isPlanId } from "~/lib/plans";
import { creditPoolFor } from "~/lib/pricing";
import { randomId } from "../crypto";
import { HTTP_PAYMENT_REQUIRED, HttpError } from "../errors";
import type { PaymentReason } from "../errors";
import { otherHoldsSql } from "./credits";
import { billingSettings, hardCap } from "./settings";
import { holdGauge } from "./gauge-hold";
import { noteScratchAccount } from "./directory";
import type { BillingSettings } from "./settings";

export { billingSettings, hardCap, plans } from "./settings";
export type { BillingSettings } from "./settings";
import { spendGuard } from "./spend-guard";
import { paymentErrorFor } from "./access";
import {
  eventStatements,
  MeteringError,
  periodEnd,
  periodOf,
  resolveAccount,
  validate,
  writeAnalytics,
} from "./record";
import type { UsageInput } from "./record";
import { forgetSupporter } from "../supporter";

const DEFAULT_TTL_MS = 3_600_000;

/**
 * Out of budget. Deliberately not an HttpError, so existing handlers keep their behaviour (background work
 * skips, integrations retry); request boundaries turn it into a 402 with toHttpError().
 */
export class QuotaExceededError extends Error {
  constructor(
    readonly meter: MeterId,
    readonly limit: number,
    message = `Usage limit reached for ${meterDef(meter)?.label ?? meter}`,
    readonly accountId: string | null = null,
    readonly reason: PaymentReason = "plan_limit",
  ) {
    super(message);
    this.name = "QuotaExceededError";
  }

  /** 402 Payment Required naming the meter, why, and (internally) whose account; see withBillingContext. */
  toHttpError(message = this.message): HttpError {
    return new HttpError(HTTP_PAYMENT_REQUIRED, message, {
      code: "payment_required",
      meter: this.meter,
      reason: this.reason,
      accountId: this.accountId,
    });
  }
}

/** The billed account is disabled (dispute, admin): nothing it pays for may start. */
export class AccountDisabledError extends QuotaExceededError {
  constructor(meter: MeterId, accountId: string) {
    super(meter, 0, "This account is disabled", accountId);
    this.name = "AccountDisabledError";
  }

  override toHttpError(message = this.message): HttpError {
    return new HttpError(403, message);
  }
}

/** A platform-wide safety ceiling (not the account's budget) refused the work. */
export class PlatformLimitError extends QuotaExceededError {
  constructor(meter: MeterId, message: string) {
    super(meter, 0, message);
    this.name = "PlatformLimitError";
  }

  override toHttpError(message = this.message): HttpError {
    return new HttpError(503, message);
  }
}

type PeriodRow = { quantity: number; reserved: number };

async function periodRow(accountId: string, meter: MeterId, period: string): Promise<PeriodRow> {
  const row = await accountDb(accountId).first<PeriodRow>(
    "SELECT quantity, reserved FROM usage_periods WHERE account_id = ? AND meter = ? AND period = ?",
    accountId,
    meter,
    period,
  );
  return row ?? { quantity: 0, reserved: 0 };
}

/** Extra headroom from prepaid credit this period, in meter units: what credit covered plus what it still can. */
export async function creditHeadroom(accountId: string, meter: MeterId, period = periodOf()): Promise<number> {
  const credit = creditPoolFor(meter);
  if (!credit) return 0;
  const row = await accountDb(accountId).first<{ headroom: number }>(
    `SELECT COALESCE((SELECT credit_covered FROM usage_periods WHERE account_id = ?1 AND meter = ?2 AND period = ?3), 0)
       + MAX(0, COALESCE((SELECT balance FROM credit_balances WHERE account_id = ?1 AND meter = ?4), 0)
         - ${otherHoldsSql(credit.pool, "?1", "?2", period)}) / ?5 AS headroom`,
    accountId,
    meter,
    period,
    credit.pool,
    credit.factor,
  );
  return row?.headroom ?? 0;
}

/** Overage in cents so far this period across every priced meter (counters plus gauge peaks). */
export async function periodSpendCents(
  accountId: string,
  settings: BillingSettings,
  period = periodOf(),
): Promise<number> {
  if (!settings.plan.billsOverage) return 0;
  const db = accountDb(accountId);
  const rows = await db.all<{ meter: MeterId; quantity: number; reserved: number; credit_covered: number }>(
    `SELECT meter, quantity, reserved, credit_covered FROM usage_periods WHERE account_id = ?1 AND period = ?2
     UNION ALL SELECT meter, peak, 0, 0 FROM usage_gauges WHERE account_id = ?1 AND period = ?2`,
    accountId,
    period,
  );
  const pools = await db.all<{ meter: string; balance: number }>(
    "SELECT meter, balance FROM credit_balances WHERE account_id = ?",
    accountId,
  );
  // The pool's unspent balance is shared by its meters, so it is allotted once, and only to the part of each
  // in-flight hold past the meter's included amount (in a fixed order, so the estimate is stable).
  const balance = new Map(pools.map((row) => [row.meter, row.balance]));
  let cents = 0;
  const ordered = [...rows].sort((left, right) => left.meter.localeCompare(right.meter));
  for (const row of ordered) {
    const price = meterDef(row.meter)?.microCentsPerUnit ?? 0;
    const credit = creditPoolFor(row.meter);
    const included = settings.plan.included[row.meter] ?? 0;
    const recorded = row.quantity - row.credit_covered;
    let covered = 0;
    if (credit && row.reserved > 0) {
      const holdOver = Math.max(0, recorded + row.reserved - included) - Math.max(0, recorded - included);
      const left = balance.get(credit.pool) ?? 0;
      covered = Math.min(holdOver, left / credit.factor);
      balance.set(credit.pool, left - covered * credit.factor);
    }
    const quantity = recorded + row.reserved - covered;
    const billable = Math.max(0, quantity - included);
    cents += (billable * price) / 1e6;
  }
  return cents;
}

/**
 * False once a paid account's total overage is past its spending limit. The boundary matches spendGuard:
 * overage may reach the limit exactly but never exceed it, compared in whole microcents so float drift
 * cannot make the two disagree. A limit of 0 still allows usage within the included amounts.
 */
export async function withinSpendingLimit(accountId: string, meter: MeterId | null = null): Promise<boolean> {
  const resolved = await resolveAccount(accountId);
  return spendWithinLimit(resolved, await billingSettings(resolved), meter);
}

async function spendWithinLimit(accountId: string, settings: BillingSettings, meter: MeterId | null) {
  if (!settings.plan.billsOverage || settings.spendingLimitCents === null) return true;
  if (meter && (meterDef(meter)?.microCentsPerUnit ?? 0) === 0) return true;
  const spentMicro = Math.round((await periodSpendCents(accountId, settings)) * 1e6);
  return spentMicro <= settings.spendingLimitCents * 1e6;
}

/**
 * Read-only check that `quantity` more of `meter` fits under the account's hard cap (counting in-flight
 * reservations) and its spending limit. Soft meters only check the spending limit. Use reserveUsage when
 * concurrent callers race.
 */
export async function assertWithinQuota(accountId: string, meter: MeterId, quantity = 0): Promise<void> {
  const resolved = await resolveAccount(accountId);
  const settings = await billingSettings(resolved);
  if (settings.disabled) throw new AccountDisabledError(meter, resolved);
  const cap = hardCap(settings, meter);
  if (!(await spendWithinLimit(resolved, settings, meter))) {
    throw new QuotaExceededError(
      meter,
      settings.spendingLimitCents ?? 0,
      "Spending limit reached for this month",
      resolved,
      "spending_limit",
    );
  }
  if (cap === null) return;
  if (meterDef(meter)?.kind === "gauge") {
    if (quantity > 0) {
      if (!(await holdGauge(resolved, meter, quantity, cap)))
        throw new QuotaExceededError(meter, cap, undefined, resolved);
      return;
    }
    const gauge = await accountDb(resolved).first<{ current: number }>(
      "SELECT current FROM usage_gauges WHERE account_id = ? AND meter = ? AND period = ?",
      resolved,
      meter,
      periodOf(),
    );
    if ((gauge?.current ?? 0) >= cap) throw new QuotaExceededError(meter, cap, undefined, resolved);
    return;
  }
  const row = await periodRow(resolved, meter, periodOf());
  const limit = cap + (await creditHeadroom(resolved, meter));
  if (row.quantity + row.reserved + quantity > limit || (quantity === 0 && row.quantity + row.reserved >= limit)) {
    throw new QuotaExceededError(meter, cap, undefined, resolved);
  }
}

function upgradeHint(blocked: QuotaExceededError): boolean {
  return !(blocked instanceof AccountDisabledError || blocked instanceof PlatformLimitError);
}

/** Non-throwing form for request handlers: the 402 the plan raises for this, or null when it fits. */
export async function planBlock(accountId: string, meter: MeterId, quantity = 0): Promise<QuotaExceededError | null> {
  try {
    await assertWithinQuota(accountId, meter, quantity);
    return null;
  } catch (error) {
    if (!(error instanceof QuotaExceededError)) throw error;
    return error;
  }
}

/** The billing page that resolves a block on `accountId`: the user's own settings, or the org's. */
export async function billingUrlFor(accountId: string): Promise<string> {
  const account = await first<{ name: string; kind: string }>(
    "SELECT name, kind FROM accounts WHERE id = ?",
    accountId,
  );
  const path = account?.kind === "org" ? `/orgs/${encodeURIComponent(account.name)}/billing` : "/settings/billing";
  return new URL(path, env.PUBLIC_URL).toString();
}

/** Text-channel form (git push output, plain-text responses): the reason plus where to upgrade, or null. */
export async function planBlockReason(accountId: string, meter: MeterId, quantity = 0): Promise<string | null> {
  const blocked = await planBlock(accountId, meter, quantity);
  if (!blocked) return null;
  if (!upgradeHint(blocked)) return blocked.message;
  return `${blocked.message} on this account's plan this month. Upgrade or add credit: ${await billingUrlFor(accountId)}`;
}

/** The 402 to raise when the plan blocks this, with the billing link in the message for text-only clients. */
export async function planBlockError(accountId: string, meter: MeterId, quantity = 0): Promise<HttpError | null> {
  const blocked = await planBlock(accountId, meter, quantity);
  if (!blocked) return null;
  if (!upgradeHint(blocked)) return blocked.toHttpError();
  return blocked.toHttpError(
    `${blocked.message} on this account's plan this month. Upgrade or add credit: ${await billingUrlFor(accountId)}`,
  );
}

/** A 402 response for plain HTTP routes, JSON with the same detail server functions carry. */
export async function paymentRequiredResponse(error: QuotaExceededError, viewerId: string | null): Promise<Response> {
  const http = await paymentErrorFor(error.toHttpError(), viewerId);
  return Response.json({ error: http.message, ...http.detail }, { status: http.status });
}

/** How much more of `meter` fits under the account's hard cap this period, or null when it is uncapped. */
export async function remainingQuota(accountId: string, meter: MeterId): Promise<number | null> {
  const resolved = await resolveAccount(accountId);
  const cap = hardCap(await billingSettings(resolved), meter);
  if (cap === null) return null;
  const row = await periodRow(resolved, meter, periodOf());
  return Math.max(0, cap + (await creditHeadroom(resolved, meter)) - row.quantity - row.reserved);
}

/** Quantity recorded plus in-flight holds for `meter` this period. */
export async function periodUsage(accountId: string, meter: MeterId): Promise<PeriodRow> {
  return periodRow(await resolveAccount(accountId), meter, periodOf());
}

export type Reservation = { id: string; accountId: string; meter: MeterId; quantity: number; period: string };

/**
 * Atomically holds `quantity` of a hard-capped meter. The D1 batch is one transaction: the reservation row
 * is inserted only if `quantity + reserved + q <= cap` and the hold is added only if this attempt's nonce
 * landed, so concurrent reservations can never overshoot. `id` is the caller's idempotency key: reserving
 * the same id twice returns the original reservation without holding more.
 * Settle with commitUsage (the real amount, which may be smaller) or releaseUsage; unsettled reservations
 * are released by cron after `ttlMs`.
 */
export async function reserveUsage(input: {
  id: string;
  accountId: string;
  meter: MeterId;
  quantity: number;
  ttlMs?: number;
}): Promise<Reservation> {
  if (!Number.isFinite(input.quantity) || input.quantity < 0) throw new MeteringError("Quantity must be >= 0");
  const accountId = await resolveAccount(input.accountId);
  const db = accountDb(accountId);
  // Reservations live in the billed account's own object, so a retried id can only ever match its own (INT-A19).
  const existing = await db.first<{ quantity: number; period: string }>(
    "SELECT quantity, period FROM usage_reservations WHERE id = ?",
    input.id,
  );
  if (existing) return { ...input, accountId, quantity: existing.quantity, period: existing.period };
  const settings = await billingSettings(accountId);
  if (settings.disabled) throw new AccountDisabledError(input.meter, accountId);
  if (!(await spendWithinLimit(accountId, settings, input.meter))) {
    throw new QuotaExceededError(input.meter, 0, "Spending limit reached for this month", accountId, "spending_limit");
  }
  const period = periodOf();
  const cap = hardCap(settings, input.meter);
  const credit = creditPoolFor(input.meter);
  const now = Date.now();
  const headroom = credit
    ? `MAX(0, COALESCE((SELECT balance FROM credit_balances WHERE account_id = ?2 AND meter = ?10), 0) - ${otherHoldsSql(credit.pool, "?2", "?3", period, now)}) / ?11`
    : "0";
  const spend = spendGuard(settings, input.meter, headroom);
  const nonce = randomId();
  const ttl = input.ttlMs ?? DEFAULT_TTL_MS;
  await noteScratchAccount(accountId);
  await db.batch([
    stmt(
      "INSERT OR IGNORE INTO usage_periods (account_id, meter, period, updated_at) VALUES (?, ?, ?, ?)",
      accountId,
      input.meter,
      period,
      now,
    ),
    stmt(
      `INSERT OR IGNORE INTO usage_reservations (id, account_id, meter, period, quantity, expires_at, created_at, nonce)
       SELECT ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8 FROM usage_periods
       WHERE account_id = ?2 AND meter = ?3 AND period = ?4 AND (?9 IS NULL OR quantity + reserved + ?5 <= ?9
         + credit_covered + ${credit ? `MAX(0, COALESCE((SELECT balance FROM credit_balances WHERE account_id = ?2 AND meter = ?10), 0) - ${otherHoldsSql(credit.pool, "?2", "?3", period, now)}) / ?11` : "0"})${spend}`,
      input.id,
      accountId,
      input.meter,
      period,
      input.quantity,
      now + ttl,
      now,
      nonce,
      cap,
      ...(credit ? [credit.pool, credit.factor] : []),
    ),
    stmt(
      `UPDATE usage_periods SET reserved = reserved + ?1 WHERE account_id = ?2 AND meter = ?3 AND period = ?4
         AND EXISTS (SELECT 1 FROM usage_reservations WHERE id = ?5 AND nonce = ?6)`,
      input.quantity,
      accountId,
      input.meter,
      period,
      input.id,
      nonce,
    ),
  ]);
  const landed = await db.first<{ quantity: number; period: string }>(
    "SELECT quantity, period FROM usage_reservations WHERE id = ?",
    input.id,
  );
  if (landed) return { ...input, accountId, quantity: landed.quantity, period: landed.period };
  if (spend && !(await overCap(accountId, input.meter, input.quantity, cap, period))) {
    const limit = settings.spendingLimitCents ?? 0;
    throw new QuotaExceededError(
      input.meter,
      limit,
      "Spending limit reached for this month",
      accountId,
      "spending_limit",
    );
  }
  throw new QuotaExceededError(input.meter, cap ?? 0, undefined, accountId);
}

/** After a refused hold: whether the meter's own cap refused it, rather than the account-wide spending limit. */
async function overCap(accountId: string, meter: MeterId, quantity: number, cap: number | null, period: string) {
  if (cap === null) return false;
  const row = await periodRow(accountId, meter, period);
  return row.quantity + row.reserved + quantity > cap + (await creditHeadroom(accountId, meter, period));
}

export function releaseStatements(id: string): ScopedStatement[] {
  return [
    stmt(
      `UPDATE usage_periods SET reserved = MAX(0, reserved - (SELECT quantity FROM usage_reservations WHERE id = ?1))
       WHERE (account_id, meter, period) = (SELECT account_id, meter, period FROM usage_reservations WHERE id = ?1)`,
      id,
    ),
    stmt("DELETE FROM usage_reservations WHERE id = ?", id),
  ];
}

/**
 * Settles a reservation with the actual usage. Idempotent: the release happens once, and the event insert
 * is keyed by `idempotencyKey` (defaults to `reservation:<id>`). The event is dated inside the reservation's
 * period so the monthly and daily rollups agree even when a hold straddles midnight at month end.
 */
export async function commitUsage(
  reservation: Reservation,
  actual: Omit<UsageInput, "accountId" | "meter" | "idempotencyKey"> & { idempotencyKey?: string },
): Promise<void> {
  const input: UsageInput = {
    ...actual,
    accountId: reservation.accountId,
    meter: reservation.meter,
    idempotencyKey: actual.idempotencyKey ?? `reservation:${reservation.id}`,
  };
  validate(input);
  const at = Math.min(Date.now(), periodEnd(reservation.period));
  const statements = releaseStatements(reservation.id);
  const eventIndex = statements.length;
  if (input.quantity > 0) {
    statements.push(...eventStatements(input, reservation.accountId, reservation.period, at));
  }
  const results = await accountDb(reservation.accountId).batch(statements);
  if (input.quantity > 0 && (results[eventIndex]?.meta.changes ?? 0) > 0) {
    writeAnalytics(input, reservation.accountId, reservation.period);
  }
}

/** Releases a hold of `accountId` (the account it was reserved for; claimed scratch ids are followed). */
export async function releaseUsage(reservationId: string, accountId: string): Promise<void> {
  await accountDb(await resolveAccount(accountId)).batch(releaseStatements(reservationId));
}

/** KV flag set by the sites rollup when an account's sites reach a plan hard cap; plan changes clear it. */
export function siteCapKey(accountId: string, period = periodOf()): string {
  return `sites:over-cap:${accountId}:${period}`;
}

export async function clearSiteCap(accountId: string): Promise<void> {
  await env.KV.delete(siteCapKey(accountId));
}

/**
 * Sets the account's plan. `admin` pins it (BILL-P8) so later Stripe events leave it alone; `billing` is the
 * Stripe path and changes nothing while an admin pin is in place.
 */
export async function setPlan(accountId: string, planId: string, source: "admin" | "billing"): Promise<void> {
  if (!isPlanId(planId) || planId === "scratch") throw new MeteringError("Unknown plan");
  if (!(await first("SELECT 1 AS one FROM accounts WHERE id = ?", accountId)))
    throw new MeteringError("Unknown account");
  const pinned = source === "admin" ? 1 : 0;
  const result = await accountDb(accountId).run(
    `INSERT INTO account_billing (account_id, plan, plan_override, updated_at) VALUES (?1, ?2, ?3, ?4)
     ON CONFLICT (account_id) DO UPDATE SET plan = excluded.plan, plan_override = excluded.plan_override,
       updated_at = excluded.updated_at
     WHERE ?3 = 1 OR account_billing.plan_override = 0`,
    accountId,
    planId,
    pinned,
    Date.now(),
  );
  if (result.meta.changes > 0) await Promise.all([clearSiteCap(accountId), forgetSupporter(accountId)]);
}

/** Hands a pinned account back to Stripe; the next subscription event decides its plan. */
export async function clearPlanOverride(accountId: string): Promise<void> {
  await accountDb(accountId).run(
    "UPDATE account_billing SET plan_override = 0, updated_at = ? WHERE account_id = ?",
    Date.now(),
    accountId,
  );
}

export async function setSpendingLimit(accountId: string, cents: number | null): Promise<void> {
  if (cents !== null && (!Number.isSafeInteger(cents) || cents < 0)) throw new MeteringError("Bad spending limit");
  await accountDb(accountId).run(
    `INSERT INTO account_billing (account_id, spending_limit_cents, updated_at) VALUES (?, ?, ?)
     ON CONFLICT (account_id) DO UPDATE SET spending_limit_cents = excluded.spending_limit_cents,
       updated_at = excluded.updated_at`,
    accountId,
    cents,
    Date.now(),
  );
  await clearSiteCap(accountId);
}