fix: harden datatable migration version allocation and initial-migration bookkeeping, add tests

2026-10-20 08:03:28 +00:00 · 2026-06-22 15:58:28 +02:00
parent cdc7750248
commit 26629f7dff
5 changed files with 470 additions and 78 deletions
@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT datatable || '/' || timestamp || '_' || name AS \"path!\"\n                   FROM datatable_migrations\n                   WHERE workspace_id = $1\n                     AND datatable || '/' || timestamp || '_' || name = ANY($2)",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "path!",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "TextArray"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "12afc363f5dae345a45e8e0e4234e09d6c746c01cc32257ee782ce37e4d4cfa6"
+}
@@ -0,0 +1,24 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM datatable_migrations WHERE workspace_id = $1 AND datatable = $2 AND timestamp = $3 RETURNING name",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "name",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "3c057e0284f1219865b8f7dc6eb3c1293c4b72d0b52abeefd1e954617984a2d8"
+}
@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "pg_advisory_xact_lock",
+        "type_info": "Void"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "b3200181380df2f645bfce270dc1cf520939fa4fa4b65371d11db7416ba0b14c"
+}
@@ -1885,6 +1885,82 @@ mod tests {
            format!("text={}...", "é".repeat(27))
        );
    }
+
+    #[test]
+    fn validate_migration_name_accepts_safe_names() {
+        for name in ["initial", "add_index_to_customers", "fix-bug_2", "ABC123"] {
+            assert!(
+                validate_migration_name(name).is_ok(),
+                "{name} should be valid"
+            );
+        }
+    }
+
+    #[test]
+    fn validate_migration_name_rejects_unsafe_names() {
+        for name in [
+            "",
+            "add index",
+            "a/b",
+            "a\\b",
+            "a..b",
+            "a.b",
+            "naïve",
+            "a/../b",
+        ] {
+            assert!(
+                validate_migration_name(name).is_err(),
+                "{name} should be rejected"
+            );
+        }
+    }
+
+    #[test]
+    fn validate_datatable_path_segment_accepts_and_rejects() {
+        for ok in ["mydt", "my-dt", "main", "a b"] {
+            assert!(
+                validate_datatable_path_segment(ok).is_ok(),
+                "{ok} should be ok"
+            );
+        }
+        for bad in ["", "a/b", "a\\b", "..", "a..b", "../etc", "x/.."] {
+            assert!(
+                validate_datatable_path_segment(bad).is_err(),
+                "{bad} should be rejected"
+            );
+        }
+    }
+
+    #[test]
+    fn parse_datatable_migration_diff_path_roundtrips() {
+        assert_eq!(
+            parse_datatable_migration_diff_path("mydt/20260101000001_create_users.up.sql"),
+            Some(("mydt".to_string(), 20260101000001))
+        );
+        // up and down map to the same (datatable, timestamp) record.
+        assert_eq!(
+            parse_datatable_migration_diff_path("mydt/20260101000001_create_users.down.sql"),
+            Some(("mydt".to_string(), 20260101000001))
+        );
+        // datatable names may themselves be hyphenated.
+        assert_eq!(
+            parse_datatable_migration_diff_path("my-dt/42_x.up.sql"),
+            Some(("my-dt".to_string(), 42))
+        );
+    }
+
+    #[test]
+    fn parse_datatable_migration_diff_path_rejects_malformed() {
+        // no slash → not a migration path
+        assert_eq!(parse_datatable_migration_diff_path("nofile"), None);
+        // filename not starting with digits → no timestamp
+        assert_eq!(
+            parse_datatable_migration_diff_path("mydt/create_users.up.sql"),
+            None
+        );
+        // empty filename
+        assert_eq!(parse_datatable_migration_diff_path("mydt/"), None);
+    }
 }

 /// Resolve a source string to PgDatabase credentials with user-scoped permission checks.
@@ -3229,6 +3305,119 @@ fn validate_datatable_path_segment(datatable: &str) -> Result<()> {
    Ok(())
 }

