diff --git a/docs/content/docs/api/reference/contacts.mdx b/docs/content/docs/api/reference/contacts.mdx index e866c2cc..841c22b0 100644 --- a/docs/content/docs/api/reference/contacts.mdx +++ b/docs/content/docs/api/reference/contacts.mdx @@ -696,6 +696,10 @@ Returns a `data` array plus a `pagination` envelope. } ``` +`email_account_id`, `email_account_email` and `email_account_name` are the mailbox the email was actually sent from, which is what a campaign rotating across several mailboxes needs: the row names the mailbox that sent this email, not the campaign's pool or the one that sent the previous step. + +`opened_at` is a person's open, as it is in `engagement` and in campaign analytics. A fetch by a mail client's prefetch or a security gateway is reported as `machine_opened_at` instead, so a row is never presented as read by the recipient when only a machine touched it. At most one of the two is present. + ## List a contact's timeline `GET /contacts/:id/timeline` @@ -839,7 +843,7 @@ Auth: **Scope** `READ_CONTACTS` ยท **Org permission** `view_contacts` } ``` -`lead_status` uses the same values as the campaign Leads view: `pending`, `active`, `completed`, `replied`, `bounced`, `failed`, `unsubscribed`, or `undeliverable`. Each step carries whichever of `sent_at`, `opened_at`, `clicked_at`, `replied_at`, `bounced_at` and `failed_at` apply, plus `attempts` and `in_flight` (reserved for a worker whose result has not come back). While a branch condition is undecided, `next.step_id` is absent and `next.step_label` says the step depends on the contact's response. +`lead_status` uses the same values as the campaign Leads view: `pending`, `active`, `completed`, `replied`, `bounced`, `failed`, `unsubscribed`, or `undeliverable`. Each step carries whichever of `sent_at`, `opened_at`, `clicked_at`, `replied_at`, `bounced_at` and `failed_at` apply, plus `attempts` and `in_flight` (reserved for a worker whose result has not come back). `opened_at` is a person's open, as it is in the Leads view: a step a mail client prefetched or a security gateway scanned carries no `opened_at`. While a branch condition is undecided, `next.step_id` is absent and `next.step_label` says the step depends on the contact's response. ## List a contact's activities diff --git a/internal/models/contact.go b/internal/models/contact.go index 232ebb01..bd35030c 100644 --- a/internal/models/contact.go +++ b/internal/models/contact.go @@ -394,10 +394,14 @@ type ContactSentEmail struct { SequenceName *string `json:"step_name,omitempty"` // Engagement (from campaign_contact_progress, may be nil). - OpenedAt *time.Time `json:"opened_at,omitempty"` - ClickedAt *time.Time `json:"clicked_at,omitempty"` - RepliedAt *time.Time `json:"replied_at,omitempty"` - BouncedAt *time.Time `json:"bounced_at,omitempty"` + // OpenedAt is a person's open. An automated fetch (client prefetch, + // security gateway) lands in MachineOpenedAt instead, so the two are + // never mistaken for each other. + OpenedAt *time.Time `json:"opened_at,omitempty"` + MachineOpenedAt *time.Time `json:"machine_opened_at,omitempty"` + ClickedAt *time.Time `json:"clicked_at,omitempty"` + RepliedAt *time.Time `json:"replied_at,omitempty"` + BouncedAt *time.Time `json:"bounced_at,omitempty"` } type ContactSentEmailsResult struct { diff --git a/internal/repository/pg_contact.go b/internal/repository/pg_contact.go index 392588a8..4d446ce2 100644 --- a/internal/repository/pg_contact.go +++ b/internal/repository/pg_contact.go @@ -3117,6 +3117,11 @@ func (r *contactRepository) GetDetail(ctx context.Context, userID uuid.UUID, org // We deliberately scope by the contact's owning user via the // campaign join โ this keeps multi-tenant safety even though the // tasks table itself has no user_id column. +// +// opened_at is a person's open, as it is in campaign analytics and the +// contact's engagement summary: a fetch by a mail client's prefetch or a +// security gateway is reported separately as machine_opened_at, so this list +// never claims a recipient read mail they never opened (issue #392). func (r *contactRepository) ListSentEmails(ctx context.Context, userID, contactID uuid.UUID, limit int, beforeSentAt *time.Time, beforeTaskID *uuid.UUID) (*models.ContactSentEmailsResult, *errx.Error) { if limit <= 0 || limit > 200 { limit = 50 @@ -3137,7 +3142,9 @@ func (r *contactRepository) ListSentEmails(ctx context.Context, userID, contactI cam.id, cam.name, seq.id, seq.name, COALESCE(et.subject, seq.subject, '') AS subject, - ccp.opened_at, ccp.clicked_at, ccp.replied_at, ccp.bounced_at + CASE WHEN ccp.opened_machine THEN NULL ELSE ccp.opened_at END AS opened_at, + CASE WHEN ccp.opened_machine THEN ccp.opened_at END AS machine_opened_at, + ccp.clicked_at, ccp.replied_at, ccp.bounced_at FROM tasks t JOIN campaign_tasks ct ON ct.task_id = t.id LEFT JOIN email_accounts ea ON ea.id = t.email_account_id @@ -3171,7 +3178,7 @@ func (r *contactRepository) ListSentEmails(ctx context.Context, userID, contactI &e.CampaignID, &e.CampaignName, &e.SequenceID, &e.SequenceName, &e.Subject, - &e.OpenedAt, &e.ClickedAt, &e.RepliedAt, &e.BouncedAt, + &e.OpenedAt, &e.MachineOpenedAt, &e.ClickedAt, &e.RepliedAt, &e.BouncedAt, ); err != nil { db.CaptureError(err, "", nil, "ListSentEmails scan") return nil, errx.InternalError() @@ -3294,7 +3301,11 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, WHERE ct.campaign_id = ccp.campaign_id AND ct.contact_id = ccp.contact_id AND ct.sequence_id = ccp.sequence_id - ORDER BY t.created_at DESC + -- The task holding the step's reservation is the one that put the + -- email on the wire; a step can carry several task rows (a retry, a + -- tick that skipped as a duplicate) and only that one names the + -- mailbox the recipient saw. + ORDER BY COALESCE(t.id = ccp.dispatch_task_id, false) DESC, t.created_at DESC LIMIT 1 ) ea ON TRUE WHERE ccp.contact_id = $1 diff --git a/internal/repository/pg_contact_campaign_state.go b/internal/repository/pg_contact_campaign_state.go index 484cfb15..f24e76e1 100644 --- a/internal/repository/pg_contact_campaign_state.go +++ b/internal/repository/pg_contact_campaign_state.go @@ -137,7 +137,12 @@ func (r *contactRepository) ListCampaignStates(ctx context.Context, orgID, conta func (r *contactRepository) campaignStepsForContact(ctx context.Context, campaignID, contactID uuid.UUID) ([]models.ContactCampaignStep, string, *errx.Error) { stepQuery := ` SELECT s.id, s.name, s.subject, s.kind, s.action, s.position, - p.sent_at, p.opened_at, p.clicked_at, p.replied_at, p.bounced_at, p.failed_at, + p.sent_at, + -- A person's open, as in the Leads list this mirrors: a client + -- prefetch or a gateway scan must not read as the recipient + -- having opened the step. + CASE WHEN p.opened_machine THEN NULL ELSE p.opened_at END, + p.clicked_at, p.replied_at, p.bounced_at, p.failed_at, COALESCE(p.send_attempts, 0), p.dispatched_at, COALESCE(p.failure_reason, '') FROM sequences s LEFT JOIN campaign_contact_progress p diff --git a/internal/repository/pg_task.go b/internal/repository/pg_task.go index 5f49b27e..77fe9301 100644 --- a/internal/repository/pg_task.go +++ b/internal/repository/pg_task.go @@ -145,6 +145,10 @@ type TaskRepository interface { DirectPendingWarmupTask(ctx context.Context, accountID, targetAccountID uuid.UUID, at time.Time) (bool, error) UpdateTaskStatusWithLock(ctx context.Context, taskID uuid.UUID, status string) error UpdateTaskMessageID(ctx context.Context, taskID uuid.UUID, messageID string) error + // UpdateTaskEmailAccount repoints a task at the mailbox it is actually + // sending from. A campaign task is created before its mailbox is known, so + // the send path stamps the rotation's real pick before dispatching. + UpdateTaskEmailAccount(ctx context.Context, taskID, accountID uuid.UUID) error // Update campaign task with contact/sequence IDs (for tracking) UpdateCampaignTaskTracking(ctx context.Context, taskID, contactID, sequenceID uuid.UUID) error @@ -918,6 +922,20 @@ func (r *taskRepository) UpdateTaskMessageID(ctx context.Context, taskID uuid.UU return err } +// UpdateTaskEmailAccount records the mailbox a task is sending from. A campaign +// chain creates its successor before rotation has chosen a mailbox for it, so +// the row is seeded with the previous tick's pick and corrected here. Everything +// that attributes a send to a mailbox reads tasks.email_account_id โ the daily +// budget, the min-gap clock, rotation's last-send fallback, bounce and complaint +// rates, the contact's activity feed โ so a stale value charges one mailbox for +// another's mail (issue #392). +func (r *taskRepository) UpdateTaskEmailAccount(ctx context.Context, taskID, accountID uuid.UUID) error { + _, err := r.db.Exec(ctx, + `UPDATE tasks SET email_account_id = $1, updated_at = NOW() WHERE id = $2`, + accountID, taskID) + return err +} + // UpdateCampaignTaskTracking updates the campaign task with contact_id and sequence_id // This is called when the task is processed and we know which contact/sequence to send to // These IDs are needed for tracking pixel/click events to record progress diff --git a/internal/tasks/campaign_sender_attribution_live_test.go b/internal/tasks/campaign_sender_attribution_live_test.go new file mode 100644 index 00000000..b042a934 --- /dev/null +++ b/internal/tasks/campaign_sender_attribution_live_test.go @@ -0,0 +1,267 @@ +package tasks + +import ( + "context" + "os" + "sort" + "sync" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/warmbly/warmbly/internal/infrastructure/db" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/pkg/encrypt" + "github.com/warmbly/warmbly/internal/repository" + "github.com/warmbly/warmbly/internal/scheduler" + "github.com/warmbly/warmbly/internal/tasks/proto" +) + +// Issue #392: a campaign rotating across several mailboxes attributed every +// send to the wrong one. The chain creates its successor before rotation has +// chosen a mailbox for it, so tasks.email_account_id named the previous tick's +// pick and was never corrected. That column is what attributes a send to a +// mailbox everywhere else (today's budget, the min-gap clock, round-robin's own +// position, bounce and complaint rates, the contact's activity feed), so the +// activity said one mailbox while another one actually sent. +// +// Skipped unless WARMBLY_TEST_DB is set: +// +// WARMBLY_TEST_DB=postgres://warmbly:warmbly@localhost:15432/warmbly_dev?sslmode=disable \ +// go test ./internal/tasks/ -run Live -v + +// attributingSender records which mailbox each task was actually dispatched +// from, which is the ground truth the stamped task row has to agree with. +type attributingSender struct { + mu sync.Mutex + from map[uuid.UUID]uuid.UUID +} + +func (s *attributingSender) Send(ctx context.Context, taskID uuid.UUID, msg EmailMessage, account models.Email) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.from == nil { + s.from = map[uuid.UUID]uuid.UUID{} + } + s.from[taskID] = account.ID + return nil +} + +func (s *attributingSender) sentFrom(taskID uuid.UUID) (uuid.UUID, bool) { + s.mu.Lock() + defer s.mu.Unlock() + id, ok := s.from[taskID] + return id, ok +} + +type rotationFixture struct { + pool *pgxpool.Pool + user, org uuid.UUID + mailboxes []uuid.UUID + campaign, step uuid.UUID + leads []uuid.UUID + svc *tasksService + sender *attributingSender +} + +// newRotationFixture builds a round-robin campaign over two mailboxes with one +// lead per planned tick, so every tick sends and rotation is the only thing +// deciding which mailbox does it. +func newRotationFixture(t *testing.T, leads int) *rotationFixture { + t.Helper() + 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() }) + pool := handle.Pool + + f := &rotationFixture{ + pool: pool, user: uuid.New(), org: uuid.New(), + campaign: uuid.New(), step: uuid.New(), + mailboxes: []uuid.UUID{uuid.New(), uuid.New()}, + sender: &attributingSender{}, + } + + exec := func(sql string, args ...any) { + t.Helper() + if _, err := pool.Exec(ctx, sql, args...); err != nil { + t.Fatalf("fixture %q: %v", sql[:min(60, len(sql))], err) + } + } + exec(`INSERT INTO users (id, email, first_name, last_name) VALUES ($1, $2, 'Live', 'Rotation')`, + f.user, "rot-"+f.user.String()[:8]+"@test.local") + exec(`INSERT INTO organizations (id, name, slug, owner_user_id) VALUES ($1, 'Live Rotation', $2, $3)`, + f.org, "rot-"+f.org.String()[:8], f.user) + for _, mb := range f.mailboxes { + exec(`INSERT INTO email_accounts (id, user_id, organization_id, email, name, + signature_plain, signature_html, provider, status, campaign_limit, min_wait_time, timezone) + VALUES ($1, $2, $3, $4, 'Live', '', '', 'smtp_imap', 'active', 50, 0, 'UTC')`, + mb, f.user, f.org, "rot-"+mb.String()[:8]+"@test.local") + } + // No campaign_senders rows: the pool is "all active mailboxes", the shape a + // customer gets by default, and the one where rotation reads its position + // off each mailbox's own send count. + exec(`INSERT INTO campaigns (id, user_id, organization_id, name, description, status, + daily_limit, timezone, days, start_time, end_time, rotation_mode, updated_at, created_at) + VALUES ($1, $2, $3, 'Live Rotation', '', 'active', 50, 'UTC', 127, '00:00', '23:59', + 'round_robin', NOW(), NOW())`, f.campaign, f.user, f.org) + exec(`INSERT INTO sequences (id, campaign_id, organization_id, name, subject, + body_plain, body_html, wait_after, position, kind) + VALUES ($1, $2, $3, 'Step 1', 'Hi', 'Hello', '
Hello
', 0, 0, 'email')`, f.step, f.campaign, f.org) + for i := 0; i < leads; i++ { + id := uuid.New() + f.leads = append(f.leads, id) + exec(`INSERT INTO contacts (id, user_id, organization_id, email, first_name, last_name, company, phone, custom_fields, verification_status, created_at) + VALUES ($1, $2, $3, $4, 'Live', 'Contact', '', '', '{}', 'valid', NOW() + make_interval(secs => $5))`, + id, f.user, f.org, "rot-lead-"+id.String()[:8]+"@test.local", float64(i)) + exec(`INSERT INTO campaign_leads (campaign_id, contact_id, position) VALUES ($1, $2, $3)`, f.campaign, id, i) + } + + t.Cleanup(func() { + c := context.Background() + for _, step := range []struct { + sql string + arg any + }{ + {`DELETE FROM task_failures WHERE task_id IN (SELECT task_id FROM campaign_tasks WHERE campaign_id = $1)`, f.campaign}, + {`DELETE FROM task_execution_keys WHERE task_id IN (SELECT task_id FROM campaign_tasks WHERE campaign_id = $1)`, f.campaign}, + {`DELETE FROM tasks WHERE id IN (SELECT task_id FROM campaign_tasks WHERE campaign_id = $1)`, f.campaign}, + {`DELETE FROM campaign_tasks WHERE campaign_id = $1`, f.campaign}, + {`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 organization_id = $1`, f.org}, + {`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) + } + } + }) + + enc, err := encrypt.NewEncrypter([]byte("0123456789abcdef0123456789abcdef")) + if err != nil { + t.Fatalf("encrypter: %v", err) + } + emailRepo := repository.NewEmailRepostory(handle, enc) + campaignRepo := repository.NewCampaignRepostory(handle) + contactRepo := repository.NewContactRepostory(handle) + logRepo := repository.NewCampaignLogRepository(handle) + progressRepo := repository.NewCampaignProgressRepository(pool) + taskRepo := repository.NewTaskRepository(pool) + f.svc = &tasksService{ + tasksClient: noopTaskScheduler{}, + scheduler: scheduler.NewSchedulerService(taskRepo, repository.NewWarmupRepository(pool), progressRepo, emailRepo, campaignRepo, contactRepo, logRepo), + cipherService: noopCipher{}, + emailSender: f.sender, + taskRepo: taskRepo, + campaignProgressRepo: progressRepo, + emailRepo: emailRepo, + campaignRepo: campaignRepo, + contactRepo: contactRepo, + campaignLogRepo: logRepo, + trackedLinkRepo: repository.NewTrackedLinkRepository(pool), + } + return f +} + +// tick runs one real campaign tick on a fresh due task seeded, as the chain +// seeds it, with the mailbox the PREVIOUS send used. Returns the task id. +func (f *rotationFixture) tick(t *testing.T, seed uuid.UUID) uuid.UUID { + t.Helper() + ctx := context.Background() + if _, err := f.pool.Exec(ctx, + `UPDATE tasks SET status = 'cancelled' WHERE status = 'pending' AND id IN ( + SELECT task_id FROM campaign_tasks WHERE campaign_id = $1)`, f.campaign); err != nil { + t.Fatalf("clear pending tasks: %v", err) + } + taskID := uuid.New() + if _, err := f.pool.Exec(ctx, ` + INSERT INTO tasks (id, task_type, email_account_id, status, message_id, scheduled_at, created_at, updated_at) + VALUES ($1, 'campaign', $2, 'pending', '', NOW(), NOW(), NOW())`, taskID, seed); err != nil { + t.Fatalf("create task: %v", err) + } + if _, err := f.pool.Exec(ctx, `INSERT INTO campaign_tasks (task_id, campaign_id) VALUES ($1, $2)`, + taskID, f.campaign); err != nil { + t.Fatalf("create campaign task: %v", err) + } + if xerr := f.svc.HandleCampaignTask(&proto.ProcessTask{TaskId: taskID.String()}); xerr != nil { + t.Fatalf("campaign tick: %v", xerr) + } + return taskID +} + +func (f *rotationFixture) taskAccount(t *testing.T, taskID uuid.UUID) uuid.UUID { + t.Helper() + var id uuid.UUID + if err := f.pool.QueryRow(context.Background(), + `SELECT email_account_id FROM tasks WHERE id = $1`, taskID).Scan(&id); err != nil { + t.Fatalf("read task mailbox: %v", err) + } + return id +} + +// TestLiveRotatingSendIsAttributedToTheMailboxThatSentIt is issue #392: the +// task row must name the mailbox the email actually left from, whatever the +// chain guessed when it created the task. +func TestLiveRotatingSendIsAttributedToTheMailboxThatSentIt(t *testing.T) { + f := newRotationFixture(t, 4) + + // Seed every tick with the WRONG mailbox on purpose: the first one in the + // pool, which round-robin will stop choosing as soon as it has sent. This is + // what the chain does naturally, and what used to be left uncorrected. + seed := f.mailboxes[0] + for i := 0; i < 4; i++ { + taskID := f.tick(t, seed) + want, ok := f.sender.sentFrom(taskID) + if !ok { + t.Fatalf("tick %d dispatched nothing", i+1) + } + if got := f.taskAccount(t, taskID); got != want { + t.Fatalf("tick %d: the task is recorded against mailbox %s but the email was sent from %s", + i+1, got, want) + } + } +} + +// TestLiveRoundRobinSpreadsAcrossMailboxes is the consequence: round-robin +// reads its position from each mailbox's own send count, so mis-attributed +// sends made it rotate off the wrong tally and pile onto one mailbox. +func TestLiveRoundRobinSpreadsAcrossMailboxes(t *testing.T) { + f := newRotationFixture(t, 4) + + seed := f.mailboxes[0] + for i := 0; i < 4; i++ { + f.tick(t, seed) + } + + counts := map[uuid.UUID]int{} + for _, mb := range f.mailboxes { + var n int + if err := f.pool.QueryRow(context.Background(), ` + SELECT COUNT(*) FROM tasks t + JOIN campaign_tasks ct ON ct.task_id = t.id + WHERE ct.campaign_id = $1 AND t.email_account_id = $2 AND t.status = 'completed'`, + f.campaign, mb).Scan(&n); err != nil { + t.Fatalf("count sends: %v", err) + } + counts[mb] = n + } + got := []int{counts[f.mailboxes[0]], counts[f.mailboxes[1]]} + sort.Ints(got) + if got[0] != 2 || got[1] != 2 { + t.Fatalf("four round-robin sends across two mailboxes landed %v, want 2 each", got) + } +} diff --git a/internal/tasks/campaign_task.go b/internal/tasks/campaign_task.go index f47d2e85..3ac430d9 100644 --- a/internal/tasks/campaign_task.go +++ b/internal/tasks/campaign_task.go @@ -452,6 +452,28 @@ func (s *tasksService) HandleCampaignTask(task *proto.ProcessTask) *errx.Error { return xerr } + // STEP 9.25: Stamp the mailbox this send actually goes out from. The chain + // seeds a task with the previous tick's pick, and rotation chooses again + // here, so tasks.email_account_id (today's budget, the min-gap clock, + // rotation's own position, bounce and complaint rates, the contact's + // activity feed) has to be corrected before the send (issue #392). Failing + // it fails the task like the reservation: a send charged to another mailbox + // would let this one past its daily cap. + if account.ID != taskRecord.EmailAccountID { + if err := s.taskRepo.UpdateTaskEmailAccount(ctx, taskID, account.ID); err != nil { + errs.CaptureException(err) + s.taskRepo.RecordTaskFailure(ctx, taskID, "Could not record the sending mailbox", err.Error()) + s.recordSchedulerFailure(ctx, campaign.ID, "sender_attribution_failed", + fmt.Sprintf("Could not record which mailbox is sending to %s; retrying", contact.Email), err) + if uerr := s.taskRepo.UpdateTaskStatus(ctx, taskID, "pending"); uerr != nil { + errs.CaptureException(uerr) + } + executionStatus = "failed" + return errx.InternalError() + } + taskRecord.EmailAccountID = account.ID + } + // STEP 9.5: Resolve {{form_link:...}} markers to per-recipient form URLs // BEFORE templating, so the substituted literal survives the naive // fallback and gets wrapped by click tracking in STEP 11 like any link.