import { ABANDONED_RESERVATION_MS, IDLE_EVICTION_MS, evictIdleCachesForRepo } from "../ci/cache-api";
import { JOB_TOKEN_MARKER_TTL_MS, purgeExpiredJobTokensForRepo } from "../ci/job-token";
import { reapStaleRunnerJobsForRepo, STALE_RUNNER_MS } from "../ci/runner-api";
import { fireDueSchedulesForRepo } from "../ci/workflow-index";
import { ATTACHMENT_ORPHAN_GRACE_MS, sweepOrphanedAttachmentsForRepo } from "../attachments";
import {
  dropHiddenRefsForRepo,
  MAX_CLEANUP_ATTEMPTS,
  sweepClosedPullRefsForRepo,
  UNSWEPT_CLOSED_PULLS,
} from "../git/pull-bridge";
import { WEBHOOK_DELIVERY_RETENTION_MS } from "../integrations/retention";
import type { ScopedValue } from "../scoped-db/types";
import { withJobPerf } from "../perf";
import { indexActivity } from "../activity-index";
import { publishScopeGauges } from "../metering/scope-gauges";
import { ROLLUP_GRACE_MS, rollupScopeSiteUsage } from "../sites/metering";
import { STALE_PULL_ALIASES, sweepClosedPullAliasesForScope } from "../sites/pr-aliases";
import { publishSiteDirectory } from "../sites/directory";

/** Never wake more often than this, however soon something is due (or if a task keeps failing). */
const MIN_INTERVAL_MS = 10 * 1000;
const WEBHOOK_PRUNE_BATCH = 500;
const PENDING_INDEX_BATCH = 200;
const CLOSED_PULL_GRACE_MS = 5 * 60 * 1000;
/** Garbage-collection tasks (expired tokens, idle caches, old deliveries) run at most this often. */
const GC_INTERVAL_MS = 60 * 60 * 1000;
/** A task that threw waits this long before its next attempt. */
const FAILURE_BACKOFF_MS = 5 * 60 * 1000;

type Sql = { exec<T extends Record<string, ScopedValue>>(sql: string, ...params: ScopedValue[]): { toArray(): T[] } };

/**
 * Time-driven work of one repository, run by its RepoState alarm instead of crons that scan every repository:
 * scheduled workflows, cache eviction, job token expiry, webhook delivery retention, orphaned attachments,
 * hidden-ref cleanup retries and stale self-hosted runner jobs. A write to a task's table arms the alarm; the
 * alarm runs the tasks that are due and re-arms for the earliest next one, so idle repositories cost nothing.
 */
type Task = {
  name: string;
  /**
   * Statements that can make this task due sooner than the pending alarm. Garbage-collection tasks list only
   * inserts; their rows only become due later, so a pending alarm already covers them.
   */
  arms: RegExp;
  /** SQL returning one row `{ due }`: when the task next has work (null when none). */
  due: string;
  /** Re-run no sooner than this after the last run, even while `due` stays in the past (polling and GC tasks). */
  every?: number;
  /** Deadline-driven (schedules, retries) rather than garbage collection: computed on every arming write. */
  precise?: boolean;
  /**
   * Debounced work: an arming write records a mark in `scope_dirty` (the first one wins) and `due` reads it, so
   * the task runs once a little after a burst of writes.
   */
  marks?: boolean;
  /** Resolves true when it stopped at its batch size: there is more, so `every` doesn't hold the next run back. */
  run(repoId: string, now: number, sql: Sql): Promise<boolean | void>;
};

/** When each task last ran, kept by the RepoState in its storage. */
export type LastRuns = Record<string, number>;

const insertInto = (table: string) => new RegExp(`^\\s*INSERT\\s+(?:OR\\s+\\w+\\s+)?INTO\\s+${table}\\b`, "i");
const writeTo = (table: string) =>
  new RegExp(`^\\s*(?:INSERT\\s+(?:OR\\s+\\w+\\s+)?INTO|UPDATE|REPLACE\\s+INTO)\\s+${table}\\b`, "i");

const GAUGE_DEBOUNCE_MS = 5 * 60 * 1000;
const SITE_DIRECTORY_DEBOUNCE_MS = 60 * 1000;
/** Epoch ms of the end of the earliest site_usage day this scope hasn't folded, plus the flush grace. */
const SITE_ROLLUP_DUE = `SELECT (CAST(strftime('%s', MIN(day)) AS INTEGER) + 86400) * 1000 + ${ROLLUP_GRACE_MS} AS due
  FROM site_usage WHERE day NOT IN (SELECT day FROM site_rollups)`;

