From 7b0a6738dda8faf5c55fe5fb9d2b24ed076e76f4 Mon Sep 17 00:00:00 2001 From: Lucas Winther Date: Fri, 30 Oct 2026 21:10:39 +0100 Subject: [PATCH] feat(core): API support helpers: token list, suggestions, attachment links, connection runs, device auth, file-mode sync Co-Authored-By: Claude Opus 5.5 --- packages/core/src/attachments.ts | 68 +++++++++++- packages/core/src/auth.ts | 28 +++++ packages/core/src/connections.ts | 123 ++++++++++++++++++++-- packages/core/src/entries.ts | 31 ++++++ packages/core/test/api-support.test.ts | 140 +++++++++++++++++++++++++ 5 files changed, 378 insertions(+), 12 deletions(-) create mode 100644 packages/core/test/api-support.test.ts diff --git a/packages/core/src/attachments.ts b/packages/core/src/attachments.ts index f86b93b..6c85d32 100644 --- a/packages/core/src/attachments.ts +++ b/packages/core/src/attachments.ts @@ -3,7 +3,7 @@ import { dirname, join } from "node:path"; import type { Db } from "@sttracker/db/db"; import { newId, nowIso } from "@sttracker/db/ids"; import type { AttachmentEntity, AttachmentsTable } from "@sttracker/db/schema"; -import { ValidationError } from "./errors.ts"; +import { NotFoundError, ValidationError } from "./errors.ts"; export type Attachment = AttachmentsTable; @@ -127,15 +127,15 @@ export async function readAttachment( dataDir: string, userId: string, id: string, -): Promise<{ file: ReturnType; mime: string } | null> { +): Promise<{ file: ReturnType; mime: string; filename: string } | null> { const row = await db.k .selectFrom("attachments") - .select(["sha256", "mime"]) + .select(["sha256", "mime", "filename"]) .where("id", "=", id) .where("user_id", "=", userId) .executeTakeFirst(); if (!row) return null; - return { file: Bun.file(pathFor(dataDir, row.sha256)), mime: row.mime }; + return { file: Bun.file(pathFor(dataDir, row.sha256)), mime: row.mime, filename: row.filename }; } /** Attachments of `userId` linked to the given entity, oldest first. */ @@ -201,3 +201,63 @@ export async function unlinkAttachment( if (orphan) await rm(pathFor(dataDir, att.sha256), { force: true }); }); } + +const LINK_ENTITIES: readonly AttachmentEntity[] = ["entry", "engagement", "copy", "list", "work"]; + +/** + * Throw `NotFoundError` unless `entityId` is a live `entity` the user may attach to: their own entry, + * engagement, copy or list, or a work that is shared or theirs. Call before `storeAttachment`. + */ +export async function requireLinkTarget(db: Db, userId: string, entity: AttachmentEntity, entityId: string): Promise { + if (!LINK_ENTITIES.includes(entity)) throw new ValidationError("entity", "invalid"); + const k = db.k; + let found: unknown; + switch (entity) { + case "entry": + found = await k.selectFrom("entries").select("id").where("id", "=", entityId).where("user_id", "=", userId).where("deleted_at", "is", null).executeTakeFirst(); + break; + case "engagement": + found = await k + .selectFrom("engagements as g") + .innerJoin("entries as e", "e.id", "g.entry_id") + .select("g.id") + .where("g.id", "=", entityId) + .where("e.user_id", "=", userId) + .where("g.deleted_at", "is", null) + .where("e.deleted_at", "is", null) + .executeTakeFirst(); + break; + case "copy": + found = await k.selectFrom("copies").select("id").where("id", "=", entityId).where("user_id", "=", userId).where("deleted_at", "is", null).executeTakeFirst(); + break; + case "list": + found = await k.selectFrom("lists").select("id").where("id", "=", entityId).where("user_id", "=", userId).where("deleted_at", "is", null).executeTakeFirst(); + break; + case "work": + found = await k + .selectFrom("works") + .select("id") + .where("id", "=", entityId) + .where((eb) => eb.or([eb("owner_user_id", "is", null), eb("owner_user_id", "=", userId)])) + .executeTakeFirst(); + break; + } + if (!found) throw new NotFoundError(entity); +} + +/** The links of one of the user's attachments (empty for another user's or a missing attachment). */ +export async function listAttachmentLinks( + db: Db, + userId: string, + attachmentId: string, +): Promise<{ id: string; entity: AttachmentEntity; entityId: string }[]> { + const rows = await db.k + .selectFrom("attachment_links as l") + .innerJoin("attachments as a", "a.id", "l.attachment_id") + .select(["l.id", "l.entity", "l.entity_id"]) + .where("a.id", "=", attachmentId) + .where("a.user_id", "=", userId) + .orderBy("l.id") + .execute(); + return rows.map((r) => ({ id: r.id, entity: r.entity, entityId: r.entity_id })); +} diff --git a/packages/core/src/auth.ts b/packages/core/src/auth.ts index b10f1f6..8bf546d 100644 --- a/packages/core/src/auth.ts +++ b/packages/core/src/auth.ts @@ -198,3 +198,31 @@ export async function revokeAppToken(db: Db, userId: string, id: string): Promis .where("revoked_at", "is", null) .execute(); } + +export interface AppTokenInfo { + id: string; + name: string; + scopes: TokenScope[]; + lastUsedAt: string | null; + revokedAt: string | null; + createdAt: string; +} + +/** The user's app tokens, newest first. Metadata only: the token and its hash are never returned. */ +export async function listAppTokens(db: Db, userId: string): Promise { + const rows = await db.k + .selectFrom("app_tokens") + .select(["id", "name", "scopes", "last_used_at", "revoked_at", "created_at"]) + .where("user_id", "=", userId) + .orderBy("created_at", "desc") + .orderBy("id", "desc") + .execute(); + return rows.map((r) => ({ + id: r.id, + name: r.name, + scopes: fromJson(r.scopes, []), + lastUsedAt: r.last_used_at, + revokedAt: r.revoked_at, + createdAt: r.created_at, + })); +} diff --git a/packages/core/src/connections.ts b/packages/core/src/connections.ts index 7a2c275..bb12cc8 100644 --- a/packages/core/src/connections.ts +++ b/packages/core/src/connections.ts @@ -1,14 +1,14 @@ 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 type { ConnectionStatus, RunMode, RunStatus } 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"; +import { runSync, type Connection, type RunSummary, type SyncMode } from "./sync/run.ts"; const WEEK_MS = 7 * 24 * 60 * 60 * 1000; @@ -143,6 +143,104 @@ export async function listConnections(db: Db, userId: string): Promise): boolean { + return c.status === "paused" && !c.hasCredentials; +} + +async function ownedRow(db: Db, userId: string, id: string) { + const r = await db.k.selectFrom("source_connections").selectAll().where("id", "=", id).where("user_id", "=", userId).executeTakeFirst(); + if (!r || (r.status === "paused" && r.credentials_enc === null)) throw new NotFoundError("connection"); + return r; +} + +/** One of the user's live connections, without secrets. Deleted or foreign connections throw `NotFoundError`. */ +export async function getConnection(db: Db, userId: string, id: string): Promise { + const r = await ownedRow(db, userId, id); + return { + id: r.id, + pluginId: r.plugin_id, + name: r.name, + config: fromJson>(r.config, {}), + scheduleMinutes: r.schedule_minutes, + status: r.status, + lastRunAt: r.last_run_at, + createdAt: r.created_at, + hasCredentials: r.credentials_enc !== null, + }; +} + +export interface RunView { + id: string; + mode: RunMode; + status: RunStatus; + startedAt: string; + finishedAt: string | null; + counts: Record; + log: string; + error: string | null; +} + +/** Runs of one of the user's connections, newest first (at most `limit`, default 50). */ +export async function listRuns(db: Db, userId: string, connectionId: string, limit = 50): Promise { + await ownedRow(db, userId, connectionId); + const rows = await db.k + .selectFrom("source_runs") + .selectAll() + .where("connection_id", "=", connectionId) + .orderBy("started_at", "desc") + .orderBy("id", "desc") + .limit(Math.min(Math.max(limit, 1), 200)) + .execute(); + return rows.map((r) => ({ + id: r.id, + mode: r.mode, + status: r.status, + startedAt: r.started_at, + finishedAt: r.finished_at, + counts: fromJson>(r.counts, {}), + log: r.log, + error: r.error, + })); +} + +async function deviceContext(db: Db, secret: string, registry: PluginRegistry, userId: string, id: string) { + const r = await ownedRow(db, userId, id); + const plugin = registry.get(r.plugin_id); + if (!plugin) throw new ValidationError("pluginId", "not_installed"); + if (!plugin.deviceAuth) throw new ValidationError("pluginId", "no_device_auth"); + const conn: Connection = { + id: r.id, + userId: r.user_id, + pluginId: r.plugin_id, + name: r.name, + config: fromJson>(r.config, {}), + credentialsEnc: r.credentials_enc, + scheduleMinutes: r.schedule_minutes, + cursor: r.cursor, + status: r.status, + lastRunAt: r.last_run_at, + createdAt: r.created_at, + }; + const ctx = await makePluginContext(db, secret, conn, "device", plugin.manifest); + return { deviceAuth: plugin.deviceAuth, ctx }; +} + +/** Start a device-code sign-in for one of the user's connections. */ +export async function deviceStart(db: Db, secret: string, registry: PluginRegistry, userId: string, id: string) { + const { deviceAuth, ctx } = await deviceContext(db, secret, registry, userId, id); + return deviceAuth.start(ctx); +} + +/** Poll a device-code sign-in. On success the returned credentials are merged into the encrypted blob. */ +export async function devicePoll(db: Db, secret: string, registry: PluginRegistry, userId: string, id: string, deviceCode: string): Promise<"pending" | "connected"> { + const { deviceAuth, ctx } = await deviceContext(db, secret, registry, userId, id); + const res = await deviceAuth.poll(ctx, deviceCode); + if (res === "pending") return "pending"; + await ctx.credentials.save(res); + return "connected"; +} + export const USER_AGENT = "STTracker/0.1 (+self-hosted)"; export interface ContextOpts { @@ -215,7 +313,7 @@ export interface SyncHandlerDeps { 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 { +async function failBeforeRun(db: Db, connectionId: string, mode: RunMode, message: string): Promise { const now = nowIso(); await db.k .insertInto("source_runs") @@ -232,16 +330,25 @@ async function failBeforeRun(db: Db, connectionId: string, mode: "full" | "chang * 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 { +export interface SyncPayload { + connectionId: string; + mode?: "full" | "changes"; + /** File mode: run the plugin's `importFile` over these bytes instead of pulling. */ + file?: { name: string; base64: string }; +} + +export function syncHandler(deps: SyncHandlerDeps): (payload: SyncPayload, jc?: JobContext) => Promise { 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") { + if (!plugin) return failBeforeRun(db, full.id, payload.file ? "file" : (payload.mode ?? "full"), `Plugin '${full.plugin_id}' is not installed or failed to load.`); + let mode: SyncMode = "full"; + if (payload.file) { + mode = { file: { name: payload.file.name, bytes: new Uint8Array(Buffer.from(payload.file.base64, "base64")) } }; + } else if (plugin.pullChanges && payload.mode !== "full") { const cutoff = new Date(Date.now() - WEEK_MS).toISOString(); const recentFull = await db.k .selectFrom("source_runs") @@ -270,7 +377,7 @@ export function syncHandler(deps: SyncHandlerDeps): (payload: { connectionId: st try { creds = await readCreds(deps.secret, conn.credentialsEnc); } catch (e) { - if (e instanceof CryptoError) return failBeforeRun(db, full.id, mode, MSG_CRYPTO); + if (e instanceof CryptoError) return failBeforeRun(db, full.id, typeof mode === "string" ? mode : "file", MSG_CRYPTO); throw e; } const logs: string[] = []; diff --git a/packages/core/src/entries.ts b/packages/core/src/entries.ts index 94dbcef..142f87c 100644 --- a/packages/core/src/entries.ts +++ b/packages/core/src/entries.ts @@ -975,3 +975,34 @@ export async function rejectSuggestion(db: Db, userId: string, suggestionId: str if (s.status !== "pending") throw new ValidationError("status", "not_pending"); await db.k.updateTable("match_suggestions").set({ status: "rejected" }).where("id", "=", s.id).execute(); } + +export interface MatchSuggestion { + id: string; + workId: string; + candidateWorkId: string; + score: number; + status: "pending" | "accepted" | "rejected"; + createdAt: string; +} + +/** The user's pending match suggestions for a work, best first. Throws `NotFoundError` if the work is not visible. */ +export async function listSuggestions(db: Db, userId: string, workId: string): Promise { + await resolveVisibleWork(db.k, userId, workId); + const rows = await db.k + .selectFrom("match_suggestions") + .selectAll() + .where("user_id", "=", userId) + .where("work_id", "=", workId) + .where("status", "=", "pending") + .orderBy("score", "desc") + .orderBy("id") + .execute(); + return rows.map((r) => ({ + id: r.id, + workId: r.work_id, + candidateWorkId: r.candidate_work_id, + score: r.score, + status: r.status, + createdAt: r.created_at, + })); +} diff --git a/packages/core/test/api-support.test.ts b/packages/core/test/api-support.test.ts new file mode 100644 index 0000000..552134c --- /dev/null +++ b/packages/core/test/api-support.test.ts @@ -0,0 +1,140 @@ +import { expect, spyOn, test } from "bun:test"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { newId, nowIso } from "@sttracker/db/ids"; +import { forEachDialect } from "@sttracker/db/testing"; +import { definePlugin, type SourcePlugin } from "@sttracker/plugin-sdk/types"; +import { listAttachmentLinks, readAttachment, requireLinkTarget, storeAttachment } from "../src/attachments.ts"; +import { createAppToken, listAppTokens, revokeAppToken } from "../src/auth.ts"; +import { createConnection, deleteConnection, devicePoll, deviceStart, getConnection, listRuns, syncHandler } from "../src/connections.ts"; +import { listSuggestions, quickAdd } from "../src/entries.ts"; +import { addEngagement } from "../src/engagements.ts"; +import { NotFoundError, ValidationError } from "../src/errors.ts"; +import type { PluginRegistry } from "../src/registry.ts"; +import { loadPlugins } from "../src/registry.ts"; +import { makeUser, makeWork } from "./user-layer-fixtures.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")!; + +const fileish: SourcePlugin = definePlugin({ + manifest: { id: "fileish", name: "Fileish", version: "1", sdkVersion: 1, mediaTypes: ["book"], capabilities: ["importFile"], config: [] }, + async *importFile(_ctx, file) { + yield { + kind: "entry", + key: file.name, + raw: new TextDecoder().decode(file.bytes), + work: { title: new TextDecoder().decode(file.bytes), mediaType: "book", externalIds: { isbn13: "9780000000009" } }, + entry: { status: "planned" }, + }; + }, + deviceAuth: { + start: async () => ({ userCode: "ABCD", verificationUrl: "https://x.test/device", deviceCode: "dc", intervalSec: 5, expiresInSec: 600 }), + poll: async (_ctx, deviceCode) => (deviceCode === "dc" ? { access_token: "tok" } : "pending"), + }, +}); +const reg2: PluginRegistry = { get: (id) => (id === "fileish" ? fileish : reg.get(id)), list: () => [...reg.list(), fileish.manifest], reload: async () => {} }; + +forEachDialect("api support", (getDb) => { + test("listAppTokens returns metadata only, revoked flagged, scoped to user", async () => { + const db = getDb(); + const u = await makeUser(db); + const other = await makeUser(db, "bob"); + const t = await createAppToken(db, u, "script", ["read"]); + await createAppToken(db, other, "x", ["write"]); + await revokeAppToken(db, u, t.id); + const list = await listAppTokens(db, u); + expect(list).toHaveLength(1); + expect(list[0]).toMatchObject({ id: t.id, name: "script", scopes: ["read"], lastUsedAt: null }); + expect(list[0]!.revokedAt).not.toBeNull(); + expect(JSON.stringify(list)).not.toContain("hash"); + }); + + test("listSuggestions lists pending suggestions for a visible work", async () => { + const db = getDb(); + const u = await makeUser(db); + const draft = await quickAdd(db, u, { title: "Dune??", mediaType: "book" }); + const good = await makeWork(db, { title: "Dune", externalIds: { isbn13: "1" } }); + await db.k + .insertInto("match_suggestions") + .values({ id: newId(), user_id: u, work_id: draft.workId, candidate_work_id: good, score: 0.75, status: "pending", created_at: nowIso() }) + .execute(); + const s = await listSuggestions(db, u, draft.workId); + expect(s).toHaveLength(1); + expect(s[0]).toMatchObject({ workId: draft.workId, candidateWorkId: good, score: 0.75, status: "pending" }); + const other = await makeUser(db, "bob"); + await expect(listSuggestions(db, other, draft.workId)).rejects.toBeInstanceOf(NotFoundError); + }); + + test("requireLinkTarget checks ownership per entity; links and filename are readable", async () => { + const db = getDb(); + const dir = await mkdtemp(join(tmpdir(), "stt-att-")); + try { + const u = await makeUser(db); + const bob = await makeUser(db, "bob"); + const e = await quickAdd(db, u, { title: "Dune" }); + const g = await addEngagement(db, u, e.id, {}, "user"); + await requireLinkTarget(db, u, "entry", e.id); + await requireLinkTarget(db, u, "engagement", g.id); + await requireLinkTarget(db, u, "work", e.workId); + await expect(requireLinkTarget(db, bob, "entry", e.id)).rejects.toBeInstanceOf(NotFoundError); + await expect(requireLinkTarget(db, bob, "engagement", g.id)).rejects.toBeInstanceOf(NotFoundError); + await expect(requireLinkTarget(db, bob, "work", e.workId)).rejects.toBeInstanceOf(NotFoundError); + await expect(requireLinkTarget(db, u, "nope" as never, e.id)).rejects.toBeInstanceOf(ValidationError); + const a = await storeAttachment(db, dir, u, { bytes: new Uint8Array([1]), mime: "image/png", filename: "a.png" }, { entity: "entry", entityId: e.id }); + const links = await listAttachmentLinks(db, u, a.id); + expect(links).toEqual([{ id: expect.any(String), entity: "entry", entityId: e.id }]); + expect(await listAttachmentLinks(db, bob, a.id)).toEqual([]); + expect((await readAttachment(db, dir, u, a.id))?.filename).toBe("a.png"); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + test("getConnection hides other users and deleted connections; listRuns is newest first", async () => { + const db = getDb(); + const u = await makeUser(db); + const bob = await makeUser(db, "bob"); + const id = await createConnection(db, SECRET, u, { pluginId: "echo", name: "E", config: {} }, echo.manifest); + expect((await getConnection(db, u, id)).pluginId).toBe("echo"); + await expect(getConnection(db, bob, id)).rejects.toBeInstanceOf(NotFoundError); + const sync = syncHandler({ db, secret: SECRET, registry: reg }); + await sync({ connectionId: id, mode: "full" }); + await sync({ connectionId: id, mode: "full" }); + const runs = await listRuns(db, u, id); + expect(runs).toHaveLength(2); + expect(runs[0]!.startedAt >= runs[1]!.startedAt).toBe(true); + expect(runs[0]).toMatchObject({ mode: "full", status: "ok" }); + await expect(listRuns(db, bob, id)).rejects.toBeInstanceOf(NotFoundError); + await deleteConnection(db, u, id); + await expect(getConnection(db, u, id)).rejects.toBeInstanceOf(NotFoundError); + }); + + test("syncHandler runs file mode from a base64 payload", async () => { + const db = getDb(); + const u = await makeUser(db); + const id = await createConnection(db, SECRET, u, { pluginId: "fileish", name: "F", config: {} }, fileish.manifest); + const sync = syncHandler({ db, secret: SECRET, registry: reg2 }); + const s = await sync({ connectionId: id, file: { name: "x.txt", base64: Buffer.from("Dune").toString("base64") } }); + expect(s).toMatchObject({ status: "ok" }); + expect((await listRuns(db, u, id))[0]!.mode).toBe("file"); + const entries = await db.k.selectFrom("entries").select("id").where("user_id", "=", u).execute(); + expect(entries).toHaveLength(1); + }); + + test("deviceStart and devicePoll store credentials on success", async () => { + const db = getDb(); + const u = await makeUser(db); + const id = await createConnection(db, SECRET, u, { pluginId: "fileish", name: "F", config: {} }, fileish.manifest); + expect(await deviceStart(db, SECRET, reg2, u, id)).toMatchObject({ userCode: "ABCD", deviceCode: "dc" }); + expect(await devicePoll(db, SECRET, reg2, u, id, "other")).toBe("pending"); + expect(await devicePoll(db, SECRET, reg2, u, id, "dc")).toBe("connected"); + expect(await getConnection(db, u, id)).toMatchObject({ hasCredentials: true }); + const e = await createConnection(db, SECRET, u, { pluginId: "echo", name: "E", config: {} }, echo.manifest); + await expect(deviceStart(db, SECRET, reg2, u, e)).rejects.toBeInstanceOf(ValidationError); + }); +});