305 lines
8.9 KiB
Go
305 lines
8.9 KiB
Go
package worker
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"testing"
|
|
"time"
|
|
|
|
"wucher/internal/config"
|
|
"wucher/internal/domain/auth"
|
|
"wucher/internal/queue"
|
|
)
|
|
|
|
type stubOutboxRepo struct {
|
|
messages map[string]*auth.EmailOutboxMessage
|
|
}
|
|
|
|
func newStubOutboxRepo(messages ...*auth.EmailOutboxMessage) *stubOutboxRepo {
|
|
repo := &stubOutboxRepo{messages: map[string]*auth.EmailOutboxMessage{}}
|
|
for _, message := range messages {
|
|
if message == nil {
|
|
continue
|
|
}
|
|
cp := *message
|
|
cp.ID = append([]byte(nil), message.ID...)
|
|
cp.Body = append([]byte(nil), message.Body...)
|
|
cp.Attributes = append([]byte(nil), message.Attributes...)
|
|
repo.messages[string(cp.ID)] = &cp
|
|
}
|
|
return repo
|
|
}
|
|
|
|
func (r *stubOutboxRepo) CreateEmailOutboxMessage(_ context.Context, message *auth.EmailOutboxMessage) error {
|
|
if message == nil {
|
|
return errors.New("nil message")
|
|
}
|
|
cp := *message
|
|
cp.ID = append([]byte(nil), message.ID...)
|
|
cp.Body = append([]byte(nil), message.Body...)
|
|
cp.Attributes = append([]byte(nil), message.Attributes...)
|
|
r.messages[string(cp.ID)] = &cp
|
|
return nil
|
|
}
|
|
|
|
func (r *stubOutboxRepo) ClaimPendingEmailOutboxMessages(_ context.Context, limit int, now time.Time, lockTTL time.Duration, workerID string) ([]auth.EmailOutboxMessage, error) {
|
|
if limit <= 0 {
|
|
limit = 1
|
|
}
|
|
if lockTTL <= 0 {
|
|
lockTTL = time.Minute
|
|
}
|
|
out := make([]auth.EmailOutboxMessage, 0, limit)
|
|
staleBefore := now.Add(-lockTTL)
|
|
for _, message := range r.messages {
|
|
if len(out) >= limit {
|
|
break
|
|
}
|
|
switch {
|
|
case message.Status == auth.EmailOutboxStatusPending && !message.AvailableAt.After(now):
|
|
case message.Status == auth.EmailOutboxStatusProcessing && message.LockedAt != nil && message.LockedAt.Before(staleBefore):
|
|
default:
|
|
continue
|
|
}
|
|
message.Status = auth.EmailOutboxStatusProcessing
|
|
message.Attempts++
|
|
ts := now
|
|
message.LockedAt = &ts
|
|
message.LockedBy = workerID
|
|
cp := *message
|
|
out = append(out, cp)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (r *stubOutboxRepo) MarkEmailOutboxMessagePublished(_ context.Context, id []byte, publishedAt time.Time) error {
|
|
message := r.messages[string(id)]
|
|
if message == nil {
|
|
return errors.New("not found")
|
|
}
|
|
message.Status = auth.EmailOutboxStatusPublished
|
|
message.PublishedAt = &publishedAt
|
|
message.LockedAt = nil
|
|
message.LockedBy = ""
|
|
message.LastError = ""
|
|
return nil
|
|
}
|
|
|
|
func (r *stubOutboxRepo) MarkEmailOutboxMessageRetry(_ context.Context, id []byte, availableAt time.Time, lastErr string) error {
|
|
message := r.messages[string(id)]
|
|
if message == nil {
|
|
return errors.New("not found")
|
|
}
|
|
message.Status = auth.EmailOutboxStatusPending
|
|
message.AvailableAt = availableAt
|
|
message.LockedAt = nil
|
|
message.LockedBy = ""
|
|
message.LastError = lastErr
|
|
return nil
|
|
}
|
|
|
|
func (r *stubOutboxRepo) MarkEmailOutboxMessageDead(_ context.Context, id []byte, failedAt time.Time, lastErr string) error {
|
|
message := r.messages[string(id)]
|
|
if message == nil {
|
|
return errors.New("not found")
|
|
}
|
|
message.Status = auth.EmailOutboxStatusDead
|
|
message.AvailableAt = failedAt
|
|
message.LockedAt = nil
|
|
message.LockedBy = ""
|
|
message.LastError = lastErr
|
|
return nil
|
|
}
|
|
|
|
func (r *stubOutboxRepo) CountPendingEmailOutboxMessages(_ context.Context, now time.Time) (int64, error) {
|
|
var count int64
|
|
for _, message := range r.messages {
|
|
if (message.Status == auth.EmailOutboxStatusPending && !message.AvailableAt.After(now)) || message.Status == auth.EmailOutboxStatusProcessing {
|
|
count++
|
|
}
|
|
}
|
|
return count, nil
|
|
}
|
|
|
|
type stubQueueProducer struct {
|
|
published []queue.OutboundMessage
|
|
err error
|
|
}
|
|
|
|
func (p *stubQueueProducer) Backend() string {
|
|
return "stub"
|
|
}
|
|
|
|
func (p *stubQueueProducer) Publish(_ context.Context, message queue.OutboundMessage) error {
|
|
if p.err != nil {
|
|
return p.err
|
|
}
|
|
p.published = append(p.published, message)
|
|
return nil
|
|
}
|
|
|
|
func newOutboxRecord(t *testing.T, job queue.EmailJob) *auth.EmailOutboxMessage {
|
|
t.Helper()
|
|
serializer := queue.NewJSONSerializer("v1", true, queue.QueueTypeStandard, "")
|
|
message, err := serializer.Serialize(context.Background(), job)
|
|
if err != nil {
|
|
t.Fatalf("serialize: %v", err)
|
|
}
|
|
record, err := auth.NewEmailOutboxMessage(time.Now().UTC(), message)
|
|
if err != nil {
|
|
t.Fatalf("new outbox message: %v", err)
|
|
}
|
|
return record
|
|
}
|
|
|
|
func TestOutboxDispatcherPublishSuccess(t *testing.T) {
|
|
record := newOutboxRecord(t, queue.EmailJob{
|
|
Type: "email_verify",
|
|
To: "user@example.com",
|
|
Subject: "Verify",
|
|
Body: "hello",
|
|
})
|
|
repo := newStubOutboxRepo(record)
|
|
producer := &stubQueueProducer{}
|
|
dispatcher := NewOutboxDispatcher(config.QueueConfig{
|
|
OutboxEnabled: true,
|
|
OutboxBatchSize: 10,
|
|
OutboxDispatchTimeout: time.Second,
|
|
OutboxLockTTL: time.Minute,
|
|
OutboxMaxAttempts: 3,
|
|
WorkerBackoffMin: time.Second,
|
|
WorkerBackoffMax: time.Minute,
|
|
}, repo, producer, nil)
|
|
|
|
processed, err := dispatcher.dispatchOnce(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("dispatch once: %v", err)
|
|
}
|
|
if processed != 1 {
|
|
t.Fatalf("expected one processed message, got %d", processed)
|
|
}
|
|
if len(producer.published) != 1 {
|
|
t.Fatalf("expected one published message, got %d", len(producer.published))
|
|
}
|
|
if repo.messages[string(record.ID)].Status != auth.EmailOutboxStatusPublished {
|
|
t.Fatalf("expected message published, got %s", repo.messages[string(record.ID)].Status)
|
|
}
|
|
if got := dispatcher.Metrics().Published; got != 1 {
|
|
t.Fatalf("expected published metric=1, got %d", got)
|
|
}
|
|
}
|
|
|
|
func TestOutboxDispatcherSkipRegisterVerifyEmail(t *testing.T) {
|
|
record := newOutboxRecord(t, queue.EmailJob{
|
|
MessageType: "email_verify",
|
|
To: "user@example.com",
|
|
Subject: "Verify",
|
|
TextBody: "hello",
|
|
Metadata: map[string]string{
|
|
"skip_send": "true",
|
|
},
|
|
Tags: map[string]string{
|
|
"source": "register",
|
|
},
|
|
})
|
|
repo := newStubOutboxRepo(record)
|
|
producer := &stubQueueProducer{}
|
|
dispatcher := NewOutboxDispatcher(config.QueueConfig{
|
|
OutboxEnabled: true,
|
|
OutboxBatchSize: 10,
|
|
OutboxDispatchTimeout: time.Second,
|
|
OutboxLockTTL: time.Minute,
|
|
OutboxMaxAttempts: 3,
|
|
WorkerBackoffMin: time.Second,
|
|
WorkerBackoffMax: time.Minute,
|
|
}, repo, producer, nil)
|
|
|
|
processed, err := dispatcher.dispatchOnce(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("dispatch once: %v", err)
|
|
}
|
|
if processed != 1 {
|
|
t.Fatalf("expected one processed message, got %d", processed)
|
|
}
|
|
if len(producer.published) != 0 {
|
|
t.Fatalf("expected zero published message, got %d", len(producer.published))
|
|
}
|
|
if repo.messages[string(record.ID)].Status != auth.EmailOutboxStatusPublished {
|
|
t.Fatalf("expected message published status after skip, got %s", repo.messages[string(record.ID)].Status)
|
|
}
|
|
}
|
|
|
|
func TestOutboxDispatcherRetryOnPublishFailure(t *testing.T) {
|
|
record := newOutboxRecord(t, queue.EmailJob{
|
|
Type: "password_reset",
|
|
To: "user@example.com",
|
|
Subject: "Reset",
|
|
Body: "reset",
|
|
})
|
|
repo := newStubOutboxRepo(record)
|
|
producer := &stubQueueProducer{err: errors.New("queue unavailable")}
|
|
dispatcher := NewOutboxDispatcher(config.QueueConfig{
|
|
OutboxEnabled: true,
|
|
OutboxBatchSize: 10,
|
|
OutboxLockTTL: time.Minute,
|
|
OutboxMaxAttempts: 3,
|
|
WorkerBackoffMin: time.Second,
|
|
WorkerBackoffMax: time.Minute,
|
|
}, repo, producer, nil)
|
|
|
|
processed, err := dispatcher.dispatchOnce(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("dispatch once: %v", err)
|
|
}
|
|
if processed != 1 {
|
|
t.Fatalf("expected one processed message, got %d", processed)
|
|
}
|
|
got := repo.messages[string(record.ID)]
|
|
if got.Status != auth.EmailOutboxStatusPending {
|
|
t.Fatalf("expected message returned to pending, got %s", got.Status)
|
|
}
|
|
if got.Attempts != 1 {
|
|
t.Fatalf("expected attempts=1, got %d", got.Attempts)
|
|
}
|
|
if !got.AvailableAt.After(time.Now().Add(-time.Second)) {
|
|
t.Fatalf("expected retry to schedule next attempt in the future")
|
|
}
|
|
if metrics := dispatcher.Metrics(); metrics.Retried != 1 {
|
|
t.Fatalf("expected retried metric=1, got %d", metrics.Retried)
|
|
}
|
|
}
|
|
|
|
func TestOutboxDispatcherMarksDeadAfterMaxAttempts(t *testing.T) {
|
|
record := newOutboxRecord(t, queue.EmailJob{
|
|
Type: "invite_set_password",
|
|
To: "user@example.com",
|
|
Subject: "Invite",
|
|
Body: "invite",
|
|
})
|
|
record.Attempts = 2
|
|
repo := newStubOutboxRepo(record)
|
|
producer := &stubQueueProducer{err: errors.New("queue unavailable")}
|
|
dispatcher := NewOutboxDispatcher(config.QueueConfig{
|
|
OutboxEnabled: true,
|
|
OutboxBatchSize: 10,
|
|
OutboxLockTTL: time.Minute,
|
|
OutboxMaxAttempts: 3,
|
|
WorkerBackoffMin: time.Second,
|
|
WorkerBackoffMax: time.Minute,
|
|
}, repo, producer, nil)
|
|
|
|
processed, err := dispatcher.dispatchOnce(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("dispatch once: %v", err)
|
|
}
|
|
if processed != 1 {
|
|
t.Fatalf("expected one processed message, got %d", processed)
|
|
}
|
|
if got := repo.messages[string(record.ID)].Status; got != auth.EmailOutboxStatusDead {
|
|
t.Fatalf("expected dead status, got %s", got)
|
|
}
|
|
if got := dispatcher.Metrics().Dead; got != 1 {
|
|
t.Fatalf("expected dead metric=1, got %d", got)
|
|
}
|
|
}
|