import type { AddressInfo } from "node:net"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { parseTokens } from "../../src/config.ts"; import { createHttpServer } from "../../src/http.ts"; import { Scheduler } from "../../src/sync/scheduler.ts"; import { syncSettings } from "../../src/sync/settings.ts"; import { enabledJobs } from "../../src/sync/index.ts"; import type { SyncContext, SyncJob, SyncResult } from "../../src/sync/types.ts"; import { book, rig, type Rig } from "../helpers.ts"; let r: Rig; beforeEach(() => { r = rig(); }); afterEach(() => { vi.useRealTimers(); r.done(); }); const pings: string[] = []; const fakeFetch = (async (url: string | URL | Request) => { pings.push(String(url)); return new Response("OK"); }) as typeof fetch; const job = (name: string, run: (ctx: SyncContext) => Promise, extra: Partial = {}): SyncJob => ({ name, everyMinutes: 60, run, ...extra, }); const deferred = () => { let resolve!: (v: SyncResult) => void; const promise = new Promise((ok) => (resolve = ok)); return { promise, resolve }; }; describe("scheduler", () => { it("reports each run to sync_runs and pings Healthchecks start, success and fail", async () => { pings.length = 0; let fail = false; const s = new Scheduler( r.service, [ job("demo", async () => { if (fail) throw new Error("source down"); return { message: "2 synced", counts: { synced: 2 } }; }, { healthcheckUrl: "https://hc-ping.com/abc/" }), ], { fetch: fakeFetch, log: () => {} }, ); expect(await s.runNow("demo")).toMatchObject({ ok: true, message: "2 synced" }); fail = true; expect(await s.runNow("demo")).toMatchObject({ ok: false, message: "source down" }); expect(r.store.syncRuns("demo").map((x) => [x.ok, x.message])).toEqual([ [false, "source down"], [true, "2 synced"], ]); expect(r.store.syncJobs()[0]).toMatchObject({ job: "demo", ok: false, counts: {} }); expect(pings).toEqual(["https://hc-ping.com/abc/start", "https://hc-ping.com/abc", "https://hc-ping.com/abc/start", "https://hc-ping.com/abc/fail"]); }); it("never runs one job twice at once", async () => { const d = deferred(); let runs = 0; const s = new Scheduler(r.service, [job("slow", () => (runs++, d.promise))], { fetch: fakeFetch, log: () => {} }); const first = s.runNow("slow"); expect(await s.runNow("slow")).toMatchObject({ skipped: true }); expect(runs).toBe(1); d.resolve({}); await first; await Promise.resolve(); expect(s.isRunning("slow")).toBe(false); }); it("aborts a run that passes its timeout and stays locked until the job really ends", async () => { vi.useFakeTimers(); const d = deferred(); let aborted = false; const s = new Scheduler( r.service, [ job("stuck", (ctx) => { ctx.signal.addEventListener("abort", () => (aborted = true)); return d.promise; }, { timeoutMs: 1000 }), ], { fetch: fakeFetch, log: () => {} }, ); const run = s.runNow("stuck"); await vi.advanceTimersByTimeAsync(1001); expect(await run).toMatchObject({ ok: false, message: "Timed out after 1 seconds" }); expect(aborted).toBe(true); expect(s.isRunning("stuck")).toBe(true); d.resolve({}); await vi.advanceTimersByTimeAsync(0); expect(s.isRunning("stuck")).toBe(false); }); it("runs each job at start, staggered, then on its interval", async () => { vi.useFakeTimers(); const ran: string[] = []; const mk = (name: string) => job(name, async () => (ran.push(name), {}), { everyMinutes: 10 }); const s = new Scheduler(r.service, [mk("a"), mk("b")], { staggerMs: 1000, fetch: fakeFetch, log: () => {} }); s.start(); await vi.advanceTimersByTimeAsync(0); expect(ran).toEqual(["a"]); await vi.advanceTimersByTimeAsync(1000); expect(ran).toEqual(["a", "b"]); await vi.advanceTimersByTimeAsync(10 * 60_000); expect(ran).toEqual(["a", "b", "a", "b"]); s.stop(); await vi.advanceTimersByTimeAsync(60 * 60_000); expect(ran).toHaveLength(4); }); }); describe("sync cache", () => { it("keeps the last good values when a run fails, and shows them on the notes API", async () => { const path = r.put(book("Mistborn")); r.service.scan(); let fail = false; const s = new Scheduler( r.service, [ job("demo", async (ctx) => { if (fail) throw new Error("down"); ctx.set(path, { cover: "https://example.com/a.jpg", pages: 541 }); return {}; }), ], { fetch: fakeFetch, log: () => {} }, ); await s.runNow("demo"); const first = r.store.syncCache.get("demo", path); fail = true; await s.runNow("demo"); expect(r.store.syncCache.get("demo", path)).toEqual(first); // Values from several jobs merge per note. r.store.syncCache.set("other", path, { year: 2006 }, "2026-10-11T00:00:00Z"); const token = "r".repeat(24); const server = createHttpServer(r.service, { tokens: parseTokens(`reader:${token}`) }); await new Promise((ok) => server.listen(0, "127.0.0.1", ok)); const base = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; const get = (p: string) => fetch(base + p, { headers: { authorization: `Bearer ${token}` } }).then((x) => x.json()); try { const [listed] = (await get("/api/notes?full=true")) as { synced: unknown }[]; expect(listed!.synced).toEqual({ cover: "https://example.com/a.jpg", pages: 541, year: 2006 }); expect(((await get(`/api/note?path=${encodeURIComponent(path)}`)) as { synced: unknown }).synced).toMatchObject({ pages: 541 }); expect(await get("/api/notes")).toEqual([expect.not.objectContaining({ synced: expect.anything() })]); } finally { server.close(); } }); it("retain drops values for notes that left a job's scope", () => { r.store.syncCache.set("j", "a.md", { x: 1 }, "t"); r.store.syncCache.set("j", "b.md", { x: 2 }, "t"); r.store.syncCache.set("k", "b.md", { y: 3 }, "t"); r.store.syncCache.retain("j", ["a.md"]); expect([...r.store.syncCache.all().entries()]).toEqual([ ["a.md", { x: 1 }], ["b.md", { y: 3 }], ]); }); }); describe("writing notes as sync", () => { it("runs the agent rules: a note Greg wrote is held unless it allows agent edits", async () => { const mine = r.put(book("Mine", { progress: "1/10" })); const open = r.put(book("Open", { progress: "1/10", agentEdits: true })); r.service.scan(); const outcomes: string[] = []; const s = new Scheduler( r.service, [ job("reader", async (ctx) => { for (const p of [mine, open]) outcomes.push(ctx.apply({ op: "update", path: p, fields: { progress: "2/10" } }).status); outcomes.push(ctx.apply({ op: "update", path: open, fields: { note: "Mail me at someone@example.com" } }).status); return {}; }), ], { fetch: fakeFetch, log: () => {} }, ); await s.runNow("reader"); expect(outcomes).toEqual(["held", "accepted", "held"]); expect(r.store.heldList()[0]!.reasons.map((x) => x.code)).toContain("contact"); expect(r.store.get(open)!.data).toMatchObject({ progress: "2/10", updatedBy: "sync" }); expect(r.store.get(open)!.author).toBe("sync"); }); }); describe("settings", () => { it("enables only the jobs whose config is present", () => { const s = syncSettings({ SYNC_JOBS: "bgg, covers", SYNC_BGG_MINUTES: "720", SYNC_COVERS_HEALTHCHECK_URL: "https://hc/x", BGG_TOKEN: "t" }); expect(enabledJobs(s).map((j) => [j.name, j.everyMinutes, j.healthcheckUrl])).toEqual([ ["bgg", 720, undefined], ["covers", 15, "https://hc/x"], ]); expect(enabledJobs(syncSettings({}))).toEqual([]); }); });