diff --git a/internal/models/warmup.go b/internal/models/warmup.go index 66f3699a..758445f8 100644 --- a/internal/models/warmup.go +++ b/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"` diff --git a/internal/repository/pg_warmup.go b/internal/repository/pg_warmup.go index a42558ff..acf22dd5 100644 --- a/internal/repository/pg_warmup.go +++ b/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 { diff --git a/internal/repository/warmup_provider_routing_live_test.go b/internal/repository/warmup_provider_routing_live_test.go index e0e90672..a0f3fbb8 100644 --- a/internal/repository/warmup_provider_routing_live_test.go +++ b/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) + } +} diff --git a/internal/tasks/warmup_partner_routing_live_test.go b/internal/tasks/warmup_partner_routing_live_test.go new file mode 100644 index 00000000..ebb1f197 --- /dev/null +++ b/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) + } +}