merge: task 9 jobs, scheduler, registry, connections

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

docs/ARCHITECTURE.md

@@ -16,16 +16,16 @@ user layer or files a Changes-inbox event.
 |---|---|---|
 | `apps/server` | `Bun.serve`: `/api/v1`, `/ingest/<plugin>`, job worker, static web | planned |
 | `apps/web` | React app (HTML imports) | planned |
-| `packages/core` | domain services, one module per area (import by path, no barrel): `auth`, `crypto`, `works`, `matching`, `entries` (entries, library listing, quick add, per-user moves, match suggestions), `engagements`, `history` (`user_history`, restore), `lists`, `copies`, `custom-fields`, `attachments`, `export` (per-user bundle, CSV, restore), `backup` (scheduled bundles + SQLite `VACUUM INTO`), `sync/` (the change engine: `snapshot` hashing and versions, `apply` per-record diff and apply, `run` = `runSync`, `changes` = Changes inbox `listChanges`/`resolveChange`, `listens` = `ingestListens`) | exists (partial) |
+| `packages/core` | domain services, one module per area (import by path, no barrel): `auth`, `crypto`, `works`, `matching`, `entries` (entries, library listing, quick add, per-user moves, match suggestions), `engagements`, `history` (`user_history`, restore), `lists`, `copies`, `custom-fields`, `attachments`, `export` (per-user bundle, CSV, restore), `backup` (scheduled bundles + SQLite `VACUUM INTO`), `sync/` (the change engine: `snapshot` hashing and versions, `apply` per-record diff and apply, `run` = `runSync`, `changes` = Changes inbox `listChanges`/`resolveChange`, `listens` = `ingestListens`), `jobs` (queue, worker, scheduler), `registry` (`loadPlugins`), `connections` (CRUD, plugin context, `syncHandler`) | exists (partial) |
 | `packages/db` | Kysely over one Bun.SQL dialect (SQLite + Postgres), ids, JSON helpers, migrations, `forEachDialect` test harness | exists (schema pending) |
 | `packages/plugin-sdk` | plugin contract, `SourceRecord`, http/rate-limit helpers, fixture test harness | planned |
 | `plugins/<id>` | one source integration each (hardcover, trakt, mangabaka, listenbrainz, lastfm, csv) | planned |
 | `docs/design/mockups` | approved UI mockups (reference only) | exists |
 
 ## Data flow
