STTracker/packages/core/test/connections.test.ts

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(r.deleted_at).not.toBeNull();
    expect(await listConnections(db, u)).toEqual([]);
  });

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