feat(cli): add transient declarative schema apply

apps/cli/src/command-internal/migration-apply.unit.test.ts

@@ -15,6 +15,7 @@ import { DbConnectError } from "./db-connection.errors.ts";
 import type { DbBatchStatement, DbSession } from "./db-connection.service.ts";
 import {
   applyMigrationFile,
+  applyRenderedSqlUnits,
   applySchemaFiles,
   hasTransactionControl,
   isPipelineIncompatible,
@@ -114,6 +115,7 @@ const executedSql = (
 const run = (
   session: DbSession,
   migrationPath: string,
+  onStatementsCommitted?: Effect.Effect<void>,
 ): Effect.Effect<void, TestError | DbConnectError> =>
   Effect.gen(function* () {
     const fs = yield* FileSystem.FileSystem;
@@ -124,9 +126,123 @@ const run = (
       path,
       migrationPath,
       (message) => new TestError({ message }),
+      onStatementsCommitted,
     );
   }).pipe(Effect.provide(BunServices.layer));
 
+describe("applyRenderedSqlUnits", () => {
+  it.effect("applies mixed transaction modes in unit order without history or reset writes", () => {
+    const { session, calls } = fakeSession();
+    return applyRenderedSqlUnits(
+      session,
+      [
+        {
+          name: "tables",
+          sql: "CREATE TABLE widgets (id bigint);\nALTER TABLE widgets ENABLE ROW LEVEL SECURITY;",
+          transactionMode: "transactional",
+        },
+        {
+          name: "enum",
+          sql: "SET check_function_bodies = off;\nALTER TYPE mood ADD VALUE 'fine';",
+          transactionMode: "none",
+        },
+        {
+          name: "grants",
+          sql: "GRANT SELECT ON TABLE widgets TO anon;",
+          transactionMode: "transactional",
+        },
+      ],
+      (message) => new TestError({ message }),
+    ).pipe(
+      Effect.tap(() =>
+        Effect.sync(() => {
+          expect(calls.map(({ kind }) => kind)).toEqual(["batch", "exec", "exec", "batch"]);
+          expect(executedSql(calls)).toEqual([
+            "CREATE TABLE widgets (id bigint)",
+            "ALTER TABLE widgets ENABLE ROW LEVEL SECURITY",
+            "SET check_function_bodies = off",
+            "ALTER TYPE mood ADD VALUE 'fine'",
+            "GRANT SELECT ON TABLE widgets TO anon",
+          ]);
+          expect(executedSql(calls).some((sql) => sql === "RESET ALL")).toBe(false);
+          expect(
+            calls.some(
+              ({ sql }) => sql.includes("supabase_migrations") || sql.includes("schema_migrations"),
+            ),
+          ).toBe(false);
+          expect(calls.some(({ kind }) => kind === "query")).toBe(false);
+        }),
+      ),
+    );
+  });
+
+  it.effect("maps transactional failures with the unit-local statement index", () => {
+    const { session, calls } = fakeSession({ failOn: "missing_column" });
+    return applyRenderedSqlUnits(
+      session,
+      [
+        {
+          name: "broken",
+          sql: "SELECT 1;\nSELECT missing_column;\nSELECT 3;",
+          transactionMode: "transactional",
+        },
+        {
+          name: "not_reached",
+          sql: "SELECT 4;",
+          transactionMode: "transactional",
+        },
+      ],
+      (message) => new TestError({ message }),
+    ).pipe(
+      Effect.flip,
+      Effect.tap((error) =>
+        Effect.sync(() => {
+          expect(error.message).toContain("At statement: 1");
+          expect(error.message).toContain("SELECT missing_column");
+          expect(executedSql(calls)).not.toContain("SELECT 4");
+        }),
+      ),
+    );
+  });
+
+  it.effect(
+    "restores a stepped-down role after a sequential failure without resetting the unit",
+    () => {
+      const restoreRoleSql = "SET SESSION ROLE postgres";
+      const { session, calls } = fakeSession({
+        failOn: "missing_column",
+        restoreRoleSql,
+      });
+      return applyRenderedSqlUnits(
+        session,
+        [
+          {
+            name: "broken_nontransactional",
+            sql: "RESET ROLE;\nSELECT missing_column;",
+            transactionMode: "none",
+          },
+        ],
+        (message) => new TestError({ message }),
+      ).pipe(
+        Effect.flip,
+        Effect.tap((error) =>
+          Effect.sync(() => {
+            expect(error.message).toContain("At statement: 1");
+            expect(executedSql(calls)).toEqual([
+              "RESET ROLE",
+              restoreRoleSql,
+              "SELECT missing_column",
+              restoreRoleSql,
+            ]);
+            expect(executedSql(calls)).not.toContain("RESET ALL");
+            expect(calls.some(({ kind }) => kind === "query")).toBe(false);
+          }),
+        ),
+      );
+    },
+  );
+});
+
 describe("applyMigrationFile", () => {
   it.effect(
     "creates the history table, then runs the statements + history insert in a transaction",
@@ -417,6 +533,90 @@ describe("applyMigrationFile", () => {
     );
   });
 
+  it.effect("does not notify after a no-transaction SET preamble when later SQL fails", () => {
+    const dir = mkdtempSync(join(tmpdir(), "apply-"));
+    const file = join(dir, "20240101120000_drop_subscription.sql");
+    writeFileSync(
+      file,
+      "-- pg-delta: transaction=false\n" +
+        "SET check_function_bodies = off;\n" +
+        "DROP SUBSCRIPTION app_events;\n" +
+        "RESET ALL;",
+    );
+    const { session } = fakeSession({ failOn: "DROP SUBSCRIPTION" });
+    let committed = 0;
+    return run(
+      session,
+      file,
+      Effect.sync(() => {
+        committed += 1;
+      }),
+    ).pipe(
+      Effect.exit,
+      Effect.tap((exit) =>
+        Effect.sync(() => {
+          expect(Exit.isFailure(exit)).toBe(true);
+          expect(committed).toBe(0);
+          rmSync(dir, { recursive: true, force: true });
+        }),
+      ),
+    );
+  });
+
+  it.effect("notifies after a no-transaction statement commits before a later failure", () => {
+    const dir = mkdtempSync(join(tmpdir(), "apply-"));
+    const file = join(dir, "20240101120000_drop_subscription.sql");
+    writeFileSync(
+      file,
+      "-- pg-delta: transaction=false\n" +
+        "CREATE TABLE widgets (id bigint);\n" +
+        "DROP SUBSCRIPTION app_events;\n" +
+        "RESET ALL;",
+    );
+    const { session } = fakeSession({ failOn: "DROP SUBSCRIPTION" });
+    let committed = 0;
+    return run(
+      session,
+      file,
+      Effect.sync(() => {
+        committed += 1;
+      }),
+    ).pipe(
+      Effect.exit,
+      Effect.tap((exit) =>
+        Effect.sync(() => {
+          expect(Exit.isFailure(exit)).toBe(true);
+          expect(committed).toBe(1);
+          rmSync(dir, { recursive: true, force: true });
+        }),
+      ),
+    );
+  });
+
+  it.effect("notifies after flushing a batch before a pipeline-incompatible failure", () => {
+    const dir = mkdtempSync(join(tmpdir(), "apply-"));
+    const file = join(dir, "20240101120000_add_index.sql");
+    writeFileSync(file, "create table a (id int);\nCREATE INDEX CONCURRENTLY a_idx ON a(id);");
+    const { session } = fakeSession({ failOn: "CONCURRENTLY" });
+    let committed = 0;
+    return run(
+      session,
+      file,
+      Effect.sync(() => {
+        committed += 1;
+      }),
+    ).pipe(
+      Effect.exit,
+      Effect.tap((exit) =>
+        Effect.sync(() => {
+          expect(Exit.isFailure(exit)).toBe(true);
+          expect(committed).toBe(1);
+          rmSync(dir, { recursive: true, force: true });
+        }),
+      ),
+    );
+  });
+
   it.effect("reports a pipeline-incompatible statement failure with its statement index", () => {
     const dir = mkdtempSync(join(tmpdir(), "apply-"));
     const file = join(dir, "20240101120000_add_index.sql");

apps/cli/src/commands/db/schema/declarative/declarative.errors.ts

@@ -62,6 +62,45 @@ export class DeclarativeInvalidDbUrlError extends Data.TaggedError("DeclarativeI
   }
 }
 
+/** A migration stem would escape the migration directory or duplicate the SQL suffix. */
+export class DeclarativeInvalidMigrationStemError extends Data.TaggedError(
+  "DeclarativeInvalidMigrationStemError",
+)<{
+  readonly message: string;
+}> {
+  get [ErrorActionabilityId](): CliErrorActionabilityDeclaration {
+    return actionability.invalidInput;
+  }
+}
+
+/** Transient apply needs explicit consent when no interactive prompt is available. */
+export class DeclarativeTransientConfirmationRequiredError extends Data.TaggedError(
+  "DeclarativeTransientConfirmationRequiredError",
+)<{
+  readonly message: string;
+  readonly suggestion: string;
+}> {
+  get [ErrorActionabilityId](): CliErrorActionabilityDeclaration {
+    return actionability.provideFlags;
+  }
+}
+
+/**
+ * `--transient` plans against the already-running local database and must not
+ * `db start` as a side effect (fresh-volume start would migrate, seed, and
+ * record history before the user confirms the planned SQL).
+ */
+export class DeclarativeLocalDbNotRunningError extends Data.TaggedError(
+  "DeclarativeLocalDbNotRunningError",
+)<{
+  readonly message: string;
+  readonly suggestion: string;
+}> {
+  get [ErrorActionabilityId](): CliErrorActionabilityDeclaration {
+    return actionability.startStack;
+  }
+}
+
 /**
  * `db schema declarative generate` ran but produced no declarative files (sync's post-generate
  * guard); message text is an established output contract.

apps/cli/src/commands/db/schema/declarative/declarative.flow.ts

@@ -45,6 +45,20 @@ export function resolveDeclarativeMigrationName(name: string, file: string): str
   return name.length > 0 ? name : file;
 }
 
+export function validateDeclarativeMigrationStem(stem: string): string | undefined {
+  const candidate = stem.trim();
+  if (candidate.includes("/") || candidate.includes("\\")) {
+    return "migration names must not contain path separators";
+  }
+  if (/\.sql$/i.test(candidate)) {
+    return "migration names must not include the .sql suffix";
+  }
+  if (candidate !== stem) {
+    return "migration names must not have leading or trailing whitespace";
+  }
+  return undefined;
+}
+
 /** Whether sync applies the generated migration, prompts, or skips. */
 export type DeclarativeApplyDecision = "apply" | "skip" | "prompt";
 

apps/cli/src/commands/db/schema/declarative/declarative.flow.unit.test.ts

@@ -10,6 +10,7 @@ import {
   resolveDeclarativeMigrationName,
   resolveDeclarativeSyncApplyDecision,
   resolveStagedDeclarativeDir,
+  validateDeclarativeMigrationStem,
 } from "./declarative.flow.ts";
 
 const stuck = (message: string) => ({
@@ -448,6 +449,23 @@ describe("resolveDeclarativeMigrationName", () => {
   });
 });
 
+describe("validateDeclarativeMigrationStem", () => {
+  it.each([
+    ["nested/name", "migration names must not contain path separators"],
+    ["nested\\name", "migration names must not contain path separators"],
+    ["change.sql", "migration names must not include the .sql suffix"],
+    ["change.SQL", "migration names must not include the .sql suffix"],
+    ["change.SQL ", "migration names must not include the .sql suffix"],
+    [" add_users ", "migration names must not have leading or trailing whitespace"],
+  ])("rejects %j", (stem, expected) => {
+    expect(validateDeclarativeMigrationStem(stem)).toBe(expected);
+  });
+
+  it("accepts a plain migration stem", () => {
+    expect(validateDeclarativeMigrationStem("add_customer_status")).toBeUndefined();
+  });
+});
+
 describe("resolveDeclarativeSyncApplyDecision", () => {
   it.each([
     ["--no-apply wins", { apply: true, noApply: true, yes: true, tty: true }, "skip"],