World postgres drizzle migrator (#312)

* Use drizzle migrator

Signed-off-by: paulhenri-l <[email protected]>

* Changeset

Signed-off-by: paulhenri-l <[email protected]>

---------

Signed-off-by: paulhenri-l <[email protected]>

Author: paulhenri-l

.changeset/fluffy-peaches-buy.md

@@ -0,0 +1,5 @@
+---
+"@workflow/world-postgres": patch
+---
+
+Use drizzle migrator

packages/world-postgres/DEV_NOTES.md

@@ -0,0 +1,7 @@
+# Generate migrations
+
+The migrations are generated and managed by drizzle. When you perform schema changes you have to generate new migrations using the following command:
+
+```
+pnpm drizzle-kit generate --dialect=postgresql --schema=./src/drizzle/schema.ts --out src/drizzle/migrations
+```

packages/world-postgres/package.json

@@ -50,7 +50,7 @@
     "@workflow/world": "workspace:*",
     "@workflow/world-local": "workspace:*",
     "dotenv": "^16.4.5",
-    "drizzle-orm": "^0.31.2",
+    "drizzle-orm": "^0.44.7",
     "pg-boss": "^11.0.7",
     "postgres": "^3.4.7",
     "ulid": "^3.0.1",
@@ -62,7 +62,7 @@
     "@workflow/errors": "workspace:*",
     "@workflow/tsconfig": "workspace:*",
     "@workflow/world-testing": "workspace:*",
-    "drizzle-kit": "^0.22.7",
+    "drizzle-kit": "^0.31.6",
     "vitest": "catalog:"
   },
   "keywords": [],

packages/world-postgres/src/cli.ts

@@ -1,7 +1,8 @@
-import { readFile } from 'node:fs/promises';
 import { dirname, join } from 'node:path';
 import { fileURLToPath } from 'node:url';
 import { config } from 'dotenv';
+import { drizzle } from 'drizzle-orm/postgres-js';
+import { migrate } from 'drizzle-orm/postgres-js/migrator';
 import postgres from 'postgres';
 
 const __dirname = dirname(fileURLToPath(import.meta.url));
