import { beforeEach, describe, expect, mock, test } from "bun:test";
import { Database } from "bun:sqlite";
import { applyCombinedSchema } from "./test-schema.test-helper";
import { fakeStorage } from "./do-test-storage";

mock.module("cloudflare:workers", () => ({
  env: {},
  DurableObject: class {
    ctx: unknown;
    constructor(ctx: unknown) {
      this.ctx = ctx;
    }
  },
}));

type Statement = { sql: string; params: unknown[] };
let seeds: Record<string, Record<string, number>> = {};
let flushed: Statement[][] = [];
let failing: (statement: Statement) => boolean = () => false;

const seeded = async (sql: string, ...params: unknown[]) => {
  if (sql.startsWith("SELECT 1 AS ok")) return { ok: 1 };
  for (const [fragment, rows] of Object.entries(seeds)) {
    if (sql.includes(fragment)) return rows[String(params.at(-1))] !== undefined ? rows[String(params.at(-1))] : null;
  }
  return null;
};
const record = async (statements: Statement[]) => {
  if (statements.some(failing)) throw new Error("D1 rejected the batch");
  flushed.push(statements);
  return [];
};
const stmt = (sql: string, ...params: unknown[]): Statement => ({ sql, params });
mock.module("./db", () => ({ stmt, first: seeded, batch: record }));
// Every repository and account object shares the recorder: the tests check statements, not which store.
const scoped = (scopeId: string) => ({ scopeId, first: seeded, batch: record });
mock.module("./scoped-db", () => ({ stmt, repoDb: scoped, accountDb: scoped }));
mock.module("./scoped-db/locators", () => ({ locate: async (_kind: string, key: string) => `repo-of-${key}` }));
mock.module("./sites/caps", () => ({ afterHostingFlush: async () => undefined }));

const { Counters } = await import("./counters-do");
const { REPO_PACK_BYTES, REPO_STARS, discussionUpvotesKey, usageKey } = await import("./counter-keys");
const { backoffMs, ISOLATE_AFTER } = await import("./do-flush");

function counters() {
  const storage = fakeStorage();
  const object = new Counters({ storage } as never, {} as never);
  return { object, storage };
}

beforeEach(() => {
  seeds = {};
  flushed = [];
  failing = () => false;
});

describe("mcp-usage counters", () => {
  test("seeds from D1, accumulates calls and errors, and flushes absolute totals", async () => {
    seeds = { "FROM mcp_usage_daily": { search: { calls: 5, errors: 1 } as never } };
    const { object } = counters();
    const key = usageKey("2026-10-12", "search");
    await object.add("mcp-usage", "acct1", key, [1, 0, 0], null);
    expect(await object.add("mcp-usage", "acct1", key, [1, 1, 0], null)).toEqual([7, 2, 0]);
    await object.alarm();
    expect(flushed).toHaveLength(1);
    const [statement] = flushed[0];
    expect(statement.sql).toContain("INSERT INTO mcp_usage_daily");
    expect(statement.sql).toContain("calls = excluded.calls");
    expect(statement.params).toEqual(["acct1", "2026-10-12", "search", 7, 2]);
  });

  test("a tool name containing the separator round-trips", async () => {
    const { object } = counters();
    await object.add("mcp-usage", "acct1", usageKey("2026-10-12", "a|b"), [1, 0, 0], null);
    await object.alarm();
    expect(flushed[0][0].params.slice(1, 3)).toEqual(["2026-10-12", "a|b"]);
  });
});

describe("repo-counts", () => {
  test("stars, pack bytes and upvotes flush to their own columns", async () => {
    seeds = { "FROM discussion_votes": { disc1: { value: 1 } as never } };
    const { object } = counters();
    await object.add("repo-counts", "repo1", REPO_STARS, [1, 0, 0], null);
    await object.add("repo-counts", "repo1", REPO_PACK_BYTES, [2048, 0, 0], null);
    await object.add("repo-counts", "repo1", discussionUpvotesKey("disc1"), [1, 0, 0], null);
    await object.alarm();
    // Star counts go to D1, pack bytes and upvotes to the repository's object: one batch each.
    const sqls = flushed.flat().map((statement) => statement.sql);
    expect(sqls.some((sql) => sql.includes("stars_count = (SELECT COUNT(*) FROM stars"))).toBe(true);
    expect(sqls.some((sql) => sql.includes("repo_storage_estimates") && sql.includes("excluded.pack_bytes"))).toBe(
      true,
    );
    expect(flushed.flat().find((statement) => statement.sql.includes("discussions"))?.params).toEqual(["disc1"]);
  });
});

