[push] refactor some code

Author: capcom6

internal/sms-gateway/handlers/upstream.go

@@ -68,12 +68,12 @@ func (h *upstreamHandler) postPush(c *fiber.Ctx) error {
 			return err
 		}
 
-		event := push.Event{
-			Event: anys.ZeroDefault(v.Event, smsgateway.PushMessageEnqueued),
-			Data:  v.Data,
-		}
+		event := push.NewEvent(
+			anys.ZeroDefault(v.Event, smsgateway.PushMessageEnqueued),
+			v.Data,
+		)
 
-		if err := h.pushSvc.Enqueue(v.Token, &event); err != nil {
+		if err := h.pushSvc.Enqueue(v.Token, event); err != nil {
 			h.Logger.Error("Can't push message", zap.Error(err))
 		}
 	}

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

@@ -0,0 +1,3 @@
+package push
+
+const maxRetries = 3

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

@@ -7,22 +7,30 @@ import (
 )
 
 type Event struct {
-	Event smsgateway.PushEventType
-	Data  map[string]string
+	event smsgateway.PushEventType
+	data  map[string]string
+}
+
+func (e *Event) Event() smsgateway.PushEventType {
+	return e.event
+}
+
+func (e *Event) Data() map[string]string {
+	return e.data
 }
 
 func (e *Event) Map() map[string]string {
-	json, _ := json.Marshal(e.Data)
+	json, _ := json.Marshal(e.data)
 
 	return map[string]string{
-		"event": string(e.Event),
+		"event": string(e.event),
 		"data":  string(json),
 	}
 }
 
 func NewEvent(event smsgateway.PushEventType, data map[string]string) *Event {
 	return &Event{
-		Event: event,
-		Data:  data,
+		event: event,
+		data:  data,
 	}
 }

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

@@ -2,10 +2,7 @@ package fcm
 
 import (
 	"context"
-	"errors"
 	"fmt"
-	"maps"
-	"slices"
 	"sync"
 
 	firebase "firebase.google.com/go/v4"
@@ -71,7 +68,7 @@ func (c *Client) Send(ctx context.Context, messages map[string]domain.Event) (ma
 		}
 	}
 
