Phase 6: sync jobs, inquiry and order forms

- Sync scheduler in the content service with a per-job cache (sync_cache), sync_runs reporting and Healthchecks pings. Jobs: Archidekt, BoardGameGeek, cover art (Open Library, TMDB, AniList), Suwayomi and Kavita reading progress, Bambuddy filament and Homebox material stock. A note's own value always wins over a synced one. - PersonalReader: updates only Library notes with a matching anilistId, changes to Greg's notes are held for review; Reading series with no note show as suggestions on the admin Overview. - /api/inquiries and /api/orders: Turnstile, rate limits, a saved copy in the content service (intake token) plus email over SMTP. Orders are re-checked against the live notes: accepted only when the category's ordersOpen is true and the line has capacity. - Admin Inbox for saved submissions. Options' inStock is optional so synced stock can fill it. All settings from process.env; .env.example per service. - Sources could not be reached from the build environment; parsers are tested against fixtures labelled as not live. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012aSbo2D1qP8cMFQCykf7kx
2026-11-08 04:08:24 +00:00 · 2026-10-11 23:50:06 +00:00
parent 60dfeecd7f
commit e3b0cf60c5
76 changed files with 6050 additions and 86 deletions
@@ -0,0 +1,204 @@
+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<SyncResult>, extra: Partial<SyncJob> = {}): SyncJob => ({
+  name,
+  everyMinutes: 60,
+  run,
+  ...extra,
+});
+
+const deferred = () => {
+  let resolve!: (v: SyncResult) => void;
+  const promise = new Promise<SyncResult>((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<void>((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([]);
+  });
+});