Merge remote-tracking branch 'origin/main' into fix/issue-241

internal/models/warmup.go

@@ -108,9 +108,7 @@ type WarmupPoolHealthSummary struct {
 	ByState           map[string]int `json:"by_state"`
 	AvgSpamScore      float64        `json:"avg_spam_score"`
 	AvgSpamPlacement  float64        `json:"avg_spam_placement_rate"`
-	// SpamPlacementByProvider breaks recent spam-placement counts down by the
-	// recipient provider so the admin can see where warmup mail is being
-	// filtered (e.g. mostly at Outlook vs Gmail) rather than one flat rate.
+	// Keyed by who runs the recipient's mail, the vocabulary routing reads.
 	SpamPlacementByProvider map[string]int `json:"spam_placement_by_provider"`
 	BlockedCount            int            `json:"blocked_count"`
 	AtRiskCount             int            `json:"at_risk_count"`

internal/repository/pg_warmup.go

@@ -862,12 +862,11 @@ func (r *warmupRepository) PoolSpamPlacementRate(ctx context.Context, since time
 	return float64(placements) / float64(sent) * 100, nil
 }
 
-// PoolSpamPlacementsByProvider returns spam-placement counts grouped by the
-// recipient provider over the window, so the admin overview can show where
-// warmup mail is being filtered (e.g. mostly at Outlook vs Gmail).
+// PoolSpamPlacementsByProvider counts spam placements in the window keyed by who
+// runs the recipient's mail, not the stored connect method.
 func (r *warmupRepository) PoolSpamPlacementsByProvider(ctx context.Context, since time.Time) (map[string]int, error) {
 	query := `
-		SELECT COALESCE(NULLIF(recipient_provider, ''), 'unknown'), COUNT(*)
+		SELECT recipient_domain, COUNT(*)
 		FROM warmup_spam_reports
 		WHERE report_type = 'spam_placement' AND created_at >= $1
 		GROUP BY 1
@@ -880,12 +879,17 @@ func (r *warmupRepository) PoolSpamPlacementsByProvider(ctx context.Context, sin
 
 	out := make(map[string]int)
 	for rows.Next() {
-		var provider string
+		var domain string
 		var n int
-		if err := rows.Scan(&provider, &n); err != nil {
+		if err := rows.Scan(&domain, &n); err != nil {
 			return nil, err
 		}
-		out[provider] = n
+		// A domainless row belongs to no provider, so it stays out of custom.
+		key := "unknown"
+		if domain != "" {
+			key = string(models.ClassifyProvider(domain))
+		}
+		out[key] += n
 	}
 	return out, rows.Err()
 }
@@ -938,10 +942,13 @@ func (r *warmupRepository) SenderPlacementByProvider(ctx context.Context, sender
 		return nil, err
 	}
 
+	// A blank domain is unattributable and the send side never produces one, so
+	// counting it would demote every custom-domain partner for nobody's failure.
 	placementRows, err := r.db.Query(ctx, `
 		SELECT recipient_domain, COUNT(*)
 		FROM warmup_spam_reports
 		WHERE reported_account_id = $1 AND report_type = 'spam_placement' AND created_at >= $2
+		  AND recipient_domain <> ''
 		GROUP BY 1
 	`, senderAccountID, since)
 	if err != nil {

internal/repository/warmup_provider_routing_live_test.go

@@ -27,6 +27,7 @@ type providerRoutingFixture struct {
 	sender    uuid.UUID
 	atGoogle  uuid.UUID
 	atMSGraph uuid.UUID
+	atCustom  uuid.UUID
 }
 
 func newProviderRoutingFixture(t *testing.T) *providerRoutingFixture {
@@ -41,7 +42,7 @@ func newProviderRoutingFixture(t *testing.T) *providerRoutingFixture {
 
 	f := &providerRoutingFixture{
 		pool: pool, user: uuid.New(), org: uuid.New(),
-		sender: uuid.New(), atGoogle: uuid.New(), atMSGraph: uuid.New(),
+		sender: uuid.New(), atGoogle: uuid.New(), atMSGraph: uuid.New(), atCustom: uuid.New(),
 	}
 	exec := func(sql string, args ...any) {
 		t.Helper()
@@ -65,15 +66,16 @@ func newProviderRoutingFixture(t *testing.T) *providerRoutingFixture {
 		{f.sender, "smtp_imap", "test.local"},
 		{f.atGoogle, "gmail", "gmail.com"},
 		{f.atMSGraph, "smtp_imap", "outlook.com"},
+		{f.atCustom, "smtp_imap", "acme.test"},
 	} {
 		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, 'Route', '', '', $5, 'active', 50, 600, 'UTC')`,
 			m.id, f.user, f.org, "route-"+m.id.String()[:8]+"@"+m.domain, m.provider)
 	}
