package service import ( "context" "errors" "fmt" "testing" "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" "go.uber.org/zap/zaptest/observer" ) type workerRepo struct { Repository job model.Notification saved *model.Notification finishErr error } func (r *workerRepo) Recover(context.Context, time.Time) error { return nil } func (r *workerRepo) Claim(context.Context, string, time.Time) (*model.Notification, error) { n := r.job return &n, nil } func (r *workerRepo) Finish(_ context.Context, n *model.Notification) error { r.saved = n return r.finishErr } type workerSender struct { calls int outcome email.Outcome } func (s *workerSender) Send(context.Context, string, model.Email) email.Outcome { s.calls++ return s.outcome } func TestWorkerTerminalAndRetryOutcomes(t *testing.T) { for _, tc := range []struct { name string expiry time.Time driver string attempts uint32 outcome model.State want model.State calls int }{ {"expired", time.Now().Add(-time.Minute), "resend", 1, model.Accepted, model.Expired, 0}, {"driver changed", time.Now().Add(time.Hour), "sendgrid", 1, model.Accepted, model.Failed, 0}, {"retry exhausted", time.Now().Add(time.Hour), "resend", 5, model.RetryWait, model.Failed, 1}, {"retry expiry", time.Now().Add(time.Second), "resend", 1, model.RetryWait, model.Expired, 1}, {"ambiguous", time.Now().Add(time.Hour), "resend", 1, model.Unknown, model.Unknown, 1}, {"invalid driver outcome", time.Now().Add(time.Hour), "resend", 1, model.Queued, model.Unknown, 1}, } { t.Run(tc.name, func(t *testing.T) { repo := &workerRepo{job: model.Notification{Driver: tc.driver, ExpiresAt: tc.expiry, Attempts: tc.attempts}} sender := &workerSender{outcome: email.Outcome{State: tc.outcome}} svc := New(repo, nil, sender, "resend", zap.NewNop()) if _, err := svc.Dispatch(context.Background(), "id"); err != nil { t.Fatal(err) } if repo.saved == nil || repo.saved.State != tc.want || sender.calls != tc.calls { t.Fatal(repo.saved, sender.calls) } }) } } func TestWorkerLogsProviderFailureOnce(t *testing.T) { for _, failPersistence := range []bool{false, true} { t.Run(fmt.Sprint(failPersistence), func(t *testing.T) { core, logs := observer.New(zap.WarnLevel) repo := &workerRepo{job: model.Notification{Driver: "resend", ExpiresAt: time.Now().Add(time.Hour), Attempts: 1, Email: model.Email{Address: "private@example.com", Subject: "Private subject", Text: "Private text"}}} if failPersistence { repo.finishErr = merrors.ServiceUnavailable("storage offline") } sender := &workerSender{outcome: email.Outcome{State: model.Failed, Code: "provider_authentication_failed", HTTPStatus: 403, ProviderError: "validation_error", ProviderMessage: "The sxchnge.com domain is not verified"}} svc := New(repo, nil, sender, "resend", zap.New(core)) if _, err := svc.Dispatch(context.Background(), "id"); !errors.Is(err, repo.finishErr) { t.Fatal("unexpected persistence error", err) } entries := logs.All() if len(entries) != 1 || entries[0].Message != "Email delivery attempt finished" { t.Fatal("expected one provider outcome log", entries) } fields := entries[0].ContextMap() if fields["provider_status"] != int64(403) || fields["provider_error"] != "validation_error" || fields["provider_message"] != "The sxchnge.com domain is not verified" { t.Fatal("provider diagnostic missing", fields) } if repo.saved.FailureCode != "provider_authentication_failed" { t.Fatal("public failure code changed", repo.saved.FailureCode) } for _, field := range []string{"address", "email", "subject", "text", "html", "parameters"} { if _, ok := fields[field]; ok { t.Fatalf("private notification field %s logged", field) } } }) } }