Merge pull request #394 from warmbly/fix/round-robin-sender-attribution

feat: attribute a rotating campaign send to the mailbox that actually sent it

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
 

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 {

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

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

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

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', '<p>Hello</p>', 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)
+	}
+}

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.