Hello
', 0, 1, 'email')`, f.step, f.campaign, f.org) + exec(`INSERT INTO contacts (id, user_id, organization_id, email, first_name, last_name, company, phone, custom_fields) + VALUES ($1, $2, $3, $4, 'Live', 'Contact', '', '', '{}')`, + f.contact, f.user, f.org, "lead-"+f.contact.String()[:8]+"@test.local") + exec(`INSERT INTO campaign_leads (campaign_id, contact_id, position) VALUES ($1, $2, 0)`, f.campaign, f.contact) + + t.Cleanup(func() { + c := context.Background() + for _, step := range []struct { + sql string + arg any + }{ + {`DELETE FROM task_failures WHERE task_id IN (SELECT id FROM tasks WHERE email_account_id = $1)`, f.mailbox}, + {`DELETE FROM campaign_tasks WHERE task_id IN (SELECT id FROM tasks WHERE email_account_id = $1)`, f.mailbox}, + {`DELETE FROM tasks WHERE email_account_id = $1`, f.mailbox}, + {`DELETE FROM campaign_logs WHERE campaign_id = $1`, f.campaign}, + {`DELETE FROM campaign_daily_sends WHERE campaign_id = $1`, f.campaign}, + {`DELETE FROM campaign_contact_progress WHERE campaign_id = $1`, f.campaign}, + {`DELETE FROM campaign_leads WHERE campaign_id = $1`, f.campaign}, + {`DELETE FROM sequences WHERE campaign_id = $1`, f.campaign}, + {`DELETE FROM campaigns WHERE id = $1`, f.campaign}, + {`DELETE FROM email_accounts WHERE id = $1`, f.mailbox}, + {`DELETE FROM contacts WHERE organization_id = $1`, f.org}, + {`DELETE FROM organizations WHERE id = $1`, f.org}, + {`DELETE FROM users WHERE id = $1`, f.user}, + } { + if _, err := pool.Exec(c, step.sql, step.arg); err != nil { + t.Errorf("cleanup %q: %v", step.sql, err) + } + } + }) + return f +} + +// stampSend does what the backend's campaign tick does after handing a send +// to the worker: a completed task, the progress stamp, and the day's counters. +func (f *sendResultFixture) stampSend(t *testing.T, s *JobsService) uuid.UUID { + t.Helper() + ctx := context.Background() + taskID := uuid.New() + now := time.Now() + created, err := s.TaskRepo.CreateTaskWithLock(ctx, &repository.Task{ + ID: taskID, TaskType: "campaign", EmailAccountID: f.mailbox, Status: "pending", ScheduledAt: &now, + }, &repository.CampaignTask{TaskID: taskID, CampaignID: &f.campaign}) + if err != nil || !created { + t.Fatalf("create task: created=%v err=%v", created, err) + } + if err := s.TaskRepo.UpdateCampaignTaskTracking(ctx, taskID, f.contact, f.step); err != nil { + t.Fatalf("tracking: %v", err) + } + if err := s.CampaignProgressRepo.RecordEmailSent(ctx, f.campaign, f.contact, f.step); err != nil { + t.Fatalf("record sent: %v", err) + } + if err := s.CampaignRepo.IncrementCampaignDailySend(ctx, f.campaign, true); err != nil { + t.Fatalf("increment daily: %v", err) + } + if err := s.TaskRepo.UpdateTaskStatusWithLock(ctx, taskID, "completed"); err != nil { + t.Fatalf("complete task: %v", err) + } + return taskID +} + +func TestLiveHandleEmailFailedWalksBackAndRetriesUntilCap(t *testing.T) { + dsn := os.Getenv("WARMBLY_TEST_DB") + if dsn == "" { + t.Skip("WARMBLY_TEST_DB not set") + } + ctx := context.Background() + handle, err := db.New(ctx, dsn) + if err != nil { + t.Fatalf("connect: %v", err) + } + t.Cleanup(func() { handle.Pool.Close() }) + + f := newSendResultFixture(t, handle) + s := &JobsService{ + TaskRepo: repository.NewTaskRepository(handle.Pool), + CampaignRepo: repository.NewCampaignRepostory(handle), + CampaignProgressRepo: repository.NewCampaignProgressRepository(handle.Pool), + CampaignLogRepo: repository.NewCampaignLogRepository(handle), + ContactRepo: repository.NewContactRepostory(handle), + } + + var sentAt *time.Time + var attempts int + readProgress := func() { + t.Helper() + if err := handle.Pool.QueryRow(ctx, `SELECT sent_at, send_attempts FROM campaign_contact_progress + WHERE campaign_id = $1 AND contact_id = $2 AND sequence_id = $3`, f.campaign, f.contact, f.step). + Scan(&sentAt, &attempts); err != nil { + t.Fatalf("read progress: %v", err) + } + } + nextPair := func() *repository.ContactSequencePair { + t.Helper() + pair, _, err := s.CampaignProgressRepo.FindNextRoutedPair(ctx, f.campaign, "created_at", "asc", "", false, false) + if err != nil { + t.Fatalf("next pair: %v", err) + } + return pair + } + + // First failure: the stamped step is walked back, the day's counters give + // the send back, the task is failed, the completed campaign reopens, and + // the step is routable again. + taskID := f.stampSend(t, s) + readProgress() + if sentAt == nil { + t.Fatal("precondition: step should be stamped sent") + } + if nextPair() != nil { + t.Fatal("precondition: a stamped single-step lead has nothing left to route") + } + if err := s.HandleEmailFailed(ctx, models.SendEmailResult{ + TaskID: taskID, Success: false, + Error: &models.EmailSendError{Code: "UNSUPPORTED", Message: "worker could not send"}, + }); err != nil { + t.Fatalf("handle failed: %v", err) + } + readProgress() + if sentAt != nil || attempts != 1 { + t.Fatalf("after first failure: sent_at=%v attempts=%d, want NULL/1", sentAt, attempts) + } + var taskStatus string + if err := handle.Pool.QueryRow(ctx, `SELECT status FROM tasks WHERE id = $1`, taskID).Scan(&taskStatus); err != nil { + t.Fatal(err) + } + if taskStatus != "failed" { + t.Fatalf("task status = %s, want failed", taskStatus) + } + var sent, newLeads int + if err := handle.Pool.QueryRow(ctx, `SELECT emails_sent, new_leads_started FROM campaign_daily_sends + WHERE campaign_id = $1 AND send_date = CURRENT_DATE`, f.campaign).Scan(&sent, &newLeads); err != nil { + t.Fatal(err) + } + if sent != 0 || newLeads != 0 { + t.Fatalf("daily counters after walk-back: sent=%d new=%d, want 0/0", sent, newLeads) + } + var campaignStatus string + if err := handle.Pool.QueryRow(ctx, `SELECT status FROM campaigns WHERE id = $1`, f.campaign).Scan(&campaignStatus); err != nil { + t.Fatal(err) + } + if campaignStatus != "active" { + t.Fatalf("campaign status = %s, want active (reopened)", campaignStatus) + } + if pair := nextPair(); pair == nil || pair.SequenceID != f.step || !pair.IsNewLead { + t.Fatalf("after walk-back the step should be routable as a new lead, got %+v", pair) + } + var logs int + if err := handle.Pool.QueryRow(ctx, `SELECT COUNT(*) FROM campaign_logs + WHERE campaign_id = $1 AND event_type = 'email_failed' AND metadata->>'level' = 'error'`, f.campaign).Scan(&logs); err != nil { + t.Fatal(err) + } + if logs != 1 { + t.Fatalf("campaign log entries = %d, want 1", logs) + } + + // A duplicate result for the same task is a no-op. + if err := s.HandleEmailFailed(ctx, models.SendEmailResult{TaskID: taskID, LegacyErrorMsg: "again"}); err != nil { + t.Fatalf("duplicate: %v", err) + } + readProgress() + if attempts != 1 { + t.Fatalf("duplicate result changed attempts to %d", attempts) + } + + // Keep failing: the step stays routable until the cap, then the lead is + // dropped and marked failed. + for i := 2; i <= config.CampaignSendMaxAttempts; i++ { + id := f.stampSend(t, s) + if err := s.HandleEmailFailed(ctx, models.SendEmailResult{TaskID: id, LegacyErrorMsg: "still broken"}); err != nil { + t.Fatalf("attempt %d: %v", i, err) + } + readProgress() + if attempts != i { + t.Fatalf("attempt %d recorded as %d", i, attempts) + } + if i < config.CampaignSendMaxAttempts && nextPair() == nil { + t.Fatalf("after attempt %d the step should still be routable", i) + } + } + if nextPair() != nil { + t.Fatal("after the attempt cap the lead should be dropped from routing") + } + counts, xerr := s.ContactRepo.CampaignLeadCounts(ctx, f.org.String(), f.campaign.String()) + if xerr != nil { + t.Fatalf("lead counts: %v", xerr) + } + if counts.Failed != 1 || counts.Queued != 0 || counts.Total != 1 { + t.Fatalf("lead counts = %+v, want failed=1", counts) + } +} diff --git a/internal/app/consumer/events.go b/internal/app/consumer/events.go index 2e6ebd21..e96f4b40 100644 --- a/internal/app/consumer/events.go +++ b/internal/app/consumer/events.go @@ -39,6 +39,11 @@ func (w *JobsService) InitEvents() { Register(w, models.JobEventTypeSyncState, w.HandleSyncState) Register(w, models.JobEventTypeTokenUpdate, w.HandleTokenUpdate) + // Send outcomes. A worker reports every send it was handed; a failure + // walks back what the control plane stamped at hand-off. + Register(w, models.JobEventTypeEmailSent, w.HandleEmailSent) + Register(w, models.JobEventTypeEmailFailed, w.HandleEmailFailed) + // Email error handlers Register(w, models.JobEventTypeEmailAuthError, w.HandleEmailAuthError) Register(w, models.JobEventTypeEmailDisabled, w.HandleEmailDisabled) diff --git a/internal/app/consumer/service.go b/internal/app/consumer/service.go index a4f3a97c..f8381e5d 100644 --- a/internal/app/consumer/service.go +++ b/internal/app/consumer/service.go @@ -59,6 +59,16 @@ type JobsService struct { // dead-worker job moves or strands their mailboxes. Nil disables it. Notifier OrgNotifier + // Send-outcome handling (EMAIL_SENT / EMAIL_FAILED from workers). The task + // and campaign progress the control plane stamped at hand-off are walked + // back here when a worker reports it could not send. Nil TaskRepo disables + // the handlers. + TaskRepo repository.TaskRepository + CampaignRepo repository.CampaignRepository + CampaignProgressRepo repository.CampaignProgressRepository + CampaignLogRepo repository.CampaignLogRepository + ContactRepo repository.ContactRepository + eventHandlers map[models.JobEventType]func(ctx context.Context, body any) error } diff --git a/internal/app/email/loader.go b/internal/app/email/loader.go index 956d79e9..609e03ea 100644 --- a/internal/app/email/loader.go +++ b/internal/app/email/loader.go @@ -86,6 +86,29 @@ func (s *emailService) reconcileWorkerAccounts(ctx context.Context, lastPublishe } } +// ReloadWorkerAccounts publishes every active mailbox assigned to workerID +// back onto it. Called from the boot heartbeat, so a restarted worker is +// sending and syncing again within seconds rather than after the reconciler's +// next republish window. +func (s *emailService) ReloadWorkerAccounts(ctx context.Context, workerID uuid.UUID) { + ids, err := s.emailRepository.ListActiveAccountsByWorker(ctx, workerID) + if err != nil { + log.Warn().Err(err).Str("worker_id", workerID.String()).Msg("worker boot reload: list accounts failed") + return + } + loaded := 0 + for _, id := range ids { + if err := s.LoadAccountOntoWorker(ctx, id); err != nil { + log.Warn().Err(err).Str("email_id", id.String()).Str("worker_id", workerID.String()).Msg("worker boot reload: load account failed") + continue + } + loaded++ + } + if len(ids) > 0 { + log.Info().Str("worker_id", workerID.String()).Int("loaded", loaded).Int("assigned", len(ids)).Msg("worker boot reload: mailboxes re-shipped") + } +} + // loadAccountBestEffort loads a freshly onboarded account onto its worker without // blocking or failing the onboarding response; the reconciler is the safety net. func (s *emailService) loadAccountBestEffort(ctx context.Context, accountID uuid.UUID) { diff --git a/internal/app/email/service.go b/internal/app/email/service.go index 3fecd4c3..17c783d4 100644 --- a/internal/app/email/service.go +++ b/internal/app/email/service.go @@ -70,6 +70,9 @@ type EmailService interface { // StartWorkerReconciler periodically ensures every active mailbox is // assigned to a worker and loaded onto it (blocks until ctx is cancelled). StartWorkerReconciler(ctx context.Context, interval time.Duration) + // ReloadWorkerAccounts re-ships every active mailbox assigned to one + // worker, for a worker that just booted and holds none in memory. + ReloadWorkerAccounts(ctx context.Context, workerID uuid.UUID) } type emailService struct { diff --git a/internal/app/worker/event_send_email.go b/internal/app/worker/event_send_email.go index d16def42..5b253e5f 100644 --- a/internal/app/worker/event_send_email.go +++ b/internal/app/worker/event_send_email.go @@ -3,6 +3,7 @@ package worker import ( "bytes" "context" + "errors" "fmt" "io" "time" @@ -29,9 +30,13 @@ func (w *WorkerService) HandleSendEmail(ctx context.Context, sendEmail models.Se w.mailManager.RUnlock() if !exists { + // The mailbox is not loaded here: it is still being added (its + // ADD_EMAIL is queued behind this send), or this worker restarted and + // the reconciler has not re-shipped it yet. Leave the send for + // redelivery a few times so a queued ADD_EMAIL gets processed first, + // then report the failure so the control plane retries the step. err := fmt.Errorf("email account %s not found in worker", sendEmail.EmailID.String()) - w.sendEmailFailure(sendEmail.TaskID, sendEmail.EmailID, mail, err.Error()) - return err + return w.failSend(ctx, sendEmail, err.Error(), true) } // Decrypt subject @@ -50,8 +55,7 @@ func (w *WorkerService) HandleSendEmail(ctx context.Context, sendEmail models.Se bodyPlain, bodyHTML, attachmentRefs, err := w.fetchEmailBody(ctx, sendEmail.OrgID, sendEmail.BodyS3Key) if err != nil { log.Error().Err(err).Str("s3_key", sendEmail.BodyS3Key).Msg("Failed to fetch email body from S3") - w.sendEmailFailure(sendEmail.TaskID, sendEmail.EmailID, mail, fmt.Sprintf("failed to fetch email body: %v", err)) - return err + return w.failSend(ctx, sendEmail, fmt.Sprintf("failed to fetch email body: %v", err), true) } // Fetch each attachment's bytes from object storage by key. A fetch failure @@ -59,8 +63,7 @@ func (w *WorkerService) HandleSendEmail(ctx context.Context, sendEmail models.Se attachments, err := w.fetchAttachments(ctx, attachmentRefs) if err != nil { log.Error().Err(err).Str("task_id", sendEmail.TaskID.String()).Msg("Failed to fetch attachment bytes from S3") - w.sendEmailFailure(sendEmail.TaskID, sendEmail.EmailID, mail, fmt.Sprintf("failed to fetch attachment: %v", err)) - return err + return w.failSend(ctx, sendEmail, fmt.Sprintf("failed to fetch attachment: %v", err), true) } // Use unified Send method @@ -220,7 +223,30 @@ func (w *WorkerService) sendEmailSuccess(taskID uuid.UUID, messageID, providerMs } } -// sendEmailError sends a structured error result back to the jobs service +// sendNotLoadedRedeliveries is how many bus deliveries a send may burn waiting +// for a transient condition (mailbox not loaded yet, object storage blip) +// before the worker reports it failed. Each redelivery is a second apart, and +// the bus stops redelivering at ten, so this stays well inside that. +const sendNotLoadedRedeliveries = 5 + +// failSend reports a send the worker could not attempt. A retryable condition +// is first left for bus redelivery (returning an error naks the message) so a +// queued ADD_EMAIL or a storage blip can clear; once the redeliveries are +// spent, when the bus does not redeliver, or when the condition is not +// retryable, the failure result is produced and the message is acked, handing +// the retry to the control plane. +func (w *WorkerService) failSend(ctx context.Context, sendEmail models.SendEmail, reason string, retryable bool) error { + if d := deliveryOf(ctx); retryable && d.redelivers && d.attempt < sendNotLoadedRedeliveries { + return errors.New(reason) + } + w.sendEmailFailure(sendEmail.TaskID, sendEmail.EmailID, nil, reason) + return nil +} + +// sendEmailError reports a failed send attempt. The per-task result is always +// an EMAIL_FAILED so the control plane has one result channel to walk the +// send back on; account-level conditions (auth, disabled, rate limit, server +// error) additionally raise their own typed event carrying the full context. func (w *WorkerService) sendEmailError(taskID uuid.UUID, emailID uuid.UUID, mail *wmail.WMail, mailErr *errx.MailError) { // Determine the appropriate event type based on error eventType := wmail.DetermineErrorEventType(mailErr) @@ -236,14 +262,15 @@ func (w *WorkerService) sendEmailError(taskID uuid.UUID, emailID uuid.UUID, mail SentAt: time.Now(), } - if err := w.Produce(eventType, taskID.String(), result); err != nil { - log.Error().Err(err).Str("task_id", taskID.String()).Msg("Failed to produce email error event") + if err := w.Produce(models.JobEventTypeEmailFailed, taskID.String(), result); err != nil { + log.Error().Err(err).Str("task_id", taskID.String()).Msg("Failed to produce email failed event") } - // For critical auth/disabled errors, also send a separate error event with full context + // Account-level conditions also raise their typed event with full context if eventType == models.JobEventTypeEmailAuthError || eventType == models.JobEventTypeEmailDisabled || - eventType == models.JobEventTypeEmailRateLimited { + eventType == models.JobEventTypeEmailRateLimited || + eventType == models.JobEventTypeEmailServerError { userInfo := mailErr.GetUserErrorInfo() errorEvent := models.EmailErrorEvent{ diff --git a/internal/app/worker/heartbeat.go b/internal/app/worker/heartbeat.go index 262559d1..f0e22942 100644 --- a/internal/app/worker/heartbeat.go +++ b/internal/app/worker/heartbeat.go @@ -7,6 +7,13 @@ import ( ) func (s *WorkerService) Heartbeat(ctx context.Context) { + // Beat at once: the control plane refuses to hand sends to a worker with + // no heartbeat key, so waiting a full interval would make a fresh worker + // unusable for its first 90 seconds. + if err := s.heartbeat(ctx); err != nil { + log.Println("Failed to do heartbeat", err) + } + ticker := time.NewTicker(90 * time.Second) defer ticker.Stop() diff --git a/internal/app/worker/receive.go b/internal/app/worker/receive.go index cbf1a7db..7b69f75d 100644 --- a/internal/app/worker/receive.go +++ b/internal/app/worker/receive.go @@ -8,6 +8,21 @@ import ( "github.com/warmbly/warmbly/internal/models" ) +type deliveryKey struct{} + +// delivery is what a handler may know about the bus message it is processing. +type delivery struct { + attempt int // 1-based delivery count; 0 when unknown + redelivers bool // a handler error gets the message delivered again +} + +// deliveryOf returns the bus delivery details for the message a handler is +// processing (zero value when the handler was not invoked from Receive). +func deliveryOf(ctx context.Context) delivery { + d, _ := ctx.Value(deliveryKey{}).(delivery) + return d +} + // Receive is the eventbus.Handler that drives the worker's event loop. It // decodes the wire payload via the injected codec.Codec and dispatches to // HandleEvent. @@ -17,7 +32,7 @@ func (w *WorkerService) Receive(ctx context.Context, msg eventbus.Message) error return err } - hctx, cancel := context.WithTimeout(ctx, 30*time.Second) + hctx, cancel := context.WithTimeout(context.WithValue(ctx, deliveryKey{}, delivery{attempt: msg.Attempt, redelivers: msg.Redelivers}), 30*time.Second) defer cancel() return w.HandleEvent(hctx, &event) } diff --git a/internal/app/worker/wmail/send.go b/internal/app/worker/wmail/send.go index 69123bab..6187bb61 100644 --- a/internal/app/worker/wmail/send.go +++ b/internal/app/worker/wmail/send.go @@ -422,7 +422,7 @@ func DetermineErrorEventType(err *errx.MailError) models.JobEventType { } switch err.Code { - case errx.MailErrorCodeGoogleAuth, errx.MailErrorCodeAuthenticationFailed: + case errx.MailErrorCodeGoogleAuth, errx.MailErrorCodeAuthenticationFailed, errx.MailErrorCodeInvalidCredentials: return models.JobEventTypeEmailAuthError case errx.MailErrorCodeAccountSuspended, errx.MailErrorCodeAuthorizationFailed: diff --git a/internal/config/constants.go b/internal/config/constants.go index 41ee05a8..66e8db6a 100644 --- a/internal/config/constants.go +++ b/internal/config/constants.go @@ -73,6 +73,12 @@ const ( // delay that the scheduler would then turn into an unreachable send time. SequenceWaitAfterMax = 60 + // CampaignSendMaxAttempts is how many times one (contact, step) may be + // handed to a worker before the lead is marked failed and dropped from the + // campaign. A worker-reported failure clears the step's sent_at so the next + // tick retries it; this bounds that loop for a mailbox that can never send. + CampaignSendMaxAttempts = 5 + // Webhook/integration fan-out throttle. Caps how many events of a single // type one org can fan out to its webhooks + integration sinks // (Slack/Discord/CRM) per minute — the backstop against a campaign "notify" diff --git a/internal/events/publisher.go b/internal/events/publisher.go index 803535f4..d865af7f 100644 --- a/internal/events/publisher.go +++ b/internal/events/publisher.go @@ -92,6 +92,14 @@ func (p *publisher) PublishSendEmail(ctx context.Context, workerID uuid.UUID, pa // Store email body (and attachment refs) in S3. The attachment refs ride // inside the emsg blob so the worker receives them via BodyS3Key without any // change to the Avro event contract. + if p.bus == nil { + return fmt.Errorf("event bus not configured; cannot hand send %s to a worker", params.TaskID) + } + if p.storageClient == nil { + // The worker reads the body from object storage; without it the send + // would be published body-less and fail there. + return fmt.Errorf("object storage not configured; cannot hand send %s to a worker", params.TaskID) + } s3Key, err := p.storeEmailBody(ctx, params.TaskID, params.OrgID, params.BodyPlain, params.BodyHTML, params.Attachments) if err != nil { return fmt.Errorf("failed to store email body: %w", err) diff --git a/internal/infrastructure/db/migrations/000089_campaign_send_outcome.down.sql b/internal/infrastructure/db/migrations/000089_campaign_send_outcome.down.sql new file mode 100644 index 00000000..0aa6e923 --- /dev/null +++ b/internal/infrastructure/db/migrations/000089_campaign_send_outcome.down.sql @@ -0,0 +1,4 @@ +ALTER TABLE campaign_contact_progress + DROP COLUMN IF EXISTS failure_reason, + DROP COLUMN IF EXISTS failed_at, + DROP COLUMN IF EXISTS send_attempts; diff --git a/internal/infrastructure/db/migrations/000089_campaign_send_outcome.up.sql b/internal/infrastructure/db/migrations/000089_campaign_send_outcome.up.sql new file mode 100644 index 00000000..6e9c445e --- /dev/null +++ b/internal/infrastructure/db/migrations/000089_campaign_send_outcome.up.sql @@ -0,0 +1,8 @@ +-- Worker send outcomes close the loop on campaign sends. A step is stamped +-- sent_at when the control plane hands it to a worker; when the worker reports +-- that the send failed, sent_at is cleared again so the step is retried, and the +-- failure is kept here so the lead can be shown as failed once retries run out. +ALTER TABLE campaign_contact_progress + ADD COLUMN IF NOT EXISTS send_attempts integer NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS failed_at timestamptz, + ADD COLUMN IF NOT EXISTS failure_reason text NOT NULL DEFAULT ''; diff --git a/internal/infrastructure/db/migrations/000090_connect_wizard_steps.down.sql b/internal/infrastructure/db/migrations/000090_connect_wizard_steps.down.sql new file mode 100644 index 00000000..450b663f --- /dev/null +++ b/internal/infrastructure/db/migrations/000090_connect_wizard_steps.down.sql @@ -0,0 +1 @@ +-- Connections are ordinary step data the user can edit; there is nothing to undo. diff --git a/internal/infrastructure/db/migrations/000090_connect_wizard_steps.up.sql b/internal/infrastructure/db/migrations/000090_connect_wizard_steps.up.sql new file mode 100644 index 00000000..d18586f1 --- /dev/null +++ b/internal/infrastructure/db/migrations/000090_connect_wizard_steps.up.sql @@ -0,0 +1,28 @@ +-- Campaigns created by the wizard before connections were written at creation +-- hold their follow-ups as disconnected steps, so only the first email ever +-- sent. Link the steps of every campaign that has two or more email steps and +-- no connection anywhere, in position order, which is the sequence the wizard +-- showed ("Follow-up 1, after N days"). +WITH candidates AS ( + SELECT campaign_id + FROM sequences + GROUP BY campaign_id + HAVING COUNT(*) >= 2 + AND bool_and(kind = 'email') + AND bool_and(NOT COALESCE( + jsonb_typeof(conditions->'branches') = 'array' + AND jsonb_array_length(conditions->'branches') > 0, false)) +), +ordered AS ( + SELECT s.id, + LEAD(s.id) OVER (PARTITION BY s.campaign_id ORDER BY s.position ASC, s.created_at ASC) AS next_id + FROM sequences s + JOIN candidates c ON c.campaign_id = s.campaign_id +) +UPDATE sequences s +SET conditions = jsonb_build_object('branches', jsonb_build_array(jsonb_build_object( + 'branch_id', gen_random_uuid()::text, + 'target_step_id', o.next_id::text, + 'conditions', '[]'::jsonb))) +FROM ordered o +WHERE o.id = s.id AND o.next_id IS NOT NULL; diff --git a/internal/infrastructure/eventbus/bus.go b/internal/infrastructure/eventbus/bus.go index 1a273b52..359aea71 100644 --- a/internal/infrastructure/eventbus/bus.go +++ b/internal/infrastructure/eventbus/bus.go @@ -76,6 +76,15 @@ type Message struct { Topic string Key string Payload []byte + // Attempt is the 1-based delivery count of this message (NATS reports it + // from the consumer's redelivery metadata; Kafka always reports 1). A + // handler that retries by returning an error can read it to know when the + // broker is about to stop redelivering and give up cleanly instead. + Attempt int + // Redelivers is true when a handler error leaves the message for another + // delivery (NATS). Kafka commits regardless, so a handler must not count + // on a retry there and should finish what it can on this delivery. + Redelivers bool } // Subject normalises a topic name to the dot-separated form by replacing any diff --git a/internal/infrastructure/eventbus/kafka.go b/internal/infrastructure/eventbus/kafka.go index 8303f517..479dc2a2 100644 --- a/internal/infrastructure/eventbus/kafka.go +++ b/internal/infrastructure/eventbus/kafka.go @@ -140,6 +140,7 @@ func (b *KafkaBus) Subscribe(ctx context.Context, topics []string, group string, Topic: topic, Key: string(msg.Key), Payload: msg.Value, + Attempt: 1, }); err != nil { log.Error().Err(err).Str("topic", topic).Msg("eventbus kafka handler error") return err diff --git a/internal/infrastructure/eventbus/nats.go b/internal/infrastructure/eventbus/nats.go index 490d0fbc..87a8e1f3 100644 --- a/internal/infrastructure/eventbus/nats.go +++ b/internal/infrastructure/eventbus/nats.go @@ -249,10 +249,16 @@ func (b *NATSBus) Subscribe(ctx context.Context, topics []string, group string, // on the exact separator and should use Subject() if it needs to // compare against a Kafka-style topic name. topic := strings.TrimPrefix(m.Subject(), b.prefix+".") + attempt := 1 + if meta, merr := m.Metadata(); merr == nil && meta.NumDelivered > 0 { + attempt = int(meta.NumDelivered) + } if err := invokeHandler(hctx, handler, Message{ - Topic: topic, - Key: key, - Payload: m.Data(), + Topic: topic, + Key: key, + Payload: m.Data(), + Attempt: attempt, + Redelivers: true, }); err != nil { log.Error().Err(err).Str("subject", m.Subject()).Msg("eventbus nats handler error") // Nak with a short delay so transient errors don't hot-loop. diff --git a/internal/models/contact.go b/internal/models/contact.go index 6b022811..35312820 100644 --- a/internal/models/contact.go +++ b/internal/models/contact.go @@ -65,6 +65,7 @@ type ContactCampaignProgress struct { // completed — every sequence step was sent, no reply (done, nothing left to send) // replied — the contact has replied (terminal/positive) // bounced — a send hard-bounced (terminal/negative) + // failed — the mailbox could not send a step after every retry (terminal/negative) // unsubscribed — the contact is unsubscribed/suppressed (terminal) Status string `json:"status"` Sent int `json:"sent"` @@ -77,6 +78,9 @@ type ContactCampaignProgress struct { // step actually sent ("Email 2", a custom step name, or an action label). // Empty when nothing has been sent yet (status "pending"). CurrentStep string `json:"current_step,omitempty"` + // FailureReason is the worker's reason for the last failed send, set only + // when Status is "failed". + FailureReason string `json:"failure_reason,omitempty"` } // Lead status constants for ContactCampaignProgress.Status. @@ -86,6 +90,7 @@ const ( LeadStatusCompleted = "completed" LeadStatusReplied = "replied" LeadStatusBounced = "bounced" + LeadStatusFailed = "failed" LeadStatusUnsubscribed = "unsubscribed" ) @@ -93,7 +98,7 @@ const ( // Used to gate the single-campaign Leads-view `lead_status` search filter. func ValidLeadStatus(s string) bool { switch s { - case LeadStatusPending, LeadStatusActive, LeadStatusCompleted, LeadStatusReplied, LeadStatusBounced, LeadStatusUnsubscribed: + case LeadStatusPending, LeadStatusActive, LeadStatusCompleted, LeadStatusReplied, LeadStatusBounced, LeadStatusFailed, LeadStatusUnsubscribed: return true default: return false @@ -118,8 +123,8 @@ type ContactsResult struct { // CampaignLeadCounts are per-status lead totals within a single campaign, // derived the same way as ContactCampaignProgress.Status (unsubscribed > -// bounced > replied > completed > processing > queued). Drives the Leads-view -// scope chips. +// bounced > replied > failed > completed > processing > queued). Drives the +// Leads-view scope chips. type CampaignLeadCounts struct { Total int `json:"total"` Queued int `json:"queued"` // pending: a lead, no email sent yet @@ -127,6 +132,7 @@ type CampaignLeadCounts struct { Completed int `json:"completed"` // done: every step sent, no reply Replied int `json:"replied"` Bounced int `json:"bounced"` + Failed int `json:"failed"` // a step could not be sent after every retry Unsubscribed int `json:"unsubscribed"` } diff --git a/internal/models/sequence.go b/internal/models/sequence.go index 7ca68f3d..4553d04c 100644 --- a/internal/models/sequence.go +++ b/internal/models/sequence.go @@ -27,9 +27,11 @@ type Sequence struct { X float64 `json:"x"` Y float64 `json:"y"` - // Conditions is the per-step branching tree. When empty (`{}` / no - // branches), the step keeps the default linear behaviour (advance to the - // next position). When populated, the scheduler evaluates the contact's + // Conditions is the per-step routing: the connections out of this step. + // Routing follows connections only. When empty (`{}` / no branches) the + // step has no outgoing path and the contact's flow ends there; position + // orders the canvas and picks the entry step, it never advances a contact + // by itself. When populated, the scheduler evaluates the contact's // engagement against these branches at schedule time to decide which step // (or stop) comes next. Stored as a single jsonb column on `sequences`. Conditions json.RawMessage `json:"conditions,omitempty"` @@ -220,8 +222,9 @@ type SequencePosition struct { // `conditions` jsonb column. Branches are evaluated in declared order; the first // branch whose conditions ALL match wins. A winning branch routes the contact to // its TargetSequenceID (any step in the campaign), or stops them when the target -// is nil. When no branch matches (or Branches is empty) the scheduler keeps the -// default linear progression (advance to the next step by position). +// is nil. When no branch matches (or Branches is empty) the contact's flow +// ends at this step; a plain "go there next" link is a branch with no +// conditions. type BranchConditions struct { Branches []Branch `json:"branches,omitempty"` } diff --git a/internal/repository/pg_campaign.go b/internal/repository/pg_campaign.go index df677114..712f7682 100644 --- a/internal/repository/pg_campaign.go +++ b/internal/repository/pg_campaign.go @@ -87,6 +87,14 @@ type CampaignRepository interface { // CountNewLeadsStartedToday returns new_leads_started for the current UTC // day (0 when no row exists yet). CountNewLeadsStartedToday(ctx context.Context, campaignID uuid.UUID) (int, error) + // DecrementCampaignDailySend gives back a counted send the worker could not + // deliver, against the day it was counted on (sentAt), so the new-lead cap + // and the campaign daily limit do not charge for mail that never left. + DecrementCampaignDailySend(ctx context.Context, campaignID uuid.UUID, sentAt time.Time, newLead bool) error + // ReopenAfterSendFailure flips a campaign that completed while a send was + // still in flight back to active, so the failed step is retried instead of + // being finalised as done. Returns true when the status changed. + ReopenAfterSendFailure(ctx context.Context, campaignID uuid.UUID) (bool, error) // ── Campaign-scoped tracking domain (feature 5) ───────────────────── // SetCampaignTrackingDomainVerified flips the verified flag / timestamp on @@ -482,8 +490,12 @@ func (r *campaignRepository) Create(ctx context.Context, userID string, orgID *u // Initial sequences. Position is the array index; wait_after defaults // to 0 for the first step and 3 days for any follow-ups so a default - // wizard run still produces something usable. + // wizard run still produces something usable. Steps given together are a + // linear sequence, so each one is connected to the next: routing follows + // connections only (a step with no outgoing connection ends the flow), and + // a follow-up that is listed but not connected would never send. if len(data.Sequences) > 0 { + stepIDs := make([]uuid.UUID, 0, len(data.Sequences)) for i, seq := range data.Sequences { waitAfter := 0 if i > 0 { @@ -513,16 +525,22 @@ func (r *campaignRepository) Create(ctx context.Context, userID string, orgID *u body_plain, body_html, body_sync, body_code, wait_after, position ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) + RETURNING id ` seqParams := []any{ campaign.ID, orgID, seq.Name, seq.Subject, seq.BodyPlain, bodyHTML, bodySync, bodyCode, waitAfter, i + 1, } - if _, err := tx.Exec(ctx, seqInsert, seqParams...); err != nil { - db.CaptureError(err, seqInsert, seqParams, "exec") + var stepID uuid.UUID + if err := tx.QueryRow(ctx, seqInsert, seqParams...).Scan(&stepID); err != nil { + db.CaptureError(err, seqInsert, seqParams, "queryrow") return nil, errx.InternalError() } + stepIDs = append(stepIDs, stepID) + } + if err := connectLinearSequenceTx(ctx, tx, stepIDs); err != nil { + return nil, errx.InternalError() } } @@ -1798,6 +1816,37 @@ func (r *campaignRepository) AdvanceRampLevel(ctx context.Context, campaignID uu return err } +// DecrementCampaignDailySend reverses IncrementCampaignDailySend for a send +// that failed in the worker. Clamped at zero; a missing row is left alone. +func (r *campaignRepository) DecrementCampaignDailySend(ctx context.Context, campaignID uuid.UUID, sentAt time.Time, newLead bool) error { + newLeadDec := 0 + if newLead { + newLeadDec = 1 + } + _, err := r.DB.Exec(ctx, ` + UPDATE campaign_daily_sends + SET emails_sent = GREATEST(emails_sent - 1, 0), + new_leads_started = GREATEST(new_leads_started - $3, 0) + WHERE campaign_id = $1 AND send_date = $2::date + `, campaignID, sentAt.UTC(), newLeadDec) + return err +} + +// ReopenAfterSendFailure moves a completed campaign back to active. Only the +// completed state is touched: a paused campaign stays paused and picks the +// retry up when it is resumed. +func (r *campaignRepository) ReopenAfterSendFailure(ctx context.Context, campaignID uuid.UUID) (bool, error) { + tag, err := r.DB.Exec(ctx, ` + UPDATE campaigns + SET status = 'active', last_status_change_at = NOW(), updated_at = NOW() + WHERE id = $1 AND status = 'completed' + `, campaignID) + if err != nil { + return false, err + } + return tag.RowsAffected() > 0, nil +} + // IncrementCampaignDailySend bumps today's per-campaign send counters. newLead // also increments new_leads_started (a position-1 send) so the new-lead cap can // read it back. @@ -1877,3 +1926,27 @@ func (r *campaignRepository) UpdateStatusWithLock(ctx context.Context, campaignI } return err } + +// connectLinearSequenceTx links steps in order with an unconditional +// connection from each to the next, the same shape the sequence canvas writes +// when a step is dragged onto another ("just go there after the wait"). +func connectLinearSequenceTx(ctx context.Context, tx pgx.Tx, stepIDs []uuid.UUID) error { + for i := 0; i+1 < len(stepIDs); i++ { + next := stepIDs[i+1] + conditions, err := json.Marshal(models.BranchConditions{ + Branches: []models.Branch{{ + BranchID: uuid.New().String(), + TargetSequenceID: &next, + }}, + }) + if err != nil { + return err + } + const q = `UPDATE sequences SET conditions = $2 WHERE id = $1` + if _, err := tx.Exec(ctx, q, stepIDs[i], conditions); err != nil { + db.CaptureError(err, q, []any{stepIDs[i]}, "exec") + return err + } + } + return nil +} diff --git a/internal/repository/pg_campaign_progress.go b/internal/repository/pg_campaign_progress.go index 555a365a..a4e389e4 100644 --- a/internal/repository/pg_campaign_progress.go +++ b/internal/repository/pg_campaign_progress.go @@ -4,10 +4,13 @@ import ( "context" "database/sql" "encoding/json" + "errors" "time" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" + "github.com/warmbly/warmbly/internal/config" "github.com/warmbly/warmbly/internal/models" ) @@ -74,6 +77,16 @@ type CampaignSequencePair struct { type CampaignProgressRepository interface { // Record email status RecordEmailSent(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID) error + // RecordSendFailure walks back a step the worker could not send: sent_at is + // cleared so routing offers the step again, and the attempt is counted. It + // returns the attempts so far and whether the lead has now exhausted + // config.CampaignSendMaxAttempts. rolledBack is false when there was + // nothing to walk back (a duplicate result, or the send was already retried). + RecordSendFailure(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID, reason string) (attempts int, exhausted bool, rolledBack bool, err error) + // HasSentSteps reports whether the contact has any other step of the + // campaign stamped sent, which is what decides if a failed send was the + // lead's first step (a "new lead" for the daily new-lead counter). + HasSentSteps(ctx context.Context, campaignID, contactID uuid.UUID) (bool, error) RecordEmailOpened(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID, machine bool) error RecordEmailClicked(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID) error RecordEmailReplied(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID) error @@ -149,19 +162,62 @@ func NewCampaignProgressRepository(db *pgxpool.Pool) CampaignProgressRepository return &campaignProgressRepository{db: db} } -// RecordEmailSent records that an email was sent +// RecordEmailSent stamps a step as handed to the worker. A retry after a +// worker-reported failure clears the failure marker again; send_attempts is +// kept as history so the retry cap still holds. func (r *campaignProgressRepository) RecordEmailSent(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID) error { query := ` INSERT INTO campaign_contact_progress (campaign_id, contact_id, sequence_id, sent_at) VALUES ($1, $2, $3, NOW()) ON CONFLICT (campaign_id, contact_id, sequence_id) - DO UPDATE SET sent_at = NOW() + DO UPDATE SET sent_at = NOW(), failed_at = NULL, failure_reason = '' ` _, err := r.db.Exec(ctx, query, campaignID, contactID, sequenceID) return err } +// RecordSendFailure clears sent_at on a stamped step and counts the attempt. +// Only a row that is currently stamped sent is touched, so a duplicate worker +// result after the step was already walked back (or re-sent) is a no-op. +func (r *campaignProgressRepository) RecordSendFailure(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID, reason string) (int, bool, bool, error) { + if len(reason) > 500 { + reason = reason[:500] + } + query := ` + UPDATE campaign_contact_progress + SET sent_at = NULL, + send_attempts = send_attempts + 1, + failed_at = NOW(), + failure_reason = $4 + WHERE campaign_id = $1 AND contact_id = $2 AND sequence_id = $3 + AND sent_at IS NOT NULL + RETURNING send_attempts + ` + var attempts int + err := r.db.QueryRow(ctx, query, campaignID, contactID, sequenceID, reason).Scan(&attempts) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return 0, false, false, nil + } + return 0, false, false, err + } + return attempts, attempts >= config.CampaignSendMaxAttempts, true, nil +} + +// HasSentSteps reports whether any step of the campaign is stamped sent for +// the contact. +func (r *campaignProgressRepository) HasSentSteps(ctx context.Context, campaignID, contactID uuid.UUID) (bool, error) { + var has bool + err := r.db.QueryRow(ctx, ` + SELECT EXISTS ( + SELECT 1 FROM campaign_contact_progress + WHERE campaign_id = $1 AND contact_id = $2 AND sent_at IS NOT NULL + ) + `, campaignID, contactID).Scan(&has) + return has, err +} + // RecordEmailOpened records that an email was opened. machine marks automated // fetches (Apple MPP prefetch, UA-less clients): the first open stamps // opened_at with the flag, and a later HUMAN open upgrades a machine open to @@ -538,16 +594,20 @@ func (r *campaignProgressRepository) GetLatestCampaignSequenceForContact(ctx con } // FindNextRoutedPair selects the next (contact, step) to send by FOLLOWING THE -// FLOW graph instead of linear position order. For each contact, the next step -// is the route out of their last-sent step: +// FLOW graph. For each contact, the next step is the route out of their +// last-sent step: // 1. conditional branches (first match wins, evaluated against engagement), -// 2. then the explicit "else" catch-all branch (empty conditions; target nil = STOP), -// 3. then linear position+1 — but ONLY when the step defines no branches at all -// (so plain linear campaigns keep working unchanged). +// 2. then the explicit "else" catch-all branch (empty conditions; target nil = STOP). +// +// There is no implicit advance by position: a step with no outgoing connection +// ends the contact's flow. The campaign wizard connects the steps it creates +// in order, and the canvas connects steps as they are dragged. // // A contact who has never been sent starts at the entry step (position 1). A // step is sent only if the route reaches it, so branch-only steps are never sent // linearly, and a routed step that was already sent (a loop) stops the contact. +// A contact whose step failed in the worker more than CampaignSendMaxAttempts +// times is dropped, the same way a bounced or suppressed contact is. // // Conditions are evaluated SEND-RELATIVE with a three-valued result: a contact // whose next step isn't decidable yet (an engagement window still open) is not @@ -787,6 +847,12 @@ func (r *campaignProgressRepository) FindNextRoutedPair(ctx context.Context, cam SELECT 1 FROM campaign_contact_progress b WHERE b.contact_id = cl.contact_id AND b.bounced_at IS NOT NULL ) + AND NOT EXISTS ( + SELECT 1 FROM campaign_contact_progress f + WHERE f.campaign_id = $1 AND f.contact_id = cl.contact_id + AND f.sent_at IS NULL AND f.failed_at IS NOT NULL + AND f.send_attempts >= $2 + ) AND NOT EXISTS ( SELECT 1 FROM suppressed_recipients sr JOIN campaigns camp ON camp.organization_id = sr.organization_id @@ -797,7 +863,7 @@ func (r *campaignProgressRepository) FindNextRoutedPair(ctx context.Context, cam ORDER BY ` + orderPrefix + contactOrder + ` ` + dir + ` ` - rows, err := r.db.Query(ctx, query, campaignID) + rows, err := r.db.Query(ctx, query, campaignID, config.CampaignSendMaxAttempts) if err != nil { return nil, nil, err } diff --git a/internal/repository/pg_contact.go b/internal/repository/pg_contact.go index a897128a..f13a1b18 100644 --- a/internal/repository/pg_contact.go +++ b/internal/repository/pg_contact.go @@ -705,8 +705,9 @@ func (r *contactRepository) Search( // Lead status filter (single-campaign Leads view only) // ----------------------------- // The derived status has no stored column, so reproduce the same priority - // chain used on read (unsubscribed > bounced > replied > processing > - // queued) as a boolean predicate over the campaign's progress rows. Only + // chain used on read (unsubscribed > bounced > replied > failed > + // processing > queued) as a boolean predicate over the campaign's progress + // rows. Only // meaningful with exactly one campaign bound; ignored otherwise. if filters.LeadStatus != "" && singleCampaignPlaceholder != "" { if clause := leadStatusClause(filters.LeadStatus, singleCampaignPlaceholder); clause != "" { @@ -828,7 +829,14 @@ func (r *contactRepository) Search( 'clicked', COUNT(*) FILTER (WHERE p.clicked_at IS NOT NULL), 'replied', COUNT(*) FILTER (WHERE p.replied_at IS NOT NULL), 'bounced', COUNT(*) FILTER (WHERE p.bounced_at IS NOT NULL), - 'last_at', MAX(GREATEST(p.sent_at, p.opened_at, p.clicked_at, p.replied_at, p.bounced_at)), + -- Steps the mailbox could not send after every retry. + 'failed', COUNT(*) FILTER (WHERE p.sent_at IS NULL AND p.failed_at IS NOT NULL AND p.send_attempts >= %[2]d), + 'failure_reason', ( + SELECT fp.failure_reason FROM campaign_contact_progress fp + WHERE fp.campaign_id = %[1]s AND fp.contact_id = c.id AND fp.failed_at IS NOT NULL + ORDER BY fp.failed_at DESC LIMIT 1 + ), + 'last_at', MAX(GREATEST(p.sent_at, p.opened_at, p.clicked_at, p.replied_at, p.bounced_at, p.failed_at)), -- Total email steps in the sequence, to tell "still sending" (active) -- apart from "every step sent" (completed/done). 'total_steps', (SELECT COUNT(*) FROM sequences st WHERE st.campaign_id = %[1]s AND st.kind = 'email'), @@ -861,7 +869,7 @@ func (r *contactRepository) Search( ) FROM campaign_contact_progress p WHERE p.campaign_id = %[1]s AND p.contact_id = c.id - )`, singleCampaignPlaceholder) + )`, singleCampaignPlaceholder, config.CampaignSendMaxAttempts) } // Main query. @@ -969,6 +977,8 @@ func (r *contactRepository) Search( Clicked int `json:"clicked"` Replied int `json:"replied"` Bounced int `json:"bounced"` + Failed int `json:"failed"` + FailReason *string `json:"failure_reason"` TotalSteps int `json:"total_steps"` LastAt *time.Time `json:"last_at"` Step *string `json:"step"` @@ -985,6 +995,8 @@ func (r *contactRepository) Search( status = models.LeadStatusBounced case lp.Replied > 0: status = models.LeadStatusReplied + case lp.Failed > 0: + status = models.LeadStatusFailed case lp.Sent > 0 && lp.TotalSteps > 0 && lp.Sent >= lp.TotalSteps: // Every email step has been sent and the contact hasn't replied // or bounced: the sequence is exhausted, so the lead is done. @@ -996,6 +1008,10 @@ func (r *contactRepository) Search( if lp.Step != nil { currentStep = *lp.Step } + failureReason := "" + if status == models.LeadStatusFailed && lp.FailReason != nil { + failureReason = *lp.FailReason + } c.CampaignLead = &models.ContactCampaignProgress{ Status: status, Sent: lp.Sent, @@ -1005,6 +1021,7 @@ func (r *contactRepository) Search( Bounced: lp.Bounced, LastActivityAt: lp.LastAt, CurrentStep: currentStep, + FailureReason: failureReason, } } @@ -1151,7 +1168,7 @@ func (r *contactRepository) DistinctCustomFieldKeys(ctx context.Context, orgID u // leadStatusClause builds the WHERE predicate for a derived lead status inside // ONE campaign, matching pg_contact Search's read-time derivation exactly: -// unsubscribed > bounced > replied > completed > processing(active) > +// unsubscribed > bounced > replied > failed > completed > processing(active) > // queued(pending). `cp` is the already-bound placeholder for that campaign id // (e.g. "$5"). Returns "" for an unknown status (the caller then applies no lead // filter). @@ -1164,6 +1181,13 @@ func leadStatusClause(status, cp string) string { ) } sent, replied, bounced := has("sent_at"), has("replied_at"), has("bounced_at") + // failed: a step the mailbox could not send after every retry (sent_at was + // walked back and the attempt cap is spent). + failed := fmt.Sprintf( + "EXISTS (SELECT 1 FROM campaign_contact_progress p WHERE p.campaign_id = %s AND p.contact_id = c.id "+ + "AND p.sent_at IS NULL AND p.failed_at IS NOT NULL AND p.send_attempts >= %d)", + cp, config.CampaignSendMaxAttempts, + ) // allSent: every email step of the campaign has been sent to this contact. allSent := fmt.Sprintf( "((SELECT COUNT(*) FROM sequences st WHERE st.campaign_id = %[1]s AND st.kind = 'email') > 0 "+ @@ -1178,12 +1202,14 @@ func leadStatusClause(status, cp string) string { return fmt.Sprintf("(c.subscribed AND %s)", bounced) case models.LeadStatusReplied: return fmt.Sprintf("(c.subscribed AND NOT %s AND %s)", bounced, replied) + case models.LeadStatusFailed: + return fmt.Sprintf("(c.subscribed AND NOT %s AND NOT %s AND %s)", bounced, replied, failed) case models.LeadStatusCompleted: - return fmt.Sprintf("(c.subscribed AND NOT %s AND NOT %s AND %s AND %s)", bounced, replied, sent, allSent) + return fmt.Sprintf("(c.subscribed AND NOT %s AND NOT %s AND NOT %s AND %s AND %s)", bounced, replied, failed, sent, allSent) case models.LeadStatusActive: - return fmt.Sprintf("(c.subscribed AND NOT %s AND NOT %s AND %s AND NOT %s)", bounced, replied, sent, allSent) + return fmt.Sprintf("(c.subscribed AND NOT %s AND NOT %s AND NOT %s AND %s AND NOT %s)", bounced, replied, failed, sent, allSent) case models.LeadStatusPending: - return fmt.Sprintf("(c.subscribed AND NOT %s AND NOT %s AND NOT %s)", bounced, replied, sent) + return fmt.Sprintf("(c.subscribed AND NOT %s AND NOT %s AND NOT %s AND NOT %s)", bounced, replied, failed, sent) default: return "" } @@ -1192,22 +1218,24 @@ func leadStatusClause(status, cp string) string { // CampaignLeadCounts returns per-status lead totals for one campaign (the // campaign Leads view scope chips). A single aggregate over the campaign's // leads joined to their contact and a rolled-up view of their progress, so the -// buckets follow the same unsubscribed > bounced > replied > completed > -// processing > queued priority as the row-level derived status. Scoped to the -// org through the contacts join. +// buckets follow the same unsubscribed > bounced > replied > failed > +// completed > processing > queued priority as the row-level derived status. +// Scoped to the org through the contacts join. func (r *contactRepository) CampaignLeadCounts(ctx context.Context, orgID, campaignID string) (*models.CampaignLeadCounts, *errx.Error) { // A lead is "done" (completed) when every email step has been sent and it // hasn't replied or bounced; "processing" when some but not all steps sent. const done = "ts.total_steps > 0 AND COALESCE(pr.sent_steps, 0) >= ts.total_steps" + const live = "c.subscribed AND NOT COALESCE(pr.has_bounced, false) AND NOT COALESCE(pr.has_replied, false) AND NOT COALESCE(pr.has_failed, false)" query := fmt.Sprintf(` SELECT COUNT(*) AS total, COUNT(*) FILTER (WHERE NOT c.subscribed) AS unsubscribed, COUNT(*) FILTER (WHERE c.subscribed AND COALESCE(pr.has_bounced, false)) AS bounced, COUNT(*) FILTER (WHERE c.subscribed AND NOT COALESCE(pr.has_bounced, false) AND COALESCE(pr.has_replied, false)) AS replied, - COUNT(*) FILTER (WHERE c.subscribed AND NOT COALESCE(pr.has_bounced, false) AND NOT COALESCE(pr.has_replied, false) AND COALESCE(pr.has_sent, false) AND (%[1]s)) AS completed, - COUNT(*) FILTER (WHERE c.subscribed AND NOT COALESCE(pr.has_bounced, false) AND NOT COALESCE(pr.has_replied, false) AND COALESCE(pr.has_sent, false) AND NOT (%[1]s)) AS processing, - COUNT(*) FILTER (WHERE c.subscribed AND NOT COALESCE(pr.has_bounced, false) AND NOT COALESCE(pr.has_replied, false) AND NOT COALESCE(pr.has_sent, false)) AS queued + COUNT(*) FILTER (WHERE c.subscribed AND NOT COALESCE(pr.has_bounced, false) AND NOT COALESCE(pr.has_replied, false) AND COALESCE(pr.has_failed, false)) AS failed, + COUNT(*) FILTER (WHERE %[2]s AND COALESCE(pr.has_sent, false) AND (%[1]s)) AS completed, + COUNT(*) FILTER (WHERE %[2]s AND COALESCE(pr.has_sent, false) AND NOT (%[1]s)) AS processing, + COUNT(*) FILTER (WHERE %[2]s AND NOT COALESCE(pr.has_sent, false)) AS queued FROM campaign_leads cl JOIN contacts c ON c.id = cl.contact_id AND c.organization_id = $2 CROSS JOIN (SELECT COUNT(*) AS total_steps FROM sequences st WHERE st.campaign_id = $1 AND st.kind = 'email') ts @@ -1216,15 +1244,16 @@ func (r *contactRepository) CampaignLeadCounts(ctx context.Context, orgID, campa bool_or(p.sent_at IS NOT NULL) AS has_sent, bool_or(p.replied_at IS NOT NULL) AS has_replied, bool_or(p.bounced_at IS NOT NULL) AS has_bounced, + bool_or(p.sent_at IS NULL AND p.failed_at IS NOT NULL AND p.send_attempts >= $3) AS has_failed, COUNT(*) FILTER (WHERE p.sent_at IS NOT NULL) AS sent_steps FROM campaign_contact_progress p WHERE p.campaign_id = cl.campaign_id AND p.contact_id = cl.contact_id ) pr ON true WHERE cl.campaign_id = $1 - `, done) + `, done, live) out := &models.CampaignLeadCounts{} - if err := r.DB.QueryRow(ctx, query, campaignID, orgID).Scan( - &out.Total, &out.Unsubscribed, &out.Bounced, &out.Replied, &out.Completed, &out.Processing, &out.Queued, + if err := r.DB.QueryRow(ctx, query, campaignID, orgID, config.CampaignSendMaxAttempts).Scan( + &out.Total, &out.Unsubscribed, &out.Bounced, &out.Replied, &out.Failed, &out.Completed, &out.Processing, &out.Queued, ); err != nil { if err == pgx.ErrNoRows { return out, nil diff --git a/internal/repository/pg_email.go b/internal/repository/pg_email.go index 2d02ac5c..89ecb0f8 100644 --- a/internal/repository/pg_email.go +++ b/internal/repository/pg_email.go @@ -102,6 +102,9 @@ type EmailRepository interface { // worker reconciler uses it to (re)load accounts onto their assigned workers // after onboarding, worker restarts, or reassignment. ListActiveWorkerAccounts(ctx context.Context) ([]uuid.UUID, error) + // ListActiveAccountsByWorker returns the ids of the active mailboxes + // assigned to one worker, for reloading them after that worker restarts. + ListActiveAccountsByWorker(ctx context.Context, workerID uuid.UUID) ([]uuid.UUID, error) } type emailRepository struct { @@ -220,6 +223,25 @@ func (r *emailRepository) ListActiveWorkerAccounts(ctx context.Context) ([]uuid. return ids, rows.Err() } +func (r *emailRepository) ListActiveAccountsByWorker(ctx context.Context, workerID uuid.UUID) ([]uuid.UUID, error) { + const query = `SELECT id FROM email_accounts WHERE status = 'active' AND worker_id = $1` + rows, err := r.DB.Query(ctx, query, workerID) + if err != nil { + return nil, err + } + defer rows.Close() + + var ids []uuid.UUID + for rows.Next() { + var id uuid.UUID + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + return ids, rows.Err() +} + func (r *emailRepository) CountForOrganization(ctx context.Context, orgID uuid.UUID) (int, *errx.Error) { var count int query := `SELECT COUNT(*) FROM email_accounts WHERE organization_id = $1` diff --git a/internal/tasks/campaign_task.go b/internal/tasks/campaign_task.go index 193e7d0f..76b95a5f 100644 --- a/internal/tasks/campaign_task.go +++ b/internal/tasks/campaign_task.go @@ -564,16 +564,23 @@ func (s *tasksService) HandleCampaignTask(task *proto.ProcessTask) *errx.Error { } if err := s.emailSender.Send(ctx, taskID, emailMsg, *account); err != nil { - // Failed to send to worker, record failure + // The send never reached a worker (none assigned, worker offline, bus + // or storage down). Nothing is stamped sent; the task is dead-lettered + // for the retry loop and the chain is re-seeded by the reconciler. s.taskRepo.RecordTaskFailure(ctx, taskID, "Send failed", err.Error()) if s.campaignLogRepo != nil { s.campaignLogRepo.CreateLog(ctx, &repository.CampaignLogEntry{ CampaignID: campaign.ID, EventType: "email_failed", - Message: fmt.Sprintf("Failed to send to %s", contact.Email), + Message: fmt.Sprintf("Could not hand %s's email to a sending worker, will retry: %s", contact.Email, err.Error()), Metadata: map[string]interface{}{ - "contact_id": contact.ID.String(), - "error": err.Error(), + "level": "error", + "code": "WORKER_UNAVAILABLE", + "contact_id": contact.ID.String(), + "sequence_id": sequence.ID.String(), + "account_id": account.ID.String(), + "error": err.Error(), + "will_retry": true, }, }) } diff --git a/internal/tasks/email_sender.go b/internal/tasks/email_sender.go index 94404e03..64cfa80a 100644 --- a/internal/tasks/email_sender.go +++ b/internal/tasks/email_sender.go @@ -2,10 +2,12 @@ package tasks import ( "context" + "errors" "fmt" "strings" "github.com/google/uuid" + "github.com/redis/go-redis/v9" "github.com/warmbly/warmbly/internal/events" "github.com/warmbly/warmbly/internal/models" "github.com/warmbly/warmbly/internal/repository" @@ -41,9 +43,23 @@ type EmailSender interface { Send(ctx context.Context, taskID uuid.UUID, msg EmailMessage, account models.Email) error } +// WorkerLiveness reports whether a worker is still heartbeating. Satisfied by +// repository.WorkerRepository. +type WorkerLiveness interface { + IsWorkerLive(ctx context.Context, workerID uuid.UUID) (bool, error) +} + +// ErrWorkerOffline is returned by Send when the mailbox's worker has stopped +// heartbeating. The send is not published: a command queued for a worker that +// is gone is never executed and never answered, which would leave the step +// looking sent forever. The worker reconciler moves the mailbox to a live +// worker and the dead-letter retry replays the task. +var ErrWorkerOffline = errors.New("the mailbox's sending worker is offline") + type emailSender struct { emailRepo repository.EmailRepository publisher events.Publisher + liveness WorkerLiveness } // NewEmailSender creates a new email sender @@ -54,6 +70,12 @@ func NewEmailSender(emailRepo repository.EmailRepository, publisher events.Publi } } +// WireWorkerLiveness makes Send refuse to publish to a worker that is not +// heartbeating. Optional; without it Send trusts the assignment. +func (s *emailSender) WireWorkerLiveness(l WorkerLiveness) { + s.liveness = l +} + // Send publishes an email to the worker service for sending func (s *emailSender) Send(ctx context.Context, taskID uuid.UUID, msg EmailMessage, account models.Email) error { // Get worker ID for this email account @@ -61,6 +83,15 @@ func (s *emailSender) Send(ctx context.Context, taskID uuid.UUID, msg EmailMessa if workerID == nil { return fmt.Errorf("no worker assigned to email account %s", account.ID) } + if s.liveness != nil { + live, err := s.liveness.IsWorkerLive(ctx, *workerID) + if err != nil { + return fmt.Errorf("check worker %s liveness: %w", workerID, err) + } + if !live { + return fmt.Errorf("%w (worker %s, mailbox %s)", ErrWorkerOffline, workerID, account.Email) + } + } // Email content is sealed with the organization DEK; an account without an // organization cannot be encrypted for transport. @@ -115,3 +146,39 @@ func generateMessageID(fromEmail string) string { // Generate unique ID return fmt.Sprintf("<%s@%s>", uuid.New().String(), domain) } + +// HeartbeatChecker reports whether a worker's short-lived heartbeat key is +// present. Satisfied by *cache.Cache (go-redis Exists). +type HeartbeatChecker interface { + Exists(ctx context.Context, keys ...string) *redis.IntCmd +} + +// workerLiveness combines the registry's view of a worker (active, seen in the +// last ten minutes) with its three-minute heartbeat key, so a crashed worker +// stops receiving sends within minutes rather than at the end of the registry +// window. A heartbeat lookup error fails open to the registry's answer: a +// cache blip must not stop every send. +type workerLiveness struct { + repo WorkerLiveness + heartbeat HeartbeatChecker +} + +// NewWorkerLiveness builds the liveness check Send uses. heartbeat may be nil. +func NewWorkerLiveness(repo WorkerLiveness, heartbeat HeartbeatChecker) WorkerLiveness { + return &workerLiveness{repo: repo, heartbeat: heartbeat} +} + +func (l *workerLiveness) IsWorkerLive(ctx context.Context, workerID uuid.UUID) (bool, error) { + live, err := l.repo.IsWorkerLive(ctx, workerID) + if err != nil || !live { + return live, err + } + if l.heartbeat == nil { + return true, nil + } + n, herr := l.heartbeat.Exists(ctx, "worker:heartbeat:"+workerID.String()).Result() + if herr != nil { + return true, nil + } + return n > 0, nil +} diff --git a/web/src/components/app/contacts/ContactsTable.tsx b/web/src/components/app/contacts/ContactsTable.tsx index 60f17511..65383c2b 100644 --- a/web/src/components/app/contacts/ContactsTable.tsx +++ b/web/src/components/app/contacts/ContactsTable.tsx @@ -990,6 +990,7 @@ const LEAD_META: Record< completed: { label: "Done", dot: "bg-indigo-500", text: "text-indigo-700", Icon: CheckIcon }, replied: { label: "Replied", dot: "bg-emerald-500", text: "text-emerald-700", Icon: CornerUpLeftIcon }, bounced: { label: "Bounced", dot: "bg-rose-500", text: "text-rose-600", Icon: AlertTriangleIcon }, + failed: { label: "Failed", dot: "bg-rose-500", text: "text-rose-600", Icon: AlertTriangleIcon }, unsubscribed: { label: "Unsubscribed", dot: "bg-slate-300", text: "text-slate-400", Icon: BanIcon }, }; @@ -997,9 +998,16 @@ function LeadStatusPill({ lead }: { lead?: ContactCampaignProgress | null }) { const status: LeadStatus = lead?.status ?? "pending"; const meta = LEAD_META[status]; const Icon = meta.Icon; + // A failed lead carries the worker's reason; surface it on hover since the + // pill itself only has room for the word. + const title = + status === "failed" && lead?.failure_reason + ? `Could not send: ${lead.failure_reason}` + : undefined; return ( {status === "active" ? ( @@ -1030,6 +1038,7 @@ function LeadProgressStrip({ completed: 0, replied: 0, bounced: 0, + failed: 0, unsubscribed: 0, }; for (const ct of contacts) c[ct.campaign_lead?.status ?? "pending"]++; @@ -1045,6 +1054,7 @@ function LeadProgressStrip({ { key: "replied", color: "bg-emerald-500" }, { key: "pending", color: "bg-slate-300" }, { key: "bounced", color: "bg-rose-400" }, + { key: "failed", color: "bg-rose-500" }, { key: "unsubscribed", color: "bg-slate-200" }, ]; @@ -1069,6 +1079,7 @@ function LeadProgressStrip({