describe("repo-counts flush against the real schema", () => {
  test("stars and upvotes recount source rows, so deletes that bypassed the counter are not undone", async () => {
    const db = new Database(":memory:");
    applyCombinedSchema(db);
    db.exec(`INSERT INTO accounts (id, name, kind, created_at) VALUES ('u1', 'u1', 'user', 0), ('u2', 'u2', 'user', 0);
      INSERT INTO repos (id, owner_id, name, visibility, default_branch, artifacts_name, artifacts_remote, created_at, updated_at)
        VALUES ('repo1', 'u1', 'r', 'public', 'main', 'a', 'x', 0, 0);
      INSERT INTO stars VALUES ('u1', 'repo1', 0), ('u2', 'repo1', 0);
      INSERT INTO discussion_votes VALUES ('disc1', 'u1', 0), ('disc1', 'u2', 0);`);
    const { object } = counters();
    seeds = {
      "FROM stars": { repo1: { value: 2 } as never },
      "FROM discussion_votes": { disc1: { value: 2 } as never },
    };
    await object.add("repo-counts", "repo1", REPO_STARS, [1, 0, 0], null);
    await object.add("repo-counts", "repo1", discussionUpvotesKey("disc1"), [1, 0, 0], null);
    db.exec("PRAGMA foreign_keys = ON; DELETE FROM accounts WHERE id = 'u2'");
    await object.alarm();
    const starStatement = flushed.flat().find((statement) => statement.sql.includes("stars_count"));
    const voteStatement = flushed.flat().find((statement) => statement.sql.includes("discussion_votes"));
    if (!starStatement || !voteStatement) throw new Error("missing flush statements");
    db.query(starStatement.sql).run(...(starStatement.params as never[]));
    expect(db.query("SELECT stars_count FROM repos WHERE id = 'repo1'").get()).toEqual({ stars_count: 1 });
    expect(
      db
        .query(voteStatement.sql.replace(/UPDATE \w+ SET upvotes = (\(.*\)) WHERE id = \?1/s, "SELECT $1 AS n"))
        .get(...(voteStatement.params as never[])),
    ).toEqual({ n: 1 });
  });
});

describe("repo-counts seeding", () => {
  test("a star already written to D1 is not counted twice when it seeds the counter", async () => {
    seeds = { "FROM stars": { repo1: { value: 4 } as never } };
    const { object } = counters();
    expect(await object.add("repo-counts", "repo1", REPO_STARS, [1, 0, 0], null)).toEqual([4, 0, 0]);
    expect(await object.add("repo-counts", "repo1", REPO_STARS, [-1, 0, 0], null)).toEqual([3, 0, 0]);
  });

  test("pack bytes add the push on top of the stored estimate", async () => {
    seeds = { "FROM repo_storage_estimates": { repo1: { value: 100 } as never } };
    const { object } = counters();
    expect(await object.add("repo-counts", "repo1", REPO_PACK_BYTES, [50, 0, 0], null)).toEqual([150, 0, 0]);
  });
});

describe("flush failures", () => {
  test("backs off exponentially and keeps rows dirty", async () => {
    const { object, storage } = counters();
    await object.add("release-downloads", "repo1", "asset1", [1, 0, 0], null);
    failing = () => true;
    const before = Date.now();
    await object.alarm();
    const first = (storage.alarmAt ?? 0) - before;
    await object.alarm();
    const second = (storage.alarmAt ?? 0) - before;
    expect(second).toBeGreaterThan(first);
    expect(first).toBeGreaterThanOrEqual(backoffMs(1, 60_000) - 1000);
    failing = () => false;
    await object.alarm();
    expect(flushed.at(-1)?.[0].params).toEqual([1, "asset1"]);
    expect(await storage.get("attempts")).toBeUndefined();
  });

  test("after repeated failures a poison row is dead-lettered and the rest flush", async () => {
    const { object } = counters();
    await object.add("release-downloads", "repo1", "bad", [1, 0, 0], null);
    await object.add("release-downloads", "repo1", "good", [3, 0, 0], null);
    failing = (statement) => statement.params.includes("bad");
    for (let attempt = 0; attempt <= ISOLATE_AFTER; attempt++) await object.alarm();
    expect(flushed.flat().map((statement) => statement.params)).toEqual([[3, "good"]]);
    const dead = await object.deadLetters();
    expect(dead.map((row) => row.key)).toEqual(["bad"]);
    await object.alarm();
    expect(flushed).toHaveLength(1);
  });
});

describe("reset keeps unflushed increments", () => {
  test("pack bytes added since the last flush survive a reset and land on top of the recomputed total", async () => {
    seeds = { "FROM repo_storage_estimates": { repo1: { value: 100 } as never } };
    const { object } = counters();
    await object.add("repo-counts", "repo1", REPO_PACK_BYTES, [50, 0, 0], null);
    await object.alarm();
    expect(flushed.at(-1)?.[0].params[1]).toBe(150);
    await object.add("repo-counts", "repo1", REPO_PACK_BYTES, [30, 0, 0], null);
    seeds = { "FROM repo_storage_estimates": { repo1: { value: 1000 } as never } };
    await object.reset(REPO_PACK_BYTES);
    await object.alarm();
    expect(flushed.at(-1)?.[0].params[1]).toBe(1030);
    expect(await object.add("repo-counts", "repo1", REPO_PACK_BYTES, [5, 0, 0], null)).toEqual([1035, 0, 0]);
  });

  test("a reset before the first flush keeps everything added since the seed", async () => {
    seeds = { "FROM repo_storage_estimates": { repo1: { value: 100 } as never } };
    const { object } = counters();
    await object.add("repo-counts", "repo1", REPO_PACK_BYTES, [20, 0, 0], null);
    seeds = { "FROM repo_storage_estimates": { repo1: { value: 400 } as never } };
    await object.reset(REPO_PACK_BYTES);
    expect(await object.totals("repo-counts", "repo1", REPO_PACK_BYTES)).toEqual([420, 0, 0]);
  });

  test("a clean row is simply dropped and reseeded", async () => {
    seeds = { "FROM repo_storage_estimates": { repo1: { value: 100 } as never } };
    const { object } = counters();
    await object.add("repo-counts", "repo1", REPO_PACK_BYTES, [1, 0, 0], null);
    await object.alarm();
    seeds = { "FROM repo_storage_estimates": { repo1: { value: 7 } as never } };
    await object.reset(REPO_PACK_BYTES);
    expect(await object.totals("repo-counts", "repo1", REPO_PACK_BYTES)).toEqual([7, 0, 0]);
  });
});