diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index eeb3b052d..6c7099d32 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -200,6 +200,19 @@ jobs: run: | go test -race -coverprofile=coverage.out -covermode=atomic ./... + - name: Linked-mailbox batching regressions (PostgreSQL) + env: + PRIMARY_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable + WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable + run: | + set -o pipefail + go run ./cmd/migrate + go test -race -v -p 1 ./internal/repository ./internal/app/analytics -run '^TestLive(WarmupPartnerCountsMatchIndividualCandidates|WarmupSummariesMatchIndividualReads|LinkedMailboxStatusBatchPreservesDetails)$' -count=1 | tee linked-mailbox-batch.log + for test in TestLiveWarmupPartnerCountsMatchIndividualCandidates TestLiveWarmupSummariesMatchIndividualReads TestLiveLinkedMailboxStatusBatchPreservesDetails; do + grep -q -- "--- PASS: $test " linked-mailbox-batch.log + done + if grep -q -- '--- SKIP:' linked-mailbox-batch.log; then exit 1; fi + - name: Campaign send admission regressions (PostgreSQL) env: PRIMARY_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable diff --git a/docs/content/docs/guides/warmbly-cloud.mdx b/docs/content/docs/guides/warmbly-cloud.mdx index 6ffdbba6a..e96b0da60 100644 --- a/docs/content/docs/guides/warmbly-cloud.mdx +++ b/docs/content/docs/guides/warmbly-cloud.mdx @@ -159,6 +159,8 @@ Pause and resume per mailbox from the same row. Warmup settings (start volume, d Removing one Cloud-managed mailbox, through its Cloud warmup controls or **Disconnect mailbox**, stops both warmup sending and receiving on Cloud before releasing its link. Cloud keeps the mailbox, sign-in and history; the instance removes its mirror. If the stop fails, removal fails and keeps the link for retry. If releasing the link fails after the stop, warmup remains stopped while you retry. This also applies to existing managed enrollments and older self-hosted clients once Cloud is updated, without disconnecting or re-enrolling other mailboxes. No blanket change is made to disconnected or idle instances, whose Cloud mailboxes may still be intentionally in use. On an older Cloud, removal may only release the mirror; stop participation directly on Cloud until it is updated. +Deleting a mailbox directly on Cloud also removes its instance association. Connecting the same address again creates a different Cloud record; it does not restore the old link. If an existing SMTP/IMAP mailbox is warming on Cloud but missing from its instance's linked list, contact support with the local mailbox ID and connection ID. Do not delete or re-enroll it to repair the association, and do not send passwords or connection tokens. Support can use the verified IDs to restore only the association, preserving the current Cloud mailbox, credentials, history and participation. + Disconnecting an entire instance is not an individual mailbox stop. Cloud-managed mailboxes retain their existing warmup settings and can continue independently on Cloud. SMTP/IMAP enrollments still delete their Cloud credential and history on removal. Pausing warmup only stops sending; use **Stop all diagnostic participation** to stop receiving too. Sends already accepted by a provider cannot be recalled. ### When the count stays at zero diff --git a/internal/app/analytics/linked_status_live_test.go b/internal/app/analytics/linked_status_live_test.go new file mode 100644 index 000000000..a5420c995 --- /dev/null +++ b/internal/app/analytics/linked_status_live_test.go @@ -0,0 +1,107 @@ +package analytics + +import ( + "context" + "reflect" + "sync/atomic" + "testing" + "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/infrastructure/db" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/pkg/encrypt" + "github.com/warmbly/warmbly/internal/repository" +) + +type statusQueryCounter struct{ queries atomic.Int64 } + +func (c *statusQueryCounter) TraceQueryStart(ctx context.Context, _ *pgx.Conn, _ pgx.TraceQueryStartData) context.Context { + c.queries.Add(1) + return ctx +} + +func (*statusQueryCounter) TraceQueryEnd(context.Context, *pgx.Conn, pgx.TraceQueryEndData) {} + +func TestLiveLinkedMailboxStatusBatchPreservesDetails(t *testing.T) { + f := newRampFixture(t, 12, 3, 2, 30) + exec := func(sql string, args ...any) { + t.Helper() + if _, err := f.pool.Exec(t.Context(), sql, args...); err != nil { + t.Fatal(err) + } + } + exec(`UPDATE organizations SET risk_state='trusted' WHERE id=$1`, f.org) + exec(`UPDATE email_accounts SET test_mode=NULL WHERE id IN ($1,$2)`, f.mailbox, f.gmail) + ids := []uuid.UUID{f.mailbox} + for i := 0; i < 14; i++ { + id := uuid.New() + provider := "gmail" + if i < 4 { + provider = "smtp_imap" + } + exec(`INSERT INTO email_accounts (id, user_id, organization_id, email, name, signature_plain, + signature_html, provider, status, campaign_limit, min_wait_time, timezone, + warmup, warmup_base, warmup_increase, warmup_max, test_mode) + VALUES ($1,$2,$3,$4,'Linked','','',$5,'active',50,600,'UTC', + NOW()-interval '12 days',3,2,30,NULL)`, + id, f.user, f.org, "linked-"+id.String()+"@example.test", provider) + ids = append(ids, id) + } + for _, id := range append(append([]uuid.UUID{}, ids...), f.gmail) { + exec(`INSERT INTO warmup_pool_participants (pool_id, email_account_id, participant_role, health_state) + VALUES ($1,$2,'sender_receiver','healthy')`, models.WarmupPoolPremiumID, id) + } + // A restrictive row and a paused mailbox must retain the same detail in the list. + exec(`UPDATE warmup_pool_participants SET health_state='quarantined', blocked_until=NOW()+interval '2 days' WHERE email_account_id=$1`, ids[1]) + exec(`UPDATE email_accounts SET warmup_paused_at=NOW() WHERE id=$1`, ids[2]) + counter := &statusQueryCounter{} + poolConfig := f.pool.Config() + poolConfig.ConnConfig.Tracer = counter + pool, err := pgxpool.NewWithConfig(t.Context(), poolConfig) + if err != nil { + t.Fatal(err) + } + t.Cleanup(pool.Close) + enc, err := encrypt.NewEncrypter([]byte("0123456789abcdef0123456789abcdef")) + if err != nil { + t.Fatal(err) + } + handle := &db.DB{Pool: pool} + svc := NewService(repository.NewAnalyticsRepository(handle), repository.NewEmailRepostory(handle, enc), + repository.NewCampaignRepostory(handle), repository.NewEmailAccountErrorRepository(handle), repository.NewWarmupRepository(pool)) + start := time.Now() + individual := make(map[uuid.UUID]models.EmailAccountStatus) + for _, id := range ids { + status, xerr := svc.GetAccountStatusDetail(t.Context(), f.org, id) + if xerr != nil { + t.Fatal(xerr) + } + individual[id] = *status + } + oldQueries, oldDuration := counter.queries.Swap(0), time.Since(start) + start = time.Now() + page, xerr := svc.GetAccountStatusDetailsPage(t.Context(), f.org, ids, "", config.AccountStatusLimitMax) + if xerr != nil || len(page.Data) != 15 { + t.Fatalf("page=%+v error=%v", page, xerr) + } + newQueries, newDuration := counter.queries.Swap(0), time.Since(start) + for _, status := range page.Data { + want := individual[status.ID] + if !reflect.DeepEqual(status.WarmupStatus, want.WarmupStatus) || !reflect.DeepEqual(status.WarmupHealth, want.WarmupHealth) || !reflect.DeepEqual(status.Errors, want.Errors) || !reflect.DeepEqual(status.DailyUsage, want.DailyUsage) { + t.Fatalf("%s: batched=%+v individual=%+v", status.ID, status, want) + } + } + if newQueries >= oldQueries { + t.Fatalf("batch queries=%d individual=%d", newQueries, oldQueries) + } + t.Logf("15 mixed-provider mailbox details: individual %d queries in %s; batch %d queries in %s", oldQueries, oldDuration, newQueries, newDuration) + foreign := newRampFixture(t, 12, 3, 2, 30) + page, xerr = svc.GetAccountStatusDetailsPage(t.Context(), f.org, []uuid.UUID{ids[0], foreign.mailbox}, "", config.AccountStatusLimitMax) + if xerr != nil || len(page.Data) != 1 || page.Data[0].ID != ids[0] { + t.Fatalf("foreign organization leaked: %+v %v", page, xerr) + } +} diff --git a/internal/app/analytics/service.go b/internal/app/analytics/service.go index 22e1bb2b8..03dfcd4df 100644 --- a/internal/app/analytics/service.go +++ b/internal/app/analytics/service.go @@ -12,6 +12,7 @@ import ( "github.com/warmbly/warmbly/internal/models" "github.com/warmbly/warmbly/internal/repository" "github.com/warmbly/warmbly/internal/utils/paging" + "golang.org/x/sync/errgroup" ) type AnalyticsService interface { @@ -40,6 +41,7 @@ type AnalyticsService interface { // batched across the page, so its cost is bounded by the page, not by the // total inventory. GetAccountStatusesPage(ctx context.Context, orgID uuid.UUID, emailIDs []uuid.UUID, cursor string, limit int32) (*models.EmailAccountStatusesResult, *errx.Error) + GetAccountStatusDetailsPage(ctx context.Context, orgID uuid.UUID, emailIDs []uuid.UUID, cursor string, limit int32) (*models.EmailAccountStatusesResult, *errx.Error) // Usage overview GetUsageOverview(ctx context.Context, orgID, userID uuid.UUID, period string) (*models.UsageOverview, *errx.Error) @@ -499,6 +501,10 @@ func (s *analyticsService) GetAllAccountStatuses(ctx context.Context, orgID uuid // is batched across the page, and the shared placement rate is read once, so // the page's cost is bounded by the page size, not by the total inventory. func (s *analyticsService) GetAccountStatusesPage(ctx context.Context, orgID uuid.UUID, emailIDs []uuid.UUID, cursor string, limit int32) (*models.EmailAccountStatusesResult, *errx.Error) { + return s.accountStatusesPage(ctx, orgID, emailIDs, cursor, limit, 1) +} + +func (s *analyticsService) accountStatusesPage(ctx context.Context, orgID uuid.UUID, emailIDs []uuid.UUID, cursor string, limit int32, concurrency int) (*models.EmailAccountStatusesResult, *errx.Error) { // The opaque cursor decodes to the keyset position; an invalid token is a // 400, never a silently ignored full-inventory scan. cursorID, xerr := paging.DecodeCursor(cursor) @@ -564,22 +570,63 @@ func (s *analyticsService) GetAccountStatusesPage(ctx context.Context, orgID uui } day := now.Format("2006-01-02") + result.Data = make([]models.EmailAccountStatus, len(emails)) + g, assemblyCtx := errgroup.WithContext(ctx) + g.SetLimit(concurrency) for i := range emails { - email := emails[i] - u := usage[email.ID] - if u == nil { - // Deleted between the page read and the usage read: report an empty - // day rather than drop the row or fail the whole page. - u = &models.AccountDailyUsage{Date: day, CampaignLimit: email.CampaignLimit, WarmupLimit: email.WarmupMax} - } - warmupHealth := s.warmupHealthFromParticipant(ctx, email.ID, healthByID[email.ID]) - status := s.assembleAccountStatus(ctx, &email, u, errsByID[email.ID], warmupHealth, campByID[email.ID] > 0, rates, false) - result.Data = append(result.Data, *status) + g.Go(func() error { + if err := assemblyCtx.Err(); err != nil { + return err + } + email := emails[i] + u := usage[email.ID] + if u == nil { + // Deletion between the page and usage reads leaves an empty day, not a dropped row. + u = &models.AccountDailyUsage{Date: day, CampaignLimit: email.CampaignLimit, WarmupLimit: email.WarmupMax} + } + warmupHealth := s.warmupHealthFromParticipant(assemblyCtx, email.ID, healthByID[email.ID]) + status := s.assembleAccountStatus(assemblyCtx, &email, u, errsByID[email.ID], warmupHealth, campByID[email.ID] > 0, rates, false) + result.Data[i] = *status + return assemblyCtx.Err() + }) + } + if err := g.Wait(); err != nil { + return nil, errx.InternalError() } return result, nil } +// GetAccountStatusDetailsPage retains the detail target without scanning partner cohorts per mailbox. +func (s *analyticsService) GetAccountStatusDetailsPage(ctx context.Context, orgID uuid.UUID, emailIDs []uuid.UUID, cursor string, limit int32) (*models.EmailAccountStatusesResult, *errx.Error) { + // Residual mailbox-specific reads use four slots, never a goroutine per mailbox. + result, xerr := s.accountStatusesPage(ctx, orgID, emailIDs, cursor, limit, 4) + if xerr != nil || s.warmupRepo == nil { + return result, xerr + } + byPool := make(map[string][]uuid.UUID) + for _, status := range result.Data { + if w, h := status.WarmupStatus, status.WarmupHealth; w != nil && !w.Paused && w.TargetVolume > 0 && h != nil && h.PoolType != "" { + byPool[h.PoolType] = append(byPool[h.PoolType], status.ID) + } + } + for pool, ids := range byPool { + counts, err := s.warmupRepo.WarmupPartnerCounts(ctx, pool, ids) + if err != nil { + return nil, errx.InternalError() + } + for i := range result.Data { + status := &result.Data[i] + if reachable, ok := counts[status.ID]; ok && status.WarmupStatus != nil && reachable < status.WarmupStatus.TargetVolume { + w := status.WarmupStatus + w.PartnerLimit = &models.WarmupPartnerLimit{Reachable: reachable, RampTarget: w.TargetVolume} + w.TargetVolume = max(reachable, status.DailyUsage.WarmupSent) + } + } + } + return result, nil +} + func (s *analyticsService) GetUsageOverview(ctx context.Context, orgID, userID uuid.UUID, period string) (*models.UsageOverview, *errx.Error) { now := time.Now().UTC() var from time.Time diff --git a/internal/app/poollink/list_mailboxes_test.go b/internal/app/poollink/list_mailboxes_test.go new file mode 100644 index 000000000..fa9848da6 --- /dev/null +++ b/internal/app/poollink/list_mailboxes_test.go @@ -0,0 +1,194 @@ +package poollink + +import ( + "context" + "errors" + "fmt" + "reflect" + "testing" + "time" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/app/analytics" + "github.com/warmbly/warmbly/internal/config" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" +) + +type listMailboxRepo struct { + repository.PoolLinkRepository + inst *models.PoolLinkInstance + rows []models.PoolLinkMailbox + err error +} + +func (r *listMailboxRepo) ListMailboxes(_ context.Context, instance uuid.UUID) ([]models.PoolLinkMailbox, error) { + if instance != r.inst.ID { + panic("wrong instance") + } + return r.rows, r.err +} + +func (r *listMailboxRepo) ListStanding(_ context.Context, instance uuid.UUID) ([]models.PoolLinkMailboxStanding, error) { + if instance != r.inst.ID { + panic("wrong standing instance") + } + return []models.PoolLinkMailboxStanding{{RemoteID: r.rows[0].RemoteID, Health: &models.WarmupHealthInfo{State: "quarantined"}}}, nil +} + +type listMailboxEmails struct { + repository.EmailRepository + org uuid.UUID + accounts map[uuid.UUID]models.Email + calls int + err *errx.Error +} + +func (r *listMailboxEmails) Search(_ context.Context, org, search string, cursor, tag *string, limit int32, ids []uuid.UUID) (*models.EmailsResult, *errx.Error) { + if org != r.org.String() || search != "" || cursor != nil || tag != nil || len(ids) > int(limit) || len(ids) == 0 { + panic("list must be org-scoped and bounded to its nonempty association batch") + } + r.calls++ + if r.err != nil { + return nil, r.err + } + out := &models.EmailsResult{} + for i := len(ids) - 1; i >= 0; i-- { + if a, ok := r.accounts[ids[i]]; ok { + out.Data = append(out.Data, a) + } + } + return out, nil +} + +type listMailboxAnalytics struct { + analytics.AnalyticsService + org uuid.UUID + statuses map[uuid.UUID]models.EmailAccountStatus + calls int + err *errx.Error +} + +func (a *listMailboxAnalytics) GetAccountStatusDetailsPage(_ context.Context, org uuid.UUID, ids []uuid.UUID, cursor string, limit int32) (*models.EmailAccountStatusesResult, *errx.Error) { + if org != a.org || len(ids) > int(limit) || len(ids) == 0 || cursor != "" { + panic("status page must be scoped to this instance's association batch") + } + a.calls++ + if a.err != nil { + return nil, a.err + } + out := &models.EmailAccountStatusesResult{} + for _, id := range ids { + if status, ok := a.statuses[id]; ok { + out.Data = append(out.Data, status) + } + } + return out, nil +} + +type listMailboxWarmup struct { + repository.WarmupRepository + calls int + err error +} + +func (r *listMailboxWarmup) GetWarmupSummaries(_ context.Context, ids []uuid.UUID, from, to time.Time) (map[uuid.UUID]repository.WarmupSummary, error) { + if len(ids) == 0 || len(ids) > config.AccountStatusLimitMax || to.Sub(from) != 7*24*time.Hour { + panic("unexpected summary batch/window") + } + r.calls++ + out := make(map[uuid.UUID]repository.WarmupSummary) + for _, id := range ids { + out[id] = repository.WarmupSummary{Sent: 42, Replied: 6, SpamPlaced: 2} + } + return out, r.err +} + +func newMailboxList(n int) (*service, *listMailboxRepo, *listMailboxEmails, *listMailboxAnalytics, *listMailboxWarmup) { + inst := &models.PoolLinkInstance{ID: uuid.New(), OrganizationID: uuid.New()} + r := &listMailboxRepo{inst: inst} + e := &listMailboxEmails{org: inst.OrganizationID, accounts: make(map[uuid.UUID]models.Email)} + a := &listMailboxAnalytics{org: inst.OrganizationID, statuses: make(map[uuid.UUID]models.EmailAccountStatus)} + w := &listMailboxWarmup{} + for i := range n { + id := uuid.New() + provider := "gmail" + if i%3 == 0 { + provider = "smtp_imap" + } + enrolled := time.Now().Add(-time.Duration(i) * time.Hour) + r.rows = append(r.rows, models.PoolLinkMailbox{InstanceID: inst.ID, RemoteID: uuid.New(), EmailAccountID: id, EnrolledAt: enrolled, Managed: provider == "gmail"}) + mode := models.TestParticipationDiagnostic + e.accounts[id] = models.Email{ID: id, OrganizationID: &inst.OrganizationID, Email: fmt.Sprintf("box-%d@example.test", i), Name: "Mailbox", Provider: provider, Status: "active", TestMode: &mode, TestSendEnabled: true, TestReceiveEnabled: true, WarmupBase: 3, WarmupMax: 30, WarmupIncrease: 2, Timezone: "Europe/Budapest", AuthState: "pass"} + a.statuses[id] = models.EmailAccountStatus{ID: id, DailyUsage: models.AccountDailyUsage{WarmupSent: i}, WarmupHealth: &models.WarmupHealthInfo{State: "quarantined", Reason: "placement"}, WarmupStatus: &models.WarmupStatusInfo{TargetVolume: 5, PartnerLimit: &models.WarmupPartnerLimit{Reachable: 5, RampTarget: 20}}, Errors: []models.AccountError{{Title: "sync"}}} + } + return &service{repo: r, emails: e, analytics: a, warmup: w}, r, e, a, w +} + +func TestListMailboxesBatchesAllProvidersAndPreservesState(t *testing.T) { + for _, n := range []int{0, 15, config.AccountStatusLimitMax + 1} { + t.Run(fmt.Sprint(n), func(t *testing.T) { + s, r, e, a, w := newMailboxList(n) + before := make(map[uuid.UUID]models.Email) + for id, acc := range e.accounts { + before[id] = acc + } + states, xerr := s.ListMailboxes(t.Context(), r.inst) + if xerr != nil || len(states) != n { + t.Fatalf("len=%d error=%v", len(states), xerr) + } + batches := (n + config.AccountStatusLimitMax - 1) / config.AccountStatusLimitMax + if e.calls != batches || a.calls != batches || w.calls != batches { + t.Fatalf("calls=%d/%d/%d want %d batch reads", e.calls, a.calls, w.calls, batches) + } + for i, st := range states { + m := r.rows[i] + acc := e.accounts[m.EmailAccountID] + if st.RemoteID != m.RemoteID || st.EmailAccountID != acc.ID || st.Email != acc.Email || st.Provider != acc.Provider || st.Managed != m.Managed || !st.EnrolledAt.Equal(m.EnrolledAt) || st.AuthState != acc.AuthState || st.Settings.Max != acc.WarmupMax || st.Settings.Timezone != acc.Timezone { + t.Fatalf("mailbox metadata/order changed: %+v", st) + } + if st.Participation.Mode != *acc.TestMode || !st.Participation.Send || !st.Participation.Receive || st.SentToday != i || st.Sent7d != 42 || st.Replied7d != 6 || st.SpamPlaced7d != 2 || st.Health.State != "quarantined" || st.Warmup.PartnerLimit.Reachable != 5 || len(st.Errors) != 1 { + t.Fatalf("mailbox status changed: %+v", st) + } + } + if !reflect.DeepEqual(before, e.accounts) { + t.Fatal("listing changed mailbox data") + } + }) + } +} + +func TestListMailboxesFailsInsteadOfReturningIncompleteAssociations(t *testing.T) { + for _, fail := range []string{"associations", "accounts", "missing account", "analytics", "missing status", "summary"} { + t.Run(fail, func(t *testing.T) { + s, r, e, a, w := newMailboxList(15) + switch fail { + case "associations": + r.err = errors.New("unavailable") + case "accounts": + e.err = errx.InternalError() + case "missing account": + delete(e.accounts, r.rows[0].EmailAccountID) + case "analytics": + a.err = errx.InternalError() + case "missing status": + delete(a.statuses, r.rows[0].EmailAccountID) + case "summary": + w.err = errors.New("unavailable") + } + if states, xerr := s.ListMailboxes(t.Context(), r.inst); xerr == nil || states != nil { + t.Fatalf("incomplete success: %+v %v", states, xerr) + } + }) + } +} + +func TestListStandingDoesNotDependOnMailboxDetails(t *testing.T) { + s, r, e, a, w := newMailboxList(15) + e.err, a.err, w.err = errx.InternalError(), errx.InternalError(), errors.New("unavailable") + states, xerr := s.ListStanding(t.Context(), r.inst) + if xerr != nil || len(states) != 1 || states[0].Health.State != "quarantined" || e.calls != 0 || a.calls != 0 || w.calls != 0 { + t.Fatalf("standing depended on full telemetry: %+v %v", states, xerr) + } +} diff --git a/internal/app/poollink/service.go b/internal/app/poollink/service.go index 14e52382c..79541768f 100644 --- a/internal/app/poollink/service.go +++ b/internal/app/poollink/service.go @@ -567,12 +567,55 @@ func (s *service) ListMailboxes(ctx context.Context, inst *models.PoolLinkInstan return nil, errx.InternalError() } out := make([]models.PoolLinkMailboxState, 0, len(rows)) - for _, m := range rows { - state, xerr := s.state(ctx, inst, &m) + for start := 0; start < len(rows); start += config.AccountStatusLimitMax { + batch := rows[start:min(start+config.AccountStatusLimitMax, len(rows))] + ids := make([]uuid.UUID, len(batch)) + for i, m := range batch { + ids[i] = m.EmailAccountID + } + accounts, xerr := s.emails.Search(ctx, inst.OrganizationID.String(), "", nil, nil, config.AccountStatusLimitMax, ids) if xerr != nil { - continue + return nil, xerr + } + byID := make(map[uuid.UUID]models.Email, len(accounts.Data)) + for _, acc := range accounts.Data { + byID[acc.ID] = acc + } + statuses := make(map[uuid.UUID]models.EmailAccountStatus) + if s.analytics != nil { + page, xerr := s.analytics.GetAccountStatusDetailsPage(ctx, inst.OrganizationID, ids, "", config.AccountStatusLimitMax) + if xerr != nil { + return nil, xerr + } + for _, status := range page.Data { + statuses[status.ID] = status + } + } + summaries := make(map[uuid.UUID]repository.WarmupSummary) + if s.warmup != nil { + now := time.Now() + summaries, err = s.warmup.GetWarmupSummaries(ctx, ids, now.Add(-7*24*time.Hour), now) + if err != nil { + return nil, errx.InternalError() + } + } + for _, m := range batch { + acc, ok := byID[m.EmailAccountID] + if !ok { + return nil, errx.InternalError() + } + st := mailboxState(&acc, &m) + if s.analytics != nil { + status, ok := statuses[acc.ID] + if !ok { + return nil, errx.InternalError() + } + applyMailboxStatus(st, &status) + } + summary := summaries[acc.ID] + st.Sent7d, st.Replied7d, st.SpamPlaced7d = summary.Sent, summary.Replied, summary.SpamPlaced + out = append(out, *st) } - out = append(out, *state) } return out, nil } @@ -601,33 +644,14 @@ func (s *service) state(ctx context.Context, inst *models.PoolLinkInstance, m *m if xerr != nil { return nil, xerr } - st := &models.PoolLinkMailboxState{ - Participation: &models.DiagnosticParticipation{Mode: models.TestParticipationLegacy, Send: acc.TestSendingAllowed(), Receive: acc.TestReceivingAllowed(), SharedDailyLimit: acc.SharedDailyLimit, RollingRecipientLimit: acc.RollingRecipientLimit}, - RemoteID: m.RemoteID, - EmailAccountID: acc.ID, - Email: acc.Email, - Name: acc.Name, - Provider: acc.Provider, - Status: acc.Status, - EnrolledAt: m.EnrolledAt, - Managed: m.Managed, - AuthState: acc.AuthState, - Settings: models.PoolLinkWarmupSettings{ - Base: acc.WarmupBase, Max: acc.WarmupMax, Increase: acc.WarmupIncrease, ReplyRate: acc.WarmupReplyRate, - StartTime: acc.WarmupStartTime, EndTime: acc.WarmupEndTime, Days: acc.WarmupDays, Timezone: acc.ClockTimezone(), - }, - } - if acc.TestMode != nil { - st.Participation.Mode = *acc.TestMode + if acc.OrganizationID == nil || *acc.OrganizationID != inst.OrganizationID { + return nil, ErrMailboxNotFound } + st := mailboxState(acc, m) if s.analytics != nil { - // Detail carries the partner cap, so a target no partner can meet is never shown as one. status, xerr := s.analytics.GetAccountStatusDetail(ctx, inst.OrganizationID, acc.ID) if xerr == nil && status != nil { - st.Warmup = status.WarmupStatus - st.Health = status.WarmupHealth - st.Errors = status.Errors - st.SentToday = status.DailyUsage.WarmupSent + applyMailboxStatus(st, status) } else if xerr != nil { log.Warn().Str("account_id", acc.ID.String()).Str("error", xerr.Message).Msg("pool link: account status unavailable") } @@ -649,6 +673,36 @@ func (s *service) state(ctx context.Context, inst *models.PoolLinkInstance, m *m return st, nil } +func mailboxState(acc *models.Email, m *models.PoolLinkMailbox) *models.PoolLinkMailboxState { + st := &models.PoolLinkMailboxState{ + Participation: &models.DiagnosticParticipation{Mode: models.TestParticipationLegacy, Send: acc.TestSendingAllowed(), Receive: acc.TestReceivingAllowed(), SharedDailyLimit: acc.SharedDailyLimit, RollingRecipientLimit: acc.RollingRecipientLimit}, + RemoteID: m.RemoteID, + EmailAccountID: acc.ID, + Email: acc.Email, + Name: acc.Name, + Provider: acc.Provider, + Status: acc.Status, + EnrolledAt: m.EnrolledAt, + Managed: m.Managed, + AuthState: acc.AuthState, + Settings: models.PoolLinkWarmupSettings{ + Base: acc.WarmupBase, Max: acc.WarmupMax, Increase: acc.WarmupIncrease, ReplyRate: acc.WarmupReplyRate, + StartTime: acc.WarmupStartTime, EndTime: acc.WarmupEndTime, Days: acc.WarmupDays, Timezone: acc.ClockTimezone(), + }, + } + if acc.TestMode != nil { + st.Participation.Mode = *acc.TestMode + } + return st +} + +func applyMailboxStatus(st *models.PoolLinkMailboxState, status *models.EmailAccountStatus) { + st.Warmup = status.WarmupStatus + st.Health = status.WarmupHealth + st.Errors = status.Errors + st.SentToday = status.DailyUsage.WarmupSent +} + func (s *service) PatchMailbox(ctx context.Context, inst *models.PoolLinkInstance, remoteID uuid.UUID, patch models.PoolLinkMailboxPatch) (*models.PoolLinkMailboxState, *errx.Error) { if patch.Participation != nil && !patch.Participation.Valid() { return nil, errx.ErrInvalid diff --git a/internal/repository/pg_warmup.go b/internal/repository/pg_warmup.go index 7809e2379..9265af8df 100644 --- a/internal/repository/pg_warmup.go +++ b/internal/repository/pg_warmup.go @@ -200,6 +200,7 @@ type WarmupRepository interface { LastWarmupSendFailure(ctx context.Context, accountID uuid.UUID, since time.Time) (*models.WarmupSendFailure, error) ListMailboxLoadingIncidents(ctx context.Context, now time.Time) ([]MailboxLoadingIncident, error) GetWarmupStatistics(ctx context.Context, accountID uuid.UUID, from, to time.Time) ([]WarmupStatistic, error) + GetWarmupSummaries(ctx context.Context, accountIDs []uuid.UUID, from, to time.Time) (map[uuid.UUID]WarmupSummary, error) GetOrCreateDailyStats(ctx context.Context, accountID uuid.UUID, date time.Time, targetVolume int) (*WarmupStatistic, error) // Pool-wide placement analytics (admin overview) @@ -245,6 +246,7 @@ type WarmupRepository interface { // tier, plus proven free mailboxes when a premium tier is thin. The // scheduler caps volume on the same set the selector draws from. WarmupPartnerCandidates(ctx context.Context, poolType string, senderID uuid.UUID) ([]models.WarmupPartnerCandidate, error) + WarmupPartnerCounts(ctx context.Context, poolType string, senderIDs []uuid.UUID) (map[uuid.UUID]int, error) GetRecentPartnerDomainCounts(ctx context.Context, accountID uuid.UUID, since time.Time) (map[string]int, error) // GetPartnerDiversity counts confirmed partners reached in the window. GetPartnerDiversity(ctx context.Context, accountID uuid.UUID, since time.Time) (WarmupPartnerDiversity, error) diff --git a/internal/repository/pg_warmup_partner_counts.go b/internal/repository/pg_warmup_partner_counts.go new file mode 100644 index 000000000..9f1d8e155 --- /dev/null +++ b/internal/repository/pg_warmup_partner_counts.go @@ -0,0 +1,139 @@ +package repository + +import ( + "context" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/config" + "github.com/warmbly/warmbly/internal/models" +) + +// WarmupPartnerCounts shares cohort scans while retaining each sender's eligibility and exclusions. +func (r *warmupRepository) WarmupPartnerCounts(ctx context.Context, poolType string, senderIDs []uuid.UUID) (map[uuid.UUID]int, error) { + out := make(map[uuid.UUID]int, len(senderIDs)) + if len(senderIDs) == 0 { + return out, nil + } + for _, id := range senderIDs { + out[id] = 0 + } + orgs, err := r.eligiblePartnerSenders(ctx, poolType, senderIDs) + if err != nil || len(orgs) == 0 { + return out, err + } + suffix := candidateSuffixFor(poolType) + own, err := r.queryPartnerCandidates(ctx, ` + SELECT wpp.email_account_id AS id, ea.email, ea.organization_id, ea.provider::text AS provider, ea.mail_host + FROM warmup_pool_participants wpp + JOIN warmup_pools wp ON wpp.pool_id = wp.id + JOIN email_accounts ea ON ea.id = wpp.email_account_id + WHERE wp.pool_type = $1 AND `+partnerEligibleSQL, + suffix, poolType, models.WarmupPartnerOwnTier, poolType) + if err != nil { + return nil, err + } + ownIDs := make(map[uuid.UUID]bool, len(own)) + for _, c := range own { + ownIDs[c.ID] = true + } + for id := range orgs { + out[id] = len(own) + if ownIDs[id] { + out[id]-- + } + } + if borrowFrom, ok := models.WarmupPoolBorrowsFrom(poolType); ok { + borrowed, err := r.queryPartnerCandidates(ctx, ` + SELECT wpp.email_account_id AS id, ea.email, ea.organization_id, ea.provider::text AS provider, ea.mail_host + FROM warmup_pool_participants wpp + JOIN warmup_pools wp ON wpp.pool_id = wp.id + JOIN email_accounts ea ON ea.id = wpp.email_account_id + JOIN organizations o ON o.id = ea.organization_id + WHERE wp.pool_type = $1 AND `+partnerEligibleSQL+` AND `+partnerProvenSQL, + suffix, borrowFrom, models.WarmupPartnerBorrowed, borrowFrom, config.WarmupPoolFallbackMinAgeDays) + if err != nil { + return nil, err + } + byOrg := make(map[uuid.UUID]int) + for _, c := range borrowed { + if c.OrganizationID != nil { + byOrg[*c.OrganizationID]++ + } + } + for id, org := range orgs { + out[id] += len(borrowed) - byOrg[org] + } + } + if returnTo, ok := models.WarmupPoolReturnsTo(poolType); ok { + returns, err := r.queryPartnerCandidates(ctx, ` + SELECT wpp.email_account_id AS id, ea.email, ea.organization_id, ea.provider::text AS provider, ea.mail_host + FROM warmup_pool_participants wpp + JOIN warmup_pools wp ON wpp.pool_id = wp.id + JOIN email_accounts ea ON ea.id = wpp.email_account_id + WHERE wp.pool_type = $1 AND `+partnerEligibleSQL+` + AND wpp.email_account_id IN ( + SELECT wr.sender_account_id FROM warmup_received wr + WHERE wr.email_account_id = ANY($2::uuid[]) + AND wr.created_at >= NOW() - make_interval(days => $3))`, + suffix, returnTo, models.WarmupPartnerReturn, returnTo, senderIDs, config.WarmupPoolReturnVisitDays) + if err != nil { + return nil, err + } + returnIDs := make(map[uuid.UUID]bool, len(returns)) + for _, c := range returns { + returnIDs[c.ID] = true + } + rows, err := resultDB(ctx, r.db).Query(ctx, ` + SELECT DISTINCT ea.id, wr.sender_account_id + FROM warmup_received wr + JOIN email_accounts ea ON ea.id = wr.email_account_id + JOIN warmup_pool_participants wpp ON wpp.email_account_id = ea.id + JOIN warmup_pools wp ON wp.id = wpp.pool_id + JOIN organizations o ON o.id = ea.organization_id + WHERE ea.id = ANY($3::uuid[]) AND wp.pool_type = $1 AND ea.status = 'active' + AND `+partnerProvenSQL+` + AND wr.created_at >= NOW() - make_interval(days => $4)`, + poolType, config.WarmupPoolFallbackMinAgeDays, senderIDs, config.WarmupPoolReturnVisitDays) + if err != nil { + return nil, err + } + defer rows.Close() + for rows.Next() { + var sender, partner uuid.UUID + if err := rows.Scan(&sender, &partner); err != nil { + return nil, err + } + if _, eligible := orgs[sender]; eligible && returnIDs[partner] { + out[sender]++ + } + } + if err := rows.Err(); err != nil { + return nil, err + } + } + return out, nil +} + +func (r *warmupRepository) eligiblePartnerSenders(ctx context.Context, poolType string, senderIDs []uuid.UUID) (map[uuid.UUID]uuid.UUID, error) { + rows, err := resultDB(ctx, r.db).Query(ctx, ` + SELECT ea.id, ea.organization_id + FROM warmup_pool_participants wpp + JOIN warmup_pools wp ON wp.id = wpp.pool_id + JOIN email_accounts ea ON ea.id = wpp.email_account_id + WHERE ea.id = ANY($1::uuid[]) AND wp.pool_type = $2 + AND wpp.participant_role = 'sender_receiver' + AND `+testSenderSQL+` AND `+poolEligibleSQL, senderIDs, poolType) + if err != nil { + return nil, err + } + defer rows.Close() + out := make(map[uuid.UUID]uuid.UUID) + for rows.Next() { + var id, org uuid.UUID + if err := rows.Scan(&id, &org); err != nil { + return nil, err + } + out[id] = org + } + return out, rows.Err() +} diff --git a/internal/repository/pg_warmup_summary.go b/internal/repository/pg_warmup_summary.go new file mode 100644 index 000000000..955faf514 --- /dev/null +++ b/internal/repository/pg_warmup_summary.go @@ -0,0 +1,50 @@ +package repository + +import ( + "context" + "time" + + "github.com/google/uuid" +) + +type WarmupSummary struct { + Sent, Replied, SpamPlaced int +} + +func (r *warmupRepository) GetWarmupSummaries(ctx context.Context, accountIDs []uuid.UUID, from, to time.Time) (map[uuid.UUID]WarmupSummary, error) { + out := make(map[uuid.UUID]WarmupSummary, len(accountIDs)) + if len(accountIDs) == 0 { + return out, nil + } + rows, err := resultDB(ctx, r.db).Query(ctx, ` + WITH stats AS ( + SELECT email_account_id, SUM(emails_sent) AS sent, + SUM(emails_replied) FILTER (WHERE date <= DATE($3)) AS replied + FROM warmup_statistics + WHERE email_account_id = ANY($1::uuid[]) AND date >= DATE($2) + GROUP BY email_account_id + ), spam AS ( + SELECT reported_account_id, COUNT(*) AS placed + FROM warmup_spam_reports + WHERE reported_account_id = ANY($1::uuid[]) + AND created_at >= $2 AND report_type = 'spam_placement' + GROUP BY reported_account_id + ) + SELECT a.id, COALESCE(s.sent, 0), COALESCE(s.replied, 0), COALESCE(p.placed, 0) + FROM unnest($1::uuid[]) a(id) + LEFT JOIN stats s ON s.email_account_id = a.id + LEFT JOIN spam p ON p.reported_account_id = a.id`, accountIDs, from, to) + if err != nil { + return nil, err + } + defer rows.Close() + for rows.Next() { + var id uuid.UUID + var summary WarmupSummary + if err := rows.Scan(&id, &summary.Sent, &summary.Replied, &summary.SpamPlaced); err != nil { + return nil, err + } + out[id] = summary + } + return out, rows.Err() +} diff --git a/internal/repository/warmup_batch_live_test.go b/internal/repository/warmup_batch_live_test.go new file mode 100644 index 000000000..43fe3f601 --- /dev/null +++ b/internal/repository/warmup_batch_live_test.go @@ -0,0 +1,124 @@ +package repository + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/config" +) + +func TestLiveWarmupPartnerCountsMatchIndividualCandidates(t *testing.T) { + f := newPartnerOrgFixture(t) + r := &warmupRepository{db: f.pool} + ids := []uuid.UUID{f.sender, f.sibling, f.outside, uuid.New()} + assertParity := func(pool string) { + t.Helper() + counts, err := r.WarmupPartnerCounts(t.Context(), pool, ids) + if err != nil { + t.Fatal(err) + } + for _, id := range ids { + candidates, err := r.WarmupPartnerCandidates(t.Context(), pool, id) + if err != nil || counts[id] != len(candidates) { + t.Fatalf("%s/%s: count=%d individual=%d error=%v", pool, id, counts[id], len(candidates), err) + } + } + } + assertParity("premium") + f.moveToFree(t, f.sibling, config.WarmupPoolFallbackMinAgeDays+1) + f.moveToFree(t, f.outside, config.WarmupPoolFallbackMinAgeDays+1) + assertParity("premium") + assertParity("free") + f.received(t, f.outside, f.sender, time.Now().Add(-time.Hour)) + f.received(t, f.outside, f.sender, time.Now().Add(-2*time.Hour)) + f.received(t, f.sibling, f.sender, time.Now().Add(-time.Duration(config.WarmupPoolReturnVisitDays+1)*24*time.Hour)) + assertParity("free") + for _, change := range []string{ + `UPDATE warmup_pool_participants SET health_state='watch' WHERE email_account_id=$1`, + `UPDATE warmup_pool_participants SET health_state='healthy', joined_at=NOW() WHERE email_account_id=$1`, + `UPDATE warmup_pool_participants SET joined_at=NOW()-interval '30 days', health_state='quarantined', blocked_until=NULL WHERE email_account_id=$1`, + `UPDATE warmup_pool_participants SET blocked_until=NOW()-interval '1 hour' WHERE email_account_id=$1`, + `UPDATE warmup_pool_participants SET participant_role='recipient_only' WHERE email_account_id=$1`, + `UPDATE warmup_pool_participants SET participant_role='sender_receiver', health_state='healthy', blocked_until=NULL WHERE email_account_id=$1`, + `UPDATE email_accounts SET test_mode='diagnostic', test_send_enabled=false WHERE id=$1`, + `UPDATE email_accounts SET test_mode=NULL WHERE id=$1`, + } { + f.exec(change, f.outside) + assertParity("free") + assertParity("premium") + } + f.exec(`UPDATE organizations SET risk_state='restricted' WHERE id=$1`, f.other) + assertParity("premium") + assertParity("free") + cancelled, cancel := context.WithCancel(t.Context()) + cancel() + if _, err := r.WarmupPartnerCounts(cancelled, "premium", ids); err == nil { + t.Fatal("cancelled query reported counts") + } +} + +func TestLiveWarmupSummariesMatchIndividualReads(t *testing.T) { + f := newPartnerOrgFixture(t) + r := &warmupRepository{db: f.pool} + now := time.Now().UTC() + since := now.Add(-7 * 24 * time.Hour) + for _, row := range []struct { + date time.Time + sent int + }{ + {since.Add(-24 * time.Hour), 100}, + {since, 3}, + {now, 7}, + {now.Add(24 * time.Hour), 2}, + } { + f.exec(`INSERT INTO warmup_statistics (email_account_id, date, emails_sent, emails_replied, target_volume) + VALUES ($1, DATE($2), $3, 2, 10)`, f.sender, row.date, row.sent) + } + for _, row := range []struct { + at time.Time + kind string + }{ + {since.Add(-time.Second), "spam_placement"}, + {since, "spam_placement"}, + {now, "spam_placement"}, + {now, "user_complaint"}, + } { + f.exec(`INSERT INTO warmup_spam_reports (reporter_account_id, reported_account_id, message_id, report_type, created_at) + VALUES ($1, $2, $3, $4, $5)`, f.sibling, f.sender, uuid.NewString(), row.kind, row.at) + } + ids := []uuid.UUID{f.sender, f.sibling} + summaries, err := r.GetWarmupSummaries(t.Context(), ids, since, now) + if err != nil { + t.Fatal(err) + } + for _, id := range ids { + sent, err := r.SumWarmupSentSince(t.Context(), id, since) + if err != nil { + t.Fatal(err) + } + placed, _, err := r.CountWarmupSpamReportsSince(t.Context(), id, since) + if err != nil { + t.Fatal(err) + } + stats, err := r.GetWarmupStatistics(t.Context(), id, since, now) + if err != nil { + t.Fatal(err) + } + replied := 0 + for _, stat := range stats { + replied += stat.EmailsReplied + } + want := WarmupSummary{Sent: sent, Replied: replied, SpamPlaced: placed} + if summaries[id] != want { + t.Fatalf("%s: got %+v want %+v", id, summaries[id], want) + } + } + if summaries[f.sender] != (WarmupSummary{Sent: 12, Replied: 4, SpamPlaced: 2}) || summaries[f.sibling] != (WarmupSummary{}) { + t.Fatalf("window/isolation mismatch: %+v", summaries) + } + if empty, err := r.GetWarmupSummaries(t.Context(), nil, since, now); err != nil || len(empty) != 0 { + t.Fatalf("empty request: %+v %v", empty, err) + } +}