import { describe, expect, test } from "vitest";

import type { PackCatalogRow } from "@/worker/do/repo/db/schema";
import {
  AUTO_COMPACTION_MAX_SOURCE_BYTES,
  AUTO_COMPACTION_MAX_SOURCE_OBJECTS,
  catalogNeedsCompaction,
  selectCompactionWork,
} from "@/worker/do/repo/catalog/compaction/plan";

function packRow(args: {
  seqLo: number;
  seqHi?: number;
  tier?: number;
  kind?: PackCatalogRow["kind"];
  objectCount?: number;
  packBytes?: number;
}): PackCatalogRow {
  const seqHi = args.seqHi ?? args.seqLo;
  const tier = args.tier ?? 0;
  return {
    packKey: `pack-t${tier}-${args.seqLo}-${seqHi}.pack`,
    kind: args.kind ?? "receive",
    state: "active",
    tier,
    seqLo: args.seqLo,
    seqHi,
    objectCount: args.objectCount ?? 10,
    packBytes: args.packBytes ?? 1024,
    idxBytes: 128,
    createdAt: args.seqLo,
    supersededBy: null,
  };
}

function selectedSeqs(rows: PackCatalogRow[]): number[] {
  const selection = selectCompactionWork(rows);
  expect(selection.status).toBe("ready");
  if (selection.status !== "ready") return [];
  return selection.plan.sourcePacks.map((pack) => pack.seqLo);
}

describe("automatic compaction planning", () => {
  test("skips an oversized oldest base pack and compacts the next closed window", () => {
    const rows = [
      packRow({ seqLo: 5, objectCount: 113, packBytes: 71_514 }),
      packRow({ seqLo: 4, objectCount: 1_142, packBytes: 283_549 }),
      packRow({ seqLo: 3, objectCount: 2_042, packBytes: 673_559 }),
      packRow({ seqLo: 2, objectCount: 1_025, packBytes: 348_376 }),
      packRow({
        seqLo: 1,
        objectCount: 79_056,
        packBytes: AUTO_COMPACTION_MAX_SOURCE_BYTES + 1,
      }),
    ];

    expect(selectedSeqs(rows)).toEqual([2, 3, 4, 5]);
    expect(catalogNeedsCompaction(rows)).toBe(true);
  });

  test("preserves oldest-first behavior when the oldest closed window fits the budget", () => {
    const rows = [
      packRow({ seqLo: 5 }),
      packRow({ seqLo: 4 }),
      packRow({ seqLo: 3 }),
      packRow({ seqLo: 2 }),
      packRow({ seqLo: 1 }),
    ];

    expect(selectedSeqs(rows)).toEqual([1, 2, 3, 4]);
  });

  test("rejects a window whose sequence interval overlaps an unselected active pack", () => {
    const rows = [
      packRow({ seqLo: 6 }),
      packRow({ seqLo: 5 }),
      packRow({ seqLo: 4 }),
      packRow({ seqLo: 3 }),
      packRow({ seqLo: 1 }),
      packRow({ seqLo: 2, tier: 1 }),
    ];

    expect(selectedSeqs(rows)).toEqual([3, 4, 5, 6]);
  });

  test("rejects source windows overlapped by a compact range from another tier", () => {
    const rows = [
      packRow({ seqLo: 8 }),
      packRow({ seqLo: 7 }),
      packRow({ seqLo: 6 }),
      packRow({ seqLo: 5 }),
      packRow({ seqLo: 4 }),
      packRow({ seqLo: 3 }),
      packRow({ seqLo: 2 }),
      packRow({ seqLo: 1 }),
      packRow({ seqLo: 1, seqHi: 4, tier: 1, kind: "compact" }),
    ];

    expect(selectedSeqs(rows)).toEqual([5, 6, 7, 8]);
  });

  test("uses a later overflowing tier when an earlier tier only has over-budget windows", () => {
    const perPackObjects = Math.floor(AUTO_COMPACTION_MAX_SOURCE_OBJECTS / 4) + 1;
    const rows = [
      packRow({ seqLo: 5, objectCount: perPackObjects }),
      packRow({ seqLo: 4, objectCount: perPackObjects }),
      packRow({ seqLo: 3, objectCount: perPackObjects }),
      packRow({ seqLo: 2, objectCount: perPackObjects }),
      packRow({ seqLo: 1, objectCount: perPackObjects }),
      packRow({ seqLo: 14, tier: 1 }),
      packRow({ seqLo: 13, tier: 1 }),
      packRow({ seqLo: 12, tier: 1 }),
      packRow({ seqLo: 11, tier: 1 }),
      packRow({ seqLo: 10, tier: 1 }),
    ];

    const selection = selectCompactionWork(rows);
    expect(selection.status).toBe("ready");
    if (selection.status !== "ready") return;
    expect(selection.plan.sourceTier).toBe(1);
    expect(selection.plan.sourcePacks.map((pack) => pack.seqLo)).toEqual([10, 11, 12, 13]);
  });

  test("finds a later closed window in a large catalog with many overlapped windows", () => {
    const rows: PackCatalogRow[] = [];
    for (let seq = 604; seq >= 1; seq--) {
      rows.push(packRow({ seqLo: seq }));
    }
    rows.push(packRow({ seqLo: 1, seqHi: 600, tier: 1, kind: "compact" }));

    expect(selectedSeqs(rows)).toEqual([601, 602, 603, 604]);
  });

  test("blocks automatic compaction when every closed window exceeds the budget", () => {
    const perPackObjects = Math.floor(AUTO_COMPACTION_MAX_SOURCE_OBJECTS / 4) + 1;
    const rows = [
      packRow({ seqLo: 5, objectCount: perPackObjects }),
      packRow({ seqLo: 4, objectCount: perPackObjects }),
      packRow({ seqLo: 3, objectCount: perPackObjects }),
      packRow({ seqLo: 2, objectCount: perPackObjects }),
      packRow({ seqLo: 1, objectCount: perPackObjects }),
    ];

    const selection = selectCompactionWork(rows);
    expect(selection.status).toBe("blocked");
    if (selection.status !== "blocked") return;
    expect(selection.blocked.reason).toBe("over-budget");
    expect(selection.blocked.smallestWindowObjects).toBe(perPackObjects * 4);
    expect(catalogNeedsCompaction(rows)).toBe(false);
  });

  test("blocks automatic compaction when no closed sequence window exists", () => {
    const rows = [
      packRow({ seqLo: 9 }),
      packRow({ seqLo: 7 }),
      packRow({ seqLo: 5 }),
      packRow({ seqLo: 3 }),
      packRow({ seqLo: 1 }),
      packRow({ seqLo: 8, tier: 1 }),
      packRow({ seqLo: 6, tier: 1 }),
      packRow({ seqLo: 4, tier: 1 }),
      packRow({ seqLo: 2, tier: 1 }),
    ];

    const selection = selectCompactionWork(rows);
    expect(selection.status).toBe("blocked");
    if (selection.status !== "blocked") return;
    expect(selection.blocked.reason).toBe("non-contiguous-window");
    expect(catalogNeedsCompaction(rows)).toBe(false);
  });
});