[push] migrate to external cache

Author: capcom6

go.mod

@@ -18,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

@@ -262,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/sms-gateway/app.go

@@ -43,7 +43,7 @@ var Module = fx.Module(
 	openapi.Module(),
 	handlers.Module,
 	auth.Module,
-	push.Module,
+	push.Module(),
 	db.Module,
 	cache.Module(),
 	pubsub.Module(),

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

@@ -2,49 +2,51 @@ package push
 
 import (
 	"context"
-	"errors"
+	"fmt"
 
 	"github.com/android-sms-gateway/server/internal/sms-gateway/modules/push/fcm"
 	"github.com/android-sms-gateway/server/internal/sms-gateway/modules/push/upstream"
 	"go.uber.org/fx"
 	"go.uber.org/zap"
 )
 
-var Module = fx.Module(
-	"push",
-	fx.Decorate(func(log *zap.Logger) *zap.Logger {
-		return log.Named("push")
-	}),
-	fx.Provide(newMetrics, fx.Private),
-	fx.Provide(
-		func(cfg Config, lc fx.Lifecycle) (c client, err error) {
-			switch cfg.Mode {
-			case ModeFCM:
-				c, err = fcm.New(cfg.ClientOptions)
-			case ModeUpstream:
-				c, err = upstream.New(cfg.ClientOptions)
-			default:
-				return nil, errors.New("invalid push mode")
-			}
+func Module() fx.Option {
+	return fx.Module(
+		"push",
+		fx.Decorate(func(log *zap.Logger) *zap.Logger {
+			return log.Named("push")
+		}),
+		fx.Provide(newMetrics, fx.Private),
+		fx.Provide(
+			func(cfg Config, lc fx.Lifecycle) (c client, err error) {
+				switch cfg.Mode {
+				case ModeFCM:
+					c, err = fcm.New(cfg.ClientOptions)
+				case ModeUpstream:
+					c, err = upstream.New(cfg.ClientOptions)
+				default:
+					return nil, fmt.Errorf("invalid push mode: %q", cfg.Mode)
+				}
 
-			if err != nil {
-				return nil, err
-			}
+				if err != nil {
+					return nil, err
+				}
 
-			lc.Append(fx.Hook{
-				OnStart: func(ctx context.Context) error {
-					return c.Open(ctx)
-				},
-				OnStop: func(ctx context.Context) error {
-					return c.Close(ctx)
-				},
-			})
+				lc.Append(fx.Hook{
+					OnStart: func(ctx context.Context) error {
+						return c.Open(ctx)
+					},
+					OnStop: func(ctx context.Context) error {
+						return c.Close(ctx)
+					},
+				})
 
-			return c, nil
-		},
-		fx.Private,
-	),
-	fx.Provide(
-		New,
-	),
-)
+				return c, nil
+			},
+			fx.Private,
+		),
+		fx.Provide(
+			New,
+		),
+	)
+}

internal/sms-gateway/modules/push/service.go

@@ -5,14 +5,19 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/android-sms-gateway/server/internal/sms-gateway/cache"
 	"github.com/android-sms-gateway/server/internal/sms-gateway/modules/push/types"
-	"github.com/capcom6/go-helpers/cache"
-	"github.com/capcom6/go-helpers/maps"
+	cacheImpl "github.com/android-sms-gateway/server/pkg/cache"
+	"github.com/samber/lo"
 
-	"go.uber.org/fx"
 	"go.uber.org/zap"
 )
 
