fix(core): soft-delete source connections with deleted_at

Migration 0003 adds source_connections.deleted_at. deleteConnection sets it; listing, lookups,
updates, the scheduler and the sync handler ignore deleted connections, so a paused
credential-less connection stays visible and can be resumed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Lucas WintherandClaude Opus 5.5 committed 2026-10-30 21:25:59 +01:00
1 parent 0a540208e1
commit e3a759d768
8 files changed
+81 -16

No files matched your search

+22 -12
View File
@@ -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<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);
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<void> {
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<ConnectionView[]> {
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<Connectio
}));
}
/** A deleted connection is paused with its credentials wiped (`deleteConnection`). */
export function isDeletedConnection(c: Pick<ConnectionView, "status" | "hasCredentials">): 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.`);