88 lines
2.8 KiB
Go
88 lines
2.8 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"time"
|
|
|
|
"git.sendico.io/tech/sxchnge/notifications/email"
|
|
"git.sendico.io/tech/sxchnge/notifications/model"
|
|
merrors "git.sendico.io/tech/sxchnge/shared/errors"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
// Dispatch consumes an opaque job ID from JetStream. Duplicate messages are safe.
|
|
func (s *Service) Dispatch(ctx context.Context, id string) (bool, error) {
|
|
now := time.Now().UTC()
|
|
if err := s.repo.Recover(ctx, now); err != nil {
|
|
return false, err
|
|
}
|
|
n, err := s.repo.Claim(ctx, id, now)
|
|
if errors.Is(err, merrors.ErrNoData) {
|
|
return false, nil
|
|
}
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
var outcome email.Outcome
|
|
switch {
|
|
case !n.ExpiresAt.After(now):
|
|
n.State = model.Expired
|
|
n.FailureCode = "expired"
|
|
case n.Driver != s.driver:
|
|
n.State = model.Failed
|
|
n.FailureCode = "driver_changed"
|
|
default:
|
|
deadline := now.Add(20 * time.Second)
|
|
if n.ExpiresAt.Before(deadline) {
|
|
deadline = n.ExpiresAt
|
|
}
|
|
sendCtx, cancel := context.WithDeadline(ctx, deadline)
|
|
outcome = s.sender.Send(sendCtx, n.ID.Hex(), n.Email)
|
|
cancel()
|
|
n.State = outcome.State
|
|
n.FailureCode = outcome.Code
|
|
if n.State == model.RetryWait {
|
|
delay := time.Duration(1<<min(n.Attempts, 8)) * time.Second
|
|
if outcome.RetryAfter > delay {
|
|
delay = outcome.RetryAfter
|
|
}
|
|
n.NextAttemptAt = time.Now().UTC().Add(delay)
|
|
if n.Attempts >= 5 {
|
|
n.State = model.Failed
|
|
n.FailureCode = "retry_exhausted"
|
|
} else if !n.NextAttemptAt.Before(n.ExpiresAt) {
|
|
n.State = model.Expired
|
|
n.FailureCode = "expired"
|
|
}
|
|
} else if n.State != model.Accepted && n.State != model.Failed && n.State != model.Unknown {
|
|
n.State = model.Unknown
|
|
n.FailureCode = "invalid_driver_outcome"
|
|
}
|
|
}
|
|
n.UpdatedAt = time.Now().UTC()
|
|
// Persist the result even when shutdown canceled the provider request.
|
|
finishCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
|
|
defer cancel()
|
|
fields := []zap.Field{zap.String("notification_id", n.ID.Hex()), zap.String("driver", n.Driver), zap.String("state", string(n.State)), zap.Uint32("attempts", n.Attempts), zap.String("failure_code", n.FailureCode)}
|
|
if outcome.HTTPStatus != 0 {
|
|
fields = append(fields, zap.Int("provider_status", outcome.HTTPStatus))
|
|
}
|
|
if outcome.ProviderError != "" {
|
|
fields = append(fields, zap.String("provider_error", outcome.ProviderError))
|
|
}
|
|
if outcome.ProviderMessage != "" {
|
|
fields = append(fields, zap.String("provider_message", outcome.ProviderMessage))
|
|
}
|
|
// Keep the provider outcome observable even if its local result write fails.
|
|
if n.State == model.Accepted {
|
|
s.logger.Info("Email accepted by provider", fields...)
|
|
} else {
|
|
s.logger.Warn("Email delivery attempt finished", fields...)
|
|
}
|
|
if err = s.repo.Finish(finishCtx, n); err != nil {
|
|
return true, err
|
|
}
|
|
return true, nil
|
|
}
|