+const (
+	cachePrefixEvents    = "events:"
+	cachePrefixBlacklist = "blacklist:"
+)
+
 type Config struct {
 	Mode Mode
 
@@ -22,53 +27,51 @@ type Config struct {
 	Timeout  time.Duration
 }
 
-type Params struct {
-	fx.In
-
-	Config Config
-
-	Client  client
-	Metrics *metrics
-
-	Logger *zap.Logger
-}
-
 type Service struct {
 	config Config
 
-	client  client
-	metrics *metrics
-
-	cache     *cache.Cache[eventWrapper]
-	blacklist *cache.Cache[struct{}]
+	client    client
+	events    cache.Cache
+	blacklist cache.Cache
 
-	logger *zap.Logger
+	metrics *metrics
+	logger  *zap.Logger
 }
 
-func New(params Params) *Service {
-	if params.Config.Timeout == 0 {
-		params.Config.Timeout = time.Second
+func New(
+	config Config,
+	client client,
+	cacheFactory cache.Factory,
+	metrics *metrics,
+	logger *zap.Logger,
+) (*Service, error) {
+	events, err := cacheFactory.New(cachePrefixEvents)
+	if err != nil {
+		return nil, fmt.Errorf("can't create events cache: %w", err)
 	}
-	if params.Config.Debounce < 5*time.Second {
-		params.Config.Debounce = 5 * time.Second
+
+	blacklist, err := cacheFactory.New(cachePrefixBlacklist)
+	if err != nil {
+		return nil, fmt.Errorf("can't create blacklist cache: %w", err)
 	}
 
-	return &Service{
-		config: params.Config,
+	config.Timeout = max(config.Timeout, time.Second)
+	config.Debounce = max(config.Debounce, 5*time.Second)
 
-		client:  params.Client,
-		metrics: params.Metrics,
+	return &Service{
+		config: config,
 
-		cache: cache.New[eventWrapper](cache.Config{}),
-		blacklist: cache.New[struct{}](cache.Config{
-			TTL: blacklistTimeout,
-		}),
+		client:    client,
+		events:    events,
+		blacklist: blacklist,
 
-		logger: params.Logger,
-	}
+		metrics: metrics,
+		logger:  logger,
+	}, nil
 }
 
-// Run runs the service with the provided context if a debounce is set.
+// Run starts a ticker that triggers the sendAll function every debounce interval.
+// It runs indefinitely until the provided context is canceled.
 func (s *Service) Run(ctx context.Context) {
 	ticker := time.NewTicker(s.config.Debounce)
 	defer ticker.Stop()
@@ -85,19 +88,28 @@ func (s *Service) Run(ctx context.Context) {
 
 // Enqueue adds the data to the cache and immediately sends all messages if the debounce is 0.
 func (s *Service) Enqueue(token string, event types.Event) error {
-	if _, err := s.blacklist.Get(token); err == nil {
+	ctx, cancel := context.WithTimeout(context.Background(), s.config.Timeout)
+	defer cancel()
+
+	if _, err := s.blacklist.Get(ctx, token); err == nil {
 		s.metrics.IncBlacklist(BlacklistOperationSkipped)
 		s.logger.Debug("Skipping blacklisted token", zap.String("token", token))
 		return nil
 	}
 
 	wrapper := eventWrapper{
-		token:   token,
-		event:   &event,
-		retries: 0,
+		Token:   token,
+		Event:   event,
+		Retries: 0,
+	}
+	wrapperData, err := wrapper.serialize()
+	if err != nil {
+		s.metrics.IncError(1)
+		return fmt.Errorf("can't serialize event wrapper: %w", err)
 	}
 
-	if err := s.cache.Set(token, wrapper); err != nil {
+	if err := s.events.Set(ctx, wrapper.key(), wrapperData); err != nil {
+		s.metrics.IncError(1)
 		return fmt.Errorf("can't add message to cache: %w", err)
 	}
 
@@ -108,20 +120,43 @@ func (s *Service) Enqueue(token string, event types.Event) error {
 
 // sendAll sends messages to all targets from the cache after initializing the service.
 func (s *Service) sendAll(ctx context.Context) {
-	targets := s.cache.Drain()
-	if len(targets) == 0 {
+	rawEvents, err := s.events.Drain(ctx)
+	if err != nil {
+		s.logger.Error("Can't drain cache", zap.Error(err))
 		return
 	}
 
-	messages := maps.MapValues(targets, func(w eventWrapper) types.Event {
-		return *w.event
-	})
+	if len(rawEvents) == 0 {
+		return
+	}
+
+	wrappers := lo.MapEntries(
+		rawEvents,
+		func(key string, value []byte) (string, *eventWrapper) {
+			wrapper := new(eventWrapper)
+			if err := wrapper.deserialize(value); err != nil {
+				s.metrics.IncError(1)
+				s.logger.Error("Failed to deserialize event wrapper", zap.String("key", key), zap.Binary("value", value), zap.Error(err))
+				return "", nil
+			}
+
+			return wrapper.Token, wrapper
+		},
+	)
+	delete(wrappers, "")
+
+	messages := lo.MapValues(
+		wrappers,
+		func(value *eventWrapper, key string) Event {
+			return value.Event
+		},
+	)
 
 	s.logger.Info("Sending messages", zap.Int("count", len(messages)))
-	ctx, cancel := context.WithTimeout(ctx, s.config.Timeout)
+	sendCtx, cancel := context.WithTimeout(ctx, s.config.Timeout)
 	defer cancel()
 
-	errs, err := s.client.Send(ctx, messages)
+	errs, err := s.client.Send(sendCtx, messages)
 	if len(errs) == 0 && err == nil {
 		s.logger.Info("Messages sent successfully", zap.Int("count", len(messages)))
 		return
@@ -138,11 +173,11 @@ func (s *Service) sendAll(ctx context.Context) {
 	for token, sendErr := range errs {
 		s.logger.Error("Can't send message", zap.Error(sendErr), zap.String("token", token))
 
-		wrapper := targets[token]
-		wrapper.retries++
+		wrapper := wrappers[token]
+		wrapper.Retries++
 
-		if wrapper.retries >= maxRetries {
-			if err := s.blacklist.Set(token, struct{}{}); err != nil {
+		if wrapper.Retries >= maxRetries {
+			if err := s.blacklist.Set(ctx, token, []byte{}, cacheImpl.WithTTL(blacklistTimeout)); err != nil {
 				s.logger.Warn("Can't add to blacklist", zap.String("token", token), zap.Error(err))
 			}
 
@@ -154,8 +189,16 @@ func (s *Service) sendAll(ctx context.Context) {
 			continue
 		}
 
-		if setErr := s.cache.SetOrFail(token, wrapper); setErr != nil {
-			s.logger.Info("Can't set message to cache", zap.Error(setErr))
+		wrapperData, err := wrapper.serialize()
+		if err != nil {
+			s.metrics.IncError(1)
+			s.logger.Error("Can't serialize event wrapper", zap.Error(err))
+			continue
+		}
+
+		if setErr := s.events.SetOrFail(ctx, wrapper.key(), wrapperData); setErr != nil {
+			s.logger.Warn("Can't set message to cache", zap.Error(setErr))
+			continue
 		}
 
 		s.metrics.IncRetry(RetryOutcomeRetried)

internal/sms-gateway/modules/push/types.go

@@ -2,6 +2,7 @@ package push
 
 import (
 	"context"
+	"encoding/json"
 
 	"github.com/android-sms-gateway/server/internal/sms-gateway/modules/push/types"
 )
@@ -17,12 +18,24 @@ type Event = types.Event
 
 type client interface {
 	Open(ctx context.Context) error
-	Send(ctx context.Context, messages map[string]types.Event) (map[string]error, error)
+	Send(ctx context.Context, messages map[string]Event) (map[string]error, error)
 	Close(ctx context.Context) error
 }
 
 type eventWrapper struct {
-	token   string
-	event   *types.Event
-	retries int
+	Token   string `json:"token"`
+	Event   Event  `json:"event"`
+	Retries int    `json:"retries"`
+}
+
+func (e *eventWrapper) key() string {
+	return e.Token + ":" + string(e.Event.Type)
+}
+
+func (e *eventWrapper) serialize() ([]byte, error) {
+	return json.Marshal(e)
+}
+
+func (e *eventWrapper) deserialize(data []byte) error {
+	return json.Unmarshal(data, e)
 }

internal/sms-gateway/modules/push/types/types.go

@@ -5,6 +5,6 @@ import (
 )
 
 type Event struct {
-	Type smsgateway.PushEventType
-	Data map[string]string
+	Type smsgateway.PushEventType `json:"type"`
+	Data map[string]string        `json:"data"`
 }