add circuit breaker implementation and refactor SMS sending logic

Author: Armin

internal/operator/operatorA/service.go

@@ -2,9 +2,23 @@ package operatorA
 
 import (
 	"context"
+	"sms-gateway/app"
 	"sms-gateway/internal/model"
 )
 
 type OA struct{}
 
-func (o OA) Send(ctx context.Context, s model.SMS) {}
+func (o OA) Send(ctx context.Context, s model.SMS) error {
+
+	//return errors.New("fall down")  for test circuit breaker
+
+	for _, v := range s.Recipients {
+		app.Logger.Info("your sms has sent ",
+			"user id : ", s.CustomerID,
+			"msg : ", s.Text,
+			"number : ", v,
+			"operator:", "A")
+	}
+
+	return nil
+}

internal/operator/operatorB/service.go

@@ -2,9 +2,21 @@ package operatorB
 
 import (
 	"context"
+	"sms-gateway/app"
 	"sms-gateway/internal/model"
 )
 
 type OB struct{}
 
-func (o OB) Send(ctx context.Context, s model.SMS) {}
+func (o OB) Send(ctx context.Context, s model.SMS) error {
+
+	for _, v := range s.Recipients {
+		app.Logger.Info("your sms has sent ",
+			"user id : ", s.CustomerID,
+			"msg : ", s.Text,
+			"number : ", v,
+			"operator:", "B")
+	}
+
+	return nil
+}

internal/operator/service.go

@@ -2,9 +2,87 @@ package operator
 
 import (
 	"context"
+	"fmt"
+	"log/slog"
+	"time"
+
 	"sms-gateway/internal/model"
+	operatorA "sms-gateway/internal/operator/operatorA"
+	operatorB "sms-gateway/internal/operator/operatorB"
+	"sms-gateway/pkg/circuitbreaker"
 )
 
 type Operator interface {
-	Send(ctx context.Context, s model.SMS)
+	Send(ctx context.Context, s model.SMS) error
+}
+
+var (
+	primaryOperator  Operator = operatorA.OA{}
+	fallbackOperator Operator = operatorB.OB{}
+
+	operatorBreaker = circuitbreaker.New(circuitbreaker.Config{
+		FailureThreshold: 3,
+		SuccessThreshold: 2,
+		OpenTimeout:      5 * time.Second,
+	})
+
+	retryBackoff    = 200 * time.Millisecond
+	maxSendRetries  = 2
+	operatorTimeout = 2 * time.Second
+)
+
+func Send(ctx context.Context, s model.SMS) (string, error) {
+	if provider, err := dispatch(ctx, "operatorA", primaryOperator, operatorBreaker, s); err == nil {
+		return provider, nil
+	}
+
+	slog.Warn("primary operator failed, falling back")
+
+	return dispatch(ctx, "operatorB", fallbackOperator, nil, s)
+}
+
+func dispatch(ctx context.Context, name string, op Operator, breaker *circuitbreaker.Breaker, s model.SMS) (string, error) {
+	if breaker != nil {
+		if err := breaker.Allow(); err != nil {
+			return "", err
+		}
+	}
+
+	var lastErr error
+	backoff := retryBackoff
+
+	for attempt := 0; attempt <= maxSendRetries; attempt++ {
+		sendCtx, cancel := context.WithTimeout(ctx, operatorTimeout)
+		err := op.Send(sendCtx, s)
+		cancel()
+
+		if err == nil {
+			if breaker != nil {
+				breaker.MarkSuccess()
+			}
+			return name, nil
+		}
+
+		lastErr = err
+		if breaker != nil {
+			breaker.MarkFailure()
+		}
+
+		if attempt == maxSendRetries {
+			break
+		}
+
+		select {
+		case <-time.After(backoff):
+			backoff *= 2
+		case <-ctx.Done():
+			return "", ctx.Err()
+		}
+	}
+
+	if lastErr == nil {
+		return "", fmt.Errorf("%s failed without an explicit error", name)
+	}
+
+	return "", lastErr
 }

internal/sms/consumer.go

