refact: pubsub: use postgres listen/notify (#268)

database/migrations/022_events.up.sql

@@ -0,0 +1,13 @@
+CREATE TYPE getstronger.event_topic AS ENUM (
+    'FollowedUser',
+    'RequestTraced',
+    'WorkoutCommentPosted'
+);
+
+CREATE TABLE getstronger.events
+(
+    id         UUID PRIMARY KEY        NOT NUll DEFAULT uuid_generate_v4(),
+    topic      getstronger.event_topic NOT NULL,
+    payload    JSONB                   NOT NULL,
+    created_at TIMESTAMP               NOT NULL DEFAULT (NOW() AT TIME ZONE 'UTC')
+);

go.mod

@@ -16,6 +16,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/jackc/pgx/v5 v5.7.1
 	github.com/joho/godotenv v1.5.1
+	github.com/lib/pq v1.10.9
 	github.com/rs/cors v1.11.1
 	github.com/stretchr/testify v1.10.0
 	github.com/testcontainers/testcontainers-go v0.34.0
@@ -24,6 +25,7 @@ require (
 	github.com/volatiletech/sqlboiler/v4 v4.17.1
 	github.com/volatiletech/strmangle v0.0.8
 	go.uber.org/fx v1.23.0
+	go.uber.org/mock v0.5.0
 	go.uber.org/zap v1.27.0
 	golang.org/x/crypto v0.31.0
 	golang.org/x/net v0.33.0
@@ -70,7 +72,6 @@ require (
 	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/klauspost/compress v1.17.4 // indirect
-	github.com/lib/pq v1.10.9 // indirect
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
 	github.com/magiconair/properties v1.8.7 // indirect
 	github.com/moby/docker-image-spec v1.3.1 // indirect

go.sum

@@ -631,6 +631,8 @@ go.uber.org/fx v1.23.0 h1:lIr/gYWQGfTwGcSXWXu4vP5Ws6iqnNEIY+F/aFzCKTg=
 go.uber.org/fx v1.23.0/go.mod h1:o/D9n+2mLP6v1EG+qsdT1O8wKopYAsqZasju97SDFCU=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+go.uber.org/mock v0.5.0 h1:KAMbZvZPyBPWgD14IrIQ38QCyjwpvVVV6K/bHl1IwQU=
+go.uber.org/mock v0.5.0/go.mod h1:ge71pBPLYDk7QIi1LupWxdAykm7KIEFchiOqd6z7qMM=
 go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
 go.uber.org/multierr v1.10.0 h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ=
 go.uber.org/multierr v1.10.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=

server/db/db.go

@@ -11,9 +11,13 @@ import (
 )
 
 func New(c *config.Config) (*sql.DB, error) {
-	db, err := sql.Open("pgx", fmt.Sprintf("postgresql://%s:%s@%s/%s", c.DB.User, c.DB.Password, net.JoinHostPort(c.DB.Host, c.DB.Port), c.DB.Name))
+	db, err := sql.Open("pgx", ConnectionString(c))
 	if err != nil {
 		return nil, fmt.Errorf("open db: %w", err)
 	}
 	return db, nil
 }
+
+func ConnectionString(c *config.Config) string {
+	return fmt.Sprintf("postgresql://%s:%s@%s/%s?sslmode=disable", c.DB.User, c.DB.Password, net.JoinHostPort(c.DB.Host, c.DB.Port), c.DB.Name)
+}

server/db/module.go

@@ -3,13 +3,22 @@ package db
 import (
 	"context"
 	"database/sql"
+	"time"
 
+	"github.com/lib/pq"
 	"go.uber.org/fx"
+
+	"github.com/crlssn/getstronger/server/config"
 )
 
 func Module() fx.Option {
 	return fx.Module("db", fx.Options(
-		fx.Provide(New),
+		fx.Provide(
+			New,
+			func(c *config.Config) *pq.Listener {
+				return pq.NewListener(ConnectionString(c), time.Second, time.Minute, nil)
+			},
+		),
 		fx.Invoke(func(l fx.Lifecycle, db *sql.DB) {
 			l.Append(fx.Hook{
 				OnStart: func(ctx context.Context) error {