diff --git a/apps/server/src/routes/sources.ts b/apps/server/src/routes/sources.ts index b8a27fb..8cb6061 100644 --- a/apps/server/src/routes/sources.ts +++ b/apps/server/src/routes/sources.ts @@ -4,7 +4,6 @@ import { devicePoll, deviceStart, getConnection, - isDeletedConnection, listConnections, listRuns, updateConnection, @@ -48,7 +47,7 @@ export const sourceRoutes = [ path: "sources", summary: "Your source connections (never includes secrets).", auth: "any:read", - handler: async ({ deps, user }) => (await listConnections(deps.db, user.id)).filter((c) => !isDeletedConnection(c)), + handler: async ({ deps, user }) => listConnections(deps.db, user.id), }), route({ method: "POST", diff --git a/packages/core/src/connections.ts b/packages/core/src/connections.ts index bb12cc8..bab549c 100644 --- a/packages/core/src/connections.ts +++ b/packages/core/src/connections.ts @@ -77,6 +77,7 @@ export async function createConnection(db: Db, secret: string, userId: string, i status: "active", last_run_at: null, created_at: nowIso(), + deleted_at: null, }) .execute(); return id; @@ -94,7 +95,7 @@ export interface ConnectionPatch { export async function updateConnection(db: Db, secret: string, userId: string, id: string, patch: ConnectionPatch, manifest: PluginManifest): Promise { checkSchedule(patch.scheduleMinutes); await db.k.transaction().execute(async (tx) => { - let q = tx.selectFrom("source_connections").selectAll().where("id", "=", id).where("user_id", "=", userId); + let q = tx.selectFrom("source_connections").selectAll().where("id", "=", id).where("user_id", "=", userId).where("deleted_at", "is", null); if (db.dialect === "postgres") q = q.forUpdate(); const row = await q.executeTakeFirst(); if (!row) throw new NotFoundError("connection"); @@ -116,20 +117,27 @@ export async function updateConnection(db: Db, secret: string, userId: string, i }); } -/** Soft delete: the connection is paused and its credentials are wiped. Records and runs stay. */ +/** Soft delete: `deleted_at` is set, the connection is paused and its credentials are wiped. Records and runs stay. */ export async function deleteConnection(db: Db, userId: string, id: string): Promise { const r = await db.k .updateTable("source_connections") - .set({ status: "paused", credentials_enc: null }) + .set({ status: "paused", credentials_enc: null, deleted_at: nowIso() }) .where("id", "=", id) .where("user_id", "=", userId) + .where("deleted_at", "is", null) .executeTakeFirst(); if (r.numUpdatedRows !== 1n) throw new NotFoundError("connection"); } -/** Never returns secrets: plain config plus `hasCredentials`. */ +/** The user's live (not deleted) connections. Never returns secrets: plain config plus `hasCredentials`. */ export async function listConnections(db: Db, userId: string): Promise { - const rows = await db.k.selectFrom("source_connections").selectAll().where("user_id", "=", userId).orderBy("created_at").execute(); + const rows = await db.k + .selectFrom("source_connections") + .selectAll() + .where("user_id", "=", userId) + .where("deleted_at", "is", null) + .orderBy("created_at") + .execute(); return rows.map((r) => ({ id: r.id, pluginId: r.plugin_id, @@ -143,14 +151,15 @@ 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"); + const r = await db.k + .selectFrom("source_connections") + .selectAll() + .where("id", "=", id) + .where("user_id", "=", userId) + .where("deleted_at", "is", null) + .executeTakeFirst(); + if (!r) throw new NotFoundError("connection"); return r; } @@ -342,6 +351,7 @@ export function syncHandler(deps: SyncHandlerDeps): (payload: SyncPayload, 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.deleted_at !== null) throw new DropJobError("connection is deleted"); 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.file ? "file" : (payload.mode ?? "full"), `Plugin '${full.plugin_id}' is not installed or failed to load.`); diff --git a/packages/core/src/jobs.ts b/packages/core/src/jobs.ts index 4620857..155ce2c 100644 --- a/packages/core/src/jobs.ts +++ b/packages/core/src/jobs.ts @@ -198,6 +198,7 @@ export function startScheduler(db: Db, opts: { tickMs?: number } = {}): { stop() .selectFrom("source_connections") .select(["id", "schedule_minutes", "last_run_at"]) .where("status", "!=", "paused") + .where("deleted_at", "is", null) .where("schedule_minutes", "is not", null) .execute(); const now = Date.now(); diff --git a/packages/core/test/api-support.test.ts b/packages/core/test/api-support.test.ts index 552134c..b5a26c0 100644 --- a/packages/core/test/api-support.test.ts +++ b/packages/core/test/api-support.test.ts @@ -7,7 +7,9 @@ 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 { createConnection, deleteConnection, devicePoll, deviceStart, getConnection, listConnections, listRuns, syncHandler, updateConnection } from "../src/connections.ts"; +import { startScheduler } from "../src/jobs.ts"; +import { DropJobError } from "../src/jobs.ts"; import { listSuggestions, quickAdd } from "../src/entries.ts"; import { addEngagement } from "../src/engagements.ts"; import { NotFoundError, ValidationError } from "../src/errors.ts"; @@ -114,6 +116,41 @@ forEachDialect("api support", (getDb) => { await expect(getConnection(db, u, id)).rejects.toBeInstanceOf(NotFoundError); }); + test("a paused connection without credentials stays listed and can be unpaused; a deleted one is gone", async () => { + const db = getDb(); + const u = await makeUser(db); + const id = await createConnection(db, SECRET, u, { pluginId: "echo", name: "E", config: {} }, echo.manifest); + await updateConnection(db, SECRET, u, id, { status: "paused" }, echo.manifest); + expect((await listConnections(db, u)).map((c) => c.id)).toEqual([id]); + expect((await getConnection(db, u, id)).status).toBe("paused"); + await updateConnection(db, SECRET, u, id, { status: "active" }, echo.manifest); + expect((await getConnection(db, u, id)).status).toBe("active"); + + await deleteConnection(db, u, id); + const row = await db.k.selectFrom("source_connections").selectAll().where("id", "=", id).executeTakeFirstOrThrow(); + expect(row).toMatchObject({ status: "paused", credentials_enc: null }); + expect(row.deleted_at).not.toBeNull(); + expect(await listConnections(db, u)).toEqual([]); + await expect(getConnection(db, u, id)).rejects.toBeInstanceOf(NotFoundError); + await expect(updateConnection(db, SECRET, u, id, { status: "active" }, echo.manifest)).rejects.toBeInstanceOf(NotFoundError); + await expect(deleteConnection(db, u, id)).rejects.toBeInstanceOf(NotFoundError); + }); + + test("deleted connections are never scheduled and their queued syncs are dropped", async () => { + const db = getDb(); + const u = await makeUser(db); + const id = await createConnection(db, SECRET, u, { pluginId: "echo", name: "E", config: {}, scheduleMinutes: 1 }, echo.manifest); + // Simulate a delete that left status "error" (e.g. a row edited by hand): deleted_at alone must exclude it. + await deleteConnection(db, u, id); + await db.k.updateTable("source_connections").set({ status: "error" }).where("id", "=", id).execute(); + const sched = startScheduler(db, { tickMs: 3_600_000 }); + await sched.tickNow(); + sched.stop(); + expect(await db.k.selectFrom("jobs").select("id").execute()).toEqual([]); + const sync = syncHandler({ db, secret: SECRET, registry: reg }); + await expect(sync({ connectionId: id })).rejects.toBeInstanceOf(DropJobError); + }); + test("syncHandler runs file mode from a base64 payload", async () => { const db = getDb(); const u = await makeUser(db); diff --git a/packages/core/test/connections.test.ts b/packages/core/test/connections.test.ts index 1859c9d..cd8dfaf 100644 --- a/packages/core/test/connections.test.ts +++ b/packages/core/test/connections.test.ts @@ -55,7 +55,8 @@ forEachDialect("connections", (getDb) => { r = await raw(id); expect(r.status).toBe("paused"); expect(r.credentials_enc).toBeNull(); - expect((await listConnections(db, u))[0]!.hasCredentials).toBe(false); + expect(r.deleted_at).not.toBeNull(); + expect(await listConnections(db, u)).toEqual([]); }); test("cursor persists on set; http uses manifest rate limit", async () => { diff --git a/packages/db/src/migrations/0003_source_connections_deleted_at.ts b/packages/db/src/migrations/0003_source_connections_deleted_at.ts new file mode 100644 index 0000000..9b4763c --- /dev/null +++ b/packages/db/src/migrations/0003_source_connections_deleted_at.ts @@ -0,0 +1,13 @@ +import type { Kysely } from "kysely"; +import type { Migration } from "kysely/migration"; + +/** Soft delete for source connections: a deleted connection is distinct from one the user merely paused. */ +export const sourceConnectionsDeletedAt: Migration = { + async up(db: Kysely) { + await db.schema.alterTable("source_connections").addColumn("deleted_at", "text").execute(); + }, + + async down(db: Kysely) { + await db.schema.alterTable("source_connections").dropColumn("deleted_at").execute(); + }, +}; diff --git a/packages/db/src/migrations/index.ts b/packages/db/src/migrations/index.ts index 4dc1ca0..fd6adc3 100644 --- a/packages/db/src/migrations/index.ts +++ b/packages/db/src/migrations/index.ts @@ -1,9 +1,11 @@ import type { Migration } from "kysely/migration"; import { initial } from "./0001_initial.ts"; import { worksPositionUnique } from "./0002_works_position_unique.ts"; +import { sourceConnectionsDeletedAt } from "./0003_source_connections_deleted_at.ts"; /** All migrations, keyed by sortable name. Imported statically so bundles need no filesystem. */ export const migrations: Record = { "0001_initial": initial, "0002_works_position_unique": worksPositionUnique, + "0003_source_connections_deleted_at": sourceConnectionsDeletedAt, }; diff --git a/packages/db/src/schema.ts b/packages/db/src/schema.ts index 81c27bf..0d85c9d 100644 --- a/packages/db/src/schema.ts +++ b/packages/db/src/schema.ts @@ -269,6 +269,8 @@ export interface SourceConnectionsTable { status: ConnectionStatus; last_run_at: string | null; created_at: string; + /** Set by `deleteConnection`; deleted connections are hidden, never scheduled and never synced. */ + deleted_at: string | null; } export interface SourceRunsTable {