const TASKS: Task[] = [
  {
    // Each finished day of the scope's site counters is folded into the billed meters once.
    name: "site-usage-rollup",
    arms: /^\s*INSERT\s+(?:OR\s+\w+\s+)?INTO\s+site_usage\b/i,
    due: SITE_ROLLUP_DUE,
    precise: true,
    run: (scopeId, now) => rollupScopeSiteUsage(scopeId, now),
  },
  {
    // Storage gauges follow the tables they sum, reported to the billed accounts (metering/scope-gauges.ts).
    name: "gauges",
    arms: /^\s*(?:INSERT\s+(?:OR\s+\w+\s+)?INTO|UPDATE|REPLACE\s+INTO|DELETE\s+FROM)\s+(action_caches|repo_storage_estimates|release_assets|releases|attachments|sites|site_deployments)\b/i,
    due: `SELECT at + ${GAUGE_DEBOUNCE_MS} AS due FROM scope_dirty WHERE name = 'gauges'`,
    precise: true,
    marks: true,
    run: (repoId) => publishScopeGauges(repoId),
  },
  {
    // The directory copy of the scope's sites (admin counts) follows the sites table a minute behind.
    name: "site-directory",
    arms: /^\s*(?:INSERT\s+(?:OR\s+\w+\s+)?INTO|UPDATE|REPLACE\s+INTO|DELETE\s+FROM)\s+sites\b/i,
    due: `SELECT at + ${SITE_DIRECTORY_DEBOUNCE_MS} AS due FROM scope_dirty WHERE name = 'site-directory'`,
    precise: true,
    marks: true,
    run: (scopeId, now, sql) => publishSiteDirectory(scopeId, sql, now),
  },
  {
    name: "activity-index",
    arms: insertInto("activity_index_pending"),
    due: "SELECT MIN(at) AS due FROM activity_index_pending",
    precise: true,
    run: async (_repoId, _now, sql) => {
      const pending = sql
        .exec<{ rowid: number; actor_id: string; repo_id: string; at: number; counted: number }>(
          "SELECT rowid, actor_id, repo_id, at, counted FROM activity_index_pending ORDER BY rowid LIMIT ?",
          PENDING_INDEX_BATCH,
        )
        .toArray();
      if (pending.length === 0) return false;
      await indexActivity(
        pending.map((row) => ({ actorId: row.actor_id, repoId: row.repo_id, at: row.at, counted: row.counted === 1 })),
      );
      sql.exec("DELETE FROM activity_index_pending WHERE rowid <= ?", pending.at(-1)?.rowid ?? 0);
      return pending.length === PENDING_INDEX_BATCH;
    },
  },
  {
    name: "schedules",
    arms: writeTo("workflow_schedules"),
    due: "SELECT MIN(next_run_at) AS due FROM workflow_schedules",
    precise: true,
    run: (repoId, now) => fireDueSchedulesForRepo(repoId, now),
  },
  {
    name: "caches",
    arms: insertInto("action_caches"),
    due: `SELECT MIN(last_used_at) + ${IDLE_EVICTION_MS} AS due FROM action_caches`,
    every: GC_INTERVAL_MS,
    run: (repoId, now) => evictIdleCachesForRepo(repoId, now),
  },
  {
    // An abandoned upload reservation blocks its key until evicted, so it is a deadline, not garbage.
    name: "cache-reservations",
    arms: insertInto("action_caches"),
    due: `SELECT MIN(created_at) + ${ABANDONED_RESERVATION_MS} AS due FROM action_caches WHERE size < 0`,
    precise: true,
    run: (repoId, now) => evictIdleCachesForRepo(repoId, now),
  },
  {
    name: "job-tokens",
    arms: /^\s*INSERT\s+(?:OR\s+\w+\s+)?INTO\s+(job_tokens|job_token_pushes)\b/i,
    due: `SELECT MIN(due) AS due FROM (SELECT MIN(expires_at) AS due FROM job_tokens
      UNION ALL SELECT MIN(created_at) + ${JOB_TOKEN_MARKER_TTL_MS} FROM job_token_pushes)`,
    every: GC_INTERVAL_MS,
    run: (repoId, now) => purgeExpiredJobTokensForRepo(repoId, now),
  },
  {
    name: "webhook-deliveries",
    arms: insertInto("webhook_deliveries"),
    due: `SELECT MIN(created_at) + ${WEBHOOK_DELIVERY_RETENTION_MS} AS due FROM webhook_deliveries`,
    every: GC_INTERVAL_MS,
    run: async (_repoId, now, sql) => {
      const removed = sql
        .exec<{ id: string }>(
          "DELETE FROM webhook_deliveries WHERE id IN (SELECT id FROM webhook_deliveries WHERE created_at < ? LIMIT ?) RETURNING id",
          now - WEBHOOK_DELIVERY_RETENTION_MS,
          WEBHOOK_PRUNE_BATCH,
        )
        .toArray();
      return removed.length === WEBHOOK_PRUNE_BATCH;
    },
  },
  {
    name: "attachments",
    arms: insertInto("attachments"),
    due: `SELECT MIN(created_at) + ${ATTACHMENT_ORPHAN_GRACE_MS} AS due FROM attachments`,
    every: GC_INTERVAL_MS,
    run: (repoId, now) => sweepOrphanedAttachmentsForRepo(repoId, now),
  },
  {
    name: "hidden-refs",
    arms: writeTo("hidden_ref_cleanups"),
    due: `SELECT MIN(next_attempt_at) AS due FROM hidden_ref_cleanups WHERE attempts < ${MAX_CLEANUP_ATTEMPTS}`,
    precise: true,
    run: (repoId) => dropHiddenRefsForRepo(repoId),
  },
  {
    // A grace period after closing lets the close-time drop (which runs right away) win first. Only a write that
    // closes something (sets closed_at) can make it due sooner.
    name: "closed-pull-refs",
    arms: /^\s*UPDATE\s+issues\b[\s\S]*\bclosed_at\b/i,
    due: `SELECT MIN(COALESCE(i.closed_at, 0)) + ${CLOSED_PULL_GRACE_MS} AS due FROM pull_requests p
      JOIN issues i ON i.id = p.issue_id WHERE ${UNSWEPT_CLOSED_PULLS}`,
    // A drop that keeps failing leaves the pull due; retry it at a modest pace rather than at the floor interval.
    every: CLOSED_PULL_GRACE_MS * 3,
    precise: true,
    run: (repoId) => sweepClosedPullRefsForRepo(repoId),
  },
  {
    // Pull request and commit preview hosts of pulls that closed without the close hook dropping them.
    name: "closed-pull-aliases",
    arms: /^\s*(?:UPDATE\s+issues\b[\s\S]*\bstate\b|DELETE\s+FROM\s+issues\b|INSERT\s+(?:OR\s+\w+\s+)?INTO\s+site_pr_aliases\b)/i,
    due: `SELECT MIN(COALESCE(i.closed_at, 0)) + ${CLOSED_PULL_GRACE_MS} AS due FROM ${STALE_PULL_ALIASES}`,
    every: CLOSED_PULL_GRACE_MS * 3,
    precise: true,
    run: (scopeId) => sweepClosedPullAliasesForScope(scopeId),
  },
  {
    // Only claiming a job (status set to in_progress) can start the clock.
    name: "stale-runner-jobs",
    arms: /^\s*UPDATE\s+workflow_jobs\b[\s\S]*'in_progress'/i,
    due: `SELECT MIN(started_at) + ${STALE_RUNNER_MS} AS due FROM workflow_jobs
      WHERE status = 'in_progress' AND runner_id IS NOT NULL`,
    every: STALE_RUNNER_MS,
    precise: true,
    run: (repoId, now) => reapStaleRunnerJobsForRepo(repoId, now),
  },
];

