Add temporary Postgres consumer within Kafka to save message history

Author: Leo7Deng

backend/Makefile

@@ -4,7 +4,7 @@ create_migration:
 	migrate create -ext=sql -dir=postgres/migrations -seq init
 
 migrate_up:
-	migrate -path=postgres/migrations -database "postgresql://${PSQL_USER}:${PSQL_PASSWORD}@localhost:${PSQL_PORT}/${PSQL_DBNAME}?sslmode=disable" -verbose up
+	migrate -path=postgres/migrations -database "postgresql://${PSQL_USER}:${PSQL_PASSWORD}@${PSQL_HOST}:${PSQL_PORT}/${PSQL_DBNAME}?sslmode=disable" -verbose up
 
 migrate_down:
 	migrate -path=postgres/migrations -database "postgresql://${PSQL_USER}:${PSQL_PASSWORD}@${PSQL_HOST}:${PSQL_PORT}/${PSQL_DBNAME}?sslmode=${PSQL_SSLMODE}" -verbose down

backend/kafka/consumer.go

@@ -6,17 +6,21 @@ import (
 	"errors"
 	"fmt"
 	"log"
+	"os"
 	"sync"
 
 	"github.com/IBM/sarama"
 	"github.com/Leo7Deng/ChatApp/cassandra"
 	"github.com/Leo7Deng/ChatApp/models"
 	"github.com/Leo7Deng/ChatApp/websockets"
 	"github.com/gocql/gocql"
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgxpool"
 )
 
 var hub *websockets.Hub
 var cassandraSession *gocql.Session
+var pool *pgxpool.Pool
 
 func WebsocketConsumer(ctx context.Context, websocketHub *websockets.Hub) {
 	hub = websocketHub
@@ -32,6 +36,14 @@ func CassandraConsumer(ctx context.Context, session *gocql.Session) {
 	runConsumer(ctx, groupID, handler)
 }
 
+func PostgresConsumer(ctx context.Context, postgresPool *pgxpool.Pool) {
+	pool = postgresPool
+	groupID := "postgres-group"
+	handler := &PostgresConsumerHandler{}
+	runConsumer(ctx, groupID, handler)
+}
+
+
 func runConsumer(ctx context.Context, groupID string, handler sarama.ConsumerGroupHandler) {
 	config := sarama.NewConfig()
 	config.Consumer.Return.Errors = true
@@ -74,6 +86,7 @@ func runConsumer(ctx context.Context, groupID string, handler sarama.ConsumerGro
 
 type WebsocketConsumerHandler struct {hub *websockets.Hub}
 type CassandraConsumerHandler struct {}
+type PostgresConsumerHandler struct {}
 
 // Setup is run at the beginning of a new session, before ConsumeClaim
 func (consumer *WebsocketConsumerHandler) Setup(sarama.ConsumerGroupSession) error {
@@ -82,6 +95,9 @@ func (consumer *WebsocketConsumerHandler) Setup(sarama.ConsumerGroupSession) err
 func (consumer *CassandraConsumerHandler) Setup(sarama.ConsumerGroupSession) error {
 	return nil
 }
+func (consumer *PostgresConsumerHandler) Setup(sarama.ConsumerGroupSession) error {
+	return nil
+}
 
 // Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited
 func (consumer *WebsocketConsumerHandler) Cleanup(sarama.ConsumerGroupSession) error {
@@ -90,6 +106,9 @@ func (consumer *WebsocketConsumerHandler) Cleanup(sarama.ConsumerGroupSession) e
 func (consumer *CassandraConsumerHandler) Cleanup(sarama.ConsumerGroupSession) error {
 	return nil
 }
+func (consumer *PostgresConsumerHandler) Cleanup(sarama.ConsumerGroupSession) error {
+	return nil
+}
 
 // ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().
 // Once the Messages() channel is closed, the Handler must finish its processing
@@ -159,4 +178,57 @@ func (consumer *CassandraConsumerHandler) ConsumeClaim(session sarama.ConsumerGr
 			return nil
 		}
 	}
+}
+
+func (consumer *PostgresConsumerHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+	for {
+		select {
+		case message, ok := <-claim.Messages():
+			if !ok {
+				log.Printf("message channel was closed")
+				return nil
+			}
+			partition, offset := message.Partition, message.Offset
+			var websocketMessage models.WebsocketMessage
+			err := json.Unmarshal(message.Value, &websocketMessage)
+			if err != nil {
+				fmt.Printf("Failed to unmarshal message: %v\n", err)
+			}
+			log.Printf("Postgres consumer: %s | Partition: %d | Offset: %d\n", message.Value, partition, offset)
+			type InsertMessage struct {
+				MessageID string `json:"message_id"`
+				Message models.Message `json:"message"`
+			}
+			insertMessage := InsertMessage{
+				MessageID: uuid.New().String(),
+				Message: models.Message{
+					CircleID: websocketMessage.Message.CircleID,
+					AuthorID: websocketMessage.Message.AuthorID,
+					Content:  websocketMessage.Message.Content,
+					CreatedAt: websocketMessage.Message.CreatedAt,
+				},
+			}
+			conn, err := pool.Acquire(session.Context())
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "Unable to acquire a connection from the pool: %v\n", err)
+				return err
+			}
+			defer conn.Release()
+
+			_, err = conn.Exec(
+				session.Context(),
+				"INSERT INTO messages (message_id, circle_id, author_id, content) VALUES ($1, $2, $3, $4)",
+				insertMessage.MessageID,
+				insertMessage.Message.CircleID,
+				insertMessage.Message.AuthorID,
+				insertMessage.Message.Content,
+			)
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "Error inserting into messages: %v\n", err)
+				return err
+			}
+		case <-session.Context().Done():
+			return nil
+		}
+	}
 }

