import { Database } from "bun:sqlite"; import { beforeAll, describe, expect, test } from "bun:test"; import { cp, mkdir, mkdtemp, readFile, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { deflateSync } from "node:zlib"; import { type Bucket, openBucket } from "../src/bucket/reader.ts"; import { readSave } from "../src/bucket/repo.ts"; import { syncAll, syncRepo } from "../src/bucket/sync.ts"; import { fileAt, filesAt, flagReason, headId, history, objectBytes, repoStats, saveByPrefix, timeline, } from "../src/mirror/queries.ts"; import { applyHidden, closeMirror, mirrorState, openMirror } from "../src/mirror/store.ts"; import { openKeprDb } from "../src/server/db.ts"; import { Browser, instance } from "./browser.ts"; import { type Fixture, makeKohBucket } from "./fixture.ts"; const text = (bytes: Uint8Array | null) => new TextDecoder().decode(bytes ?? new Uint8Array()); async function syncTarget(fx: Fixture) { const dataDir = await mkdtemp(join(tmpdir(), "kepr-data-")); const target = { bucket: openBucket({ kind: "folder", path: fx.bucketDir }), kepr: openKeprDb(dataDir), mirrorDir: join(dataDir, "mirror"), }; return { ...target, mirror: () => openMirror(target.mirrorDir, fx.repo) }; } describe("a first sync of a koh bucket", () => { let fx: Fixture; let t: Awaited>; beforeAll(async () => { fx = await makeKohBucket(); t = await syncTarget(fx); await syncAll(t); }); const [head, checkpoint, second, first] = [0, 1, 2, 3].map((i) => () => fx.saveIds[i] as string); test("registers the repo private, with a clean last sync", () => { const row = t.kepr.query("SELECT public, last_sync_at, last_sync_error FROM repos").get() as { public: number; last_sync_at: number; last_sync_error: string | null; }; expect(row.public).toBe(0); expect(row.last_sync_error).toBeNull(); expect(row.last_sync_at).toBeGreaterThan(0); }); test("history is newest first and leaves out the checkpoint", () => { expect(headId(t.mirror())).toBe(head()); expect(history(t.mirror(), 10, 0).map((s) => s.id)).toEqual([head(), second(), first()]); expect(history(t.mirror(), 1, 1).map((s) => s.message)).toEqual(["add readme and entry point"]); expect(timeline(t.mirror()).map((s) => s.id)).toEqual([first(), second(), head()]); expect(saveByPrefix(t.mirror(), checkpoint())?.is_checkpoint).toBe(true); }); test("finds a save by a unique prefix of six or more characters", () => { const save = saveByPrefix(t.mirror(), second().slice(0, 6)); expect([save?.id, save?.agent, save?.session]).toEqual([second(), "test-agent", "fixture"]); expect(saveByPrefix(t.mirror(), second().slice(0, 5))).toBeNull(); expect(saveByPrefix(t.mirror(), "zzzzzz")).toBeNull(); }); test("reads files and their bytes at any save", () => { const paths = filesAt(t.mirror(), head()).map((f) => f.path); expect(paths).toContain("src/lib/deep/util.ts"); const now = fileAt(t.mirror(), head(), "src/main.ts"); const then = fileAt(t.mirror(), second(), "src/main.ts"); expect(text(objectBytes(t.mirror(), now?.hash as string))).toContain("hello"); expect(text(objectBytes(t.mirror(), then?.hash as string))).toContain("'hi'"); expect(fileAt(t.mirror(), first(), "src/main.ts")).toBeNull(); expect(fileAt(t.mirror(), head(), "docs/logo.png")?.size).toBe(70); }); test("counts saves without checkpoints, and lines and language at head", () => { const stats = repoStats(t.mirror()); expect(stats.saves).toBe(3); expect(stats.language).toBe("TypeScript"); // Source lines only: util.ts 2 + main.ts 2. The README, guide, secret and .kohignore are not // code, and vendor/big.c's 100 lines of C are someone else's. expect(stats.lines).toBe(4); expect(stats.updated).toBe(saveByPrefix(t.mirror(), head())?.created_at ?? -1); }); test("stores the flag event", () => { const flag = t.mirror().query("SELECT kind, save_id, reason FROM events").get(); expect(flag).toEqual({ kind: "flag", save_id: first(), reason: "holds a secret" }); }); test("a save's flag state is its newest flag or clear event, by time and then ID", async () => { expect(flagReason(t.mirror(), first())).toBe("holds a secret"); expect(flagReason(t.mirror(), head())).toBeNull(); const mirror = openMirror(await mkdtemp(join(tmpdir(), "kepr-flags-")), "flags"); const event = mirror.query( "INSERT INTO events (id, kind, save_id, reason, created_at, machine) VALUES (?, ?, 'aaaaaaaaaaaa', ?, ?, 'm')", ); event.run("000000000001", "flag", "first", 10); expect(flagReason(mirror, "aaaaaaaaaaaa")).toBe("first"); event.run("000000000002", "clear", null, 20); expect(flagReason(mirror, "aaaaaaaaaaaa")).toBeNull(); // Same second: the larger ID is the newer. event.run("000000000003", "flag", "again", 20); expect(flagReason(mirror, "aaaaaaaaaaaa")).toBe("again"); event.run("000000000000", "clear", null, 20); expect(flagReason(mirror, "aaaaaaaaaaaa")).toBe("again"); }); test("a strip reaches the mirror and drops the object", async () => { const secret = fileAt(t.mirror(), head(), "secret.txt")?.hash as string; expect(objectBytes(t.mirror(), secret)).not.toBeNull(); await fx.strip("secret.txt"); await syncAll(t); for (const id of fx.saveIds) expect(fileAt(t.mirror(), id, "secret.txt")).toBeNull(); expect(objectBytes(t.mirror(), secret)).toBeNull(); expect(fileAt(t.mirror(), head(), "src/main.ts")).not.toBeNull(); }); }); test("a hidden path that is then stripped stays gone when it is unhidden", async () => { const fx = await makeKohBucket("hidden-strip"); const t = await syncTarget(fx); await syncAll(t); const secret = fileAt(t.mirror(), fx.saveIds[0] as string, "secret.txt")?.hash as string; t.kepr.query("UPDATE repos SET hidden_paths = 'secret.txt' WHERE name = ?").run(fx.repo); await syncAll(t); for (const id of fx.saveIds) expect(fileAt(t.mirror(), id, "secret.txt")).toBeNull(); expect(objectBytes(t.mirror(), secret)).not.toBeNull(); await fx.strip("secret.txt"); await syncAll(t); applyHidden(t.mirror(), []); for (const id of fx.saveIds) expect(fileAt(t.mirror(), id, "secret.txt")).toBeNull(); expect(objectBytes(t.mirror(), secret)).toBeNull(); expect(fileAt(t.mirror(), fx.saveIds[0] as string, "src/main.ts")).not.toBeNull(); }); test("a whole six-character ID finds its own save, not a longer one it begins", async () => { const mirror = openMirror(await mkdtemp(join(tmpdir(), "kepr-mirror-")), "ids"); const add = mirror.query( "INSERT INTO saves (id, parent, message, created_at, machine) VALUES (?, NULL, ?, 1, 'm')", ); add.run("af96c8", "migrated from Koh 2"); add.run("af96c8123456", "made by Koh 3"); expect(saveByPrefix(mirror, "af96c8")?.message).toBe("migrated from Koh 2"); expect(saveByPrefix(mirror, "af96c81")?.message).toBe("made by Koh 3"); }); test("a sync lands history in batches, so one cut short resumes where it stopped", async () => { const fx = await makeKohBucket("batched"); const bucket = openBucket({ kind: "folder", path: fx.bucketDir }); const mirror = openMirror(await mkdtemp(join(tmpdir(), "kepr-mirror-")), fx.repo); // The bucket goes away as soon as the first batch has landed. const failing: Bucket = { ...bucket, get: (key) => key.includes("/objects/") && mirrorState(mirror, "head") !== null ? Promise.reject(new Error("connection reset")) : bucket.get(key), }; await expect(syncRepo(failing, mirror, fx.repo, [], 1)).rejects.toThrow("connection reset"); const first = fx.saveIds[3] as string; expect(mirrorState(mirror, "head")).toBe(first); expect(history(mirror, 10, 0).map((s) => s.id)).toEqual([first]); await syncRepo(bucket, mirror, fx.repo, [], 1); expect(headId(mirror)).toBe(fx.saveIds[0] as string); expect(history(mirror, 10, 0)).toHaveLength(3); expect(flagReason(mirror, first)).toBe("holds a secret"); }); test("a save larger than a batch is stored ahead of itself, and a sync cut short within it resumes", async () => { const fx = await makeKohBucket("ahead"); const bucket = openBucket({ kind: "folder", path: fx.bucketDir }); const mirror = openMirror(await mkdtemp(join(tmpdir(), "kepr-mirror-")), fx.repo); // Reads go out in groups, so the cut falls on the first object the second save adds. const first = await readSave(bucket, fx.repo, fx.saveIds.at(-1) as string); const firstObjects = new Set(first.files.map((file) => file.hash)).size; let objectGets = 0; const failing: Bucket = { ...bucket, get: (key) => key.includes("/objects/") && ++objectGets > firstObjects ? Promise.reject(new Error("connection reset")) : bucket.get(key), }; await expect(syncRepo(failing, mirror, fx.repo, [], 1)).rejects.toThrow("connection reset"); const stored = () => mirror.query<{ n: number }, []>("SELECT COUNT(*) AS n FROM objects").get()?.n; expect(mirrorState(mirror, "head")).toBeNull(); expect(stored()).toBe(firstObjects); await syncRepo(bucket, mirror, fx.repo, [], 1); expect(headId(mirror)).toBe(fx.saveIds[0] as string); expect(history(mirror, 10, 0)).toHaveLength(3); const guide = fileAt(mirror, fx.saveIds[0] as string, "docs/guide.md"); expect(text(objectBytes(mirror, guide?.hash as string))).toBe( "# Guide\n\nRead me. It is short.\n", ); }); test("object reads overlap, but never past the bytes allowed in flight", async () => { const fx = await makeKohBucket("in-flight"); const bucket = openBucket({ kind: "folder", path: fx.bucketDir }); const peak = async (readBytes?: number) => { let reading = 0; let most = 0; const counting: Bucket = { ...bucket, get: async (key) => { if (!key.includes("/objects/")) return bucket.get(key); most = Math.max(most, ++reading); await Bun.sleep(5); reading--; return bucket.get(key); }, }; const mirror = openMirror(await mkdtemp(join(tmpdir(), "kepr-mirror-")), fx.repo); await syncRepo(counting, mirror, fx.repo, [], undefined, readBytes); expect(headId(mirror)).toBe(fx.saveIds[0] as string); return most; }; expect(await peak()).toBeGreaterThan(1); // Every object is larger than one byte, so each is read alone. expect(await peak(1)).toBe(1); }); test("a read the gate admits after the batch filled waits for the next batch", async () => { const fx = await makeKohBucket("deferred"); const bucket = openBucket({ kind: "folder", path: fx.bucketDir }); const mirror = openMirror(await mkdtemp(join(tmpdir(), "kepr-mirror-")), fx.repo); const stored = () => mirror.query<{ n: number }, []>("SELECT COUNT(*) AS n FROM objects").get()?.n ?? 0; let read = 0; const unstored: number[] = []; const counting: Bucket = { ...bucket, get: async (key) => { // Each object fills a one-byte batch, so it is stored before the next read starts. if (key.includes("/objects/")) unstored.push(read++ - stored()); return bucket.get(key); }, }; await syncRepo(counting, mirror, fx.repo, [], 1, 1); expect(read).toBeGreaterThan(2); expect(new Set(unstored)).toEqual(new Set([0])); }); test("a failed read reports only once every other read has settled", async () => { const fx = await makeKohBucket("settled"); const bucket = openBucket({ kind: "folder", path: fx.bucketDir }); const mirror = openMirror(await mkdtemp(join(tmpdir(), "kepr-mirror-")), fx.repo); let reading = 0; let reads = 0; const failing: Bucket = { ...bucket, get: async (key) => { if (!key.includes("/objects/")) return bucket.get(key); if (reads++ === 0) throw new Error("connection reset"); reading++; await Bun.sleep(30); reading--; return bucket.get(key); }, }; await expect(syncRepo(failing, mirror, fx.repo)).rejects.toThrow("connection reset"); expect(reading).toBe(0); }); test("a new save syncs incrementally and a bad format is recorded as the last sync error", async () => { const fx = await makeKohBucket("growing"); const t = await syncTarget(fx); await syncAll(t); await writeFile(join(fx.repoDir, "new.txt"), "new\n"); Bun.spawnSync(["koh", "save", "add new"], { cwd: fx.repoDir, env: { PATH: process.env.PATH, HOME: join(fx.repoDir, "../../home"), NO_COLOR: "1" }, }); await syncAll(t); expect(history(t.mirror(), 1, 0)[0]?.message).toBe("add new"); await writeFile(join(fx.bucketDir, fx.repo, "format"), "koh-remote 2"); await syncAll(t); const row = t.kepr.query("SELECT last_sync_error FROM repos").get() as { last_sync_error: string; }; expect(row.last_sync_error).toContain('remote format "koh-remote 2"'); }); test("a whole-history strip reaches every save, those arriving after it too, whatever their time (§10.2)", async () => { const fx = await makeKohBucket("late"); const view = await mkdtemp(join(tmpdir(), "kepr-view-")); const viewRepo = join(view, fx.repo); await cp(join(fx.bucketDir, fx.repo), viewRepo, { recursive: true }); // A strip older than every save here: a strip bans its path from all of history, so no save // keeps it, however much later it was made or arrives. await mkdir(join(viewRepo, "events"), { recursive: true }); await writeFile( join(viewRepo, "events", "0000000000bb"), JSON.stringify({ id: "0000000000bb", kind: "strip", save_id: "", path: "secret.txt", reason: "a secret got saved", created_at: 1, machine: "elsewhere", }), ); const oldest = fx.saveIds.at(-1) as string; const newest = fx.saveIds[0] as string; // First sync: the strip and the oldest save only. await writeFile(join(viewRepo, "head"), oldest); const t = await syncTarget({ ...fx, bucketDir: view }); await syncAll(t); expect(fileAt(t.mirror(), oldest, "secret.txt")).toBeNull(); // Then the rest arrive, each made long after the strip. await writeFile(join(viewRepo, "head"), newest); await syncAll(t); expect(headId(t.mirror())).toBe(newest); for (const id of fx.saveIds) expect(fileAt(t.mirror(), id, "secret.txt")).toBeNull(); expect(fileAt(t.mirror(), newest, "src/main.ts")).not.toBeNull(); }); describe("the mirror's schema version", () => { test("a version 4 mirror reopens and any other version is refused as disposable", async () => { const dir = await mkdtemp(join(tmpdir(), "kepr-mirror-")); const created = openMirror(dir, "one"); created.run("PRAGMA wal_checkpoint(TRUNCATE)"); await Bun.write(join(dir, "copy.db"), Bun.file(join(dir, "one.db"))); expect(openMirror(dir, "copy").query("PRAGMA user_version").get()).toEqual({ user_version: 4, }); for (const version of [1, 5]) { const other = new Database(join(dir, `v${version}.db`)); other.run(`PRAGMA user_version = ${version}`); other.close(); expect(() => openMirror(dir, `v${version}`)).toThrow("disposable"); } }); test("a version 2 mirror moves to 4 in place, emptied only when it deleted hidden rows", async () => { const dir = await mkdtemp(join(tmpdir(), "kepr-mirror-")); for (const [name, hidden] of [ ["plain", ""], ["hiding", "secret.txt"], ] as const) { const db = openMirror(dir, name); for (const table of ["hidden_files", "touches", "facts", "known_objects"]) { db.run(`DROP TABLE ${table}`); } db.run("PRAGMA user_version = 2"); db.query("INSERT INTO state (key, value) VALUES ('head', 'abcdef123456'), ('hidden', ?)").run( hidden, ); closeMirror(dir, name); } const head = (name: string) => mirrorState(openMirror(dir, name), "head"); expect(head("plain")).toBe("abcdef123456"); expect(head("hiding")).toBeNull(); for (const name of ["plain", "hiding"]) { const db = openMirror(dir, name); expect(db.query("PRAGMA user_version").get()).toEqual({ user_version: 4 }); expect(db.query("SELECT COUNT(*) AS n FROM hidden_files").get()).toEqual({ n: 0 }); } }); test("touches recorded save by save match touches worked out all at once", async () => { const fx = await makeKohBucket("stepwise"); const t = await syncTarget(fx); await syncAll(t); await writeFile(join(fx.repoDir, "src/main.ts"), "// Entry point.\nconsole.log('again');\n"); await writeFile(join(fx.repoDir, "added.txt"), "new\n"); Bun.spawnSync(["koh", "save", "change main, add a file"], { cwd: fx.repoDir, env: { PATH: process.env.PATH, HOME: join(fx.repoDir, "../../home"), NO_COLOR: "1" }, }); Bun.spawnSync(["koh", "offer"], { cwd: fx.repoDir, env: { PATH: process.env.PATH, HOME: join(fx.repoDir, "../../home"), NO_COLOR: "1" }, }); await syncAll(t); const touches = () => t.mirror().query("SELECT path, n FROM touches ORDER BY n, path").all(); const stepwise = touches(); expect(stepwise).toContainEqual(expect.objectContaining({ path: "added.txt" })); // Reopened as version 3, the mirror works every save's touches out again in one go. for (const table of ["touches", "facts", "known_objects"]) { t.mirror().run(`DROP TABLE ${table}`); } t.mirror().run("PRAGMA user_version = 3"); closeMirror(t.mirrorDir, fx.repo); expect(touches()).toEqual(stepwise); }); test("a recount cut short is done when the mirror next opens", async () => { const fx = await makeKohBucket("recount"); const t = await syncTarget(fx); await syncAll(t); const known = () => t.mirror().query("SELECT COUNT(*) AS n FROM known_objects").get(); const counted = known(); // What a landing leaves if Kepr stops before its recount commits. t.mirror().run("DELETE FROM known_objects"); t.mirror().run("DELETE FROM state WHERE key = 'stats_rule'"); closeMirror(t.mirrorDir, fx.repo); expect(known()).toEqual(counted); }); test("a version 3 mirror gains the touches and facts a sync would have recorded", async () => { const fx = await makeKohBucket("derived"); const t = await syncTarget(fx); await syncAll(t); const derived = () => [ t.mirror().query("SELECT path, n FROM touches ORDER BY n, path").all(), t.mirror().query("SELECT * FROM facts ORDER BY path").all(), ]; const recorded = derived(); expect((recorded[0] as unknown[]).length).toBeGreaterThan(0); for (const table of ["touches", "facts", "known_objects"]) { t.mirror().run(`DROP TABLE ${table}`); } t.mirror().run("PRAGMA user_version = 3"); closeMirror(t.mirrorDir, fx.repo); expect(derived()).toEqual(recorded); }); }); test("a strip that arrives older than the saves already mirrored still strips them (§10.2, R6)", async () => { const fx = await makeKohBucket("older-strip"); const t = await syncTarget(fx); await syncAll(t); // Every save in the mirror holds secret.txt … for (const id of fx.saveIds) expect(fileAt(t.mirror(), id, "secret.txt")).not.toBeNull(); // … and now a strip arrives whose own time is older than all of them, which is what a // strip made on another machine and offered late looks like. Every save is reached, whatever // its time. const events = join(fx.bucketDir, fx.repo, "events"); await mkdir(events, { recursive: true }); await writeFile( join(events, "0000000000aa"), JSON.stringify({ id: "0000000000aa", kind: "strip", save_id: "", path: "secret.txt", reason: "a secret got saved", created_at: 1, machine: "elsewhere", }), ); await syncAll(t); for (const id of fx.saveIds) expect(fileAt(t.mirror(), id, "secret.txt")).toBeNull(); expect(fileAt(t.mirror(), fx.saveIds[0] as string, "src/main.ts")).not.toBeNull(); }); test("a save with an unsafe path lands without it, keeps syncing, and the admin status names it (A3)", async () => { const fx = await makeKohBucket("quarantine"); const head = fx.saveIds[0] as string; const key = join(fx.bucketDir, fx.repo, "saves", head); const record = JSON.parse(await readFile(key, "utf8")); record.files.push({ ...record.files[0], path: "Icon\r" }); await writeFile(key, JSON.stringify(record)); const inst = await instance(fx.bucketDir); const mirror = openMirror(join(inst.dataDir, "mirror"), fx.repo); expect(inst.kepr.query("SELECT last_sync_error FROM repos").get()).toEqual({ last_sync_error: null, }); expect(filesAt(mirror, head).map((file) => file.path)).toContain("README.md"); expect(filesAt(mirror, head).map((file) => file.path)).not.toContain("Icon\r"); await writeFile(join(fx.repoDir, "later.txt"), "later\n"); const env = { PATH: process.env.PATH, HOME: join(fx.repoDir, "../../home"), NO_COLOR: "1" }; Bun.spawnSync(["koh", "save", "a later save"], { cwd: fx.repoDir, env }); Bun.spawnSync(["koh", "offer"], { cwd: fx.repoDir, env }); await syncAll({ bucket: openBucket({ kind: "folder", path: fx.bucketDir }), kepr: inst.kepr, mirrorDir: join(inst.dataDir, "mirror"), }); expect(history(mirror, 1, 0)[0]?.message).toBe("a later save"); const browser = new Browser(inst.app); await browser.signIn(); const page = await (await browser.get("/admin/repos")).text(); expect(page).toContain("1 path quarantined: Icon\\x0d"); }); test("a strip event with an illegal path is ignored, as Koh ignores it (§5.5)", async () => { const fx = await makeKohBucket("illegal-strip"); for (const [id, path] of [ ["aaaaaaaaaaaa", "src//"], ["bbbbbbbbbbbb", ""], ["cccccccccccc", "src\ud800"], ]) { const event = { id, kind: "strip", save_id: "", path, reason: "", created_at: 1, machine: "x" }; await writeFile(join(fx.bucketDir, fx.repo, "events", id as string), JSON.stringify(event)); } const t = await syncTarget(fx); await syncAll(t); expect(t.kepr.query("SELECT last_sync_error FROM repos").get()).toEqual({ last_sync_error: null, }); expect(filesAt(t.mirror(), fx.saveIds[0] as string).map((file) => file.path)).toContain( "src/main.ts", ); }); test("an object that inflates past 512 MiB is quarantined without stopping the sync (A3)", async () => { const fx = await makeKohBucket("oversized"); const head = fx.saveIds[0] as string; const key = join(fx.bucketDir, fx.repo, "saves", head); const record = JSON.parse(await readFile(key, "utf8")); const hash = "ab".repeat(32); record.files.push({ path: "huge.bin", hash, executable: false }); await writeFile(key, JSON.stringify(record)); await writeFile( join(fx.bucketDir, fx.repo, "objects", hash), deflateSync(new Uint8Array(512 * 1024 * 1024 + 1)), ); const t = await syncTarget(fx); await syncAll(t); expect(t.kepr.query("SELECT last_sync_error FROM repos").get()).toEqual({ last_sync_error: null, }); expect(filesAt(t.mirror(), head).map((file) => file.path)).not.toContain("huge.bin"); expect(JSON.parse(mirrorState(t.mirror(), "quarantined") ?? "[]")).toEqual(["huge.bin"]); }); test("the line count and the Atlas's facts skip a file over 8 MiB (A3)", async () => { const fx = await makeKohBucket("large-text"); const t = await syncTarget(fx); await syncAll(t); const before = repoStats(t.mirror()).lines; await writeFile(join(fx.repoDir, "big.txt"), "x\n".repeat(4.5 * 1024 * 1024)); const env = { PATH: process.env.PATH, HOME: join(fx.repoDir, "../../home"), NO_COLOR: "1" }; Bun.spawnSync(["koh", "save", "add a big text file"], { cwd: fx.repoDir, env }); Bun.spawnSync(["koh", "offer"], { cwd: fx.repoDir, env }); await syncAll(t); expect(history(t.mirror(), 1, 0)[0]?.message).toBe("add a big text file"); expect(repoStats(t.mirror()).lines).toBe(before); expect( t.mirror().query("SELECT lines, description FROM facts WHERE path = 'big.txt'").get(), ).toEqual({ lines: null, description: null }); });