/** Whether a statement may change when one of the repository's tasks is next due. */
export function armsRepoHousekeeping(sql: string): boolean {
  return TASKS.some((task) => task.arms.test(sql));
}

/** Debounced tasks a statement makes due. */
export function repoMarksOf(sql: string): string[] {
  return TASKS.filter((task) => task.marks && task.arms.test(sql)).map((task) => task.name);
}

/** Records debounced work; the first mark wins, so a stream of writes can't postpone it forever. */
export function markRepo(sql: Sql, names: string[], now: number): void {
  for (const name of names) sql.exec("INSERT OR IGNORE INTO scope_dirty (name, at) VALUES (?, ?)", name, now);
}

/** Whether it may make a deadline-driven task due sooner than an alarm that is already pending. */
export function armsPreciseRepoTask(sql: string): boolean {
  return TASKS.some((task) => task.precise && task.arms.test(sql));
}

function dueOf(sql: Sql, task: Task, lastRuns: LastRuns): number | null {
  const row = sql.exec<{ due: number | null }>(task.due).toArray()[0];
  if (row?.due == null) return null;
  const last = lastRuns[task.name];
  const failed = lastRuns[`${task.name}:failed`];
  let due = Number(row.due);
  if (task.every && last !== undefined) due = Math.max(due, last + task.every);
  if (failed !== undefined) due = Math.max(due, failed + FAILURE_BACKOFF_MS);
  return due;
}

/** When the alarm should next run, or null when nothing is waiting. */
export function nextRepoDue(sql: Sql, now: number, lastRuns: LastRuns): number | null {
  const times = TASKS.map((task) => dueOf(sql, task, lastRuns)).filter((due): due is number => due !== null);
  if (times.length === 0) return null;
  return Math.max(Math.min(...times), now + MIN_INTERVAL_MS);
}

/**
 * Runs every due task and records when it ran; one failing never stops the others (it stays due and is retried
 * on the next alarm).
 */
export async function runRepoHousekeeping(repoId: string, sql: Sql, now: number, lastRuns: LastRuns): Promise<void> {
  for (const task of TASKS) {
    const due = dueOf(sql, task, lastRuns);
    if (due === null || due > now) continue;
    lastRuns[task.name] = now;
    // Cleared before running: writes during the run mark it again.
    if (task.marks) sql.exec("DELETE FROM scope_dirty WHERE name = ?", task.name);
    try {
      const more = await withJobPerf(`alarm.repo.${task.name}`, () => task.run(repoId, now, sql));
      delete lastRuns[`${task.name}:failed`];
      // A full batch means a backlog: let the next alarm (at the floor interval) continue it.
      if (more === true) delete lastRuns[task.name];
    } catch (error) {
      lastRuns[`${task.name}:failed`] = now;
      if (task.marks) markRepo(sql, [task.name], now);
      console.error("Repository housekeeping task failed", task.name, repoId, error);
    }
  }
}