Send recv (#18)

First implementation of send/recv. Few things might move around in the
future (e.g., were we create the notifier connection). Also this PR does
not handle breaks on the connection checking for NOTIFY.


How it works:
- Send/Recv functions are special durable steps
- Recv, before checking in the DB if the message exists, will wait for
PG notifications for a set timeout.
- A notification listener loop is created with the system database, and
"launched" with it.

Some details:
- Use the builtin PGX `WaitForNotification` and `OnNotification`
mechanisms.
- A recv call will create a new channel and share it with the
notification listener. Upon receiving a notification, the notification
listener will check if a channel was shared for a given payload, and
write a value on the channel to signal the receiver.


About what can be sent: to send non built-in types, users must register
them with `encoding/gob` because that's what we use to serialize the
message. Usual constraints apply (e.g., no exported fields <-> not
serializable). Also note that `encoding/gob` cannot encode raw `nil`
values.

Author: maxdml

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,