@@ -21,33 +22,31 @@ async function setupDatabase() {
   );
 
   try {
-    const sql = postgres(connectionString);
+    const pgClient = postgres(connectionString, { max: 1 });
+    const db = drizzle(pgClient);
 
     // Read the migration SQL file
     // The migrations are in src/drizzle/migrations, and this CLI is in dist/
     // So we need to go up one level from dist/ to reach src/
-    const migrationPath = join(
+    const migrationsFolder = join(
       __dirname,
       '..',
       'src',
       'drizzle',
-      'migrations',
-      '0000_redundant_smasher.sql'
+      'migrations'
     );
-    const migrationSQL = await readFile(migrationPath, 'utf-8');
+    console.log(`📂 Running migrations from: ${migrationsFolder}`);
 
     // Execute the migration
-    await sql.unsafe(migrationSQL);
+    await migrate(db, {
+      migrationsFolder,
+      migrationsTable: 'workflow_migrations',
+      migrationsSchema: 'workflow_drizzle',
+    });
 
     console.log('✅ Database schema created successfully!');
-    console.log('\nCreated tables:');
-    console.log('  - workflow_runs');
-    console.log('  - workflow_events');
-    console.log('  - workflow_steps');
-    console.log('  - workflow_hooks');
-    console.log('  - workflow_stream_chunks');
-
-    await sql.end();
+
+    await pgClient.end();
     process.exit(0);
   } catch (error) {
     console.error('❌ Failed to setup database:', error);

packages/world-postgres/src/drizzle/index.ts

@@ -5,6 +5,7 @@ import * as Schema from './schema.js';
 export { Schema };
 
 export type Drizzle = ReturnType<typeof createClient>;
+
 export function createClient(postgres: Postgres.Sql) {
   return drizzle(postgres, { schema: Schema });
 }

packages/world-postgres/src/drizzle/migrations/0000_redundant_smasher.sql renamed to packages/world-postgres/src/drizzle/migrations/0000_cultured_the_anarchist.sql

@@ -1,3 +1,5 @@
+CREATE SCHEMA "workflow";
+--> statement-breakpoint
 DO $$ BEGIN
  CREATE TYPE "public"."step_status" AS ENUM('pending', 'running', 'completed', 'failed', 'cancelled');
 EXCEPTION
@@ -10,7 +12,7 @@ EXCEPTION
  WHEN duplicate_object THEN null;
 END $$;
 --> statement-breakpoint
-CREATE TABLE IF NOT EXISTS "workflow_events" (
+CREATE TABLE IF NOT EXISTS "workflow"."workflow_events" (
 	"id" varchar PRIMARY KEY NOT NULL,
 	"type" varchar NOT NULL,
 	"correlation_id" varchar,
@@ -19,7 +21,7 @@ CREATE TABLE IF NOT EXISTS "workflow_events" (
 	"payload" jsonb
 );
 --> statement-breakpoint
-CREATE TABLE IF NOT EXISTS "workflow_hooks" (
+CREATE TABLE IF NOT EXISTS "workflow"."workflow_hooks" (
 	"run_id" varchar NOT NULL,
 	"hook_id" varchar PRIMARY KEY NOT NULL,
 	"token" varchar NOT NULL,
@@ -30,7 +32,7 @@ CREATE TABLE IF NOT EXISTS "workflow_hooks" (
 	"metadata" jsonb
 );
 --> statement-breakpoint
-CREATE TABLE IF NOT EXISTS "workflow_runs" (
+CREATE TABLE IF NOT EXISTS "workflow"."workflow_runs" (
 	"id" varchar PRIMARY KEY NOT NULL,
 	"output" jsonb,
 	"deployment_id" varchar NOT NULL,
@@ -46,7 +48,7 @@ CREATE TABLE IF NOT EXISTS "workflow_runs" (
 	"started_at" timestamp
 );
 --> statement-breakpoint
-CREATE TABLE IF NOT EXISTS "workflow_steps" (
+CREATE TABLE IF NOT EXISTS "workflow"."workflow_steps" (
 	"run_id" varchar NOT NULL,
 	"step_id" varchar PRIMARY KEY NOT NULL,
 	"step_name" varchar NOT NULL,
@@ -63,7 +65,7 @@ CREATE TABLE IF NOT EXISTS "workflow_steps" (
 	"retry_after" timestamp
 );
 --> statement-breakpoint
-CREATE TABLE IF NOT EXISTS "workflow_stream_chunks" (
+CREATE TABLE IF NOT EXISTS "workflow"."workflow_stream_chunks" (
 	"id" varchar NOT NULL,
 	"stream_id" varchar NOT NULL,
 	"data" "bytea" NOT NULL,
@@ -72,11 +74,11 @@ CREATE TABLE IF NOT EXISTS "workflow_stream_chunks" (
 	CONSTRAINT "workflow_stream_chunks_stream_id_id_pk" PRIMARY KEY("stream_id","id")
 );
 --> statement-breakpoint
-CREATE INDEX IF NOT EXISTS "workflow_events_run_id_index" ON "workflow_events" USING btree ("run_id");--> statement-breakpoint
-CREATE INDEX IF NOT EXISTS "workflow_events_correlation_id_index" ON "workflow_events" USING btree ("correlation_id");--> statement-breakpoint
-CREATE INDEX IF NOT EXISTS "workflow_hooks_run_id_index" ON "workflow_hooks" USING btree ("run_id");--> statement-breakpoint
-CREATE INDEX IF NOT EXISTS "workflow_hooks_token_index" ON "workflow_hooks" USING btree ("token");--> statement-breakpoint
-CREATE INDEX IF NOT EXISTS "workflow_runs_name_index" ON "workflow_runs" USING btree ("name");--> statement-breakpoint
-CREATE INDEX IF NOT EXISTS "workflow_runs_status_index" ON "workflow_runs" USING btree ("status");--> statement-breakpoint
-CREATE INDEX IF NOT EXISTS "workflow_steps_run_id_index" ON "workflow_steps" USING btree ("run_id");--> statement-breakpoint
-CREATE INDEX IF NOT EXISTS "workflow_steps_status_index" ON "workflow_steps" USING btree ("status");
+CREATE INDEX IF NOT EXISTS "workflow_events_run_id_index" ON "workflow"."workflow_events" USING btree ("run_id");--> statement-breakpoint
+CREATE INDEX IF NOT EXISTS "workflow_events_correlation_id_index" ON "workflow"."workflow_events" USING btree ("correlation_id");--> statement-breakpoint
+CREATE INDEX IF NOT EXISTS "workflow_hooks_run_id_index" ON "workflow"."workflow_hooks" USING btree ("run_id");--> statement-breakpoint
+CREATE INDEX IF NOT EXISTS "workflow_hooks_token_index" ON "workflow"."workflow_hooks" USING btree ("token");--> statement-breakpoint
+CREATE INDEX IF NOT EXISTS "workflow_runs_name_index" ON "workflow"."workflow_runs" USING btree ("name");--> statement-breakpoint
+CREATE INDEX IF NOT EXISTS "workflow_runs_status_index" ON "workflow"."workflow_runs" USING btree ("status");--> statement-breakpoint
+CREATE INDEX IF NOT EXISTS "workflow_steps_run_id_index" ON "workflow"."workflow_steps" USING btree ("run_id");--> statement-breakpoint
+CREATE INDEX IF NOT EXISTS "workflow_steps_status_index" ON "workflow"."workflow_steps" USING btree ("status");