-	return errs, errors.Join(slices.Collect(maps.Values(errs))...)
+	return errs, nil
 }
 
 func (c *Client) Close(ctx context.Context) error {

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

@@ -7,6 +7,7 @@ import (
 
 	"github.com/android-sms-gateway/server/internal/sms-gateway/modules/push/domain"
 	"github.com/capcom6/go-helpers/cache"
+	"github.com/capcom6/go-helpers/maps"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
@@ -99,15 +100,15 @@ func (s *Service) Run(ctx context.Context) {
 func (s *Service) Enqueue(token string, event *domain.Event) error {
 	wrapper := eventWrapper{
 		token:   token,
-		event:   *event,
+		event:   event,
 		retries: 0,
 	}
 
 	if err := s.cache.Set(token, wrapper); err != nil {
 		return fmt.Errorf("can't add message to cache: %w", err)
 	}
 
-	s.enqueuedCounter.WithLabelValues(string(event.Event)).Inc()
+	s.enqueuedCounter.WithLabelValues(string(event.Event())).Inc()
 
 	return nil
 }
@@ -119,36 +120,41 @@ func (s *Service) sendAll(ctx context.Context) {
 		return
 	}
 
-	messages := make(map[string]domain.Event, len(targets))
-	for _, w := range targets {
-		if w.retries >= 3 {
-			s.retriesCounter.WithLabelValues("max_attempts").Inc()
-			s.logger.Warn("Retries exceeded", zap.String("token", w.token))
-			continue
-		}
+	messages := maps.MapValues(targets, func(w eventWrapper) domain.Event {
+		return *w.event
+	})
+
+	s.logger.Info("Sending messages", zap.Int("count", len(messages)))
+	ctx, cancel := context.WithTimeout(ctx, s.config.Timeout)
+	defer cancel()
 
-		messages[w.token] = w.event
+	errs, err := s.client.Send(ctx, messages)
+	if len(errs) == 0 && err == nil {
+		s.logger.Info("Messages sent successfully", zap.Int("count", len(messages)))
+		return
 	}
 
-	if len(messages) == 0 {
+	if err != nil {
+		s.logger.Error("Can't send messages", zap.Error(err))
 		return
 	}
 
-	s.logger.Info("Sending messages", zap.Int("count", len(messages)))
-	ctx, cancel := context.WithTimeout(ctx, s.config.Timeout)
-	if errs, err := s.client.Send(ctx, messages); err != nil {
-		for token := range errs {
-			wrapper := targets[token]
-			wrapper.retries++
-			if s.cache.SetOrFail(token, wrapper) != nil {
-				s.logger.Info("Can't set message to cache", zap.Error(err))
-			} else {
-				s.logger.Info("Retrying message", zap.String("token", token))
-			}
-			s.retriesCounter.WithLabelValues("retried").Inc()
+	for token, sendErr := range errs {
+		s.logger.Error("Can't send message", zap.Error(sendErr), zap.String("token", token))
+
+		wrapper := targets[token]
+		wrapper.retries++
+
+		if wrapper.retries >= maxRetries {
+			s.retriesCounter.WithLabelValues(string(RetryOutcomeMaxAttempts)).Inc()
+			s.logger.Warn("Retries exceeded", zap.String("token", token))
+			continue
 		}
 
-		s.logger.Error("Can't send messages", zap.Error(err))
+		if setErr := s.cache.SetOrFail(token, wrapper); setErr != nil {
+			s.logger.Info("Can't set message to cache", zap.Error(setErr))
+		}
+
+		s.retriesCounter.WithLabelValues(string(RetryOutcomeRetried)).Inc()
 	}
-	cancel()
 }

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

@@ -11,6 +11,8 @@ import (
 type Mode string
 type Event = domain.Event
 
+var NewEvent = domain.NewEvent
+
 const (
 	ModeFCM      Mode = "fcm"
 	ModeUpstream Mode = "upstream"
@@ -22,9 +24,16 @@ type client interface {
 	Close(ctx context.Context) error
 }
 
+type RetryOutcome string
+
+const (
+	RetryOutcomeMaxAttempts RetryOutcome = "max_attempts"
+	RetryOutcomeRetried     RetryOutcome = "retried"
+)
+
 type eventWrapper struct {
 	token   string
-	event   domain.Event
+	event   *domain.Event
 	retries int
 }
 

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

@@ -48,28 +48,28 @@ func (c *Client) Send(ctx context.Context, messages map[string]domain.Event) (ma
 	for address, data := range messages {
 		payload = append(payload, smsgateway.PushNotification{
 			Token: address,
-			Event: data.Event,
-			Data:  data.Data,
+			Event: data.Event(),
+			Data:  data.Data(),
 		})
 	}
 
 	payloadBytes, err := json.Marshal(payload)
 
 	if err != nil {
-		return c.sendError(messages, fmt.Errorf("can't marshal payload: %w", err))
+		return nil, fmt.Errorf("can't marshal payload: %w", err)
 	}
 
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, BASE_URL+"/push", bytes.NewReader(payloadBytes))
 	if err != nil {
-		return c.sendError(messages, fmt.Errorf("can't create request: %w", err))
+		return nil, fmt.Errorf("can't create request: %w", err)
 	}
 
 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("User-Agent", "android-sms-gateway/1.x (server; golang)")
 
 	resp, err := c.client.Do(req)
 	if err != nil {
-		return c.sendError(messages, fmt.Errorf("can't send request: %w", err))
+		return c.mapErrors(messages, fmt.Errorf("can't send request: %w", err)), nil
 	}
 
 	defer func() {
@@ -78,16 +78,16 @@ func (c *Client) Send(ctx context.Context, messages map[string]domain.Event) (ma
 	}()
 
 	if resp.StatusCode >= 400 {
-		return c.sendError(messages, fmt.Errorf("unexpected status code: %d", resp.StatusCode))
+		return c.mapErrors(messages, fmt.Errorf("unexpected status code: %d", resp.StatusCode)), nil
 	}
 
 	return nil, nil
 }
 
-func (c *Client) sendError(messages map[string]domain.Event, err error) (map[string]error, error) {
+func (c *Client) mapErrors(messages map[string]domain.Event, err error) map[string]error {
 	return maps.MapValues(messages, func(e domain.Event) error {
 		return err
-	}), err
+	})
 }
 
 func (c *Client) Close(ctx context.Context) error {