-	// Only the two recipients join the pool; the sender does not, so the
-	// participants map must contain exactly them.
-	for _, id := range []uuid.UUID{f.atGoogle, f.atMSGraph} {
+	// Only the recipients join the pool; the sender does not, so it must never
+	// appear in the participants map.
+	for _, id := range []uuid.UUID{f.atGoogle, f.atMSGraph, f.atCustom} {
 		exec(`INSERT INTO warmup_pool_participants (pool_id, email_account_id, participant_role, health_state)
 		      VALUES ($1, $2, 'sender_receiver', 'healthy')`, premiumPoolID, id)
 	}
@@ -121,13 +123,20 @@ func (f *providerRoutingFixture) send(t *testing.T, recipient uuid.UUID, n int,
 }
 
 func (f *providerRoutingFixture) placement(t *testing.T, domain string, n int) {
+	t.Helper()
+	f.placementFrom(t, "smtp_imap", domain, n)
+}
+
+// placementFrom writes the row the way the consumer does: recipient_provider is
+// the connect method the mailbox uses, recipient_domain is who its mail is at.
+func (f *providerRoutingFixture) placementFrom(t *testing.T, connectMethod, domain string, n int) {
 	t.Helper()
 	for i := 0; i < n; i++ {
 		if _, err := f.pool.Exec(context.Background(),
 			`INSERT INTO warmup_spam_reports (id, reporter_account_id, reported_account_id, message_id,
-			     report_type, recipient_domain, created_at)
-			 VALUES (gen_random_uuid(), $1, $1, $2, 'spam_placement', $3, NOW())`,
-			f.sender, "m-"+uuid.New().String(), domain); err != nil {
+			     report_type, recipient_provider, recipient_domain, created_at)
+			 VALUES (gen_random_uuid(), $1, $1, $2, 'spam_placement', $3, $4, NOW())`,
+			f.sender, "m-"+uuid.New().String(), connectMethod, domain); err != nil {
 			t.Fatalf("insert placement: %v", err)
 		}
 	}
@@ -222,3 +231,70 @@ func TestLiveProviderRoutingIgnoresFailedSends(t *testing.T) {
 		t.Errorf("rate = %v, want 0.5; counting the failures would have read 0.05", m.Rate())
 	}
 }
+
+// An unattributed placement (the recipient account could not be resolved when
+// it was recorded) has no domain, so it belongs to no provider. Charging it to
+// ProviderCustom would demote every custom-domain partner for a failure that
+// was never theirs, and the send side can never produce a blank domain to put
+// underneath it.
+func TestLiveProviderRoutingIgnoresUnattributedPlacement(t *testing.T) {
+	handle, _ := liveContactDB(t)
+	f := newProviderRoutingFixture(t)
+	repo := NewWarmupRepository(handle.Pool)
+
+	f.send(t, f.atCustom, 10, "completed")
+	f.placement(t, "", 5)
+
+	got, err := repo.SenderPlacementByProvider(context.Background(), f.sender, time.Now().Add(-24*time.Hour))
+	if err != nil {
+		t.Fatalf("SenderPlacementByProvider: %v", err)
+	}
+	c := got["custom"]
+	if c.Sends != 10 {
+		t.Errorf("sends = %d, want the 10 that went to the custom domain", c.Sends)
+	}
+	if c.Placements != 0 || c.Rate() != 0 {
+		t.Errorf("custom = %+v (rate %v), want the domainless placements ignored", c, c.Rate())
+	}
+}
+
+// The admin overview has to name the same providers routing does. The stored
+// recipient_provider column is the connect method, which files a custom-domain
+// Microsoft 365 mailbox under smtp_imap: the bucket an operator most needs
+// split, and one routing never uses.
+func TestLivePoolPlacementsUseTheRoutingVocabulary(t *testing.T) {
+	handle, _ := liveContactDB(t)
+	f := newProviderRoutingFixture(t)
+	repo := NewWarmupRepository(handle.Pool)
+	ctx := context.Background()
+	since := time.Now().Add(-24 * time.Hour)
+
+	// The rollup is pool-wide, so measure what these rows added to it.
+	before, err := repo.PoolSpamPlacementsByProvider(ctx, since)
+	if err != nil {
+		t.Fatalf("PoolSpamPlacementsByProvider: %v", err)
+	}
+
+	f.placementFrom(t, "smtp_imap", "outlook.com", 3)
+	f.placementFrom(t, "gmail", "", 2)
+
+	after, err := repo.PoolSpamPlacementsByProvider(ctx, since)
+	if err != nil {
+		t.Fatalf("PoolSpamPlacementsByProvider: %v", err)
+	}
+	delta := func(k string) int { return after[k] - before[k] }
+
+	if got := delta("microsoft"); got != 3 {
+		t.Errorf("microsoft delta = %d, want 3: mail run by Microsoft over plain IMAP", got)
+	}
+	if got := delta("smtp_imap"); got != 0 {
+		t.Errorf("smtp_imap delta = %d, want 0: the connect method is not a provider", got)
+	}
+	// Domainless rows stay their own bucket rather than inflating custom.
+	if got := delta("unknown"); got != 2 {
+		t.Errorf("unknown delta = %d, want 2", got)
+	}
+	if got := delta("custom"); got != 0 {
+		t.Errorf("custom delta = %d, want 0", got)
+	}
+}

internal/tasks/warmup_partner_routing_live_test.go

@@ -0,0 +1,220 @@
+package tasks
+
+import (
+	"context"
+	"os"
+	"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"
+)
+
+// Issue #143 end to end: the per-provider placement signal has to reach the
+// partner the selector actually returns, not just the query that computes it.
+// The repository tests prove the numbers; this proves selectWarmupPartner
+// wires them into the weight. Skipped unless WARMBLY_TEST_DB is set:
+//
+//	WARMBLY_TEST_DB=postgres://warmbly:warmbly@localhost:15432/warmbly_dev?sslmode=disable \
+//	  go test ./internal/tasks/ -run LiveWarmupPartner -v
+
+// freePoolID is the seeded free pool. It is used here rather than the premium
+// one because the selector reads EVERY participant of the pool, and the free
+// pool is the one no fixture or seed puts mailboxes in.
+const freePoolID = "77777777-aaaa-0000-0000-000000000001"
+
+type partnerRoutingFixture struct {
+	pool     *pgxpool.Pool
+	svc      *tasksService
+	sender   models.Email
+	user     uuid.UUID
+	org      uuid.UUID
+	atGoogle uuid.UUID
+	atMS     uuid.UUID
+}
+
+func newPartnerRoutingFixture(t *testing.T) *partnerRoutingFixture {
+	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() })
+
+	var pools int
+	if err := handle.Pool.QueryRow(ctx, `SELECT count(*) FROM warmup_pools WHERE id = $1`, freePoolID).Scan(&pools); err != nil || pools == 0 {
+		t.Skip("free warmup pool not seeded in this database")
+	}
+	// A pick is weighted across the WHOLE pool, so a stray participant would
+	// dilute the measurement into a meaningless pass.
+	var occupied int
+	if err := handle.Pool.QueryRow(ctx, `SELECT count(*) FROM warmup_pool_participants WHERE pool_id = $1`, freePoolID).Scan(&occupied); err != nil {
+		t.Fatalf("count free pool: %v", err)
+	}
+	if occupied != 0 {
+		t.Skip("free pool already has participants; cannot isolate the measurement")
+	}
+
+	f := &partnerRoutingFixture{
+		pool: handle.Pool, user: uuid.New(), org: uuid.New(),
+		atGoogle: uuid.New(), atMS: uuid.New(),
+	}
+	senderID := uuid.New()
+	exec := func(sql string, args ...any) {
+		t.Helper()
+		if _, err := handle.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, 'Pick', 'Test')`,
+		f.user, "pick-"+f.user.String()[:8]+"@test.local")
+	exec(`INSERT INTO organizations (id, name, slug, owner_user_id) VALUES ($1, 'Pick Test', $2, $3)`,
+		f.org, "pick-"+f.org.String()[:8], f.user)
+	for _, m := range []struct {
+		id     uuid.UUID
+		domain string
+	}{
+		{senderID, "test.local"},
+		{f.atGoogle, "gmail.com"},
+		{f.atMS, "outlook.com"},
+	} {
+		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, 'Pick', '', '', 'smtp_imap', 'active', 50, 600, 'UTC')`,
+			m.id, f.user, f.org, "pick-"+m.id.String()[:8]+"@"+m.domain)
+	}
+	for _, id := range []uuid.UUID{f.atGoogle, f.atMS} {
+		exec(`INSERT INTO warmup_pool_participants (pool_id, email_account_id, participant_role, health_state)
+		      VALUES ($1, $2, 'sender_receiver', 'healthy')`, freePoolID, id)
+	}
+
+	t.Cleanup(func() {
+		c := context.Background()
+		for _, step := range []struct {
+			sql string
+			arg any
+		}{
+			{`DELETE FROM warmup_spam_reports WHERE reported_account_id = $1`, senderID},
+			{`DELETE FROM warmup_tokens WHERE sender_account_id = $1`, senderID},
+			{`DELETE FROM tasks WHERE email_account_id = $1`, senderID},
+			{`DELETE FROM warmup_pool_participants WHERE email_account_id IN (SELECT id FROM email_accounts WHERE organization_id = $1)`, f.org},
+			{`DELETE FROM email_accounts WHERE organization_id = $1`, f.org},
+			{`DELETE FROM organizations WHERE id = $1`, f.org},
+			{`DELETE FROM users WHERE id = $1`, f.user},
+		} {
+			if _, err := handle.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)
+	}
+	f.svc = &tasksService{
+		warmupRepo: repository.NewWarmupRepository(handle.Pool),
+		emailRepo:  repository.NewEmailRepostory(handle, enc),
+	}
+	// The stored tier picks the pool outright, and the risk read ahead of it
+	// fails open, so the selector runs with no feature gate or org-risk repo.
+	f.sender = models.Email{ID: senderID, OrganizationID: &f.org, WarmupPoolType: "free"}
+	return f
+}
+
+// history writes n completed warmup sends to one partner, backdated two days:
+// inside the seven-day placement window, but outside the same-day exclusion
+// that would drop both candidates before weighting ever runs.
+func (f *partnerRoutingFixture) history(t *testing.T, recipient uuid.UUID, n int) {
+	t.Helper()
+	for i := 0; i < n; i++ {
+		taskID := uuid.New()
+		if _, err := f.pool.Exec(context.Background(),
+			`INSERT INTO tasks (id, task_type, email_account_id, status, message_id)
+			 VALUES ($1, 'warmup', $2, 'completed', '')`, taskID, f.sender.ID); err != nil {
+			t.Fatalf("insert task: %v", err)
+		}
+		if _, err := f.pool.Exec(context.Background(),
+			`INSERT INTO warmup_tokens (token, task_id, sender_account_id, recipient_account_id, created_at)
+			 VALUES (gen_random_uuid(), $1, $2, $3, NOW() - INTERVAL '2 days')`,
+			taskID, f.sender.ID, recipient); err != nil {
+			t.Fatalf("insert token: %v", err)
+		}
+	}
+}
+
+func (f *partnerRoutingFixture) junked(t *testing.T, domain string, n int) {
+	t.Helper()
+	for i := 0; i < n; i++ {
+		if _, err := f.pool.Exec(context.Background(),
+			`INSERT INTO warmup_spam_reports (id, reporter_account_id, reported_account_id, message_id,
+			     report_type, recipient_domain, created_at)
+			 VALUES (gen_random_uuid(), $1, $1, $2, 'spam_placement', $3, NOW() - INTERVAL '1 day')`,
+			f.sender.ID, "m-"+uuid.New().String(), domain); err != nil {
+			t.Fatalf("insert placement: %v", err)
+		}
+	}
+}
+
+// picks runs the real selector n times and reports how often each partner won.
+func (f *partnerRoutingFixture) picks(t *testing.T, n int) (google, microsoft int) {
+	t.Helper()
+	ctx := context.Background()
+	for i := 0; i < n; i++ {
+		partner, err := f.svc.selectWarmupPartner(ctx, f.sender)
+		if err != nil {
+			t.Fatalf("selectWarmupPartner: %v", err)
+		}
+		switch partner.ID {
+		case f.atGoogle:
+			google++
+		case f.atMS:
+			microsoft++
+		default:
+			t.Fatalf("selector returned a mailbox outside the fixture: %s", partner.ID)
+		}
+	}
+	return google, microsoft
+}
+
+// The whole point of #143: a sender landing in junk only at Microsoft stops
+// being handed Microsoft partners, without an aggregate health band tripping.
+func TestLiveWarmupPartnerRoutesAwayFromTheProviderItLandsInJunkAt(t *testing.T) {
+	f := newPartnerRoutingFixture(t)
+	const rounds = 200
+
+	// Equal history on both sides, so the domain-diversity weight cannot be
+	// what moves the split.
+	f.history(t, f.atGoogle, 10)
+	f.history(t, f.atMS, 10)
+
+	baseGoogle, baseMS := f.picks(t, rounds)
+	if baseGoogle < rounds*35/100 || baseGoogle > rounds*65/100 {
+		t.Fatalf("baseline split is not even: google %d, microsoft %d of %d", baseGoogle, baseMS, rounds)
+	}
+
+	// 6 of the 10 Microsoft sends were filtered into junk. Nothing about the
+	// Google side changed.
+	f.junked(t, "outlook.com", 6)
+
+	google, microsoft := f.picks(t, rounds)
+	// weight ratio is 1 : 1/(1+4*0.6), so google should take ~77%.
+	if google <= rounds*60/100 {
+		t.Errorf("placement signal did not reach the selector: google %d, microsoft %d of %d (baseline was %d/%d)",
+			google, microsoft, rounds, baseGoogle, baseMS)
+	}
+	// Downweighted, never excluded: a sender that stops mailing a provider
+	// entirely can never discover it recovered there.
+	if microsoft == 0 {
+		t.Errorf("microsoft was excluded outright over %d picks; the penalty must only downweight", rounds)
+	}
+}