migration match python

Author: maxdml

dbos/migrations/1_initial_dbos_schema.sql

@@ -1,19 +1,14 @@
--- 001_initial_dbos_schema.up.sql
-
--- Create the dbos schema
-CREATE SCHEMA IF NOT EXISTS dbos;
-
 -- Enable uuid extension for generating UUIDs
 CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
 
--- Create workflow_status table
-CREATE TABLE IF NOT EXISTS dbos.workflow_status (
+CREATE TABLE dbos.workflow_status (
     workflow_uuid TEXT PRIMARY KEY,
     status TEXT,
     name TEXT,
     authenticated_user TEXT,
     assumed_role TEXT,
     authenticated_roles TEXT,
+    request TEXT,
     output TEXT,
     error TEXT,
     executor_id TEXT,
@@ -27,34 +22,21 @@ CREATE TABLE IF NOT EXISTS dbos.workflow_status (
     queue_name TEXT,
     workflow_timeout_ms BIGINT,
     workflow_deadline_epoch_ms BIGINT,
+    inputs TEXT,
     started_at_epoch_ms BIGINT,
     deduplication_id TEXT,
-    inputs TEXT,
     priority INTEGER NOT NULL DEFAULT 0
 );
 
--- Create indexes for workflow_status
-CREATE INDEX IF NOT EXISTS workflow_status_created_at_index ON dbos.workflow_status (created_at);
-CREATE INDEX IF NOT EXISTS workflow_status_executor_id_index ON dbos.workflow_status (executor_id);
-CREATE INDEX IF NOT EXISTS workflow_status_status_index ON dbos.workflow_status (status);
+CREATE INDEX workflow_status_created_at_index ON dbos.workflow_status (created_at);
+CREATE INDEX workflow_status_executor_id_index ON dbos.workflow_status (executor_id);
+CREATE INDEX workflow_status_status_index ON dbos.workflow_status (status);
 
--- Create unique constraint for queue_name and deduplication_id
-DO $$
-BEGIN
-    IF NOT EXISTS (
-        SELECT 1 FROM information_schema.table_constraints
-        WHERE constraint_name = 'uq_workflow_status_queue_name_dedup_id'
-        AND table_name = 'workflow_status'
-        AND table_schema = 'dbos'
-    ) THEN
-        ALTER TABLE dbos.workflow_status
-        ADD CONSTRAINT uq_workflow_status_queue_name_dedup_id
-        UNIQUE (queue_name, deduplication_id);
-    END IF;
-END $$;
+ALTER TABLE dbos.workflow_status
+ADD CONSTRAINT uq_workflow_status_queue_name_dedup_id
+UNIQUE (queue_name, deduplication_id);
 
--- Create operation_outputs table
-CREATE TABLE IF NOT EXISTS dbos.operation_outputs (
+CREATE TABLE dbos.operation_outputs (
     workflow_uuid TEXT NOT NULL,
     function_id INTEGER NOT NULL,
     function_name TEXT NOT NULL DEFAULT '',
@@ -66,7 +48,7 @@ CREATE TABLE IF NOT EXISTS dbos.operation_outputs (
         ON UPDATE CASCADE ON DELETE CASCADE
 );
 
-CREATE TABLE IF NOT EXISTS dbos.notifications (
+CREATE TABLE dbos.notifications (
     destination_uuid TEXT NOT NULL,
     topic TEXT,
     message TEXT NOT NULL,
@@ -75,8 +57,7 @@ CREATE TABLE IF NOT EXISTS dbos.notifications (
     FOREIGN KEY (destination_uuid) REFERENCES dbos.workflow_status(workflow_uuid) 
         ON UPDATE CASCADE ON DELETE CASCADE
 );
--- Create index for notifications
-CREATE INDEX IF NOT EXISTS idx_workflow_topic ON dbos.notifications (destination_uuid, topic);
+CREATE INDEX idx_workflow_topic ON dbos.notifications (destination_uuid, topic);
 
 -- Create notification function
 CREATE OR REPLACE FUNCTION dbos.notifications_function() RETURNS TRIGGER AS $$
@@ -89,22 +70,11 @@ END;
 $$ LANGUAGE plpgsql;
 
 -- Create notification trigger
-DO $$
-BEGIN
-    IF NOT EXISTS (
-        SELECT 1 FROM information_schema.triggers
-        WHERE trigger_name = 'dbos_notifications_trigger'
-        AND event_object_table = 'notifications'
-        AND event_object_schema = 'dbos'
-    ) THEN
-        CREATE TRIGGER dbos_notifications_trigger
-        AFTER INSERT ON dbos.notifications
-        FOR EACH ROW EXECUTE FUNCTION dbos.notifications_function();
-    END IF;
-END $$;
+CREATE TRIGGER dbos_notifications_trigger
+AFTER INSERT ON dbos.notifications
+FOR EACH ROW EXECUTE FUNCTION dbos.notifications_function();
 
--- Create workflow_events table
-CREATE TABLE IF NOT EXISTS dbos.workflow_events (
+CREATE TABLE dbos.workflow_events (
     workflow_uuid TEXT NOT NULL,
     key TEXT NOT NULL,
     value TEXT NOT NULL,
@@ -124,16 +94,26 @@ END;
 $$ LANGUAGE plpgsql;
 
 -- Create events trigger
-DO $$
-BEGIN
-    IF NOT EXISTS (
-        SELECT 1 FROM information_schema.triggers
-        WHERE trigger_name = 'dbos_workflow_events_trigger'
-        AND event_object_table = 'workflow_events'
-        AND event_object_schema = 'dbos'
-    ) THEN
-        CREATE TRIGGER dbos_workflow_events_trigger
-        AFTER INSERT ON dbos.workflow_events
-        FOR EACH ROW EXECUTE FUNCTION dbos.workflow_events_function();
-    END IF;
-END $$;
+CREATE TRIGGER dbos_workflow_events_trigger
+AFTER INSERT ON dbos.workflow_events
+FOR EACH ROW EXECUTE FUNCTION dbos.workflow_events_function();
+
+CREATE TABLE dbos.streams (
+    workflow_uuid TEXT NOT NULL,
+    key TEXT NOT NULL,
+    value TEXT NOT NULL,
+    "offset" INTEGER NOT NULL,
+    PRIMARY KEY (workflow_uuid, key, "offset"),
+    FOREIGN KEY (workflow_uuid) REFERENCES dbos.workflow_status(workflow_uuid)
+        ON UPDATE CASCADE ON DELETE CASCADE
+);
+
+CREATE TABLE dbos.event_dispatch_kv (
+    service_name TEXT NOT NULL,
+    workflow_fn_name TEXT NOT NULL,
+    key TEXT NOT NULL,
+    value TEXT,
+    update_seq NUMERIC(38,0),
+    update_time NUMERIC(38,15),
+    PRIMARY KEY (service_name, workflow_fn_name, key)
+);