starred/warmbly
1
0
Fork 0
mirror of https://github.com/warmbly/warmbly.git synced 2026-10-31 08:13:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: cut database load from issue 993 with placement, sweep and unibox indexes, redundant index drops, identical-result rewrites of contact facets, dashboard stats, campaign progress, advisor bounces and routing pre-filter, bounded delivery evidence, cold-ramp skip, held advisor sections, ANALYZE after restore and hidden-tab realtime deferral

Browse Source
This commit is contained in:
Matthew Meszaros
2026-10-17 05:18:45 +02:00
parent 0911c5390b
commit ae9eca5c7e
52 changed files with 1102 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/backend/main.go
+16 -1
View File
@@ -2001,7 +2001,8 @@ func main() {
// snapshot instead of running the planner (lead supply, per-mailbox
// history) on the request. A no-op without the snapshot store or planner.
if campaignService != nil {
go campaignService.StartSendPlanSnapshotter(ctx, time.Minute)
interval, maxAge := sendPlanSnapshotTiming()
go campaignService.StartSendPlanSnapshotter(ctx, interval, maxAge)
}
// Segment-linked campaigns: enrol contacts that drifted into a linked
@@ -2674,6 +2675,20 @@ func localTasksPollInterval() time.Duration {
return d
}
// sendPlanSnapshotTiming reads SEND_PLAN_SNAPSHOT_INTERVAL (default 1m) and
// SEND_PLAN_SNAPSHOT_MAX_AGE (default three intervals), both Go durations.
func sendPlanSnapshotTiming() (time.Duration, time.Duration) {
interval, err := time.ParseDuration(getenvDefault("SEND_PLAN_SNAPSHOT_INTERVAL", "1m"))
if err != nil || interval <= 0 {
interval = time.Minute
}
maxAge, err := time.ParseDuration(os.Getenv("SEND_PLAN_SNAPSHOT_MAX_AGE"))
if err != nil || maxAge <= 0 {
maxAge = 3 * interval
}
return interval, maxAge
}
// wsHealthURL derives the realtime service's HTTP /health URL from the
// public websocket URI (ws[s]://host[:port]/... -> http[s]://host[:port]/health).
func wsHealthURL(wsURI string) string {
cmd/warmblyctl/backup.go
+19
View File
@@ -305,6 +305,11 @@ func runRestore(ctx context.Context, args []string) error {
return err
}
fmt.Println(" database restored")
if err := analyzeDatabase(ctx, dsn); err != nil {
warn("the database restored, but ANALYZE did not finish, so queries plan without statistics until autovacuum gathers them. Run ANALYZE by hand:\n%s", err)
} else {
fmt.Println(" statistics analyzed")
}
if !*noBlobs && m.BlobFiles > 0 {
root := blobRoot()
@@ -491,6 +496,20 @@ func restoreDatabase(ctx context.Context, dsn, bundle string) error {
return nil
}
// analyzeDatabase gathers planner statistics, which a dump does not carry.
func analyzeDatabase(ctx context.Context, dsn string) error {
cmd := pgCommand(ctx, "psql", dsn, "-v", "ON_ERROR_STOP=1", "-q", "-c", "ANALYZE;")
var stderr strings.Builder
cmd.Stderr = &stderr
if err := cmd.Run(); err != nil {
if msg := strings.TrimSpace(stderr.String()); msg != "" {
return errors.New(msg)
}
return err
}
return nil
}
// pgCommand builds a libpq invocation with the password moved out of argv.
//
// A connection string on the command line is readable by anyone on the host who
docs/content/docs/development/configuration.mdx
+2
View File
@@ -687,6 +687,8 @@ Per-mailbox `test_mode`, `test_send_enabled` and `test_receive_enabled` separate
|---|---|---|
| `TASKS_PROVIDER` | `local` (an in-process Postgres poller) or `gcloud` (Cloud Tasks) | `local` |
| `TASKS_LOCAL_POLL_INTERVAL` | How often the local poller looks for due work | `1s` |
| `SEND_PLAN_SNAPSHOT_INTERVAL` | How often the backend re-walks the send plan of every active campaign whose stored plan is out of date. Each walk reads all of that campaign's leads, so a longer interval lowers database load on large campaigns | `1m` |
| `SEND_PLAN_SNAPSHOT_MAX_AGE` | How old a stored send plan may be before it counts as out of date. An out-of-date plan is still served, with `stale: true`, while a fresh one is computed in the background | three times `SEND_PLAN_SNAPSHOT_INTERVAL` (`3m`) |
| `BILLING_PROVIDER` | `none` (every feature unlocked, no trial expiry; the dashboard reports the workspace as self-hosted rather than on a free tier and hides billing) or `stripe` | `none` |
| `STRIPE_SECRET_KEY`, `STRIPE_WEBHOOK_SECRET`, `STRIPE_PUBLISHABLE_KEY` | Required together when `BILLING_PROVIDER=stripe`. The backend exits at boot if any is missing | unset |
docs/content/docs/development/data-control.mdx
+1 -1
View File
@@ -222,7 +222,7 @@ docker compose -p warmbly exec backend warmblyctl restore --file /data/blobs/war
docker compose -p warmbly restart
```
The restore empties the schema and replays the dump, so it replaces everything currently on that instance and asks you to type `restore` first.
The restore empties the schema and replays the dump, so it replaces everything currently on that instance and asks you to type `restore` first. It then runs `ANALYZE`, since a dump carries no planner statistics; a failure there only warns.
Before it does anything it compares the bundle's `CREDENTIALS_ENCRYPTION_KEY` and `KMS_LOCAL_MASTER_KEY` against the destination's and refuses to continue when they differ, printing the two lines to put in `.env`. That check is the point of the command: without it a restore looks like it worked and every mailbox fails to authenticate days later, with no error that names the cause.
docs/content/docs/development/warmblyctl.mdx
+2
View File
@@ -499,6 +499,8 @@ Before anything is written it compares the bundle's `CREDENTIALS_ENCRYPTION_KEY`
Blobs are unpacked over the blob root additively: files the bundle knows about are overwritten, anything else is left alone.
Once the dump has loaded, the restore runs `ANALYZE`, because a dump carries no planner statistics. If that step fails the restore still completes and prints a warning; run `ANALYZE` in `psql` yourself afterwards.
[Data control](/development/data-control/#moving-an-instance) walks the whole move to a new host.
## org list
internal/app/advisor/advisor.go
+39 -9
View File
@@ -125,9 +125,21 @@ type Detector struct {
// it matters. It grounds the narrator so the copy explains the real rule
// rather than paraphrasing the numbers back at the user.
About string
// Reads lists the snapshot sections it depends on beyond the mailboxes.
Reads []repository.AdvisorSection
Run func(s *repository.AdvisorSnapshot) []Finding
}
// held reports whether a section this detector reads failed to load.
func (d Detector) held(s *repository.AdvisorSnapshot) bool {
for _, sec := range d.Reads {
if s.SectionFailed(sec) {
return true
}
}
return false
}
// AllDetectors returns every detector in evaluation order. Order does not
// affect correctness (findings are sorted by severity for display) but keeping
// the categories grouped makes the run log readable.
@@ -154,18 +166,11 @@ func DetectorAbout() map[string]string {
// Detect runs every detector that the org's settings leave enabled and returns
// the findings sorted most-urgent-first.
func Detect(s *repository.AdvisorSnapshot, settings *models.AdvisorSettings) []Finding {
muted := map[string]bool{}
for _, k := range settings.MutedDetectors {
muted[k] = true
}
mutedCat := map[string]bool{}
for _, c := range settings.MutedCategories {
mutedCat[c] = true
}
muted := mutedBy(settings)
out := []Finding{}
for _, d := range AllDetectors() {
if muted[d.Key] || mutedCat[string(d.Category)] {
if muted(d) || d.held(s) {
continue
}
for _, f := range d.Run(s) {
@@ -194,6 +199,31 @@ func Detect(s *repository.AdvisorSnapshot, settings *models.AdvisorSettings) []F
return out
}
// heldDetectors is the enabled detectors whose open findings stand unchanged
// this run, because a section they read did not load.
func heldDetectors(s *repository.AdvisorSnapshot, settings *models.AdvisorSettings) []string {
muted := mutedBy(settings)
out := []string{}
for _, d := range AllDetectors() {
if !muted(d) && d.held(s) {
out = append(out, d.Key)
}
}
return out
}
func mutedBy(settings *models.AdvisorSettings) func(Detector) bool {
muted := map[string]bool{}
for _, k := range settings.MutedDetectors {
muted[k] = true
}
mutedCat := map[string]bool{}
for _, c := range settings.MutedCategories {
mutedCat[c] = true
}
return func(d Detector) bool { return muted[d.Key] || mutedCat[string(d.Category)] }
}
// --- shared helpers -------------------------------------------------------
// rate returns n/total as a percentage, or 0 when there is nothing to divide by.
internal/app/advisor/detect_campaign.go
+11
View File
@@ -16,66 +16,77 @@ func campaignDetectors() []Detector {
Key: "campaign_no_followups",
Category: models.AdvisorCategoryCampaign,
About: "A cold campaign with a single email and no follow-up steps. Most replies to cold outreach arrive on a follow-up rather than the first send, so a one-step campaign gives up most of its own results.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns},
Run: detectNoFollowUps,
},
{
Key: "campaign_reply_rate_low",
Category: models.AdvisorCategoryCampaign,
About: "A campaign's reply rate at volume. A very low reply rate on a large sample means the offer, the audience, or the copy is wrong, and sending more of it will cost sender reputation without producing results.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns},
Run: detectReplyRateLow,
},
{
Key: "campaign_step_dropoff",
Category: models.AdvisorCategoryCampaign,
About: "One sequence step performing far worse than the campaign's best step. Because every step reaches the same audience, a step that underperforms the others by a wide margin is a copy problem isolated to that message.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns, repository.AdvisorSectionSteps},
Run: detectStepDropoff,
},
{
Key: "campaign_unsubscribe_header_off",
Category: models.AdvisorCategoryCampaign,
About: "Whether a campaign sends the List-Unsubscribe header. Google's bulk sender rules require one-click unsubscribe on marketing mail, and an easy unsubscribe is what stops an uninterested recipient reaching for the spam button instead.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns},
Run: detectUnsubscribeHeaderOff,
},
{
Key: "campaign_capacity_shortfall",
Category: models.AdvisorCategoryCampaign,
About: "A campaign whose daily limit exceeds what its attached mailboxes can safely send. The campaign will silently deliver less than configured, or push mailboxes past their caps.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns},
Run: detectCapacityShortfall,
},
{
Key: "campaign_no_senders",
Category: models.AdvisorCategoryCampaign,
About: "A running campaign with no active mailbox resolving for it. Its sender_selection says how it picks mailboxes: by tag (no active mailbox carries its tags), picked by hand (those mailboxes were disconnected), or every active mailbox (the workspace has none active). The campaign is live and sending nothing.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns},
Run: detectNoSenders,
},
{
Key: "campaign_followup_spacing",
Category: models.AdvisorCategoryCampaign,
About: "The gap between follow-up steps. Too close reads as pestering and drives unsubscribes and complaints; too far apart and the recipient no longer remembers the first email.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns, repository.AdvisorSectionSteps},
Run: detectFollowUpSpacing,
},
{
Key: "campaign_list_exhaustion",
Category: models.AdvisorCategoryCampaign,
About: "A running campaign about to run out of leads. Campaigns do not announce that they have finished their list; they just stop producing results.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns},
Run: detectListExhaustion,
},
{
Key: "campaign_no_ab_test",
Category: models.AdvisorCategoryCampaign,
About: "A high-volume campaign running a single variant. At this volume an A/B test costs nothing and is the only way to know whether the copy is the constraint.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns},
Run: detectNoABTest,
},
{
Key: "campaign_narrow_window",
Category: models.AdvisorCategoryCampaign,
About: "A sending window too short to fit the campaign's daily volume at safe pacing. The campaign either underdelivers or compresses its sends into a burst.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns},
Run: detectNarrowWindow,
},
{
Key: "campaign_duplicate_enrollment",
Category: models.AdvisorCategoryCampaign,
About: "Contacts enrolled in more than one running campaign at the same time. To the recipient this is two unrelated strangers pitching in the same week, which is how a sender earns a complaint rather than a reply.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionOrg},
Run: detectDuplicateEnrollment,
},
}
internal/app/advisor/detect_copy.go
+8
View File
@@ -23,48 +23,56 @@ func copyDetectors() []Detector {
Key: "copy_broken_template",
Category: models.AdvisorCategoryCopy,
About: "Copy that the template engine cannot parse. On send it silently degrades to the naive replacement path, so conditionals and merge variables ship to the recipient as literal text. This is the single most visible mistake in cold outreach and instantly marks a message as bulk.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns, repository.AdvisorSectionSteps},
Run: detectBrokenTemplate,
},
{
Key: "copy_spam_phrases",
Category: models.AdvisorCategoryCopy,
About: "Phrases that both spam filters and human readers treat as bulk-mail markers. The filter cost is real but secondary; the main cost is that these phrases make a message read as something nobody wrote to anyone in particular.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns, repository.AdvisorSectionSteps},
Run: detectSpamPhrases,
},
{
Key: "copy_too_long",
Category: models.AdvisorCategoryCopy,
About: "Cold emails long enough that a busy recipient will not read them. Length is the most reliable predictor of a cold email being ignored on a phone, which is where most of them are opened.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns, repository.AdvisorSectionSteps},
Run: detectCopyTooLong,
},
{
Key: "copy_subject_too_long",
Category: models.AdvisorCategoryCopy,
About: "Subject lines long enough to be truncated in the inbox list, especially on mobile clients, so the part that would earn the open is never seen.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns, repository.AdvisorSectionSteps},
Run: detectSubjectTooLong,
},
{
Key: "copy_too_many_links",
Category: models.AdvisorCategoryCopy,
About: "Link count in a cold email. Multiple links in a first-touch email is a strong bulk-mail signal to filters, and it splits the reader's attention away from the single thing you want them to do.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns, repository.AdvisorSectionSteps},
Run: detectTooManyLinks,
},
{
Key: "copy_shouty_subject",
Category: models.AdvisorCategoryCopy,
About: "Subject lines in capitals or stacked with exclamation marks. This is a bulk-mail signal for filters and reads as shouting to a person.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns, repository.AdvisorSectionSteps},
Run: detectShoutySubject,
},
{
Key: "copy_reads_as_bulk",
Category: models.AdvisorCategoryCopy,
About: "Copy that a calibrated reader model places at the bulk-mail end of a personal-to-bulk scale, or that makes a claim a spam filter objects to (guaranteed results, free money, prizes, pressure to act now). Unlike the phrase list this judges the whole email as its recipient would, so it catches copy that avoids every trigger word and still reads as a blast. Only runs when the operator has configured TypeSafe.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns, repository.AdvisorSectionSteps},
Run: detectReadsAsBulk,
},
{
Key: "copy_no_clear_ask",
Category: models.AdvisorCategoryCopy,
About: "Copy that asks the reader for nothing, or for several things at once. A cold email earns a reply by making one small, specific request; with none there is nothing to answer, and with several the reader answers none. Judged by a calibrated reader model, and only when the operator has configured TypeSafe.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns, repository.AdvisorSectionSteps},
Run: detectNoClearAsk,
},
}
internal/app/advisor/detect_list.go
+5
View File
@@ -16,30 +16,35 @@ func listDetectors() []Detector {
Key: "list_role_addresses",
Category: models.AdvisorCategoryList,
About: "The share of a campaign's audience that is a shared inbox (info@, sales@, support@). These rarely reply, are frequently monitored by people with no interest in the offer, and complain at a higher rate than named recipients.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns, repository.AdvisorSectionLists},
Run: detectRoleAddresses,
},
{
Key: "list_free_mail_heavy",
Category: models.AdvisorCategoryList,
About: "The share of a B2B campaign's audience on consumer mailbox domains. A cold list that is mostly consumer addresses is usually scraped rather than sourced, and both reply and complaint rates reflect that.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns, repository.AdvisorSectionLists},
Run: detectFreeMailHeavy,
},
{
Key: "list_suppressed_share",
Category: models.AdvisorCategoryList,
About: "Leads already on the organization's suppression list. They are skipped at send time, so a large share means the campaign delivers far less than its numbers suggest.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns, repository.AdvisorSectionLists},
Run: detectSuppressedShare,
},
{
Key: "list_missing_personalization_data",
Category: models.AdvisorCategoryList,
About: "Contacts missing the fields the copy personalizes on, which is how an email ends up opening with 'Hi ,'. This is a data problem that surfaces as a copy problem.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns, repository.AdvisorSectionLists, repository.AdvisorSectionSteps},
Run: detectMissingPersonalizationData,
},
{
Key: "list_unsubscribed_enrolled",
Category: models.AdvisorCategoryList,
About: "Contacts who have unsubscribed but are still enrolled in a campaign. Mailing someone who opted out is the fastest route to a complaint and, in several jurisdictions, is not merely impolite.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionCampaigns, repository.AdvisorSectionLists},
Run: detectUnsubscribedEnrolled,
},
}
internal/app/advisor/detect_mailbox.go
+1
View File
@@ -40,6 +40,7 @@ func mailboxDetectors() []Detector {
Key: "mailbox_concentration",
Category: models.AdvisorCategoryMailbox,
About: "How many mailboxes carry the org's cold volume. Spreading sending across more mailboxes and more sending identities is the core safety property of the platform; concentrating it into one or two mailboxes gives up that protection.",
Reads: []repository.AdvisorSection{repository.AdvisorSectionOrg},
Run: detectMailboxConcentration,
},
{
internal/app/advisor/detect_test.go
+114
View File
@@ -1272,3 +1272,117 @@ func TestDetectWarmupMailNotArriving(t *testing.T) {
t.Fatal("a healthy mailbox is reported")
}
}
// busySnapshot fires as many section-reading detectors as one fixture can.
func busySnapshot() *repository.AdvisorSnapshot {
campaignID, otherID := uuid.New(), uuid.New()
heavy := healthyMailbox()
heavy.ColdSent7d, heavy.ColdSent30d = 5000, 20000
snap := snapshotOf(heavy, healthyMailbox())
snap.Campaigns = []repository.AdvisorCampaign{
{
ID: campaignID, Name: "Outbound", Status: "active",
DailyLimit: 900, StartTime: "09:00", EndTime: "09:30",
SenderCount: 1, SenderCapacity: 50, PickedSenders: 1, SenderStrategy: "explicit",
StepCount: 2, EmailStepCount: 2, VariantCount: 1,
Sent: 3000, Opened: 300, Replied: 1, Bounced: 10,
LeadsTotal: 1000, LeadsRemaining: 5,
},
{ID: otherID, Name: "Solo", Status: "active", StepCount: 1, EmailStepCount: 1, DailyLimit: 10, Sent: 10},
}
snap.Steps = []repository.AdvisorStep{
{
ID: uuid.New(), CampaignID: campaignID, Kind: "email", Position: 0, Name: "Opener", WaitAfter: 1,
Subject: "ACT NOW!! LIMITED TIME OFFER, THIS IS NOT SPAM AND IT IS 100% FREE FOR YOU AND YOUR WHOLE TEAM TODAY",
BodyPlain: "Hi {{.FirstName}},\n\n{{if .Company}}Act now, click here, buy now, limited time, no obligation, 100% free, risk free. " + strings.Repeat("More words here. ", 200),
Sent: 2000, Replied: 40,
},
{
ID: uuid.New(), CampaignID: campaignID, Kind: "email", Position: 1, Name: "Bump",
Subject: "re", BodyPlain: "Hi {{.FirstName}}, any thoughts?",
Sent: 1000, Replied: 0,
},
}
snap.Lists = map[uuid.UUID]repository.AdvisorListStats{
campaignID: {CampaignID: campaignID, Total: 1000, RoleAddresses: 400, FreeMail: 700, Suppressed: 200, Unsubscribed: 30, MissingFirstName: 500},
}
snap.Org = repository.AdvisorOrgStats{Mailboxes: 2, ActiveMailboxes: 2, RunningCampaigns: 2, DuplicateContacts: 150}
return snap
}
// A detector that reads a section it does not declare would resolve its
// findings on a run where that section failed to load.
func TestDetectorsDeclareEverySectionTheyRead(t *testing.T) {
blank := map[repository.AdvisorSection]func(*repository.AdvisorSnapshot){
repository.AdvisorSectionCampaigns: func(s *repository.AdvisorSnapshot) { s.Campaigns = nil },
repository.AdvisorSectionSteps: func(s *repository.AdvisorSnapshot) { s.Steps = nil },
repository.AdvisorSectionLists: func(s *repository.AdvisorSnapshot) { s.Lists = map[uuid.UUID]repository.AdvisorListStats{} },
repository.AdvisorSectionOrg: func(s *repository.AdvisorSnapshot) { s.Org = repository.AdvisorOrgStats{} },
}
fingerprints := func(fs []Finding) string {
out := make([]string, 0, len(fs))
for _, f := range fs {
out = append(out, f.Fingerprint())
}
return strings.Join(out, ",")
}
base := busySnapshot()
fired := 0
for _, d := range AllDetectors() {
full := d.Run(base)
if len(d.Reads) > 0 && len(full) > 0 {
fired++
}
for sec, clear := range blank {
declared := false
for _, r := range d.Reads {
declared = declared || r == sec
}
if declared {
continue
}
s := *base
clear(&s)
if fingerprints(d.Run(&s)) != fingerprints(full) {
t.Errorf("detector %q reads the %s section but does not declare it", d.Key, sec)
}
}
}
if fired < 20 {
t.Errorf("the fixture fires only %d section-reading detectors; the check has no teeth", fired)
}
}
func TestFailedSectionHoldsItsDetectors(t *testing.T) {
snap := busySnapshot()
snap.Failed = map[repository.AdvisorSection]error{repository.AdvisorSectionLists: errors.New("statement timeout")}
settings := defaults()
for _, f := range Detect(snap, settings) {
if strings.HasPrefix(f.Key, "list_") {
t.Errorf("%s opened from a list section that did not load", f.Key)
}
}
held := map[string]bool{}
for _, k := range heldDetectors(snap, settings) {
held[k] = true
}
for _, d := range AllDetectors() {
if want := strings.HasPrefix(d.Key, "list_"); held[d.Key] != want {
t.Errorf("detector %q held = %v, want %v", d.Key, held[d.Key], want)
}
}
// A muted detector is not held, so its findings still close as before.
settings.MutedDetectors = []string{"list_role_addresses"}
for _, k := range heldDetectors(snap, settings) {
if k == "list_role_addresses" {
t.Error("a muted detector was held open")
}
}
snap.Failed = nil
if got := heldDetectors(snap, settings); len(got) != 0 {
t.Errorf("a complete snapshot held %v", got)
}
}
internal/app/advisor/service.go
+7 -2
View File
@@ -188,6 +188,10 @@ func (s *service) Evaluate(ctx context.Context, orgID uuid.UUID, trigger string)
return nil, err
}
for sec, serr := range snapshot.Failed {
log.Printf("advisor: org %s: %s section did not load, keeping its findings: %v", orgID, sec, serr)
}
snapshot.TrackingHost = s.trackingHost
if s.domainAuth != nil {
snapshot.DomainAuthEnforced, snapshot.DomainAuthGrace = s.domainAuth.DomainAuth(ctx)
@@ -213,8 +217,9 @@ func (s *service) Evaluate(ctx context.Context, orgID uuid.UUID, trigger string)
stored = append(stored, row)
}
// Anything the detectors did not produce this pass is no longer true.
closed, err := s.repo.ResolveMissing(ctx, orgID, keep)
// Anything the detectors did not produce this pass is no longer true,
// except what a detector held for a section that did not load.
closed, err := s.repo.ResolveMissing(ctx, orgID, keep, heldDetectors(snapshot, settings))
if err != nil {
log.Printf("advisor: resolve missing for org %s: %v", orgID, err)
}
internal/app/campaign/handlers.go
+1 -1
View File
@@ -1411,7 +1411,7 @@ func (s *campaignService) SendPlan(ctx context.Context, orgID uuid.UUID, campaig
// stalled snapshotter would otherwise serve old figures as fresh.
now := time.Now()
plan.Stale = snap.VersionKey != key || snap.Day != planBudgetDay(now) ||
now.Sub(snap.ComputedAt) > sendPlanSnapshotMaxAge
now.Sub(snap.ComputedAt) > s.sendPlanMaxAge()
if plan.Stale {
s.refreshPlanAsync(campaign, orgID, key)
}
internal/app/campaign/send_plan_snapshot.go
+13 -2
View File
@@ -111,7 +111,10 @@ func (s *campaignService) refreshPlanAsync(campaign *models.Campaign, orgID uuid
// claim means exactly one backend replica walks the fleet each tick rather than
// every replica repeating the work; runOnBoot seeds snapshots promptly after a
// restart. A no-op without the snapshot store or a planner.
func (s *campaignService) StartSendPlanSnapshotter(ctx context.Context, interval time.Duration) {
func (s *campaignService) StartSendPlanSnapshotter(ctx context.Context, interval, maxAge time.Duration) {
if maxAge > 0 {
s.planMaxAge.Store(int64(maxAge))
}
if s.planSnapshotRepo == nil || s.campaignRepository == nil {
return
}
@@ -127,6 +130,14 @@ func (s *campaignService) StartSendPlanSnapshotter(ctx context.Context, interval
})
}
// sendPlanMaxAge is the configured staleness window, else sendPlanSnapshotMaxAge.
func (s *campaignService) sendPlanMaxAge() time.Duration {
if v := s.planMaxAge.Load(); v > 0 {
return time.Duration(v)
}
return sendPlanSnapshotMaxAge
}
// snapshotActiveCampaignsOnce walks every active campaign once, skipping those
// whose stored snapshot is already current, and logs the pass's work.
func (s *campaignService) snapshotActiveCampaignsOnce(ctx context.Context) {
@@ -190,7 +201,7 @@ func (s *campaignService) snapshotCampaign(ctx context.Context, id uuid.UUID) st
// missing snapshots are recomputed.
if snap, err := s.planSnapshotRepo.Get(ctx, orgID, campaign.ID); err == nil && snap != nil &&
snap.VersionKey == key && snap.Day == planBudgetDay(time.Now()) &&
time.Since(snap.ComputedAt) <= sendPlanSnapshotMaxAge {
time.Since(snap.ComputedAt) <= s.sendPlanMaxAge() {
return "skipped"
}
cctx, cancel := context.WithTimeout(ctx, sendPlanSnapshotPerCompute)
internal/app/campaign/service.go
+6 -2
View File
@@ -2,6 +2,7 @@ package campaign
import (
"context"
"sync/atomic"
"time"
"github.com/google/uuid"
@@ -39,8 +40,9 @@ type CampaignService interface {
// StartSendPlanSnapshotter runs the background loop that re-walks every
// active campaign's send plan on an interval and stores it, so SendPlan
// serves a stored snapshot instead of computing on the request. A no-op
// without the snapshot store or a planner.
StartSendPlanSnapshotter(ctx context.Context, interval time.Duration)
// without the snapshot store or a planner. maxAge is when a stored
// snapshot counts as stale; zero keeps the built-in window.
StartSendPlanSnapshotter(ctx context.Context, interval, maxAge time.Duration)
Update(ctx context.Context, orgID, id string, data *models.UpdateCampaign) (*models.Campaign, *errx.Error)
// Delete removes an organization's campaign outright. A running campaign
// is stopped as part of it: its pending tasks are cancelled in the same
@@ -143,6 +145,8 @@ type campaignService struct {
// on the request. Optional/nil-safe: without it SendPlan computes inline
// through the single-flight cache, as it did before.
planSnapshotRepo repository.CampaignSendPlanSnapshotRepository
// planMaxAge overrides sendPlanSnapshotMaxAge when set (nanoseconds).
planMaxAge atomic.Int64
}
// SegmentCounter is the slice of the segment service Estimate needs.
internal/app/emailverify/evidence.go
+4 -3
View File
@@ -112,12 +112,13 @@ func (e *Evidence) Explain(ctx context.Context, contactID uuid.UUID) *models.Con
}
// CreditCleanDeliveries turns sends that never bounced into evidence, then
// rescores the contacts credited. Returns how many contacts changed.
func (e *Evidence) CreditCleanDeliveries(ctx context.Context, limit int) (int, error) {
// rescores the contacts credited. Returns how many contacts changed. A
// lookback > 0 only considers steps sent within it before the window.
func (e *Evidence) CreditCleanDeliveries(ctx context.Context, limit int, lookback time.Duration) (int, error) {
if e == nil || e.repo == nil {
return 0, nil
}
ids, err := e.repo.CreditCleanDeliveries(ctx, time.Duration(config.VerificationDeliveryWindowHours)*time.Hour, limit)
ids, err := e.repo.CreditRecentCleanDeliveries(ctx, time.Duration(config.VerificationDeliveryWindowHours)*time.Hour, lookback, limit)
if err != nil {
return 0, err
}
internal/infrastructure/db/migrations/000301_tasks_placement_by_account_index.down.sql
+1
View File
@@ -0,0 +1 @@
DROP INDEX CONCURRENTLY IF EXISTS public.idx_tasks_placement_by_account;
internal/infrastructure/db/migrations/000301_tasks_placement_by_account_index.up.sql
+3
View File
@@ -0,0 +1,3 @@
-- Alone in its file for CONCURRENTLY. Serves the placement half of the daily send count.
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_tasks_placement_by_account
ON public.tasks (email_account_id) WHERE task_type = 'placement';
internal/infrastructure/db/migrations/000302_drop_suppressed_recipients_org_email_index.down.sql
+2
View File
@@ -0,0 +1,2 @@
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_suppressed_recipients_org_email
ON public.suppressed_recipients USING btree (organization_id, email);
internal/infrastructure/db/migrations/000302_drop_suppressed_recipients_org_email_index.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Covered by the suppressed_recipients_unique constraint on the same columns.
DROP INDEX CONCURRENTLY IF EXISTS public.idx_suppressed_recipients_org_email;
internal/infrastructure/db/migrations/000303_drop_campaign_progress_org_sent_index.down.sql
+2
View File
@@ -0,0 +1,2 @@
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_campaign_progress_org_sent
ON public.campaign_contact_progress USING btree (campaign_id, sent_at DESC) WHERE (sent_at IS NOT NULL);
internal/infrastructure/db/migrations/000303_drop_campaign_progress_org_sent_index.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Covered by idx_campaign_progress_sent, which a btree scan reads in either direction.
DROP INDEX CONCURRENTLY IF EXISTS public.idx_campaign_progress_org_sent;
internal/infrastructure/db/migrations/000304_drop_campaign_progress_campaign_index.down.sql
+2
View File
@@ -0,0 +1,2 @@
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_campaign_progress_campaign
ON public.campaign_contact_progress USING btree (campaign_id);
internal/infrastructure/db/migrations/000304_drop_campaign_progress_campaign_index.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Covered by the primary key, which leads with campaign_id.
DROP INDEX CONCURRENTLY IF EXISTS public.idx_campaign_progress_campaign;
internal/infrastructure/db/migrations/000305_drop_unibox_thread_labels_org_thread_index.down.sql
+2
View File
@@ -0,0 +1,2 @@
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unibox_thread_labels_org_thread
ON public.unibox_thread_labels USING btree (organization_id, thread_id);
internal/infrastructure/db/migrations/000305_drop_unibox_thread_labels_org_thread_index.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Covered by the primary key (organization_id, thread_id, category_id).
DROP INDEX CONCURRENTLY IF EXISTS public.idx_unibox_thread_labels_org_thread;
internal/infrastructure/db/migrations/000306_drop_api_key_usage_key_index.down.sql
+2
View File
@@ -0,0 +1,2 @@
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_api_key_usage_key
ON public.api_key_usage_logs USING btree (api_key_id, created_at DESC);
internal/infrastructure/db/migrations/000306_drop_api_key_usage_key_index.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Covered by idx_api_key_usage_key_time_status, which has the same leading columns.
DROP INDEX CONCURRENTLY IF EXISTS public.idx_api_key_usage_key;
internal/infrastructure/db/migrations/000307_drop_email_message_map_user_email_index.down.sql
+2
View File
@@ -0,0 +1,2 @@
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_email_message_map_user_email
ON public.email_message_map USING btree (user_id, email_id);
internal/infrastructure/db/migrations/000307_drop_email_message_map_user_email_index.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Covered by the primary key (user_id, email_id, message_id).
DROP INDEX CONCURRENTLY IF EXISTS public.idx_email_message_map_user_email;
internal/infrastructure/db/migrations/000308_send_result_effects_delivered_index.down.sql
+1
View File
@@ -0,0 +1 @@
DROP INDEX CONCURRENTLY IF EXISTS public.idx_send_result_effects_delivered;
internal/infrastructure/db/migrations/000308_send_result_effects_delivered_index.up.sql
+3
View File
@@ -0,0 +1,3 @@
-- Alone in its file for CONCURRENTLY. Serves the seven-day cleanup of delivered effects.
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_send_result_effects_delivered
ON public.send_result_effects (delivered_at) WHERE delivered_at IS NOT NULL;
internal/infrastructure/db/migrations/000309_unibox_emails_missing_body_index.down.sql
+1
View File
@@ -0,0 +1 @@
DROP INDEX CONCURRENTLY IF EXISTS public.idx_unibox_emails_missing_body;
internal/infrastructure/db/migrations/000309_unibox_emails_missing_body_index.up.sql
+3
View File
@@ -0,0 +1,3 @@
-- Alone in its file for CONCURRENTLY. Lets the body-text backfill read only the rows it fills.
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unibox_emails_missing_body
ON public.unibox_emails (id) WHERE body_text = '';
internal/infrastructure/db/migrations/000310_unibox_emails_reply_repair_index.down.sql
+1
View File
@@ -0,0 +1 @@
DROP INDEX CONCURRENTLY IF EXISTS public.idx_unibox_emails_reply_repair;
internal/infrastructure/db/migrations/000310_unibox_emails_reply_repair_index.up.sql
+5
View File
@@ -0,0 +1,5 @@
-- Alone in its file for CONCURRENTLY. Same predicate as the campaign reply repair scan.
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unibox_emails_reply_repair
ON public.unibox_emails (id) INCLUDE (created_at)
WHERE campaign_reply_processed_at IS NULL
AND folder NOT IN ('sent', 'drafts') AND provider_folder NOT IN ('sent', 'drafts');
internal/infrastructure/db/migrations/000311_unibox_emails_human_thread_index.down.sql
+1
View File
@@ -0,0 +1 @@
DROP INDEX CONCURRENTLY IF EXISTS public.idx_unibox_emails_human_thread;
internal/infrastructure/db/migrations/000311_unibox_emails_human_thread_index.up.sql
+4
View File
@@ -0,0 +1,4 @@
-- Alone in its file for CONCURRENTLY. Serves the person-written half of the automated-conversation test.
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unibox_emails_human_thread
ON public.unibox_emails (thread_id) INCLUDE (email_id)
WHERE NOT automated AND folder NOT IN ('sent', 'drafts', 'spam', 'trash');
internal/infrastructure/db/migrations/000312_unibox_emails_inbox_unseen_index.down.sql
+1
View File
@@ -0,0 +1 @@
DROP INDEX CONCURRENTLY IF EXISTS public.idx_unibox_emails_inbox_unseen;
internal/infrastructure/db/migrations/000312_unibox_emails_inbox_unseen_index.up.sql
+4
View File
@@ -0,0 +1,4 @@
-- Alone in its file for CONCURRENTLY. Serves the unread badge and unread counts.
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unibox_emails_inbox_unseen
ON public.unibox_emails (email_id) INCLUDE (thread_id, user_id, automated)
WHERE NOT seen AND folder = 'inbox';
internal/jobs/verification_evidence.go
+34 -10
View File
@@ -8,6 +8,13 @@ import (
"github.com/warmbly/warmbly/internal/jobrun"
)
const (
// deliveryEvidenceLookback bounds a regular run to steps that came due recently.
deliveryEvidenceLookback = 7 * 24 * time.Hour
// deliveryEvidenceFullEvery is how often a run walks every sent step instead.
deliveryEvidenceFullEvery = 24 * time.Hour
)
// DeliveryEvidenceJob turns campaign sends that never bounced into
// verification evidence: a delivery the recipient's server kept is the
// strongest proof the mailbox exists, and it costs nothing to observe.
@@ -15,6 +22,7 @@ type DeliveryEvidenceJob struct {
evidence *emailverifyapp.Evidence
interval time.Duration
batch int
lastFull time.Time
}
func NewDeliveryEvidenceJob(evidence *emailverifyapp.Evidence, interval time.Duration, batch int) *DeliveryEvidenceJob {
@@ -27,15 +35,31 @@ func NewDeliveryEvidenceJob(evidence *emailverifyapp.Evidence, interval time.Dur
// Start runs the job on its interval until ctx ends. A full batch repeats
// at once so a backlog drains.
func (j *DeliveryEvidenceJob) Start(ctx context.Context) {
jobrun.Loop(ctx, "delivery_evidence", j.interval, false, func(ctx context.Context) error {
for {
n, err := j.evidence.CreditCleanDeliveries(ctx, j.batch)
if err != nil {
return err
}
if n < j.batch || ctx.Err() != nil {
return nil
}
jobrun.Loop(ctx, "delivery_evidence", j.interval, false, j.run)
}
// run credits recent steps, and every step on the first run and once a day after.
func (j *DeliveryEvidenceJob) run(ctx context.Context) error {
full := j.lastFull.IsZero() || time.Since(j.lastFull) >= deliveryEvidenceFullEvery
lookback := deliveryEvidenceLookback
if full {
lookback = 0
}
started := time.Now()
for {
n, err := j.evidence.CreditCleanDeliveries(ctx, j.batch, lookback)
if err != nil {
return err
}
})
if ctx.Err() != nil {
return nil
}
if n < j.batch {
break
}
}
if full {
j.lastFull = started
}
return nil
}
internal/repository/pg_advisor.go
+13 -5
View File
@@ -224,6 +224,8 @@ type AdvisorSnapshot struct {
// a step with no entry is simply not judged, and no copy detector that
// reads a verdict fires on it.
CopyJudgments map[uuid.UUID]copyjudge.Verdict
// Failed holds each section that did not load this run, with its error.
Failed map[AdvisorSection]error
}
// AdvisorRepository persists findings and loads the evaluation snapshot.
@@ -237,8 +239,9 @@ type AdvisorRepository interface {
// dismissed one stays dismissed until it resolves first.
UpsertFinding(ctx context.Context, f *models.AdvisorFinding) (*models.AdvisorFinding, bool, error)
// ResolveMissing closes every open/snoozed/applied finding for the org whose
// fingerprint is not in keep, returning how many it closed.
ResolveMissing(ctx context.Context, orgID uuid.UUID, keep []string) (int, error)
// fingerprint is not in keep and whose detector is not in keepDetectors,
// returning how many it closed.
ResolveMissing(ctx context.Context, orgID uuid.UUID, keep, keepDetectors []string) (int, error)
// ExpireSnoozes flips snoozed findings whose window has passed back to open.
ExpireSnoozes(ctx context.Context) (int, error)
@@ -438,14 +441,19 @@ func evidenceHashOf(f *models.AdvisorFinding) string {
return f.EvidenceHash()
}
func (r *advisorRepository) ResolveMissing(ctx context.Context, orgID uuid.UUID, keep []string) (int, error) {
func (r *advisorRepository) ResolveMissing(ctx context.Context, orgID uuid.UUID, keep, keepDetectors []string) (int, error) {
// A nil slice binds as NULL, and NOT (x = ANY(NULL)) would match no row.
if keepDetectors == nil {
keepDetectors = []string{}
}
query := `
UPDATE advisor_findings
SET status = 'resolved', resolved_at = NOW()
WHERE organization_id = $1
AND status IN ('open', 'snoozed', 'applied', 'dismissed')
AND NOT (fingerprint = ANY($2))`
tag, err := r.db.Exec(ctx, query, orgID, keep)
AND NOT (fingerprint = ANY($2))
AND NOT (detector_key = ANY($3))`
tag, err := r.db.Exec(ctx, query, orgID, keep, keepDetectors)
if err != nil {
return 0, err
}
internal/repository/pg_advisor_snapshot.go
+44 -9
View File
@@ -9,10 +9,37 @@ import (
"github.com/warmbly/warmbly/internal/models"
)
// AdvisorSection names a snapshot section that loads on its own.
type AdvisorSection string
const (
AdvisorSectionCampaigns AdvisorSection = "campaigns"
AdvisorSectionSteps AdvisorSection = "steps"
AdvisorSectionLists AdvisorSection = "lists"
AdvisorSectionOrg AdvisorSection = "org"
)
// SectionFailed reports whether sec did not load this run.
func (s *AdvisorSnapshot) SectionFailed(sec AdvisorSection) bool {
_, failed := s.Failed[sec]
return failed
}
func (s *AdvisorSnapshot) fail(sec AdvisorSection, err error) {
if err == nil {
return
}
if s.Failed == nil {
s.Failed = map[AdvisorSection]error{}
}
s.Failed[sec] = err
}
// LoadSnapshot reads one org's advisor view. Every section is best-effort: a
// single failing rollup must not cost the org its whole evaluation, because a
// partial snapshot still produces correct findings for the parts that loaded
// (detectors are written to no-op on absent data rather than guess).
// partial snapshot still produces correct findings for the parts that loaded.
// A section that fails is recorded in Failed, so the detectors that read it
// are held for the run rather than run on empty data.
func (r *advisorRepository) LoadSnapshot(ctx context.Context, orgID uuid.UUID, now time.Time) (*AdvisorSnapshot, error) {
snap := &AdvisorSnapshot{
OrganizationID: orgID,
@@ -29,10 +56,14 @@ func (r *advisorRepository) LoadSnapshot(ctx context.Context, orgID uuid.UUID, n
}
snap.Mailboxes = mailboxes
snap.Campaigns, _ = r.loadCampaigns(ctx, orgID)
snap.Steps, _ = r.loadSteps(ctx, orgID)
snap.Lists, _ = r.loadListStats(ctx, orgID)
snap.Org, _ = r.loadOrgStats(ctx, orgID)
snap.Campaigns, err = r.loadCampaigns(ctx, orgID)
snap.fail(AdvisorSectionCampaigns, err)
snap.Steps, err = r.loadSteps(ctx, orgID)
snap.fail(AdvisorSectionSteps, err)
snap.Lists, err = r.loadListStats(ctx, orgID)
snap.fail(AdvisorSectionLists, err)
snap.Org, err = r.loadOrgStats(ctx, orgID)
snap.fail(AdvisorSectionOrg, err)
return snap, nil
}
@@ -77,15 +108,19 @@ func (r *advisorRepository) loadMailboxes(ctx context.Context, orgID uuid.UUID,
AND t.completed_at > NOW() - INTERVAL '30 days'
AND ` + taskDispatchedEmail + `
) sent ON true
LEFT JOIN LATERAL (
-- One pass per org: no index bounds a single mailbox's events, so a lateral rescanned the window per mailbox.
LEFT JOIN (
SELECT
t2.email_account_id,
COUNT(*) FILTER (WHERE de.event_type = 'bounce') AS bounces,
COUNT(*) FILTER (WHERE de.event_type = 'complaint') AS complaints
FROM deliverability_events de
JOIN tasks t2 ON t2.id = de.task_id
WHERE t2.email_account_id = ea.id
WHERE t2.email_account_id IN (SELECT id FROM email_accounts WHERE organization_id = $1)
AND de.event_type IN ('bounce', 'complaint')
AND de.created_at > NOW() - INTERVAL '30 days'
) dl ON true
GROUP BY t2.email_account_id
) dl ON dl.email_account_id = ea.id
LEFT JOIN LATERAL (
SELECT
COALESCE(SUM(emails_sent), 0) AS sent7,
internal/repository/pg_analytics.go
+12 -7
View File
@@ -781,13 +781,7 @@ func (r *analyticsRepository) GetDashboardOverallStats(ctx context.Context, orgI
COUNT(CASE WHEN ccp.opened_at IS NOT NULL AND NOT ccp.opened_machine AND ccp.sent_at >= $2 AND ccp.sent_at <= $3 THEN 1 END) as total_opens,
COUNT(CASE WHEN ccp.opened_at IS NOT NULL AND ccp.opened_machine AND ccp.sent_at >= $2 AND ccp.sent_at <= $3 THEN 1 END) as machine_opens,
COUNT(CASE WHEN ccp.clicked_at IS NOT NULL AND ccp.sent_at >= $2 AND ccp.sent_at <= $3 THEN 1 END) as total_clicks,
COUNT(CASE WHEN ccp.clicked_at IS NULL AND ccp.sent_at >= $2 AND ccp.sent_at <= $3 AND EXISTS (
SELECT 1 FROM email_link_clicks lc
WHERE lc.campaign_id = ccp.campaign_id AND lc.contact_id = ccp.contact_id AND lc.sequence_id = ccp.sequence_id AND lc.machine
) AND NOT EXISTS (
SELECT 1 FROM email_link_clicks lc
WHERE lc.campaign_id = ccp.campaign_id AND lc.contact_id = ccp.contact_id AND lc.sequence_id = ccp.sequence_id AND NOT lc.machine
) THEN 1 END) as machine_clicks,
COUNT(CASE WHEN ccp.clicked_at IS NULL AND ccp.sent_at >= $2 AND ccp.sent_at <= $3 AND mo.campaign_id IS NOT NULL THEN 1 END) as machine_clicks,
COUNT(CASE WHEN ccp.replied_at IS NOT NULL AND ccp.sent_at >= $2 AND ccp.sent_at <= $3 THEN 1 END) as total_replies,
COUNT(CASE WHEN ccp.bounced_at IS NOT NULL AND ccp.sent_at >= $2 AND ccp.sent_at <= $3 THEN 1 END) as total_bounces,
(SELECT COUNT(*) FROM campaigns ac WHERE ac.organization_id = $1 AND ac.status = 'active'` + inCampaignScope("ac.id", scope, 4) + `) as active_campaigns,
@@ -795,6 +789,15 @@ func (r *analyticsRepository) GetDashboardOverallStats(ctx context.Context, orgI
FROM campaign_contact_progress ccp
JOIN campaigns c ON c.id = ccp.campaign_id
JOIN sequences s ON s.id = ccp.sequence_id AND s.kind = 'email'
-- Steps with at least one click, every one of them machine (machine is NOT NULL).
LEFT JOIN (
SELECT lc.campaign_id, lc.contact_id, lc.sequence_id
FROM email_link_clicks lc
JOIN campaigns lk ON lk.id = lc.campaign_id
WHERE lk.organization_id = $1` + inCampaignScope("lc.campaign_id", scope, 4) + `
GROUP BY lc.campaign_id, lc.contact_id, lc.sequence_id
HAVING bool_and(lc.machine)
) mo ON mo.campaign_id = ccp.campaign_id AND mo.contact_id = ccp.contact_id AND mo.sequence_id = ccp.sequence_id
WHERE c.organization_id = $1` + inCampaignScope("c.id", scope, 4) + `
AND ccp.sent_at >= $2 AND ccp.sent_at <= $3
`
@@ -1018,6 +1021,8 @@ func (r *analyticsRepository) GetDashboardDailyTrend(ctx context.Context, orgID
JOIN sequences s ON s.id = ccp.sequence_id AND s.kind = 'email'
WHERE c.organization_id = $1
AND ccp.sent_at IS NOT NULL
-- Index range with a day of slack for any zone offset; the ::date tests decide.
AND ccp.sent_at >= $2::date - 1 AND ccp.sent_at < $3::date + 2
AND ccp.sent_at::date >= $2
AND ccp.sent_at::date <= $3` + inCampaignScope("c.id", scope, 4) + `
GROUP BY sent_at::date
internal/repository/pg_campaign_progress.go
+231 -56
View File
@@ -1155,6 +1155,8 @@ func (r *campaignProgressRepository) GetCampaignProgress(ctx context.Context, ca
COUNT(*) FILTER (WHERE ccp.bounced_at IS NOT NULL) AS emails_bounced,
COUNT(*) FILTER (WHERE ccp.complained_at IS NOT NULL) AS emails_complained
FROM campaign_contact_progress ccp
-- At most one match per row (sequences.id is the key), so no row is duplicated or dropped.
LEFT JOIN sequences es ON es.id = ccp.sequence_id AND es.kind = 'email'
WHERE ccp.campaign_id = $1
`
@@ -1409,6 +1411,9 @@ func (r *campaignProgressRepository) FindRoutedPairs(ctx context.Context, campai
// soon as the batch is full. The ordering, holds, due/not-due and
// sender-pacing decisions below are exactly the ones the single-query scan
// made; only the shape of the SQL work changed.
// Each chunk first reads only its leads' latest sent step, so a lead whose
// flow has ended or whose next step cannot be due yet is never hydrated;
// a pass that finds nothing due walks again unpruned for the next-due time.
// p1args carries phase one's bound parameters. The custom-field key is the
// only part of this ORDER BY that comes from a customer, so it is bound
@@ -1475,80 +1480,136 @@ func (r *campaignProgressRepository) FindRoutedPairs(ctx context.Context, campai
hydrateQuery := routedLeadsQuery("", " AND cl.contact_id = ANY($3)")
chunkSize := config.CampaignRoutedCandidateChunk
floors := router.dueFloors()
// inputs keeps every hydrated lead, so the next-due walk below reuses them.
inputs := make(map[uuid.UUID]routeInput)
hydrate := func(ids []uuid.UUID) error {
hydrated, herr := r.hydrateRoutedLeads(ctx, hydrateQuery, campaignID, ids)
for id, in := range hydrated {
inputs[id] = in
}
return herr
}
// notYetDue are the leads skipped because no route out of their step can be
// due yet; they matter only to next-due, so only a pass with no pair reads them.
var notYetDue []uuid.UUID
for start := 0; start < len(ordered); start += chunkSize {
end := start + chunkSize
if end > len(ordered) {
end = len(ordered)
}
window := ordered[start:end]
hydrated, herr := r.hydrateRoutedLeads(ctx, hydrateQuery, campaignID, window)
if herr != nil {
facts, ferr := r.leadRouteFacts(ctx, campaignID, window)
if ferr != nil {
return nil, nil, false, ferr
}
// An inert lead adds neither a pair nor a next-due time, so it is never hydrated.
want := make([]uuid.UUID, 0, len(window))
for _, contactID := range window {
f, ok := facts[contactID]
switch {
case !ok || router.inert(f, excludeNewLeads):
case !router.mayBeDue(f, floors):
notYetDue = append(notYetDue, contactID)
default:
want = append(want, contactID)
}
}
if len(want) == 0 {
continue
}
if herr := hydrate(want); herr != nil {
return nil, nil, false, herr
}
// Route the window in its phase-one order so the batch keeps routing
// order across chunk boundaries.
for _, contactID := range window {
in, ok := hydrated[contactID]
if !ok {
// A pre-send gate (bounced, failed past the ceiling, suppressed,
// unsubscribed, undeliverable) dropped the lead in hydration,
// exactly as the single-query WHERE would have. It is not a
// candidate, so it contributes no next-due time either.
continue
}
if in.lastSeq == nil && excludeNewLeads {
continue
}
res := router.route(campaignID, contactID, in)
// Held with no end: there is nothing to wake up for, so it must not
// be reported as a next-due time either — that would park the
// campaign's chain on a moment that never arrives. Checked before the
// condition window, which a held lead can also be sitting in.
if res.Hold != nil && res.Hold.Until == nil {
continue
}
if res.WaitUntil != nil {
// Not decidable yet — remember the soonest window so the scheduler
// can re-check exactly then instead of guessing or completing.
noteDue(*res.WaitUntil, false)
continue
}
if res.Target == nil {
continue // reached the end / a STOP / a step already received
}
// When is this step due? Skip it (but remember when) if not yet: the
// contacts behind this one may be sendable right now.
if res.DueAt != nil && res.DueAt.After(router.dueBy) {
noteDue(*res.DueAt, false)
continue
}
// Due, but the mailbox this lead's conversation belongs to has
// nothing left today. Wait for it — the address a contact has already
// heard from is worth more than sending the follow-up a few hours
// earlier from a stranger. Only an email step waits: an action or
// wait node sends nothing, so no mailbox has to be free for it.
if back, ok := paced.reopening(in.sender); ok && router.isEmailStep(*res.Target) {
noteDue(back, true)
continue
}
pairs = append(pairs, ContactSequencePair{ContactID: contactID, SequenceID: *res.Target, IsNewLead: res.IsNewLead, Instant: res.Instant, NotBefore: res.DueAt, AssignedSender: in.sender})
if len(pairs) >= limit {
// A full batch makes the deferral times collected so far
// irrelevant: the caller is sending, not waiting. Stop before
// hydrating any further chunk.
return pairs, nil, false, nil
}
if routeWindow(campaignID, router, want, inputs, excludeNewLeads, paced, limit, &pairs, noteDue) {
return pairs, nil, false, nil
}
}
// A due pair makes the deferral times irrelevant: the caller is sending.
if len(pairs) > 0 {
return pairs, nil, false, nil
}
if len(notYetDue) > 0 {
// Nothing is due: hydrate the skipped leads and walk everyone in order for next-due.
nextDue, waitingOnSender = nil, false
for start := 0; start < len(notYetDue); start += chunkSize {
end := start + chunkSize
if end > len(notYetDue) {
end = len(notYetDue)
}
if herr := hydrate(notYetDue[start:end]); herr != nil {
return nil, nil, false, herr
}
}
if routeWindow(campaignID, router, ordered, inputs, excludeNewLeads, paced, limit, &pairs, noteDue) || len(pairs) > 0 {
return pairs, nil, false, nil
}
}
// Nobody sendable now. Hand back the soonest moment somebody will be so the
// scheduler defers until then rather than completing.
return nil, nextDue, waitingOnSender, nil
}
// routeWindow routes one hydrated window in order, appending due pairs and
// noting next-due times. It reports true once the batch holds limit pairs.
func routeWindow(campaignID uuid.UUID, router *campaignRouter, window []uuid.UUID, hydrated map[uuid.UUID]routeInput, excludeNewLeads bool, paced PacedSenders, limit int, pairs *[]ContactSequencePair, noteDue func(time.Time, bool)) bool {
// Route the window in its phase-one order so the batch keeps routing
// order across chunk boundaries.
for _, contactID := range window {
in, ok := hydrated[contactID]
if !ok {
// A pre-send gate (bounced, failed past the ceiling, suppressed,
// unsubscribed, undeliverable) dropped the lead in hydration,
// exactly as the single-query WHERE would have. It is not a
// candidate, so it contributes no next-due time either.
continue
}
if in.lastSeq == nil && excludeNewLeads {
continue
}
res := router.route(campaignID, contactID, in)
// Held with no end: there is nothing to wake up for, so it must not
// be reported as a next-due time either — that would park the
// campaign's chain on a moment that never arrives. Checked before the
// condition window, which a held lead can also be sitting in.
if res.Hold != nil && res.Hold.Until == nil {
continue
}
if res.WaitUntil != nil {
// Not decidable yet — remember the soonest window so the scheduler
// can re-check exactly then instead of guessing or completing.
noteDue(*res.WaitUntil, false)
continue
}
if res.Target == nil {
continue // reached the end / a STOP / a step already received
}
// When is this step due? Skip it (but remember when) if not yet: the
// contacts behind this one may be sendable right now.
if res.DueAt != nil && res.DueAt.After(router.dueBy) {
noteDue(*res.DueAt, false)
continue
}
// Due, but the mailbox this lead's conversation belongs to has
// nothing left today. Wait for it — the address a contact has already
// heard from is worth more than sending the follow-up a few hours
// earlier from a stranger. Only an email step waits: an action or
// wait node sends nothing, so no mailbox has to be free for it.
if back, ok := paced.reopening(in.sender); ok && router.isEmailStep(*res.Target) {
noteDue(back, true)
continue
}
*pairs = append(*pairs, ContactSequencePair{ContactID: contactID, SequenceID: *res.Target, IsNewLead: res.IsNewLead, Instant: res.Instant, NotBefore: res.DueAt, AssignedSender: in.sender})
if len(*pairs) >= limit {
// A full batch makes the deferral times collected so far
// irrelevant: the caller is sending, not waiting. Stop before
// hydrating any further chunk.
return true
}
}
return false
}
// orderedCandidateIDs enumerates a campaign's leads in routing order without
// the history, classification, suppression or sent-set work the full scan
// does, so the ordered candidate window is cheap to produce. The pre-send
@@ -1578,6 +1639,52 @@ func (r *campaignProgressRepository) orderedCandidateIDs(ctx context.Context, or
return ids, rows.Err()
}
// leadFacts is what routing needs to tell whether a lead can act at all: when
// it entered, and the step(s) holding its latest sent_at.
type leadFacts struct {
addedAt *time.Time
// lastAt is the latest sent_at, nil for a lead with nothing sent.
lastAt *time.Time
// lastSeqs are every step sent at lastAt, since routing may read any of them.
lastSeqs []uuid.UUID
}
// leadRouteFacts reads leadFacts for a chunk of leads; one absent from the
// result is no longer a lead of the campaign.
func (r *campaignProgressRepository) leadRouteFacts(ctx context.Context, campaignID uuid.UUID, ids []uuid.UUID) (map[uuid.UUID]leadFacts, error) {
rows, err := resultDB(ctx, r.db).Query(ctx, `
SELECT cl.contact_id, cl.added_at, p.sequence_id, p.sent_at
FROM campaign_leads cl
LEFT JOIN campaign_contact_progress p
ON p.campaign_id = cl.campaign_id AND p.contact_id = cl.contact_id AND p.sent_at IS NOT NULL
WHERE cl.campaign_id = $1 AND cl.contact_id = ANY($2)`, campaignID, ids)
if err != nil {
return nil, err
}
defer rows.Close()
out := make(map[uuid.UUID]leadFacts, len(ids))
for rows.Next() {
var contactID uuid.UUID
var addedAt, sentAt *time.Time
var seq *uuid.UUID
if serr := rows.Scan(&contactID, &addedAt, &seq, &sentAt); serr != nil {
return nil, serr
}
f := out[contactID]
f.addedAt = addedAt
if seq != nil && sentAt != nil {
switch {
case f.lastAt == nil || sentAt.After(*f.lastAt):
f.lastAt, f.lastSeqs = sentAt, []uuid.UUID{*seq}
case sentAt.Equal(*f.lastAt):
f.lastSeqs = append(f.lastSeqs, *seq)
}
}
out[contactID] = f
}
return out, rows.Err()
}
// hydrateRoutedLeads runs the shared routedLeadsQuery for one bounded chunk of
// already ordered candidate ids, returning the routing input keyed by contact.
// A candidate a pre-send gate excludes is simply absent from the result.
@@ -2150,6 +2257,72 @@ func (cr *campaignRouter) isEmailStep(id uuid.UUID) bool {
return ok && cr.steps[idx].kind == "email"
}
// dueFloors maps each step to the shortest delay route() can put on any step
// after it; a step with no live target out of it is absent.
func (cr *campaignRouter) dueFloors() map[uuid.UUID]time.Duration {
out := make(map[uuid.UUID]time.Duration, len(cr.steps))
for _, s := range cr.steps {
for i := range s.bc.Branches {
b := &s.bc.Branches[i]
if b.TargetSequenceID == nil {
continue
}
ti, live := cr.idxByID[*b.TargetSequenceID]
if !live {
continue
}
d := 24 * time.Hour * time.Duration(BranchWaitAfter(cr.steps[ti].waitAfter, branchIsInstant(b)))
if s.waitMinutes > 0 {
d += time.Duration(s.waitMinutes) * time.Minute
}
if cur, ok := out[s.id]; !ok || d < cur {
out[s.id] = d
}
}
}
return out
}
// mayBeDue is false only when route() cannot return a step due by dueBy for
// any lead with these facts: holds and history only ever push DueAt later.
func (cr *campaignRouter) mayBeDue(f leadFacts, floors map[uuid.UUID]time.Duration) bool {
if f.lastAt == nil {
return cr.entryDelay <= 0 || !cr.enteredAt(f.addedAt).Add(cr.entryDelay).After(cr.dueBy)
}
for _, seq := range f.lastSeqs {
if d, ok := floors[seq]; ok && !f.lastAt.Add(d).After(cr.dueBy) {
return true
}
}
return false
}
// inert is true for a lead FindRoutedPairs neither sends nor waits for: a new
// lead while new leads are excluded, or one whose current step has no branches.
func (cr *campaignRouter) inert(f leadFacts, excludeNewLeads bool) bool {
if f.lastAt == nil {
return excludeNewLeads
}
for _, seq := range f.lastSeqs {
if idx, ok := cr.idxByID[seq]; ok && len(cr.steps[idx].bc.Branches) > 0 {
return false
}
}
return true
}
// endSteps are the steps with no outgoing branch, where every lead stops.
func (cr *campaignRouter) endSteps() []uuid.UUID {
// Never nil: pgx sends a nil slice as NULL, and <> ALL(NULL) drops every row.
out := []uuid.UUID{}
for _, s := range cr.steps {
if len(s.bc.Branches) == 0 {
out = append(out, s.id)
}
}
return out
}
// enteredAt is when a lead entered the campaign. campaign_leads.added_at is
// nullable and deliberately not backfilled, so a lead that predates the column
// falls back to the campaign's own creation time — which keeps turning an entry
@@ -2548,7 +2721,9 @@ func (r *campaignProgressRepository) LeadSupply(ctx context.Context, campaignID
if router == nil {
return out, nil
}
rows, err := resultDB(ctx, r.db).Query(ctx, routedLeadsQuery("c.created_at ASC", ""), campaignID, config.CampaignSendMaxAttempts)
// Order-free (only counts and a minimum), and a lead whose last step ends the
// flow routes to nothing, so neither is paid for.
rows, err := resultDB(ctx, r.db).Query(ctx, routedLeadsQuery("", " AND (lp.sequence_id IS NULL OR lp.sequence_id <> ALL($3))"), campaignID, config.CampaignSendMaxAttempts, router.endSteps())
if err != nil {
return nil, err
}
internal/repository/pg_contact.go
+4 -3
View File
@@ -2084,9 +2084,10 @@ func (r *contactRepository) SearchCounts(ctx context.Context, orgID string) (*mo
COUNT(*) FILTER (WHERE COALESCE(cl.campaign_count, 0) = 0)
FROM contacts c
LEFT JOIN (
SELECT contact_id, COUNT(campaign_id) AS campaign_count
FROM campaign_leads
GROUP BY contact_id
SELECT l.contact_id, COUNT(l.campaign_id) AS campaign_count
FROM campaign_leads l
JOIN contacts lc ON lc.id = l.contact_id AND lc.organization_id = $1
GROUP BY l.contact_id
) cl ON c.id = cl.contact_id
WHERE c.organization_id = $1
`
internal/repository/pg_verification_evidence.go
+20 -2
View File
@@ -33,6 +33,9 @@ type VerificationEvidenceRepository interface {
// step sent at least `window` ago that never bounced and is not yet in
// the ledger. Returns the contacts credited.
CreditCleanDeliveries(ctx context.Context, window time.Duration, limit int) ([]uuid.UUID, error)
// CreditRecentCleanDeliveries is CreditCleanDeliveries limited to steps sent
// within lookback before the window; a full pass still catches older ones.
CreditRecentCleanDeliveries(ctx context.Context, window, lookback time.Duration, limit int) ([]uuid.UUID, error)
}
// ContactVerdict is a contact's last verdict (the check's own status) and any re-check waiting on it.
@@ -162,6 +165,15 @@ func (r *verificationEvidenceRepository) SetScore(ctx context.Context, contactID
}
func (r *verificationEvidenceRepository) CreditCleanDeliveries(ctx context.Context, window time.Duration, limit int) ([]uuid.UUID, error) {
return r.creditCleanDeliveries(ctx, window, 0, limit)
}
func (r *verificationEvidenceRepository) CreditRecentCleanDeliveries(ctx context.Context, window, lookback time.Duration, limit int) ([]uuid.UUID, error) {
return r.creditCleanDeliveries(ctx, window, lookback, limit)
}
// creditCleanDeliveries bounds sent_at only when lookback > 0, as its own statement text so a generic plan keeps the index range.
func (r *verificationEvidenceRepository) creditCleanDeliveries(ctx context.Context, window, lookback time.Duration, limit int) ([]uuid.UUID, error) {
if limit <= 0 {
limit = 1000
}
@@ -175,6 +187,13 @@ func (r *verificationEvidenceRepository) CreditCleanDeliveries(ctx context.Conte
// change went to a different mailbox and are not evidence about this one.
// dispatched_at, not sent_at: sent_at is stamped when the worker's result
// lands, which for a send already on the bus can be after the edit.
params := []any{window.Seconds(), limit}
since := ""
if lookback > 0 {
since = `
AND p.sent_at >= NOW() - make_interval(secs => $3)`
params = append(params, (window + lookback).Seconds())
}
query := `
WITH due AS (
SELECT p.contact_id, p.campaign_id, p.sequence_id, p.sent_at
@@ -182,7 +201,7 @@ func (r *verificationEvidenceRepository) CreditCleanDeliveries(ctx context.Conte
JOIN contacts c ON c.id = p.contact_id
WHERE p.sent_at IS NOT NULL AND p.bounced_at IS NULL
AND ` + progressIsEmailStep("p") + `
AND p.sent_at < NOW() - make_interval(secs => $1)
AND p.sent_at < NOW() - make_interval(secs => $1)` + since + `
AND (c.verification_evidence_reset_at IS NULL OR COALESCE(p.dispatched_at, p.sent_at) > c.verification_evidence_reset_at)
AND NOT EXISTS (
SELECT 1 FROM contact_verification_evidence e
@@ -200,7 +219,6 @@ func (r *verificationEvidenceRepository) CreditCleanDeliveries(ctx context.Conte
)
SELECT DISTINCT contact_id FROM ins
`
params := []any{window.Seconds(), limit}
rows, err := resultDB(ctx, r.DB).Query(ctx, query, params...)
if err != nil {
db.CaptureError(err, query, params, "query")
internal/repository/routed_pairs_prune_test.go
+178
View File
@@ -0,0 +1,178 @@
package repository
import (
"math/rand"
"testing"
"time"
"github.com/google/uuid"
"github.com/warmbly/warmbly/internal/models"
)
// FindRoutedPairs skips hydrating a lead when its facts say route() cannot
// offer it (mayBeDue) or cannot offer or wake for it (inert). Those skips are
// only safe if they never disagree with route() itself, so this drives both
// over random flows and leads and fails on any lead the skip would lose.
func TestRoutedPairsPruneNeverDropsWhatRouteWouldOffer(t *testing.T) {
rng := rand.New(rand.NewSource(993))
now := time.Now()
conditions := [][]models.BranchCondition{
nil,
{{Field: "opened", Operator: "within_days", Value: intPtr(1)}},
{{Field: "not_opened", Operator: "within_days", Value: intPtr(3)}},
{{Field: "clicked", Operator: "ever"}},
{{Field: "replied", Operator: "within_days", Value: intPtr(2)}},
{{Field: "not_replied", Operator: "within_days", Value: intPtr(1)}},
{{Field: "reply_positive", Operator: "ever"}},
}
ago := func(maxHours int) *time.Time {
at := now.Add(-time.Duration(rng.Intn(maxHours*60+1)) * time.Minute)
return &at
}
maybe := func(at *time.Time) *time.Time {
if rng.Intn(2) == 0 {
return nil
}
return at
}
for i := 0; i < 20000; i++ {
n := 1 + rng.Intn(5)
ids := make([]uuid.UUID, n)
for j := range ids {
ids[j] = uuid.New()
}
pick := func() uuid.UUID {
if rng.Intn(8) == 0 {
return uuid.New() // a deleted step
}
return ids[rng.Intn(n)]
}
cr := &campaignRouter{
idxByID: map[uuid.UUID]int{},
replyFlowSteps: map[uuid.UUID]bool{},
now: now,
dueBy: now.Add(time.Minute),
stopOnReply: rng.Intn(2) == 0,
entryDelay: []time.Duration{0, 0, time.Hour, 72 * time.Hour}[rng.Intn(4)],
campaignCreatedAt: now.Add(-48 * time.Hour),
}
for j, id := range ids {
s := routeStep{id: id, kind: []string{"email", "email", "action"}[rng.Intn(3)], waitAfter: rng.Intn(4)}
if s.kind == "action" && rng.Intn(2) == 0 {
s.waitMinutes = 30 + rng.Intn(600)
}
for b := rng.Intn(4); b > 0; b-- {
br := models.Branch{BranchID: uuid.NewString(), Conditions: conditions[rng.Intn(len(conditions))]}
if rng.Intn(5) > 0 {
target := pick()
br.TargetSequenceID = &target
}
switch rng.Intn(3) {
case 1:
v := true
br.Instant = &v
case 2:
v := false
br.Instant = &v
}
s.bc.Branches = append(s.bc.Branches, br)
}
cr.steps = append(cr.steps, s)
cr.idxByID[id] = j
if rng.Intn(3) == 0 {
cr.replyFlowSteps[id] = true
}
}
cr.entry = ids[0]
floors := cr.dueFloors()
in := routeInput{addedAt: maybe(ago(96)), hasReplied: rng.Intn(3) == 0}
if rng.Intn(4) > 0 {
last := pick()
in.lastSeq = &last
in.sentAt = ago(24 * 6)
in.openedAt, in.clickedAt, in.repliedAt = maybe(ago(48)), maybe(ago(48)), maybe(ago(48))
in.replyClass = []string{"", "positive", "negative"}[rng.Intn(3)]
}
for k := rng.Intn(n + 1); k > 0; k-- {
in.sentIDs = append(in.sentIDs, pick())
}
switch rng.Intn(4) {
case 1:
in.pausedAt = ago(72)
case 2:
in.pausedAt = ago(72)
until := now.Add(time.Duration(rng.Intn(96)-48) * time.Hour)
in.pausedUntil = &until
}
f := leadFacts{addedAt: in.addedAt}
if in.lastSeq != nil {
f.lastAt = in.sentAt
f.lastSeqs = []uuid.UUID{*in.lastSeq}
if rng.Intn(4) == 0 {
// A tie on sent_at: hydration may read either step.
f.lastSeqs = append(f.lastSeqs, pick())
}
}
res := cr.route(uuid.New(), uuid.New(), in)
heldForever := res.Hold != nil && res.Hold.Until == nil
offered := !heldForever && res.WaitUntil == nil && res.Target != nil &&
(res.DueAt == nil || !res.DueAt.After(cr.dueBy))
if offered && !cr.mayBeDue(f, floors) {
t.Fatalf("case %d: route offers %v (due %v) but mayBeDue is false", i, *res.Target, res.DueAt)
}
for _, excludeNew := range []bool{false, true} {
if !cr.inert(f, excludeNew) {
continue
}
if in.lastSeq == nil {
if !excludeNew {
t.Fatalf("case %d: a new lead is inert while new leads are allowed", i)
}
continue
}
if res.Target != nil || res.WaitUntil != nil || res.Hold != nil {
t.Fatalf("case %d: inert lead routes to target=%v wait=%v hold=%v", i, res.Target, res.WaitUntil, res.Hold)
}
}
}
}
// The floor is the least delay any branch can carry: an instant branch drops
// the target's wait, and a wait node adds its own minutes to every route out.
func TestRoutedPairsDueFloors(t *testing.T) {
a, b, c, end := uuid.New(), uuid.New(), uuid.New(), uuid.New()
instant := true
cr := &campaignRouter{
steps: []routeStep{
{id: a, kind: "email", bc: models.BranchConditions{Branches: []models.Branch{
{TargetSequenceID: &b},
{TargetSequenceID: &c, Conditions: []models.BranchCondition{{Field: "opened", Operator: "ever"}}, Instant: &instant},
}}},
{id: b, kind: "action", waitAfter: 3, waitMinutes: 90, bc: models.BranchConditions{Branches: []models.Branch{{TargetSequenceID: &end}}}},
{id: c, kind: "email", waitAfter: 2, bc: models.BranchConditions{Branches: []models.Branch{{TargetSequenceID: nil}}}},
{id: end, kind: "email", waitAfter: 1},
},
idxByID: map[uuid.UUID]int{a: 0, b: 1, c: 2, end: 3},
}
floors := cr.dueFloors()
if got := floors[a]; got != 0 {
t.Fatalf("floor(a) = %v, want 0 (instant branch)", got)
}
if got, want := floors[b], 24*time.Hour+90*time.Minute; got != want {
t.Fatalf("floor(b) = %v, want %v", got, want)
}
if _, ok := floors[c]; ok {
t.Fatal("a step whose only branch stops has a floor")
}
if _, ok := floors[end]; ok {
t.Fatal("a step with no branches has a floor")
}
if got := cr.endSteps(); len(got) != 1 || got[0] != end {
t.Fatalf("endSteps = %v, want [%v]", got, end)
}
}
internal/scheduler/cold_ramp.go
+8 -3
View File
@@ -16,10 +16,13 @@ func (s *schedulerService) coldRampStates(ctx context.Context, accounts []models
states := make(map[uuid.UUID]repository.ColdRampState, len(accounts))
ids := make([]uuid.UUID, 0, len(accounts))
for _, a := range accounts {
ids = append(ids, a.ID)
states[a.ID] = repository.ColdRampState{WarmupStartedAt: a.Warmup}
// Every reader ignores the state of a mailbox that never warmed.
if a.Warmup != nil {
ids = append(ids, a.ID)
}
}
if s.warmupRepo == nil || len(accounts) == 0 {
if s.warmupRepo == nil || len(ids) == 0 {
return states
}
observed, err := s.warmupRepo.ColdRampStateForAccounts(ctx, ids, time.Now().Add(-warmupramp.LookbackWindow))
@@ -27,7 +30,9 @@ func (s *schedulerService) coldRampStates(ctx context.Context, accounts []models
return states
}
for _, a := range accounts {
states[a.ID] = observed[a.ID].WithKnownWarmup(a.Warmup)
if a.Warmup != nil {
states[a.ID] = observed[a.ID].WithKnownWarmup(a.Warmup)
}
}
return states
}
web/src/hooks/useRealtimeEvents.ts
+42 -14
View File
@@ -7,7 +7,12 @@ import { useUserProfile } from './context/user'
import { markSelfMutation } from '@/lib/realtime/selfActivity'
import { announceCampaignDeleted } from '@/lib/realtime/campaignDeleted'
import { MAILBOX_REMOVAL_KEYS } from '@/lib/api/hooks/app/emails/invalidateAfterMailboxRemoval'
import { createDeliveryDeduper, createRefreshCoalescer } from '@/lib/realtime/refreshCoalescer'
import {
createDeliveryDeduper,
createHiddenTabGate,
createRefreshCoalescer,
type QueryMatch,
} from '@/lib/realtime/refreshCoalescer'
// What an inbox event moves besides its own thread; refreshed through the coalescer, never per event.
const INBOX_AGGREGATES: QueryKey[] = [
@@ -33,13 +38,17 @@ export function useRealtimeEvents() {
const updateDeal = useAppStore((s) => s.updateDeal)
const setSubscription = useAppStore((s) => s.setSubscription)
// Every realtime invalidation goes through this, so a hidden tab refetches once on return instead of per event.
const [gate] = useState(() => createHiddenTabGate(queryClient))
useEffect(() => () => gate.dispose(), [gate])
const invalidate = useCallback(
(queryKeys: QueryKey[]) => {
for (const queryKey of queryKeys) {
void queryClient.invalidateQueries({ queryKey })
gate.invalidate({ queryKey })
}
},
[queryClient],
[gate],
)
// Warmup deliveries arrive all day across a whole pool, so their refreshes
@@ -67,13 +76,16 @@ export function useRealtimeEvents() {
}, [])
// One aggregate refresh per quiet interval; per-thread work once per delivery, not per channel.
const [inboxRefresh] = useState(() => createRefreshCoalescer(queryClient))
const [inboxRefresh] = useState(() => createRefreshCoalescer(queryClient, { invalidate: gate.invalidate }))
const [isDuplicateDelivery] = useState(() => createDeliveryDeduper())
useEffect(() => () => inboxRefresh.dispose(), [inboxRefresh])
// Mailbox events arrive in bursts (a fleet reconnect, a pool re-evaluation),
// and each one refetches the whole mailbox list and every status chunk.
const [accountRefresh] = useState(() => createRefreshCoalescer(queryClient))
const [accountRefresh] = useState(() => createRefreshCoalescer(queryClient, { invalidate: gate.invalidate }))
useEffect(() => () => accountRefresh.dispose(), [accountRefresh])
// One send publishes several campaign events within a second or two; they share one refetch.
const [campaignRefresh] = useState(() => createRefreshCoalescer(queryClient, { invalidate: gate.invalidate }))
useEffect(() => () => campaignRefresh.dispose(), [campaignRefresh])
const handleRealtimeEvent = useCallback(
(payload: Record<string, unknown>) => {
@@ -304,7 +316,7 @@ export function useRealtimeEvents() {
if (campaignId) {
const status = getString('status')
updateCampaign(campaignId, status ? { status } : {})
invalidate([
campaignRefresh.add([
['campaigns', campaignId],
['campaigns', campaignId, 'logs'],
['analytics', 'campaigns', campaignId],
@@ -312,15 +324,23 @@ export function useRealtimeEvents() {
['analytics', 'campaigns', campaignId, 'hourly'],
])
}
invalidate([
campaignRefresh.add([
['campaigns', 'list'],
['analytics'],
])
// Other contacts' campaign-state (a scheduler preview per campaign) is left alone.
void queryClient.invalidateQueries({
queryKey: ['contacts'],
predicate: (q) => !contactId || q.queryKey[2] !== 'campaign-state' || q.queryKey[1] === contactId,
})
// A reply can open a CRM task or deal (reply intents, reply actions), which publish no event of their own.
if (includes('EMAIL_REPLIED')) campaignRefresh.add([['crm', 'tasks'], ['crm', 'deals']])
// Sends and progress record no verification evidence; opens, clicks, replies and bounces do.
const movesVerification = !includes('EMAIL_SENT', 'TASK_PROGRESS')
const contactsMatch: QueryMatch = (q) => {
const [, scope, part] = q.queryKey
// Custom field keys are never written by a send, open, click, reply or bounce.
if (scope === 'custom-fields') return false
if (scope === 'verification') return movesVerification
// Other contacts' campaign-state (a scheduler preview per campaign) is left alone.
return !contactId || part !== 'campaign-state' || scope === contactId
}
campaignRefresh.add([['contacts']], contactsMatch)
return
}
@@ -373,11 +393,17 @@ export function useRealtimeEvents() {
return
}
if (includes('CRM_TASK', 'TASK')) {
if (includes('CRM_TASK')) {
invalidate([['crm', 'tasks'], ['crm', 'deals']])
return
}
// Campaign task status and progress pulse: system-created CRM tasks and deals publish nothing of their own, so refresh coalesced.
if (event.startsWith('TASK_')) {
campaignRefresh.add([['crm', 'tasks'], ['crm', 'deals']])
return
}
if (includes('SUBSCRIPTION', 'PLAN', 'BILLING', 'LIMIT')) {
setSubscription(payload as any)
invalidate([
@@ -598,7 +624,7 @@ export function useRealtimeEvents() {
invalidate([['campaigns', entityId], ['segments', 'campaign', entityId]])
// A campaign write can enrol leads (linking a segment), so only the
// contact lists scoped to that campaign move, not every list.
void queryClient.invalidateQueries({
gate.invalidate({
predicate: (q) => {
const [root, kind, options] = q.queryKey as [unknown, unknown, { campaign_ids?: string[] } | undefined]
return root === 'contacts' && kind === 'list' && !!options?.campaign_ids?.includes(entityId)
@@ -626,7 +652,9 @@ export function useRealtimeEvents() {
[
accountRefresh,
addUniboxEmail,
campaignRefresh,
currentOrg?.id,
gate,
inboxRefresh,
invalidate,
isDuplicateDelivery,
web/src/lib/realtime/refreshCoalescer.test.ts
+110 -2
View File
@@ -1,6 +1,6 @@
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { QueryClient } from "@tanstack/react-query";
import { createDeliveryDeduper, createRefreshCoalescer } from "./refreshCoalescer";
import { QueryClient, QueryObserver } from "@tanstack/react-query";
import { createDeliveryDeduper, createHiddenTabGate, createRefreshCoalescer } from "./refreshCoalescer";
describe("createRefreshCoalescer", () => {
let client: QueryClient;
@@ -82,6 +82,114 @@ describe("createRefreshCoalescer", () => {
});
});
describe("createRefreshCoalescer with a match", () => {
let client: QueryClient;
beforeEach(() => {
vi.useFakeTimers();
client = new QueryClient();
for (const key of [["contacts", "list"], ["contacts", "a", "campaign-state"], ["contacts", "b", "campaign-state"], ["contacts", "c", "campaign-state"]]) {
client.setQueryData(key, 1);
}
});
afterEach(() => {
vi.useRealTimers();
client.clear();
});
const invalidated = () =>
client.getQueryCache().getAll().filter((q) => q.state.isInvalidated).map((q) => q.queryKey.join("/")).sort();
it("refreshes the union of what each narrowed add asked for", () => {
const c = createRefreshCoalescer(client, { delayMs: 10 });
const notOtherState = (id: string) => (q: { queryKey: readonly unknown[] }) => q.queryKey[2] !== "campaign-state" || q.queryKey[1] === id;
c.add([["contacts"]], notOtherState("a"));
c.add([["contacts"]], notOtherState("b"));
vi.advanceTimersByTime(10);
expect(invalidated()).toEqual(["contacts/a/campaign-state", "contacts/b/campaign-state", "contacts/list"]);
});
it("lets an unnarrowed add widen a narrowed one", () => {
const c = createRefreshCoalescer(client, { delayMs: 10 });
c.add([["contacts"]], (q) => q.queryKey[1] === "list");
c.add([["contacts"]]);
vi.advanceTimersByTime(10);
expect(invalidated()).toHaveLength(4);
});
it("does not treat a narrowed broader key as covering a narrower one", () => {
const c = createRefreshCoalescer(client, { delayMs: 10 });
c.add([["contacts"]], (q) => q.queryKey[1] === "list");
c.add([["contacts", "c"]]);
vi.advanceTimersByTime(10);
expect(invalidated()).toEqual(["contacts/c/campaign-state", "contacts/list"]);
});
});
describe("createHiddenTabGate", () => {
let client: QueryClient;
let visibility: DocumentVisibilityState;
const setVisibility = (state: DocumentVisibilityState) => {
visibility = state;
document.dispatchEvent(new Event("visibilitychange"));
};
const observe = (key: string[]) => {
const fn = vi.fn(async () => 1);
const observer = new QueryObserver(client, { queryKey: key, queryFn: fn });
const unsubscribe = observer.subscribe(() => {});
return { fn, unsubscribe };
};
beforeEach(() => {
client = new QueryClient();
visibility = "visible";
vi.spyOn(document, "visibilityState", "get").mockImplementation(() => visibility);
});
afterEach(() => {
vi.restoreAllMocks();
client.clear();
});
it("refetches at once on a visible tab", async () => {
const gate = createHiddenTabGate(client);
const list = observe(["campaigns", "list"]);
await vi.waitFor(() => expect(client.getQueryData(["campaigns", "list"])).toBe(1));
gate.invalidate({ queryKey: ["campaigns"] });
await vi.waitFor(() => expect(list.fn).toHaveBeenCalledTimes(2));
list.unsubscribe();
gate.dispose();
});
it("defers a hidden tab's refetches to one on return, keeping the unread badge live", async () => {
const gate = createHiddenTabGate(client);
const list = observe(["campaigns", "list"]);
const badge = observe(["unibox", "unseen-count"]);
const overview = observe(["unibox", "overview"]);
await vi.waitFor(() => expect(client.isFetching()).toBe(0));
setVisibility("hidden");
for (let i = 0; i < 5; i++) {
gate.invalidate({ queryKey: ["campaigns", "list"] });
gate.invalidate({ queryKey: ["unibox"] });
}
await vi.waitFor(() => expect(badge.fn).toHaveBeenCalledTimes(6));
expect(list.fn).toHaveBeenCalledTimes(1);
expect(overview.fn).toHaveBeenCalledTimes(1);
expect(client.getQueryState(["campaigns", "list"])?.isInvalidated).toBe(true);
setVisibility("visible");
await vi.waitFor(() => expect(list.fn).toHaveBeenCalledTimes(2));
await vi.waitFor(() => expect(overview.fn).toHaveBeenCalledTimes(2));
setVisibility("hidden");
setVisibility("visible");
await Promise.resolve();
expect(list.fn).toHaveBeenCalledTimes(2);
for (const o of [list, badge, overview]) o.unsubscribe();
gate.dispose();
});
});
describe("createDeliveryDeduper", () => {
beforeEach(() => vi.useFakeTimers());
afterEach(() => vi.useRealTimers());
web/src/lib/realtime/refreshCoalescer.ts
+105 -17
View File
@@ -1,10 +1,21 @@
// Bounds realtime refetches: one invalidation per quiet interval (capped), and one handling per delivered event.
import { hashKey, partialMatchKey, type QueryClient, type QueryKey } from "@tanstack/react-query";
import {
hashKey,
partialMatchKey,
type InvalidateOptions,
type InvalidateQueryFilters,
type Query,
type QueryClient,
type QueryKey,
} from "@tanstack/react-query";
export type QueryMatch = (query: Query) => boolean;
export type Invalidate = (filters: InvalidateQueryFilters, options?: InvalidateOptions) => void;
export interface RefreshCoalescer {
/** Queues keys (prefixes) for the next flush. */
add(keys: readonly QueryKey[]): void;
/** Queues keys (prefixes) for the next flush, optionally narrowed to the queries `match` accepts. */
add(keys: readonly QueryKey[], match?: QueryMatch): void;
/** Drops anything pending without invalidating it. */
dispose(): void;
}
@@ -14,18 +25,29 @@ export interface RefreshCoalescerOptions {
delayMs?: number;
/** Longest a key may wait while adds keep arriving. */
maxWaitMs?: number;
/** Runs each flushed invalidation; defaults to the client's own. */
invalidate?: Invalidate;
}
// Kept above the backend's overviewFreshFor (internal/app/unibox/overview_cache.go).
export const REFRESH_DELAY_MS = 1500;
export const REFRESH_MAX_WAIT_MS = 5000;
interface Pending {
key: QueryKey;
match?: QueryMatch;
}
export function createRefreshCoalescer(
queryClient: QueryClient,
{ delayMs = REFRESH_DELAY_MS, maxWaitMs = REFRESH_MAX_WAIT_MS }: RefreshCoalescerOptions = {},
{
delayMs = REFRESH_DELAY_MS,
maxWaitMs = REFRESH_MAX_WAIT_MS,
invalidate = (filters, options) => void queryClient.invalidateQueries(filters, options),
}: RefreshCoalescerOptions = {},
): RefreshCoalescer {
// Each pending key with the time it was last added.
const pending = new Map<string, { key: QueryKey; at: number }>();
const pending = new Map<string, Pending & { at: number }>();
let timer: ReturnType<typeof setTimeout> | null = null;
let firstAt = 0;
@@ -36,36 +58,44 @@ export function createRefreshCoalescer(
timer = setTimeout(flush, Math.max(0, Math.min(delayMs, firstAt + maxWaitMs - now)));
};
const add = (keys: readonly QueryKey[]) => {
if (keys.length === 0) return;
const queue = (entries: readonly Pending[]) => {
if (entries.length === 0) return;
const now = Date.now();
for (const key of keys) pending.set(hashKey(key), { key, at: now });
for (const { key, match } of entries) {
const hash = hashKey(key);
const prev = pending.get(hash);
const prevMatch = prev?.match;
// Two narrowed adds of one key refresh what either asked for; an unnarrowed one refreshes everything.
const merged = !prev ? match : prevMatch && match ? (q: Query) => prevMatch(q) || match(q) : undefined;
pending.set(hash, { key, match: merged, at: now });
}
schedule();
};
const add = (keys: readonly QueryKey[], match?: QueryMatch) => queue(keys.map((key) => ({ key, match })));
const flush = () => {
timer = null;
const now = Date.now();
const entries = [...pending.values()];
pending.clear();
const keys = entries.map((e) => e.key);
// A flush forced by the cap may beat the last event's writes, so recent keys get a trailing flush too.
const again: QueryKey[] = entries.filter((e) => now - e.at < delayMs).map((e) => e.key);
for (const key of keys) {
const again: Pending[] = entries.filter((e) => now - e.at < delayMs).map(({ key, match }) => ({ key, match }));
for (const { key, match } of entries) {
// A broader pending key already covers this one.
if (keys.some((other) => other !== key && other.length < key.length && partialMatchKey(key, other))) {
if (entries.some((o) => o.key !== key && !o.match && o.key.length < key.length && partialMatchKey(key, o.key))) {
continue;
}
// A running fetch finishes rather than restarts, and is re-checked next round since it may predate the event.
for (const query of queryClient.getQueryCache().findAll({ queryKey: key, fetchStatus: "fetching" })) {
again.push(query.queryKey);
for (const query of queryClient.getQueryCache().findAll({ queryKey: key, fetchStatus: "fetching", predicate: match })) {
again.push({ key: query.queryKey });
}
void queryClient.invalidateQueries(
{ queryKey: key, predicate: (query) => query.state.fetchStatus !== "fetching" },
invalidate(
{ queryKey: key, predicate: (query) => query.state.fetchStatus !== "fetching" && (!match || match(query)) },
{ cancelRefetch: false },
);
}
if (again.length > 0) add(again);
queue(again);
};
return {
@@ -78,6 +108,64 @@ export function createRefreshCoalescer(
};
}
// Shown on a hidden tab (its title and favicon badge), so these keep refetching there.
const HIDDEN_TAB_LIVE_KEYS: readonly QueryKey[] = [["unibox", "unseen-count"]];
export interface HiddenTabGate {
/** Invalidates now; on a hidden tab only marks stale, and refetches once the tab is visible again. */
invalidate: Invalidate;
/** Forgets deferred refetches and stops listening. */
dispose(): void;
}
export function createHiddenTabGate(queryClient: QueryClient): HiddenTabGate {
// Hashes of the active queries a hidden tab left stale.
const deferred = new Set<string>();
let listening = false;
const hidden = () => typeof document !== "undefined" && document.visibilityState === "hidden";
const live: QueryMatch = (query) => HIDDEN_TAB_LIVE_KEYS.some((key) => partialMatchKey(query.queryKey, key));
// On document, so it runs before the focus manager's window listener and a focus refetch joins this one.
const onVisibility = () => {
if (hidden()) return;
stop();
if (deferred.size === 0) return;
const hashes = new Set(deferred);
deferred.clear();
void queryClient.refetchQueries({ type: "active", predicate: (query) => hashes.has(query.queryHash) });
};
const stop = () => {
if (!listening) return;
document.removeEventListener("visibilitychange", onVisibility);
listening = false;
};
const invalidate: Invalidate = (filters, options) => {
if (!hidden()) {
void queryClient.invalidateQueries(filters, options);
return;
}
const match = filters.predicate;
const stale: InvalidateQueryFilters = { ...filters, predicate: (q) => !live(q) && (!match || match(q)) };
for (const query of queryClient.getQueryCache().findAll({ ...stale, type: "active" })) deferred.add(query.queryHash);
void queryClient.invalidateQueries({ ...stale, refetchType: "none" });
void queryClient.invalidateQueries({ ...filters, predicate: (q) => live(q) && (!match || match(q)) }, options);
if (!listening) {
document.addEventListener("visibilitychange", onVisibility);
listening = true;
}
};
return {
invalidate,
dispose: () => {
stop();
deferred.clear();
},
};
}
/**
* Returns true when the same delivery was already seen within `windowMs`. The
* key should name the event and its publish (type, timestamp, ids), so a