[pubsub] introduce memory and redis implementations

.gitignore

@@ -1,6 +1,6 @@
 # File created using '.gitignore Generator' for Visual Studio Code: https://bit.ly/vscode-gig
-# Created by https://www.toptal.com/developers/gitignore/api/windows,visualstudiocode,terraform,macos,linux,go,dotenv
-# Edit at https://www.toptal.com/developers/gitignore?templates=windows,visualstudiocode,terraform,macos,linux,go,dotenv
+# Created by https://www.toptal.com/developers/gitignore/api/windows,visualstudiocode,macos,linux,go,dotenv,terraform
+# Edit at https://www.toptal.com/developers/gitignore?templates=windows,visualstudiocode,macos,linux,go,dotenv,terraform
 
 ### dotenv ###
 .env
@@ -157,17 +157,16 @@ $RECYCLE.BIN/
 # Windows shortcuts
 *.lnk
 
-# End of https://www.toptal.com/developers/gitignore/api/windows,visualstudiocode,terraform,macos,linux,go,dotenv
+# End of https://www.toptal.com/developers/gitignore/api/windows,visualstudiocode,macos,linux,go,dotenv,terraform
 
 # Custom rules (everything added below won't be overriden by 'Generate .gitignore File' if you use 'Update' option)
 
-/tmp
+dist/
+tmp/
 
 /configs/*
 !/configs/*.example.*
 
-/certs
-
-/api/docs.go
-
-go.work*
+go.work*
+coverage.*
+benchmark.*

Makefile

@@ -6,11 +6,37 @@ ifeq ($(OS),Windows_NT)
 	extension = .exe
 endif
 
-.DEFAULT_GOAL := build
+# Default target
+all: fmt lint test benchmark
 
-init:
+fmt:
+	golangci-lint fmt
+
+# Lint the code using golangci-lint
+lint:
+	golangci-lint run --timeout=5m
+
+# Run tests with coverage
+test:
+	go test -race -shuffle=on -count=1 -covermode=atomic -coverpkg=./... -coverprofile=coverage.out ./...
+
+# Run benchmarks
+benchmark:
+	go test -run=^$$ -bench=. -benchmem ./... | tee benchmark.txt
+
+# Download dependencies
+deps:
 	go mod download
 
+# Clean up generated files
+clean:
+	go clean -cache -testcache
+	rm -f coverage.out benchmark.txt
+
+###
+
+init: deps
+
 init-dev: init
 	go install github.com/air-verse/air@latest \
 		&& go install github.com/swaggo/swag/cmd/swag@latest \
@@ -31,11 +57,7 @@ db-upgrade-raw:
 run:
 	go run cmd/$(project_name)/main.go
 
-lint:
-	golangci-lint run ./...
-
-test:
-	go test -race -coverprofile=coverage.out -covermode=atomic ./...
+test-e2e: test
 	cd test/e2e && go test -count=1 .
 
 build:
@@ -53,7 +75,7 @@ docker:
 docker-dev:
 	docker compose -f deployments/docker-compose/docker-compose.dev.yml up --build
 
-clean:
+docker-clean:
 	docker compose -f deployments/docker-compose/docker-compose.yml down --volumes
 
-.PHONY: init init-dev air db-upgrade db-upgrade-raw run test build install docker docker-dev api-docs view-docs clean
+.PHONY: all fmt lint test benchmark deps clean init init-dev air ngrok db-upgrade db-upgrade-raw run test-e2e build install docker-build docker docker-dev docker-clean

configs/config.example.yml

@@ -24,8 +24,10 @@ fcm: # firebase cloud messaging config
   credentials_json: "{}" # firebase credentials json (for public mode only) [FCM__CREDENTIALS_JSON]
   timeout_seconds: 1 # push notification send timeout [FCM__TIMEOUT_SECONDS]
   debounce_seconds: 5 # push notification debounce (>= 5s) [FCM__DEBOUNCE_SECONDS]
-cache: # cache config
-  url: memory:// # cache url (memory:// or redis://) [CACHE__URL]
 tasks: # tasks config
   hashing: # hashing task (hashes processed messages for privacy purposes)
     interval_seconds: 15 # hashing interval in seconds [TASKS__HASHING__INTERVAL_SECONDS]
+cache: # cache config
+  url: memory:// # cache url (memory:// or redis://) [CACHE__URL]
+pubsub: # pubsub config
+  url: memory:// # pubsub url (memory:// or redis://) [PUBSUB__URL]

go.mod

@@ -5,7 +5,6 @@ go 1.24.1
 require (
 	firebase.google.com/go/v4 v4.12.1
 	github.com/android-sms-gateway/client-go v1.9.5
-	github.com/android-sms-gateway/core v1.0.1
 	github.com/ansrivas/fiberprometheus/v2 v2.6.1
 	github.com/capcom6/go-helpers v0.3.0
 	github.com/capcom6/go-infra-fx v0.4.0
@@ -19,6 +18,7 @@ require (
 	github.com/nyaruka/phonenumbers v1.4.0
 	github.com/prometheus/client_golang v1.19.1
 	github.com/redis/go-redis/v9 v9.9.0
+	github.com/samber/lo v1.52.0
 	github.com/swaggo/swag v1.16.6
 	go.uber.org/fx v1.24.0
 	go.uber.org/zap v1.27.0

go.sum

@@ -34,8 +34,6 @@ github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 h1:d+Bc7a5rLufV
 github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578/go.mod h1:uGdkoq3SwY9Y+13GIhn11/XLaGBb4BfwItxLd5jeuXE=
 github.com/android-sms-gateway/client-go v1.9.5 h1:fHrE1Pi3rKUdPVMmI9evKW0iyjB5bMIhFRxyq1wVQ+o=
 github.com/android-sms-gateway/client-go v1.9.5/go.mod h1:DQsReciU1xcaVW3T5Z2bqslNdsAwCFCtghawmA6g6L4=
-github.com/android-sms-gateway/core v1.0.1 h1:7QyqyW3UQSQmEXQuUgXjZwHSnOd65DTxHUyhXQi6gpc=
-github.com/android-sms-gateway/core v1.0.1/go.mod h1:HXczGDCKxTeuiwadPElczCx/y3Y6Wamc5kl5nFp5rVM=
 github.com/andybalholm/brotli v1.1.0 h1:eLKJA0d02Lf0mVpIDgYnqXcUn0GqVmEFny3VuID1U3M=
 github.com/andybalholm/brotli v1.1.0/go.mod h1:sms7XGricyQI9K10gOSf56VKKWS4oLer58Q+mhRPtnY=
 github.com/ansrivas/fiberprometheus/v2 v2.6.1 h1:wac3pXaE6BYYTF04AC6K0ktk6vCD+MnDOJZ3SK66kXM=
@@ -264,6 +262,8 @@ github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ=
 github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
 github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
 github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
+github.com/samber/lo v1.52.0 h1:Rvi+3BFHES3A8meP33VPAxiBZX/Aws5RxrschYGjomw=
+github.com/samber/lo v1.52.0/go.mod h1:4+MXEGsJzbKGaUEQFKBq2xtfuznW9oz/WrgyzMzRoM0=
 github.com/segmentio/asm v1.2.0 h1:9BQrFxC+YOHJlTlHGkTrFWf59nbL3XnCoFLTwDCI7ys=
 github.com/segmentio/asm v1.2.0/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs=
 github.com/sethvargo/go-retry v0.2.4 h1:T+jHEQy/zKJf5s95UkguisicE0zuF9y7+/vgz08Ocec=

internal/config/config.go

@@ -16,6 +16,7 @@ type Config struct {
 	SSE      SSE       `yaml:"sse"`      // server-sent events config
 	Messages Messages  `yaml:"messages"` // messages config
 	Cache    Cache     `yaml:"cache"`    // cache (memory or redis) config
+	PubSub   PubSub    `yaml:"pubsub"`   // pubsub (memory or redis) config
 }
 
 type Gateway struct {
@@ -81,6 +82,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{
@@ -113,4 +118,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"
@@ -122,4 +123,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"
@@ -42,9 +43,10 @@ var Module = fx.Module(
 	openapi.Module(),
 	handlers.Module,
 	auth.Module,
-	push.Module,
+	push.Module(),
 	db.Module,
 	cache.Module(),
+	pubsub.Module(),
 	events.Module,
 	messages.Module(),
 	health.Module,

internal/sms-gateway/cache/factory.go

@@ -4,7 +4,6 @@ import (
 	"fmt"
 	"net/url"
 
-	"github.com/android-sms-gateway/core/redis"
 	"github.com/android-sms-gateway/server/pkg/cache"
 )
 
@@ -40,13 +39,14 @@ func NewFactory(config Config) (Factory, error) {
 			},
 		}, 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 &factory{
 			new: func(name string) (Cache, error) {
-				return cache.NewRedis(client, name, 0), nil
+				return cache.NewRedis(cache.RedisConfig{
+					Client: nil,
+					URL:    config.URL,
+					Prefix: keyPrefix + name,
+					TTL:    0,
+				})
 			},
 		}, nil
 	default:
@@ -56,5 +56,5 @@ func NewFactory(config Config) (Factory, error) {
 
 // New implements Factory.
 func (f *factory) New(name string) (Cache, error) {
-	return f.new(keyPrefix + name)
+	return f.new(name)
 }

internal/sms-gateway/modules/events/events.go

@@ -6,15 +6,15 @@ import (
 	"github.com/android-sms-gateway/client-go/smsgateway"
 )
 
-func NewMessageEnqueuedEvent() *Event {
+func NewMessageEnqueuedEvent() Event {
 	return NewEvent(smsgateway.PushMessageEnqueued, nil)
 }
 
-func NewWebhooksUpdatedEvent() *Event {
+func NewWebhooksUpdatedEvent() Event {
 	return NewEvent(smsgateway.PushWebhooksUpdated, nil)
 }
 
-func NewMessagesExportRequestedEvent(since, until time.Time) *Event {
+func NewMessagesExportRequestedEvent(since, until time.Time) Event {
 	return NewEvent(
 		smsgateway.PushMessagesExportRequested,
 		map[string]string{
@@ -24,6 +24,6 @@ func NewMessagesExportRequestedEvent(since, until time.Time) *Event {
 	)
 }
 
-func NewSettingsUpdatedEvent() *Event {
+func NewSettingsUpdatedEvent() Event {
 	return NewEvent(smsgateway.PushSettingsUpdated, nil)
 }

internal/sms-gateway/modules/events/metrics.go

@@ -19,8 +19,11 @@ const (
 	DeliveryTypeSSE     = "sse"
 	DeliveryTypeUnknown = "unknown"
 
-	FailureReasonQueueFull      = "queue_full"
-	FailureReasonProviderFailed = "provider_failed"
+	FailureReasonSerializationError = "serialization_error"
+	FailureReasonPublishError       = "publish_error"
+	FailureReasonProviderFailed     = "provider_failed"
+
+	EventTypeUnknown = "unknown"
 )
 
 // metrics contains all Prometheus metrics for the events module

internal/sms-gateway/modules/events/module.go

@@ -14,11 +14,18 @@ var Module = fx.Module(
 	}),
 	fx.Provide(newMetrics, fx.Private),
 	fx.Provide(NewService),
-	fx.Invoke(func(lc fx.Lifecycle, svc *Service) {
+	fx.Invoke(func(lc fx.Lifecycle, svc *Service, logger *zap.Logger, sh fx.Shutdowner) {
 		ctx, cancel := context.WithCancel(context.Background())
 		lc.Append(fx.Hook{
 			OnStart: func(_ context.Context) error {
-				go svc.Run(ctx)
+				go func() {
+					if err := svc.Run(ctx); err != nil {
+						logger.Error("Error running events service", zap.Error(err))
+						if err := sh.Shutdown(fx.ExitCode(1)); err != nil {
+							logger.Error("Failed to shutdown", zap.Error(err))
+						}
+					}
+				}()
 				return nil
 			},
 			OnStop: func(_ context.Context) error {