@@ -29,7 +29,7 @@ func StartConsumers(ctx context.Context) error {
 
 			app.Logger.Info("got message", "message", message)
 
-			if err := sendSmsToProvider(ctx, message); err != nil {
+			if err := sendSms(ctx, message); err != nil {
 				return err
 			}
 
@@ -48,13 +48,15 @@ func StartConsumers(ctx context.Context) error {
 		func(ctx context.Context, evt amqp091.Delivery) error {
 			var message model.SMS
 			if err := json.Unmarshal(evt.Body, &message); err != nil {
+				evt.Ack(false)
 				app.Logger.Error("cannot unmarshal sms", "error", err)
 				return fmt.Errorf("cannot unmarshal sms : %w", err)
 			}
 
 			app.Logger.Info("got message", "message", message)
 
-			if err := sendSmsToProvider(ctx, message); err != nil {
+			if err := sendSms(ctx, message); err != nil {
+				evt.Ack(false)
 				return err
 			}
 

internal/sms/service.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"sms-gateway/app"
 	"sms-gateway/internal/model"
+	"sms-gateway/internal/operator"
 	"strings"
 )
 
@@ -16,30 +17,41 @@ const (
 	Done           State = "done"
 )
 
-func sendSmsToProvider(ctx context.Context, s model.SMS) error {
+func sendSms(ctx context.Context, s model.SMS) error {
 	if err := UpdateSMS(ctx, s, Init); err != nil {
 		return err
 	}
 
-	if err := UpdateSMS(ctx, s, Done); err != nil {
+	provider, err := operator.Send(ctx, s)
+	if err != nil {
 		return err
 	}
+
+	if err := UpdateSMS(ctx, s, Done, provider); err != nil {
+		return err
+	}
+
 	return nil
 }
 
-func UpdateSMS(ctx context.Context, s model.SMS, state State) error {
+func UpdateSMS(ctx context.Context, s model.SMS, state State, provider ...string) error {
 	if len(s.Recipients) == 0 {
 		return errors.New("no  Recipients")
 	}
 
+	var providerName string
+	if len(provider) > 0 {
+		providerName = provider[0]
+	}
+
 	valueStrings := make([]string, 0, len(s.Recipients))
-	valueArgs := make([]any, 0, len(s.Recipients)*4)
+	valueArgs := make([]any, 0, len(s.Recipients)*5)
 	for _, recipient := range s.Recipients {
-		valueStrings = append(valueStrings, "(?, ?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)")
-		valueArgs = append(valueArgs, s.CustomerID, s.Type, state, recipient)
+		valueStrings = append(valueStrings, "(?, ?, ?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)")
+		valueArgs = append(valueArgs, s.CustomerID, s.Type, state, recipient, providerName)
 	}
-	query := `	INSERT INTO sms_status 
-				(user_id,type, status, recipient, created_at, updated_at) 
+	query := ` INSERT INTO sms_status 
+				(user_id,type, status, recipient, provider, created_at, updated_at) 
 				VALUES ` + strings.Join(valueStrings, ",")
 	if _, err := app.DB.ExecContext(ctx, query, valueArgs...); err != nil {
 		return err

pkg/circuitbreaker/circuitbreaker.go

@@ -0,0 +1,104 @@
+package circuitbreaker
+
+import (
+	"errors"
+	"sync"
+	"time"
+)
+
+type State int
+
+const (
+	StateClosed State = iota
+	StateOpen
+	StateHalfOpen
+)
+
+type Config struct {
+	FailureThreshold int
+	SuccessThreshold int
+	OpenTimeout      time.Duration
+}
+
+type Breaker struct {
+	mu             sync.Mutex
+	state          State
+	failureCount   int
+	successCount   int
+	reopenDeadline time.Time
+	cfg            Config
+}
+
+var ErrOpen = errors.New("circuit open")
+
+func New(cfg Config) *Breaker {
+	if cfg.FailureThreshold <= 0 {
+		cfg.FailureThreshold = 3
+	}
+	if cfg.SuccessThreshold <= 0 {
+		cfg.SuccessThreshold = 1
+	}
+	if cfg.OpenTimeout <= 0 {
+		cfg.OpenTimeout = 5 * time.Second
+	}
+
+	return &Breaker{cfg: cfg}
+}
+
+func (cb *Breaker) Allow() error {
+	cb.mu.Lock()
+	defer cb.mu.Unlock()
+
+	now := time.Now()
+	switch cb.state {
+	case StateOpen:
+		if now.Before(cb.reopenDeadline) {
+			return ErrOpen
+		}
+		cb.transitionTo(StateHalfOpen)
+	}
+
+	return nil
+}
+
+func (cb *Breaker) MarkSuccess() {
+	cb.mu.Lock()
+	defer cb.mu.Unlock()
+
+	if cb.state == StateHalfOpen {
+		cb.successCount++
+		if cb.successCount >= cb.cfg.SuccessThreshold {
+			cb.reset()
+		}
+		return
+	}
+
+	cb.reset()
+}
+
+func (cb *Breaker) MarkFailure() {
+	cb.mu.Lock()
+	defer cb.mu.Unlock()
+
+	cb.failureCount++
+	if cb.failureCount >= cb.cfg.FailureThreshold {
+		cb.trip()
+	}
+}
+
+func (cb *Breaker) transitionTo(state State) {
+	cb.state = state
+	cb.failureCount = 0
+	cb.successCount = 0
+	if state == StateOpen {
+		cb.reopenDeadline = time.Now().Add(cb.cfg.OpenTimeout)
+	}
+}
+
+func (cb *Breaker) reset() {
+	cb.transitionTo(StateClosed)
+}
+
+func (cb *Breaker) trip() {
+	cb.transitionTo(StateOpen)
+}