feat: sync datatable migrations as files via the workspace export

2026-10-20 08:03:28 +00:00 · 2026-06-17 19:52:28 +02:00
parent 541a1407e0
commit bd5a047169
7 changed files with 249 additions and 201 deletions
@@ -0,0 +1,19 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO datatable_migrations (workspace_id, datatable, timestamp, name, code_up, code_down) VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (workspace_id, datatable, timestamp) DO UPDATE SET name = EXCLUDED.name, code_up = EXCLUDED.code_up, code_down = EXCLUDED.code_down",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Int8",
+        "Varchar",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "c5639d48c92d6863f16f97de91eae590d0ed2f4b7831956bb429a47c478f7c1f"
+}
@@ -0,0 +1,46 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT datatable, timestamp, name, code_up, code_down FROM datatable_migrations WHERE workspace_id = $1 ORDER BY datatable, timestamp",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "datatable",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "timestamp",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 2,
+        "name": "name",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "code_up",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 4,
+        "name": "code_down",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      true
+    ]
+  },
+  "hash": "f89b024949f29eb5142744635914b94fd84c1bb64e9659f798df2888848894bd"
+}
@@ -4577,6 +4577,44 @@ paths:
              schema:
                type: string

+  /w/{workspace}/workspaces/upsert_datatable_migration/{datatable_name}:
+    post:
+      summary: insert or update a single datatable migration at an explicit version
+      operationId: upsertDatatableMigration
+      tags:
+        - workspace
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - name: datatable_name
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              required: [timestamp, name, code_up]
+              properties:
+                timestamp:
+                  type: integer
+                  format: int64
+                name:
+                  type: string
+                code_up:
+                  type: string
+                code_down:
+                  type: string
+      responses:
+        "200":
+          description: status
+          content:
+            text/plain:
+              schema:
+                type: string
+
  /w/{workspace}/workspaces/generate_initial_datatable_migration/{datatable_name}:
    post:
      summary: snapshot the current schema as an already-installed initial migration
@@ -1501,6 +1501,34 @@ pub(crate) async fn tarball_workspace(
            .await?;
    }

+    {
+        // Data table migrations live in the `datatable_migrations` table; surface
+        // them in the export as `datatable_migrations/<datatable>/<version>_<name>`
+        // .up.sql (and .down.sql when present) so `wmill sync` treats them like any
+        // other workspace item.
+        let migrations = sqlx::query!(
+            "SELECT datatable, timestamp, name, code_up, code_down FROM datatable_migrations \
+             WHERE workspace_id = $1 ORDER BY datatable, timestamp",
+            &w_id
+        )
+        .fetch_all(&mut *tx)
+        .await?;
+        for m in migrations {
+            let base = format!(
+                "datatable_migrations/{}/{}_{}",
+                m.datatable, m.timestamp, m.name
+            );
+            archive
+                .write_to_archive(&m.code_up, &format!("{base}.up.sql"))
+                .await?;
+            if let Some(code_down) = m.code_down {
+                archive
+                    .write_to_archive(&code_down, &format!("{base}.down.sql"))
+                    .await?;
+            }
+        }
+    }
+
    archive.finish().await?;

    let file = tokio::fs::File::open(&file_path).await?;
@@ -3,12 +3,14 @@ import * as path from "node:path";
 import * as log from "../core/log.ts";
 import { colors } from "@cliffy/ansi/colors";
 import * as wmill from "../../gen/services.gen.ts";
-import type { DatatableMigration } from "../../gen/types.gen.ts";
-import { readTextFile, readTextFileSync } from "../utils/utils.ts";
+import { readTextFile } from "../utils/utils.ts";
 import { Confirm } from "@cliffy/prompt/confirm";

 // Migrations live under <cwd>/datatable_migrations/<datatable>/, one folder per
 // target data table, as `<timestamp>_<name>.up.sql` (and optional `.down.sql`).
+// They are synced as ordinary workspace files (see the workspace tarball export
+// and the `datatable_migration` handling in sync.ts); this module only holds the
+// `wmill datatable migrate` command helpers and the per-file push primitive.
 const MIGRATIONS_DIR = "datatable_migrations";

 // Migration names map directly onto file names and the DB `name` column.
