Add PostgreSQL Messaging Pub/Sub Watermill

cmd/server/app/serve.go

@@ -16,6 +16,7 @@
 package app
 
 import (
+	"database/sql"
 	"fmt"
 	"log"
 	"net/url"
@@ -65,13 +66,29 @@ var serveCmd = &cobra.Command{
 		if err != nil {
 			return fmt.Errorf("unable to connect to database: %w", err)
 		}
-		defer dbConn.Close()
+		defer func(dbConn *sql.DB) {
+			err := dbConn.Close()
+			if err != nil {
+				log.Printf("error closing database connection: %v", err)
+			}
+		}(dbConn)
 
 		store := db.NewStore(dbConn)
 
+		dbConnEvents, _, err := cfg.DatabaseQueue.GetDBConnection(ctx)
+		if err != nil {
+			return fmt.Errorf("unable to connect to events database: %w", err)
+		}
+		defer func(dbConnEvents *sql.DB) {
+			err := dbConnEvents.Close()
+			if err != nil {
+				log.Printf("error closing events database connection: %v", err)
+			}
+		}(dbConnEvents)
+
 		errg, ctx := errgroup.WithContext(ctx)
 
-		evt, err := events.Setup(ctx, &cfg.Events)
+		evt, err := events.Setup(ctx, &cfg.Events, dbConnEvents)
 		if err != nil {
 			log.Printf("Failed to set up eventer: %v", err)
 			return err

config/config.yaml.example

@@ -100,6 +100,6 @@ github:
     events: ["*"]
 
 events:
-  driver: go-channel
+  driver: postgresql
   router_close_timeout: 10
   go-channel: {}

deployment/helm_tests/basic.yaml-out

@@ -130,7 +130,7 @@ data:
         events: ["*"]
 
     events:
-      driver: go-channel
+      driver: postgresql
       router_close_timeout: 10
       go-channel: {}
 
@@ -738,7 +738,7 @@ data:
         events: ["*"]
 
     events:
-      driver: go-channel
+      driver: postgresql
       router_close_timeout: 10
       go-channel: {}
 

deployment/helm_tests/sidecar.yaml-out

@@ -130,7 +130,7 @@ data:
         events: ["*"]
 
     events:
-      driver: go-channel
+      driver: postgresql
       router_close_timeout: 10
       go-channel: {}
 
@@ -757,7 +757,7 @@ data:
         events: ["*"]
 
     events:
-      driver: go-channel
+      driver: postgresql
       router_close_timeout: 10
       go-channel: {}
 

docker-compose.yaml

@@ -25,6 +25,7 @@ services:
       "--http-host=0.0.0.0",
       "--metric-host=0.0.0.0",
       "--db-host=postgres",
+      "--db-events-host=postgres_queue",
       "--config=/app/config.yaml",
       # If you don't want to store your GitHub client ID and secret in the main
       # config file, point to them here:
@@ -60,6 +61,8 @@ services:
     depends_on:
       postgres:
         condition: service_healthy
+      postgres_queue:
+        condition: service_healthy  
       keycloak:
         condition: service_healthy
 #      migrate:
@@ -109,6 +112,23 @@ services:
         retries: 5
       networks:
         - app_net
+  postgres_queue:
+    container_name: postgres_queue_container
+    image: postgres:15-alpine
+    restart: always
+    environment:
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_DB: minder-queue
+    networks:
+      - app_net
+    ports:
+      - "5433:5432"  
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready"]
+      interval: 10s
+      timeout: 5s
+      retries: 5      
 
   keycloak:
     container_name: keycloak_container

go.mod

@@ -72,6 +72,7 @@ require (
 )
 
 require (
+	github.com/ThreeDotsLabs/watermill-sql/v2 v2.0.0 // indirect
 	github.com/atotto/clipboard v0.1.4 // indirect
 	github.com/aws/aws-sdk-go-v2/service/ecr v1.20.2 // indirect
 	github.com/aws/aws-sdk-go-v2/service/ecrpublic v1.18.2 // indirect

go.sum

@@ -112,6 +112,8 @@ github.com/ThalesIgnite/crypto11 v1.2.5/go.mod h1:ILDKtnCKiQ7zRoNxcp36Y1ZR8LBPmR
 github.com/ThreeDotsLabs/watermill v1.1.1/go.mod h1:Qd1xNFxolCAHCzcMrm6RnjW0manbvN+DJVWc1MWRFlI=
 github.com/ThreeDotsLabs/watermill v1.3.5 h1:50JEPEhMGZQMh08ct0tfO1PsgMOAOhV3zxK2WofkbXg=
 github.com/ThreeDotsLabs/watermill v1.3.5/go.mod h1:O/u/Ptyrk5MPTxSeWM5vzTtZcZfxXfO9PK9eXTYiFZY=
+github.com/ThreeDotsLabs/watermill-sql/v2 v2.0.0 h1:wswlLYY0Jc0tloj3lty4Y+VTEA8AM1vYfrIDwWtqyJk=
+github.com/ThreeDotsLabs/watermill-sql/v2 v2.0.0/go.mod h1:83l/4sKaLHwoHJlrAsDLaXcHN+QOHHntAAyabNmiuO4=
 github.com/acomagu/bufpipe v1.0.4 h1:e3H4WUzM3npvo5uv95QuJM3cQspFNtFBzvJ2oNjKIDQ=
 github.com/acomagu/bufpipe v1.0.4/go.mod h1:mxdxdup/WdsKVreO5GpW4+M/1CE2sMG4jeGJ2sYmHc4=
 github.com/agnivade/levenshtein v1.1.1 h1:QY8M92nrzkmr798gCo3kmMyqXFzdQVpxLlGPRBij0P8=

internal/config/config.go

@@ -36,6 +36,7 @@ type Config struct {
 	Tracing       TracingConfig      `mapstructure:"tracing"`
 	Metrics       MetricsConfig      `mapstructure:"metrics"`
 	Database      DatabaseConfig     `mapstructure:"database"`
+	DatabaseQueue DatabaseConfig     `mapstructure:"database_queue"`
 	Identity      IdentityConfig     `mapstructure:"identity"`
 	Salt          CryptoConfig       `mapstructure:"salt"`
 	Auth          AuthConfig         `mapstructure:"auth"`

internal/config/db.go

@@ -154,6 +154,39 @@
 		return err
 	}
 
-	return util.BindConfigFlagWithShort(
+	err = util.BindConfigFlagWithShort(
 		v, flags, "database.sslmode", "db-sslmode", "s", "disable", "Database sslmode", flags.StringP)
+
+	err = util.BindConfigFlagWithShort(
+		v, flags, "database_queue.dbhost", "db-events-host", "Q", "localhost", "Database Events host", flags.StringP)
+	if err != nil {
+		return err
+	}
+
+	err = util.BindConfigFlag(
+		v, flags, "database_queue.dbport", "db-events-port", 5432, "Database Events port", flags.Int)
+	if err != nil {
+		return err
+	}
+
+	err = util.BindConfigFlagWithShort(
+		v, flags, "database_queue.dbuser", "db-events-user", "U", "postgres", "Database Events user", flags.StringP)
+	if err != nil {
+		return err
+	}
+
+	err = util.BindConfigFlagWithShort(
+		v, flags, "database_queue.dbpass", "db-events-pass", "W", "postgres", "Database Events password", flags.StringP)
+	if err != nil {
+		return err
+	}
+
+	err = util.BindConfigFlagWithShort(
+		v, flags, "database_queue.dbname", "db-events-name", "N", "minder-queue", "Database Events name", flags.StringP)
+	if err != nil {
+		return err
+	}
+
+	return util.BindConfigFlagWithShort(
+		v, flags, "database_queue.sslmode", "db-events-sslmode", "M", "disable", "Database Events sslmode", flags.StringP)
 }

internal/controlplane/handlers_user_test.go

@@ -261,7 +261,7 @@ func TestCreateUser_gRPC(t *testing.T) {
 			evt, err := events.Setup(context.Background(), &config.EventConfig{
 				Driver:    "go-channel",
 				GoChannel: config.GoChannelEventConfig{},
-			})
+			}, nil)
 			require.NoError(t, err, "failed to setup eventer")
 			server, err := NewServer(mockStore, evt, NewMetrics(), &config.Config{
 				Salt: config.DefaultConfigForTest().Salt,
@@ -468,7 +468,7 @@ func TestDeleteUser_gRPC(t *testing.T) {
 			evt, err := events.Setup(context.Background(), &config.EventConfig{
 				Driver:    "go-channel",
 				GoChannel: config.GoChannelEventConfig{},
-			})
+			}, nil)
 			require.NoError(t, err, "failed to setup eventer")
 			server, err := NewServer(mockStore, evt, NewMetrics(), &config.Config{
 				Salt: config.DefaultConfigForTest().Salt,

internal/controlplane/server_test.go

@@ -73,7 +73,7 @@ func newDefaultServer(t *testing.T, mockStore *mockdb.MockStore) *Server {
 	evt, err := events.Setup(context.Background(), &config.EventConfig{
 		Driver:    "go-channel",
 		GoChannel: config.GoChannelEventConfig{},
-	})
+	}, nil)
 	require.NoError(t, err, "failed to setup eventer")
 
 	var c *config.Config

internal/eea/eea_test.go

@@ -56,7 +56,7 @@ func TestAggregator(t *testing.T) {
 			BufferSize:                     concurrentEvents,
 			BlockPublishUntilSubscriberAck: true,
 		},
-	})
+	}, nil)
 	require.NoError(t, err)
 
 	// we'll wait 2 seconds for the lock to be available

internal/engine/executor_test.go

@@ -265,7 +265,7 @@ default allow = true`,
 		GoChannel: config.GoChannelEventConfig{
 			BlockPublishUntilSubscriberAck: true,
 		},
-	})
+	}, nil)
 	require.NoError(t, err, "failed to setup eventer")
 
 	go func() {

internal/events/eventer.go

@@ -19,13 +19,15 @@ package events
 
 import (
 	"context"
+	"database/sql"
 	"errors"
 	"fmt"
 	"reflect"
 	"runtime"
 	"time"
 
 	"github.com/ThreeDotsLabs/watermill"
+	watermillsql "github.com/ThreeDotsLabs/watermill-sql/v2/pkg/sql"
 	"github.com/ThreeDotsLabs/watermill/components/metrics"
 	"github.com/ThreeDotsLabs/watermill/message"
 	"github.com/ThreeDotsLabs/watermill/message/router/middleware"
@@ -98,7 +100,7 @@ var _ message.Publisher = (*Eventer)(nil)
 
 // Setup creates an Eventer object which isolates the watermill setup code
 // TODO: pass in logger
-func Setup(ctx context.Context, cfg *config.EventConfig) (*Eventer, error) {
+func Setup(ctx context.Context, cfg *config.EventConfig, db *sql.DB) (*Eventer, error) {
 	if cfg == nil {
 		return nil, errors.New("event config is nil")
 	}
@@ -137,7 +139,7 @@ func Setup(ctx context.Context, cfg *config.EventConfig) (*Eventer, error) {
 		middleware.Recoverer,
 	)
 
-	pub, sub, err := instantiateDriver(cfg.Driver, cfg)
+	pub, sub, err := instantiateDriver(cfg.Driver, cfg, db)
 	if err != nil {
 		return nil, fmt.Errorf("failed instantiating driver: %w", err)
 	}
@@ -159,10 +161,12 @@ func Setup(ctx context.Context, cfg *config.EventConfig) (*Eventer, error) {
 	}, nil
 }
 
-func instantiateDriver(driver string, cfg *config.EventConfig) (message.Publisher, message.Subscriber, error) {
+func instantiateDriver(driver string, cfg *config.EventConfig, db *sql.DB) (message.Publisher, message.Subscriber, error) {
 	switch driver {
 	case "go-channel":
 		return buildGoChannelDriver(cfg)
+	case "postgresql":
+		return buildPostgreSQLDriver(db)
 	default:
 		return nil, nil, fmt.Errorf("unknown driver %s", driver)
 	}
@@ -177,6 +181,35 @@ func buildGoChannelDriver(cfg *config.EventConfig) (message.Publisher, message.S
 	return pubsub, pubsub, nil
 }
 
+func buildPostgreSQLDriver(db *sql.DB) (message.Publisher, message.Subscriber, error) {
+	publisher, err := watermillsql.NewPublisher(
+		db,
+		watermillsql.PublisherConfig{
+			SchemaAdapter:        watermillsql.DefaultPostgreSQLSchema{},
+			AutoInitializeSchema: true,
+		},
+		watermill.NewStdLogger(false, false),
+	)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to create SQL publisher: %w", err)
+	}
+
+	subscriber, err := watermillsql.NewSubscriber(
+		db,
+		watermillsql.SubscriberConfig{
+			SchemaAdapter:    watermillsql.DefaultPostgreSQLSchema{},
+			OffsetsAdapter:   watermillsql.DefaultPostgreSQLOffsetsAdapter{},
+			InitializeSchema: true,
+		},
+		watermill.NewStdLogger(false, false),
+	)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to create SQL subscriber: %w", err)
+	}
+
+	return publisher, subscriber, nil
+}
+
 // Close closes the router
 func (e *Eventer) Close() error {
 	//nolint:gosec // It's fine if there's an error as long as we close the router

internal/events/eventer_test.go

@@ -126,7 +126,7 @@ func TestEventer(t *testing.T) {
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
 
-			eventer, err := events.Setup(ctx, driverConfig())
+			eventer, err := events.Setup(ctx, driverConfig(), nil)
 			if err != nil {
 				t.Errorf("Setup() error = %v", err)
 				return