Send recv

dbos/dbos.go

@@ -103,6 +103,8 @@ func Launch() error {
 	}
 	fmt.Println("DBOS: System database initialized")
 
+	systemDB.Launch(context.Background())
+
 	// Create context with cancel function for queue runner
 	ctx, cancel := context.WithCancel(context.Background())
 

dbos/migrations/000001_initial_dbos_schema.down.sql

@@ -1,5 +1,11 @@
 -- 001_initial_dbos_schema.down.sql
 
+-- Drop trigger first
+DROP TRIGGER IF EXISTS dbos_notifications_trigger ON dbos.notifications;
+
+-- Drop function
+DROP FUNCTION IF EXISTS dbos.notifications_function();
+
 -- Drop tables in reverse order to respect foreign key constraints
 DROP TABLE IF EXISTS dbos.workflow_events;
 DROP TABLE IF EXISTS dbos.notifications;

dbos/migrations/000001_initial_dbos_schema.up.sql

@@ -68,6 +68,21 @@ CREATE TABLE dbos.notifications (
 -- Create index for notifications
 CREATE INDEX idx_workflow_topic ON dbos.notifications (destination_uuid, topic);
 
+-- Create notification function
+CREATE OR REPLACE FUNCTION dbos.notifications_function() RETURNS TRIGGER AS $$
+DECLARE
+    payload text := NEW.destination_uuid || '::' || NEW.topic;
+BEGIN
+    PERFORM pg_notify('dbos_notifications_channel', payload);
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+-- Create notification trigger
+CREATE TRIGGER dbos_notifications_trigger
+AFTER INSERT ON dbos.notifications
+FOR EACH ROW EXECUTE FUNCTION dbos.notifications_function();
+
 -- Create workflow_events table
 CREATE TABLE dbos.workflow_events (
     workflow_uuid TEXT NOT NULL,