merge master into pap-1167-app-ui-bundle

packages/db/src/backup-lib.test.ts

@@ -176,4 +176,49 @@ describeEmbeddedPostgres("runDatabaseBackup", () => {
     },
     60_000,
   );
+
+  it(
+    "restores statements incrementally when backup comments precede the first breakpoint",
+    async () => {
+      const restoreConnectionString = await createTempDatabase();
+      const restoreSql = postgres(restoreConnectionString, { max: 1, onnotice: () => {} });
+      const backupDir = createTempDir("paperclip-db-restore-manual-");
+      const backupFile = path.join(backupDir, "manual.sql");
+
+      try {
+        await fs.promises.writeFile(
+          backupFile,
+          [
+            "-- Paperclip database backup",
+            "-- Created: 2026-04-06T00:00:00.000Z",
+            "",
+            "BEGIN;",
+            "-- paperclip statement breakpoint 69f6f3f1-42fd-46a6-bf17-d1d85f8f3900",
+            "CREATE TABLE public.restore_stream_test (id integer primary key, payload text not null);",
+            "-- paperclip statement breakpoint 69f6f3f1-42fd-46a6-bf17-d1d85f8f3900",
+            "INSERT INTO public.restore_stream_test (id, payload)",
+            "VALUES (1, 'hello');",
+            "-- paperclip statement breakpoint 69f6f3f1-42fd-46a6-bf17-d1d85f8f3900",
+            "COMMIT;",
+            "-- paperclip statement breakpoint 69f6f3f1-42fd-46a6-bf17-d1d85f8f3900",
+          ].join("\n"),
+          "utf8",
+        );
+
+        await runDatabaseRestore({
+          connectionString: restoreConnectionString,
+          backupFile,
+        });
+
+        const rows = await restoreSql.unsafe<{ payload: string }[]>(`
+          SELECT payload
+          FROM public.restore_stream_test
+        `);
+        expect(rows).toEqual([{ payload: "hello" }]);
+      } finally {
+        await restoreSql.end();
+      }
+    },
+    20_000,
+  );
 });

packages/db/src/backup-lib.ts

@@ -1,6 +1,6 @@
-import { createWriteStream, existsSync, mkdirSync, readdirSync, statSync, unlinkSync } from "node:fs";
-import { readFile } from "node:fs/promises";
+import { createReadStream, createWriteStream, existsSync, mkdirSync, readdirSync, statSync, unlinkSync } from "node:fs";
 import { basename, resolve } from "node:path";
