com.gregkilmer.www/services/content/test/sync/scheduler.test.ts

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([]);
  });
});