[push] support multiple events per token per batch

Author: capcom6

internal/sms-gateway/modules/push/fcm/client.go

@@ -52,25 +52,25 @@ func (c *Client) Open(ctx context.Context) error {
 	return nil
 }
 
-func (c *Client) Send(ctx context.Context, messages map[string]types.Event) (map[string]error, error) {
-	errs := make(map[string]error, len(messages))
-	for address, payload := range messages {
-		eventMap, err := eventToMap(payload)
+func (c *Client) Send(ctx context.Context, messages []types.Message) ([]error, error) {
+	errs := make([]error, len(messages))
+
+	for i, message := range messages {
+		data, err := eventToMap(message.Event)
 		if err != nil {
-			errs[address] = fmt.Errorf("can't marshal event: %w", err)
+			errs[i] = fmt.Errorf("can't marshal event: %w", err)
 			continue
 		}
 
 		_, err = c.client.Send(ctx, &messaging.Message{
-			Data: eventMap,
+			Data: data,
 			Android: &messaging.AndroidConfig{
 				Priority: "high",
 			},
-			Token: address,
+			Token: message.Token,
 		})
-
 		if err != nil {
-			errs[address] = fmt.Errorf("can't send message to %s: %w", address, err)
+			errs[i] = fmt.Errorf("can't send message: %w", err)
 		}
 	}
 

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

@@ -5,13 +5,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 )
 
-type RetryOutcome string
-
-const (
-	RetryOutcomeRetried     RetryOutcome = "retried"
-	RetryOutcomeMaxAttempts RetryOutcome = "max_attempts"
-)
-
 type BlacklistOperation string
 
 const (
@@ -21,7 +14,7 @@ const (
 
 type metrics struct {
 	enqueuedCounter  *prometheus.CounterVec
-	retriesCounter   *prometheus.CounterVec
+	retriesCounter   prometheus.Counter
 	blacklistCounter *prometheus.CounterVec
 	errorsCounter    *prometheus.CounterVec
 }
@@ -35,12 +28,12 @@ func newMetrics() *metrics {
 			Help:      "Total number of messages enqueued",
 		}, []string{"event"}),
 
-		retriesCounter: promauto.NewCounterVec(prometheus.CounterOpts{
+		retriesCounter: promauto.NewCounter(prometheus.CounterOpts{
 			Namespace: "sms",
 			Subsystem: "push",
 			Name:      "retries_total",
 			Help:      "Total retry attempts",
-		}, []string{"outcome"}),
+		}),
 
 		blacklistCounter: promauto.NewCounterVec(prometheus.CounterOpts{
 			Namespace: "sms",
@@ -62,8 +55,8 @@ func (m *metrics) IncEnqueued(event string) {
 	m.enqueuedCounter.WithLabelValues(event).Inc()
 }
 
-func (m *metrics) IncRetry(outcome RetryOutcome) {
-	m.retriesCounter.WithLabelValues(string(outcome)).Inc()
+func (m *metrics) IncRetry() {
+	m.retriesCounter.Inc()
 }
 
 func (m *metrics) IncBlacklist(operation BlacklistOperation) {

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

@@ -130,77 +130,90 @@ func (s *Service) sendAll(ctx context.Context) {
 		return
 	}
 
-	wrappers := lo.MapEntries(
-		rawEvents,
-		func(key string, value []byte) (string, *eventWrapper) {
+	wrappers := lo.FilterMap(
+		lo.Values(rawEvents),
+		func(value []byte, _ int) (*eventWrapper, bool) {
 			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
+				s.logger.Error("Failed to deserialize event wrapper", zap.Binary("value", value), zap.Error(err))
+				return nil, false
 			}
 
-			return wrapper.Token, wrapper
+			return wrapper, true
 		},
 	)
-	delete(wrappers, "")
 
-	messages := lo.MapValues(
+	messages := lo.Map(
 		wrappers,
-		func(value *eventWrapper, key string) Event {
-			return value.Event
+		func(wrapper *eventWrapper, _ int) types.Message {
+			return types.Message{
+				Token: wrapper.Token,
+				Event: wrapper.Event,
+			}
 		},
 	)
 
-	s.logger.Info("Sending messages", zap.Int("count", len(messages)))
+	totalMessages := len(messages)
+	if totalMessages == 0 {
+		return
+	}
+
+	s.logger.Info("sending messages", zap.Int("total", totalMessages))
+
 	sendCtx, cancel := context.WithTimeout(ctx, s.config.Timeout)
 	defer cancel()
-
 	errs, err := s.client.Send(sendCtx, messages)
 	if len(errs) == 0 && err == nil {
-		s.logger.Info("Messages sent successfully", zap.Int("count", len(messages)))
+		s.logger.Info("messages sent successfully", zap.Int("total", totalMessages))
 		return
 	}
 
 	if err != nil {
-		s.metrics.IncError(len(messages))
-		s.logger.Error("Can't send messages", zap.Error(err))
+		s.metrics.IncError(totalMessages)
+		s.logger.Error("failed to send messages", zap.Int("total", totalMessages), zap.Error(err))
 		return
 	}
 
-	s.metrics.IncError(len(errs))
+	totalErrors := lo.CountBy(errs, func(err error) bool { return err != nil })
+	s.metrics.IncError(totalErrors)
+
+	for i, err := range errs {
+		if err == nil {
+			continue
+		}
 
-	for token, sendErr := range errs {
-		s.logger.Error("Can't send message", zap.Error(sendErr), zap.String("token", token))
+		wrapper := wrappers[i]
+		token := wrapper.Token
 
-		wrapper := wrappers[token]
 		wrapper.Retries++
 
 		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))
+				s.logger.Warn("failed to blacklist", zap.String("token", token), zap.Error(err))
+				continue
 			}
 
 			s.metrics.IncBlacklist(BlacklistOperationAdded)
-			s.metrics.IncRetry(RetryOutcomeMaxAttempts)
-			s.logger.Warn("Retries exceeded, blacklisting token",
+			s.logger.Warn("retries exceeded, blacklisting token",
 				zap.String("token", token),
-				zap.Duration("ttl", blacklistTimeout))
+				zap.Duration("ttl", blacklistTimeout),
+			)
 			continue
 		}
 
 		wrapperData, err := wrapper.serialize()
 		if err != nil {
 			s.metrics.IncError(1)
-			s.logger.Error("Can't serialize event wrapper", zap.Error(err))
+			s.logger.Error("failed to 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))
+			s.logger.Warn("failed to set message to cache", zap.String("key", wrapper.key()), zap.Error(setErr))
 			continue
 		}
 
-		s.metrics.IncRetry(RetryOutcomeRetried)
+		s.metrics.IncRetry()
 	}
 }

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

@@ -18,7 +18,7 @@ type Event = types.Event
 
 type client interface {
 	Open(ctx context.Context) error
-	Send(ctx context.Context, messages map[string]Event) (map[string]error, error)
+	Send(ctx context.Context, messages []types.Message) ([]error, error)
 	Close(ctx context.Context) error
 }
 

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

@@ -4,6 +4,11 @@ import (
 	"github.com/android-sms-gateway/client-go/smsgateway"
 )
 
+type Message struct {
+	Token string
+	Event Event
+}
+
 type Event struct {
 	Type smsgateway.PushEventType `json:"type"`
 	Data map[string]string        `json:"data"`

internal/sms-gateway/modules/push/upstream/client.go

@@ -11,7 +11,7 @@ import (
 
 	"github.com/android-sms-gateway/client-go/smsgateway"
 	"github.com/android-sms-gateway/server/internal/sms-gateway/modules/push/types"
-	"github.com/capcom6/go-helpers/maps"
+	"github.com/samber/lo"
 )
 
 const BASE_URL = "https://api.sms-gate.app/upstream/v1"
@@ -42,18 +42,19 @@ func (c *Client) Open(ctx context.Context) error {
 	return nil
 }
 
-func (c *Client) Send(ctx context.Context, messages map[string]types.Event) (map[string]error, error) {
-	payload := make(smsgateway.UpstreamPushRequest, 0, len(messages))
+func (c *Client) Send(ctx context.Context, messages []types.Message) ([]error, error) {
+	payload := lo.Map(
+		messages,
+		func(item types.Message, _ int) smsgateway.PushNotification {
+			return smsgateway.PushNotification{
+				Token: item.Token,
+				Event: item.Event.Type,
+				Data:  item.Event.Data,
+			}
+		},
+	)
 
-	for address, data := range messages {
-		payload = append(payload, smsgateway.PushNotification{
-			Token: address,
-			Event: data.Type,
-			Data:  data.Data,
-		})
-	}
-
-	payloadBytes, err := json.Marshal(payload)
+	payloadBytes, err := json.Marshal(smsgateway.UpstreamPushRequest(payload))
 
 	if err != nil {
 		return nil, fmt.Errorf("can't marshal payload: %w", err)
@@ -65,11 +66,11 @@ func (c *Client) Send(ctx context.Context, messages map[string]types.Event) (map
 	}
 
 	req.Header.Set("Content-Type", "application/json")
-	req.Header.Set("User-Agent", "android-sms-gateway/1.x (server; golang)")
+	req.Header.Set("User-Agent", "sms-gate/1.x (server; golang)")
 
 	resp, err := c.client.Do(req)
 	if err != nil {
-		return c.mapErrors(messages, fmt.Errorf("can't send request: %w", err)), nil
+		return nil, fmt.Errorf("can't send request: %w", err)
 	}
 
 	defer func() {
@@ -84,10 +85,13 @@ func (c *Client) Send(ctx context.Context, messages map[string]types.Event) (map
 	return nil, nil
 }
 
-func (c *Client) mapErrors(messages map[string]types.Event, err error) map[string]error {
-	return maps.MapValues(messages, func(e types.Event) error {
-		return err
-	})
+func (c *Client) mapErrors(messages []types.Message, err error) []error {
+	return lo.Map(
+		messages,
+		func(_ types.Message, _ int) error {
+			return err
+		},
+	)
 }
 
 func (c *Client) Close(ctx context.Context) error {