+import { createInterface } from "node:readline";
 import postgres from "postgres";
 
 export type RunDatabaseBackupOptions = {
@@ -147,6 +147,42 @@ function tableKey(schemaName: string, tableName: string): string {
   return `${schemaName}.${tableName}`;
 }
 
+async function* readRestoreStatements(backupFile: string): AsyncGenerator<string> {
+  const stream = createReadStream(backupFile, { encoding: "utf8" });
+  const reader = createInterface({
+    input: stream,
+    crlfDelay: Infinity,
+  });
+  let statementLines: string[] = [];
+
+  const flushStatement = () => {
+    const statement = statementLines.join("\n").trim();
+    statementLines = [];
+    return statement;
+  };
+
+  try {
+    for await (const line of reader) {
+      if (line === STATEMENT_BREAKPOINT) {
+        const statement = flushStatement();
+        if (statement.length > 0) {
+          yield statement;
+        }
+        continue;
+      }
+      statementLines.push(line);
+    }
+
+    const trailingStatement = flushStatement();
+    if (trailingStatement.length > 0) {
+      yield trailingStatement;
+    }
+  } finally {
+    reader.close();
+    stream.destroy();
+  }
+}
+
 export function createBufferedTextFileWriter(filePath: string, maxBufferedBytes = DEFAULT_BACKUP_WRITE_BUFFER_BYTES) {
   const stream = createWriteStream(filePath, { encoding: "utf8" });
   const flushThreshold = Math.max(1, Math.trunc(maxBufferedBytes));
@@ -650,13 +686,7 @@ export async function runDatabaseRestore(opts: RunDatabaseRestoreOptions): Promi
 
   try {
     await sql`SELECT 1`;
-    const contents = await readFile(opts.backupFile, "utf8");
-    const statements = contents
-      .split(STATEMENT_BREAKPOINT)
-      .map((statement) => statement.trim())
-      .filter((statement) => statement.length > 0);
-
-    for (const statement of statements) {
+    for await (const statement of readRestoreStatements(opts.backupFile)) {
       await sql.unsafe(statement).execute();
     }
   } catch (error) {

packages/db/src/client.test.ts

@@ -401,4 +401,70 @@ describeEmbeddedPostgres("applyPendingMigrations", () => {
     },
     20_000,
   );
+
+  it(
+    "replays migration 0050 safely when projects.env already exists",
+    async () => {
+      const connectionString = await createTempDatabase();
+
+      await applyPendingMigrations(connectionString);
+
+      const sql = postgres(connectionString, { max: 1, onnotice: () => {} });
+      try {
+        const stiffLuckmanHash = await migrationHash("0050_stiff_luckman.sql");
+
+        await sql.unsafe(
+          `DELETE FROM "drizzle"."__drizzle_migrations" WHERE hash = '${stiffLuckmanHash}'`,
+        );
+
+        const columns = await sql.unsafe<{ column_name: string }[]>(
+          `
+            SELECT column_name
+            FROM information_schema.columns
+            WHERE table_schema = 'public'
+              AND table_name = 'projects'
+              AND column_name = 'env'
+          `,
+        );
+        expect(columns).toHaveLength(1);
+      } finally {
+        await sql.end();
+      }
+
+      const pendingState = await inspectMigrations(connectionString);
+      expect(pendingState).toMatchObject({
+        status: "needsMigrations",
+        pendingMigrations: ["0050_stiff_luckman.sql"],
+        reason: "pending-migrations",
+      });
+
+      await applyPendingMigrations(connectionString);
+
+      const finalState = await inspectMigrations(connectionString);
+      expect(finalState.status).toBe("upToDate");
+
+      const verifySql = postgres(connectionString, { max: 1, onnotice: () => {} });
+      try {
+        const columns = await verifySql.unsafe<{ column_name: string; is_nullable: string; data_type: string }[]>(
+          `
+            SELECT column_name, is_nullable, data_type
+            FROM information_schema.columns
+            WHERE table_schema = 'public'
+              AND table_name = 'projects'
+              AND column_name = 'env'
+          `,
+        );
+        expect(columns).toEqual([
+          expect.objectContaining({
+            column_name: "env",
+            is_nullable: "YES",
+            data_type: "jsonb",
+          }),
+        ]);
+      } finally {
+        await verifySql.end();
+      }
+    },
+    20_000,
+  );
 });

packages/db/src/migrations/0050_stiff_luckman.sql

@@ -0,0 +1 @@
+ALTER TABLE "projects" ADD COLUMN IF NOT EXISTS "env" jsonb;

packages/db/src/migrations/meta/_journal.json

@@ -355,6 +355,13 @@
     {
       "idx": 50,
       "version": "7",
+      "when": 1775487782768,
+      "tag": "0050_stiff_luckman",
+      "breakpoints": true
+    },
+    {
+      "idx": 51,
+      "version": "7",
       "when": 1775524651831,
       "tag": "0051_young_korg",
       "breakpoints": true

packages/db/src/schema/projects.ts

@@ -1,4 +1,5 @@
 import { pgTable, uuid, text, timestamp, date, index, jsonb } from "drizzle-orm/pg-core";
+import type { AgentEnvConfig } from "@paperclipai/shared";
 import { companies } from "./companies.js";
 import { goals } from "./goals.js";
 import { agents } from "./agents.js";
@@ -15,6 +16,7 @@ export const projects = pgTable(
     leadAgentId: uuid("lead_agent_id").references(() => agents.id),
     targetDate: date("target_date"),
     color: text("color"),
+    env: jsonb("env").$type<AgentEnvConfig>(),
     pauseReason: text("pause_reason"),
     pausedAt: timestamp("paused_at", { withTimezone: true }),
     executionWorkspacePolicy: jsonb("execution_workspace_policy").$type<Record<string, unknown>>(),