-Sync lifecycle (`packages/core/src/sync/`; steps 1–2 are Task 9's job worker, planned):
+Sync lifecycle (`packages/core/src/sync/`; steps 1–2 are `jobs.ts` and `connections.ts`):
-1. The scheduler enqueues a job.
+1. The scheduler enqueues a `sync` job (lockKey `sync:<connectionId>`) for each non-paused connection that is due.
-2. The worker loads the connection and plugin and calls `runSync(deps, plugin, connectionId, mode)`, injecting
+2. The worker claims the job; `syncHandler` loads the connection and plugin and calls `runSync(deps, plugin, connectionId, mode)`, injecting
    `makeContext` (http client, credentials, cursor). `mode` picks `pull`, `pullChanges` or `importFile`.
 3. `runSync` opens a `source_runs` row and iterates the plugin's `SourceRecord`s one at a time.
    - `listen` records are batched into `ingestListens` (insert-or-ignore into `listens`; no snapshot, no events).

docs/DECISIONS.md

@@ -4,6 +4,44 @@ Append-only. Never edit or delete an entry. Supersede it with a new one that lin
 One entry per decision a future maintainer would otherwise have to reverse-engineer.
 No index: `grep -n '^## ' docs/DECISIONS.md` lists every entry (date, title, line number).
 
+## 2026-10-30: Job failure semantics and plugin reload limits
+**Status:** accepted
+**Context:** Task 9 fix round 1.
+**Decision:**
+- **Failures before `runSync` never retry.** A missing/paused connection drops the job (`DropJobError`: `done`, message in
+  `last_error`). A missing plugin or undecryptable credentials (`CryptoError`) insert a failed `source_runs` row, set the
+  connection to `error`, advance `last_run_at` and throw `NonRetryableError` (job `failed`, no backoff). The scheduler still
+  picks `error` connections, one interval later, so fixing the cause heals them.
+- **Stale locks.** Every worker tick re-queues `running` jobs whose `locked_at` is over an hour old (attempts unchanged), so a
+  crash cannot hold a lockKey forever. Job bookkeeping errors are logged (`[jobs]`), never unhandled; `stop()` never rejects
+  and aborts the `signal` passed to handlers (`handler(payload, { signal })`).
+- **Reload** evicts the entry file from `require.cache` and imports with `?v=<Date.now()>` (Bun ignores the query alone for
+  cached `.ts`). Only entry files reload: changes in a plugin's other modules need a server restart.
+- `credentials.save` and `updateConnection` read-merge-write in one transaction (row lock on Postgres). `ctx.log` is buffered and
+  appended to the run's `source_runs.log` when the run ends.
+
+## 2026-10-30: Jobs, plugin registry, connections: rate limit in the manifest, secret split, per-connection lock
+**Status:** accepted
+**Context:** Task 9 (`packages/core/src/{jobs,registry,connections}.ts`).
+**Decision:**
+- **Rate limit lives in the manifest** (`PluginManifest.rateLimit?: { perMinute, burst? }`), not in plugin code. The host builds
+  `ctx.http` from it (default 60/min, burst 5, UA `STTracker/0.1 (+self-hosted)`), so every plugin is throttled the same way
+  and the per-plugin `RATE_PER_MINUTE` constants are gone.
+- **Secrets are split off config.** Config fields typed `"secret"` in the manifest are stored only in the encrypted
+  `credentials_enc` blob (with OAuth tokens a plugin saves), never in `config`. `makePluginContext` merges them back into
+  `ctx.config`; `listConnections` returns plain config and `hasCredentials` only. `updateConnection` keeps secrets that are
+  omitted and replaces plain config wholesale. `credentials.save` merges and persists immediately, so a token refreshed
+  mid-pull survives a crash.
+- **Constructor deviation.** `createConnection`/`updateConnection` take the plugin `manifest` as a trailing argument (needed
+  for the split), and `makePluginContext` is async (it decrypts) and takes the manifest before an optional `http`.
+- **One sync per connection.** Sync jobs carry `lockKey = "sync:<connectionId>"`; `enqueue` returns null while a
+  `queued|running` job holds the key. On Postgres the check and insert run under `pg_advisory_xact_lock(hashtext(key))`.
+  Claiming is `update … where status = 'queued'` checked on affected rows. Retries back off `2^attempts` minutes, 5 attempts, then `failed`.
+- **Mode choice.** The sync handler runs `full` when asked, when the plugin lacks `pullChanges`, or when no `ok` full run started
+  within 7 days; else `changes`. A failed run is rethrown so the worker retries it.
+**Consequences:** Reload re-imports plugin modules with `?v=<Date.now()>`; old module instances are not freed by Bun.
+
+
 ## 2026-10-30: Sync engine: hash composition, anti-flip-flop, no revive, root ancestor, dedupe
 **Status:** accepted
 **Context:** Task 8 built the change engine (`packages/core/src/sync/`). Spec §8.3 fixes the outcomes; these

docs/TODO.md

@@ -24,8 +24,11 @@ Small, specific, independently actionable. Remove only when actually done.
     file, and compare steps and token use.
 
 ## Known issues
-- [ ] Sync: two concurrent runs of the same connection can both insert a new `source_records` row for one key
+- [ ] Plugins: reload re-imports entry files only; edits to a plugin's other modules need a server restart.
-  (one fails on the unique key and is logged as a record error). The job worker should lock per connection.
+- [ ] Plugins: rate limiting is per run (each run builds its own client), not per plugin (spec §11); share a limiter per plugin in Phase 2.
+- [ ] Plugins: `ctx.log` reaches `source_runs.log` only after the run ends, so a crash mid-run loses it.
+- [ ] Sync: two concurrent runs of one connection outside the job queue (e.g. a manual run) can still race on `source_records`.
+  Queued syncs are serialised by `lockKey`.
 - [ ] Restore: a new child work whose parent+position collides with an existing child of a re-matched parent fails
   with `ImportError("conflict")` instead of being merged or repositioned.
 - [ ] Restore: an unlink of identical attachment bytes between the pre-write and the commit could remove the file;

packages/core/src/connections.ts

@@ -0,0 +1,297 @@
+import type { Db } from "@sttracker/db/db";
+import { newId, nowIso } from "@sttracker/db/ids";
+import { fromJson, toJson } from "@sttracker/db/json";
+import type { ConnectionStatus } from "@sttracker/db/schema";
+import { createHttpClient, type HttpClientOptions } from "@sttracker/plugin-sdk/http";
+import type { HttpClient, PluginContext, PluginManifest } from "@sttracker/plugin-sdk/types";
+import { decryptJson, encryptJson } from "./crypto.ts";
+import { CryptoError, NotFoundError, ValidationError } from "./errors.ts";
+import { DropJobError, NonRetryableError, type JobContext } from "./jobs.ts";
+import type { PluginRegistry } from "./registry.ts";
+import { runSync, type Connection, type RunSummary } from "./sync/run.ts";
+
+const WEEK_MS = 7 * 24 * 60 * 60 * 1000;
+
+export interface ConnectionView {
+  id: string;
+  pluginId: string;
+  name: string;
+  config: Record<string, unknown>;
+  scheduleMinutes: number | null;
+  status: ConnectionStatus;
+  lastRunAt: string | null;
+  createdAt: string;
+  hasCredentials: boolean;
+}
+
+export interface ConnectionInput {
+  pluginId: string;
+  name: string;
+  config: Record<string, unknown>;
+  credentials?: Record<string, string>;
+  scheduleMinutes?: number | null;
+}
+
+type Creds = Record<string, string>;
+
+/** Split incoming config: fields typed `"secret"` in the manifest go to the credentials blob, the rest stay config. */
+function splitConfig(manifest: PluginManifest, config: Record<string, unknown>): { plain: Record<string, unknown>; secrets: Creds } {
+  const secretKeys = new Set(manifest.config.filter((f) => f.type === "secret").map((f) => f.key));
+  const plain: Record<string, unknown> = {};
+  const secrets: Creds = {};
+  for (const [k, v] of Object.entries(config)) {
+    if (!secretKeys.has(k)) plain[k] = v;
+    else if (typeof v === "string" && v !== "") secrets[k] = v;
+    else if (v !== undefined && v !== null && v !== "") throw new ValidationError(k, "secret_must_be_string");
+  }
+  return { plain, secrets };
+}
+
+async function readCreds(secret: string, enc: string | null): Promise<Creds | null> {
+  return enc === null ? null : decryptJson<Creds>(secret, enc);
+}
+
+const encOrNull = async (secret: string, c: Creds): Promise<string | null> => (Object.keys(c).length === 0 ? null : encryptJson(secret, c));
+
+function checkSchedule(m: number | null | undefined): void {
+  if (m !== undefined && m !== null && (!Number.isInteger(m) || m < 1)) throw new ValidationError("scheduleMinutes", "must_be_positive_integer");
+}
+
+export async function createConnection(db: Db, secret: string, userId: string, input: ConnectionInput, manifest: PluginManifest): Promise<string> {
+  if (manifest.id !== input.pluginId) throw new ValidationError("pluginId", "manifest_mismatch");
+  if (input.name.trim() === "") throw new ValidationError("name", "required");
+  checkSchedule(input.scheduleMinutes);
+  const { plain, secrets } = splitConfig(manifest, input.config);
+  const id = newId();
+  await db.k
+    .insertInto("source_connections")
+    .values({
+      id,
+      user_id: userId,
+      plugin_id: input.pluginId,
+      name: input.name.trim(),
+      config: toJson(plain),
+      credentials_enc: await encOrNull(secret, { ...secrets, ...input.credentials }),
+      schedule_minutes: input.scheduleMinutes ?? null,
+      cursor: null,
+      status: "active",
+      last_run_at: null,
+      created_at: nowIso(),
+    })
+    .execute();
+  return id;
+}
+
+export interface ConnectionPatch {
+  name?: string;
+  config?: Record<string, unknown>;
+  credentials?: Record<string, string>;
+  scheduleMinutes?: number | null;
+  status?: ConnectionStatus;
+}
+
+/** Secrets omitted from `config` are kept; plain config is replaced wholesale. Read-modify-write runs in one transaction. */
+export async function updateConnection(db: Db, secret: string, userId: string, id: string, patch: ConnectionPatch, manifest: PluginManifest): Promise<void> {
+  checkSchedule(patch.scheduleMinutes);
+  await db.k.transaction().execute(async (tx) => {
+    let q = tx.selectFrom("source_connections").selectAll().where("id", "=", id).where("user_id", "=", userId);
+    if (db.dialect === "postgres") q = q.forUpdate();
+    const row = await q.executeTakeFirst();
+    if (!row) throw new NotFoundError("connection");
+    if (row.plugin_id !== manifest.id) throw new ValidationError("pluginId", "manifest_mismatch");
+    const set: Record<string, unknown> = {};
+    if (patch.name !== undefined) {
+      if (patch.name.trim() === "") throw new ValidationError("name", "required");
+      set.name = patch.name.trim();
+    }
+    if (patch.scheduleMinutes !== undefined) set.schedule_minutes = patch.scheduleMinutes;
+    if (patch.status !== undefined) set.status = patch.status;
+    if (patch.config !== undefined || patch.credentials !== undefined) {
+      const split = patch.config ? splitConfig(manifest, patch.config) : { plain: undefined, secrets: {} };
+      if (split.plain) set.config = toJson(split.plain);
+      const merged = { ...((await readCreds(secret, row.credentials_enc)) ?? {}), ...split.secrets, ...patch.credentials };
+      set.credentials_enc = await encOrNull(secret, merged);
+    }
+    if (Object.keys(set).length > 0) await tx.updateTable("source_connections").set(set).where("id", "=", id).execute();
+  });
+}
+
+/** Soft delete: the connection is paused and its credentials are wiped. Records and runs stay. */
+export async function deleteConnection(db: Db, userId: string, id: string): Promise<void> {
+  const r = await db.k
+    .updateTable("source_connections")
+    .set({ status: "paused", credentials_enc: null })
+    .where("id", "=", id)
+    .where("user_id", "=", userId)
+    .executeTakeFirst();
+  if (r.numUpdatedRows !== 1n) throw new NotFoundError("connection");
+}
+
+/** Never returns secrets: plain config plus `hasCredentials`. */
+export async function listConnections(db: Db, userId: string): Promise<ConnectionView[]> {
+  const rows = await db.k.selectFrom("source_connections").selectAll().where("user_id", "=", userId).orderBy("created_at").execute();
+  return rows.map((r) => ({
+    id: r.id,
+    pluginId: r.plugin_id,
+    name: r.name,
+    config: fromJson<Record<string, unknown>>(r.config, {}),
+    scheduleMinutes: r.schedule_minutes,
+    status: r.status,
+    lastRunAt: r.last_run_at,
+    createdAt: r.created_at,
+    hasCredentials: r.credentials_enc !== null,
+  }));
+}
+
+export const USER_AGENT = "STTracker/0.1 (+self-hosted)";
+
+export interface ContextOpts {
+  /** Replaces the rate-limited client entirely. */
+  http?: HttpClient;
+  /** Overrides for the built client (fetch, sleep, clock); used by tests. */
+  httpOptions?: Pick<HttpClientOptions, "fetchImpl" | "sleep" | "now">;
+  signal?: AbortSignal;
+  /** Receives `ctx.log` messages. */
+  onLog?: (msg: string) => void;
+}
+
+/** Synchronous context builder over already-decrypted credentials. */
+function buildContext(db: Db, secret: string, conn: Connection, manifest: PluginManifest, initial: Creds | null, opts: ContextOpts): PluginContext {
+  let creds = initial;
+  let cursor = conn.cursor;
+  return {
+    http:
+      opts.http ??
+      createHttpClient({
+        ratePerMinute: manifest.rateLimit?.perMinute ?? 60,
+        burst: manifest.rateLimit?.burst ?? 5,
+        userAgent: USER_AGENT,
+        ...opts.httpOptions,
+      }),
+    log: (msg) => opts.onLog?.(msg),
+    config: { ...conn.config, ...(initial ?? {}) },
+    credentials: {
+      get: async () => creds,
+      async save(c) {
+        // Read the current blob, merge, and write in one transaction so concurrent saves cannot lose keys.
+        creds = await db.k.transaction().execute(async (tx) => {
+          let q = tx.selectFrom("source_connections").select("credentials_enc").where("id", "=", conn.id);
+          if (db.dialect === "postgres") q = q.forUpdate();
+          const row = await q.executeTakeFirstOrThrow();
+          const merged = { ...((await readCreds(secret, row.credentials_enc)) ?? {}), ...c };
+          await tx.updateTable("source_connections").set({ credentials_enc: await encryptJson(secret, merged) }).where("id", "=", conn.id).execute();
+          return merged;
+        });
+      },
+    },
+    cursor: {
+      get: async () => cursor,
+      async set(c) {
+        await db.k.updateTable("source_connections").set({ cursor: c }).where("id", "=", conn.id).execute();
+        cursor = c;
+      },
+    },
+    now: () => new Date(),
+    signal: opts.signal ?? new AbortController().signal,
+  };
+}
+
+/**
+ * Build the plugin context for one run. `ctx.http` is rate limited from `manifest.rateLimit` (default 60/min, burst 5)
+ * unless `opts.http` is given. `credentials.save` merges into the current blob and persists it encrypted immediately;
+ * `cursor.set` persists immediately.
+ */
+export async function makePluginContext(db: Db, secret: string, conn: Connection, _runId: string, manifest: PluginManifest, opts: ContextOpts = {}): Promise<PluginContext> {
+  return buildContext(db, secret, conn, manifest, await readCreds(secret, conn.credentialsEnc), opts);
+}
+
+export interface SyncHandlerDeps {
+  db: Db;
+  secret: string;
+  registry: PluginRegistry;
+  http?: HttpClient;
+}
+
+const MSG_CRYPTO = "Cannot decrypt this source's credentials. Was STTRACKER_SECRET changed? Reconnect the source to fix it.";
+
+/** Record a run that failed before `runSync` could start, flag the connection, and stop the job without retry. */
+async function failBeforeRun(db: Db, connectionId: string, mode: "full" | "changes", message: string): Promise<never> {
+  const now = nowIso();
+  await db.k
+    .insertInto("source_runs")
+    .values({ id: newId(), connection_id: connectionId, mode, started_at: now, finished_at: now, status: "failed", counts: "{}", log: message, error: message })
+    .execute();
+  await db.k.updateTable("source_connections").set({ status: "error", last_run_at: now }).where("id", "=", connectionId).execute();
+  throw new NonRetryableError(message);
+}
+
+/**
+ * The `"sync"` job handler. Payload `{ connectionId, mode? }`. Runs `full` when asked, when the plugin cannot
+ * `pullChanges`, or when the last successful full run is over a week old; otherwise `changes`.
+ * A missing or paused connection drops the job. A missing plugin or undecryptable credentials record a failed
+ * run, mark the connection `error` and fail the job without retry. A failed run is rethrown so the worker retries.
+ * `ctx.log` output is appended to the run's `source_runs.log`.
+ */
+export function syncHandler(deps: SyncHandlerDeps): (payload: { connectionId: string; mode?: "full" | "changes" }, jc?: JobContext) => Promise<RunSummary | void> {
+  return async (payload, jc) => {
+    const { db } = deps;
+    const full = await db.k.selectFrom("source_connections").selectAll().where("id", "=", payload.connectionId).executeTakeFirst();
+    if (!full) throw new DropJobError("connection not found");
+    if (full.status === "paused") throw new DropJobError("connection is paused");
+    const plugin = deps.registry.get(full.plugin_id);
+    if (!plugin) return failBeforeRun(db, full.id, payload.mode ?? "full", `Plugin '${full.plugin_id}' is not installed or failed to load.`);
+    let mode: "full" | "changes" = "full";
+    if (plugin.pullChanges && payload.mode !== "full") {
+      const cutoff = new Date(Date.now() - WEEK_MS).toISOString();
+      const recentFull = await db.k
+        .selectFrom("source_runs")
+        .select("id")
+        .where("connection_id", "=", full.id)
+        .where("mode", "=", "full")
+        .where("status", "=", "ok")
+        .where("started_at", ">=", cutoff)
+        .executeTakeFirst();
+      if (recentFull) mode = "changes";
+    }
+    const conn: Connection = {
+      id: full.id,
+      userId: full.user_id,
+      pluginId: full.plugin_id,
+      name: full.name,
+      config: fromJson<Record<string, unknown>>(full.config, {}),
+      credentialsEnc: full.credentials_enc,
+      scheduleMinutes: full.schedule_minutes,
+      cursor: full.cursor,
+      status: full.status,
+      lastRunAt: full.last_run_at,
+      createdAt: full.created_at,
+    };
+    let creds: Creds | null;
+    try {
+      creds = await readCreds(deps.secret, conn.credentialsEnc);
+    } catch (e) {
+      if (e instanceof CryptoError) return failBeforeRun(db, full.id, mode, MSG_CRYPTO);
+      throw e;
+    }
+    const logs: string[] = [];
+    let runId: string | null = null;
+    const makeContext = (c: Connection, id: string) => {
+      runId = id;
+      return buildContext(db, deps.secret, c, plugin.manifest, creds, { http: deps.http, signal: jc?.signal, onLog: (m) => void logs.push(m) });
+    };
+    let summary: RunSummary;
+    try {
+      summary = await runSync({ db, makeContext }, plugin, full.id, mode);
+    } finally {
+      if (runId !== null && logs.length > 0) {
+        const cur = await db.k.selectFrom("source_runs").select("log").where("id", "=", runId).executeTakeFirst();
+        if (cur) {
+          const log = cur.log === "" ? logs.join("\n") : `${logs.join("\n")}\n${cur.log}`;
+          await db.k.updateTable("source_runs").set({ log }).where("id", "=", runId).execute();
+        }
+      }
+    }
+    if (summary.status === "failed") throw new Error(summary.error ?? "sync failed");
+    return summary;
+  };
+}

packages/core/src/jobs.ts

@@ -0,0 +1,221 @@
+import type { Db } from "@sttracker/db/db";
+import { newId, nowIso } from "@sttracker/db/ids";
+import { fromJson, toJson } from "@sttracker/db/json";
+import { sql } from "kysely";
+
+const MAX_ATTEMPTS = 5;
+
+const STALE_LOCK_MS = 60 * 60 * 1000;
+const HEARTBEAT_MS = 5 * 60 * 1000;
+
+export interface JobContext {
+  /** Aborted when the worker is stopped. */
+  signal: AbortSignal;
+}
+export type JobHandler = (payload: any, ctx: JobContext) => Promise<unknown>;
+
+/** Throw from a handler to fail the job at once, with no retry. */
+export class NonRetryableError extends Error {
+  override readonly name = "NonRetryableError";
+}
+/** Throw from a handler to drop the job: it is marked `done` and the message is kept in `last_error`. */
+export class DropJobError extends Error {
+  override readonly name = "DropJobError";
+}
+
+const logErr = (err: unknown) => console.error("[jobs]", err);
+
+/**
+ * Queue a job. Returns its id, or null when a `queued|running` job already holds `lockKey`.
+ * On Postgres the check and insert run under an advisory transaction lock so two callers cannot both pass.
+ */
+export async function enqueue(
+  db: Db,
+  type: string,
+  payload: unknown,
+  opts: { runAt?: string; lockKey?: string } = {},
+): Promise<string | null> {
+  const lockKey = opts.lockKey ?? null;
+  return db.k.transaction().execute(async (tx) => {
+    if (lockKey !== null) {
+      if (db.dialect === "postgres") await sql`select pg_advisory_xact_lock(hashtext(${lockKey}))`.execute(tx);
+      const held = await tx
+        .selectFrom("jobs")
+        .select("id")
+        .where("lock_key", "=", lockKey)
+        .where("status", "in", ["queued", "running"])
+        .executeTakeFirst();
+      if (held) return null;
+    }
+    const id = newId();
+    const now = nowIso();
+    await tx
+      .insertInto("jobs")
+      .values({
+        id,
+        type,
+        payload: toJson(payload ?? null),
+        run_at: opts.runAt ?? now,
+        status: "queued",
+        attempts: 0,
+        last_error: null,
+        locked_at: null,
+        lock_key: lockKey,
+        created_at: now,
+      })
+      .execute();
+    return id;
+  });
+}
+
+/** Claim one queued job atomically; true when this caller won it. */
+async function claim(db: Db, id: string): Promise<boolean> {
+  const r = await db.k
+    .updateTable("jobs")
+    .set({ status: "running", locked_at: nowIso() })
+    .where("id", "=", id)
+    .where("status", "=", "queued")
+    .executeTakeFirst();
+  return r.numUpdatedRows === 1n;
+}
+
+async function runJob(db: Db, handlers: Record<string, JobHandler>, job: { id: string; type: string; payload: string; attempts: number }, signal: AbortSignal, heartbeatMs: number): Promise<void> {
+  // Keep `locked_at` fresh so another process's reaper does not steal a long-running job.
+  const beat = setInterval(() => {
+    db.k.updateTable("jobs").set({ locked_at: nowIso() }).where("id", "=", job.id).where("status", "=", "running").execute().catch(logErr);
+  }, heartbeatMs);
+  try {
+    try {
+      const h = handlers[job.type];
+      if (!h) throw new NonRetryableError(`no handler for job type "${job.type}"`);
+      await h(fromJson<unknown>(job.payload, null), { signal });
+      await db.k.updateTable("jobs").set({ status: "done", locked_at: null, last_error: null }).where("id", "=", job.id).execute();
+    } catch (e) {
+      const attempts = job.attempts + 1;
+      const err = e instanceof Error ? e.message : String(e);
+      if (signal.aborted && !(e instanceof DropJobError) && !(e instanceof NonRetryableError)) {
+        // Interrupted by shutdown: back to the queue untouched, no attempt burned, no backoff.
+        await db.k.updateTable("jobs").set({ status: "queued", locked_at: null }).where("id", "=", job.id).execute();
+      } else if (e instanceof DropJobError) {
+        await db.k.updateTable("jobs").set({ status: "done", locked_at: null, last_error: err }).where("id", "=", job.id).execute();
+      } else if (e instanceof NonRetryableError || attempts >= MAX_ATTEMPTS) {
+        await db.k.updateTable("jobs").set({ status: "failed", attempts, last_error: err, locked_at: null }).where("id", "=", job.id).execute();
+      } else {
+        const runAt = new Date(Date.now() + 2 ** attempts * 60_000).toISOString();
+        await db.k
+          .updateTable("jobs")
+          .set({ status: "queued", attempts, last_error: err, run_at: runAt, locked_at: null })
+          .where("id", "=", job.id)
+          .execute();
+      }
+    }
+  } catch (e) {
+    logErr(e);
+  } finally {
+    clearInterval(beat);
+  }
+}
+
+/** Re-queue jobs a crashed process left `running`, so their lockKey cannot be held forever. */
+async function reapStale(db: Db, staleMs: number, mine: Set<string>): Promise<void> {
+  const cutoff = new Date(Date.now() - staleMs).toISOString();
+  let q = db.k.updateTable("jobs").set({ status: "queued", locked_at: null }).where("status", "=", "running").where("locked_at", "<", cutoff);
+  if (mine.size > 0) q = q.where("id", "not in", [...mine]);
+  await q.execute();
+}
+
+/** Poll for due jobs and run them with bounded concurrency. Failures retry with backoff `2^attempts` minutes, then `failed`. */
+export function startWorker(
+  db: Db,
+  handlers: Record<string, JobHandler>,
+  opts: { pollMs?: number; concurrency?: number; staleMs?: number; heartbeatMs?: number } = {},
+): { stop(): Promise<void> } {
+  const pollMs = opts.pollMs ?? 1000;
+  const concurrency = opts.concurrency ?? 2;
+  const staleMs = opts.staleMs ?? STALE_LOCK_MS;
+  const heartbeatMs = opts.heartbeatMs ?? HEARTBEAT_MS;
+  const inflight = new Set<Promise<void>>();
+  const mine = new Set<string>();
+  const abort = new AbortController();
+  let stopped = false;
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  let tickP: Promise<void> = Promise.resolve();
+
+  async function tick(): Promise<void> {
+    await reapStale(db, staleMs, mine);
+    while (!stopped && inflight.size < concurrency) {
+      const due = await db.k
+        .selectFrom("jobs")
+        .select(["id", "type", "payload", "attempts"])
+        .where("status", "=", "queued")
+        .where("run_at", "<=", nowIso())
+        .orderBy("run_at")
+        .limit(concurrency - inflight.size)
+        .execute();
+      let started = 0;
+      for (const job of due) {
+        if (stopped || inflight.size >= concurrency) break;
+        if (!(await claim(db, job.id))) continue;
+        started++;
+        mine.add(job.id);
+        const p: Promise<void> = runJob(db, handlers, job, abort.signal, heartbeatMs)
+          .catch(logErr)
+          .finally(() => {
+            inflight.delete(p);
+            mine.delete(job.id);
+          });
+        inflight.add(p);
+      }
+      if (started === 0) break;
+    }
+  }
+
+  function schedule() {
+    if (stopped) return;
+    timer = setTimeout(() => {
+      tickP = tick().catch(logErr).finally(schedule);
+    }, pollMs);
+  }
+  tickP = tick().catch(logErr).finally(schedule);
+
+  return {
+    async stop() {
+      stopped = true;
+      abort.abort();
+      if (timer) clearTimeout(timer);
+      await tickP.catch(logErr);
+      await Promise.allSettled([...inflight]);
+    },
+  };
+}
+
+/** Every `tickMs`, enqueue a `sync` job for each non-paused (active or errored) connection whose schedule has come due. */
+export function startScheduler(db: Db, opts: { tickMs?: number } = {}): { stop(): void; tickNow(): Promise<void> } {
+  const tickMs = opts.tickMs ?? 60_000;
+  let stopped = false;
+  async function tick() {
+    const rows = await db.k
+      .selectFrom("source_connections")
+      .select(["id", "schedule_minutes", "last_run_at"])
+      .where("status", "!=", "paused")
+      .where("schedule_minutes", "is not", null)
+      .execute();
+    const now = Date.now();
+    for (const r of rows) {
+      if (stopped) return;
+      const due = r.last_run_at === null || Date.parse(r.last_run_at) + r.schedule_minutes! * 60_000 <= now;
+      if (due) await enqueue(db, "sync", { connectionId: r.id }, { lockKey: `sync:${r.id}` });
+    }
+  }
+  const run = () => tick().catch(logErr);
+  void run();
+  const timer = setInterval(() => void run(), tickMs);
+  return {
+    /** Run one scheduling pass now (tests, manual triggers). */
+    tickNow: run,
+    stop() {
+      stopped = true;
+      clearInterval(timer);
+    },
+  };
+}

packages/core/src/registry.ts

@@ -0,0 +1,65 @@
+import { readdir } from "node:fs/promises";
+import { join } from "node:path";
+import { pathToFileURL } from "node:url";
+import type { PluginManifest, SourcePlugin } from "@sttracker/plugin-sdk/types";
+
+export interface PluginRegistry {
+  get(id: string): SourcePlugin | undefined;
+  list(): PluginManifest[];
+  reload(): Promise<void>;
+}
+
+function validShape(p: unknown): p is SourcePlugin {
+  const m = (p as SourcePlugin | undefined)?.manifest;
+  return (
+    !!m &&
+    typeof m.id === "string" &&
+    m.id.trim() !== "" &&
+    m.sdkVersion === 1 &&
+    typeof m.name === "string" &&
+    Array.isArray(m.capabilities) &&
+    Array.isArray(m.config)
+  );
+}
+
+async function scan(dirs: string[]): Promise<Map<string, SourcePlugin>> {
+  const found = new Map<string, SourcePlugin>();
+  const stamp = Date.now();
+  for (const dir of dirs) {
+    let names: string[];
+    try {
+      names = (await readdir(dir, { withFileTypes: true })).filter((d) => d.isDirectory()).map((d) => d.name).sort();
+    } catch (e) {
+      console.warn(`plugins: cannot read ${dir}: ${e instanceof Error ? e.message : String(e)}`);
+      continue;
+    }
+    for (const name of names) {
+      const file = join(dir, name, "index.ts");
+      if (!(await Bun.file(file).exists())) continue;
+      try {
+        // Bun ignores the `?v=` query for cached .ts modules; evicting the cache entry is what makes reload work.
+        delete require.cache[file];
+        const mod = (await import(`${pathToFileURL(file).href}?v=${stamp}`)) as { default?: unknown };
+        if (!validShape(mod.default)) throw new Error("default export is not a valid plugin (manifest, sdkVersion 1)");
+        const id = mod.default.manifest.id;
+        if (found.has(id)) throw new Error(`duplicate plugin id "${id}"`);
+        found.set(id, mod.default);
+      } catch (e) {
+        console.warn(`plugins: skipping ${file}: ${e instanceof Error ? e.message : String(e)}`);
+      }
+    }
+  }
+  return found;
+}
+
+/** Load every `<dir>/<name>/index.ts` plugin. A bad plugin is logged and skipped, never fatal. */
+export async function loadPlugins(dirs: string[]): Promise<PluginRegistry> {
+  let plugins = await scan(dirs);
+  return {
+    get: (id) => plugins.get(id),
+    list: () => [...plugins.values()].map((p) => p.manifest),
+    async reload() {
+      plugins = await scan(dirs);
+    },
+  };
+}

packages/core/test/connections.test.ts

@@ -0,0 +1,164 @@
+import { expect, spyOn, test } from "bun:test";
+import { join } from "node:path";
+import { forEachDialect } from "@sttracker/db/testing";
+import { decryptJson } from "../src/crypto.ts";
+import { DropJobError, NonRetryableError } from "../src/jobs.ts";
+import { createConnection, deleteConnection, listConnections, makePluginContext, syncHandler, updateConnection } from "../src/connections.ts";
+import { loadPlugins } from "../src/registry.ts";
+import { makeUser } from "./sync/helpers.ts";
+import { newId, nowIso } from "@sttracker/db/ids";
+import type { Connection } from "../src/sync/run.ts";
+
+const SECRET = "test-secret";
+const warn = spyOn(console, "warn").mockImplementation(() => {});
+const reg = await loadPlugins([join(import.meta.dir, "fixtures/plugins")]);
+warn.mockRestore();
+const echo = reg.get("echo")!;
+
+forEachDialect("connections", (getDb) => {
+  const raw = (id: string) => getDb().k.selectFrom("source_connections").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
+  const asConn = async (id: string): Promise<Connection> => {
+    const r = await raw(id);
+    return { id, userId: r.user_id, pluginId: r.plugin_id, name: r.name, config: JSON.parse(r.config), credentialsEnc: r.credentials_enc, scheduleMinutes: null, cursor: r.cursor, status: r.status, lastRunAt: null, createdAt: r.created_at };
+  };
+
+  test("secrets live only in the encrypted blob and never in list", async () => {
+    const db = getDb();
+    const u = await makeUser(db);
+    const id = await createConnection(db, SECRET, u, { pluginId: "echo", name: "E", config: { label: "hi", token: "s3cret" } }, echo.manifest);
+    const r = await raw(id);
+    expect(JSON.parse(r.config)).toEqual({ label: "hi" });
+    expect(r.credentials_enc).not.toContain("s3cret");
+    expect(await decryptJson(SECRET, r.credentials_enc!)).toEqual({ token: "s3cret" });
+    const [v] = await listConnections(db, u);
+    expect(JSON.stringify(v)).not.toContain("s3cret");
+    expect(v).toMatchObject({ id, config: { label: "hi" }, hasCredentials: true });
+    const ctx = await makePluginContext(db, SECRET, await asConn(id), "r", echo.manifest);
+    expect(ctx.config).toEqual({ label: "hi", token: "s3cret" });
+    expect(await ctx.credentials.get()).toEqual({ token: "s3cret" });
+  });
+
+  test("update keeps omitted secrets; delete pauses and clears credentials", async () => {
+    const db = getDb();
+    const u = await makeUser(db);
+    const id = await createConnection(db, SECRET, u, { pluginId: "echo", name: "E", config: { token: "t1" } }, echo.manifest);
+    await updateConnection(db, SECRET, u, id, { config: { label: "new" }, scheduleMinutes: 30 }, echo.manifest);
+    let r = await raw(id);
+    expect(JSON.parse(r.config)).toEqual({ label: "new" });
+    expect(r.schedule_minutes).toBe(30);
+    expect(await decryptJson(SECRET, r.credentials_enc!)).toEqual({ token: "t1" });
+    await updateConnection(db, SECRET, u, id, { config: { token: "t2" } }, echo.manifest);
+    expect(await decryptJson(SECRET, (await raw(id)).credentials_enc!)).toEqual({ token: "t2" });
+    const other = await makeUser(db, "bob");
+    await expect(deleteConnection(db, other, id)).rejects.toThrow();
+    await deleteConnection(db, u, id);
+    r = await raw(id);
+    expect(r.status).toBe("paused");
+    expect(r.credentials_enc).toBeNull();
+    expect((await listConnections(db, u))[0]!.hasCredentials).toBe(false);
+  });
+
+  test("cursor persists on set; http uses manifest rate limit", async () => {
+    const db = getDb();
+    const u = await makeUser(db);
+    const id = await createConnection(db, SECRET, u, { pluginId: "echo", name: "E", config: {} }, echo.manifest);
+    const ctx = await makePluginContext(db, SECRET, await asConn(id), "r", echo.manifest);
+    await ctx.cursor.set("c1");
+    expect((await raw(id)).cursor).toBe("c1");
+    expect(await ctx.cursor.get()).toBe("c1");
+  });
+
+  test("credentials.save persists encrypted before pull continues", async () => {
+    const db = getDb();
+    const u = await makeUser(db);
+    const id = await createConnection(db, SECRET, u, { pluginId: "echo", name: "E", config: { label: "explode", token: "t" } }, echo.manifest);
+    const handler = syncHandler({ db, secret: SECRET, registry: reg });
+    await expect(handler({ connectionId: id })).rejects.toThrow();
+    const r = await raw(id);
+    expect(await decryptJson(SECRET, r.credentials_enc!)).toEqual({ token: "t", refresh: "new-refresh" });
+    expect(r.status).toBe("error");
+  });
+
+  test("sync handler: full first, changes after a recent full, full when payload says so or full is stale", async () => {
+    const db = getDb();
+    const u = await makeUser(db);
+    const id = await createConnection(db, SECRET, u, { pluginId: "echo", name: "E", config: { label: "ok" } }, echo.manifest);
+    const handler = syncHandler({ db, secret: SECRET, registry: reg });
+    const modes = async () => (await db.k.selectFrom("source_runs").select(["mode", "status"]).where("connection_id", "=", id).orderBy("started_at").orderBy("id").execute()).map((r) => `${r.mode}:${r.status}`);
+    await handler({ connectionId: id });
+    await handler({ connectionId: id });
+    await handler({ connectionId: id, mode: "full" });
+    expect(await modes()).toEqual(["full:ok", "changes:ok", "full:ok"]);
+    await db.k.updateTable("source_runs").set({ started_at: new Date(Date.now() - 8 * 86400_000).toISOString() }).where("connection_id", "=", id).execute();
+    await handler({ connectionId: id });
+    expect((await modes()).at(-1)).toBe("full:ok");
+    await db.k.insertInto("source_runs").values({ id: newId(), connection_id: id, mode: "full", started_at: nowIso(), finished_at: null, status: "failed", counts: "{}", log: "", error: "x" }).execute();
+    await handler({ connectionId: id });
+    expect((await modes()).at(-1)).toBe("changes:ok");
+  });
+
+  const setup = async (config: Record<string, unknown>) => {
+    const db = getDb();
+    const u = await makeUser(db);
+    const id = await createConnection(db, SECRET, u, { pluginId: "echo", name: "E", config }, echo.manifest);
+    return { db, id };
+  };
+  const failedRuns = (id: string) => getDb().k.selectFrom("source_runs").selectAll().where("connection_id", "=", id).execute();
+
+  test("missing plugin: failed run, connection error, no retry", async () => {
+    const { db, id } = await setup({});
+    await db.k.updateTable("source_connections").set({ plugin_id: "gone" }).where("id", "=", id).execute();
+    await expect(syncHandler({ db, secret: SECRET, registry: reg })({ connectionId: id })).rejects.toBeInstanceOf(NonRetryableError);
+    const runs = await failedRuns(id);
+    expect(runs).toHaveLength(1);
+    expect(runs[0]).toMatchObject({ status: "failed", mode: "full", error: "Plugin 'gone' is not installed or failed to load." });
+    const c = await raw(id);
+    expect(c.status).toBe("error");
+    expect(c.last_run_at).not.toBeNull();
+  });
+
+  test("undecryptable credentials: failed run, connection error, no retry", async () => {
+    const { db, id } = await setup({ token: "t" });
+    await expect(syncHandler({ db, secret: "other-secret", registry: reg })({ connectionId: id })).rejects.toBeInstanceOf(NonRetryableError);
+    const runs = await failedRuns(id);
+    expect(runs[0]!.error).toBe("Cannot decrypt this source's credentials. Was STTRACKER_SECRET changed? Reconnect the source to fix it.");
+    expect((await raw(id)).status).toBe("error");
+  });
+
+  test("missing or paused connection drops the job", async () => {
+    const { db, id } = await setup({});
+    const h = syncHandler({ db, secret: SECRET, registry: reg });
+    await expect(h({ connectionId: newId() })).rejects.toBeInstanceOf(DropJobError);
+    await db.k.updateTable("source_connections").set({ status: "paused" }).where("id", "=", id).execute();
+    await expect(h({ connectionId: id })).rejects.toBeInstanceOf(DropJobError);
+    expect(await failedRuns(id)).toHaveLength(0);
+  });
+
+  test("ctx.log lands in the run's log", async () => {
+    const { db, id } = await setup({ label: "ok" });
+    const res = await syncHandler({ db, secret: SECRET, registry: reg })({ connectionId: id });
+    const run = await db.k.selectFrom("source_runs").selectAll().where("id", "=", res!.runId).executeTakeFirstOrThrow();
+    expect(run.log).toContain("echo pulled");
+  });
+
+  test("concurrent credentials.save calls keep every key", async () => {
+    const { db, id } = await setup({ token: "t" });
+    const conn = await asConn(id);
+    const ctx = await makePluginContext(db, SECRET, conn, "r", echo.manifest);
+    await Promise.all([ctx.credentials.save({ a: "1" }), ctx.credentials.save({ b: "2" })]);
+    expect(await decryptJson(SECRET, (await raw(id)).credentials_enc!)).toEqual({ token: "t", a: "1", b: "2" });
+  });
+
+  test("ctx.http is limited by manifest.rateLimit", async () => {
+    const { db, id } = await setup({});
+    let t = 0;
+    const sleeps: number[] = [];
+    const manifest = { ...echo.manifest, rateLimit: { perMinute: 60, burst: 1 } };
+    const ctx = await makePluginContext(db, SECRET, await asConn(id), "r", manifest, {
+      httpOptions: { fetchImpl: (async () => Response.json({})) as unknown as typeof fetch, now: () => t, sleep: async (ms) => { sleeps.push(ms); t += ms; } },
+    });
+    await ctx.http.json("https://example.test/a");
+    await ctx.http.json("https://example.test/b");
+    expect(sleeps).toEqual([1000]);
+  });
+});

packages/core/test/fixtures/plugins/broken/index.ts

@@ -0,0 +1 @@
+export default { nope: true };

packages/core/test/fixtures/plugins/echo/index.ts

@@ -0,0 +1,36 @@
+import { definePlugin, type SourceRecord } from "@sttracker/plugin-sdk/types";
+
+export default definePlugin({
+  manifest: {
+    id: "echo",
+    name: "Echo",
+    version: "0.0.1",
+    sdkVersion: 1,
+    mediaTypes: ["book"],
+    capabilities: ["pull", "pullChanges"],
+    config: [
+      { key: "label", label: "Label", type: "string" },
+      { key: "token", label: "Token", type: "secret" },
+    ],
+    rateLimit: { perMinute: 30, burst: 2 },
+  },
+  async *pull(ctx) {
+    const cfg = ctx.config;
+    if (cfg.label === "explode") {
+      await ctx.credentials.save({ refresh: "new-refresh" });
+      throw new Error("boom after save");
+    }
+    ctx.log("echo pulled");
+    const rec: SourceRecord = {
+      kind: "entry",
+      key: "echo-1",
+      raw: { label: cfg.label ?? null },
+      work: { title: String(cfg.label ?? "Echo"), mediaType: "book", externalIds: { isbn13: "9780000000002" } },
+      entry: { status: "completed" },
+    };
+    yield rec;
+  },
+  async *pullChanges(ctx) {
+    yield* this.pull!(ctx);
+  },
+});

packages/core/test/jobs.test.ts

@@ -0,0 +1,195 @@
+import { expect, spyOn, test } from "bun:test";
+import { forEachDialect } from "@sttracker/db/testing";
+import { DropJobError, enqueue, NonRetryableError, startScheduler, startWorker } from "../src/jobs.ts";
+import { makeConnection, makeUser } from "./sync/helpers.ts";
+
+const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
+
+forEachDialect("jobs", (getDb) => {
+  test("enqueue with same lockKey returns null", async () => {
+    const db = getDb();
+    expect(await enqueue(db, "sync", {}, { lockKey: "sync:a" })).toBeString();
+    expect(await enqueue(db, "sync", {}, { lockKey: "sync:a" })).toBeNull();
+    expect(await enqueue(db, "sync", {}, { lockKey: "sync:b" })).toBeString();
+    expect(await enqueue(db, "x", {})).toBeString();
+    expect(await enqueue(db, "x", {})).toBeString();
+  });
+
+  test("concurrent enqueue with one lockKey admits exactly one", async () => {
+    const db = getDb();
+    const ids = await Promise.all(Array.from({ length: 6 }, () => enqueue(db, "sync", {}, { lockKey: "k" })));
+    expect(ids.filter((i) => i !== null)).toHaveLength(1);
+  });
+
+  test("worker runs a job and releases the lock", async () => {
+    const db = getDb();
+    const seen: unknown[] = [];
+    const id = await enqueue(db, "t", { n: 1 }, { lockKey: "l" });
+    const w = startWorker(db, { t: async (p) => void seen.push(p) }, { pollMs: 10 });
+    await sleep(150);
+    await w.stop();
+    expect(seen).toEqual([{ n: 1 }]);
+    expect((await db.k.selectFrom("jobs").selectAll().where("id", "=", id!).executeTakeFirstOrThrow()).status).toBe("done");
+    expect(await enqueue(db, "t", {}, { lockKey: "l" })).toBeString();
+  });
+
+  test("a job is claimed by only one of two workers", async () => {
+    const db = getDb();
+    let runs = 0;
+    const h = { t: async () => { runs++; await sleep(30); } };
+    await enqueue(db, "t", {});
+    const a = startWorker(db, h, { pollMs: 5 });
+    const b = startWorker(db, h, { pollMs: 5 });
+    await sleep(200);
+    await Promise.all([a.stop(), b.stop()]);
+    expect(runs).toBe(1);
+  });
+
+  test("worker retries then fails", async () => {
+    const db = getDb();
+    let calls = 0;
+    const id = (await enqueue(db, "bad", {}))!;
+    const w = startWorker(db, { bad: async () => { calls++; throw new Error("nope"); } }, { pollMs: 10 });
+    const row = () => db.k.selectFrom("jobs").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
+    for (let attempt = 1; attempt <= 5; attempt++) {
+      for (let i = 0; i < 100 && (await row()).attempts < attempt; i++) await sleep(10);
+      const r = await row();
+      expect(r.attempts).toBe(attempt);
+      expect(r.last_error).toBe("nope");
+      if (attempt < 5) {
+        expect(r.status).toBe("queued");
+        const delay = Date.parse(r.run_at) - Date.now();
+        expect(delay).toBeGreaterThan(2 ** attempt * 60_000 - 5_000);
+        expect(delay).toBeLessThanOrEqual(2 ** attempt * 60_000);
+        await db.k.updateTable("jobs").set({ run_at: new Date(Date.now() - 1000).toISOString() }).where("id", "=", id).execute();
+      }
+    }
+    await w.stop();
+    expect((await row()).status).toBe("failed");
+    expect(calls).toBe(5);
+  });
+
+  test("scheduler enqueues due connections only", async () => {
+    const db = getDb();
+    const u = await makeUser(db);
+    const due = await makeConnection(db, u, "p");
+    const fresh = await makeConnection(db, u, "p");
+    const manual = await makeConnection(db, u, "p");
+    const errored = await makeConnection(db, u, "p");
+    const paused = await makeConnection(db, u, "p");
+    const old = new Date(Date.now() - 120 * 60_000).toISOString();
+    const recent = new Date(Date.now() - 5 * 60_000).toISOString();
+    const set = (id: string, v: object) => db.k.updateTable("source_connections").set(v).where("id", "=", id).execute();
+    await set(due, { schedule_minutes: 60, last_run_at: old });
+    await set(fresh, { schedule_minutes: 60, last_run_at: recent });
+    await set(errored, { schedule_minutes: 60, last_run_at: old, status: "error" });
+    await set(paused, { schedule_minutes: 60, last_run_at: old, status: "paused" });
+    void manual;
+    const s = startScheduler(db, { tickMs: 3_600_000 });
+    await s.tickNow();
+    await s.tickNow();
+    s.stop();
+    const jobs = await db.k.selectFrom("jobs").selectAll().execute();
+    expect(jobs.map((j) => j.lock_key).sort()).toEqual([`sync:${due}`, `sync:${errored}`].sort());
+    expect(JSON.parse(jobs.find((j) => j.lock_key === `sync:${due}`)!.payload)).toEqual({ connectionId: due });
+  });
+
+  const row = (db: ReturnType<typeof getDb>, id: string) => db.k.selectFrom("jobs").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
+  async function until(fn: () => Promise<boolean>) {
+    for (let i = 0; i < 200 && !(await fn()); i++) await sleep(10);
+  }
+
+  test("NonRetryableError fails at once; DropJobError completes with last_error", async () => {
+    const db = getDb();
+    const a = (await enqueue(db, "a", {}))!;
+    const b = (await enqueue(db, "b", {}))!;
+    const w = startWorker(db, { a: async () => { throw new NonRetryableError("fatal"); }, b: async () => { throw new DropJobError("gone"); } }, { pollMs: 10 });
+    await until(async () => (await row(db, a)).status !== "queued" && (await row(db, b)).status !== "queued");
+    await w.stop();
+    expect(await row(db, a)).toMatchObject({ status: "failed", attempts: 1, last_error: "fatal" });
+    expect(await row(db, b)).toMatchObject({ status: "done", last_error: "gone" });
+  });
+
+  test("a stale running job is re-queued and then runs; a fresh one is left", async () => {
+    const db = getDb();
+    const stale = (await enqueue(db, "t", {}, { lockKey: "s" }))!;
+    const fresh = (await enqueue(db, "never", {}))!;
+    await db.k.updateTable("jobs").set({ status: "running", locked_at: new Date(Date.now() - 2 * 3600_000).toISOString(), attempts: 2 }).where("id", "=", stale).execute();
+    await db.k.updateTable("jobs").set({ status: "running", locked_at: nowStr() }).where("id", "=", fresh).execute();
+    let ran = 0;
+    const w = startWorker(db, { t: async () => void ran++ }, { pollMs: 10 });
+    await until(async () => (await row(db, stale)).status === "done");
+    await w.stop();
+    expect(ran).toBe(1);
+    expect((await row(db, stale)).attempts).toBe(2);
+    expect((await row(db, fresh)).status).toBe("running");
+  });
+
+  test("failing bookkeeping is logged and stop() does not reject", async () => {
+    const db = getDb();
+    const err = spyOn(console, "error").mockImplementation(() => {});
+    await enqueue(db, "t", {});
+    let updates = 0;
+    const k = new Proxy(db.k, {
+      get(t, p) {
+        const v = Reflect.get(t, p, t);
+        if (p === "updateTable") return (...a: unknown[]) => { if (++updates > 1) throw new Error("db down"); return (v as Function).apply(t, a); };
+        return typeof v === "function" ? v.bind(t) : v;
+      },
+    });
+    let called = false;
+    const w = startWorker({ ...db, k } as typeof db, { t: async () => void (called = true) }, { pollMs: 10 });
+    await until(async () => called && err.mock.calls.length > 0);
+    await expect(w.stop()).resolves.toBeUndefined();
+    expect(err.mock.calls.some((c) => c[0] === "[jobs]")).toBe(true);
+    err.mockRestore();
+  });
+
+  test("a job running past staleMs is not reaped by its own worker and runs once", async () => {
+    const db = getDb();
+    const id = (await enqueue(db, "t", {}, { lockKey: "slow" }))!;
+    let runs = 0;
+    const w = startWorker(db, { t: async () => { runs++; await sleep(300); } }, { pollMs: 10, staleMs: 50, heartbeatMs: 20 });
+    await until(async () => (await row(db, id)).status === "done");
+    await w.stop();
+    expect(runs).toBe(1);
+  });
+
+  test("heartbeat refreshes locked_at", async () => {
+    const db = getDb();
+    const id = (await enqueue(db, "t", {}))!;
+    let release!: () => void;
+    const gate = new Promise<void>((r) => (release = r));
+    const w = startWorker(db, { t: () => gate }, { pollMs: 10, heartbeatMs: 20 });
+    await until(async () => (await row(db, id)).status === "running");
+    const first = (await row(db, id)).locked_at!;
+    await until(async () => (await row(db, id)).locked_at !== first);
+    expect((await row(db, id)).locked_at).not.toBe(first);
+    release();
+    await w.stop();
+  });
+
+  test("stop() re-queues an aborted job without burning an attempt or backoff", async () => {
+    const db = getDb();
+    const id = (await enqueue(db, "t", {}))!;
+    const before = (await row(db, id)).run_at;
+    let started = false;
+    const w = startWorker(db, { t: (_p, { signal }) => new Promise<void>((_, rej) => { started = true; signal.addEventListener("abort", () => rej(new Error("aborted"))); }) }, { pollMs: 10 });
+    await until(async () => started);
+    await w.stop();
+    expect(await row(db, id)).toMatchObject({ status: "queued", attempts: 0, run_at: before, locked_at: null });
+  });
+
+  test("stop() aborts the signal handed to handlers", async () => {
+    const db = getDb();
+    await enqueue(db, "t", {});
+    let aborted = false;
+    let started = false;
+    const w = startWorker(db, { t: (_p, { signal }) => new Promise<void>((res) => { started = true; signal.addEventListener("abort", () => { aborted = true; res(); }); }) }, { pollMs: 10 });
+    await until(async () => started);
+    await w.stop();
+    expect(aborted).toBe(true);
+  });
+});
+
+const nowStr = () => new Date().toISOString();

packages/core/test/registry.test.ts

@@ -0,0 +1,44 @@
+import { expect, spyOn, test } from "bun:test";
+import { mkdtemp, mkdir, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { loadPlugins } from "../src/registry.ts";
+
+const dir = join(import.meta.dir, "fixtures/plugins");
+
+test("registry loads echo plugin and skips broken one", async () => {
+  const warn = spyOn(console, "warn").mockImplementation(() => {});
+  const reg = await loadPlugins([dir, join(dir, "does-not-exist")]);
+  expect(reg.list().map((m) => m.id)).toEqual(["echo"]);
+  expect(reg.get("echo")?.manifest.name).toBe("Echo");
+  expect(reg.get("broken")).toBeUndefined();
+  expect(warn.mock.calls.some((c) => String(c[0]).includes("broken"))).toBe(true);
+  await reg.reload();
+  expect(reg.list()).toHaveLength(1);
+  warn.mockRestore();
+});
+
+test("duplicate ids across dirs keep the first", async () => {
+  const warn = spyOn(console, "warn").mockImplementation(() => {});
+  const reg = await loadPlugins([dir, dir]);
+  expect(reg.list()).toHaveLength(1);
+  warn.mockRestore();
+});
+
+test("reload picks up an edited entry file", async () => {
+  const root = await mkdtemp(join(tmpdir(), "sttracker-plugins-"));
+  try {
+    await mkdir(join(root, "hot"));
+    const write = (v: string) =>
+      writeFile(join(root, "hot", "index.ts"), `export default { manifest: { id: "hot", name: "Hot", version: "${v}", sdkVersion: 1, mediaTypes: [], capabilities: [], config: [] } };\n`);
+    await write("1.0.0");
+    const reg = await loadPlugins([root]);
+    expect(reg.get("hot")?.manifest.version).toBe("1.0.0");
+    await new Promise((r) => setTimeout(r, 5));
+    await write("2.0.0");
+    await reg.reload();
+    expect(reg.get("hot")?.manifest.version).toBe("2.0.0");
+  } finally {
+    await rm(root, { recursive: true, force: true });
+  }
+});

packages/plugin-sdk/src/types.ts

@@ -7,7 +7,7 @@ export type Precision = "day" | "month" | "year" | "unknown";
 export interface PartialDate { value: string; precision: Precision }
 export type Capability = "pull" | "pullChanges" | "push" | "ownedLibrary" | "ingest" | "lookup" | "importFile" | "exportFile";
 export interface ConfigField { key: string; label: string; type: "string" | "secret" | "number" | "boolean" | "select"; required?: boolean; options?: string[]; help?: string }
-export interface PluginManifest { id: string; name: string; version: string; sdkVersion: 1; mediaTypes: MediaType[]; capabilities: Capability[]; config: ConfigField[]; attribution?: string }
+export interface PluginManifest { id: string; name: string; version: string; sdkVersion: 1; mediaTypes: MediaType[]; capabilities: Capability[]; config: ConfigField[]; attribution?: string; rateLimit?: { perMinute: number; burst?: number } }
 export interface WorkHint { title: string; mediaType: MediaType; year?: number; creators?: { name: string; role?: string }[]; externalIds: Record<string, string>; coverUrl?: string; description?: string; parent?: WorkHint; position?: { season?: number; episode?: number; index?: number } }
 export interface EntryFields { status?: Status; rating?: number | null; review?: string | null; notes?: string | null; favorite?: boolean; tags?: string[] }
 export interface EngagementFields { startedAt?: PartialDate | null; finishedAt?: PartialDate | null; progress?: Record<string, number | string>; platform?: string | null; rating?: number | null; notes?: string | null }

plugins/hardcover/index.ts

@@ -4,8 +4,6 @@ import { gql, LISTS_QUERY, PAGE_SIZE, SEARCH_QUERY, USER_BOOKS_QUERY } from "./s
 import { mapList, mapUserBook } from "./src/map.ts";
 import type { HcList, HcUserBook } from "./src/map.ts";
 
-export { RATE_PER_MINUTE } from "./src/graphql.ts";
-
 async function* pageUserBooks(ctx: PluginContext, where: Record<string, unknown> | null): AsyncGenerator<HcUserBook> {
   for (let offset = 0; ; offset += PAGE_SIZE) {
     const d = await gql<{ me: { user_books: HcUserBook[] }[] }>(ctx, USER_BOOKS_QUERY, { limit: PAGE_SIZE, offset, where });
@@ -45,6 +43,7 @@ export default definePlugin({
     name: "Hardcover",
     version: "0.1.0",
     sdkVersion: 1,
+    rateLimit: { perMinute: 50 },
     mediaTypes: ["book"],
     capabilities: ["pull", "pullChanges", "lookup"],
     config: [{ key: "token", label: "Personal access token", type: "secret", required: true }],

plugins/hardcover/src/graphql.ts

@@ -2,7 +2,6 @@ import { HttpError } from "@sttracker/plugin-sdk/http";
 import type { PluginContext } from "@sttracker/plugin-sdk/types";
 
 export const ENDPOINT = "https://api.hardcover.app/v1/graphql";
-export const RATE_PER_MINUTE = 50;
 export const PAGE_SIZE = 100;
 
 const BOOK = `id title release_year contributions { author { name } }`;

plugins/lastfm/index.ts

@@ -7,6 +7,7 @@ export default definePlugin({
     name: "Last.fm",
     version: "0.1.0",
     sdkVersion: 1,
+    rateLimit: { perMinute: 240 },
     mediaTypes: ["track", "album", "artist"],
     capabilities: ["pull", "pullChanges"],
     config: [

plugins/listenbrainz/index.ts

@@ -7,6 +7,7 @@ export default definePlugin({
     name: "ListenBrainz",
     version: "0.1.0",
     sdkVersion: 1,
+    rateLimit: { perMinute: 60 },
     mediaTypes: ["track", "album", "artist"],
     capabilities: ["pull", "pullChanges"],
     config: [

plugins/mangabaka/index.ts

@@ -8,6 +8,7 @@ export default definePlugin({
     name: "MangaBaka",
     version: "0.1.0",
     sdkVersion: 1,
+    rateLimit: { perMinute: 150 },
     mediaTypes: ["manga", "comic"],
     capabilities: ["pull", "pullChanges", "lookup"],
     config: [{ key: "token", label: "API token", type: "secret", required: true, help: "Sent as the x-api-key header." }],

plugins/mangabaka/src/api.ts

@@ -1,8 +1,6 @@
 import type { PluginContext } from "@sttracker/plugin-sdk/types";
 
 export const BASE_URL = "https://api.mangabaka.org";
-/** The host builds the HttpClient; it should use this rate for this plugin. */
-export const RATE_PER_MINUTE = 150;
 
 export interface ApiSource { id?: number | string | null }
 export interface ApiSeries {

plugins/trakt/index.ts

@@ -12,6 +12,7 @@ export default definePlugin({
     name: "Trakt",
     version: "0.1.0",
     sdkVersion: 1,
+    rateLimit: { perMinute: 100 },
     mediaTypes: ["movie", "show", "season", "episode"],
     capabilities: ["pull", "pullChanges", "lookup"],
     config: [

plugins/trakt/src/api.ts

@@ -2,8 +2,6 @@ import { HttpError } from "@sttracker/plugin-sdk/http";
 import type { PluginContext } from "@sttracker/plugin-sdk/types";
 
 export const BASE = "https://api.trakt.tv";
-/** Rate the host should configure for this plugin's HttpClient. */
-export const RATE_PER_MINUTE = 100;
 const REFRESH_WINDOW_MS = 24 * 60 * 60 * 1000;
 const REDIRECT_URI = "urn:ietf:wg:oauth:2.0:oob";