first commit

internal/config/config.go

@@ -0,0 +1,48 @@
+package config
+
+import (
+	"log"
+	"os"
+	"strings"
+
+	"github.com/joho/godotenv"
+)
+
+type Config struct {
+	DBHost     string
+	DBPort     string
+	DBUser     string
+	DBPassword string
+	DBName     string
+	DBSSLMode  string
+	Topics     []string
+	LogDebug   bool
+}
+
+func Load() *Config {
+	_ = godotenv.Load()
+
+	return &Config{
+		DBHost:     getEnv("DB_HOST", "localhost"),
+		DBPort:     getEnv("DB_PORT", "5432"),
+		DBUser:     getEnv("DB_USER", "postgres"),
+		DBPassword: getEnv("DB_PASSWORD", "postgres"),
+		DBName:     getEnv("DB_NAME", "middleware"),
+		DBSSLMode:  getEnv("DB_SSLMODE", "disable"),
+		Topics:     strings.Split(getEnv("WATERMILL_TOPICS", "example_topic"), ","),
+		LogDebug:   getEnv("LOG_DEBUG", "false") == "true",
+	}
+}
+
+func getEnv(key, def string) string {
+	if val, ok := os.LookupEnv(key); ok {
+		return val
+	}
+	return def
+}
+
+func Must(err error) {
+	if err != nil {
+		log.Fatalf("❌ %v", err)
+	}
+}

internal/db/postgres.go

@@ -0,0 +1,16 @@
+package db
+
+import (
+	"database/sql"
+	"fmt"
+
+	_ "github.com/lib/pq"
+)
+
+func Connect(host, port, user, password, dbname, sslmode string) (*sql.DB, error) {
+	dsn := fmt.Sprintf(
+		"host=%s port=%s user=%s password=%s dbname=%s sslmode=%s",
+		host, port, user, password, dbname, sslmode,
+	)
+	return sql.Open("postgres", dsn)
+}

internal/pubsub/publisher.go

@@ -0,0 +1,19 @@
+package pubsub
+
+import (
+	"database/sql"
+
+	"github.com/ThreeDotsLabs/watermill"
+	sqlwatermill "github.com/ThreeDotsLabs/watermill-sql/pkg/sql"
+	"github.com/ThreeDotsLabs/watermill/message"
+)
+
+func NewPublisher(db *sql.DB, logger watermill.LoggerAdapter) (message.Publisher, error) {
+	return sqlwatermill.NewPublisher(
+		db,
+		sqlwatermill.PublisherConfig{
+			SchemaAdapter: sqlwatermill.DefaultPostgreSQLSchema{},
+		},
+		logger,
+	)
+}

internal/pubsub/subscriber.go

@@ -0,0 +1,23 @@
+package pubsub
+
+import (
+	"time"
+
+	"database/sql"
+
+	"github.com/ThreeDotsLabs/watermill"
+	sqlwatermill "github.com/ThreeDotsLabs/watermill-sql/pkg/sql"
+	"github.com/ThreeDotsLabs/watermill/message"
+)
+
+func NewSubscriber(db *sql.DB, logger watermill.LoggerAdapter) (message.Subscriber, error) {
+	return sqlwatermill.NewSubscriber(
+		db,
+		sqlwatermill.SubscriberConfig{
+			SchemaAdapter:  sqlwatermill.DefaultPostgreSQLSchema{},
+			OffsetsAdapter: sqlwatermill.DefaultPostgreSQLOffsetsAdapter{},
+			PollInterval:   3 * time.Second,
+		},
+		logger,
+	)
+}

internal/pubsub/table.go

@@ -0,0 +1,40 @@
+package pubsub
+
+import (
+	"database/sql"
+	"fmt"
+	"log"
+)
+
+func CreateTopicTables(db *sql.DB, topic string) {
+	topicTable := fmt.Sprintf("watermill_%s", topic)
+	offsetTable := fmt.Sprintf("watermill_offsets_%s", topic)
+
+	_, err := db.Exec(fmt.Sprintf(`
+		CREATE TABLE IF NOT EXISTS %s (
+			"offset" BIGSERIAL PRIMARY KEY,
+			uuid UUID NOT NULL,
+			payload BYTEA NOT NULL,
+			metadata JSONB NOT NULL DEFAULT '{}'::jsonb,
+			created_at TIMESTAMP NOT NULL DEFAULT NOW()
+		);
+	`, topicTable))
+	if err != nil {
+		log.Fatalf("❌ Failed to ensure topic table %s: %v", topicTable, err)
+	}
+
+	_, err = db.Exec(fmt.Sprintf(`
+		CREATE TABLE IF NOT EXISTS %s (
+			consumer_group VARCHAR(255) NOT NULL,
+			"offset" BIGINT NOT NULL DEFAULT 0,
+			offset_acked BIGINT NOT NULL DEFAULT 0,
+			offset_consumed BIGINT NOT NULL DEFAULT 0,
+			PRIMARY KEY (consumer_group)
+		);
+	`, offsetTable))
+	if err != nil {
+		log.Fatalf("❌ Failed to ensure offsets table %s: %v", offsetTable, err)
+	}
+
+	log.Printf("✅ Watermill tables ensured for topic: %s", topic)
+}