Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 | 1x 1x 1x 1x 16x 3x 7x 4x 4x 4x 11x 2x 2x 1x 1x 2x 22x 22x 22x 22x 22x 22x 1x 1x 21x 20x 20x 20x 1x 1x 19x 19x 1x 1x 18x 18x 18x 1x 1x 17x 54x 17x 1x 1x 16x 16x 16x 16x 1x 1x 15x 15x 3x 3x 1x 1x 1x 2x 12x 3x 3x 3x 2x 3x 9x 3x 3x 5x 5x 5x 1x 1x 1x 22x 22x 22x 22x 22x 20x 2x 2x | /**
* The RRM automated ladder — send queue consumer
* (`docs/operations/rrm-ladder-design.md` §7).
*
* Per message: load the step, the run and the prospect, call the gateway,
* then classify the result into one of three outcome classes (§7's table).
* Idempotent on the step's own state — a step no longer `enqueued` when this
* runs (already `sent`/`failed`/`cancelled` by an earlier delivery of the
* same message, or picked up by the 30-minute stuck-`enqueued` sweep) is a
* no-op, never a re-send.
*
* `services/rrm/gateway.service.ts` (the ONLY send path — see its own header)
* owns `message_sent` / `send_blocked`, `last_contacted_at` and the
* `queued → contacted` stage move; this file never duplicates them. It
* appends its OWN `message_failed` event (payload carries `stepKey`/`runId`,
* which the gateway's events do not) whenever a step permanently fails, and
* ALWAYS appends one — before any retry/ack — when the failure looks like a
* `368`/`131031` account-restriction code, because that is the one signal
* the scheduler's auto-halt (§6.1) cannot compute from `rrm_scheduled_steps`
* alone.
*
* That code shows up in exactly one place: the gateway's own send attempt
* (its checklist step 9) THROWS on a Graph/network failure rather than
* returning `{ok:false}` — it has already logged its own detail-free
* `message_failed` by the time it does. This file therefore wraps ONLY the
* `sendToProspect` call in its own try/catch, inspects the thrown error's
* message for the restriction pattern, and only then either fails the step
* permanently (a restriction will not clear on retry) or re-throws so the
* queue's normal retry/DLQ policy applies. A returned `{ok:false, decision:
* "error"}` is a different thing entirely — a caller-contract problem inside
* the gateway itself (missing prospect / text / template name) — and is
* handled as a permanent failure in the switch below, never retried.
*/
import { and, eq } from "drizzle-orm";
import type { DrizzleD1Database } from "drizzle-orm/d1";
import {
RRM_CONFIG_DEFAULTS,
RRM_CONFIG_KEYS,
RrmConfigDal,
} from "../../dal/rrm/config.dal";
import { RrmEventsDal } from "../../dal/rrm/events.dal";
import { RrmProspectsDal } from "../../dal/rrm/prospects.dal";
import { getDb } from "../../db";
import * as schema from "../../db/schema";
import type { RrmScheduledStep, RrmSequenceRun } from "../../db/schema/rrm";
import { logger } from "../../lib/logger";
import { sendToProspect } from "../../services/rrm/gateway.service";
import {
countInFlightOrSent,
loadWindowConfig,
templateKeysOf,
} from "../../services/rrm/scheduler.service";
import { IST_OFFSET_MS, isWithinSendWindow, nextWindowStart } from "./ist";
import { parseSteps, renderVariables, type StepDef } from "./sequence";
const RUNS = schema.rrmSequenceRuns;
const STEPS = schema.rrmScheduledSteps;
const SEQUENCES = schema.rrmSequences;
const DAY_MS = 86_400_000;
/** Epoch ms of 00:00 IST on the IST calendar day containing `ms` — same
* formula as `scheduler.service.ts`'s private helper of the same purpose. */
function istDayStartMs(ms: number): number {
return Math.floor((ms + IST_OFFSET_MS) / DAY_MS) * DAY_MS - IST_OFFSET_MS;
}
export type RrmSendQueueMessage = { runId: string; stepKey: string };
type ConsumerCtx = {
db: DrizzleD1Database<typeof schema>;
env: CloudflareBindings;
config: RrmConfigDal;
events: RrmEventsDal;
prospects: RrmProspectsDal;
};
/** `368` (temporarily restricted) / `131031` (account restricted) — the
* login-outage guards L1 depends on (design §6.1). */
function isAccountRestriction(detail: string | undefined): boolean {
return typeof detail === "string" && /368|131031/.test(detail);
}
async function recordMessageFailed(
events: RrmEventsDal,
prospectId: string,
step: RrmScheduledStep,
decision: string,
detail: string | undefined,
now: Date,
): Promise<void> {
await events.append({
prospectId,
type: "message_failed",
actorType: "system",
payload: { stepKey: step.stepKey, runId: step.runId, decision, detail },
occurredAt: now,
});
}
async function reschedule(
db: DrizzleD1Database<typeof schema>,
events: RrmEventsDal,
step: RrmScheduledStep,
config: RrmConfigDal,
prospectId: string,
now: Date,
): Promise<void> {
const windowConfig = await loadWindowConfig(config);
const dueAt = isWithinSendWindow(
now.getTime(),
windowConfig.startHour,
windowConfig.endHour,
windowConfig.holidays,
)
? now
: new Date(
nextWindowStart(
now.getTime(),
windowConfig.startHour,
windowConfig.endHour,
windowConfig.holidays,
0,
),
);
await db.batch([
db
.update(STEPS)
.set({ state: "pending", enqueuedAt: null, dueAt })
.where(eq(STEPS.id, step.id)),
events.buildEventStatement({
prospectId,
type: "requeued",
actorType: "system",
payload: { stepKey: step.stepKey, runId: step.runId },
occurredAt: now,
}),
]);
}
async function markPermanentFailure(
db: DrizzleD1Database<typeof schema>,
step: RrmScheduledStep,
state: "failed" | "skipped",
decision: string,
): Promise<void> {
await db
.update(STEPS)
.set({ state, cancelReason: decision })
.where(eq(STEPS.id, step.id));
}
/** Applies N7's `onComplete` and closes the run. Only ever reached on a
* *successful* final send — see the header comment and the sweep in
* `scheduler.service.ts` for the failure-path safety net. */
async function applyOnCompleteAndCloseRun(
prospects: RrmProspectsDal,
config: RrmConfigDal,
db: DrizzleD1Database<typeof schema>,
run: RrmSequenceRun,
now: Date,
): Promise<void> {
const staged = await prospects.setStage(run.prospectId, "not_now", {
actorType: "system",
reason: "ladder_completed_no_reply",
});
if (staged.changed) {
// `setStage` already stamped `snooze_until` at its own hardcoded
// 60-day default (`NOT_NOW_SNOOZE_DAYS`, `prospects.dal.ts`). Overwrite
// it with the CONFIGURED value so raising `snooze_days` on Ramp &
// pacing takes effect without a code change.
const snoozeDays = await config.getNumber(
RRM_CONFIG_KEYS.snoozeDays,
RRM_CONFIG_DEFAULTS[RRM_CONFIG_KEYS.snoozeDays],
);
await prospects.snooze(run.prospectId, snoozeDays, {
actorType: "system",
reason: "ladder_completed_no_reply",
});
}
// `changed:false` means DNC (or some other terminal state) raced this
// send — leave the snooze alone, but the run still closes below.
await db
.update(RUNS)
.set({ state: "completed", exitReason: "completed", endedAt: now })
.where(eq(RUNS.id, run.id));
}
async function handleOne(
ctx: ConsumerCtx,
body: RrmSendQueueMessage,
): Promise<void> {
const { db, env, config, events, prospects } = ctx;
const now = new Date();
const { runId, stepKey } = body;
const stepRows = await db
.select()
.from(STEPS)
.where(and(eq(STEPS.runId, runId), eq(STEPS.stepKey, stepKey)))
.limit(1);
const step = stepRows[0];
if (!step) {
logger.error(
`[RRM send-consumer] no scheduled step for run ${runId}/${stepKey}`,
);
return;
}
// Idempotency: a step no longer `enqueued` was already handled by an
// earlier delivery of this message, or reclaimed by the stuck-enqueued
// sweep. Either way, sending again would be a duplicate template.
if (step.state !== "enqueued") return;
const runRows = await db
.select()
.from(RUNS)
.where(eq(RUNS.id, runId))
.limit(1);
const run = runRows[0];
if (!run) {
await markPermanentFailure(db, step, "failed", "run_not_found");
return;
}
const prospect = await prospects.findById(run.prospectId);
if (!prospect) {
await markPermanentFailure(db, step, "failed", "prospect_not_found");
return;
}
const seqRows = await db
.select()
.from(SEQUENCES)
.where(eq(SEQUENCES.id, run.sequenceId))
.limit(1);
const seqRow = seqRows[0];
if (!seqRow) {
await markPermanentFailure(db, step, "failed", "sequence_not_found");
return;
}
// `parseSteps` throws on a corrupt sequence — let it propagate to the
// outer per-message catch, which retries (a bad sequence definition is a
// real operational problem worth surfacing loudly, not swallowing here).
const steps: StepDef[] = parseSteps(seqRow);
const def = steps.find((s) => s.key === stepKey);
if (!def || def.kind !== "template") {
await markPermanentFailure(db, step, "failed", "not_a_template_step");
return;
}
// The gateway's own cap check (design §5 point 5) is the PER-PROSPECT
// frequency cap. The GLOBAL daily cap is this pipeline's concept, so it
// has to be re-checked here — the scheduler verified it at enqueue time,
// but other steps may have been sent in the minutes since.
const dailyCap = await config.getNumber(
RRM_CONFIG_KEYS.dailySendCap,
RRM_CONFIG_DEFAULTS[RRM_CONFIG_KEYS.dailySendCap],
);
const templateKeys = templateKeysOf(steps);
const sentToday = await countInFlightOrSent(
db,
templateKeys,
new Date(istDayStartMs(now.getTime())),
);
if (sentToday >= dailyCap) {
await reschedule(db, events, step, config, prospect.id, now);
return;
}
// The gateway's own send attempt (step 9 of its checklist) THROWS on a
// Graph/network failure — it has already appended its own `message_failed`
// event by the time it does (gateway.service.ts), but that event carries
// no error detail. A `368`/`131031` account-restriction code — the one
// signal the auto-halt (§6.1) cannot compute any other way — is only
// readable from the thrown error's own message, so it is caught here,
// specifically, rather than left to the outer per-message catch.
let result: Awaited<ReturnType<typeof sendToProspect>>;
try {
result = await sendToProspect(
{ db, env, now },
{
prospectId: prospect.id,
intent: "template",
templateName: def.template,
variables: renderVariables(def, prospect),
stepKey,
runId,
actor: { type: "system" },
},
);
} catch (err) {
const detail = err instanceof Error ? err.message : String(err);
if (isAccountRestriction(detail)) {
// Always on the record — and never retried: a restriction code will
// not clear itself on attempt #2.
await recordMessageFailed(
events,
prospect.id,
step,
"error",
detail,
now,
);
await markPermanentFailure(db, step, "failed", "error");
return;
}
// Graph 5xx/429/network-shaped: let the queue retry (max 3 → DLQ).
throw err;
}
if (result.ok) {
await db.update(STEPS).set({ state: "sent" }).where(eq(STEPS.id, step.id));
await db
.update(RUNS)
.set({ currentStep: stepKey })
.where(eq(RUNS.id, run.id));
if (def.onComplete) {
await applyOnCompleteAndCloseRun(prospects, config, db, run, now);
}
return;
}
switch (result.decision) {
case "quiet_hours":
case "halted":
// The gateway already wrote its own `send_blocked` event for this
// decision — this module's `requeued` event is the step-level
// bookkeeping the scheduler's sweep also writes on its own
// stuck-`enqueued` path, kept consistent here.
await reschedule(db, events, step, config, prospect.id, now);
return;
case "suppressed":
case "dnc":
case "no_consent":
case "window_closed":
case "not_configured":
case "error":
// `no_consent` belongs here, not with the retryable pacing outcomes:
// nothing the scheduler does on its own turns `none` into a lawful
// basis. Only an operator recording one through Lane E does, and
// that re-enters the ladder from the top.
//
// It was missing entirely, and the omission was expensive because
// this switch has no `default`. A consent-blocked step fell straight
// out of `handleOne`, was acked, and stayed `enqueued`; the
// stuck-enqueued sweep then returned it to `pending`, the next tick
// re-enqueued it, and the gateway blocked it again — forever. Each
// lap spent daily-cap and hourly budget without sending anything and
// wrote another `send_blocked` event, which latches the ramp's
// blocks-since-release condition shut so the NEXT release is refused
// with no visible cause.
//
// `error` here is a caller-contract decision from the gateway itself
// (prospect not found / missing text or template name) — never a
// Graph failure, those are thrown (handled above) — so retrying
// cannot help.
await recordMessageFailed(
events,
prospect.id,
step,
result.decision,
result.detail,
now,
);
await markPermanentFailure(db, step, "failed", result.decision);
return;
case "frequency":
// Per-prospect frequency/lifetime cap — this step will never become
// sendable again on its own, but it is a pacing outcome, not a
// delivery failure, so the step is `skipped` rather than `failed`.
await recordMessageFailed(
events,
prospect.id,
step,
result.decision,
result.detail,
now,
);
await markPermanentFailure(db, step, "skipped", result.decision);
return;
}
}
export async function handleRrmSendQueue(
batch: MessageBatch<RrmSendQueueMessage>,
env: CloudflareBindings,
): Promise<void> {
const db = getDb(env.DB);
const ctx: ConsumerCtx = {
db,
env,
config: new RrmConfigDal(db),
events: new RrmEventsDal(db),
prospects: new RrmProspectsDal(db),
};
for (const message of batch.messages) {
try {
await handleOne(ctx, message.body);
message.ack();
} catch (err) {
logger.error("[RRM send-consumer] processing error, retrying:", err);
message.retry();
}
}
}
|