diff --git a/cmd/consumer/main.go b/cmd/consumer/main.go index 57ba736f..cb4eea6b 100644 --- a/cmd/consumer/main.go +++ b/cmd/consumer/main.go @@ -240,6 +240,7 @@ func main() { EmailAccountErrorRepository: emailAccountErrorRepo, WarmupRepo: warmupRepo, WarmupContentRepo: repository.NewWarmupContentRepository(primaryDB.Pool), + WarmupEngagementRepo: repository.NewWarmupEngagementRepository(primaryDB.Pool), WarmupService: warmupService, WorkerRepo: workerRepo, Publisher: eventsPublisher, @@ -268,6 +269,11 @@ func main() { // Start warmup health evaluation sweep (every hour) go jobsService.StartWarmupHealthSweep(ctx, 1*time.Hour) + // Drains the durable delayed-engagement schedule (read/important/star) so the + // recipient-side dwell survives worker restarts. Short interval keeps the + // effective dwell close to the requested value. + go jobsService.StartWarmupEngagementPoller(ctx, 30*time.Second) + // Start dead worker detection (every 5 minutes) go jobsService.StartDeadWorkerDetection(ctx, 5*time.Minute) diff --git a/internal/app/consumer/event_new_email.go b/internal/app/consumer/event_new_email.go index 890e5720..28d54a0c 100644 --- a/internal/app/consumer/event_new_email.go +++ b/internal/app/consumer/event_new_email.go @@ -2,9 +2,11 @@ package jobs import ( "context" + "encoding/json" "fmt" "slices" "strings" + "time" "github.com/google/uuid" "github.com/rs/zerolog/log" @@ -151,31 +153,67 @@ func (s *JobsService) performWarmupActions(ctx context.Context, e *models.JobEve settings := s.getGenerationSettings(ctx) actions, delaySeconds := engagementPlan(e.Message.EmailID, settings.Engagement) + immediate, delayed := splitEngagementLegs(actions) - action := &models.WarmupEmailAction{ + base := models.WarmupEmailAction{ UserID: e.UserID, EmailID: e.Message.EmailID, GmailID: e.Message.GmailID, UID: e.Message.UID, MailboxUIDValidity: e.Message.Mailbox, - Actions: actions, - DelaySeconds: delaySeconds, } - // Look up the worker ID from the email account + // Resolve the receiving mailbox's worker once. + var workerID *uuid.UUID if s.EmailRepository != nil { - account, xerr := s.EmailRepository.GetByID(ctx, e.Message.EmailID) - if xerr == nil && account != nil && account.WorkerID != nil { - s.Publisher.PublishWarmupAction(ctx, *account.WorkerID, action) - } else if account != nil && account.WorkerID == nil { - // No assigned worker (mid-migration / just-unassigned / assignment - // lag): the warmup mail can't be foldered or engaged with. Log it - // instead of dropping silently so the gap is observable. - log.Warn(). - Str("email_id", e.Message.EmailID.String()). - Msg("Warmup actions skipped: recipient mailbox has no assigned worker") + if account, xerr := s.EmailRepository.GetByID(ctx, e.Message.EmailID); xerr == nil && account != nil { + workerID = account.WorkerID } } + if workerID == nil { + // No assigned worker (mid-migration / just-unassigned / assignment lag): + // the warmup mail can't be foldered or engaged with. Log instead of + // dropping silently so the gap is observable. + log.Warn(). + Str("email_id", e.Message.EmailID.String()). + Msg("Warmup actions skipped: recipient mailbox has no assigned worker") + return + } + + // Immediate, durable leg (folder + spam-rescue): publish to the worker now. + if len(immediate) > 0 { + act := base + act.Actions = immediate + s.Publisher.PublishWarmupAction(ctx, *workerID, &act) + } + + if len(delayed) == 0 { + return + } + + act := base + act.Actions = delayed + + // Delayed leg (read / important / star): with no dwell (or no durable store + // available) publish immediately; otherwise persist it to the durable + // schedule so a worker restart mid-dwell can't drop it. The poller publishes + // it when fire_at passes. + if delaySeconds <= 0 || s.WarmupEngagementRepo == nil { + s.Publisher.PublishWarmupAction(ctx, *workerID, &act) + return + } + + payload, err := json.Marshal(act) + if err != nil { + log.Warn().Err(err).Str("email_id", e.Message.EmailID.String()).Msg("Failed to marshal delayed warmup engagement; publishing immediately") + s.Publisher.PublishWarmupAction(ctx, *workerID, &act) + return + } + fireAt := time.Now().Add(time.Duration(delaySeconds) * time.Second) + if err := s.WarmupEngagementRepo.EnqueuePendingEngagement(ctx, e.Message.EmailID, payload, fireAt); err != nil { + log.Warn().Err(err).Str("email_id", e.Message.EmailID.String()).Msg("Failed to enqueue delayed warmup engagement; publishing immediately") + s.Publisher.PublishWarmupAction(ctx, *workerID, &act) + } } // recipientProviderDomain best-effort resolves a recipient mailbox's provider diff --git a/internal/app/consumer/service.go b/internal/app/consumer/service.go index e6168062..e71d5b5d 100644 --- a/internal/app/consumer/service.go +++ b/internal/app/consumer/service.go @@ -23,6 +23,7 @@ type JobsService struct { EmailAccountErrorRepository repository.EmailAccountErrorRepository WarmupRepo repository.WarmupRepository WarmupContentRepo repository.WarmupContentRepository + WarmupEngagementRepo repository.WarmupEngagementRepository WarmupService warmupapp.Service WorkerRepo repository.WorkerRepository diff --git a/internal/app/consumer/warmup_engagement.go b/internal/app/consumer/warmup_engagement.go index 2ff8b4c6..111183fb 100644 --- a/internal/app/consumer/warmup_engagement.go +++ b/internal/app/consumer/warmup_engagement.go @@ -83,6 +83,23 @@ func engagementPlan(accountID uuid.UUID, e models.WarmupEngagementSettings) (act return actions, delaySeconds } +// splitEngagementLegs separates the reputation-critical, durable actions +// (foldering + spam-rescue, published immediately) from the low-stakes +// engagement-timing signals (read / important / star) that carry the +// recipient-side dwell. The dwell is now applied durably in the control plane +// (a fire_at row drained by the poller), not by an in-process worker timer, so +// a worker restart can no longer drop the delayed leg. +func splitEngagementLegs(actions []string) (immediate, delayed []string) { + for _, a := range actions { + if a == "move_to_warmbly" || a == "remove_from_spam" { + immediate = append(immediate, a) + } else { + delayed = append(delayed, a) + } + } + return immediate, delayed +} + // rollPct rolls a biased percentage chance. The persona bias nudges a given // mailbox consistently above/below the configured rate so mailboxes differ. func rollPct(rate int, bias float64) bool { diff --git a/internal/app/consumer/warmup_engagement_poller.go b/internal/app/consumer/warmup_engagement_poller.go new file mode 100644 index 00000000..503e662e --- /dev/null +++ b/internal/app/consumer/warmup_engagement_poller.go @@ -0,0 +1,66 @@ +package jobs + +import ( + "context" + "encoding/json" + "time" + + "github.com/rs/zerolog/log" + + "github.com/warmbly/warmbly/internal/models" +) + +// StartWarmupEngagementPoller drains due delayed-engagement rows and publishes +// them to the worker. This is the durable replacement for the worker's old +// in-process dwell timer: because the schedule lives in Postgres, a worker (or +// consumer) restart can no longer drop the delayed read/important/star signals. +func (s *JobsService) StartWarmupEngagementPoller(ctx context.Context, interval time.Duration) { + if s.WarmupEngagementRepo == nil || s.Publisher == nil { + return + } + + ticker := time.NewTicker(interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + s.drainDueEngagements(ctx) + } + } +} + +func (s *JobsService) drainDueEngagements(ctx context.Context) { + cctx, cancel := context.WithTimeout(ctx, 2*time.Minute) + defer cancel() + + due, err := s.WarmupEngagementRepo.ClaimDuePendingEngagements(cctx, 200) + if err != nil { + log.Warn().Err(err).Msg("warmup engagement poller: claim failed") + return + } + + for _, p := range due { + var action models.WarmupEmailAction + if err := json.Unmarshal(p.Payload, &action); err != nil { + log.Warn().Err(err).Str("id", p.ID.String()).Msg("warmup engagement poller: bad payload, dropping") + continue + } + + // Re-resolve the worker at fire time so a mid-dwell reassignment routes + // to the current worker (the payload deliberately doesn't bake one in). + if s.EmailRepository == nil { + continue + } + account, xerr := s.EmailRepository.GetByID(cctx, action.EmailID) + if xerr != nil || account == nil || account.WorkerID == nil { + // Mailbox now unassigned — drop (best-effort low-stakes engagement). + continue + } + + action.DelaySeconds = 0 // dwell already elapsed; run immediately + s.Publisher.PublishWarmupAction(cctx, *account.WorkerID, &action) + } +} diff --git a/internal/infrastructure/db/migrations/000008_warmup_pending_engagements.down.sql b/internal/infrastructure/db/migrations/000008_warmup_pending_engagements.down.sql new file mode 100644 index 00000000..5ad6d30b --- /dev/null +++ b/internal/infrastructure/db/migrations/000008_warmup_pending_engagements.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS public.warmup_pending_engagements; diff --git a/internal/infrastructure/db/migrations/000008_warmup_pending_engagements.up.sql b/internal/infrastructure/db/migrations/000008_warmup_pending_engagements.up.sql new file mode 100644 index 00000000..14ffff42 --- /dev/null +++ b/internal/infrastructure/db/migrations/000008_warmup_pending_engagements.up.sql @@ -0,0 +1,23 @@ +-- Durable dwell for delayed warmup engagement actions. +-- +-- The recipient-side "dwell" delay for the low-stakes engagement signals +-- (mark_read / mark_important / star) previously lived only in a worker-process +-- time.AfterFunc timer, so a worker restart mid-dwell dropped those signals with +-- no trace. This table is the durable schedule: the consumer enqueues the +-- delayed leg here with a fire_at, and a consumer-side poller publishes it to +-- the worker when due. The reputation-critical leg (folder + spam-rescue) is +-- still published immediately and is unaffected. +-- +-- Control-plane only: written and drained by the consumer (Postgres-backed), +-- never by the worker. + +CREATE TABLE public.warmup_pending_engagements ( + id uuid DEFAULT gen_random_uuid() NOT NULL, + email_account_id uuid NOT NULL, + payload jsonb NOT NULL, + fire_at timestamp with time zone NOT NULL, + created_at timestamp with time zone DEFAULT now() NOT NULL, + CONSTRAINT warmup_pending_engagements_pkey PRIMARY KEY (id) +); + +CREATE INDEX idx_warmup_pending_engagements_due ON public.warmup_pending_engagements USING btree (fire_at); diff --git a/internal/repository/pg_warmup_engagement.go b/internal/repository/pg_warmup_engagement.go new file mode 100644 index 00000000..ce527a1c --- /dev/null +++ b/internal/repository/pg_warmup_engagement.go @@ -0,0 +1,81 @@ +package repository + +import ( + "context" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" +) + +// PendingEngagement is a delayed warmup engagement action awaiting its dwell. +// Payload is a JSON-encoded models.WarmupEmailAction (the delayed leg, with +// DelaySeconds already consumed — the worker runs it immediately on receipt). +type PendingEngagement struct { + ID uuid.UUID + EmailAccountID uuid.UUID + Payload []byte + FireAt time.Time +} + +// WarmupEngagementRepository is the durable schedule for dwell-delayed warmup +// engagement actions, so a worker restart can't drop them (the old in-process +// timer did). Control-plane only; drained by the consumer-side poller. +type WarmupEngagementRepository interface { + // EnqueuePendingEngagement stores a delayed engagement leg to fire at fireAt. + EnqueuePendingEngagement(ctx context.Context, accountID uuid.UUID, payload []byte, fireAt time.Time) error + // ClaimDuePendingEngagements atomically removes and returns up to limit rows + // whose fire_at has passed, so each is delivered exactly once across pollers. + ClaimDuePendingEngagements(ctx context.Context, limit int) ([]PendingEngagement, error) +} + +type warmupEngagementRepository struct { + db *pgxpool.Pool +} + +// NewWarmupEngagementRepository creates a new warmup engagement repository. +func NewWarmupEngagementRepository(db *pgxpool.Pool) WarmupEngagementRepository { + return &warmupEngagementRepository{db: db} +} + +func (r *warmupEngagementRepository) EnqueuePendingEngagement(ctx context.Context, accountID uuid.UUID, payload []byte, fireAt time.Time) error { + _, err := r.db.Exec(ctx, + `INSERT INTO warmup_pending_engagements (email_account_id, payload, fire_at) VALUES ($1, $2, $3)`, + accountID, payload, fireAt) + return err +} + +// ClaimDuePendingEngagements uses DELETE ... RETURNING over a FOR UPDATE SKIP +// LOCKED subselect so concurrent pollers never claim the same row and a claimed +// row is removed in the same statement (delivered at most once). +func (r *warmupEngagementRepository) ClaimDuePendingEngagements(ctx context.Context, limit int) ([]PendingEngagement, error) { + if limit <= 0 { + limit = 100 + } + query := ` + DELETE FROM warmup_pending_engagements + WHERE id IN ( + SELECT id FROM warmup_pending_engagements + WHERE fire_at <= NOW() + ORDER BY fire_at + LIMIT $1 + FOR UPDATE SKIP LOCKED + ) + RETURNING id, email_account_id, payload, fire_at + ` + rows, err := r.db.Query(ctx, query, limit) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []PendingEngagement + for rows.Next() { + var p PendingEngagement + if err := rows.Scan(&p.ID, &p.EmailAccountID, &p.Payload, &p.FireAt); err != nil { + return nil, err + } + out = append(out, p) + } + return out, rows.Err() +}