import { flushAccountDigests } from "../notify/deliver";
import { releaseStatements } from "../metering/quota";
import { sendAccountBudgetAlerts } from "../metering/cron";
import { gaugeStatement, periodOf, resolveAccount } from "../metering/record";
import { sumScopeGauges } from "../metering/scope-gauges";
import { publishUsageLeaderboard } from "../metering/leaderboard";
import { accountDb } from "../scoped-db";
import { nextRunnerOffline, reportOfflineRunners } from "../notify/runners";
import { closeExpiredContainerSessions, EXPIRING_SESSION_DUE } from "../editor/container-meter";
import { accountTaskHelpers, mark, registerAccountTasks } from "./housekeeping";
import type { AccountTask } from "./housekeeping";

const { insertInto, writeTo, scalar } = accountTaskHelpers;

const MINUTE_MS = 60 * 1000;
/** Budget alerts look at the period summary a few minutes after usage moves, not on every event. */
const ALERT_DEBOUNCE_MS = 5 * MINUTE_MS;
const GAUGE_DEBOUNCE_MS = MINUTE_MS;
const RESERVATIONS_PER_RUN = 200;
/** Debounced tasks that follow the account's usage and gauges. */
const USAGE_MOVED = ["alerts", "usage-leaderboard"];

/** First instant of the month after `now` (UTC), when gauges carry into the new billing period. */
function nextPeriodStart(now: number): number {
  const date = new Date(now);
  return Date.UTC(date.getUTCFullYear(), date.getUTCMonth() + 1, 1);
}

/**
 * The account's deadline-driven work (see housekeeping.ts for the machinery). Gauge and usage writers mark
 * `alerts`, so budget alerts follow usage without any sweep over accounts.
 */
const TASKS: AccountTask[] = [
  {
    name: "digest",
    arms: insertInto("notification_digest_items"),
    due: (sql) => scalar(sql, "SELECT MIN(flush_after) AS due FROM notification_digest_items"),
    run: (accountId, now) => flushAccountDigests(accountId, now),
  },
  {
    // Before `reservations`: an expiring session is settled from its hold while the hold is still there.
    name: "editor-sessions",
    arms: insertInto("editor_container_sessions"),
    due: (sql) => scalar(sql, EXPIRING_SESSION_DUE),
    run: async (accountId, now) => void (await closeExpiredContainerSessions(accountId, now)),
  },
  {
    // A hold whose holder crashed must stop counting against the quota when it expires.
    name: "reservations",
    arms: insertInto("usage_reservations"),
    due: (sql) => scalar(sql, "SELECT MIN(expires_at) AS due FROM usage_reservations"),
    run: async (_accountId, now, sql) => {
      const expired = sql
        .exec<{ id: string }>(
          "SELECT id FROM usage_reservations WHERE expires_at < ? LIMIT ?",
          now,
          RESERVATIONS_PER_RUN,
        )
        .toArray();
      for (const row of expired)
        for (const statement of releaseStatements(row.id)) sql.exec(statement.sql, ...statement.params);
    },
  },
  {
    name: "scope-gauges",
    marks: writeTo("scope_gauges"),
    debounce: GAUGE_DEBOUNCE_MS,
    run: async (accountId, now, sql) => {
      sumScopeGauges(sql, accountId, now);
      mark(sql, USAGE_MOVED, now);
    },
  },
  {
    name: "lfs-gauge",
    marks: writeTo("lfs_usage"),
    debounce: GAUGE_DEBOUNCE_MS,
    run: async (accountId, now, sql) => {
      const bytes = scalar(sql, "SELECT storage_bytes AS due FROM lfs_usage WHERE owner_id = ?", accountId) ?? 0;
      const billed = await resolveAccount(accountId);
      const statement = gaugeStatement(billed, "storage.lfs", bytes, now);
      if (billed === accountId) {
        sql.exec(statement.sql, ...statement.params);
        mark(sql, USAGE_MOVED, now);
      } else {
        // A claimed scratch account's storage is billed to its claimer.
        await accountDb(billed).batch([statement]);
      }
    },
  },
  {
    // Gauges are billed per period at their peak; a new period starts from the level the old one ended at.
    name: "gauge-carryover",
    due: (sql, now) => {
      const period = periodOf(now);
      const behind = sql
        .exec(
          `SELECT 1 AS one FROM usage_gauges g WHERE g.current > 0 AND g.period < ?1
             AND NOT EXISTS (SELECT 1 FROM usage_gauges h WHERE h.account_id = g.account_id AND h.meter = g.meter AND h.period >= ?1)
           LIMIT 1`,
          period,
        )
        .toArray();
      if (behind.length > 0) return now;
      const stored = sql.exec("SELECT 1 AS one FROM usage_gauges WHERE current > 0 LIMIT 1").toArray();
      return stored.length > 0 ? nextPeriodStart(now) : null;
    },
    run: async (_accountId, now, sql) => {
      const period = periodOf(now);
      sql.exec(
        `INSERT INTO usage_gauges (account_id, meter, period, current, peak, updated_at)
         SELECT g.account_id, g.meter, ?1, g.current, g.current, ?2 FROM usage_gauges g
         WHERE g.current > 0 AND g.period < ?1
           AND g.period = (SELECT MAX(h.period) FROM usage_gauges h WHERE h.account_id = g.account_id AND h.meter = g.meter)
         ON CONFLICT (account_id, meter, period) DO NOTHING`,
        period,
        now,
      );
      mark(sql, ["usage-leaderboard"], now);
    },
  },
  {
    // Heartbeats only push a runner's offline time later, so they arm lazily.
    name: "offline-runners",
    arms: /^\s*UPDATE\s+runners\s+SET\s+last_seen_at\b/i,
    lazy: true,
    due: (sql, now) => nextRunnerOffline(sql, now),
    run: (_accountId, now, sql) => reportOfflineRunners(sql, now),
  },
  {
    name: "usage-leaderboard",
    marks: /^\s*INSERT\s+(?:OR\s+\w+\s+)?INTO\s+(usage_events|usage_periods|usage_gauges)\b/i,
    debounce: ALERT_DEBOUNCE_MS,
    run: (accountId, now, sql) => publishUsageLeaderboard(accountId, sql, now),
  },
  {
    name: "alerts",
    marks: /^\s*INSERT\s+(?:OR\s+\w+\s+)?INTO\s+(usage_events|usage_gauges)\b/i,
    debounce: ALERT_DEBOUNCE_MS,
    run: (accountId, now) => sendAccountBudgetAlerts(accountId, now),
  },
];

registerAccountTasks(TASKS);