add migration installer API endpoints for no-CLI platforms

Author: jumski

pkgs/edge-worker/deno.lock

@@ -134,6 +134,16 @@
         "parallel": false
       }
     },
+    "db-clean:ensure": {
+      "executor": "nx:run-commands",
+      "local": true,
+      "cache": false,
+      "options": {
+        "cwd": "pkgs/edge-worker",
+        "commands": ["./scripts/ensure-db-clean"],
+        "parallel": false
+      }
+    },
     "test:unit": {
       "dependsOn": ["^build"],
       "executor": "nx:run-commands",
@@ -160,6 +170,19 @@
         "parallel": false
       }
     },
+    "test:migrations": {
+      "dependsOn": ["db-clean:ensure", "^build"],
+      "executor": "nx:run-commands",
+      "local": true,
+      "inputs": ["default", "^production"],
+      "options": {
+        "cwd": "pkgs/edge-worker",
+        "commands": [
+          "deno test --config deno.test.json --allow-all --env=supabase/functions/.env tests/migrations/"
+        ],
+        "parallel": false
+      }
+    },
     "serve:functions:e2e": {
       "executor": "nx:run-commands",
       "dependsOn": ["^build"],

pkgs/edge-worker/project.json

@@ -0,0 +1,20 @@
+#!/bin/bash
+set -e
+
+# Clean up any existing containers
+echo "Shutting down existing clean database containers..."
+docker compose -f tests/db-clean/compose.yaml down --volumes --remove-orphans
+
+# Start fresh containers
+echo "Starting clean database..."
+docker compose -f tests/db-clean/compose.yaml up --detach
+
+# Wait for database to be ready
+echo "Waiting for database to be available..."
+./scripts/wait-for-localhost 5433
+
+# Additional pause to ensure database is fully initialized
+echo "Waiting for database initialization..."
+sleep 3
+
+echo "Clean database is ready on port 5433!"

pkgs/edge-worker/scripts/ensure-db-clean

@@ -0,0 +1,127 @@
+/**
+ * MigrationRunner - Applies pgflow migrations to a database
+ *
+ * This class handles:
+ * - Creating the pgflow_installer schema and tracking table
+ * - Listing migration status (pending/applied)
+ * - Applying pending migrations with advisory locking
+ */
+
+import type postgres from 'postgres';
+import { MIGRATIONS } from './loader.ts';
+import type { MigrationWithStatus, ApplyResult, MigrationApplyResult } from './types.ts';
+
+// Advisory lock key for migration serialization
+// Using a fixed key to ensure only one migration runs at a time
+const MIGRATION_LOCK_KEY = 0x706766_6c6f77; // 'pgflow' in hex
+
+export class MigrationRunner {
+  constructor(private sql: postgres.Sql) {}
+
+  /**
+   * Lists all migrations with their current status (pending/applied)
+   */
+  async list(): Promise<MigrationWithStatus[]> {
+    await this.ensureInstallerSchema();
+    const appliedSet = await this.getAppliedTimestamps();
+
+    return MIGRATIONS.map((m) => ({
+      timestamp: m.timestamp,
+      filename: m.filename,
+      status: appliedSet.has(m.timestamp) ? 'applied' : 'pending',
+    }));
+  }
+
+  /**
+   * Applies all pending migrations
+   * Each migration runs in its own transaction with advisory locking
+   */
+  async up(): Promise<ApplyResult> {
+    await this.ensureInstallerSchema();
+    const appliedSet = await this.getAppliedTimestamps();
+
+    const results: MigrationApplyResult[] = [];
+    let applied = 0;
+    let skipped = 0;
+
+    for (const migration of MIGRATIONS) {
+      if (appliedSet.has(migration.timestamp)) {
+        results.push({ timestamp: migration.timestamp, status: 'skipped' });
+        skipped++;
+        continue;
+      }
+
+      try {
+        // Run each migration in its own transaction with advisory lock
+        await this.sql.begin(async (tx) => {
+          // Acquire advisory lock (released at transaction end)
+          await tx`SELECT pg_advisory_xact_lock(${MIGRATION_LOCK_KEY})`;
+
+          // Double-check migration hasn't been applied (race condition protection)
+          const existing = await tx`
+            SELECT 1 FROM pgflow_installer.migrations
+            WHERE timestamp = ${migration.timestamp}
+          `;
+
+          if (existing.length > 0) {
+            // Another process applied it while we were waiting for lock
+            return;
+          }
+
+          // Execute migration SQL
+          await tx.unsafe(migration.content);
+
+          // Record that migration was applied
+          await tx`
+            INSERT INTO pgflow_installer.migrations (timestamp)
+            VALUES (${migration.timestamp})
+          `;
+        });
+
+        results.push({ timestamp: migration.timestamp, status: 'applied' });
+        applied++;
+      } catch (error) {
+        // Migration failed - return partial results with error
+        return {
+          success: false,
+          applied,
+          skipped,
+          total: MIGRATIONS.length,
+          results,
+          error: `Failed at ${migration.timestamp}: ${error instanceof Error ? error.message : 'Unknown error'}`,
+        };
+      }
+    }
+
+    return {
+      success: true,
+      applied,
+      skipped,
+      total: MIGRATIONS.length,
+      results,
+    };
+  }
+
+  /**
+   * Ensures the installer schema and tracking table exist
+   */
+  private async ensureInstallerSchema(): Promise<void> {
+    await this.sql`CREATE SCHEMA IF NOT EXISTS pgflow_installer`;
+    await this.sql`
+      CREATE TABLE IF NOT EXISTS pgflow_installer.migrations (
+        timestamp TEXT PRIMARY KEY,
+        applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
+      )
+    `;
+  }
+
+  /**
+   * Gets the set of already-applied migration timestamps
+   */
+  private async getAppliedTimestamps(): Promise<Set<string>> {
+    const rows = await this.sql<{ timestamp: string }[]>`
+      SELECT timestamp FROM pgflow_installer.migrations
+    `;
+    return new Set(rows.map((r) => r.timestamp));
+  }
+}

pkgs/edge-worker/src/control-plane/migrations/MigrationRunner.ts

@@ -0,0 +1,9 @@
+export { MigrationRunner } from './MigrationRunner.ts';
+export { MIGRATIONS, MIGRATIONS_BY_TIMESTAMP } from './loader.ts';
+export type {
+  Migration,
+  MigrationStatus,
+  MigrationWithStatus,
+  MigrationApplyResult,
+  ApplyResult,
+} from './types.ts';

pkgs/edge-worker/src/control-plane/migrations/index.ts

@@ -0,0 +1,60 @@
+/**
+ * Migration Loader - Deno-specific
+ *
+ * Uses Deno's sync filesystem APIs at module initialization time to load
+ * SQL migration files from @pgflow/core package.
+ *
+ * IMPORTANT: These sync APIs (readDirSync, readTextFileSync) are ONLY available
+ * during module initialization (top-level code). They are NOT available in HTTP
+ * handlers or callbacks. This code runs at import time, which is allowed.
+ *
+ * From Supabase docs:
+ * > "You can safely use the following synchronous Deno APIs during initial script evaluation:
+ * > - Deno.readDirSync, Deno.readTextFileSync, Deno.readFileSync"
+ *
+ * @see https://supabase.com/docs/guides/functions/ephemeral-storage
+ */
+
+import type { Migration } from './types.ts';
+
+// Resolve path to @pgflow/core package entry point
+const pkgEntryPath = import.meta.resolve('@pgflow/core');
+
+// Handle both development and production paths:
+// - Dev/test: pkgEntryPath = .../core/src/index.ts -> migrations at .../core/supabase/migrations/
+// - Production: pkgEntryPath = .../dist/index.js -> migrations at .../dist/supabase/migrations/
+const isDevPath = pkgEntryPath.includes('/src/index.ts');
+const migrationsDir = isDevPath
+  ? new URL('../supabase/migrations/', pkgEntryPath)
+  : new URL('./supabase/migrations/', pkgEntryPath);
+
+/**
+ * All migrations loaded from @pgflow/core, sorted by timestamp
+ */
+export const MIGRATIONS: Migration[] = [];
+
+// Load migrations at module initialization time
+for (const entry of Deno.readDirSync(migrationsDir)) {
+  if (entry.isFile && entry.name.endsWith('.sql') && !entry.name.startsWith('atlas')) {
+    // Extract 14-digit timestamp from filename (e.g., "20250429164909_...")
+    const match = entry.name.match(/^(\d{14})_/);
+    if (match) {
+      const content = Deno.readTextFileSync(new URL(entry.name, migrationsDir));
+      MIGRATIONS.push({
+        timestamp: match[1],
+        filename: entry.name,
+        content,
+      });
+    }
+  }
+}
+
+// Sort by timestamp (ascending)
+MIGRATIONS.sort((a, b) => a.timestamp.localeCompare(b.timestamp));
+
+/**
+ * Migrations keyed by timestamp for O(1) lookup
+ */
+export const MIGRATIONS_BY_TIMESTAMP: Record<string, Migration> = Object.fromEntries(
+  MIGRATIONS.map((m) => [m.timestamp, m])
+);