From a92e40b4d1fdebdae7e67e1989d613ed32f5c10f Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Wed, 7 Oct 2026 06:11:08 +0000 Subject: [PATCH] feat: show Cloud warmup analytics and placement alongside local history on linked self-hosted instances with scoped report APIs --- cmd/backend/main.go | 3 + docs/content/docs/api/endpoints.mdx | 4 + docs/content/docs/guides/warmbly-cloud.mdx | 10 ++ internal/api/handler/analytics.go | 7 + internal/api/handler/poollink_reports.go | 46 ++++++ internal/api/handler/poollink_reports_test.go | 101 ++++++++++++ internal/api/routes.go | 2 + internal/app/analytics/cloud_reports.go | 38 +++++ internal/app/analytics/cloud_reports_test.go | 119 +++++++++++++++ internal/app/analytics/service.go | 16 ++ internal/app/analytics/warmup_placement.go | 31 +++- internal/app/cloudlink/service.go | 2 + internal/app/cloudlink/warmup_reports.go | 93 +++++++++++ internal/app/cloudlink/warmup_reports_test.go | 129 ++++++++++++++++ internal/app/poollink/service.go | 2 + internal/app/poollink/warmup_reports.go | 64 ++++++++ internal/app/poollink/warmup_reports_test.go | 85 +++++++++++ internal/models/analytics.go | 2 +- internal/models/warmup_report.go | 144 ++++++++++++++++++ internal/models/warmup_report_test.go | 50 ++++++ internal/repository/cloudlink_reports.go | 29 ++++ internal/repository/pg_analytics.go | 11 +- internal/repository/pg_cloudlink.go | 1 + internal/repository/pg_poollink.go | 1 + internal/repository/pg_warmup_placement.go | 68 ++++----- internal/repository/poollink_reports.go | 27 ++++ .../warmup_analytics_scope_live_test.go | 45 ++++++ .../repository/warmup_placement_live_test.go | 13 ++ internal/repository/warmup_report.go | 40 +++++ .../components/app/emails/InboxDetails.tsx | 15 +- 30 files changed, 1155 insertions(+), 43 deletions(-) create mode 100644 internal/api/handler/poollink_reports.go create mode 100644 internal/api/handler/poollink_reports_test.go create mode 100644 internal/app/analytics/cloud_reports.go create mode 100644 internal/app/analytics/cloud_reports_test.go create mode 100644 internal/app/cloudlink/warmup_reports.go create mode 100644 internal/app/cloudlink/warmup_reports_test.go create mode 100644 internal/app/poollink/warmup_reports.go create mode 100644 internal/app/poollink/warmup_reports_test.go create mode 100644 internal/models/warmup_report.go create mode 100644 internal/models/warmup_report_test.go create mode 100644 internal/repository/cloudlink_reports.go create mode 100644 internal/repository/poollink_reports.go create mode 100644 internal/repository/warmup_report.go diff --git a/cmd/backend/main.go b/cmd/backend/main.go index df14988ca..f9aac39b5 100644 --- a/cmd/backend/main.go +++ b/cmd/backend/main.go @@ -1292,6 +1292,9 @@ func main() { cloudLinkRepository := repository.NewCloudLinkRepository(primaryDB.Pool, credEncrypter) emailService.WireCloudLink(cloudLinkRepository) cloudLinkService = cloudlink.NewService(cloudLinkRepository, emailRepostory, emailService) + if reports, ok := analyticsService.(analytics.CloudWarmupReportsAware); ok { + reports.WireCloudWarmupReports(cloudLinkService) + } // Deleting a mailbox must also revoke its cloud-held credential. emailService.WireCloudUnenroll(cloudLinkService) emailService.WireCloudCredentials(cloudLinkService) diff --git a/docs/content/docs/api/endpoints.mdx b/docs/content/docs/api/endpoints.mdx index 959b41dc7..519014464 100644 --- a/docs/content/docs/api/endpoints.mdx +++ b/docs/content/docs/api/endpoints.mdx @@ -310,6 +310,10 @@ The `/crm/settings` through `/crm/lists` routes are CRM provider mode ([HubSpot] Analytics reads use the selected workspace for JWT callers and the key's workspace for API-key callers. This includes warmup and usage totals, so teammates see the same workspace history regardless of which member connected a mailbox or created a campaign. +`GET /analytics/warmup` and `GET /analytics/warmup/placement` include detailed Cloud warmup reports for enrolled mailboxes on a linked self-hosted instance, alongside local warmup history. The optional `email_id` is always the local mailbox ID; workspace reports request only that workspace's enrollments. Cloud failures are returned as errors, not empty or partial reports. Daily warmup stats include `active`, whether that date had a warmup plan, so combining sources counts each active UTC date once. + +Cloud exposes `POST /pool-link/instance/analytics/warmup` and `POST /pool-link/instance/analytics/warmup/placement` for instance tokens only, not API keys. Both accept `{ "remote_ids": ["uuid"], "from": "YYYY-MM-DD", "to": "YYYY-MM-DD" }`, with 1 to 100 enrolled remote IDs and up to 366 inclusive UTC days. Invalid ranges or batch sizes return `400`; any ID not enrolled by the calling instance refuses the whole request with `pool_link_mailbox_not_found`. The first returns daily counts, including `active`; the second returns raw `daily`, `hosts`, `sent`, `unconfirmed` and `windows` counts with instance-side remote IDs. Daily placement includes the lookback before `from` needed for rolling rates, hosts and send counts cover the requested range, and windows cover the current trailing window. + `GET /analytics/inbox-tagging` accepts `limit`, opaque `cursor`, and `needs_review` query parameters and returns `data`, aggregate `summary`, and `pagination`. Each row carries `actions`, the list of what the verdict was allowed to do (`hold`, `stop`, `task`, `suppress`), and `summary.acted` counts the rows that did anything. `return_date` (`YYYY-MM-DD`, or `null`) is the out-of-office return date the model was asked to confirm, with its answer under `answers.return_date`. Invalid cursors, limits, or booleans return `400`. `GET /analytics/accounts` returns account health and usage in the standard `data` plus `pagination` envelope with an opaque `cursor`. Pass `email_ids` (comma-separated, up to 200) to ask only for the mailboxes a page shows, or `limit` (`1`-`1000`, default `1000`) and `cursor` to walk the whole inventory. Per-mailbox reads are batched across the page, so a page's cost does not grow with the total inventory, and a workspace with more mailboxes than one page reaches the rest through `next_cursor` instead of having them silently dropped. Invalid cursors, limits, or ids return `400`. diff --git a/docs/content/docs/guides/warmbly-cloud.mdx b/docs/content/docs/guides/warmbly-cloud.mdx index 50e5744eb..cf1d626f9 100644 --- a/docs/content/docs/guides/warmbly-cloud.mdx +++ b/docs/content/docs/guides/warmbly-cloud.mdx @@ -27,6 +27,16 @@ Arrivals awaiting verification are stored in your instance's database and retrie Unenrolling an SMTP/IMAP mailbox, disconnecting the instance, or deleting the mailbox altogether deletes its credential on the cloud immediately. Deleting an enrolled mailbox revokes its enrollment before the record goes, so the pool cannot be left holding a password for a mailbox your instance no longer has; if the cloud cannot be reached at that moment the delete is refused, the mailbox is put back as it was, and retrying it once the cloud is reachable completes it. A mailbox that was signed in through Warmbly Cloud stays in the cloud workspace when you remove it from the instance; disconnecting the instance removes its mirrors of such mailboxes, since they cannot send without the link. +## Warmup analytics on your instance + +Enrolled mailboxes show Cloud's daily sends, replies, received warmup mail and target volumes in **Mailboxes > Analytics**. Their warmup placement reports include daily inbox, tabs, spam and rescued counts, provider and host breakdowns, and the trailing inbox rate. Workspace reports combine Cloud-enrolled mailboxes with mailboxes warming locally. Earlier local warmup history stays visible too. + +Reports are fetched from Cloud when you open or refresh them, using the existing instance link. They are not copied into your instance's warmup tables and do not wait for the five-minute standing poll. Only enrolled mailboxes belonging to the selected local workspace are requested. A Cloud failure shows a load error and a retry action, not an empty chart. + +Existing enrollments work without reconnecting or re-enrolling. Deploy the reporting support on Cloud before upgrading self-hosted instances. An older Cloud deployment without these endpoints cannot supply detailed reports; summary cards and standing synchronization continue working. No database migration or history backfill is needed. + +Once a mailbox is unenrolled, the instance no longer requests its Cloud history. SMTP/IMAP unenrollment deletes the Cloud mailbox and its warmup history; a Cloud-managed mailbox's history stays in its Cloud workspace. The instance's earlier local history is unchanged. + ## Linking your instance The link is offered on a self-hosted instance as the last step of onboarding when you first sign in, as a banner on **Mailboxes** until you connect or dismiss it, under **Settings > Warmbly Cloud** at any time, and on **Sending domains** when a [root redirect](#root-redirects-served-by-warmbly-cloud) could be served from Cloud. Whichever you use, the steps are the same: diff --git a/internal/api/handler/analytics.go b/internal/api/handler/analytics.go index 566c99217..db5906b35 100644 --- a/internal/api/handler/analytics.go +++ b/internal/api/handler/analytics.go @@ -28,6 +28,9 @@ func (h *Handler) GetWarmupAnalytics(c *gin.Context) { if emailIDStr := c.Query("email_id"); emailIDStr != "" { if id, err := uuid.Parse(emailIDStr); err == nil { emailAccountID = &id + } else { + errx.Handle(c, errx.New(errx.BadRequest, "email_id must be a UUID")) + return } } @@ -51,6 +54,10 @@ func (h *Handler) GetWarmupAnalytics(c *gin.Context) { errx.Handle(c, errx.New(errx.BadRequest, "Invalid to date format (expected YYYY-MM-DD)")) return } + if to.Before(from) { + errx.Handle(c, errx.New(errx.BadRequest, "from must be on or before to")) + return + } analytics, xerr := h.AnalyticsService.GetWarmupAnalytics(c.Request.Context(), *orgID, emailAccountID, from, to) if xerr != nil { diff --git a/internal/api/handler/poollink_reports.go b/internal/api/handler/poollink_reports.go new file mode 100644 index 000000000..006e065cd --- /dev/null +++ b/internal/api/handler/poollink_reports.go @@ -0,0 +1,46 @@ +package handler + +import ( + "net/http" + + "github.com/gin-gonic/gin" + "github.com/warmbly/warmbly/internal/api/middleware" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" +) + +func (h *Handler) PoolLinkWarmupStats(c *gin.Context) { + h.poolLinkWarmupReport(c, false) +} + +func (h *Handler) PoolLinkWarmupPlacement(c *gin.Context) { + h.poolLinkWarmupReport(c, true) +} + +func (h *Handler) poolLinkWarmupReport(c *gin.Context, placement bool) { + inst := middleware.GetPoolLinkInstance(c) + if inst == nil { + errx.JSON(c, errx.ErrUnauthorized) + return + } + var req models.PoolLinkWarmupReportRequest + if err := c.ShouldBindJSON(&req); err != nil { + errx.JSON(c, errx.InvalidBody(err)) + return + } + if placement { + report, xerr := h.PoolLinkService.WarmupPlacementData(c.Request.Context(), inst, req) + if xerr != nil { + errx.JSON(c, xerr) + return + } + c.JSON(http.StatusOK, report) + return + } + stats, xerr := h.PoolLinkService.WarmupStats(c.Request.Context(), inst, req) + if xerr != nil { + errx.JSON(c, xerr) + return + } + c.JSON(http.StatusOK, stats) +} diff --git a/internal/api/handler/poollink_reports_test.go b/internal/api/handler/poollink_reports_test.go new file mode 100644 index 000000000..bdb3e82e4 --- /dev/null +++ b/internal/api/handler/poollink_reports_test.go @@ -0,0 +1,101 @@ +package handler + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/gin-gonic/gin" + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/api/middleware" + "github.com/warmbly/warmbly/internal/app/poollink" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" +) + +type instanceReportService struct { + poollink.Service + calls int + failure *errx.Error +} + +func (s *instanceReportService) AuthenticateInstance(_ context.Context, token, _ string) (*models.PoolLinkInstance, *errx.Error) { + if token != "instance-test-token" { + return nil, errx.ErrUnauthorized + } + return &models.PoolLinkInstance{ID: uuid.New()}, nil +} +func (s *instanceReportService) WarmupStats(_ context.Context, inst *models.PoolLinkInstance, req models.PoolLinkWarmupReportRequest) ([]models.WarmupDailyStats, *errx.Error) { + s.calls++ + if inst == nil { + panic("no authenticated instance") + } + if s.failure != nil { + return nil, s.failure + } + if _, _, err := req.Range(); err != nil { + return nil, errx.New(errx.BadRequest, err.Error()) + } + return []models.WarmupDailyStats{{Date: req.From, EmailsSent: 10, Active: true}}, nil +} +func (s *instanceReportService) WarmupPlacementData(_ context.Context, _ *models.PoolLinkInstance, req models.PoolLinkWarmupReportRequest) (*models.WarmupPlacementData, *errx.Error) { + s.calls++ + if s.failure != nil { + return nil, s.failure + } + if _, _, err := req.Range(); err != nil { + return nil, errx.New(errx.BadRequest, err.Error()) + } + return &models.WarmupPlacementData{}, nil +} + +func TestInstanceReportHandlersRequireInstanceTokens(t *testing.T) { + gin.SetMode(gin.TestMode) + for _, placement := range []bool{false, true} { + svc := &instanceReportService{} + h := &Handler{PoolLinkService: svc} + m := &middleware.Handler{PoolLinkService: svc} + router := gin.New() + handler := h.PoolLinkWarmupStats + if placement { + handler = h.PoolLinkWarmupPlacement + } + router.POST("/report", m.PoolLinkAuthMiddleware(), handler) + for _, tc := range []struct { + auth, body string + status int + }{ + {"", "{}", 401}, + {"Bearer ordinary-api-key", "{}", 401}, + {"Bearer revoked-instance-token", "{}", 401}, + {"Bearer instance-test-token", "not JSON", 400}, + {"Bearer instance-test-token", "{}", 400}, + {"Bearer instance-test-token", `{"remote_ids":["` + uuid.NewString() + `"],"from":"2026-10-01","to":"2026-10-02"}`, 200}, + } { + before := svc.calls + req := httptest.NewRequest(http.MethodPost, "/report", strings.NewReader(tc.body)) + req.Header.Set("Authorization", tc.auth) + response := httptest.NewRecorder() + router.ServeHTTP(response, req) + if response.Code != tc.status { + t.Fatalf("placement=%v status=%d wanted=%d body=%s", placement, response.Code, tc.status, response.Body.String()) + } + if tc.status == 401 && svc.calls != before { + t.Fatal("unauthorized call reached reports") + } + if !placement && tc.status == 200 && !strings.Contains(response.Body.String(), `"active":true`) { + t.Fatal("active day was not transported") + } + } + svc.failure = errx.ErrServiceDown + req := httptest.NewRequest(http.MethodPost, "/report", strings.NewReader("{}")) + req.Header.Set("Authorization", "Bearer instance-test-token") + response := httptest.NewRecorder() + router.ServeHTTP(response, req) + if response.Code != http.StatusServiceUnavailable { + t.Fatalf("failure returned %d", response.Code) + } + } +} diff --git a/internal/api/routes.go b/internal/api/routes.go index 24cff11d8..e93847ce3 100644 --- a/internal/api/routes.go +++ b/internal/api/routes.go @@ -1651,6 +1651,8 @@ func Run( poolLinkInstance.DELETE("", h.PoolLinkInstanceDisconnect) poolLinkInstance.GET("/mailboxes", h.PoolLinkInstanceMailboxes) poolLinkInstance.GET("/standing", h.PoolLinkInstanceStanding) + poolLinkInstance.POST("/analytics/warmup", h.PoolLinkWarmupStats) + poolLinkInstance.POST("/analytics/warmup/placement", h.PoolLinkWarmupPlacement) poolLinkInstance.POST("/mailboxes", h.PoolLinkEnroll) poolLinkInstance.GET("/mailboxes/:remoteId", h.PoolLinkGetMailbox) poolLinkInstance.PATCH("/mailboxes/:remoteId", h.PoolLinkPatchMailbox) diff --git a/internal/app/analytics/cloud_reports.go b/internal/app/analytics/cloud_reports.go new file mode 100644 index 000000000..546ca2dc2 --- /dev/null +++ b/internal/app/analytics/cloud_reports.go @@ -0,0 +1,38 @@ +package analytics + +import ( + "context" + "time" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" +) + +type CloudWarmupReports interface { + WarmupStats(ctx context.Context, orgID uuid.UUID, id *uuid.UUID, from, to time.Time) ([]models.WarmupDailyStats, *errx.Error) + WarmupPlacementData(ctx context.Context, orgID uuid.UUID, id *uuid.UUID, from, to time.Time) (*models.WarmupPlacementData, *errx.Error) +} + +type CloudWarmupReportsAware interface { + WireCloudWarmupReports(CloudWarmupReports) +} + +func (s *analyticsService) WireCloudWarmupReports(reports CloudWarmupReports) { + s.cloudReports = reports +} + +func (s *analyticsService) GetWarmupStatsForAccounts(ctx context.Context, orgID uuid.UUID, ids []uuid.UUID, from, to time.Time) ([]models.WarmupDailyStats, *errx.Error) { + return s.analyticsRepo.GetWarmupStatsForAccounts(ctx, orgID, ids, from, to) +} + +func (s *analyticsService) GetWarmupPlacementDataForAccounts(ctx context.Context, orgID uuid.UUID, ids []uuid.UUID, from, to time.Time) (*models.WarmupPlacementData, *errx.Error) { + if s.placementRepo == nil { + return nil, errx.InternalError() + } + data, err := s.placementRepo.ForAccounts(ctx, orgID, ids, from, to) + if err != nil { + return nil, errx.InternalError() + } + return data, nil +} diff --git a/internal/app/analytics/cloud_reports_test.go b/internal/app/analytics/cloud_reports_test.go new file mode 100644 index 000000000..5fec7eabc --- /dev/null +++ b/internal/app/analytics/cloud_reports_test.go @@ -0,0 +1,119 @@ +package analytics + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" +) + +type cloudWarmupStub struct { + stats []models.WarmupDailyStats + placement *models.WarmupPlacementData + failure *errx.Error + calls int +} + +func (s *cloudWarmupStub) WarmupStats(context.Context, uuid.UUID, *uuid.UUID, time.Time, time.Time) ([]models.WarmupDailyStats, *errx.Error) { + s.calls++ + return s.stats, s.failure +} +func (s *cloudWarmupStub) WarmupPlacementData(context.Context, uuid.UUID, *uuid.UUID, time.Time, time.Time) (*models.WarmupPlacementData, *errx.Error) { + s.calls++ + return s.placement, s.failure +} + +type reportPlacementRepo struct { + repository.WarmupPlacementRepository + data models.WarmupPlacementData +} + +func (r reportPlacementRepo) Daily(context.Context, uuid.UUID, *uuid.UUID, time.Time, time.Time) ([]repository.WarmupPlacementDayRow, error) { + return r.data.Daily, nil +} +func (r reportPlacementRepo) Hosts(context.Context, uuid.UUID, *uuid.UUID, time.Time, time.Time) ([]repository.WarmupPlacementHostRow, error) { + return r.data.Hosts, nil +} +func (r reportPlacementRepo) Sent(context.Context, uuid.UUID, *uuid.UUID, time.Time, time.Time) ([]repository.WarmupSenderDayCount, error) { + return r.data.Sent, nil +} +func (r reportPlacementRepo) Unconfirmed(context.Context, uuid.UUID, *uuid.UUID, time.Time, time.Time, time.Time) ([]repository.WarmupSenderDayCount, error) { + return r.data.Unconfirmed, nil +} +func (r reportPlacementRepo) Rates(context.Context, uuid.UUID, *uuid.UUID, time.Time) (map[uuid.UUID]models.WarmupPlacementWindow, error) { + return r.data.Windows, nil +} + +func TestCloudAndLocalWarmupHistoryMergeByDay(t *testing.T) { + cloud := &cloudWarmupStub{stats: []models.WarmupDailyStats{{Date: "2026-10-01", EmailsSent: 9, EmailsReplied: 3, EmailsReceived: 4, TargetVolume: 10, Active: true}}} + s := &analyticsService{analyticsRepo: warmupAnalyticsRepoStub{stats: []models.WarmupDailyStats{ + {Date: "2026-09-30", EmailsSent: 2, TargetVolume: 2, Active: true}, + {Date: "2026-10-01", EmailsSent: 1, EmailsReceived: 2, TargetVolume: 2, Active: true}, + }}, cloudReports: cloud} + report, xerr := s.GetWarmupAnalytics(context.Background(), uuid.New(), nil, time.Time{}, time.Now()) + if xerr != nil || len(report.DailyStats) != 2 || report.Summary.TotalSent != 12 || report.Summary.TotalReplied != 3 || report.Summary.TotalReceived != 6 || report.Summary.DaysActive != 2 || report.Summary.AverageDaily != 6 || report.Summary.ReplyRate != 25 { + t.Fatalf("report=%+v error=%v", report, xerr) + } + cloud.failure = errx.ErrServiceDown + if report, xerr := s.GetWarmupAnalytics(context.Background(), uuid.New(), nil, time.Time{}, time.Now()); xerr == nil || report != nil { + t.Fatalf("returned a partial report during Cloud failure: %+v, %v", report, xerr) + } +} + +func TestMixedPlacementUsesCountsAndCloudLookbackNotAveragedRates(t *testing.T) { + localID, cloudID := uuid.New(), uuid.New() + day := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC) + local := reportPlacementRepo{data: models.WarmupPlacementData{ + Daily: []models.WarmupPlacementDayRow{{SenderID: localID, Email: "local@test.local", Date: "2026-10-01", Group: "google", Spam: 2}}, + Hosts: []models.WarmupPlacementHostRow{{Group: "google", Host: "google_workspace", Spam: 2}}, + Sent: []models.WarmupSenderDayCount{{SenderID: localID, Email: "local@test.local", Date: "2026-10-01", Count: 2}}, + Windows: map[uuid.UUID]models.WarmupPlacementWindow{localID: {Major: models.WarmupPlacementTally{Spam: 10}, All: models.WarmupPlacementTally{Spam: 10}}}, + }} + cloud := &cloudWarmupStub{placement: &models.WarmupPlacementData{ + Daily: []models.WarmupPlacementDayRow{ + {SenderID: cloudID, Email: "cloud@test.local", Date: "2026-09-30", Group: "google", Inbox: 18, Spam: 2}, + {SenderID: cloudID, Email: "cloud@test.local", Date: "2026-10-01", Group: "google", Inbox: 8, Spam: 2}, + }, + Hosts: []models.WarmupPlacementHostRow{{Group: "google", Host: "google_workspace", Inbox: 8, Spam: 2}}, + Sent: []models.WarmupSenderDayCount{{SenderID: cloudID, Email: "cloud@test.local", Date: "2026-10-01", Count: 10}}, + Windows: map[uuid.UUID]models.WarmupPlacementWindow{cloudID: {Major: models.WarmupPlacementTally{Inbox: 90, Spam: 10}, All: models.WarmupPlacementTally{Inbox: 90, Spam: 10}}}, + }} + s := &analyticsService{placementRepo: local, cloudReports: cloud} + report, xerr := s.GetWarmupPlacement(context.Background(), uuid.New(), nil, day, day) + if xerr != nil || report.Summary.Delivered != 12 || report.Summary.Sent != 12 || report.Rate.InboxRate == nil || *report.Rate.InboxRate != 81.82 || len(report.Mailboxes) != 2 { + t.Fatalf("report=%+v error=%v", report, xerr) + } + if len(report.Providers) != 1 || len(report.Providers[0].Hosts) != 1 || report.Providers[0].Hosts[0].Delivered != 12 || report.Daily[0].RollingInboxRate == nil || *report.Daily[0].RollingInboxRate != 81.25 { + t.Fatalf("hosts or rolling lookback were not merged: %+v, %+v", report.Providers, report.Daily) + } + cloud.failure = errx.ErrServiceDown + if report, xerr := s.GetWarmupPlacement(context.Background(), uuid.New(), nil, day, day); xerr == nil || report != nil { + t.Fatalf("partial placement: %+v, %v", report, xerr) + } +} + +type foreignReportEmailRepo struct{ repository.EmailRepository } + +func (foreignReportEmailRepo) Get(context.Context, string, string) (*models.Email, *errx.Error) { + return nil, errx.ErrNotFound +} + +func TestCloudReportsRefuseMailboxesOutsideTheLocalWorkspace(t *testing.T) { + cloud := &cloudWarmupStub{} + s := &analyticsService{analyticsRepo: warmupAnalyticsRepoStub{}, emailRepo: foreignReportEmailRepo{}, cloudReports: cloud} + id, org := uuid.New(), uuid.New() + day := time.Now().UTC() + if _, xerr := s.GetWarmupAnalytics(context.Background(), org, &id, day, day); xerr != errx.ErrNotFound { + t.Fatalf("stats ownership: %v", xerr) + } + if _, xerr := s.GetWarmupPlacement(context.Background(), org, &id, day, day); xerr != errx.ErrNotFound { + t.Fatalf("placement ownership: %v", xerr) + } + if cloud.calls != 0 { + t.Fatal("foreign mailbox was requested from Cloud") + } +} diff --git a/internal/app/analytics/service.go b/internal/app/analytics/service.go index 0ead2b12b..ffc27f1b5 100644 --- a/internal/app/analytics/service.go +++ b/internal/app/analytics/service.go @@ -19,6 +19,8 @@ type AnalyticsService interface { GetWarmupAnalytics(ctx context.Context, orgID uuid.UUID, emailAccountID *uuid.UUID, from, to time.Time) (*models.WarmupAnalytics, *errx.Error) // GetWarmupPlacement is where warmup mail landed, for one mailbox or the workspace. GetWarmupPlacement(ctx context.Context, orgID uuid.UUID, emailAccountID *uuid.UUID, from, to time.Time) (*models.WarmupPlacementReport, *errx.Error) + GetWarmupStatsForAccounts(ctx context.Context, orgID uuid.UUID, ids []uuid.UUID, from, to time.Time) ([]models.WarmupDailyStats, *errx.Error) + GetWarmupPlacementDataForAccounts(ctx context.Context, orgID uuid.UUID, ids []uuid.UUID, from, to time.Time) (*models.WarmupPlacementData, *errx.Error) // Campaign analytics // GetCampaignAnalytics reads the performance of the sends inside period, @@ -60,6 +62,7 @@ type analyticsService struct { lifecycleRepo repository.SendLifecycleRepository // placementRepo reads warmup placement history. Optional/nil-safe. placementRepo repository.WarmupPlacementRepository + cloudReports CloudWarmupReports } func NewService( @@ -85,6 +88,19 @@ func (s *analyticsService) GetWarmupAnalytics(ctx context.Context, orgID uuid.UU return nil, xerr } + if s.cloudReports != nil { + if emailAccountID != nil { + if _, xerr := s.emailRepo.Get(ctx, orgID.String(), emailAccountID.String()); xerr != nil { + return nil, xerr + } + } + cloud, xerr := s.cloudReports.WarmupStats(ctx, orgID, emailAccountID, from, to) + if xerr != nil { + return nil, xerr + } + dailyStats = models.MergeWarmupStats(dailyStats, cloud) + } + // Calculate summary var totalSent, totalReplied, totalReceived, totalTarget, daysActive int for _, day := range dailyStats { diff --git a/internal/app/analytics/warmup_placement.go b/internal/app/analytics/warmup_placement.go index 3f6b25c15..010640b7c 100644 --- a/internal/app/analytics/warmup_placement.go +++ b/internal/app/analytics/warmup_placement.go @@ -14,7 +14,7 @@ import ( ) // MaxWarmupPlacementDays bounds one placement report. -const MaxWarmupPlacementDays = 366 +const MaxWarmupPlacementDays = models.WarmupReportMaxDays // WireWarmupPlacement attaches the placement history. func (s *analyticsService) WireWarmupPlacement(r repository.WarmupPlacementRepository) { @@ -122,6 +122,15 @@ func (s *analyticsService) GetWarmupPlacement(ctx context.Context, orgID uuid.UU if err != nil { return nil, errx.InternalError() } + if s.cloudReports != nil { + cloud, xerr := s.cloudReports.WarmupPlacementData(ctx, orgID, emailID, from, to) + if xerr != nil { + return nil, xerr + } + data := &models.WarmupPlacementData{Daily: dayRows, Hosts: hostRows, Sent: sentRows, Unconfirmed: unconfirmedRows, Windows: windows} + data.Add(cloud) + dayRows, hostRows, sentRows, unconfirmedRows, windows = data.Daily, data.Hosts, data.Sent, data.Unconfirmed, data.Windows + } rates := headlineRates(windows) b := newPlacementBuilder(from, to) @@ -336,8 +345,19 @@ func (b *placementBuilder) mailboxes(names map[uuid.UUID]string, rates map[uuid. // placementProviders groups the window's host rows by recipient group, in // display order, busiest host first. func placementProviders(rows []repository.WarmupPlacementHostRow) []models.WarmupPlacementProvider { + byHost := make(map[[2]string]repository.WarmupPlacementHostRow) + for _, row := range rows { + key := [2]string{row.Group, row.Host} + host := byHost[key] + host.Group, host.Host = row.Group, row.Host + host.Inbox += row.Inbox + host.Tabs += row.Tabs + host.Spam += row.Spam + host.Rescued += row.Rescued + byHost[key] = host + } byGroup := make(map[string]*models.WarmupPlacementProvider) - for _, r := range rows { + for _, r := range byHost { p, ok := byGroup[r.Group] if !ok { p = &models.WarmupPlacementProvider{Group: r.Group, Hosts: make([]models.WarmupPlacementHost, 0)} @@ -355,7 +375,12 @@ func placementProviders(rows []repository.WarmupPlacementHostRow) []models.Warmu continue } p.Finish() - sort.SliceStable(p.Hosts, func(i, j int) bool { return p.Hosts[i].Delivered > p.Hosts[j].Delivered }) + sort.Slice(p.Hosts, func(i, j int) bool { + if p.Hosts[i].Delivered == p.Hosts[j].Delivered { + return p.Hosts[i].Host < p.Hosts[j].Host + } + return p.Hosts[i].Delivered > p.Hosts[j].Delivered + }) out = append(out, *p) } return out diff --git a/internal/app/cloudlink/service.go b/internal/app/cloudlink/service.go index d6edf5a35..9c133ca22 100644 --- a/internal/app/cloudlink/service.go +++ b/internal/app/cloudlink/service.go @@ -104,6 +104,8 @@ type Service interface { Disconnect(ctx context.Context) *errx.Error ListMailboxes(ctx context.Context, orgID uuid.UUID) ([]models.CloudLinkMailboxRow, *errx.Error) + WarmupStats(ctx context.Context, orgID uuid.UUID, id *uuid.UUID, from, to time.Time) ([]models.WarmupDailyStats, *errx.Error) + WarmupPlacementData(ctx context.Context, orgID uuid.UUID, id *uuid.UUID, from, to time.Time) (*models.WarmupPlacementData, *errx.Error) Enroll(ctx context.Context, orgID, accountID uuid.UUID) (*models.CloudLinkMailboxRow, *errx.Error) Unenroll(ctx context.Context, orgID, accountID uuid.UUID) *errx.Error // RefreshCredentials re-sends an enrolled mailbox's credential and ramp after they change here. diff --git a/internal/app/cloudlink/warmup_reports.go b/internal/app/cloudlink/warmup_reports.go new file mode 100644 index 000000000..ce197b75f --- /dev/null +++ b/internal/app/cloudlink/warmup_reports.go @@ -0,0 +1,93 @@ +package cloudlink + +import ( + "context" + "net/http" + "time" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" +) + +func (s *service) reportMailboxes(ctx context.Context, orgID uuid.UUID, id *uuid.UUID) ([]models.CloudLinkMailbox, *errx.Error) { + rows, err := s.repo.ListForOrg(ctx, orgID, id) + if err != nil { + return nil, errx.InternalError() + } + return rows, nil +} + +func reportRequest(mailboxes []models.CloudLinkMailbox, from, to time.Time) models.PoolLinkWarmupReportRequest { + req := models.PoolLinkWarmupReportRequest{From: from.Format(time.DateOnly), To: to.Format(time.DateOnly)} + for _, mailbox := range mailboxes { + req.RemoteIDs = append(req.RemoteIDs, mailbox.RemoteID) + } + return req +} + +func (s *service) WarmupStats(ctx context.Context, orgID uuid.UUID, id *uuid.UUID, from, to time.Time) ([]models.WarmupDailyStats, *errx.Error) { + mailboxes, xerr := s.reportMailboxes(ctx, orgID, id) + if xerr != nil || len(mailboxes) == 0 { + return nil, xerr + } + l, xerr := s.link(ctx) + if xerr != nil { + return nil, xerr + } + var out []models.WarmupDailyStats + c := s.clientFor(l) + for start := from; !start.After(to); { + end := start.AddDate(0, 0, models.WarmupReportMaxDays-1) + if end.After(to) { + end = to + } + for i := 0; i < len(mailboxes); i += models.WarmupReportBatchSize { + batch := mailboxes[i:min(i+models.WarmupReportBatchSize, len(mailboxes))] + var rows []models.WarmupDailyStats + if xerr := c.do(ctx, http.MethodPost, "/instance/analytics/warmup", reportRequest(batch, start, end), &rows); xerr != nil { + return nil, xerr + } + if rows == nil { + return nil, errx.NewWithIdentifier(errx.ServiceUnavailable, "cloud_link_bad_response", "Warmbly Cloud returned an unreadable report.") + } + out = models.MergeWarmupStats(out, rows) + } + start = end.AddDate(0, 0, 1) + } + return out, nil +} + +func (s *service) WarmupPlacementData(ctx context.Context, orgID uuid.UUID, id *uuid.UUID, from, to time.Time) (*models.WarmupPlacementData, *errx.Error) { + mailboxes, xerr := s.reportMailboxes(ctx, orgID, id) + if xerr != nil || len(mailboxes) == 0 { + return nil, xerr + } + l, xerr := s.link(ctx) + if xerr != nil { + return nil, xerr + } + out := &models.WarmupPlacementData{} + c := s.clientFor(l) + // Placement includes per-sender days; keep a year of rows within the client's response limit. + const batchSize = 10 + for i := 0; i < len(mailboxes); i += batchSize { + batch := mailboxes[i:min(i+batchSize, len(mailboxes))] + var part *models.WarmupPlacementData + if xerr := c.do(ctx, http.MethodPost, "/instance/analytics/warmup/placement", reportRequest(batch, from, to), &part); xerr != nil { + return nil, xerr + } + if part == nil || part.Daily == nil || part.Hosts == nil || part.Sent == nil || part.Unconfirmed == nil || part.Windows == nil { + return nil, errx.NewWithIdentifier(errx.ServiceUnavailable, "cloud_link_bad_response", "Warmbly Cloud returned an unreadable report.") + } + localIDs := make(map[uuid.UUID]uuid.UUID, len(batch)) + for _, mailbox := range batch { + localIDs[mailbox.RemoteID] = mailbox.EmailAccountID + } + if err := part.RemapSenders(localIDs); err != nil { + return nil, errx.NewWithIdentifier(errx.ServiceUnavailable, "cloud_link_bad_response", "Warmbly Cloud returned an unreadable report.") + } + out.Add(part) + } + return out, nil +} diff --git a/internal/app/cloudlink/warmup_reports_test.go b/internal/app/cloudlink/warmup_reports_test.go new file mode 100644 index 000000000..16fbf6814 --- /dev/null +++ b/internal/app/cloudlink/warmup_reports_test.go @@ -0,0 +1,129 @@ +package cloudlink + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" +) + +type warmupReportRepo struct { + repository.CloudLinkRepository + link *models.CloudLink + org uuid.UUID + mailboxes []models.CloudLinkMailbox +} + +func (r warmupReportRepo) Get(context.Context) (*models.CloudLink, error) { return r.link, nil } +func (r warmupReportRepo) ListForOrg(_ context.Context, org uuid.UUID, id *uuid.UUID) ([]models.CloudLinkMailbox, error) { + out := make([]models.CloudLinkMailbox, 0) + if org == r.org { + for _, mailbox := range r.mailboxes { + if id == nil || mailbox.EmailAccountID == *id { + out = append(out, mailbox) + } + } + } + return out, nil +} + +func TestWarmupReportsBatchAndRemapExistingEnrollments(t *testing.T) { + t.Setenv("APP_ENV", "dev") + org := uuid.New() + var mailboxes []models.CloudLinkMailbox + for i := 0; i < 101; i++ { + mailboxes = append(mailboxes, models.CloudLinkMailbox{EmailAccountID: uuid.New(), RemoteID: uuid.New(), Managed: i%2 == 0}) + } + statsCalls, placementCalls := 0, 0 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.Header.Get("Authorization") != "Bearer test" { + t.Errorf("unexpected request method or authorization") + } + var req models.PoolLinkWarmupReportRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Error(err) + return + } + if _, _, err := req.Range(); err != nil { + t.Error(err) + return + } + switch r.URL.Path { + case "/v1/pool-link/instance/analytics/warmup": + statsCalls++ + _ = json.NewEncoder(w).Encode([]models.WarmupDailyStats{{Date: req.From, EmailsSent: len(req.RemoteIDs), Active: true}}) + case "/v1/pool-link/instance/analytics/warmup/placement": + placementCalls++ + data := models.WarmupPlacementData{Hosts: []models.WarmupPlacementHostRow{}, Windows: map[uuid.UUID]models.WarmupPlacementWindow{}} + for _, id := range req.RemoteIDs { + data.Daily = append(data.Daily, models.WarmupPlacementDayRow{SenderID: id, Date: req.From, Inbox: 1}) + data.Sent = append(data.Sent, models.WarmupSenderDayCount{SenderID: id, Count: 1}) + data.Unconfirmed = append(data.Unconfirmed, models.WarmupSenderDayCount{SenderID: id, Count: 1}) + data.Windows[id] = models.WarmupPlacementWindow{Major: models.WarmupPlacementTally{Inbox: 1}} + } + _ = json.NewEncoder(w).Encode(data) + default: + t.Errorf("unexpected path: %s", r.URL.Path) + http.NotFound(w, r) + } + })) + defer srv.Close() + s := &service{repo: warmupReportRepo{link: &models.CloudLink{CloudURL: srv.URL, Token: "test"}, org: org, mailboxes: mailboxes}} + from := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + stats, xerr := s.WarmupStats(context.Background(), org, nil, from, from.AddDate(0, 0, 366)) + if xerr != nil || statsCalls != 4 || len(stats) != 2 || stats[0].EmailsSent != 101 || stats[1].EmailsSent != 101 || !stats[0].Active { + t.Fatalf("stats=%+v error=%v calls=%d", stats, xerr, statsCalls) + } + data, xerr := s.WarmupPlacementData(context.Background(), org, nil, from, from) + if xerr != nil || placementCalls != 11 || len(data.Daily) != 101 || len(data.Windows) != 101 { + t.Fatalf("placement error=%v calls=%d data=%+v", xerr, placementCalls, data) + } + for i, m := range mailboxes { + if data.Daily[i].SenderID != m.EmailAccountID || data.Sent[i].SenderID != m.EmailAccountID || data.Unconfirmed[i].SenderID != m.EmailAccountID || data.Windows[m.EmailAccountID].Major.Inbox != 1 { + t.Fatalf("remote IDs were not mapped: %+v", data.Daily[i]) + } + } + before := statsCalls + if rows, xerr := s.WarmupStats(context.Background(), uuid.New(), nil, from, from); xerr != nil || len(rows) != 0 || statsCalls != before { + t.Fatalf("another workspace reached Cloud: %v", xerr) + } + if _, xerr := s.WarmupStats(context.Background(), org, &mailboxes[0].EmailAccountID, from, from); xerr != nil || statsCalls != before+1 { + t.Fatalf("single mailbox report: %v", xerr) + } +} + +func TestWarmupReportsDoNotTurnCloudFailuresIntoEmptyActivity(t *testing.T) { + t.Setenv("APP_ENV", "dev") + for _, tc := range []struct { + name string + status int + body string + }{ + {"old cloud", 404, "{}"}, {"outage", 503, "{}"}, {"revoked", 401, "{}"}, + {"invalid JSON", 200, "bad"}, {"empty response", 200, ""}, {"null response", 200, "null"}, + {"missing fields", 200, "{}"}, + } { + t.Run(tc.name, func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(tc.status) + _, _ = w.Write([]byte(tc.body)) + })) + defer srv.Close() + org, local := uuid.New(), uuid.New() + s := &service{repo: warmupReportRepo{link: &models.CloudLink{CloudURL: srv.URL}, org: org, mailboxes: []models.CloudLinkMailbox{{EmailAccountID: local, RemoteID: uuid.New()}}}} + day := time.Now().UTC() + if _, xerr := s.WarmupStats(context.Background(), org, &local, day, day); xerr == nil { + t.Fatal("stats failure was swallowed") + } + if _, xerr := s.WarmupPlacementData(context.Background(), org, &local, day, day); xerr == nil { + t.Fatal("placement failure was swallowed") + } + }) + } +} diff --git a/internal/app/poollink/service.go b/internal/app/poollink/service.go index 34782adf6..54f6f7221 100644 --- a/internal/app/poollink/service.go +++ b/internal/app/poollink/service.go @@ -74,6 +74,8 @@ type Service interface { Enroll(ctx context.Context, inst *models.PoolLinkInstance, req models.PoolLinkEnrollRequest) (*models.PoolLinkMailboxState, *errx.Error) ListMailboxes(ctx context.Context, inst *models.PoolLinkInstance) ([]models.PoolLinkMailboxState, *errx.Error) + WarmupStats(ctx context.Context, inst *models.PoolLinkInstance, req models.PoolLinkWarmupReportRequest) ([]models.WarmupDailyStats, *errx.Error) + WarmupPlacementData(ctx context.Context, inst *models.PoolLinkInstance, req models.PoolLinkWarmupReportRequest) (*models.WarmupPlacementData, *errx.Error) // ListStanding is the warmup standing of every enrolled mailbox, which the // instance polls so its own send gates hold the cloud's verdict. ListStanding(ctx context.Context, inst *models.PoolLinkInstance) ([]models.PoolLinkMailboxStanding, *errx.Error) diff --git a/internal/app/poollink/warmup_reports.go b/internal/app/poollink/warmup_reports.go new file mode 100644 index 000000000..874ccf8d4 --- /dev/null +++ b/internal/app/poollink/warmup_reports.go @@ -0,0 +1,64 @@ +package poollink + +import ( + "context" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" +) + +func (s *service) reportAccounts(ctx context.Context, inst *models.PoolLinkInstance, remoteIDs []uuid.UUID) ([]uuid.UUID, map[uuid.UUID]uuid.UUID, *errx.Error) { + enrolled, err := s.repo.ListReportMailboxes(ctx, inst.ID, remoteIDs) + if err != nil { + return nil, nil, errx.InternalError() + } + byRemote := make(map[uuid.UUID]uuid.UUID, len(enrolled)) + for _, mailbox := range enrolled { + byRemote[mailbox.RemoteID] = mailbox.EmailAccountID + } + ids := make([]uuid.UUID, 0, len(remoteIDs)) + remotes := make(map[uuid.UUID]uuid.UUID, len(remoteIDs)) + for _, remote := range remoteIDs { + id, ok := byRemote[remote] + if !ok { + return nil, nil, ErrMailboxNotFound + } + if _, seen := remotes[id]; !seen { + ids = append(ids, id) + remotes[id] = remote + } + } + return ids, remotes, nil +} + +func (s *service) WarmupStats(ctx context.Context, inst *models.PoolLinkInstance, req models.PoolLinkWarmupReportRequest) ([]models.WarmupDailyStats, *errx.Error) { + from, to, err := req.Range() + if err != nil { + return nil, errx.New(errx.BadRequest, err.Error()) + } + ids, _, xerr := s.reportAccounts(ctx, inst, req.RemoteIDs) + if xerr != nil { + return nil, xerr + } + return s.analytics.GetWarmupStatsForAccounts(ctx, inst.OrganizationID, ids, from, to) +} + +func (s *service) WarmupPlacementData(ctx context.Context, inst *models.PoolLinkInstance, req models.PoolLinkWarmupReportRequest) (*models.WarmupPlacementData, *errx.Error) { + from, to, err := req.Range() + if err != nil { + return nil, errx.New(errx.BadRequest, err.Error()) + } + ids, remotes, xerr := s.reportAccounts(ctx, inst, req.RemoteIDs) + if xerr != nil { + return nil, xerr + } + data, xerr := s.analytics.GetWarmupPlacementDataForAccounts(ctx, inst.OrganizationID, ids, from, to) + if xerr != nil { + return nil, xerr + } + if err := data.RemapSenders(remotes); err != nil { + return nil, errx.InternalError() + } + return data, nil +} diff --git a/internal/app/poollink/warmup_reports_test.go b/internal/app/poollink/warmup_reports_test.go new file mode 100644 index 000000000..21bf10577 --- /dev/null +++ b/internal/app/poollink/warmup_reports_test.go @@ -0,0 +1,85 @@ +package poollink + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/app/analytics" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" +) + +type reportMailboxRepo struct { + repository.PoolLinkRepository + instance uuid.UUID + mailbox models.PoolLinkMailbox +} + +func (r reportMailboxRepo) ListReportMailboxes(_ context.Context, instance uuid.UUID, requested []uuid.UUID) ([]models.PoolLinkMailbox, error) { + if instance != r.instance { + return nil, nil + } + for _, id := range requested { + if id == r.mailbox.RemoteID { + return []models.PoolLinkMailbox{r.mailbox}, nil + } + } + return nil, nil +} + +type reportAnalytics struct { + analytics.AnalyticsService + org uuid.UUID + ids []uuid.UUID + calls int +} + +func (a *reportAnalytics) GetWarmupStatsForAccounts(_ context.Context, org uuid.UUID, ids []uuid.UUID, _, _ time.Time) ([]models.WarmupDailyStats, *errx.Error) { + a.org, a.ids = org, ids + a.calls++ + return []models.WarmupDailyStats{{Date: "2026-10-01", EmailsSent: 9, Active: true}}, nil +} + +func (a *reportAnalytics) GetWarmupPlacementDataForAccounts(_ context.Context, org uuid.UUID, ids []uuid.UUID, _, _ time.Time) (*models.WarmupPlacementData, *errx.Error) { + a.org, a.ids = org, ids + a.calls++ + return &models.WarmupPlacementData{ + Daily: []models.WarmupPlacementDayRow{{SenderID: ids[0], Inbox: 9}}, + Sent: []models.WarmupSenderDayCount{{SenderID: ids[0], Count: 10}}, + Unconfirmed: []models.WarmupSenderDayCount{{SenderID: ids[0], Count: 1}}, + Windows: map[uuid.UUID]models.WarmupPlacementWindow{ids[0]: {Major: models.WarmupPlacementTally{Inbox: 9}}}, + }, nil +} + +func TestReportsScopeAndMapTheCallingInstancesMailboxes(t *testing.T) { + inst := &models.PoolLinkInstance{ID: uuid.New(), OrganizationID: uuid.New()} + mailbox := models.PoolLinkMailbox{RemoteID: uuid.New(), EmailAccountID: uuid.New()} + a := &reportAnalytics{} + s := &service{repo: reportMailboxRepo{instance: inst.ID, mailbox: mailbox}, analytics: a} + req := models.PoolLinkWarmupReportRequest{RemoteIDs: []uuid.UUID{mailbox.RemoteID, mailbox.RemoteID}, From: "2026-10-01", To: "2026-10-02"} + stats, xerr := s.WarmupStats(context.Background(), inst, req) + if xerr != nil || len(stats) != 1 || !stats[0].Active || a.org != inst.OrganizationID || len(a.ids) != 1 || a.ids[0] != mailbox.EmailAccountID { + t.Fatalf("stats=%+v error=%v scope=%v/%v", stats, xerr, a.org, a.ids) + } + data, xerr := s.WarmupPlacementData(context.Background(), inst, req) + if xerr != nil || data.Daily[0].SenderID != mailbox.RemoteID || data.Sent[0].SenderID != mailbox.RemoteID || data.Unconfirmed[0].SenderID != mailbox.RemoteID || data.Windows[mailbox.RemoteID].Major.Inbox != 9 { + t.Fatalf("data=%+v error=%v", data, xerr) + } + calls := a.calls + for _, ids := range [][]uuid.UUID{{uuid.New()}, {mailbox.RemoteID, uuid.New()}, {mailbox.EmailAccountID}} { + req.RemoteIDs = ids + if _, xerr := s.WarmupStats(context.Background(), inst, req); xerr != ErrMailboxNotFound || a.calls != calls { + t.Fatalf("unauthorized IDs reached analytics: %v, calls=%d", xerr, a.calls) + } + if _, xerr := s.WarmupPlacementData(context.Background(), inst, req); xerr != ErrMailboxNotFound || a.calls != calls { + t.Fatalf("unauthorized placement IDs reached analytics: %v", xerr) + } + } + req.RemoteIDs = []uuid.UUID{mailbox.RemoteID} + if _, xerr := s.WarmupStats(context.Background(), &models.PoolLinkInstance{ID: uuid.New()}, req); xerr != ErrMailboxNotFound { + t.Fatalf("another instance accessed the report: %v", xerr) + } +} diff --git a/internal/models/analytics.go b/internal/models/analytics.go index 0f3f99344..d60e71324 100644 --- a/internal/models/analytics.go +++ b/internal/models/analytics.go @@ -67,7 +67,7 @@ type WarmupDailyStats struct { TargetVolume int `json:"target_volume"` // Active is whether the mailbox had a warmup plan that day. A day with // arrivals but no plan still lists, but does not count as a day active. - Active bool `json:"-"` + Active bool `json:"active"` } // Campaign Analytics diff --git a/internal/models/warmup_report.go b/internal/models/warmup_report.go new file mode 100644 index 000000000..ec69d5cee --- /dev/null +++ b/internal/models/warmup_report.go @@ -0,0 +1,144 @@ +package models + +import ( + "fmt" + "sort" + "time" + + "github.com/google/uuid" +) + +const ( + WarmupReportBatchSize = 100 + WarmupReportMaxDays = 366 +) + +type PoolLinkWarmupReportRequest struct { + RemoteIDs []uuid.UUID `json:"remote_ids"` + From string `json:"from"` + To string `json:"to"` +} + +func (r PoolLinkWarmupReportRequest) Range() (time.Time, time.Time, error) { + if len(r.RemoteIDs) == 0 || len(r.RemoteIDs) > WarmupReportBatchSize { + return time.Time{}, time.Time{}, fmt.Errorf("remote_ids must contain 1 to %d mailboxes", WarmupReportBatchSize) + } + period, err := ParseDayRange(r.From, r.To) + if err != nil { + return time.Time{}, time.Time{}, err + } + if period == nil || period.To.Sub(period.From).Hours()/24 >= WarmupReportMaxDays { + return time.Time{}, time.Time{}, fmt.Errorf("a warmup report requires a range of at most %d days", WarmupReportMaxDays) + } + return period.From, period.To, nil +} + +type WarmupPlacementDayRow struct { + SenderID uuid.UUID `json:"sender_id"` + Email string `json:"email"` + Date string `json:"date"` + Group string `json:"group"` + Inbox int `json:"inbox"` + Tabs int `json:"tabs"` + Spam int `json:"spam"` + Rescued int `json:"rescued"` +} + +type WarmupPlacementHostRow struct { + Group string `json:"group"` + Host string `json:"host"` + Inbox int `json:"inbox"` + Tabs int `json:"tabs"` + Spam int `json:"spam"` + Rescued int `json:"rescued"` +} + +type WarmupSenderDayCount struct { + SenderID uuid.UUID `json:"sender_id"` + Email string `json:"email"` + Date string `json:"date"` + Count int `json:"count"` +} + +// WarmupPlacementData carries counts, not rounded rates, so mixed pools are weighted correctly. +type WarmupPlacementData struct { + Daily []WarmupPlacementDayRow `json:"daily"` + Hosts []WarmupPlacementHostRow `json:"hosts"` + Sent []WarmupSenderDayCount `json:"sent"` + Unconfirmed []WarmupSenderDayCount `json:"unconfirmed"` + Windows map[uuid.UUID]WarmupPlacementWindow `json:"windows"` +} + +func MergeWarmupStats(local, cloud []WarmupDailyStats) []WarmupDailyStats { + byDate := make(map[string]WarmupDailyStats, len(local)+len(cloud)) + for _, rows := range [][]WarmupDailyStats{local, cloud} { + for _, row := range rows { + day := byDate[row.Date] + day.Date = row.Date + day.EmailsSent += row.EmailsSent + day.EmailsReplied += row.EmailsReplied + day.EmailsReceived += row.EmailsReceived + day.TargetVolume += row.TargetVolume + day.Active = day.Active || row.Active + byDate[row.Date] = day + } + } + out := make([]WarmupDailyStats, 0, len(byDate)) + for _, day := range byDate { + out = append(out, day) + } + sort.Slice(out, func(i, j int) bool { return out[i].Date < out[j].Date }) + return out +} + +func (d *WarmupPlacementData) Add(other *WarmupPlacementData) { + if other == nil { + return + } + d.Daily = append(d.Daily, other.Daily...) + d.Hosts = append(d.Hosts, other.Hosts...) + d.Sent = append(d.Sent, other.Sent...) + d.Unconfirmed = append(d.Unconfirmed, other.Unconfirmed...) + if d.Windows == nil { + d.Windows = make(map[uuid.UUID]WarmupPlacementWindow) + } + for id, window := range other.Windows { + current := d.Windows[id] + current.Add(window) + d.Windows[id] = current + } +} + +func (d *WarmupPlacementData) RemapSenders(ids map[uuid.UUID]uuid.UUID) error { + remap := func(id *uuid.UUID) error { + target, ok := ids[*id] + if !ok { + return fmt.Errorf("unrequested warmup report sender") + } + *id = target + return nil + } + for i := range d.Daily { + if err := remap(&d.Daily[i].SenderID); err != nil { + return err + } + } + for _, rows := range [][]WarmupSenderDayCount{d.Sent, d.Unconfirmed} { + for i := range rows { + if err := remap(&rows[i].SenderID); err != nil { + return err + } + } + } + windows := make(map[uuid.UUID]WarmupPlacementWindow, len(d.Windows)) + for id, window := range d.Windows { + if err := remap(&id); err != nil { + return err + } + current := windows[id] + current.Add(window) + windows[id] = current + } + d.Windows = windows + return nil +} diff --git a/internal/models/warmup_report_test.go b/internal/models/warmup_report_test.go new file mode 100644 index 000000000..d6bdafbde --- /dev/null +++ b/internal/models/warmup_report_test.go @@ -0,0 +1,50 @@ +package models + +import ( + "encoding/json" + "testing" + + "github.com/google/uuid" +) + +func TestWarmupReportRangeIsBounded(t *testing.T) { + for _, tc := range []struct { + from, to string + count int + valid bool + }{ + {"2026-01-01", "2026-01-01", 1, true}, + {"2024-01-01", "2024-12-31", 100, true}, + {"2024-01-01", "2025-01-01", 1, false}, + {"2026-01-02", "2026-01-01", 1, false}, + {"", "", 1, false}, + {"2026-01-01", "", 1, false}, + {"invalid", "2026-01-01", 1, false}, + {"2026-01-01", "2026-01-01", 0, false}, + {"2026-01-01", "2026-01-01", 101, false}, + } { + req := PoolLinkWarmupReportRequest{RemoteIDs: make([]uuid.UUID, tc.count), From: tc.from, To: tc.to} + from, to, err := req.Range() + if (err == nil) != tc.valid || (err == nil && (from.Location().String() != "UTC" || to.Before(from))) { + t.Fatalf("%+v: from=%v to=%v error=%v", tc, from, to, err) + } + } +} + +func TestMergeWarmupStatsPreservesActiveDaysAcrossTransport(t *testing.T) { + raw, err := json.Marshal([]WarmupDailyStats{{Date: "2026-10-01", EmailsSent: 9, TargetVolume: 10, Active: true}}) + if err != nil { + t.Fatal(err) + } + var cloud []WarmupDailyStats + if err := json.Unmarshal(raw, &cloud); err != nil { + t.Fatal(err) + } + out := MergeWarmupStats([]WarmupDailyStats{ + {Date: "2026-10-02", EmailsReceived: 2}, + {Date: "2026-10-01", EmailsSent: 1, EmailsReplied: 1, TargetVolume: 2, Active: true}, + }, cloud) + if len(out) != 2 || out[0].Date != "2026-10-01" || out[0].EmailsSent != 10 || out[0].TargetVolume != 12 || !out[0].Active || out[1].Active { + t.Fatalf("merged=%+v", out) + } +} diff --git a/internal/repository/cloudlink_reports.go b/internal/repository/cloudlink_reports.go new file mode 100644 index 000000000..e94fe19a7 --- /dev/null +++ b/internal/repository/cloudlink_reports.go @@ -0,0 +1,29 @@ +package repository + +import ( + "context" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/models" +) + +func (r *cloudLinkRepository) ListForOrg(ctx context.Context, orgID uuid.UUID, accountID *uuid.UUID) ([]models.CloudLinkMailbox, error) { + query := `SELECT ` + cloudLinkMailboxColumns + ` FROM cloud_link_mailboxes + WHERE email_account_id IN ( + SELECT id FROM email_accounts WHERE organization_id = $1 AND ($2::uuid IS NULL OR id = $2) + ) ORDER BY enrolled_at` + rows, err := r.db.Query(ctx, query, orgID, accountID) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]models.CloudLinkMailbox, 0) + for rows.Next() { + m, err := scanCloudLinkMailbox(rows) + if err != nil { + return nil, err + } + out = append(out, *m) + } + return out, rows.Err() +} diff --git a/internal/repository/pg_analytics.go b/internal/repository/pg_analytics.go index 6f4e07cd2..197aff664 100644 --- a/internal/repository/pg_analytics.go +++ b/internal/repository/pg_analytics.go @@ -16,6 +16,7 @@ import ( type AnalyticsRepository interface { // Warmup analytics GetWarmupStats(ctx context.Context, orgID uuid.UUID, emailAccountID *uuid.UUID, from, to time.Time) ([]models.WarmupDailyStats, *errx.Error) + GetWarmupStatsForAccounts(ctx context.Context, orgID uuid.UUID, emailIDs []uuid.UUID, from, to time.Time) ([]models.WarmupDailyStats, *errx.Error) // Campaign analytics. A nil period is the campaign's whole history; a // period scopes the performance figures to the sends inside it. @@ -67,6 +68,10 @@ func NewAnalyticsRepository(db *db.DB) AnalyticsRepository { } func (r *analyticsRepository) GetWarmupStats(ctx context.Context, orgID uuid.UUID, emailAccountID *uuid.UUID, from, to time.Time) ([]models.WarmupDailyStats, *errx.Error) { + return r.GetWarmupStatsForAccounts(ctx, orgID, singleWarmupAccount(emailAccountID), from, to) +} + +func (r *analyticsRepository) GetWarmupStatsForAccounts(ctx context.Context, orgID uuid.UUID, emailIDs []uuid.UUID, from, to time.Time) ([]models.WarmupDailyStats, *errx.Error) { // Sends come from the daily plan rows, arrivals from the verified receipts, // joined both ways so a day the mailbox was written to but did not send // still shows what came in. Receipts are bucketed on the UTC day, which is @@ -84,7 +89,7 @@ func (r *analyticsRepository) GetWarmupStats(ctx context.Context, orgID uuid.UUI WHERE ea.organization_id = $1 AND ws.date >= $2 AND ws.date <= $3 - AND ($4::uuid IS NULL OR ws.email_account_id = $4) + AND ($4::uuid[] IS NULL OR ws.email_account_id = ANY($4)) GROUP BY ws.date ), received AS ( @@ -96,7 +101,7 @@ func (r *analyticsRepository) GetWarmupStats(ctx context.Context, orgID uuid.UUI WHERE ea.organization_id = $1 AND wr.created_at >= $2::timestamptz AND wr.created_at < ($3::timestamptz + interval '1 day') - AND ($4::uuid IS NULL OR wr.email_account_id = $4) + AND ($4::uuid[] IS NULL OR wr.email_account_id = ANY($4)) GROUP BY 1 ) SELECT @@ -111,7 +116,7 @@ func (r *analyticsRepository) GetWarmupStats(ctx context.Context, orgID uuid.UUI ORDER BY 1 ASC ` - params := []any{orgID, from, to, emailAccountID} + params := []any{orgID, from, to, emailIDs} rows, err := r.DB.Query(ctx, query, params...) if err != nil { diff --git a/internal/repository/pg_cloudlink.go b/internal/repository/pg_cloudlink.go index 6950d4436..4b02445d7 100644 --- a/internal/repository/pg_cloudlink.go +++ b/internal/repository/pg_cloudlink.go @@ -29,6 +29,7 @@ type CloudLinkRepository interface { UnenrollAll(ctx context.Context) error GetByAccount(ctx context.Context, accountID uuid.UUID) (*models.CloudLinkMailbox, error) List(ctx context.Context) ([]models.CloudLinkMailbox, error) + ListForOrg(ctx context.Context, orgID uuid.UUID, accountID *uuid.UUID) ([]models.CloudLinkMailbox, error) // IsEnrolled is the hot-path check the warmup task and reconciler use. IsEnrolled(ctx context.Context, accountID uuid.UUID) (bool, error) diff --git a/internal/repository/pg_poollink.go b/internal/repository/pg_poollink.go index b6bac7956..e3804889d 100644 --- a/internal/repository/pg_poollink.go +++ b/internal/repository/pg_poollink.go @@ -36,6 +36,7 @@ type PoolLinkRepository interface { GetMailboxByRemote(ctx context.Context, instanceID, remoteID uuid.UUID) (*models.PoolLinkMailbox, error) GetMailboxByAccount(ctx context.Context, accountID uuid.UUID) (*models.PoolLinkMailbox, error) ListMailboxes(ctx context.Context, instanceID uuid.UUID) ([]models.PoolLinkMailbox, error) + ListReportMailboxes(ctx context.Context, instanceID uuid.UUID, remoteIDs []uuid.UUID) ([]models.PoolLinkMailbox, error) // ListStanding is every enrolled mailbox's warmup standing in one read. ListStanding(ctx context.Context, instanceID uuid.UUID) ([]models.PoolLinkMailboxStanding, error) DeleteMailbox(ctx context.Context, instanceID, remoteID uuid.UUID) error diff --git a/internal/repository/pg_warmup_placement.go b/internal/repository/pg_warmup_placement.go index c4719192c..67b38c9c4 100644 --- a/internal/repository/pg_warmup_placement.go +++ b/internal/repository/pg_warmup_placement.go @@ -12,34 +12,13 @@ import ( ) // WarmupPlacementDayRow is one sender's placement at one recipient group on one UTC day. -type WarmupPlacementDayRow struct { - SenderID uuid.UUID - Email string - Date string - Group string - Inbox int - Tabs int - Spam int - Rescued int -} +type WarmupPlacementDayRow = models.WarmupPlacementDayRow // WarmupPlacementHostRow is the window's placement at one mail host. -type WarmupPlacementHostRow struct { - Group string - Host string - Inbox int - Tabs int - Spam int - Rescued int -} +type WarmupPlacementHostRow = models.WarmupPlacementHostRow // WarmupSenderDayCount is a per-sender, per-UTC-day count. -type WarmupSenderDayCount struct { - SenderID uuid.UUID - Email string - Date string - Count int -} +type WarmupSenderDayCount = models.WarmupSenderDayCount // WarmupPlacementSender is the sending mailbox a counted placement belongs to. type WarmupPlacementSender struct { @@ -67,6 +46,7 @@ type WarmupPlacementRepository interface { Unconfirmed(ctx context.Context, orgID uuid.UUID, senderID *uuid.UUID, from, to, cutoff time.Time) ([]WarmupSenderDayCount, error) // Rates is each sender's trailing-window placement since the given day. Rates(ctx context.Context, orgID uuid.UUID, senderID *uuid.UUID, since time.Time) (map[uuid.UUID]models.WarmupPlacementWindow, error) + ForAccounts(ctx context.Context, orgID uuid.UUID, senderIDs []uuid.UUID, from, to time.Time) (*models.WarmupPlacementData, error) } type warmupPlacementRepository struct { @@ -172,6 +152,10 @@ func (r *warmupPlacementRepository) SweepUnplaced(ctx context.Context, cutoff ti } func (r *warmupPlacementRepository) Daily(ctx context.Context, orgID uuid.UUID, senderID *uuid.UUID, from, to time.Time) ([]WarmupPlacementDayRow, error) { + return r.daily(ctx, orgID, singleWarmupAccount(senderID), from, to) +} + +func (r *warmupPlacementRepository) daily(ctx context.Context, orgID uuid.UUID, senderIDs []uuid.UUID, from, to time.Time) ([]WarmupPlacementDayRow, error) { const query = ` SELECT p.sender_account_id, ea.email, p.date::text, p.recipient_group, SUM(p.inbox)::int, SUM(p.tabs)::int, SUM(p.spam)::int, SUM(p.rescued)::int @@ -179,10 +163,10 @@ func (r *warmupPlacementRepository) Daily(ctx context.Context, orgID uuid.UUID, JOIN email_accounts ea ON ea.id = p.sender_account_id WHERE ea.organization_id = $1 AND p.date >= $2 AND p.date <= $3 - AND ($4::uuid IS NULL OR p.sender_account_id = $4) + AND ($4::uuid[] IS NULL OR p.sender_account_id = ANY($4)) GROUP BY 1, 2, 3, 4 ` - params := []any{orgID, from, to, senderID} + params := []any{orgID, from, to, senderIDs} rows, err := r.db.Query(ctx, query, params...) if err != nil { db.CaptureError(err, query, params, "query") @@ -201,6 +185,10 @@ func (r *warmupPlacementRepository) Daily(ctx context.Context, orgID uuid.UUID, } func (r *warmupPlacementRepository) Hosts(ctx context.Context, orgID uuid.UUID, senderID *uuid.UUID, from, to time.Time) ([]WarmupPlacementHostRow, error) { + return r.hosts(ctx, orgID, singleWarmupAccount(senderID), from, to) +} + +func (r *warmupPlacementRepository) hosts(ctx context.Context, orgID uuid.UUID, senderIDs []uuid.UUID, from, to time.Time) ([]WarmupPlacementHostRow, error) { const query = ` SELECT p.recipient_group, p.recipient_host, SUM(p.inbox)::int, SUM(p.tabs)::int, SUM(p.spam)::int, SUM(p.rescued)::int @@ -208,10 +196,10 @@ func (r *warmupPlacementRepository) Hosts(ctx context.Context, orgID uuid.UUID, JOIN email_accounts ea ON ea.id = p.sender_account_id WHERE ea.organization_id = $1 AND p.date >= $2 AND p.date <= $3 - AND ($4::uuid IS NULL OR p.sender_account_id = $4) + AND ($4::uuid[] IS NULL OR p.sender_account_id = ANY($4)) GROUP BY 1, 2 ` - params := []any{orgID, from, to, senderID} + params := []any{orgID, from, to, senderIDs} rows, err := r.db.Query(ctx, query, params...) if err != nil { db.CaptureError(err, query, params, "query") @@ -230,19 +218,27 @@ func (r *warmupPlacementRepository) Hosts(ctx context.Context, orgID uuid.UUID, } func (r *warmupPlacementRepository) Sent(ctx context.Context, orgID uuid.UUID, senderID *uuid.UUID, from, to time.Time) ([]WarmupSenderDayCount, error) { + return r.sent(ctx, orgID, singleWarmupAccount(senderID), from, to) +} + +func (r *warmupPlacementRepository) sent(ctx context.Context, orgID uuid.UUID, senderIDs []uuid.UUID, from, to time.Time) ([]WarmupSenderDayCount, error) { const query = ` SELECT ws.email_account_id, ea.email, ws.date::text, SUM(ws.emails_sent)::int FROM warmup_statistics ws JOIN email_accounts ea ON ea.id = ws.email_account_id WHERE ea.organization_id = $1 AND ws.date >= $2 AND ws.date <= $3 - AND ($4::uuid IS NULL OR ws.email_account_id = $4) + AND ($4::uuid[] IS NULL OR ws.email_account_id = ANY($4)) GROUP BY 1, 2, 3 ` - return r.senderDayCounts(ctx, query, []any{orgID, from, to, senderID}) + return r.senderDayCounts(ctx, query, []any{orgID, from, to, senderIDs}) } func (r *warmupPlacementRepository) Unconfirmed(ctx context.Context, orgID uuid.UUID, senderID *uuid.UUID, from, to, cutoff time.Time) ([]WarmupSenderDayCount, error) { + return r.unconfirmed(ctx, orgID, singleWarmupAccount(senderID), from, to, cutoff) +} + +func (r *warmupPlacementRepository) unconfirmed(ctx context.Context, orgID uuid.UUID, senderIDs []uuid.UUID, from, to, cutoff time.Time) ([]WarmupSenderDayCount, error) { // A token is written before the send, so only completed sends count. const query = ` SELECT wt.sender_account_id, ea.email, (wt.created_at AT TIME ZONE 'UTC')::date::text, COUNT(*)::int @@ -254,10 +250,10 @@ func (r *warmupPlacementRepository) Unconfirmed(ctx context.Context, orgID uuid. AND wt.created_at < ($3::timestamptz + interval '1 day') AND wt.created_at < $5 AND wt.consumed_at IS NULL - AND ($4::uuid IS NULL OR wt.sender_account_id = $4) + AND ($4::uuid[] IS NULL OR wt.sender_account_id = ANY($4)) GROUP BY 1, 2, 3 ` - return r.senderDayCounts(ctx, query, []any{orgID, from, to, senderID, cutoff}) + return r.senderDayCounts(ctx, query, []any{orgID, from, to, senderIDs, cutoff}) } func (r *warmupPlacementRepository) senderDayCounts(ctx context.Context, query string, params []any) ([]WarmupSenderDayCount, error) { @@ -279,6 +275,10 @@ func (r *warmupPlacementRepository) senderDayCounts(ctx context.Context, query s } func (r *warmupPlacementRepository) Rates(ctx context.Context, orgID uuid.UUID, senderID *uuid.UUID, since time.Time) (map[uuid.UUID]models.WarmupPlacementWindow, error) { + return r.rates(ctx, orgID, singleWarmupAccount(senderID), since) +} + +func (r *warmupPlacementRepository) rates(ctx context.Context, orgID uuid.UUID, senderIDs []uuid.UUID, since time.Time) (map[uuid.UUID]models.WarmupPlacementWindow, error) { const query = ` SELECT p.sender_account_id, SUM(p.inbox) FILTER (WHERE p.recipient_group <> 'other')::int, @@ -289,10 +289,10 @@ func (r *warmupPlacementRepository) Rates(ctx context.Context, orgID uuid.UUID, JOIN email_accounts ea ON ea.id = p.sender_account_id WHERE ea.organization_id = $1 AND p.date >= $2 - AND ($3::uuid IS NULL OR p.sender_account_id = $3) + AND ($3::uuid[] IS NULL OR p.sender_account_id = ANY($3)) GROUP BY 1 ` - params := []any{orgID, since, senderID} + params := []any{orgID, since, senderIDs} rows, err := r.db.Query(ctx, query, params...) if err != nil { db.CaptureError(err, query, params, "query") diff --git a/internal/repository/poollink_reports.go b/internal/repository/poollink_reports.go new file mode 100644 index 000000000..c239d0a60 --- /dev/null +++ b/internal/repository/poollink_reports.go @@ -0,0 +1,27 @@ +package repository + +import ( + "context" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/models" +) + +func (r *poolLinkRepository) ListReportMailboxes(ctx context.Context, instanceID uuid.UUID, remoteIDs []uuid.UUID) ([]models.PoolLinkMailbox, error) { + const query = `SELECT instance_id, remote_id, email_account_id, enrolled_at, managed, last_token_at + FROM pool_link_mailboxes WHERE instance_id = $1 AND remote_id = ANY($2::uuid[])` + rows, err := r.db.Query(ctx, query, instanceID, remoteIDs) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]models.PoolLinkMailbox, 0, len(remoteIDs)) + for rows.Next() { + m, err := scanPoolLinkMailbox(rows) + if err != nil { + return nil, err + } + out = append(out, *m) + } + return out, rows.Err() +} diff --git a/internal/repository/warmup_analytics_scope_live_test.go b/internal/repository/warmup_analytics_scope_live_test.go index 601beab0c..0942038ae 100644 --- a/internal/repository/warmup_analytics_scope_live_test.go +++ b/internal/repository/warmup_analytics_scope_live_test.go @@ -60,6 +60,51 @@ func TestLiveWarmupAnalyticsAreOrganizationScoped(t *testing.T) { }) repo := &analyticsRepository{DB: handle} + instanceID, otherInstanceID, remoteID, otherRemoteID := uuid.New(), uuid.New(), uuid.New(), uuid.New() + exec(`INSERT INTO pool_link_instances (id, organization_id, name, token_hash) VALUES ($1, $3, 'Reports', $4), ($2, $3, 'Other', $5)`, + instanceID, otherInstanceID, orgID, uuid.NewString(), uuid.NewString()) + exec(`INSERT INTO pool_link_mailboxes (instance_id, remote_id, email_account_id) VALUES ($1, $3, $5), ($2, $4, $6)`, + instanceID, otherInstanceID, remoteID, otherRemoteID, accountID, secondAccountID) + exec(`INSERT INTO cloud_link_mailboxes (email_account_id, remote_id, managed) VALUES ($1, $3, true), ($2, $4, false)`, + accountID, secondAccountID, remoteID, otherRemoteID) + t.Cleanup(func() { + if _, err := pool.Exec(context.Background(), `DELETE FROM pool_link_instances WHERE id = ANY($1::uuid[])`, []uuid.UUID{instanceID, otherInstanceID}); err != nil { + t.Errorf("instance cleanup: %v", err) + } + }) + poolRepo := NewPoolLinkRepository(pool) + linked, err := poolRepo.ListReportMailboxes(ctx, instanceID, []uuid.UUID{remoteID, otherRemoteID}) + if err != nil || len(linked) != 1 || linked[0].EmailAccountID != accountID { + t.Fatalf("instance-scoped reports = %+v, %v", linked, err) + } + cloudRepo := &cloudLinkRepository{db: pool} + local, err := cloudRepo.ListForOrg(ctx, orgID, nil) + if err != nil || len(local) != 2 { + t.Fatalf("existing enrollments = %+v, %v", local, err) + } + local, err = cloudRepo.ListForOrg(ctx, orgID, &secondAccountID) + if err != nil || len(local) != 1 || local[0].EmailAccountID != secondAccountID { + t.Fatalf("selected enrollment = %+v, %v", local, err) + } + local, err = cloudRepo.ListForOrg(ctx, uuid.New(), &accountID) + if err != nil || len(local) != 0 { + t.Fatalf("foreign enrollment = %+v, %v", local, err) + } + selected, xerr := repo.GetWarmupStatsForAccounts(ctx, orgID, []uuid.UUID{secondAccountID, secondAccountID}, day, day) + if xerr != nil || len(selected) != 1 || selected[0].EmailsSent != 5 || selected[0].EmailsReceived != 0 { + t.Fatalf("selected accounts = %+v, %v", selected, xerr) + } + for _, tc := range []struct { + org uuid.UUID + ids []uuid.UUID + }{ + {orgID, []uuid.UUID{}}, {uuid.New(), []uuid.UUID{accountID}}, {orgID, []uuid.UUID{uuid.New()}}, + } { + selected, xerr := repo.GetWarmupStatsForAccounts(ctx, tc.org, tc.ids, day, day) + if xerr != nil || len(selected) != 0 { + t.Fatalf("unselected accounts = %+v, %v", selected, xerr) + } + } stats, xerr := repo.GetWarmupStats(ctx, orgID, &accountID, day, day) if xerr != nil { t.Fatalf("GetWarmupStats: %v", xerr) diff --git a/internal/repository/warmup_placement_live_test.go b/internal/repository/warmup_placement_live_test.go index 4dec370b4..2a042efad 100644 --- a/internal/repository/warmup_placement_live_test.go +++ b/internal/repository/warmup_placement_live_test.go @@ -164,6 +164,19 @@ func TestLiveWarmupPlacementRollup(t *testing.T) { if err != nil || len(sent) != 1 || sent[0].Count != 4 { t.Fatalf("sent = %+v, %v; want 4 today", sent, err) } + data, err := repo.ForAccounts(ctx, orgID, []uuid.UUID{sender, stranger}, today.AddDate(0, 0, -3), today) + if err != nil || len(data.Daily) != 2 || len(data.Hosts) != 2 || len(data.Sent) != 1 || len(data.Unconfirmed) != 1 || data.Windows[sender].All.Inbox != 2 { + t.Fatalf("scoped report = %+v, %v", data, err) + } + for _, tc := range []struct { + org uuid.UUID + ids []uuid.UUID + }{{orgID, []uuid.UUID{}}, {otherOrg, []uuid.UUID{sender}}, {orgID, []uuid.UUID{recipient}}} { + data, err := repo.ForAccounts(ctx, tc.org, tc.ids, today.AddDate(0, 0, -3), today) + if err != nil || len(data.Daily)+len(data.Hosts)+len(data.Sent)+len(data.Unconfirmed)+len(data.Windows) != 0 { + t.Fatalf("unselected placement report = %+v, %v", data, err) + } + } } // Receipts no live arrival counted (history, or a consumer mid-upgrade) are diff --git a/internal/repository/warmup_report.go b/internal/repository/warmup_report.go new file mode 100644 index 000000000..0168d7def --- /dev/null +++ b/internal/repository/warmup_report.go @@ -0,0 +1,40 @@ +package repository + +import ( + "context" + "time" + + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/models" +) + +func singleWarmupAccount(id *uuid.UUID) []uuid.UUID { + if id == nil { + return nil + } + return []uuid.UUID{*id} +} + +func (r *warmupPlacementRepository) ForAccounts(ctx context.Context, orgID uuid.UUID, ids []uuid.UUID, from, to time.Time) (*models.WarmupPlacementData, error) { + d := &models.WarmupPlacementData{} + var err error + lookback := from.AddDate(0, 0, -(models.WarmupPlacementWindowDays - 1)) + if d.Daily, err = r.daily(ctx, orgID, ids, lookback, to); err != nil { + return nil, err + } + if d.Hosts, err = r.hosts(ctx, orgID, ids, from, to); err != nil { + return nil, err + } + if d.Sent, err = r.sent(ctx, orgID, ids, from, to); err != nil { + return nil, err + } + if d.Unconfirmed, err = r.unconfirmed(ctx, orgID, ids, from, to, time.Now().Add(-models.WarmupUnconfirmedAfterHours*time.Hour)); err != nil { + return nil, err + } + since := time.Now().UTC().AddDate(0, 0, -(models.WarmupPlacementWindowDays - 1)) + since = time.Date(since.Year(), since.Month(), since.Day(), 0, 0, 0, 0, time.UTC) + if d.Windows, err = r.rates(ctx, orgID, ids, since); err != nil { + return nil, err + } + return d, nil +} diff --git a/web/src/components/app/emails/InboxDetails.tsx b/web/src/components/app/emails/InboxDetails.tsx index fa93a67aa..aeb0ea044 100644 --- a/web/src/components/app/emails/InboxDetails.tsx +++ b/web/src/components/app/emails/InboxDetails.tsx @@ -520,7 +520,7 @@ function Detail({ mailbox, onClose, initialTab = "overview", canWarmup = true }:
{tab === "overview" && } {tab === "deliverability" && } - {tab === "analytics" && } + {tab === "analytics" && void warmup.refetch()} />} {tab === "warmup" && } {tab === "sending" && } {tab === "settings" && } @@ -854,12 +854,23 @@ function OverviewTab({ status, loading, mailbox }: { status?: import("@/lib/api/ /* ── Analytics ─────────────────────── */ -function AnalyticsTab({ warmup, loading }: { warmup?: import("@/lib/api/models/app/analytics/WarmupAnalytics").default; loading: boolean }) { +function AnalyticsTab({ warmup, loading, failed, retry }: { warmup?: import("@/lib/api/models/app/analytics/WarmupAnalytics").default; loading: boolean; failed: boolean; retry: () => void }) { // Tap affordance for touch devices, where the native title tooltip never fires. const [selectedDay, setSelectedDay] = useState(null); if (loading) { return
; } + if (failed) { + return ( +
+

Couldn't load warmup analytics

+

If this mailbox warms in Cloud, check the Cloud connection and try again.

+ +
+ ); + } if (!warmup || warmup.daily_stats.length === 0) { return (