backend/kafka/kafka.go

@@ -3,19 +3,26 @@ package kafka
 import (
 	"context"
 	"sync"
+
 	"github.com/Leo7Deng/ChatApp/websockets"
 	"github.com/gocql/gocql"
+	"github.com/jackc/pgx/v5/pgxpool"
 )
 
 // InitKafka starts the Kafka producer and consumer as goroutines
-func InitKafka(ctx context.Context, wg *sync.WaitGroup, hub *websockets.Hub, cassandraSession *gocql.Session) {
+func InitKafka(ctx context.Context, wg *sync.WaitGroup, hub *websockets.Hub, cassandraSession *gocql.Session, pool *pgxpool.Pool) {
 	wg.Add(2)
 
 	go func() {
 		defer wg.Done()
 		WebsocketConsumer(ctx, hub) 
 	}()
 
+	go func() {
+		defer wg.Done()
+		PostgresConsumer(ctx, pool)
+	}()
+
 	go func() {
 		defer wg.Done()
 		CassandraConsumer(ctx, cassandraSession)

backend/main.go

@@ -55,6 +55,8 @@ func main() {
 
 	postgres.ConnectPSQL()
 	defer postgres.ClosePSQL()
+	pool := postgres.GetPool()
+	defer pool.Close()
 
 	cassandraSession := cassandra.CassandraSession()
 	defer cassandraSession.Close()
@@ -63,7 +65,7 @@ func main() {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	var wg sync.WaitGroup
-	kafka.InitKafka(ctx, &wg, hub, cassandraSession)
+	kafka.InitKafka(ctx, &wg, hub, cassandraSession, pool)
 
 	go hub.Run() 
 	fmt.Println("Websocket server started", hub)

backend/postgres/client.go

@@ -11,11 +11,10 @@ import (
 var pool *pgxpool.Pool
 
 func ConnectPSQL() {
-	db_url := fmt.Sprintf("postgres://%s:%s@%s:%s/%s?sslmode=%s",
+	db_url := fmt.Sprintf("postgres://%s:%s@postgres:%s/%s?sslmode=%s",
 		os.Getenv("PSQL_USER"),
 		// os.Getenv("PSQL_PASSWORD"),
 		url.QueryEscape(os.Getenv("PSQL_PASSWORD")),
-		os.Getenv("PSQL_HOST"),
 		os.Getenv("PSQL_PORT"),
 		os.Getenv("PSQL_DBNAME"),
 		os.Getenv("PSQL_SSLMODE"),

backend/postgres/migrations/000004_init.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS messages;

backend/postgres/migrations/000004_init.up.sql

@@ -0,0 +1,7 @@
+CREATE TABLE messages (
+    message_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    circle_id INT NOT NULL,
+    author_id INT NOT NULL,
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    content TEXT NOT NULL
+);