feat(core): jobs, scheduler, plugin registry, connections

Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-10-30 20:39:17 +01:00
parent 207c92bb4d
commit 166b8ecb5e
20 changed files with 744 additions and 10 deletions
@@ -0,0 +1,229 @@
+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 } from "@sttracker/plugin-sdk/http";
+import type { HttpClient, PluginContext, PluginManifest } from "@sttracker/plugin-sdk/types";
+import { decryptJson, encryptJson } from "./crypto.ts";
+import { NotFoundError, ValidationError } from "./errors.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. */
+export async function updateConnection(db: Db, secret: string, userId: string, id: string, patch: ConnectionPatch, manifest: PluginManifest): Promise<void> {
+  const row = await db.k.selectFrom("source_connections").selectAll().where("id", "=", id).where("user_id", "=", userId).executeTakeFirst();
+  if (!row) throw new NotFoundError("connection");
+  if (row.plugin_id !== manifest.id) throw new ValidationError("pluginId", "manifest_mismatch");
+  checkSchedule(patch.scheduleMinutes);
+  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 db.k.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)";
+
+/**
+ * Build the plugin context for one run. `ctx.http` is rate limited from `manifest.rateLimit` (default 60/min, burst 5)
+ * unless `http` is given. `credentials.save` merges into the blob and persists it encrypted immediately;
+ * `cursor.set` persists immediately.
+ */
+export async function makePluginContext(db: Db, secret: string, conn: Connection, _runId: string, manifest: PluginManifest, http?: HttpClient): Promise<PluginContext> {
+  let creds = await readCreds(secret, conn.credentialsEnc);
+  let cursor = conn.cursor;
+  return {
+    http:
+      http ??
+      createHttpClient({ ratePerMinute: manifest.rateLimit?.perMinute ?? 60, burst: manifest.rateLimit?.burst ?? 5, userAgent: USER_AGENT }),
+    log: () => {},
+    config: { ...conn.config, ...(creds ?? {}) },
+    credentials: {
+      get: async () => creds,
+      async save(c) {
+        const merged = { ...(creds ?? {}), ...c };
+        await db.k.updateTable("source_connections").set({ credentials_enc: await encryptJson(secret, merged) }).where("id", "=", conn.id).execute();
+        creds = 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: new AbortController().signal,
+  };
+}
+
+export interface SyncHandlerDeps {
+  db: Db;
+  secret: string;
+  registry: PluginRegistry;
+  http?: HttpClient;
+}
+
+/**
+ * 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 failed run is recorded by `runSync` and rethrown so the worker retries with backoff.
+ */
+export function syncHandler(deps: SyncHandlerDeps): (payload: { connectionId: string; mode?: "full" | "changes" }) => Promise<RunSummary | void> {
+  return async (payload) => {
+    const { db } = deps;
+    const row = await db.k.selectFrom("source_connections").select(["id", "plugin_id", "status"]).where("id", "=", payload.connectionId).executeTakeFirst();
+    if (!row) throw new NotFoundError("connection");
+    if (row.status === "paused") return;
+    const plugin = deps.registry.get(row.plugin_id);
+    if (!plugin) throw new NotFoundError(`plugin ${row.plugin_id}`);
+    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", "=", row.id)
+        .where("mode", "=", "full")
+        .where("status", "=", "ok")
+        .where("started_at", ">=", cutoff)
+        .executeTakeFirst();
+      if (recentFull) mode = "changes";
+    }
+    const full = (await db.k.selectFrom("source_connections").selectAll().where("id", "=", row.id).executeTakeFirstOrThrow());
+    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,
+    };
+    const ctx = await makePluginContext(db, deps.secret, conn, "sync", plugin.manifest, deps.http);
+    const summary = await runSync({ db, makeContext: () => ctx }, plugin, row.id, mode);
+    if (summary.status === "failed") throw new Error(summary.error ?? "sync failed");
+    return summary;
+  };
+}
@@ -0,0 +1,168 @@
+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;
+
+export type JobHandler = (payload: any) => Promise<void>;
+
+/**
+ * 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 }) {
+  try {
+    const h = handlers[job.type];
+    if (!h) throw new Error(`no handler for job type "${job.type}"`);
+    await h(fromJson<unknown>(job.payload, null));
+    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 (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();
+    }
+  }
+}
+
+/** 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 } = {},
+): { stop(): Promise<void> } {
+  const pollMs = opts.pollMs ?? 1000;
+  const concurrency = opts.concurrency ?? 2;
+  const inflight = new Set<Promise<void>>();
+  let stopped = false;
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  let tickP: Promise<void> = Promise.resolve();
+
+  async function tick(): Promise<void> {
+    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++;
+        const p = runJob(db, handlers, job).finally(() => inflight.delete(p));
+        inflight.add(p);
+      }
+      if (started === 0) break;
+    }
+  }
+
+  function schedule() {
+    if (stopped) return;
+    timer = setTimeout(() => {
+      tickP = tick().catch(() => {}).finally(schedule);
+    }, pollMs);
+  }
+  tickP = tick().catch(() => {}).finally(schedule);
+
+  return {
+    async stop() {
+      stopped = true;
+      if (timer) clearTimeout(timer);
+      await tickP;
+      await Promise.all(inflight);
+    },
+  };
+}
+
+/** Every `tickMs`, enqueue a `sync` job for each active 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", "=", "active")
+      .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(() => {});
+  void run();
+  const timer = setInterval(() => void run(), tickMs);
+  return {
+    /** Run one scheduling pass now (tests, manual triggers). */
+    tickNow: run,
+    stop() {
+      stopped = true;
+      clearInterval(timer);
+    },
+  };
+}
@@ -0,0 +1,63 @@
+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 {
+        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);
+    },
+  };
+}