diff --git a/CLAUDE.md b/CLAUDE.md index 936121b25..31d32c8d1 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -150,6 +150,10 @@ protoc --proto_path=protos \ `app/pkg/clients/` contains the email notification client (separate from gRPC secondary adapters in domain layer). +### Reminder Batch Reporting + +The reminder pipeline is the only batch processor in the app. `ProcessReminders` (`internal/domains/notification/domain/reminder_service.go`) returns a `*BatchResult` describing per-item outcomes; the cronjob uses `HasInfraFailures()` to distinguish transient infrastructure problems from invalid-input failures for alerting. The retry loop short-circuits on a small list of non-retryable sentinels (validation errors, missing template) so retries aren't wasted on errors guaranteed to keep failing. Both lists live in `batch_result.go`; promote to a shared package only when a second consumer needs the same pattern. + ## Hexagonal Architecture Patterns ### Domain Layer (`domain/`) diff --git a/app/cmd/reminder/reminder.go b/app/cmd/reminder/reminder.go index d07e03e83..9586e8454 100644 --- a/app/cmd/reminder/reminder.go +++ b/app/cmd/reminder/reminder.go @@ -184,15 +184,22 @@ PASSWORDEXCHANGE_REMINDER_INTERVAL: Hours between reminders (1-720, default: 24) // Process reminders ctx := context.Background() - if err := reminderService.ProcessReminders(ctx, reminderConfig); err != nil { + result, err := reminderService.ProcessReminders(ctx, reminderConfig) + if err != nil { logging.Error(). Err(err). + Int("processedCount", result.SuccessCount). + Int("errorCount", result.FailureCount). + Bool("infraFailures", result.HasInfraFailures()). Str("operation", "process_reminders"). Msg("Failed to process reminders") return } logging.Info(). + Int("processedCount", result.SuccessCount). + Int("errorCount", result.FailureCount). + Int("totalProcessed", result.TotalProcessed). Str("operation", "processing_completed"). Msg("Reminder email processing completed") diff --git a/app/internal/domains/notification/domain/batch_result.go b/app/internal/domains/notification/domain/batch_result.go new file mode 100644 index 000000000..e6bcd6c92 --- /dev/null +++ b/app/internal/domains/notification/domain/batch_result.go @@ -0,0 +1,98 @@ +// Package domain – reminder batch reporting. +// +// BatchResult records per-item outcomes from ProcessReminders so callers can +// distinguish total failure from partial success, and infrastructure failures +// from invalid-input failures. Lives in this package because the reminder +// pipeline is the only batch processor in the app today; promote to a shared +// package only when a second consumer appears. +package domain + +import stderrors "errors" + +// BatchError captures one failed item in a reminder batch. +type BatchError struct { + ItemID string + Err error +} + +// BatchResult aggregates the outcome of a reminder batch. +type BatchResult struct { + TotalProcessed int + SuccessCount int + FailureCount int + Errors []BatchError +} + +// NewBatchResult returns an empty BatchResult. +func NewBatchResult() *BatchResult { return &BatchResult{} } + +// RecordSuccess increments the success counter. +func (r *BatchResult) RecordSuccess(itemID string) { + r.TotalProcessed++ + r.SuccessCount++ +} + +// RecordFailure appends a failure entry. +func (r *BatchResult) RecordFailure(itemID string, err error) { + r.TotalProcessed++ + r.FailureCount++ + r.Errors = append(r.Errors, BatchError{ItemID: itemID, Err: err}) +} + +// HasFailures reports whether any item failed. +func (r *BatchResult) HasFailures() bool { return r.FailureCount > 0 } + +// HasInfraFailures reports whether any failure looks like a transient +// infrastructure problem (queue, SMTP, storage). Useful for alerting: +// validation failures are not a system health concern, infra failures are. +func (r *BatchResult) HasInfraFailures() bool { + for _, e := range r.Errors { + if isInfraError(e.Err) { + return true + } + } + return false +} + +// nonRetryableSentinels are domain errors where retrying is guaranteed to +// keep failing — validation problems and missing-config conditions. Anything +// not in this list is treated as transient and eligible for retry. +var nonRetryableSentinels = []error{ + ErrInvalidEmailAddress, + ErrInvalidNotificationRequest, + ErrMessageUnmarshalFailed, + ErrEmptyMessageBody, + ErrTemplateNotFound, + ErrInvalidCheckAfterHours, + ErrInvalidMaxReminders, + ErrInvalidReminderInterval, +} + +// isRetryable reports whether retryWithBackoff should attempt err again. +func isRetryable(err error) bool { + for _, s := range nonRetryableSentinels { + if stderrors.Is(err, s) { + return false + } + } + return true +} + +// infraSentinels are domain errors that indicate transient infrastructure +// problems worth surfacing to monitoring. +var infraSentinels = []error{ + ErrEmailSendFailed, + ErrQueueConnectionFailed, + ErrQueueConsumeFailed, + ErrTemplateRenderFailed, + ErrSMTPAuthFailed, +} + +func isInfraError(err error) bool { + for _, s := range infraSentinels { + if stderrors.Is(err, s) { + return true + } + } + return false +} diff --git a/app/internal/domains/notification/domain/errors.go b/app/internal/domains/notification/domain/errors.go index 19a84c012..74fdc9260 100644 --- a/app/internal/domains/notification/domain/errors.go +++ b/app/internal/domains/notification/domain/errors.go @@ -3,33 +3,33 @@ package domain import "errors" var ( - // ErrInvalidEmailAddress indicates the email address is malformed + // ErrInvalidEmailAddress indicates the email address is malformed. ErrInvalidEmailAddress = errors.New("invalid email address") - // ErrEmailSendFailed indicates the email sending operation failed + // ErrEmailSendFailed indicates the email sending operation failed. ErrEmailSendFailed = errors.New("failed to send email") - // ErrQueueConnectionFailed indicates queue connection failed + // ErrQueueConnectionFailed indicates queue connection failed. ErrQueueConnectionFailed = errors.New("failed to connect to message queue") - // ErrQueueConsumeFailed indicates queue consumption failed + // ErrQueueConsumeFailed indicates queue consumption failed. ErrQueueConsumeFailed = errors.New("failed to consume from message queue") - // ErrTemplateRenderFailed indicates template rendering failed + // ErrTemplateRenderFailed indicates template rendering failed. ErrTemplateRenderFailed = errors.New("failed to render notification template") - // ErrTemplateNotFound indicates the requested template was not found + // ErrTemplateNotFound indicates the requested template was not found. ErrTemplateNotFound = errors.New("notification template not found") - // ErrInvalidNotificationRequest indicates the notification request is invalid + // ErrInvalidNotificationRequest indicates the notification request is invalid. ErrInvalidNotificationRequest = errors.New("invalid notification request") - // ErrSMTPAuthFailed indicates SMTP authentication failed + // ErrSMTPAuthFailed indicates SMTP authentication failed. ErrSMTPAuthFailed = errors.New("SMTP authentication failed") - // ErrMessageUnmarshalFailed indicates message unmarshaling failed + // ErrMessageUnmarshalFailed indicates message unmarshaling failed. ErrMessageUnmarshalFailed = errors.New("failed to unmarshal queue message") - // ErrEmptyMessageBody indicates the message body is empty + // ErrEmptyMessageBody indicates the message body is empty. ErrEmptyMessageBody = errors.New("message body is empty") ) diff --git a/app/internal/domains/notification/domain/reminder_service.go b/app/internal/domains/notification/domain/reminder_service.go index 1ca453a68..8b6760191 100644 --- a/app/internal/domains/notification/domain/reminder_service.go +++ b/app/internal/domains/notification/domain/reminder_service.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "strconv" "time" "github.com/Anthony-Bible/password-exchange/app/internal/domains/notification/ports/secondary" @@ -100,11 +101,17 @@ func NewReminderService( } } -// ProcessReminders finds and processes messages eligible for reminder emails -func (r *ReminderService) ProcessReminders(ctx context.Context, reminderConfig ReminderConfig) error { +// ProcessReminders finds and processes messages eligible for reminder emails. +// It returns a BatchResult describing per-item outcomes alongside any +// top-level error (config validation, fetch failure). When at least one +// message was processed successfully the error is nil even if other items +// failed — callers inspect the BatchResult to report partial success. +func (r *ReminderService) ProcessReminders(ctx context.Context, reminderConfig ReminderConfig) (*BatchResult, error) { + result := NewBatchResult() + // Check context cancellation early if err := ctx.Err(); err != nil { - return err + return result, err } // Validate configuration @@ -116,12 +123,12 @@ func (r *ReminderService) ProcessReminders(ctx context.Context, reminderConfig R Int("maxReminders", reminderConfig.MaxReminders). Int("reminderInterval", reminderConfig.Interval). Msg("Invalid reminder configuration") - return err + return result, err } if !reminderConfig.Enabled { r.logger.Info().Bool("enabled", false).Msg("Reminder system is disabled") - return nil + return result, nil } r.logger.Info(). @@ -145,21 +152,21 @@ func (r *ReminderService) ProcessReminders(ctx context.Context, reminderConfig R return err }, "get_unviewed_messages") if err != nil { - return fmt.Errorf("failed to get unviewed messages: %w", err) + return result, fmt.Errorf("failed to get unviewed messages: %w", err) } r.logger.Info().Int("count", len(messages)).Msg("Found messages eligible for reminders") if len(messages) == 0 { r.logger.Info().Msg("No messages found requiring reminders") - return nil + return result, nil } // Process each message with individual error recovery // Strategy: Continue processing other messages even if some fail (graceful degradation) - processedCount := 0 - errorCount := 0 for _, message := range messages { + itemID := strconv.Itoa(message.MessageID) + // Create reminder request reminderReq := ReminderRequest{ MessageID: message.MessageID, @@ -174,7 +181,7 @@ func (r *ReminderService) ProcessReminders(ctx context.Context, reminderConfig R return r.ProcessMessageReminder(ctx, reminderReq) }, fmt.Sprintf("process_message_%d", message.MessageID)) if err != nil { - errorCount++ + result.RecordFailure(itemID, err) r.logger.Error(). Err(err). Int("messageID", message.MessageID). @@ -184,36 +191,36 @@ func (r *ReminderService) ProcessReminders(ctx context.Context, reminderConfig R Msg("Failed to process reminder for message after all retry attempts") continue // Continue processing other messages } - processedCount++ + result.RecordSuccess(itemID) } r.logger.Info(). Int("totalMessages", len(messages)). - Int("processedCount", processedCount). - Int("errorCount", errorCount). + Int("processedCount", result.SuccessCount). + Int("errorCount", result.FailureCount). Msg("Reminder processing completed") // Implement graceful degradation: return success if at least some messages were processed // This allows partial success rather than all-or-nothing failure - if processedCount > 0 { + if result.SuccessCount > 0 { r.logger.Info(). - Int("processedCount", processedCount). - Int("errorCount", errorCount). - Float64("successRate", float64(processedCount)/float64(len(messages))*100). + Int("processedCount", result.SuccessCount). + Int("errorCount", result.FailureCount). + Float64("successRate", float64(result.SuccessCount)/float64(len(messages))*100). Msg("Reminder processing completed with partial success") - return nil + return result, nil } // If no messages were processed and we had errors, this indicates a more serious issue - if errorCount > 0 { + if result.FailureCount > 0 { r.logger.Error(). - Int("errorCount", errorCount). + Int("errorCount", result.FailureCount). Int("totalMessages", len(messages)). Msg("Failed to process any reminder messages") - return fmt.Errorf("failed to process any of %d reminder messages", len(messages)) + return result, fmt.Errorf("failed to process any of %d reminder messages", len(messages)) } - return nil + return result, nil } // ProcessMessageReminder sends a reminder email for a specific message @@ -341,6 +348,18 @@ func (r *ReminderService) retryWithBackoff(ctx context.Context, operation func() } lastErr = err + + // Fail fast on validation / missing-config errors. Retrying them is + // guaranteed to keep failing, and we don't want to trip the circuit + // breaker on user-supplied invalid input. + if !isRetryable(err) { + r.logger.Debug(). + Err(err). + Str("operation", operationName). + Msg("Operation failed with non-retryable error; skipping retry") + return err + } + r.circuitBreaker.RecordFailure() // Don't retry on last attempt @@ -375,7 +394,7 @@ func (r *ReminderService) retryWithBackoff(ctx context.Context, operation func() Int("maxRetries", MaxRetries). Msg("Operation failed after all retry attempts") - return fmt.Errorf("%w: %s failed after %d attempts: %v", ErrMaxRetriesExceeded, operationName, MaxRetries, lastErr) + return fmt.Errorf("%w: %s failed after %d attempts: %w", ErrMaxRetriesExceeded, operationName, MaxRetries, lastErr) } // validateReminderConfig validates all reminder configuration parameters diff --git a/app/internal/domains/notification/domain/reminder_service_test.go b/app/internal/domains/notification/domain/reminder_service_test.go index fd56fdfbe..b61fdd358 100644 --- a/app/internal/domains/notification/domain/reminder_service_test.go +++ b/app/internal/domains/notification/domain/reminder_service_test.go @@ -253,7 +253,7 @@ func TestProcessReminders_DisabledConfig_ReturnsEarly(t *testing.T) { } // Act - err := service.ProcessReminders(ctx, config) + _, err := service.ProcessReminders(ctx, config) // Assert assert.NoError(t, err) @@ -277,7 +277,7 @@ func TestProcessReminders_InvalidConfig_ReturnsError(t *testing.T) { } // Act - err := service.ProcessReminders(ctx, config) + _, err := service.ProcessReminders(ctx, config) // Assert assert.Error(t, err) @@ -303,7 +303,7 @@ func TestProcessReminders_NoMessages_ReturnsSuccess(t *testing.T) { mockStorageRepo.On("GetUnviewedMessagesForReminders", ctx, 24, 3, 24).Return([]*UnviewedMessage{}, nil) // Act - err := service.ProcessReminders(ctx, config) + _, err := service.ProcessReminders(ctx, config) // Assert assert.NoError(t, err) @@ -329,7 +329,7 @@ func TestProcessReminders_StorageError_ReturnsError(t *testing.T) { mockStorageRepo.On("GetUnviewedMessagesForReminders", ctx, 24, 3, 24).Return(nil, storageError) // Act - err := service.ProcessReminders(ctx, config) + _, err := service.ProcessReminders(ctx, config) // Assert assert.Error(t, err) @@ -371,7 +371,7 @@ func TestProcessReminders_SuccessfulProcessing_ReturnsSuccess(t *testing.T) { mockNotificationPublisher.On("PublishNotification", ctx, mock.AnythingOfType("NotificationRequest")).Return(nil) // Act - err := service.ProcessReminders(ctx, config) + _, err := service.ProcessReminders(ctx, config) // Assert assert.NoError(t, err) @@ -396,7 +396,7 @@ func TestProcessReminders_ContextCancelled_ReturnsError(t *testing.T) { } // Act - err := service.ProcessReminders(ctx, config) + _, err := service.ProcessReminders(ctx, config) // Assert assert.Error(t, err) @@ -423,7 +423,7 @@ func TestProcessReminders_StorageTimeout_HandledGracefully(t *testing.T) { mockStorageRepo.On("GetUnviewedMessagesForReminders", ctx, 24, 3, 24).Return(nil, context.DeadlineExceeded) // Act - err := service.ProcessReminders(ctx, config) + _, err := service.ProcessReminders(ctx, config) // Assert assert.Error(t, err) @@ -559,7 +559,7 @@ func TestProcessReminders_LoggingFailure_ContinuesProcessing(t *testing.T) { mockNotificationPublisher.On("PublishNotification", ctx, mock.AnythingOfType("NotificationRequest")).Return(nil) // Act - err := service.ProcessReminders(ctx, config) + _, err := service.ProcessReminders(ctx, config) // Assert // Should succeed because at least one message was processed successfully @@ -587,7 +587,7 @@ func TestProcessReminders_CircuitBreakerOpen_StopsProcessing(t *testing.T) { service.circuitBreaker.lastFailureTime = time.Now() // Act - err := service.ProcessReminders(ctx, config) + _, err := service.ProcessReminders(ctx, config) // Assert assert.Error(t, err) @@ -659,7 +659,7 @@ func TestProcessReminders_MixedResults_ContinuesProcessing(t *testing.T) { })).Return(nil) // Act - err := service.ProcessReminders(ctx, config) + _, err := service.ProcessReminders(ctx, config) // Assert // Should succeed overall despite individual failures @@ -783,7 +783,7 @@ func TestProcessReminders_ReminderInterval_RespectedCorrectly(t *testing.T) { mockNotificationPublisher.On("PublishNotification", ctx, mock.AnythingOfType("NotificationRequest")).Return(nil) // Act - err := service.ProcessReminders(ctx, config) + _, err := service.ProcessReminders(ctx, config) // Assert assert.NoError(t, err) @@ -824,3 +824,96 @@ func TestProcessMessageReminder_ValidRequest_Success(t *testing.T) { mockStorageRepo.AssertExpectations(t) mockNotificationPublisher.AssertExpectations(t) } + +// TestProcessReminders_ReturnsBatchResultWithCounts verifies the BatchResult +// contains accurate per-item counters after a mixed-success batch. +func TestProcessReminders_ReturnsBatchResultWithCounts(t *testing.T) { + mockStorageRepo, mockNotificationPublisher, mockLogger, mockConfig, mockValidation := createTestMocks() + service := NewReminderService(mockStorageRepo, mockNotificationPublisher, mockLogger, mockConfig, mockValidation) + + ctx := context.Background() + config := ReminderConfig{Enabled: true, CheckAfterHours: 24, MaxReminders: 3, Interval: 24} + + messages := []*UnviewedMessage{ + {MessageID: 1, UniqueID: "u1", RecipientEmail: "a@example.com", DaysOld: 2, Created: time.Now().Add(-48 * time.Hour)}, + {MessageID: 2, UniqueID: "u2", RecipientEmail: "b@example.com", DaysOld: 2, Created: time.Now().Add(-48 * time.Hour)}, + } + mockStorageRepo.On("GetUnviewedMessagesForReminders", ctx, 24, 3, 24).Return(messages, nil) + mockStorageRepo.On("GetReminderHistory", ctx, 1).Return([]*ReminderLogEntry{}, nil) + mockStorageRepo.On("GetReminderHistory", ctx, 2).Return([]*ReminderLogEntry{}, nil) + mockStorageRepo.On("LogReminderSent", ctx, 1, "a@example.com").Return(nil) + mockStorageRepo.On("LogReminderSent", ctx, 2, "b@example.com").Return(nil) + mockNotificationPublisher.On("PublishNotification", ctx, mock.AnythingOfType("NotificationRequest")).Return(nil) + + result, err := service.ProcessReminders(ctx, config) + + assert.NoError(t, err) + assert.NotNil(t, result) + assert.Equal(t, 2, result.TotalProcessed) + assert.Equal(t, 2, result.SuccessCount) + assert.Equal(t, 0, result.FailureCount) + assert.False(t, result.HasFailures()) +} + +// TestProcessReminders_NonRetryableError_FailsFast verifies that a Business +// error from the publisher is NOT retried — publisher should be called once. +func TestProcessReminders_NonRetryableError_FailsFast(t *testing.T) { + mockStorageRepo, mockNotificationPublisher, mockLogger, mockConfig, mockValidation := createTestMocks() + service := NewReminderService(mockStorageRepo, mockNotificationPublisher, mockLogger, mockConfig, mockValidation) + + ctx := context.Background() + config := ReminderConfig{Enabled: true, CheckAfterHours: 24, MaxReminders: 3, Interval: 24} + + messages := []*UnviewedMessage{ + {MessageID: 1, UniqueID: "u1", RecipientEmail: "a@example.com", DaysOld: 2, Created: time.Now().Add(-48 * time.Hour)}, + } + mockStorageRepo.On("GetUnviewedMessagesForReminders", ctx, 24, 3, 24).Return(messages, nil) + mockStorageRepo.On("GetReminderHistory", ctx, 1).Return([]*ReminderLogEntry{}, nil) + // ErrInvalidNotificationRequest is categorized Business — must not retry. + mockNotificationPublisher.On("PublishNotification", ctx, mock.AnythingOfType("NotificationRequest")). + Return(ErrInvalidNotificationRequest).Once() + + result, err := service.ProcessReminders(ctx, config) + + // All messages failed → top-level error returned. + assert.Error(t, err) + assert.NotNil(t, result) + assert.Equal(t, 1, result.FailureCount) + assert.Equal(t, 0, result.SuccessCount) + mockNotificationPublisher.AssertExpectations(t) + // Critical assertion: publisher invoked exactly once, no retry. + mockNotificationPublisher.AssertNumberOfCalls(t, "PublishNotification", 1) +} + +// TestProcessReminders_OperationalError_RetriesAndRecords verifies that a +// retryable error is retried, and on persistent failure the BatchError +// carries the Operational category derived from the wrapped sentinel. +func TestProcessReminders_OperationalError_RetriesAndRecords(t *testing.T) { + mockStorageRepo, mockNotificationPublisher, mockLogger, mockConfig, mockValidation := createTestMocks() + service := NewReminderService(mockStorageRepo, mockNotificationPublisher, mockLogger, mockConfig, mockValidation) + + ctx := context.Background() + config := ReminderConfig{Enabled: true, CheckAfterHours: 24, MaxReminders: 3, Interval: 24} + + messages := []*UnviewedMessage{ + {MessageID: 1, UniqueID: "u1", RecipientEmail: "a@example.com", DaysOld: 2, Created: time.Now().Add(-48 * time.Hour)}, + } + mockStorageRepo.On("GetUnviewedMessagesForReminders", ctx, 24, 3, 24).Return(messages, nil) + mockStorageRepo.On("GetReminderHistory", ctx, 1).Return([]*ReminderLogEntry{}, nil) + // ErrEmailSendFailed is categorized Operational — should be retried. + mockNotificationPublisher.On("PublishNotification", ctx, mock.AnythingOfType("NotificationRequest")). + Return(ErrEmailSendFailed) + + result, err := service.ProcessReminders(ctx, config) + + assert.Error(t, err) // all attempts failed + assert.NotNil(t, result) + assert.Equal(t, 1, result.FailureCount) + assert.Len(t, result.Errors, 1) + assert.True(t, result.HasInfraFailures(), + "persistent retryable publisher failure should surface as an infra batch failure") + // Publisher should have been invoked MaxRetries times (full retry exhaustion). + mockNotificationPublisher.AssertNumberOfCalls(t, "PublishNotification", MaxRetries) + // Silence unused-import warning if errors pkg is otherwise unused. + _ = errors.New +} diff --git a/app/internal/domains/notification/domain/service.go b/app/internal/domains/notification/domain/service.go index 2cdf0414b..b453f07ae 100644 --- a/app/internal/domains/notification/domain/service.go +++ b/app/internal/domains/notification/domain/service.go @@ -175,7 +175,7 @@ func (s *NotificationService) validateNotificationRequest(req NotificationReques } // ProcessReminders finds and processes messages eligible for reminder emails -func (s *NotificationService) ProcessReminders(ctx context.Context, config ReminderConfig) error { +func (s *NotificationService) ProcessReminders(ctx context.Context, config ReminderConfig) (*BatchResult, error) { return s.reminderService.ProcessReminders(ctx, config) } diff --git a/app/internal/domains/notification/ports/primary/service.go b/app/internal/domains/notification/ports/primary/service.go index 10b2f0189..283e5e697 100644 --- a/app/internal/domains/notification/ports/primary/service.go +++ b/app/internal/domains/notification/ports/primary/service.go @@ -2,7 +2,7 @@ package primary import ( "context" - + "github.com/Anthony-Bible/password-exchange/app/internal/domains/notification/domain" ) @@ -10,19 +10,21 @@ import ( type NotificationServicePort interface { // SendNotification sends a notification using the configured sender SendNotification(ctx context.Context, req domain.NotificationRequest) (*domain.NotificationResponse, error) - + // StartMessageProcessing starts consuming messages from the queue and processing them StartMessageProcessing(ctx context.Context, queueConn domain.QueueConnection, concurrency int) error - + // Close closes any open connections Close() error } // ReminderServicePort defines the primary port for reminder operations type ReminderServicePort interface { - // ProcessReminders finds and processes messages eligible for reminder emails - ProcessReminders(ctx context.Context, config domain.ReminderConfig) error - + // ProcessReminders finds and processes messages eligible for reminder emails. + // Returns a BatchResult describing per-item outcomes plus a top-level error + // for fail-fast conditions (validation, fetch failure, total failure). + ProcessReminders(ctx context.Context, config domain.ReminderConfig) (*domain.BatchResult, error) + // ProcessMessageReminder sends a reminder email for a specific message ProcessMessageReminder(ctx context.Context, req domain.ReminderRequest) error -} \ No newline at end of file +} diff --git a/app/internal/integration/reminder/reminder_flow_integration_test.go b/app/internal/integration/reminder/reminder_flow_integration_test.go index e210753b8..710c3f0a0 100644 --- a/app/internal/integration/reminder/reminder_flow_integration_test.go +++ b/app/internal/integration/reminder/reminder_flow_integration_test.go @@ -119,7 +119,8 @@ func TestIntegration_ReminderPipeline_EndToEnd(t *testing.T) { reminderCfg := notificationDomain.ReminderConfig{Enabled: true, CheckAfterHours: 24, MaxReminders: 3, Interval: 24} // First run: eligible message reminded, too-recent skipped. - require.NoError(t, svc.ProcessReminders(context.Background(), reminderCfg)) + _, err := svc.ProcessReminders(context.Background(), reminderCfg) + require.NoError(t, err) recipients := publisher.recipients() assert.True(t, recipients["wanted@example.com"], "eligible recipient should receive a reminder") @@ -143,18 +144,20 @@ func TestIntegration_ReminderPipeline_IncrementsOnSecondRun(t *testing.T) { svc, _ := newPipeline(dbCfg) reminderCfg := notificationDomain.ReminderConfig{Enabled: true, CheckAfterHours: 24, MaxReminders: 3, Interval: 1} - require.NoError(t, svc.ProcessReminders(context.Background(), reminderCfg)) + _, err := svc.ProcessReminders(context.Background(), reminderCfg) + require.NoError(t, err) count, _ := reminderRow(t, db, "pipeline-repeat") require.Equal(t, 1, count) // Age the existing reminder so the interval has elapsed for the second run. - _, err := db.Exec(`UPDATE email_reminders er + _, err = db.Exec(`UPDATE email_reminders er JOIN messages m ON m.messageid = er.message_id SET er.last_reminder_sent = NOW() - INTERVAL 5 HOUR WHERE m.uniqueid = ?`, "pipeline-repeat") require.NoError(t, err) - require.NoError(t, svc.ProcessReminders(context.Background(), reminderCfg)) + _, err = svc.ProcessReminders(context.Background(), reminderCfg) + require.NoError(t, err) count, _ = reminderRow(t, db, "pipeline-repeat") assert.Equal(t, 2, count, "second eligible run should increment the reminder count") }