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:
1 parent
0a540208e1
commit
e3a759d768
8 files changed
+81
-16
No files matched your search
@@ -4,7 +4,6 @@ import {
|
|||||||
devicePoll,
|
devicePoll,
|
||||||
deviceStart,
|
deviceStart,
|
||||||
getConnection,
|
getConnection,
|
||||||
isDeletedConnection,
|
|
||||||
listConnections,
|
listConnections,
|
||||||
listRuns,
|
listRuns,
|
||||||
updateConnection,
|
updateConnection,
|
||||||
@@ -48,7 +47,7 @@ export const sourceRoutes = [
|
|||||||
path: "sources",
|
path: "sources",
|
||||||
summary: "Your source connections (never includes secrets).",
|
summary: "Your source connections (never includes secrets).",
|
||||||
auth: "any:read",
|
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({
|
route({
|
||||||
method: "POST",
|
method: "POST",
|
||||||
|
|||||||
@@ -77,6 +77,7 @@ export async function createConnection(db: Db, secret: string, userId: string, i
|
|||||||
status: "active",
|
status: "active",
|
||||||
last_run_at: null,
|
last_run_at: null,
|
||||||
created_at: nowIso(),
|
created_at: nowIso(),
|
||||||
|
deleted_at: null,
|
||||||
})
|
})
|
||||||
.execute();
|
.execute();
|
||||||
return id;
|
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> {
|
export async function updateConnection(db: Db, secret: string, userId: string, id: string, patch: ConnectionPatch, manifest: PluginManifest): Promise<void> {
|
||||||
checkSchedule(patch.scheduleMinutes);
|
checkSchedule(patch.scheduleMinutes);
|
||||||
await db.k.transaction().execute(async (tx) => {
|
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();
|
if (db.dialect === "postgres") q = q.forUpdate();
|
||||||
const row = await q.executeTakeFirst();
|
const row = await q.executeTakeFirst();
|
||||||
if (!row) throw new NotFoundError("connection");
|
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> {
|
export async function deleteConnection(db: Db, userId: string, id: string): Promise<void> {
|
||||||
const r = await db.k
|
const r = await db.k
|
||||||
.updateTable("source_connections")
|
.updateTable("source_connections")
|
||||||
.set({ status: "paused", credentials_enc: null })
|
.set({ status: "paused", credentials_enc: null, deleted_at: nowIso() })
|
||||||
.where("id", "=", id)
|
.where("id", "=", id)
|
||||||
.where("user_id", "=", userId)
|
.where("user_id", "=", userId)
|
||||||
|
.where("deleted_at", "is", null)
|
||||||
.executeTakeFirst();
|
.executeTakeFirst();
|
||||||
if (r.numUpdatedRows !== 1n) throw new NotFoundError("connection");
|
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[]> {
|
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) => ({
|
return rows.map((r) => ({
|
||||||
id: r.id,
|
id: r.id,
|
||||||
pluginId: r.plugin_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) {
|
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();
|
const r = await db.k
|
||||||
if (!r || (r.status === "paused" && r.credentials_enc === null)) throw new NotFoundError("connection");
|
.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;
|
return r;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -342,6 +351,7 @@ export function syncHandler(deps: SyncHandlerDeps): (payload: SyncPayload, jc?:
|
|||||||
const { db } = deps;
|
const { db } = deps;
|
||||||
const full = await db.k.selectFrom("source_connections").selectAll().where("id", "=", payload.connectionId).executeTakeFirst();
|
const full = await db.k.selectFrom("source_connections").selectAll().where("id", "=", payload.connectionId).executeTakeFirst();
|
||||||
if (!full) throw new DropJobError("connection not found");
|
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");
|
if (full.status === "paused") throw new DropJobError("connection is paused");
|
||||||
const plugin = deps.registry.get(full.plugin_id);
|
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.`);
|
if (!plugin) return failBeforeRun(db, full.id, payload.file ? "file" : (payload.mode ?? "full"), `Plugin '${full.plugin_id}' is not installed or failed to load.`);
|
||||||
|
|||||||
@@ -198,6 +198,7 @@ export function startScheduler(db: Db, opts: { tickMs?: number } = {}): { stop()
|
|||||||
.selectFrom("source_connections")
|
.selectFrom("source_connections")
|
||||||
.select(["id", "schedule_minutes", "last_run_at"])
|
.select(["id", "schedule_minutes", "last_run_at"])
|
||||||
.where("status", "!=", "paused")
|
.where("status", "!=", "paused")
|
||||||
|
.where("deleted_at", "is", null)
|
||||||
.where("schedule_minutes", "is not", null)
|
.where("schedule_minutes", "is not", null)
|
||||||
.execute();
|
.execute();
|
||||||
const now = Date.now();
|
const now = Date.now();
|
||||||
|
|||||||
@@ -7,7 +7,9 @@ import { forEachDialect } from "@sttracker/db/testing";
|
|||||||
import { definePlugin, type SourcePlugin } from "@sttracker/plugin-sdk/types";
|
import { definePlugin, type SourcePlugin } from "@sttracker/plugin-sdk/types";
|
||||||
import { listAttachmentLinks, readAttachment, requireLinkTarget, storeAttachment } from "../src/attachments.ts";
|
import { listAttachmentLinks, readAttachment, requireLinkTarget, storeAttachment } from "../src/attachments.ts";
|
||||||
import { createAppToken, listAppTokens, revokeAppToken } from "../src/auth.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 { listSuggestions, quickAdd } from "../src/entries.ts";
|
||||||
import { addEngagement } from "../src/engagements.ts";
|
import { addEngagement } from "../src/engagements.ts";
|
||||||
import { NotFoundError, ValidationError } from "../src/errors.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);
|
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 () => {
|
test("syncHandler runs file mode from a base64 payload", async () => {
|
||||||
const db = getDb();
|
const db = getDb();
|
||||||
const u = await makeUser(db);
|
const u = await makeUser(db);
|
||||||
|
|||||||
@@ -55,7 +55,8 @@ forEachDialect("connections", (getDb) => {
|
|||||||
r = await raw(id);
|
r = await raw(id);
|
||||||
expect(r.status).toBe("paused");
|
expect(r.status).toBe("paused");
|
||||||
expect(r.credentials_enc).toBeNull();
|
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 () => {
|
test("cursor persists on set; http uses manifest rate limit", async () => {
|
||||||
|
|||||||
@@ -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<any>) {
|
||||||
|
await db.schema.alterTable("source_connections").addColumn("deleted_at", "text").execute();
|
||||||
|
},
|
||||||
|
|
||||||
|
async down(db: Kysely<any>) {
|
||||||
|
await db.schema.alterTable("source_connections").dropColumn("deleted_at").execute();
|
||||||
|
},
|
||||||
|
};
|
||||||
@@ -1,9 +1,11 @@
|
|||||||
import type { Migration } from "kysely/migration";
|
import type { Migration } from "kysely/migration";
|
||||||
import { initial } from "./0001_initial.ts";
|
import { initial } from "./0001_initial.ts";
|
||||||
import { worksPositionUnique } from "./0002_works_position_unique.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. */
|
/** All migrations, keyed by sortable name. Imported statically so bundles need no filesystem. */
|
||||||
export const migrations: Record<string, Migration> = {
|
export const migrations: Record<string, Migration> = {
|
||||||
"0001_initial": initial,
|
"0001_initial": initial,
|
||||||
"0002_works_position_unique": worksPositionUnique,
|
"0002_works_position_unique": worksPositionUnique,
|
||||||
|
"0003_source_connections_deleted_at": sourceConnectionsDeletedAt,
|
||||||
};
|
};
|
||||||
@@ -269,6 +269,8 @@ export interface SourceConnectionsTable {
|
|||||||
status: ConnectionStatus;
|
status: ConnectionStatus;
|
||||||
last_run_at: string | null;
|
last_run_at: string | null;
|
||||||
created_at: string;
|
created_at: string;
|
||||||
|
/** Set by `deleteConnection`; deleted connections are hidden, never scheduled and never synced. */
|
||||||
|
deleted_at: string | null;
|
||||||
}
|
}
|
||||||
|
|
||||||
export interface SourceRunsTable {
|
export interface SourceRunsTable {
|
||||||
|
|||||||
Reference in new issue
Block a user