import { DurableObject } from "cloudflare:workers";
import { batch, first, stmt } from "./db";

/** How long increments sit in the object before D1 sees them; reads that must be exact go to the object. */
const FLUSH_DELAY_MS = 10_000;
const RETRY_DELAY_MS = 60_000;
const FLUSH_BATCH = 200;
/** Rows untouched this long are dropped once flushed, so an object only holds its working set. */
const IDLE_ROW_MS = 3 * 24 * 60 * 60 * 1000;

export type CounterKind = "site-usage" | "release-downloads";
export type Totals = [number, number, number];
type Row = { key: string; a: number; b: number; c: number; meta: string | null };

type Kind = {
  /** Current D1 totals for a key the object has not seen yet, so taking over a live counter loses nothing. */
  seed(scope: string, key: string): Promise<Totals>;
  /** Writes absolute totals, so a retried flush is idempotent. */
  flush(scope: string, rows: Row[]): ReturnType<typeof stmt>[];
};

const KINDS: Record<CounterKind, Kind> = {
  /** scope = site id, key = UTC day, totals = requests / cpu ms / bytes out, meta = billed account id. */
  "site-usage": {
    async seed(siteId, day) {
      const row = await first<{ requests: number; cpu_ms: number; bytes_out: number }>(
        "SELECT requests, cpu_ms, bytes_out FROM site_usage WHERE site_id = ? AND day = ?",
        siteId,
        day,
      );
      return [row?.requests ?? 0, row?.cpu_ms ?? 0, row?.bytes_out ?? 0];
    },
    flush(siteId, rows) {
      return rows.map((row) =>
        stmt(
          `INSERT INTO site_usage (site_id, day, account_id, requests, cpu_ms, bytes_out)
           SELECT ?1, ?2, ?3, ?4, ?5, ?6 WHERE EXISTS (SELECT 1 FROM sites WHERE id = ?1)
           ON CONFLICT (site_id, day) DO UPDATE SET requests = excluded.requests, cpu_ms = excluded.cpu_ms,
             bytes_out = excluded.bytes_out, account_id = excluded.account_id`,
          siteId,
          row.key,
          row.meta,
          row.a,
          row.b,
          row.c,
        ),
      );
    },
  },
  /** scope = repo id, key = release asset id, totals[0] = downloads. */
  "release-downloads": {
    async seed(_repoId, assetId) {
      const row = await first<{ download_count: number }>(
        "SELECT download_count FROM release_assets WHERE id = ?",
        assetId,
      );
      return [row?.download_count ?? 0, 0, 0];
    },
    flush(_repoId, rows) {
      return rows.map((row) => stmt("UPDATE release_assets SET download_count = ? WHERE id = ?", row.a, row.key));
    },
  },
};

type Scope = { kind: CounterKind; scope: string };

/**
 * Write-heavy counters (per-request site usage, asset downloads) accumulate in one object per scope instead
 * of being an UPDATE on D1's single writer each time. D1 receives absolute totals at most every
 * FLUSH_DELAY_MS per object, so its load is bounded by the number of active scopes rather than traffic.
 */
export class Counters extends DurableObject<Env> {
  private ready = false;
  private scopeSaved = false;

  private init(): void {
    if (this.ready) return;
    this.ctx.storage.sql.exec(`CREATE TABLE IF NOT EXISTS counters (
      key TEXT PRIMARY KEY,
      a INTEGER NOT NULL,
      b INTEGER NOT NULL,
      c INTEGER NOT NULL,
      meta TEXT,
      dirty INTEGER NOT NULL DEFAULT 0,
      touched_at INTEGER NOT NULL
    )`);
    this.ready = true;
  }

  private read(key: string): Row | null {
    return (
      this.ctx.storage.sql.exec<Row>("SELECT key, a, b, c, meta FROM counters WHERE key = ?", key).toArray()[0] ?? null
    );
  }

  /** Adds to a key's totals and returns the new totals. */
  async add(kind: CounterKind, scope: string, key: string, delta: Totals, meta: string | null): Promise<Totals> {
    this.init();
    if (!this.scopeSaved) {
      await this.ctx.storage.put<Scope>("scope", { kind, scope });
      this.scopeSaved = true;
    }
    if (!this.read(key)) {
      const seeded = await KINDS[kind].seed(scope, key);
      this.ctx.storage.sql.exec(
        "INSERT OR IGNORE INTO counters (key, a, b, c, meta, touched_at) VALUES (?, ?, ?, ?, ?, ?)",
        key,
        ...seeded,
        meta,
        Date.now(),
      );
    }
    const row = this.ctx.storage.sql
      .exec<Row>(
        `UPDATE counters SET a = a + ?, b = b + ?, c = c + ?, meta = ?, dirty = 1, touched_at = ?
         WHERE key = ? RETURNING key, a, b, c, meta`,
        ...delta,
        meta,
        Date.now(),
        key,
      )
      .one();
    if ((await this.ctx.storage.getAlarm()) === null) await this.ctx.storage.setAlarm(Date.now() + FLUSH_DELAY_MS);
    return [row.a, row.b, row.c];
  }

  /** Exact current totals (D1 may lag by up to FLUSH_DELAY_MS). */
  async totals(kind: CounterKind, scope: string, key: string): Promise<Totals> {
    this.init();
    const row = this.read(key);
    return row ? [row.a, row.b, row.c] : KINDS[kind].seed(scope, key);
  }

  override async alarm(): Promise<void> {
    this.init();
    const target = await this.ctx.storage.get<Scope>("scope");
    if (!target) return;
    const sql = this.ctx.storage.sql;
    const rows = sql
      .exec<Row>("SELECT key, a, b, c, meta FROM counters WHERE dirty = 1 LIMIT ?", FLUSH_BATCH)
      .toArray();
    try {
      if (rows.length > 0) await batch(KINDS[target.kind].flush(target.scope, rows));
    } catch (error) {
      console.error("Counter flush failed", target, error);
      await this.ctx.storage.setAlarm(Date.now() + RETRY_DELAY_MS);
      return;
    }
    // Only rows unchanged since they were read are clean; anything added during the flush stays dirty.
    for (const row of rows) {
      sql.exec(
        "UPDATE counters SET dirty = 0 WHERE key = ? AND a = ? AND b = ? AND c = ? AND meta IS ?",
        row.key,
        row.a,
        row.b,
        row.c,
        row.meta,
      );
    }
    sql.exec("DELETE FROM counters WHERE dirty = 0 AND touched_at < ?", Date.now() - IDLE_ROW_MS);
    const pending = sql.exec<{ count: number }>("SELECT COUNT(*) AS count FROM counters WHERE dirty = 1").one().count;
    if (pending > 0) await this.ctx.storage.setAlarm(Date.now() + FLUSH_DELAY_MS);
  }
}