+/// Record a data table migration change as a deployed object so it is tallied
+/// into `workspace_diff` and shows up as a `datatable_migration` item in the
+/// workspace-merge diff. The diff path is `<datatable>/<timestamp>_<name>`,
+/// matching `parse_datatable_migration_diff_path`.
+async fn record_datatable_migration_deployment(
+    authed: &ApiAuthed,
+    db: &DB,
+    w_id: &str,
+    datatable: &str,
+    timestamp: i64,
+    name: &str,
+) -> Result<()> {
+    handle_deployment_metadata(
+        &authed.email,
+        &authed.username,
+        db,
+        w_id,
+        DeployedObject::DatatableMigration { path: format!("{datatable}/{timestamp}_{name}") },
+        Some(format!(
+            "Data table migration {name} ({timestamp}) on {datatable}"
+        )),
+        false,
+        None,
+    )
+    .await
+}
+
+/// Allocate the next version for a data table and insert the migration
+/// definition, in one transaction. A per-(workspace, data table) advisory lock
+/// serializes concurrent version allocation so two creates can't read the same
+/// `MAX(timestamp)` and collide on the `(workspace_id, datatable, timestamp)`
+/// primary key. The version is the current UTC `YYYYMMDDHHMMSS`, bumped past any
+/// existing version to stay unique and monotonically increasing.
+async fn insert_datatable_migration_def(
+    tx: &mut Transaction<'_, Postgres>,
+    w_id: &str,
+    datatable: &str,
+    name: &str,
+    code_up: &str,
+    code_down: Option<&str>,
+) -> Result<i64> {
+    sqlx::query!(
+        "SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))",
+        w_id,
+        datatable,
+    )
+    .execute(&mut **tx)
+    .await?;
+
+    let now_ts: i64 = Utc::now()
+        .format("%Y%m%d%H%M%S")
+        .to_string()
+        .parse()
+        .map_err(|e| Error::internal_err(format!("Failed to build migration version: {}", e)))?;
+    let max_existing: Option<i64> = sqlx::query_scalar!(
+        "SELECT MAX(timestamp) FROM datatable_migrations WHERE workspace_id = $1 AND datatable = $2",
+        w_id,
+        datatable,
+    )
+    .fetch_one(&mut **tx)
+    .await?;
+    let timestamp = match max_existing {
+        Some(m) if m >= now_ts => m + 1,
+        _ => now_ts,
+    };
+
+    sqlx::query!(
+        "INSERT INTO datatable_migrations (workspace_id, datatable, timestamp, name, code_up, code_down) \
+         VALUES ($1, $2, $3, $4, $5, $6)",
+        w_id,
+        datatable,
+        timestamp,
+        name,
+        code_up,
+        code_down,
+    )
+    .execute(&mut **tx)
+    .await?;
+
+    Ok(timestamp)
+}
+
+/// Mark a version as already installed in a data table's `_wm_migrations` table
+/// (ensuring the table exists first).
+async fn mark_datatable_version_installed(db: &DB, pg_db: &PgDatabase, version: i64) -> Result<()> {
+    let (client, connection) = pg_db.connect(Some(db)).await?;
+    tokio::spawn(async move {
+        if let Err(e) = connection.await {
+            tracing::error!("Datatable connection error: {}", e);
+        }
+    });
+    client
+        .batch_execute(
+            "CREATE TABLE IF NOT EXISTS _wm_migrations (\
+                version BIGINT PRIMARY KEY, \
+                installed_at TIMESTAMPTZ NOT NULL DEFAULT now())",
+        )
+        .await
+        .map_err(|e| {
+            Error::internal_err(format!("Failed to ensure _wm_migrations table: {}", e))
+        })?;
+    client
+        .execute(
+            "INSERT INTO _wm_migrations (version) VALUES ($1) ON CONFLICT DO NOTHING",
+            &[&version],
+        )
+        .await
+        .map_err(|e| {
+            Error::internal_err(format!("Failed to mark initial migration installed: {}", e))
+        })?;
+    Ok(())
+}
+
 /// Create a single migration for a data table. The version is generated
 /// server-side (current UTC `YYYYMMDDHHMMSS`), bumped past any existing version
 /// so it stays unique and monotonically increasing.
