[pubsub] add module

Author: capcom6

internal/config/config.go

@@ -15,6 +15,7 @@ type Config struct {
 	Tasks    Tasks     `yaml:"tasks"`    // tasks config
 	SSE      SSE       `yaml:"sse"`      // server-sent events config
 	Cache    Cache     `yaml:"cache"`    // cache (memory or redis) config
+	PubSub   PubSub    `yaml:"pubsub"`   // pubsub (memory or redis) config
 }
 
 type Gateway struct {
@@ -75,6 +76,10 @@ type Cache struct {
 	URL string `yaml:"url" envconfig:"CACHE__URL"`
 }
 
+type PubSub struct {
+	URL string `yaml:"url" envconfig:"PUBSUB__URL"`
+}
+
 var defaultConfig = Config{
 	Gateway: Gateway{Mode: GatewayModePublic},
 	HTTP: HTTP{
@@ -103,4 +108,7 @@ var defaultConfig = Config{
 	Cache: Cache{
 		URL: "memory://",
 	},
+	PubSub: PubSub{
+		URL: "memory://",
+	},
 }

internal/config/module.go

@@ -11,6 +11,7 @@ import (
 	"github.com/android-sms-gateway/server/internal/sms-gateway/modules/messages"
 	"github.com/android-sms-gateway/server/internal/sms-gateway/modules/push"
 	"github.com/android-sms-gateway/server/internal/sms-gateway/modules/sse"
+	"github.com/android-sms-gateway/server/internal/sms-gateway/pubsub"
 	"github.com/capcom6/go-infra-fx/config"
 	"github.com/capcom6/go-infra-fx/db"
 	"github.com/capcom6/go-infra-fx/http"
@@ -121,4 +122,10 @@ var Module = fx.Module(
 			URL: cfg.Cache.URL,
 		}
 	}),
+	fx.Provide(func(cfg Config) pubsub.Config {
+		return pubsub.Config{
+			URL:        cfg.PubSub.URL,
+			BufferSize: 128,
+		}
+	}),
 )

internal/sms-gateway/app.go

@@ -21,6 +21,7 @@ import (
 	"github.com/android-sms-gateway/server/internal/sms-gateway/modules/webhooks"
 	"github.com/android-sms-gateway/server/internal/sms-gateway/online"
 	"github.com/android-sms-gateway/server/internal/sms-gateway/openapi"
+	"github.com/android-sms-gateway/server/internal/sms-gateway/pubsub"
 	"github.com/capcom6/go-infra-fx/cli"
 	"github.com/capcom6/go-infra-fx/db"
 	"github.com/capcom6/go-infra-fx/http"
@@ -45,6 +46,7 @@ var Module = fx.Module(
 	push.Module,
 	db.Module,
 	cache.Module(),
+	pubsub.Module(),
 	events.Module,
 	messages.Module,
 	health.Module,

internal/sms-gateway/pubsub/config.go

@@ -0,0 +1,7 @@
+package pubsub
+
+// Config controls the PubSub backend via a URL (e.g., "memory://", "redis://...").
+type Config struct {
+	URL        string
+	BufferSize uint
+}

internal/sms-gateway/pubsub/module.go

@@ -0,0 +1,26 @@
+package pubsub
+
+import (
+	"context"
+
+	"github.com/android-sms-gateway/server/pkg/pubsub"
+	"go.uber.org/fx"
+	"go.uber.org/zap"
+)
+
+func Module() fx.Option {
+	return fx.Module(
+		"pubsub",
+		fx.Decorate(func(log *zap.Logger) *zap.Logger {
+			return log.Named("pubsub")
+		}),
+		fx.Provide(New),
+		fx.Invoke(func(pubsub pubsub.PubSub, lc fx.Lifecycle) {
+			lc.Append(fx.Hook{
+				OnStop: func(_ context.Context) error {
+					return pubsub.Close()
+				},
+			})
+		}),
+	)
+}

internal/sms-gateway/pubsub/pubsub.go

@@ -0,0 +1,40 @@
+package pubsub
+
+import (
+	"fmt"
+	"net/url"
+
+	"github.com/android-sms-gateway/core/redis"
+	"github.com/android-sms-gateway/server/pkg/pubsub"
+)
+
+const (
+	topicPrefix = "sms-gateway:"
+)
+
+func New(config Config) (pubsub.PubSub, error) {
+	if config.URL == "" {
+		config.URL = "memory://"
+	}
+
+	u, err := url.Parse(config.URL)
+	if err != nil {
+		return nil, fmt.Errorf("can't parse url: %w", err)
+	}
+
+	opts := []pubsub.Option{}
+	opts = append(opts, pubsub.WithBufferSize(config.BufferSize))
+
+	switch u.Scheme {
+	case "memory":
+		return pubsub.NewMemory(opts...), nil
+	case "redis":
+		client, err := redis.New(redis.Config{URL: config.URL})
+		if err != nil {
+			return nil, fmt.Errorf("can't create redis client: %w", err)
+		}
+		return pubsub.NewRedis(client, topicPrefix, opts...), nil
+	default:
+		return nil, fmt.Errorf("invalid scheme: %s", u.Scheme)
+	}
+}