feat: publish pubsub events (#1613)

- pipes module calls for publishing events through to the controller to
the db
- fake implementation for tests not included
- added TopicEventKey. It will be generally useful I think, but also
required for how db events are set up needing a "key" column

backend/controller/controller.go

@@ -817,6 +817,15 @@ func (s *Service) SendFSMEvent(ctx context.Context, req *connect.Request[ftlv1.S
 	return connect.NewResponse(&ftlv1.SendFSMEventResponse{}), nil
 }
 
+func (s *Service) PublishEvent(ctx context.Context, req *connect.Request[ftlv1.PublishEventRequest]) (*connect.Response[ftlv1.PublishEventResponse], error) {
+	// Publish the event.
+	err := s.dal.PublishEventForTopic(ctx, req.Msg.Topic.Module, req.Msg.Topic.Name, req.Msg.Body)
+	if err != nil {
+		return nil, connect.NewError(connect.CodeInternal, fmt.Errorf("failed to publish a event to topic %s:%s: %w", req.Msg.Topic.Module, req.Msg.Topic.Name, err))
+	}
+	return connect.NewResponse(&ftlv1.PublishEventResponse{}), nil
+}
+
 func (s *Service) callWithRequest(
 	ctx context.Context,
 	req *connect.Request[ftlv1.CallRequest],

backend/controller/dal/notify.go

@@ -121,6 +121,8 @@ func (d *DAL) publishNotification(ctx context.Context, notification event, logge
 
 	case "topics":
 		// TODO: handle topics notifications
+	case "topic_events":
+		// TODO: handle topic events notifications
 	default:
 		panic(fmt.Sprintf("unknown table %q in DB notification", notification.Table))
 	}

backend/controller/dal/pubsub.go

@@ -0,0 +1,21 @@
+package dal
+
+import (
+	"context"
+
+	"github.com/TBD54566975/ftl/backend/controller/sql"
+	"github.com/TBD54566975/ftl/internal/model"
+)
+
+func (d *DAL) PublishEventForTopic(ctx context.Context, module, topic string, payload []byte) error {
+	err := d.db.PublishEventForTopic(ctx, sql.PublishEventForTopicParams{
+		Key:     model.NewTopicEventKey(module, topic),
+		Module:  module,
+		Topic:   topic,
+		Payload: payload,
+	})
+	if err != nil {
+		return translatePGError(err)
+	}
+	return nil
+}

backend/controller/sql/queries.sql

@@ -659,6 +659,20 @@ VALUES (
   (SELECT id FROM deployments WHERE key = sqlc.arg('deployment')::deployment_key),
   sqlc.arg('sink')::TEXT);
 
+-- name: PublishEventForTopic :exec
+INSERT INTO topic_events ("key", topic_id, payload)
+VALUES (
+  sqlc.arg('key')::topic_event_key,
+  (
+    SELECT topics.id
+    FROM topics
+    INNER JOIN modules ON topics.module_id = modules.id
+    WHERE modules.name = sqlc.arg('module')::TEXT
+      AND topics.name = sqlc.arg('topic')::TEXT
+  ),
+  sqlc.arg('payload')
+);
+
 -- name: GetModuleConfiguration :one
 SELECT value
 FROM module_configuration

backend/controller/sql/schema/001_init.sql

@@ -324,13 +324,15 @@ CREATE TRIGGER topics_notify_event
     FOR EACH ROW
 EXECUTE PROCEDURE notify_event();
 
+CREATE DOMAIN topic_event_key AS TEXT;
+
 -- This table contains the actual topic data.
 CREATE TABLE topic_events (
     id BIGINT NOT NULL GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
     created_at TIMESTAMPTZ NOT NULL DEFAULT (NOW() AT TIME ZONE 'utc'),
 
+    "key" topic_event_key UNIQUE NOT NULL,
     topic_id BIGINT NOT NULL REFERENCES topics(id) ON DELETE CASCADE,
-
     payload BYTEA NOT NULL
 );