@@ -103,121 +105,45 @@ export async function rollbackMigrations(
  }
 }

-function upFileName(m: { timestamp: number; name: string }): string {
-  return `${m.timestamp}_${m.name}.up.sql`;
-}
-
-function downFileName(m: { timestamp: number; name: string }): string {
-  return `${m.timestamp}_${m.name}.down.sql`;
-}
-
 /**
- * Parse a `<timestamp>_<name>.up.sql` / `.down.sql` file name into its
- * `{ timestamp, name, kind }` parts. Returns undefined for unrelated files.
+ * Sync a single migration to the workspace based on the current on-disk state of
+ * its `<datatable>/<timestamp>_<name>.up.sql` file: upsert it when the up file
+ * exists, otherwise delete it. Called by `wmill sync push` for each changed
+ * `datatable_migration` file.
 */
-function parseMigrationFileName(
-  file: string,
-): { timestamp: number; name: string; kind: "up" | "down" } | undefined {
-  const m = file.match(/^(\d+)_(.*)\.(up|down)\.sql$/);
-  if (!m) return undefined;
-  return { timestamp: Number(m[1]), name: m[2], kind: m[3] as "up" | "down" };
-}
-
-function sortMigrations(migrations: DatatableMigration[]): DatatableMigration[] {
-  return migrations.sort(
-    (a, b) =>
-      a.datatable.localeCompare(b.datatable) || a.timestamp - b.timestamp,
-  );
-}
-
-/**
- * Push the local <cwd>/datatable_migrations/ folder to the workspace, replacing
- * the remote set. Returns true if a push was performed, false if the folder is
- * missing or already up to date.
- */
-export async function pushDatatableMigrations(
+export async function pushMigrationFromDisk(
  workspace: string,
-  opts?: { yes?: boolean; jsonOutput?: boolean },
-): Promise<boolean> {
-  const localDir = path.join(process.cwd(), MIGRATIONS_DIR);
-  if (!fs.existsSync(localDir)) {
-    return false;
+  m: { datatable: string; timestamp: number; name: string },
+): Promise<void> {
+  const dir = path.join(process.cwd(), MIGRATIONS_DIR, m.datatable);
+  const base = `${m.timestamp}_${m.name}`;
+  const upPath = path.join(dir, `${base}.up.sql`);
+
+  if (!fs.existsSync(upPath)) {
+    log.info(colors.red(`Deleting datatable_migration ${m.datatable}/${base}`));
+    await wmill.deleteDatatableMigration({
+      workspace,
+      datatableName: m.datatable,
+      timestamp: m.timestamp,
+    });
+    return;
  }

-  // Each immediate subfolder is a target data table; read its migration files.
-  const migrations: DatatableMigration[] = [];
-  for (const datatable of fs.readdirSync(localDir)) {
-    const dtDir = path.join(localDir, datatable);
-    if (!fs.statSync(dtDir).isDirectory()) continue;
+  const code_up = await readTextFile(upPath);
+  const downPath = path.join(dir, `${base}.down.sql`);
+  const code_down = fs.existsSync(downPath) ? await readTextFile(downPath) : undefined;

-    // Group .up.sql / .down.sql files by their (timestamp, name) key.
-    const byKey = new Map<
-      string,
-      { timestamp: number; name: string; code_up?: string; code_down?: string }
-    >();
-    for (const file of fs.readdirSync(dtDir)) {
-      const parsed = parseMigrationFileName(file);
-      if (!parsed) continue;
-      const key = `${parsed.timestamp}_${parsed.name}`;
-      const entry =
-        byKey.get(key) ?? { timestamp: parsed.timestamp, name: parsed.name };
-      const content = await readTextFile(path.join(dtDir, file));
-      if (parsed.kind === "up") entry.code_up = content;
-      else entry.code_down = content;
-      byKey.set(key, entry);
-    }
-
-    for (const entry of byKey.values()) {
-      if (entry.code_up === undefined) {
-        log.warn(
-          colors.yellow(
-            `Skipping ${datatable}/${entry.timestamp}_${entry.name}: missing .up.sql file`,
-          ),
-        );
-        continue;
-      }
-      migrations.push({
-        datatable,
-        timestamp: entry.timestamp,
-        name: entry.name,
-        code_up: entry.code_up,
-        ...(entry.code_down !== undefined ? { code_down: entry.code_down } : {}),
-      });
-    }
-  }
-  sortMigrations(migrations);
-
-  // Skip if the remote set is already identical.
-  let remote: DatatableMigration[] = [];
-  try {
-    remote = await wmill.listDatatableMigrations({ workspace });
-  } catch {
-    // If the endpoint is missing or unauthorized, just attempt the push.
-  }
-  if (JSON.stringify(remote) === JSON.stringify(migrations)) {
-    log.info(colors.gray("Datatable migrations up to date"));
-    return false;
-  }
-
-  await wmill.updateDatatableMigrations({
+  log.info(colors.green(`Pushing datatable_migration ${m.datatable}/${base}`));
+  await wmill.upsertDatatableMigration({
    workspace,
-    requestBody: { migrations },
+    datatableName: m.datatable,
+    requestBody: {
+      timestamp: m.timestamp,
+      name: m.name,
+      code_up,
+      ...(code_down !== undefined ? { code_down } : {}),
+    },
  });
-  log.info(
-    colors.green(`Pushed ${migrations.length} datatable migration(s)`),
-  );
-
-  // Migrations newly introduced by this push (absent from the remote set before),
-  // keyed by (datatable, timestamp) since timestamps are unique only per table.
-  const remoteKeys = new Set(remote.map((m) => `${m.datatable}\0${m.timestamp}`));
-  const newMigrations = migrations.filter(
-    (m) => !remoteKeys.has(`${m.datatable}\0${m.timestamp}`),
-  );
-  if (newMigrations.length > 0) {
-    await offerToRunNewMigrations(workspace, newMigrations, opts);
-  }
-
-  return true;
 }

 /**
@@ -225,11 +151,13 @@ export async function pushDatatableMigrations(
 * offer to run them, equivalent to `wmill datatable migrate up` on each affected
 * data table.
 */
-async function offerToRunNewMigrations(
+export async function offerToRunNewMigrations(
  workspace: string,
-  newMigrations: DatatableMigration[],
+  newMigrations: { datatable: string; timestamp: number; name: string }[],
  opts?: { yes?: boolean; jsonOutput?: boolean },
 ): Promise<void> {
+  if (newMigrations.length === 0) return;
+
  log.info(colors.green("New migrations were pushed:"));
  for (const m of newMigrations) {
    log.info(colors.gray(`  ${m.datatable}: ${m.timestamp} ${m.name}`));
@@ -254,74 +182,3 @@ async function offerToRunNewMigrations(
    await runMigrations(workspace, datatable);
  }
 }
-
-/**
- * Pull the workspace's datatable migrations into
- * <cwd>/datatable_migrations/<datatable>/ as `<timestamp>_<name>.up.sql` (and
- * `.down.sql` when a down migration exists). Files no longer present remotely
- * are removed locally.
- */
-export async function pullDatatableMigrations(
-  workspace: string,
-): Promise<boolean> {
-  const localDir = path.join(process.cwd(), MIGRATIONS_DIR);
-  let migrations: DatatableMigration[];
-  try {
-    migrations = await wmill.listDatatableMigrations({ workspace });
-  } catch (e) {
-    log.debug(`Skipping datatable migrations pull: ${e}`);
-    return false;
-  }
-
-  if (migrations.length === 0 && !fs.existsSync(localDir)) {
-    return false;
-  }
-
-  // Relative paths (`<datatable>/<file>`) that should exist after the pull.
-  const known = new Set<string>();
-  for (const m of migrations) {
-    const dtDir = path.join(localDir, m.datatable);
-    fs.mkdirSync(dtDir, { recursive: true });
-    const up = upFileName(m);
-    known.add(`${m.datatable}/${up}`);
-    writeIfChanged(path.join(dtDir, up), m.code_up);
-    if (m.code_down !== undefined && m.code_down !== null) {
-      const down = downFileName(m);
-      known.add(`${m.datatable}/${down}`);
-      writeIfChanged(path.join(dtDir, down), m.code_down);
-    }
-  }
-
-  // Delete locally-orphaned migration files across all datatable subfolders.
-  if (fs.existsSync(localDir)) {
-    for (const datatable of fs.readdirSync(localDir)) {
-      const dtDir = path.join(localDir, datatable);
-      if (!fs.statSync(dtDir).isDirectory()) continue;
-      for (const file of fs.readdirSync(dtDir)) {
-        if (!parseMigrationFileName(file)) continue;
-        if (!known.has(`${datatable}/${file}`)) {
-          try {
-            fs.unlinkSync(path.join(dtDir, file));
-          } catch {
-            // ignore
-          }
-        }
-      }
-    }
-  }
-
-  log.info(colors.green(`Pulled ${migrations.length} datatable migration(s)`));
-  return true;
-}
-
-function writeIfChanged(full: string, content: string): void {
-  let existing: string | undefined;
-  try {
-    existing = readTextFileSync(full);
-  } catch {
-    existing = undefined;
-  }
-  if (existing !== content) {
-    fs.writeFileSync(full, content, "utf-8");
-  }
-}
@@ -23,13 +23,15 @@ import {
  showDiff,
  extractNativeTriggerInfo,
  redactEncryptionKey,
+  isDatatableMigrationPath,
+  parseDatatableMigrationPath,
 } from "../../types.ts";
 import { downloadZip } from "./pull.ts";
 import { runLint, printReport, checkMissingLocks } from "../lint/lint.ts";
 import { pullSharedUi, pushSharedUi } from "../shared_ui.ts";
 import {
-  pullDatatableMigrations,
-  pushDatatableMigrations,
+  pushMigrationFromDisk,
+  offerToRunNewMigrations,
 } from "../datatable_migrations.ts";

 import {
@@ -2480,7 +2482,8 @@ const isNotWmillFile = (p: string, isDirectory: boolean) => {
      !p.startsWith("g" + SEP) &&
      !p.startsWith("users" + SEP) &&
      !p.startsWith("groups" + SEP) &&
-      !p.startsWith("dependencies" + SEP)
+      !p.startsWith("dependencies" + SEP) &&
+      !p.startsWith("datatable_migrations" + SEP)
    );
  }

@@ -2491,6 +2494,11 @@ const isNotWmillFile = (p: string, isDirectory: boolean) => {

  try {
    const typ = getTypeStrFromPath(p);
+    // Datatable migrations live under datatable_migrations/<datatable>/, outside
+    // the u/f/g namespaces, but are valid wmill files.
+    if (typ == "datatable_migration") {
+      return false;
+    }
    if (
      typ == "resource-type" ||
      typ == "settings" ||
@@ -2522,7 +2530,8 @@ export const isWhitelisted = (p: string) => {
    p == "ui" ||
    p == "users" ||
    p == "groups" ||
-    p == "dependencies"
+    p == "dependencies" ||
+    p == "datatable_migrations"
  );
 };

@@ -2617,6 +2626,11 @@ interface ChangeTracker {
 }

 async function addToChangedIfNotExists(p: string, tracker: ChangeTracker) {
+  // Datatable migration .sql files are not scripts; they're synced via the
+  // dedicated datatable_migration handler in the push loop.
+  if (isDatatableMigrationPath(p)) {
+    return;
+  }
  const isScript = exts.some((e) => p.endsWith(e)) && !isFileResource(p) && !isFilesetResource(p);
  if (isScript) {
    if (isFlowPath(p)) {
@@ -3353,11 +3367,8 @@ export async function pull(
    log.warn(`Failed to pull shared UI folder: ${e}`);
  }

-  try {
-    await pullDatatableMigrations(workspace.workspaceId);
-  } catch (e) {
-    log.warn(`Failed to pull datatable migrations folder: ${e}`);
-  }
+  // Datatable migrations are part of the workspace export now, so they flow
+  // through the normal diff/apply above as `datatable_migration` items.

  // Git-sync deployment-callback mode stops here: branch checkout + pull have
  // happened, but commit + push are the caller's job. The hub script does
@@ -4246,6 +4257,21 @@ export async function push(
    // Cache git branch at the start to avoid repeated execSync calls per change
    const cachedWsNameForPush = wsNameForFiles || (isGitRepository() ? getCurrentGitBranch() : null);

+    // Datatable migrations are two files (.up.sql/.down.sql) for one record, so
+    // dedupe upsert/delete by (datatable, version) across the whole push.
+    const pushedMigrationKeys = new Set<string>();
+    // Migrations newly added by this push (an added .up.sql) — offered to run once
+    // the push has completed.
+    const newDatatableMigrations = changes
+      .filter((c) => c.name === "added")
+      .map((c) => parseDatatableMigrationPath(c.path))
+      .filter((p) => !!p && p.kind === "up")
+      .map((p) => ({
+        datatable: p!.datatable,
+        timestamp: p!.timestamp,
+        name: p!.name,
+      }));
+
    while (queue.length > 0 || pool.size > 0) {
      // Fill the pool until we reach the effective parallelism limit.
      // During the folder-meta phase this is 1 (sequential) so no item change
@@ -4273,6 +4299,20 @@ export async function push(
          }

          for await (const change of changes) {
+            // A datatable migration is one record across two files; upsert/delete
+            // it from disk once (deduped), regardless of which file changed.
+            if (isDatatableMigrationPath(change.path)) {
+              const parsed = parseDatatableMigrationPath(change.path);
+              if (parsed) {
+                const key = `${parsed.datatable}\0${parsed.timestamp}`;
+                if (!pushedMigrationKeys.has(key)) {
+                  pushedMigrationKeys.add(key);
+                  await pushMigrationFromDisk(workspace.workspaceId, parsed);
+                }
+              }
+              continue;
+            }
+
            let stateTarget = undefined;
            if (stateful) {
              try {
@@ -4928,12 +4968,12 @@ export async function push(
      log.warn(`Failed to push shared UI folder: ${e}`);
    }
    try {
-      await pushDatatableMigrations(workspace.workspaceId, {
+      await offerToRunNewMigrations(workspace.workspaceId, newDatatableMigrations, {
        yes: opts.yes,
        jsonOutput: opts.jsonOutput,
      });
    } catch (e) {
-      log.warn(`Failed to push datatable migrations folder: ${e}`);
+      log.warn(`Failed to run new datatable migrations: ${e}`);
    }
    if (opts.jsonOutput) {
      const result = {
@@ -4979,14 +5019,7 @@ export async function push(
    } catch (e) {
      log.warn(`Failed to push shared UI folder: ${e}`);
    }
-    try {
-      await pushDatatableMigrations(workspace.workspaceId, {
-        yes: opts.yes,
-        jsonOutput: opts.jsonOutput,
-      });
-    } catch (e) {
-      log.warn(`Failed to push datatable migrations folder: ${e}`);
-    }
+    // No changes pushed, so no new datatable migrations to run.
    if (opts.jsonOutput) {
      console.log(
        JSON.stringify(
@@ -288,9 +288,33 @@ export function parseFromFile(p: string): any {
    throw new Error("Could not read file " + p);
  }
 }
+/**
+ * Parse a `datatable_migrations/<datatable>/<timestamp>_<name>.(up|down).sql`
+ * path into its parts. Returns undefined for any other path.
+ */
+export function parseDatatableMigrationPath(p: string):
+  | { datatable: string; timestamp: number; name: string; kind: "up" | "down" }
+  | undefined {
+  const parts = p.split(SEP);
+  if (parts[0] !== "datatable_migrations" || parts.length !== 3) return undefined;
+  const m = parts[2].match(/^(\d+)_(.*)\.(up|down)\.sql$/);
+  if (!m) return undefined;
+  return {
+    datatable: parts[1],
+    timestamp: Number(m[1]),
+    name: m[2],
+    kind: m[3] as "up" | "down",
+  };
+}
+
+export function isDatatableMigrationPath(p: string): boolean {
+  return parseDatatableMigrationPath(p) !== undefined;
+}
+
 export function getTypeStrFromPath(
  p: string
 ):
+  | "datatable_migration"
  | "script"
  | "variable"
  | "flow"
@@ -316,6 +340,9 @@ export function getTypeStrFromPath(
  | "settings"
  | "encryption_key"
  | "workspace_dependencies" {
+  if (isDatatableMigrationPath(p)) {
+    return "datatable_migration";
+  }
  if (isScriptModulePath(p)) {
    return "script";
  }