diff --git a/cli/src/__tests__/worktree.test.ts b/cli/src/__tests__/worktree.test.ts index 0fb6b2aee32..fc6227aa1ed 100644 --- a/cli/src/__tests__/worktree.test.ts +++ b/cli/src/__tests__/worktree.test.ts @@ -1673,7 +1673,19 @@ describe("worktree helpers", () => { } const sourceDbClient = createDb(sourceDb.connectionString); await migrate(drizzle(sourceDbClient.$client), { migrationsFolder: priorMigrations }); - await seedValidWorktreeSource(sourceDb.connectionString); + const seed = await seedValidWorktreeSource(sourceDb.connectionString); + // An older filtered JavaScript backup retained event IDs but lost the + // identity generator. The pending migration must repair that schema. + const legacyAgentId = randomUUID(); + await sourceDbClient.$client` + INSERT INTO agents (id, company_id, name, status) + VALUES (${legacyAgentId}, ${seed.companyId}, 'Legacy paused agent', 'paused') + `; + await sourceDbClient.$client` + INSERT INTO resource_lifecycle_events (company_id, resource_type, resource_id, action) + VALUES (${seed.companyId}, 'agent', ${legacyAgentId}, 'pause') + `; + await sourceDbClient.$client.unsafe('ALTER TABLE resource_lifecycle_events ALTER COLUMN id DROP IDENTITY'); await sourceDbClient.$client.unsafe(` WITH pair AS ( SELECT @@ -1786,6 +1798,16 @@ describe("worktree helpers", () => { ); const seededUsers = await targetDb.select().from(authUsers); expect(seededUsers.some((row) => row.email === "existing@paperclip.ing")).toBe(true); + const restoredEvents = await targetDb.$client` + SELECT id, action FROM resource_lifecycle_events WHERE resource_id = ${legacyAgentId} ORDER BY id + `; + expect(restoredEvents.map(row => row.action)).toEqual(["pause", "create"]); + expect(Number(restoredEvents[1].id)).toBeGreaterThan(Number(restoredEvents[0].id)); + const [identity] = await targetDb.$client` + SELECT is_identity FROM information_schema.columns + WHERE table_schema = 'public' AND table_name = 'resource_lifecycle_events' AND column_name = 'id' + `; + expect(identity.is_identity).toBe("YES"); }, ); diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 4e8bb3a3a57..8f4d17f2f06 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -529,6 +529,9 @@ all projects, including archived projects. Pending hires stay behind approval. Paused and terminated agents receive missing final status intents. A partial journal ending at pause receives resume when the current agent is running. Existing records remain intact, and rerunning the baseline does not duplicate it. +The migration also repairs the journal ID generator in older JavaScript restores +that lost identity metadata, starting above existing IDs. New JavaScript backups +preserve identity generation, sequence options, and sequence progress. Resource writes wait for the migration transaction to commit. These records represent current desired state, not reconstructed historical transitions. There is no later or runtime journal backfill. Plugins use `ctx.events.listLifecycle(companyId, limit?, afterId?)` and diff --git a/packages/db/src/backup-lib.test.ts b/packages/db/src/backup-lib.test.ts index 13c5edc9d8a..b6851534991 100644 --- a/packages/db/src/backup-lib.test.ts +++ b/packages/db/src/backup-lib.test.ts @@ -75,6 +75,34 @@ describe("createBufferedTextFileWriter", () => { }); describeEmbeddedPostgres("runDatabaseBackup", () => { + it("preserves identity generation, sequence options and progress in JavaScript backups", async () => { + const source = await createTempDatabase(); + const target = await createSiblingDatabase(source, "identity_restore_target"); + const sourceSql = postgres(source, { max: 1, onnotice: () => {} }); + const targetSql = postgres(target, { max: 1, onnotice: () => {} }); + try { + await sourceSql.unsafe(` + CREATE SCHEMA identity_fixture; + CREATE TABLE identity_fixture.records ( + id bigint GENERATED ALWAYS AS IDENTITY (START WITH 7 INCREMENT BY 2) PRIMARY KEY, + default_id integer GENERATED BY DEFAULT AS IDENTITY, + value text NOT NULL + ); + INSERT INTO identity_fixture.records (value) VALUES ('one'), ('two'); + `); + const backup = await runDatabaseBackup({ connectionString: source, backupDir: createTempDir("identity-backup-"), backupEngine: "javascript", retention: { dailyDays: 7, weeklyWeeks: 4, monthlyMonths: 1 } }); + await runDatabaseRestore({ connectionString: target, backupFile: backup.backupFile }); + expect(await targetSql`SELECT id::text, default_id, value FROM identity_fixture.records ORDER BY id`) + .toEqual([{ id: "7", default_id: 1, value: "one" }, { id: "9", default_id: 2, value: "two" }]); + expect(await targetSql`INSERT INTO identity_fixture.records (value) VALUES ('three') RETURNING id::text, default_id`) + .toEqual([{ id: "11", default_id: 3 }]); + await expect(targetSql`INSERT INTO identity_fixture.records (id, value) VALUES (99, 'explicit')`).rejects.toThrow("non-DEFAULT"); + } finally { + await sourceSql.end(); + await targetSql.end(); + } + }, 30_000); + it( "keeps the newest backup for each retained calendar month", async () => { diff --git a/packages/db/src/backup-lib.ts b/packages/db/src/backup-lib.ts index f066f22e235..a95ac3f1be8 100644 --- a/packages/db/src/backup-lib.ts +++ b/packages/db/src/backup-lib.ts @@ -43,6 +43,7 @@ export type RunDatabaseRestoreOptions = { }; type SequenceDefinition = { + is_identity: boolean; sequence_schema: string; sequence_name: string; data_type: string; @@ -621,6 +622,7 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise const allSequences = await sql` SELECT + false AS is_identity, s.sequence_schema, s.sequence_name, s.data_type, @@ -642,7 +644,26 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise WHERE ${sql.unsafe(nonSystemSchemaPredicate("s.sequence_schema"))} ORDER BY s.sequence_schema, s.sequence_name `; - const sequences = allSequences.filter( + const identitySequences = await sql` + SELECT true AS is_identity, n.nspname AS sequence_schema, seq.relname AS sequence_name, + format_type(options.seqtypid, NULL) AS data_type, + options.seqstart::text AS start_value, options.seqmin::text AS minimum_value, + options.seqmax::text AS maximum_value, options.seqincrement::text AS increment, + CASE WHEN options.seqcycle THEN 'YES' ELSE 'NO' END AS cycle_option, + owner_namespace.nspname AS owner_schema, owner.relname AS owner_table, + attribute.attname AS owner_column + FROM pg_sequence options + JOIN pg_class seq ON seq.oid = options.seqrelid + JOIN pg_namespace n ON n.oid = seq.relnamespace + JOIN pg_depend dependency ON dependency.objid = seq.oid AND dependency.deptype = 'i' + AND dependency.classid = 'pg_class'::regclass AND dependency.refclassid = 'pg_class'::regclass + JOIN pg_class owner ON owner.oid = dependency.refobjid + JOIN pg_namespace owner_namespace ON owner_namespace.oid = owner.relnamespace + JOIN pg_attribute attribute ON attribute.attrelid = owner.oid AND attribute.attnum = dependency.refobjsubid + WHERE ${sql.unsafe(nonSystemSchemaPredicate("n.nspname"))} + ORDER BY n.nspname, seq.relname + `; + const sequences = [...allSequences, ...identitySequences].filter( (seq) => !seq.owner_table || includedTableNames.has(tableKey(seq.owner_schema ?? "public", seq.owner_table)), ); @@ -684,7 +705,7 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise if (sequences.length > 0) { emit("-- Sequences"); - for (const seq of sequences) { + for (const seq of sequences.filter(seq => !seq.is_identity)) { const qualifiedSequenceName = quoteQualifiedName(seq.sequence_schema, seq.sequence_name); emitStatement(`DROP SEQUENCE IF EXISTS ${qualifiedSequenceName} CASCADE;`); emitStatement( @@ -704,11 +725,13 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise udt_name: string; is_nullable: string; column_default: string | null; + is_identity: string; + identity_generation: "ALWAYS" | "BY DEFAULT" | null; character_maximum_length: number | null; numeric_precision: number | null; numeric_scale: number | null; }[]>` - SELECT column_name, data_type, udt_schema, udt_name, is_nullable, column_default, + SELECT column_name, data_type, udt_schema, udt_name, is_nullable, column_default, is_identity, identity_generation, character_maximum_length, numeric_precision, numeric_scale FROM information_schema.columns WHERE table_schema = ${schema_name} AND table_name = ${tablename} @@ -742,7 +765,12 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise } let def = ` "${col.column_name}" ${typeStr}`; - if (col.column_default != null) def += ` DEFAULT ${col.column_default}`; + if (col.is_identity === "YES") { + const sequence = sequences.find(seq => seq.is_identity && seq.owner_schema === schema_name + && seq.owner_table === tablename && seq.owner_column === col.column_name); + if (!sequence || !col.identity_generation) throw new Error(`Missing identity sequence for ${schema_name}.${tablename}.${col.column_name}`); + def += ` GENERATED ${col.identity_generation} AS IDENTITY (SEQUENCE NAME ${quoteQualifiedName(sequence.sequence_schema, sequence.sequence_name)} START WITH ${sequence.start_value} INCREMENT BY ${sequence.increment} MINVALUE ${sequence.minimum_value} MAXVALUE ${sequence.maximum_value} ${sequence.cycle_option === "YES" ? "CYCLE" : "NO CYCLE"})`; + } else if (col.column_default != null) def += ` DEFAULT ${col.column_default}`; if (col.is_nullable === "NO") def += " NOT NULL"; colDefs.push(def); } @@ -770,7 +798,7 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise emit(""); } - const ownedSequences = sequences.filter((seq) => seq.owner_table && seq.owner_column); + const ownedSequences = sequences.filter((seq) => !seq.is_identity && seq.owner_table && seq.owner_column); if (ownedSequences.length > 0) { emit("-- Sequence ownership"); for (const seq of ownedSequences) { @@ -919,6 +947,7 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise for (const { schema_name, tablename } of tables) { const currentTableKey = tableKey(schema_name, tablename); const qualifiedTableName = quoteQualifiedName(schema_name, tablename); + const overrideIdentity = identitySequences.some(seq => seq.owner_schema === schema_name && seq.owner_table === tablename); const count = await sql.unsafe<{ n: number }[]>(`SELECT count(*)::int AS n FROM ${qualifiedTableName}`); if (excludedTableNames.has(currentTableKey) || (count[0]?.n ?? 0) === 0) continue; @@ -963,7 +992,7 @@ export async function runDatabaseBackup(opts: RunDatabaseBackupOptions): Promise const values = row.map((rawValue, index) => formatSqlValue(rawValue, cols[index]?.column_name, nullifiedColumns, cols[index]?.data_type), ); - emitStatement(`INSERT INTO ${qualifiedTableName} (${colNames}) VALUES (${values.join(", ")});`); + emitStatement(`INSERT INTO ${qualifiedTableName} (${colNames})${overrideIdentity ? " OVERRIDING SYSTEM VALUE" : ""} VALUES (${values.join(", ")});`); } await writer.drain(); } diff --git a/packages/db/src/migrations/0301_whole_venom.sql b/packages/db/src/migrations/0301_whole_venom.sql index 68273746f92..41045b2aab7 100644 --- a/packages/db/src/migrations/0301_whole_venom.sql +++ b/packages/db/src/migrations/0301_whole_venom.sql @@ -11,6 +11,19 @@ ALTER TABLE "plugin_lifecycle_acknowledgments" ADD CONSTRAINT "plugin_lifecycle_ ALTER TABLE "plugin_lifecycle_acknowledgments" ADD CONSTRAINT "plugin_lifecycle_acknowledgments_event_id_resource_lifecycle_events_id_fk" FOREIGN KEY ("event_id") REFERENCES "public"."resource_lifecycle_events"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint CREATE INDEX "plugin_lifecycle_acknowledgments_event_idx" ON "plugin_lifecycle_acknowledgments" USING btree ("event_id");--> statement-breakpoint CREATE INDEX "resource_lifecycle_events_resource_order_idx" ON "resource_lifecycle_events" USING btree ("company_id","resource_type","resource_id","id");--> statement-breakpoint +-- Older JavaScript backups lost identity metadata. Repair before seeding records. +DO $$ +DECLARE next_id bigint; +BEGIN + IF EXISTS ( + SELECT 1 FROM information_schema.columns + WHERE table_schema = 'public' AND table_name = 'resource_lifecycle_events' + AND column_name = 'id' AND is_identity = 'NO' AND column_default IS NULL + ) THEN + SELECT COALESCE(MAX("id"), 0) + 1 INTO next_id FROM "resource_lifecycle_events"; + EXECUTE format('ALTER TABLE "resource_lifecycle_events" ALTER COLUMN "id" ADD GENERATED ALWAYS AS IDENTITY (START WITH %s)', next_id); + END IF; +END $$;--> statement-breakpoint -- Seed a one-time current-state baseline before plugins can consume the journal. INSERT INTO "resource_lifecycle_events" ("company_id", "resource_type", "resource_id", "action") SELECT "company_id", 'agent', "id", 'create' diff --git a/server/src/__tests__/resource-lifecycle-events.test.ts b/server/src/__tests__/resource-lifecycle-events.test.ts index 488a1cd5732..b5dc507f11d 100644 --- a/server/src/__tests__/resource-lifecycle-events.test.ts +++ b/server/src/__tests__/resource-lifecycle-events.test.ts @@ -322,6 +322,7 @@ describePostgres("Resource lifecycle events", () => { it("keeps delivery unavailable when the baseline migration fails", async () => { await db.insert(agents).values({ companyId, name: "Legacy paused", status: "paused" }); + await db.execute(sql`ALTER TABLE resource_lifecycle_events ALTER COLUMN id DROP IDENTITY`); await db.execute(sql`DROP TABLE plugin_lifecycle_acknowledgments`); await db.execute(sql`DROP INDEX resource_lifecycle_events_resource_order_idx`); const applyDeliveryMigration = () => db.transaction(async tx => { @@ -332,12 +333,14 @@ describePostgres("Resource lifecycle events", () => { }); await db.execute(sql`ALTER TABLE resource_lifecycle_events ADD CONSTRAINT fixture_reject_baseline CHECK (action <> 'pause') NOT VALID`); try { - await expect(applyDeliveryMigration()).rejects.toThrow(); + await expect(applyDeliveryMigration()).rejects.toMatchObject({ cause: { constraint_name: "fixture_reject_baseline" } }); expect(await events()).toEqual([]); + expect((await db.execute(sql`SELECT is_identity FROM information_schema.columns WHERE table_schema = 'public' AND table_name = 'resource_lifecycle_events' AND column_name = 'id'`))[0]).toMatchObject({ is_identity: "NO" }); expect((await db.execute(sql`SELECT to_regclass('public.plugin_lifecycle_acknowledgments') AS relation`))[0]).toMatchObject({ relation: null }); } finally { await db.execute(sql`ALTER TABLE resource_lifecycle_events DROP CONSTRAINT fixture_reject_baseline`); await applyDeliveryMigration(); + expect((await db.execute(sql`SELECT is_identity FROM information_schema.columns WHERE table_schema = 'public' AND table_name = 'resource_lifecycle_events' AND column_name = 'id'`))[0]).toMatchObject({ is_identity: "YES" }); } });