@@ -3243,35 +3432,17 @@ async fn create_datatable_migration(
    validate_migration_name(&payload.name)?;
    ensure_datatable_migrations_enabled(&db, &w_id, &datatable_name).await?;

-    let now_ts: i64 = Utc::now()
-        .format("%Y%m%d%H%M%S")
-        .to_string()
-        .parse()
-        .map_err(|e| Error::internal_err(format!("Failed to build migration version: {}", e)))?;
-    let max_existing: Option<i64> = sqlx::query_scalar!(
-        "SELECT MAX(timestamp) FROM datatable_migrations WHERE workspace_id = $1 AND datatable = $2",
+    let mut tx = db.begin().await?;
+    let timestamp = insert_datatable_migration_def(
+        &mut tx,
        &w_id,
        &datatable_name,
-    )
-    .fetch_one(&db)
-    .await?;
-    let timestamp = match max_existing {
-        Some(m) if m >= now_ts => m + 1,
-        _ => now_ts,
-    };
-
-    sqlx::query!(
-        "INSERT INTO datatable_migrations (workspace_id, datatable, timestamp, name, code_up, code_down) \
-         VALUES ($1, $2, $3, $4, $5, $6)",
-        &w_id,
-        &datatable_name,
-        timestamp,
        &payload.name,
        &payload.code_up,
        payload.code_down.as_deref(),
    )
-    .execute(&db)
    .await?;
+    tx.commit().await?;

    audit_log(
        &db,
@@ -3284,6 +3455,16 @@ async fn create_datatable_migration(
    )
    .await?;

+    record_datatable_migration_deployment(
+        &authed,
+        &db,
+        &w_id,
+        &datatable_name,
+        timestamp,
+        &payload.name,
+    )
+    .await?;
+
    Ok(Json(DatatableMigration {
        datatable: datatable_name,
        timestamp,
@@ -3301,14 +3482,15 @@ async fn delete_datatable_migration(
 ) -> Result<String> {
    require_admin(authed.is_admin, &authed.username)?;

-    sqlx::query!(
+    let deleted_name = sqlx::query_scalar!(
        "DELETE FROM datatable_migrations \
-         WHERE workspace_id = $1 AND datatable = $2 AND timestamp = $3",
+         WHERE workspace_id = $1 AND datatable = $2 AND timestamp = $3 \
+         RETURNING name",
        &w_id,
        &datatable_name,
        timestamp,
    )
-    .execute(&db)
+    .fetch_optional(&db)
    .await?;

    audit_log(
@@ -3322,6 +3504,19 @@ async fn delete_datatable_migration(
    )
    .await?;

+    // Only tally a change if a migration was actually deleted.
+    if let Some(name) = deleted_name {
+        record_datatable_migration_deployment(
+            &authed,
+            &db,
+            &w_id,
+            &datatable_name,
+            timestamp,
+            &name,
+        )
+        .await?;
+    }
+
    Ok(format!(
        "Deleted migration {} from {}",
        timestamp, datatable_name
@@ -3377,6 +3572,16 @@ async fn upsert_datatable_migration(
    )
    .await?;

+    record_datatable_migration_deployment(
+        &authed,
+        &db,
+        &w_id,
+        &datatable_name,
+        payload.timestamp,
+        &payload.name,
+    )
+    .await?;
+
    Ok(format!(
        "Upserted migration {} in {}",
        payload.timestamp, datatable_name
@@ -3412,61 +3617,29 @@ async fn generate_initial_datatable_migration(
        .collect::<Vec<_>>()
        .join("\n");

-    let now_ts: i64 = Utc::now()
-        .format("%Y%m%d%H%M%S")
-        .to_string()
-        .parse()
-        .map_err(|e| Error::internal_err(format!("Failed to build migration version: {}", e)))?;
-    let max_existing: Option<i64> = sqlx::query_scalar!(
-        "SELECT MAX(timestamp) FROM datatable_migrations WHERE workspace_id = $1 AND datatable = $2",
-        &w_id,
-        &datatable_name,
-    )
-    .fetch_one(&db)
-    .await?;
-    let timestamp = match max_existing {
-        Some(m) if m >= now_ts => m + 1,
-        _ => now_ts,
-    };
+    // Record the definition first, then mark it installed. If marking fails we
+    // delete the definition, so a failure leaves no phantom "initial" (rather
+    // than a `_wm_migrations` version with no definition that the UI can't
+    // clear). The narrow window where it briefly shows "not run" is benign:
+    // running it would just no-op/fail harmlessly against the existing schema.
+    let mut tx = db.begin().await?;
+    let timestamp =
+        insert_datatable_migration_def(&mut tx, &w_id, &datatable_name, "initial", &code_up, None)
+            .await?;
+    tx.commit().await?;

-    // Mark the version installed in the data table BEFORE recording the
-    // definition, so the migration is never observable as "not run".
-    let (client, connection) = pg_db.connect(Some(&db)).await?;
-    tokio::spawn(async move {
-        if let Err(e) = connection.await {
-            tracing::error!("Datatable connection error: {}", e);
-        }
-    });
-    client
-        .batch_execute(
-            "CREATE TABLE IF NOT EXISTS _wm_migrations (\
-                version BIGINT PRIMARY KEY, \
-                installed_at TIMESTAMPTZ NOT NULL DEFAULT now())",
+    if let Err(e) = mark_datatable_version_installed(&db, &pg_db, timestamp).await {
+        let _ = sqlx::query!(
+            "DELETE FROM datatable_migrations \
+             WHERE workspace_id = $1 AND datatable = $2 AND timestamp = $3",
+            &w_id,
+            &datatable_name,
+            timestamp,
        )
-        .await
-        .map_err(|e| {
-            Error::internal_err(format!("Failed to ensure _wm_migrations table: {}", e))
-        })?;
-    client
-        .execute(
-            "INSERT INTO _wm_migrations (version) VALUES ($1) ON CONFLICT DO NOTHING",
-            &[&timestamp],
-        )
-        .await
-        .map_err(|e| {
-            Error::internal_err(format!("Failed to mark initial migration installed: {}", e))
-        })?;
-
-    sqlx::query!(
-        "INSERT INTO datatable_migrations (workspace_id, datatable, timestamp, name, code_up, code_down) \
-         VALUES ($1, $2, $3, 'initial', $4, NULL)",
-        &w_id,
-        &datatable_name,
-        timestamp,
-        &code_up,
-    )
-    .execute(&db)
-    .await?;
+        .execute(&db)
+        .await;
+        return Err(e);
+    }

    audit_log(
        &db,
@@ -3479,6 +3652,16 @@ async fn generate_initial_datatable_migration(
    )
    .await?;

+    record_datatable_migration_deployment(
+        &authed,
+        &db,
+        &w_id,
+        &datatable_name,
+        timestamp,
+        "initial",
+    )
+    .await?;
+
    Ok(Json(DatatableMigration {
        datatable: datatable_name,
        timestamp,
@@ -7466,6 +7649,7 @@ pub struct CompareSummary {
    pub folders_changed: usize,
    pub schedules_changed: usize,
    pub triggers_changed: usize,
+    pub datatable_migrations_changed: usize,
    pub conflicts: usize, // Items that are both ahead and behind
 }

@@ -7774,6 +7958,10 @@ async fn compare_workspaces(
            .iter()
            .filter(|s| s.kind.ends_with("_trigger"))
            .count(),
+        datatable_migrations_changed: visible_diffs
+            .iter()
+            .filter(|s| s.kind == "datatable_migration")
+            .count(),
        conflicts: visible_diffs
            .iter()
            .filter(|s| s.ahead > 0 && s.behind > 0)
@@ -8019,6 +8207,18 @@ async fn query_visible_items<'c>(
                .fetch_all(&mut **tx)
                .await?
            }
+            "datatable_migration" => {
+                sqlx::query_scalar!(
+                    r#"SELECT datatable || '/' || timestamp || '_' || name AS "path!"
+                   FROM datatable_migrations
+                   WHERE workspace_id = $1
+                     AND datatable || '/' || timestamp || '_' || name = ANY($2)"#,
+                    workspace_id,
+                    &paths_vec
+                )
+                .fetch_all(&mut **tx)
+                .await?
+            }
            k if TRIGGER_OR_SCHEDULE_TABLES.contains(&k) => {
                // SAFETY: `kind` comes from a hardcoded allowlist
                // TRIGGER_OR_SCHEDULE_TABLES, not user input.
@@ -0,0 +1,122 @@
+/**
+ * Unit tests for datatable-migration path parsing and local validation.
+ *
+ * These exercise pure logic with no backend:
+ *  - `parseDatatableMigrationPath` recognizes only the
+ *    `migrations/datatable/<dt>/<timestamp>_<name>.(up|down).sql` shape.
+ *  - `validateLocalMigrations` rejects the two invalid on-disk states a push
+ *    must catch: two up (or two down) files sharing a timestamp, and a
+ *    `.down.sql` with no matching `.up.sql`.
+ */
+
+import { afterEach, beforeEach, describe, expect, test } from "bun:test";
+import * as fs from "node:fs";
+import * as path from "node:path";
+import * as os from "node:os";
+import { parseDatatableMigrationPath } from "../src/types.ts";
+import { validateLocalMigrations } from "../src/commands/datatable_migrations.ts";
+
+describe("parseDatatableMigrationPath", () => {
+  test("parses up and down files of the new layout", () => {
+    expect(
+      parseDatatableMigrationPath(
+        "migrations/datatable/mydt/20260101000001_create_users.up.sql",
+      ),
+    ).toEqual({
+      datatable: "mydt",
+      timestamp: 20260101000001,
+      name: "create_users",
+      kind: "up",
+    });
+    expect(
+      parseDatatableMigrationPath(
+        "migrations/datatable/my-dt/42_x.down.sql",
+      ),
+    ).toEqual({ datatable: "my-dt", timestamp: 42, name: "x", kind: "down" });
+  });
+
+  test("rejects unrelated, legacy and malformed paths", () => {
+    for (
+      const p of [
+        // legacy top-level layout
+        "datatable_migrations/mydt/20260101000001_x.up.sql",
+        // wrong sub-namespace / depth
+        "migrations/ducklake/mydt/1_x.up.sql",
+        "migrations/datatable/1_x.up.sql",
+        "migrations/datatable/mydt/sub/1_x.up.sql",
+        // not a migration file
+        "migrations/datatable/mydt/notes.txt",
+        "migrations/datatable/mydt/x.up.sql", // no numeric timestamp prefix
+        // unrelated workspace files
+        "f/foo/bar.script.yaml",
+        "u/admin/script.ts",
+      ]
+    ) {
+      expect(parseDatatableMigrationPath(p)).toBeUndefined();
+    }
+  });
+});
+
+describe("validateLocalMigrations", () => {
+  let prevCwd: string;
+  let tmp: string;
+
+  beforeEach(() => {
+    prevCwd = process.cwd();
+    tmp = fs.mkdtempSync(path.join(os.tmpdir(), "dtmig-"));
+    process.chdir(tmp);
+  });
+  afterEach(() => {
+    process.chdir(prevCwd);
+    fs.rmSync(tmp, { recursive: true, force: true });
+  });
+
+  function write(datatable: string, file: string) {
+    const dir = path.join(tmp, "migrations", "datatable", datatable);
+    fs.mkdirSync(dir, { recursive: true });
+    fs.writeFileSync(path.join(dir, file), "-- sql\n");
+  }
+
+  test("accepts up+down pairs and up-only migrations", () => {
+    write("mydt", "20260101000001_create_users.up.sql");
+    write("mydt", "20260101000001_create_users.down.sql");
+    write("mydt", "20260101000002_add_email.up.sql"); // down is optional
+    expect(validateLocalMigrations()).toEqual([]);
+  });
+
+  test("flags two up files sharing a timestamp", () => {
+    write("mydt", "20260101000003_foo.up.sql");
+    write("mydt", "20260101000003_bar.up.sql");
+    const errors = validateLocalMigrations();
+    expect(errors.length).toBe(1);
+    expect(errors[0]).toContain("20260101000003");
+  });
+
+  test("flags two down files sharing a timestamp", () => {
+    write("mydt", "20260101000004_a.up.sql");
+    write("mydt", "20260101000004_a.down.sql");
+    write("mydt", "20260101000004_b.down.sql");
+    const errors = validateLocalMigrations();
+    // duplicate down + the b.down orphan (no b.up)
+    expect(errors.some((e) => e.includes("down") && e.includes("20260101000004"))).toBe(true);
+  });
+
+  test("flags a down file with no matching up", () => {
+    write("mydt", "20260101000005_orphan.down.sql");
+    const errors = validateLocalMigrations();
+    expect(errors.length).toBe(1);
+    expect(errors[0]).toContain("20260101000005_orphan");
+  });
+
+  test("only validates the requested datatables", () => {
+    write("bad", "20260101000006_x.up.sql");
+    write("bad", "20260101000006_y.up.sql"); // duplicate, but in 'bad'
+    write("good", "20260101000007_ok.up.sql");
+    expect(validateLocalMigrations(new Set(["good"]))).toEqual([]);
+    expect(validateLocalMigrations(new Set(["bad"])).length).toBe(1);
+  });
+
+  test("returns no errors when the migrations folder is absent", () => {
+    expect(validateLocalMigrations()).toEqual([]);
+  });
+});