/** * Step-runner — invoked by the worker scheduler when a delayed * `email-sequence.step` job wakes up. * * Responsibilities: * 1. Load run + delivery rows. * 2. Skip / cancel based on run status (unsubscribe / cancellation * mid-sequence). * 3. Evaluate per-step conditions against the cached payload. * 4. Render template via injected `templateRenderer` → { html, text }. * 5. Call injected `mailSender` to send the email with the * List-Unsubscribe header. Persist providerMessageId. * 6. Advance the run (next step or completion) via {@link advanceSequenceRun}. * * Storage + queue surfaces are injected so the runner is fully unit-testable. */ import type { SequenceRunRow, SequenceEngineStore, SequenceEngineQueue, SequenceEngineLogger, StepDeliveryStatus } from "./engine.js"; import { advanceSequenceRun } from "./engine.js"; import type { SequenceStep, SequenceTriggerContext } from "./types.js"; import { evaluateAllConditions } from "./conditions.js"; export interface RenderedMail { readonly html: string; readonly text: string; readonly subject: string; } export interface MailSendResult { readonly ok: boolean; readonly messageId?: string; readonly errorTag?: string; readonly bounced?: boolean; } export interface StepRunnerStore extends SequenceEngineStore { loadRun(runId: string): Promise; loadSteps(runId: string): Promise; loadTriggerContext(runId: string): Promise; /** Atomically transition the delivery row to a terminal status. */ markDelivery(args: { runId: string; stepIndex: number; status: StepDeliveryStatus; sentAt?: Date; bouncedAt?: Date; errorMessage?: string; providerMessageId?: string; }): Promise; } export interface StepRunnerDeps { readonly store: StepRunnerStore; readonly queue: SequenceEngineQueue; readonly logger?: SequenceEngineLogger; readonly now?: () => Date; readonly renderTemplate: ( step: SequenceStep, trigger: SequenceTriggerContext, runId: string ) => Promise | RenderedMail; readonly sendMail: (input: { to: string; subject: string; html: string; text: string; unsubscribeUrl: string; listUnsubscribeMailto?: string; }) => Promise; readonly buildUnsubscribeUrl: (trigger: SequenceTriggerContext, runId: string) => string; readonly buildUnsubscribeMailto?: (trigger: SequenceTriggerContext, runId: string) => string | undefined; } export interface StepRunResult { readonly action: "sent" | "skipped" | "failed" | "cancelled"; readonly completed: boolean; } export async function runStep( args: { runId: string; stepIndex: number }, deps: StepRunnerDeps ): Promise { const run = await deps.store.loadRun(args.runId); if (!run) { deps.logger?.warn?.({ op: "sequence.step.run_missing", runId: args.runId }, "sequence run not found"); return { action: "cancelled", completed: true }; } if (run.status === "cancelled" || run.status === "unsubscribed" || run.status === "completed" || run.status === "failed") { deps.logger?.info?.( { op: "sequence.step.run_inactive", runId: args.runId, status: run.status }, "sequence run inactive — skip" ); return { action: "cancelled", completed: true }; } const steps = await deps.store.loadSteps(args.runId); const step = steps.find((s) => s.stepIndex === args.stepIndex); if (!step) { deps.logger?.warn?.( { op: "sequence.step.unknown_step", runId: args.runId, stepIndex: args.stepIndex }, "step index out of range" ); return { action: "cancelled", completed: true }; } const trigger = await deps.store.loadTriggerContext(args.runId); if (!trigger) { deps.logger?.warn?.( { op: "sequence.step.no_trigger", runId: args.runId }, "trigger context missing — skip" ); await deps.store.markDelivery({ runId: args.runId, stepIndex: args.stepIndex, status: "skipped", errorMessage: "trigger_context_missing", }); const adv = await advanceSequenceRun( { run, steps, previousStepIndex: args.stepIndex }, { store: deps.store, queue: deps.queue, logger: deps.logger, now: deps.now } ); return { action: "skipped", completed: adv.completed }; } if (!evaluateAllConditions(trigger.payload, step.conditions)) { deps.logger?.info?.( { op: "sequence.step.conditions_false", runId: args.runId, stepIndex: args.stepIndex }, "step conditions not met — skip" ); await deps.store.markDelivery({ runId: args.runId, stepIndex: args.stepIndex, status: "skipped", }); const adv = await advanceSequenceRun( { run, steps, previousStepIndex: args.stepIndex }, { store: deps.store, queue: deps.queue, logger: deps.logger, now: deps.now } ); return { action: "skipped", completed: adv.completed }; } const rendered = await deps.renderTemplate(step, trigger, args.runId); const unsubscribeUrl = deps.buildUnsubscribeUrl(trigger, args.runId); const listUnsubscribeMailto = deps.buildUnsubscribeMailto?.(trigger, args.runId); const result = await deps.sendMail({ to: trigger.email, subject: rendered.subject, html: rendered.html, text: rendered.text, unsubscribeUrl, listUnsubscribeMailto, }); const sentAt = (deps.now ?? (() => new Date()))(); if (result.ok) { await deps.store.markDelivery({ runId: args.runId, stepIndex: args.stepIndex, status: "sent", sentAt, providerMessageId: result.messageId, }); deps.logger?.info?.( { op: "sequence.step.sent", runId: args.runId, stepIndex: args.stepIndex, providerMessageId: result.messageId }, "sequence step sent" ); const adv = await advanceSequenceRun( { run, steps, previousStepIndex: args.stepIndex }, { store: deps.store, queue: deps.queue, logger: deps.logger, now: deps.now } ); return { action: "sent", completed: adv.completed }; } if (result.bounced) { await deps.store.markDelivery({ runId: args.runId, stepIndex: args.stepIndex, status: "bounced", bouncedAt: sentAt, errorMessage: result.errorTag, }); await deps.store.updateRunStatus(args.runId, "failed", sentAt); deps.logger?.warn?.( { op: "sequence.step.bounced", runId: args.runId, stepIndex: args.stepIndex, errorTag: result.errorTag }, "sequence step bounced — run failed" ); return { action: "failed", completed: true }; } await deps.store.markDelivery({ runId: args.runId, stepIndex: args.stepIndex, status: "failed", errorMessage: result.errorTag, }); deps.logger?.error( { op: "sequence.step.failed", runId: args.runId, stepIndex: args.stepIndex, errorTag: result.errorTag }, "sequence step send failed" ); return { action: "failed", completed: false }; }