From f364e18b785aefa53fa50477b4ebf496d6fb1fab Mon Sep 17 00:00:00 2001 From: Adnaan Iqbal <64221968+artificialadnaan@users.noreply.github.com> Date: Thu, 2 Jul 2026 17:54:24 -0500 Subject: [PATCH 1/8] feat(rfp): 24h pending-RFP SLA breach email to leadership MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A triggered RFP can sit awaiting approval indefinitely with no nudge. Add an hourly worker scan that emails the RFP reviewers (leadership) once when an RFP has been pending approval for more than 24 hours. - worker/src/jobs/rfp-pending-sla.ts: scans every active office for opportunity, non-bid-board-owned deals with rfp_approval_status in (pending_outbox, pending) whose rfp_approval_requested_at is older than 24h (excludes on-hold + test-data). Emails resolveRfpReviewerEmails() — the SAME leadership set as the RFP-decline email — with a branded, Outlook-safe template (View Deal + Pending RFP queue, officeId-scoped links so cross-office reviewers don't 404). - Exactly-once + retry-safe: a claim row in public.rfp_pending_sla_email_receipts (keyed tenant_schema, deal_id, rfp_approval_requested_at) is INSERTed ON CONFLICT DO NOTHING before the send; a conflict skips (already alerted this cycle), a send failure DELETEs the claim so the next hourly scan retries. A re-triggered RFP gets a new requested_at -> a fresh cycle -> a new alert. Migration 0174 adds the ledger. - Gated behind RFP_PENDING_SLA_ENABLED (default OFF) so it merges inert; leadership opts in deliberately (flipping it on alerts on all currently-breaching RFPs once). - Hourly cron in worker/src/index.ts. Tests (runtime, CI-gated): buildRfpPendingSlaEmail (links/escaping/SLA copy), isRfpPendingSlaEnabled, findPendingRfpSlaBreaches on real SQL (PGlite — filters awaiting/>24h/opportunity/non-BBO/active/non-test/non-hold, oldest first), and the scan orchestration (inert-when-off, no-op-without-recipients, claim-then-send-once, skip-on-conflict, delete-on-send-failure). Added the shared/lib/rfpReviewerEmails vitest alias the worker was missing. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../0174_rfp_pending_sla_email_receipts.sql | 33 ++ worker/src/index.ts | 14 + worker/src/jobs/rfp-pending-sla.ts | 356 ++++++++++++++++++ .../jobs/rfp-pending-sla.runtime.test.ts | 205 ++++++++++ worker/vitest.config.ts | 3 + 5 files changed, 611 insertions(+) create mode 100644 migrations/0174_rfp_pending_sla_email_receipts.sql create mode 100644 worker/src/jobs/rfp-pending-sla.ts create mode 100644 worker/tests/jobs/rfp-pending-sla.runtime.test.ts diff --git a/migrations/0174_rfp_pending_sla_email_receipts.sql b/migrations/0174_rfp_pending_sla_email_receipts.sql new file mode 100644 index 000000000..13653cb60 --- /dev/null +++ b/migrations/0174_rfp_pending_sla_email_receipts.sql @@ -0,0 +1,33 @@ +-- 0174_rfp_pending_sla_email_receipts.sql +-- +-- Exactly-once ledger for the Pending RFP 24h SLA breach email (worker/src/jobs/rfp-pending-sla.ts). +-- +-- WHY: an RFP that stays in the "Pending RFP" bucket (stage=opportunity, not bid-board-owned, +-- rfp_approval_status in pending_outbox/pending) longer than 24h should alert leadership. Unlike the +-- decline email (0148), an SLA breach is TIME-based, not a status transition, so no DB trigger fires it — +-- the worker runs an hourly cron scan instead. Scanning hourly means the same breach is seen many times, so +-- this ledger is the exactly-once guard: the scan INSERTs a claim row ON CONFLICT DO NOTHING BEFORE sending, +-- and a send failure DELETEs the claim so the next scan retries. +-- +-- CYCLE IDENTITY: keyed (tenant_schema, deal_id, rfp_approval_requested_at). rfp_approval_requested_at is +-- the instant the RFP entered pending; a re-triggered RFP gets a NEW requested_at -> a fresh cycle -> a new +-- alert (correct). This is a PUBLIC table (one row per office+deal+cycle), same shape as 0148's +-- rfp_rejection_email_receipts; no per-tenant copy is needed. +-- +-- This ships with the WORKER (the hourly job) and is gated behind RFP_PENDING_SLA_ENABLED so it merges inert. + +CREATE TABLE IF NOT EXISTS public.rfp_pending_sla_email_receipts ( + tenant_schema text NOT NULL, + deal_id uuid NOT NULL, + rfp_approval_requested_at timestamptz NOT NULL, + deal_number text, + recipient_emails text, + resend_message_id text, + sent_at timestamptz NOT NULL DEFAULT NOW(), + created_at timestamptz NOT NULL DEFAULT NOW(), + updated_at timestamptz NOT NULL DEFAULT NOW(), + PRIMARY KEY (tenant_schema, deal_id, rfp_approval_requested_at) +); + +CREATE INDEX IF NOT EXISTS rfp_pending_sla_email_receipts_deal_idx + ON public.rfp_pending_sla_email_receipts (tenant_schema, deal_id); diff --git a/worker/src/index.ts b/worker/src/index.ts index 2e674054f..252bc65f8 100644 --- a/worker/src/index.ts +++ b/worker/src/index.ts @@ -20,6 +20,7 @@ import { runAiDisconnectDigest } from "./jobs/ai-disconnect-digest.js"; import { runAiDisconnectEscalationScan } from "./jobs/ai-disconnect-escalation.js"; import { runAiDisconnectAdminTaskGeneration } from "./jobs/ai-disconnect-admin-tasks.js"; import { runAiInterventionManagerAlerts } from "./jobs/ai-intervention-manager-alerts.js"; +import { runRfpPendingSlaScan } from "./jobs/rfp-pending-sla.js"; import { runCallRecordingCleanup } from "./jobs/call-recording-cleanup.js"; import { runCallRecordingTranscription } from "./jobs/call-recording-transcribe.js"; import { runRfpRequestDeadLetterSweep } from "./jobs/rfp-request-delivery.js"; @@ -268,6 +269,19 @@ async function main() { }, { timezone: "America/Chicago" }); console.log("[Worker] Cron scheduled: rep performance rollup at 4:00 AM CT daily"); + // Pending RFP 24h SLA: hourly scan that emails leadership once per pending cycle when an RFP has been + // awaiting approval > 24h. Inert until RFP_PENDING_SLA_ENABLED=true (the job self-gates); a receipts + // ledger keeps it exactly-once, so scanning hourly never double-sends. + cron.schedule("0 * * * *", async () => { + console.log("[Worker:cron] Running pending RFP SLA scan..."); + try { + await runRfpPendingSlaScan(); + } catch (err) { + console.error("[Worker:cron] Pending RFP SLA scan failed:", err); + } + }, { timezone: "America/Chicago" }); + console.log("[Worker] Cron scheduled: pending RFP SLA scan hourly"); + // Manager alerts: evaluate every 5 minutes and let office-local due gating decide when to send. cron.schedule("*/5 * * * *", async () => { console.log("[Worker:cron] Running intervention manager alerts..."); diff --git a/worker/src/jobs/rfp-pending-sla.ts b/worker/src/jobs/rfp-pending-sla.ts new file mode 100644 index 000000000..e6c2ec8a9 --- /dev/null +++ b/worker/src/jobs/rfp-pending-sla.ts @@ -0,0 +1,356 @@ +import { pool } from "../db.js"; +import { sendSystemEmailWithMetadata, type SendSystemEmailResult } from "../lib/system-email.js"; +// Reuse #611's frontend-URL + branded-template primitives (trockcrm.com base + hosted logo), same as the +// RFP-decline email, so this SLA email matches the look and links to the right host. +import { resolveFrontendUrl, TROCK_LOGO_EMAIL_URL } from "./project-number-email.js"; +// Leadership recipients (Takashi + Adam Shaw) — the SAME source of truth the RFP-decline email and the +// server override-review gate use (RFP_REJECTION_EMAIL_RECIPIENTS), so the notified set never drifts. +import { resolveRfpReviewerEmails } from "@trock-crm/shared/lib/rfpReviewerEmails"; +import { escapeHtml } from "../lib/email-format.js"; + +/** The Pending RFP SLA: an RFP awaiting approval longer than this many hours is a breach. */ +export const RFP_PENDING_SLA_HOURS = 24; + +const OFFICE_SLUG_REGEX = /^[a-z][a-z0-9_]*$/; +const TENANT_SCHEMA_REGEX = /^office_[a-z][a-z0-9_]*$/; + +// A minimal query shape so the job is injectable in tests (pool.query, a PGlite query, or a mock all fit). +type PgQuery = (text: string, params?: unknown[]) => Promise<{ rows: any[] }>; + +type SendEmail = ( + to: string | string[], + subject: string, + html: string, + options: { text: string; idempotencyKey: string }, +) => Promise; + +interface RunDeps { + query?: PgQuery; + sendEmail?: SendEmail; + env?: NodeJS.ProcessEnv; + logger?: Pick; + slaHours?: number; +} + +export interface RfpPendingSlaBreach { + dealId: string; + dealName: string; + dealNumber: string | null; + /** ISO-8601 instant the RFP entered pending (deals.rfp_approval_requested_at) — the SLA clock + cycle key. */ + requestedAt: string; + hoursPending: number; +} + +export interface RfpPendingSlaScanSummary { + scanned: number; + sent: number; + skipped: number; + failed: number; +} + +/** Gate the scan behind an explicit flag (default OFF) so it merges inert and leadership opts in deliberately. */ +export function isRfpPendingSlaEnabled(env: NodeJS.ProcessEnv): boolean { + return String(env.RFP_PENDING_SLA_ENABLED ?? "").trim().toLowerCase() === "true"; +} + +/** + * Deals whose RFP has been awaiting approval (pending_outbox/pending) longer than `slaHours`, in one office + * schema. Mirrors the pending-RFP bucket predicate (opportunity, not bid-board-owned, awaiting status) and + * adds the 24h clock off deals.rfp_approval_requested_at (indexed). Excludes on-hold + test-data deals so a + * paused/demo deal never alerts leadership. Oldest first. + */ +export async function findPendingRfpSlaBreaches( + query: PgQuery, + schemaName: string, + slaHours: number, +): Promise { + if (!TENANT_SCHEMA_REGEX.test(schemaName)) { + throw new Error(`Unsafe tenant schema: ${schemaName}`); + } + const result = await query( + `SELECT d.id, + d.name, + COALESCE(d.project_number, d.deal_number) AS deal_number, + d.rfp_approval_requested_at, + EXTRACT(EPOCH FROM (NOW() - d.rfp_approval_requested_at)) / 3600.0 AS hours_pending + FROM ${schemaName}.deals d + JOIN public.pipeline_stage_config psc ON psc.id = d.stage_id + WHERE psc.slug = 'opportunity' + AND d.is_bid_board_owned = false + AND d.rfp_approval_status IN ('pending_outbox', 'pending') + AND d.rfp_approval_requested_at IS NOT NULL + AND d.rfp_approval_requested_at < NOW() - ($1 || ' hours')::interval + AND d.is_active = true + AND COALESCE(d.is_test_data, false) = false + AND COALESCE(d.on_hold, false) = false + ORDER BY d.rfp_approval_requested_at ASC`, + [String(slaHours)], + ); + return result.rows.map((row) => ({ + dealId: String(row.id), + dealName: typeof row.name === "string" && row.name.trim() ? row.name : "Deal", + dealNumber: row.deal_number ?? null, + requestedAt: new Date(row.rfp_approval_requested_at).toISOString(), + hoursPending: Number(row.hours_pending), + })); +} + +/** + * Scan every active office for RFPs breaching the 24h pending SLA and email leadership once per pending + * cycle. Exactly-once + retry-safe: a claim row in public.rfp_pending_sla_email_receipts (keyed + * tenant_schema, deal_id, rfp_approval_requested_at) is inserted ON CONFLICT DO NOTHING BEFORE the send — + * a conflict means this cycle already alerted (skip), and a send failure DELETEs the claim so the next scan + * retries. A re-triggered RFP gets a fresh rfp_approval_requested_at -> a new cycle -> a new alert. + */ +export async function runRfpPendingSlaScan(deps: RunDeps = {}): Promise { + const env = deps.env ?? process.env; + const logger = deps.logger ?? console; + const slaHours = deps.slaHours ?? RFP_PENDING_SLA_HOURS; + const summary: RfpPendingSlaScanSummary = { scanned: 0, sent: 0, skipped: 0, failed: 0 }; + + if (!isRfpPendingSlaEnabled(env)) { + logger.log("[RfpPendingSla] Disabled (RFP_PENDING_SLA_ENABLED != 'true') - skipping scan"); + return summary; + } + const recipients = resolveRfpReviewerEmails(env); + if (recipients.length === 0) { + logger.error( + "[RfpPendingSla] No leadership recipients configured (RFP_REJECTION_EMAIL_RECIPIENTS) - cannot send; skipping", + ); + return summary; + } + + const query = deps.query ?? (pool.query.bind(pool) as PgQuery); + const sendEmail = deps.sendEmail ?? sendSystemEmailWithMetadata; + const frontendUrl = resolveFrontendUrl(env); + + const offices = await query(`SELECT id, slug FROM public.offices WHERE is_active = true`); + for (const office of offices.rows) { + if (!OFFICE_SLUG_REGEX.test(String(office.slug ?? ""))) { + logger.error(`[RfpPendingSla] Invalid office slug "${office.slug}" - skipping`); + continue; + } + const tenantSchema = `office_${office.slug}`; + let breaches: RfpPendingSlaBreach[]; + try { + breaches = await findPendingRfpSlaBreaches(query, tenantSchema, slaHours); + } catch (err) { + logger.error("[RfpPendingSla] Breach scan failed for office - skipping", { tenantSchema, err }); + continue; + } + summary.scanned += breaches.length; + for (const breach of breaches) { + const outcome = await processBreach(query, sendEmail, logger, { + tenantSchema, + officeId: (office.id as string | null) ?? null, + breach, + recipients, + frontendUrl, + slaHours, + }); + summary[outcome] += 1; + } + } + logger.log("[RfpPendingSla] Scan complete", summary); + return summary; +} + +async function processBreach( + query: PgQuery, + sendEmail: SendEmail, + logger: Pick, + args: { + tenantSchema: string; + officeId: string | null; + breach: RfpPendingSlaBreach; + recipients: string[]; + frontendUrl: string; + slaHours: number; + }, +): Promise<"sent" | "skipped" | "failed"> { + const { tenantSchema, officeId, breach, recipients, frontendUrl, slaHours } = args; + + // Claim this cycle atomically; a conflict means we already alerted for this pending instant. + const claim = await query( + `INSERT INTO public.rfp_pending_sla_email_receipts + (tenant_schema, deal_id, rfp_approval_requested_at, deal_number, recipient_emails, sent_at, created_at, updated_at) + VALUES ($1, $2::uuid, $3, $4, $5, NOW(), NOW(), NOW()) + ON CONFLICT (tenant_schema, deal_id, rfp_approval_requested_at) DO NOTHING + RETURNING deal_id`, + [tenantSchema, breach.dealId, breach.requestedAt, breach.dealNumber, recipients.join(", ")], + ); + if (claim.rows.length === 0) return "skipped"; + + try { + const email = buildRfpPendingSlaEmail({ + dealId: breach.dealId, + dealName: breach.dealName, + dealNumber: breach.dealNumber, + pendingSinceLabel: formatPendingSince(breach.requestedAt), + hoursPending: breach.hoursPending, + slaHours, + officeId, + frontendUrl, + }); + const result = await sendEmail(recipients, email.subject, email.html, { + text: email.text, + idempotencyKey: `rfp-pending-sla-${tenantSchema}-${breach.dealId}-${breach.requestedAt}`, + }); + if (!result.success) throw new Error("Email provider returned unsuccessful result"); + await query( + `UPDATE public.rfp_pending_sla_email_receipts + SET resend_message_id = $4, updated_at = NOW() + WHERE tenant_schema = $1 AND deal_id = $2::uuid AND rfp_approval_requested_at = $3`, + [tenantSchema, breach.dealId, breach.requestedAt, result.messageId], + ); + logger.log("[RfpPendingSla] Sent SLA breach alert", { + tenantSchema, + dealId: breach.dealId, + hoursPending: Math.floor(breach.hoursPending), + messageId: result.messageId, + }); + return "sent"; + } catch (err) { + // Roll back the claim so a transient failure re-sends on the next scan (the claim must not permanently + // suppress the alert). One deal's failure must not abort the rest of the scan, so we swallow + log here. + await query( + `DELETE FROM public.rfp_pending_sla_email_receipts + WHERE tenant_schema = $1 AND deal_id = $2::uuid AND rfp_approval_requested_at = $3`, + [tenantSchema, breach.dealId, breach.requestedAt], + ).catch(() => undefined); + logger.error("[RfpPendingSla] Failed to send SLA breach alert", { + tenantSchema, + dealId: breach.dealId, + err, + }); + return "failed"; + } +} + +function formatPendingSince(iso: string): string { + const label = new Date(iso).toLocaleString("en-US", { + timeZone: "America/Chicago", + month: "short", + day: "numeric", + year: "numeric", + hour: "numeric", + minute: "2-digit", + }); + return `${label} CT`; +} + +export function buildRfpPendingSlaEmail(input: { + dealId: string; + dealName: string; + dealNumber: string | null; + pendingSinceLabel: string; + hoursPending: number; + slaHours: number; + officeId?: string | null; + frontendUrl: string; +}) { + // officeId carries the deal's tenant so a cross-office reviewer (leadership often is) doesn't 404; mirrors + // the #611 project-number/decline link builder. Append ONLY officeId. + const officeParam = input.officeId ? `?officeId=${encodeURIComponent(input.officeId)}` : ""; + const baseUrl = input.frontendUrl.replace(/\/+$/, ""); + const dealUrl = `${baseUrl}/deals/${encodeURIComponent(input.dealId)}${officeParam}`; + const queueUrl = `${baseUrl}/deals/pending-rfp${officeParam}`; + const safeDealUrl = escapeHtml(dealUrl); + const safeQueueUrl = escapeHtml(queueUrl); + const hours = Math.floor(input.hoursPending); + + const subject = input.dealNumber + ? `RFP pending approval ${hours}h: ${input.dealNumber} (${input.dealName})` + : `RFP pending approval ${hours}h: ${input.dealName}`; + + const rows = [ + ["Deal name", input.dealName], + ["Project number", input.dealNumber ?? "Pending"], + ["Pending since", input.pendingSinceLabel], + ["Time pending", `${hours} hours (SLA ${input.slaHours}h)`], + ] as const; + + const htmlRows = rows + .map(([label, value]) => ` + + ${escapeHtml(label)} + ${escapeHtml(value)} + `) + .join(""); + + const html = ` + + + + + + RFP Pending Approval + + + + + + + +
+ + + + + + + + + + + + + + + + + + + + + + + + + + +
 
+ T Rock Construction +
+

RFP Awaiting Approval

+
+

This RFP has been waiting for approval for more than ${input.slaHours} hours. Open the deal to move it forward, or review the full Pending RFP queue.

+
+ ${htmlRows} +
+
+ + + View Deal in CRM + +
+ Open the Pending RFP queue +
+

This is an automated notification from T Rock Construction CRM. Please do not reply to this email.

+
+ +
+ +`; + + const text = + rows.map(([label, value]) => `${label}: ${value}`).join("\n") + + `\n\nView the deal in the CRM: ${dealUrl}` + + `\n\nOpen the Pending RFP queue: ${queueUrl}`; + return { subject, html, text, dealUrl, queueUrl }; +} diff --git a/worker/tests/jobs/rfp-pending-sla.runtime.test.ts b/worker/tests/jobs/rfp-pending-sla.runtime.test.ts new file mode 100644 index 000000000..56148d5ba --- /dev/null +++ b/worker/tests/jobs/rfp-pending-sla.runtime.test.ts @@ -0,0 +1,205 @@ +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { PGlite } from "@electric-sql/pglite"; +import { + buildRfpPendingSlaEmail, + findPendingRfpSlaBreaches, + isRfpPendingSlaEnabled, + runRfpPendingSlaScan, +} from "../../src/jobs/rfp-pending-sla.js"; + +describe("isRfpPendingSlaEnabled", () => { + it("is true only for an explicit 'true' (trimmed, case-insensitive)", () => { + expect(isRfpPendingSlaEnabled({ RFP_PENDING_SLA_ENABLED: "true" })).toBe(true); + expect(isRfpPendingSlaEnabled({ RFP_PENDING_SLA_ENABLED: " TRUE " })).toBe(true); + }); + it("is false when unset or not 'true' (default OFF — merges inert)", () => { + expect(isRfpPendingSlaEnabled({})).toBe(false); + expect(isRfpPendingSlaEnabled({ RFP_PENDING_SLA_ENABLED: "false" })).toBe(false); + expect(isRfpPendingSlaEnabled({ RFP_PENDING_SLA_ENABLED: "1" })).toBe(false); + expect(isRfpPendingSlaEnabled({ RFP_PENDING_SLA_ENABLED: "" })).toBe(false); + }); +}); + +describe("buildRfpPendingSlaEmail", () => { + const base = { + dealId: "deal-1", + dealName: "Sunrise ", + dealNumber: "DFW-1-17326", + pendingSinceLabel: "Jan 1, 2026, 9:00 AM CT", + hoursPending: 31, + slaHours: 24, + officeId: "office-uuid", + frontendUrl: "https://trockcrm.com", + }; + + it("subject names the deal number + name", () => { + const email = buildRfpPendingSlaEmail(base); + expect(email.subject).toContain("DFW-1-17326"); + expect(email.subject.toLowerCase()).toContain("pending"); + }); + + it("links to the deal with its officeId so a cross-office reviewer doesn't 404", () => { + const email = buildRfpPendingSlaEmail(base); + expect(email.html).toContain("https://trockcrm.com/deals/deal-1?officeId=office-uuid"); + expect(email.text).toContain("https://trockcrm.com/deals/deal-1?officeId=office-uuid"); + }); + + it("links to the Pending RFP queue", () => { + const email = buildRfpPendingSlaEmail(base); + expect(email.html).toContain("/deals/pending-rfp?officeId=office-uuid"); + }); + + it("states how long it's been pending and the SLA, and escapes the deal name", () => { + const email = buildRfpPendingSlaEmail(base); + expect(email.html).toContain("Jan 1, 2026, 9:00 AM CT"); + expect(email.html).toContain("31"); + expect(email.html).toContain("24"); + expect(email.html).toContain("Sunrise <Terraces>"); // escaped, not raw + expect(email.html).not.toContain("Sunrise "); + }); +}); + +// ---- Real-SQL (PGlite) filter test for the breach finder ---- +const U = (s: string) => `00000000-0000-0000-0000-${s.padStart(12, "0")}`; +const OPP = U("a1"); +const WON = U("a2"); +let pg: PGlite; +// eslint-disable-next-line @typescript-eslint/no-explicit-any +let query: any; + +beforeAll(async () => { + pg = new PGlite(); + await pg.exec(` + CREATE SCHEMA office_test; + CREATE TABLE public.pipeline_stage_config (id uuid PRIMARY KEY, slug varchar(100) NOT NULL); + CREATE TABLE office_test.deals ( + id uuid PRIMARY KEY, + name text, + stage_id uuid NOT NULL, + project_number text, + deal_number text, + is_bid_board_owned boolean NOT NULL DEFAULT false, + rfp_approval_status text, + rfp_approval_requested_at timestamptz, + is_active boolean NOT NULL DEFAULT true, + is_test_data boolean NOT NULL DEFAULT false, + on_hold boolean NOT NULL DEFAULT false + ); + INSERT INTO public.pipeline_stage_config (id, slug) VALUES ('${OPP}', 'opportunity'), ('${WON}', 'won'); + INSERT INTO office_test.deals + (id, name, stage_id, project_number, deal_number, is_bid_board_owned, rfp_approval_status, rfp_approval_requested_at, is_active, is_test_data, on_hold) + VALUES + -- BREACH: opportunity, not BBO, pending, 25h ago + ('${U("d1")}', 'Breach A', '${OPP}', 'P-1', 'HS-1', false, 'pending', NOW() - INTERVAL '25 hours', true, false, false), + -- BREACH: pending_outbox, 40h ago + ('${U("d2")}', 'Breach B', '${OPP}', 'P-2', 'HS-2', false, 'pending_outbox', NOW() - INTERVAL '40 hours', true, false, false), + -- not breaching: only 5h pending + ('${U("d3")}', 'Recent', '${OPP}', 'P-3', 'HS-3', false, 'pending', NOW() - INTERVAL '5 hours', true, false, false), + -- not breaching: declined (attention, not awaiting) + ('${U("d4")}', 'Declined', '${OPP}', 'P-4', 'HS-4', false, 'declined', NOW() - INTERVAL '30 hours', true, false, false), + -- not breaching: bid-board-owned + ('${U("d5")}', 'BBO', '${OPP}', 'P-5', 'HS-5', true, 'pending', NOW() - INTERVAL '30 hours', true, false, false), + -- not breaching: on hold + ('${U("d6")}', 'OnHold', '${OPP}', 'P-6', 'HS-6', false, 'pending', NOW() - INTERVAL '30 hours', true, false, true), + -- not breaching: test data + ('${U("d7")}', 'Test', '${OPP}', 'P-7', 'HS-7', false, 'pending', NOW() - INTERVAL '30 hours', true, true, false), + -- not breaching: inactive + ('${U("d8")}', 'Inactive', '${OPP}', 'P-8', 'HS-8', false, 'pending', NOW() - INTERVAL '30 hours', false, false, false), + -- not breaching: won stage (past opportunity) + ('${U("d9")}', 'Won', '${WON}', 'P-9', 'HS-9', false, 'pending', NOW() - INTERVAL '30 hours', true, false, false); + `); + query = (sql: string, params?: unknown[]) => pg.query(sql, params as never[]); +}); + +afterAll(async () => { + await pg?.close?.(); +}); + +describe("findPendingRfpSlaBreaches (real SQL)", () => { + it("returns only awaiting, >24h, opportunity, non-BBO, active, non-test, non-hold deals, oldest first", async () => { + const breaches = await findPendingRfpSlaBreaches(query, "office_test", 24); + expect(breaches.map((b) => b.dealId)).toEqual([U("d2"), U("d1")]); // 40h before 25h (oldest first) + expect(breaches[0]).toMatchObject({ dealName: "Breach B", dealNumber: "P-2" }); + expect(breaches[0].hoursPending).toBeGreaterThanOrEqual(24); + expect(typeof breaches[0].requestedAt).toBe("string"); + }); + + it("prefers project_number, falling back to deal_number", async () => { + const breaches = await findPendingRfpSlaBreaches(query, "office_test", 24); + expect(breaches.every((b) => b.dealNumber?.startsWith("P-"))).toBe(true); + }); +}); + +// ---- Orchestration test (mock query + send) for the exactly-once + retry logic ---- +function makeScanMocks(opts: { claimReturnsRow: boolean }) { + const sendEmail = vi.fn().mockResolvedValue({ success: true, messageId: "msg-1" }); + const calls: string[] = []; + const q = vi.fn(async (sql: string) => { + calls.push(sql); + if (sql.includes("FROM public.offices")) return { rows: [{ id: "office-1", slug: "beta" }] }; + if (sql.includes(".deals")) { + return { + rows: [ + { + id: "deal-1", + name: "Deal One", + deal_number: "P-1", + rfp_approval_requested_at: "2026-01-01T00:00:00.000Z", + hours_pending: 30, + }, + ], + }; + } + if (sql.includes("INSERT INTO public.rfp_pending_sla_email_receipts")) { + return { rows: opts.claimReturnsRow ? [{ deal_id: "deal-1" }] : [] }; + } + return { rows: [] }; + }); + return { sendEmail, q, calls }; +} + +const enabledEnv = { RFP_PENDING_SLA_ENABLED: "true", RFP_REJECTION_EMAIL_RECIPIENTS: "boss@trock.test" }; +const silent = { log: vi.fn(), warn: vi.fn(), error: vi.fn() }; + +describe("runRfpPendingSlaScan orchestration", () => { + beforeEach(() => vi.clearAllMocks()); + + it("is inert (no query, no send) when the flag is off", async () => { + const { sendEmail, q } = makeScanMocks({ claimReturnsRow: true }); + const summary = await runRfpPendingSlaScan({ query: q, sendEmail, env: { RFP_REJECTION_EMAIL_RECIPIENTS: "boss@trock.test" }, logger: silent }); + expect(q).not.toHaveBeenCalled(); + expect(sendEmail).not.toHaveBeenCalled(); + expect(summary.sent).toBe(0); + }); + + it("does not send (and logs) when no leadership recipients are configured", async () => { + const { sendEmail, q } = makeScanMocks({ claimReturnsRow: true }); + await runRfpPendingSlaScan({ query: q, sendEmail, env: { RFP_PENDING_SLA_ENABLED: "true" }, logger: silent }); + expect(sendEmail).not.toHaveBeenCalled(); + }); + + it("claims the receipt then sends exactly once for a breaching deal", async () => { + const { sendEmail, q, calls } = makeScanMocks({ claimReturnsRow: true }); + const summary = await runRfpPendingSlaScan({ query: q, sendEmail, env: enabledEnv, logger: silent }); + expect(sendEmail).toHaveBeenCalledTimes(1); + expect(sendEmail.mock.calls[0][0]).toEqual(["boss@trock.test"]); + expect(summary.sent).toBe(1); + // records the message id after a successful send + expect(calls.some((s) => s.includes("UPDATE public.rfp_pending_sla_email_receipts"))).toBe(true); + }); + + it("skips sending when the receipt claim conflicts (already sent this cycle)", async () => { + const { sendEmail, q } = makeScanMocks({ claimReturnsRow: false }); + const summary = await runRfpPendingSlaScan({ query: q, sendEmail, env: enabledEnv, logger: silent }); + expect(sendEmail).not.toHaveBeenCalled(); + expect(summary.skipped).toBe(1); + }); + + it("rolls back the claim (delete) and does not throw when the send fails", async () => { + const { q, calls } = makeScanMocks({ claimReturnsRow: true }); + const sendEmail = vi.fn().mockRejectedValue(new Error("provider down")); + const summary = await runRfpPendingSlaScan({ query: q, sendEmail, env: enabledEnv, logger: silent }); + expect(summary.failed).toBe(1); + expect(calls.some((s) => s.includes("DELETE FROM public.rfp_pending_sla_email_receipts"))).toBe(true); + }); +}); diff --git a/worker/vitest.config.ts b/worker/vitest.config.ts index 2818dd58e..e0b4e7064 100644 --- a/worker/vitest.config.ts +++ b/worker/vitest.config.ts @@ -10,6 +10,9 @@ export default defineConfig({ "@trock-crm/shared/schema": path.resolve(__dirname, "../shared/src/schema/index.ts"), "@trock-crm/shared/types": path.resolve(__dirname, "../shared/src/types/index.ts"), "@trock-crm/shared/utils": path.resolve(__dirname, "../shared/src/utils/normalize.ts"), + // Subpath lib exports must be aliased individually and BEFORE the catch-all below, else the broad + // "@trock-crm/shared" alias rewrites them to the schema entrypoint and resolution fails. + "@trock-crm/shared/lib/rfpReviewerEmails": path.resolve(__dirname, "../shared/src/lib/rfpReviewerEmails.ts"), "@trock-crm/shared": path.resolve(__dirname, "../shared/src/schema/index.ts"), }, }, From f12d21c4f340d4257c7668366fcd3be21a6934ec Mon Sep 17 00:00:00 2001 From: Adnaan Iqbal <64221968+artificialadnaan@users.noreply.github.com> Date: Thu, 2 Jul 2026 18:09:43 -0500 Subject: [PATCH 2/8] fix(rfp): harden the SLA scan to mirror the Pending RFP queue exactly (Codex #864) - P2 crash-suppression: switch from claim-first to record-AFTER-a-durable-send. The receipt is now written only once the email is accepted, so a crash between the decision and the send leaves no receipt and the next hourly scan retries (was: a committed claim could permanently suppress the alert). - P2 legacy stage aliases: resolve every stage id that canonicalizes to Opportunity (incl. `dd`) via toCanonicalDealStageSlug, matching the trigger route + pending-rfp-service, instead of requiring slug='opportunity'. - P2 override-resolved rows: exclude rfp_override_decision='denial_reconfirmed' and rfp_override_state='approving' (the NOT_RECONFIRMED_DENIAL / NOT_OVERRIDE_APPROVING guards the queue applies), so a row hidden from /deals/pending-rfp never triggers a leadership alert. - P2 stale send: re-check the deal is STILL an awaiting, unresolved breach for the same rfp_approval_requested_at cycle immediately before sending, so an approve/decline landing between the batch scan and the send doesn't fire a false alert. Tests updated: real-SQL fixture gains a `dd`-stage breach + override-resolved rows (asserted excluded); orchestration covers receipt-exists skip, re-check skip, and no-receipt-on-send-failure. Co-Authored-By: Claude Opus 4.8 (1M context) --- worker/src/jobs/rfp-pending-sla.ts | 117 +++++++++++++----- .../jobs/rfp-pending-sla.runtime.test.ts | 96 +++++++++----- 2 files changed, 155 insertions(+), 58 deletions(-) diff --git a/worker/src/jobs/rfp-pending-sla.ts b/worker/src/jobs/rfp-pending-sla.ts index e6c2ec8a9..efdb30d4c 100644 --- a/worker/src/jobs/rfp-pending-sla.ts +++ b/worker/src/jobs/rfp-pending-sla.ts @@ -6,6 +6,9 @@ import { resolveFrontendUrl, TROCK_LOGO_EMAIL_URL } from "./project-number-email // Leadership recipients (Takashi + Adam Shaw) — the SAME source of truth the RFP-decline email and the // server override-review gate use (RFP_REJECTION_EMAIL_RECIPIENTS), so the notified set never drifts. import { resolveRfpReviewerEmails } from "@trock-crm/shared/lib/rfpReviewerEmails"; +// Mirror the Pending RFP queue EXACTLY: awaiting-only statuses + every stage id that canonicalizes to +// Opportunity (incl. legacy `dd`). Keeps the SLA scan and the /deals/pending-rfp bucket in lockstep. +import { PENDING_RFP_AWAITING_STATUSES, toCanonicalDealStageSlug } from "@trock-crm/shared/types"; import { escapeHtml } from "../lib/email-format.js"; /** The Pending RFP SLA: an RFP awaiting approval longer than this many hours is a breach. */ @@ -53,11 +56,27 @@ export function isRfpPendingSlaEnabled(env: NodeJS.ProcessEnv): boolean { return String(env.RFP_PENDING_SLA_ENABLED ?? "").trim().toLowerCase() === "true"; } +/** Every stage id whose slug canonicalizes to Opportunity (incl. legacy aliases like `dd`) — exactly the set + * the Pending RFP queue (pending-rfp-service.ts) and the trigger route treat as Opportunity. */ +async function opportunityStageIds(query: PgQuery): Promise { + const result = await query(`SELECT id, slug FROM public.pipeline_stage_config`); + return result.rows + .filter( + (s) => + s.slug != null && + (toCanonicalDealStageSlug(String(s.slug), "normal") === "opportunity" || + toCanonicalDealStageSlug(String(s.slug), "service") === "opportunity"), + ) + .map((s) => String(s.id)); +} + /** * Deals whose RFP has been awaiting approval (pending_outbox/pending) longer than `slaHours`, in one office - * schema. Mirrors the pending-RFP bucket predicate (opportunity, not bid-board-owned, awaiting status) and - * adds the 24h clock off deals.rfp_approval_requested_at (indexed). Excludes on-hold + test-data deals so a - * paused/demo deal never alerts leadership. Oldest first. + * schema. Mirrors the Pending RFP bucket EXACTLY (pending-rfp-service.ts): every Opportunity-canonical stage + * (incl. legacy `dd`), not bid-board-owned, awaiting status, excludes re-confirmed denials + * (rfp_override_decision='denial_reconfirmed') and in-flight override approvals (rfp_override_state='approving') + * so a row hidden from the queue never alerts. Adds the 24h clock off rfp_approval_requested_at (indexed) and + * drops on-hold + test-data. Oldest first. */ export async function findPendingRfpSlaBreaches( query: PgQuery, @@ -67,6 +86,8 @@ export async function findPendingRfpSlaBreaches( if (!TENANT_SCHEMA_REGEX.test(schemaName)) { throw new Error(`Unsafe tenant schema: ${schemaName}`); } + const oppStageIds = await opportunityStageIds(query); + if (oppStageIds.length === 0) return []; const result = await query( `SELECT d.id, d.name, @@ -74,17 +95,18 @@ export async function findPendingRfpSlaBreaches( d.rfp_approval_requested_at, EXTRACT(EPOCH FROM (NOW() - d.rfp_approval_requested_at)) / 3600.0 AS hours_pending FROM ${schemaName}.deals d - JOIN public.pipeline_stage_config psc ON psc.id = d.stage_id - WHERE psc.slug = 'opportunity' + WHERE d.stage_id = ANY($1::uuid[]) AND d.is_bid_board_owned = false - AND d.rfp_approval_status IN ('pending_outbox', 'pending') + AND d.rfp_approval_status = ANY($2::text[]) AND d.rfp_approval_requested_at IS NOT NULL - AND d.rfp_approval_requested_at < NOW() - ($1 || ' hours')::interval + AND d.rfp_approval_requested_at < NOW() - ($3 || ' hours')::interval AND d.is_active = true AND COALESCE(d.is_test_data, false) = false AND COALESCE(d.on_hold, false) = false + AND COALESCE(d.rfp_override_decision, '') <> 'denial_reconfirmed' + AND COALESCE(d.rfp_override_state, '') <> 'approving' ORDER BY d.rfp_approval_requested_at ASC`, - [String(slaHours)], + [oppStageIds, [...PENDING_RFP_AWAITING_STATUSES], String(slaHours)], ); return result.rows.map((row) => ({ dealId: String(row.id), @@ -170,16 +192,20 @@ async function processBreach( ): Promise<"sent" | "skipped" | "failed"> { const { tenantSchema, officeId, breach, recipients, frontendUrl, slaHours } = args; - // Claim this cycle atomically; a conflict means we already alerted for this pending instant. - const claim = await query( - `INSERT INTO public.rfp_pending_sla_email_receipts - (tenant_schema, deal_id, rfp_approval_requested_at, deal_number, recipient_emails, sent_at, created_at, updated_at) - VALUES ($1, $2::uuid, $3, $4, $5, NOW(), NOW(), NOW()) - ON CONFLICT (tenant_schema, deal_id, rfp_approval_requested_at) DO NOTHING - RETURNING deal_id`, - [tenantSchema, breach.dealId, breach.requestedAt, breach.dealNumber, recipients.join(", ")], + // Already alerted for this pending cycle? The receipt is written only AFTER a durable send (below), so a + // crash between "decided to send" and "sent" leaves NO receipt and the next scan retries — the alert is + // never permanently suppressed. + const existing = await query( + `SELECT 1 FROM public.rfp_pending_sla_email_receipts + WHERE tenant_schema = $1 AND deal_id = $2::uuid AND rfp_approval_requested_at = $3 + LIMIT 1`, + [tenantSchema, breach.dealId, breach.requestedAt], ); - if (claim.rows.length === 0) return "skipped"; + if (existing.rows.length > 0) return "skipped"; + + // Re-check the deal is STILL a breach right before sending: a reviewer may approve/decline (or an override + // may resolve) between the batch scan and this send, and those transitions are async. Skip a stale alert. + if (!(await isStillBreaching(query, tenantSchema, breach))) return "skipped"; try { const email = buildRfpPendingSlaEmail({ @@ -197,11 +223,14 @@ async function processBreach( idempotencyKey: `rfp-pending-sla-${tenantSchema}-${breach.dealId}-${breach.requestedAt}`, }); if (!result.success) throw new Error("Email provider returned unsuccessful result"); + // Record the receipt ONLY after the send is durable, so a failed send never suppresses a retry. ON + // CONFLICT DO NOTHING guards the (single-worker) case of an overlapping run that already recorded it. await query( - `UPDATE public.rfp_pending_sla_email_receipts - SET resend_message_id = $4, updated_at = NOW() - WHERE tenant_schema = $1 AND deal_id = $2::uuid AND rfp_approval_requested_at = $3`, - [tenantSchema, breach.dealId, breach.requestedAt, result.messageId], + `INSERT INTO public.rfp_pending_sla_email_receipts + (tenant_schema, deal_id, rfp_approval_requested_at, deal_number, recipient_emails, resend_message_id, sent_at, created_at, updated_at) + VALUES ($1, $2::uuid, $3, $4, $5, $6, NOW(), NOW(), NOW()) + ON CONFLICT (tenant_schema, deal_id, rfp_approval_requested_at) DO NOTHING`, + [tenantSchema, breach.dealId, breach.requestedAt, breach.dealNumber, recipients.join(", "), result.messageId], ); logger.log("[RfpPendingSla] Sent SLA breach alert", { tenantSchema, @@ -211,13 +240,8 @@ async function processBreach( }); return "sent"; } catch (err) { - // Roll back the claim so a transient failure re-sends on the next scan (the claim must not permanently - // suppress the alert). One deal's failure must not abort the rest of the scan, so we swallow + log here. - await query( - `DELETE FROM public.rfp_pending_sla_email_receipts - WHERE tenant_schema = $1 AND deal_id = $2::uuid AND rfp_approval_requested_at = $3`, - [tenantSchema, breach.dealId, breach.requestedAt], - ).catch(() => undefined); + // No receipt was written, so the next hourly scan retries. One deal's failure must not abort the rest of + // the scan, so we swallow + log here. logger.error("[RfpPendingSla] Failed to send SLA breach alert", { tenantSchema, dealId: breach.dealId, @@ -227,6 +251,43 @@ async function processBreach( } } +/** Re-read the deal and confirm it's STILL an awaiting, unresolved, alert-eligible pending RFP for the same + * cycle — the same predicate as findPendingRfpSlaBreaches, minus the 24h clock (only grows). */ +async function isStillBreaching( + query: PgQuery, + schemaName: string, + breach: RfpPendingSlaBreach, +): Promise { + if (!TENANT_SCHEMA_REGEX.test(schemaName)) return false; + const result = await query( + `SELECT d.rfp_approval_status AS status, + d.rfp_approval_requested_at AS requested_at, + COALESCE(d.is_bid_board_owned, false) AS bbo, + COALESCE(d.is_active, false) AS active, + COALESCE(d.is_test_data, false) AS test_data, + COALESCE(d.on_hold, false) AS on_hold, + COALESCE(d.rfp_override_decision, '') AS override_decision, + COALESCE(d.rfp_override_state, '') AS override_state + FROM ${schemaName}.deals d + WHERE d.id = $1::uuid + LIMIT 1`, + [breach.dealId], + ); + const row = result.rows[0]; + if (!row) return false; + return ( + (PENDING_RFP_AWAITING_STATUSES as readonly string[]).includes(String(row.status)) && + row.requested_at != null && + new Date(row.requested_at).toISOString() === breach.requestedAt && + row.bbo !== true && + row.active === true && + row.test_data !== true && + row.on_hold !== true && + row.override_decision !== "denial_reconfirmed" && + row.override_state !== "approving" + ); +} + function formatPendingSince(iso: string): string { const label = new Date(iso).toLocaleString("en-US", { timeZone: "America/Chicago", diff --git a/worker/tests/jobs/rfp-pending-sla.runtime.test.ts b/worker/tests/jobs/rfp-pending-sla.runtime.test.ts index 56148d5ba..ab7346463 100644 --- a/worker/tests/jobs/rfp-pending-sla.runtime.test.ts +++ b/worker/tests/jobs/rfp-pending-sla.runtime.test.ts @@ -63,6 +63,7 @@ describe("buildRfpPendingSlaEmail", () => { const U = (s: string) => `00000000-0000-0000-0000-${s.padStart(12, "0")}`; const OPP = U("a1"); const WON = U("a2"); +const DD = U("a3"); // legacy stage that canonicalizes to Opportunity let pg: PGlite; // eslint-disable-next-line @typescript-eslint/no-explicit-any let query: any; @@ -83,30 +84,38 @@ beforeAll(async () => { rfp_approval_requested_at timestamptz, is_active boolean NOT NULL DEFAULT true, is_test_data boolean NOT NULL DEFAULT false, - on_hold boolean NOT NULL DEFAULT false + on_hold boolean NOT NULL DEFAULT false, + rfp_override_decision text, + rfp_override_state text ); - INSERT INTO public.pipeline_stage_config (id, slug) VALUES ('${OPP}', 'opportunity'), ('${WON}', 'won'); + INSERT INTO public.pipeline_stage_config (id, slug) VALUES ('${OPP}', 'opportunity'), ('${WON}', 'won'), ('${DD}', 'dd'); INSERT INTO office_test.deals - (id, name, stage_id, project_number, deal_number, is_bid_board_owned, rfp_approval_status, rfp_approval_requested_at, is_active, is_test_data, on_hold) + (id, name, stage_id, project_number, deal_number, is_bid_board_owned, rfp_approval_status, rfp_approval_requested_at, is_active, is_test_data, on_hold, rfp_override_decision, rfp_override_state) VALUES -- BREACH: opportunity, not BBO, pending, 25h ago - ('${U("d1")}', 'Breach A', '${OPP}', 'P-1', 'HS-1', false, 'pending', NOW() - INTERVAL '25 hours', true, false, false), + ('${U("d1")}', 'Breach A', '${OPP}', 'P-1', 'HS-1', false, 'pending', NOW() - INTERVAL '25 hours', true, false, false, NULL, NULL), -- BREACH: pending_outbox, 40h ago - ('${U("d2")}', 'Breach B', '${OPP}', 'P-2', 'HS-2', false, 'pending_outbox', NOW() - INTERVAL '40 hours', true, false, false), + ('${U("d2")}', 'Breach B', '${OPP}', 'P-2', 'HS-2', false, 'pending_outbox', NOW() - INTERVAL '40 hours', true, false, false, NULL, NULL), + -- BREACH via legacy dd stage (canonicalizes to opportunity), 35h ago + ('${U("dd")}', 'Breach DD', '${DD}', 'P-DD', 'HS-DD', false, 'pending', NOW() - INTERVAL '35 hours', true, false, false, NULL, NULL), -- not breaching: only 5h pending - ('${U("d3")}', 'Recent', '${OPP}', 'P-3', 'HS-3', false, 'pending', NOW() - INTERVAL '5 hours', true, false, false), + ('${U("d3")}', 'Recent', '${OPP}', 'P-3', 'HS-3', false, 'pending', NOW() - INTERVAL '5 hours', true, false, false, NULL, NULL), -- not breaching: declined (attention, not awaiting) - ('${U("d4")}', 'Declined', '${OPP}', 'P-4', 'HS-4', false, 'declined', NOW() - INTERVAL '30 hours', true, false, false), + ('${U("d4")}', 'Declined', '${OPP}', 'P-4', 'HS-4', false, 'declined', NOW() - INTERVAL '30 hours', true, false, false, NULL, NULL), -- not breaching: bid-board-owned - ('${U("d5")}', 'BBO', '${OPP}', 'P-5', 'HS-5', true, 'pending', NOW() - INTERVAL '30 hours', true, false, false), + ('${U("d5")}', 'BBO', '${OPP}', 'P-5', 'HS-5', true, 'pending', NOW() - INTERVAL '30 hours', true, false, false, NULL, NULL), -- not breaching: on hold - ('${U("d6")}', 'OnHold', '${OPP}', 'P-6', 'HS-6', false, 'pending', NOW() - INTERVAL '30 hours', true, false, true), + ('${U("d6")}', 'OnHold', '${OPP}', 'P-6', 'HS-6', false, 'pending', NOW() - INTERVAL '30 hours', true, false, true, NULL, NULL), -- not breaching: test data - ('${U("d7")}', 'Test', '${OPP}', 'P-7', 'HS-7', false, 'pending', NOW() - INTERVAL '30 hours', true, true, false), + ('${U("d7")}', 'Test', '${OPP}', 'P-7', 'HS-7', false, 'pending', NOW() - INTERVAL '30 hours', true, true, false, NULL, NULL), -- not breaching: inactive - ('${U("d8")}', 'Inactive', '${OPP}', 'P-8', 'HS-8', false, 'pending', NOW() - INTERVAL '30 hours', false, false, false), + ('${U("d8")}', 'Inactive', '${OPP}', 'P-8', 'HS-8', false, 'pending', NOW() - INTERVAL '30 hours', false, false, false, NULL, NULL), -- not breaching: won stage (past opportunity) - ('${U("d9")}', 'Won', '${WON}', 'P-9', 'HS-9', false, 'pending', NOW() - INTERVAL '30 hours', true, false, false); + ('${U("d9")}', 'Won', '${WON}', 'P-9', 'HS-9', false, 'pending', NOW() - INTERVAL '30 hours', true, false, false, NULL, NULL), + -- not breaching: re-confirmed denial (resolved terminal; hidden from the queue) + ('${U("da")}', 'ReconfirmedDenial', '${OPP}', 'P-A', 'HS-A', false, 'pending', NOW() - INTERVAL '30 hours', true, false, false, 'denial_reconfirmed', NULL), + -- not breaching: override approval in flight (becoming a Bid Board project) + ('${U("db")}', 'OverrideApproving', '${OPP}', 'P-B', 'HS-B', false, 'pending', NOW() - INTERVAL '30 hours', true, false, false, NULL, 'approving'); `); query = (sql: string, params?: unknown[]) => pg.query(sql, params as never[]); }); @@ -116,28 +125,39 @@ afterAll(async () => { }); describe("findPendingRfpSlaBreaches (real SQL)", () => { - it("returns only awaiting, >24h, opportunity, non-BBO, active, non-test, non-hold deals, oldest first", async () => { + it("returns only awaiting, >24h, Opportunity-canonical (incl. legacy dd), non-BBO, active, non-test, non-hold, non-override-resolved deals, oldest first", async () => { const breaches = await findPendingRfpSlaBreaches(query, "office_test", 24); - expect(breaches.map((b) => b.dealId)).toEqual([U("d2"), U("d1")]); // 40h before 25h (oldest first) + // 40h (d2) > 35h (dd, legacy stage) > 25h (d1); everything else excluded. + expect(breaches.map((b) => b.dealId)).toEqual([U("d2"), U("dd"), U("d1")]); expect(breaches[0]).toMatchObject({ dealName: "Breach B", dealNumber: "P-2" }); expect(breaches[0].hoursPending).toBeGreaterThanOrEqual(24); expect(typeof breaches[0].requestedAt).toBe("string"); }); + it("excludes re-confirmed denials and in-flight override approvals (mirrors the Pending RFP queue)", async () => { + const breaches = await findPendingRfpSlaBreaches(query, "office_test", 24); + expect(breaches.map((b) => b.dealId)).not.toContain(U("da")); // denial_reconfirmed + expect(breaches.map((b) => b.dealId)).not.toContain(U("db")); // override approving + }); + it("prefers project_number, falling back to deal_number", async () => { const breaches = await findPendingRfpSlaBreaches(query, "office_test", 24); expect(breaches.every((b) => b.dealNumber?.startsWith("P-"))).toBe(true); }); }); -// ---- Orchestration test (mock query + send) for the exactly-once + retry logic ---- -function makeScanMocks(opts: { claimReturnsRow: boolean }) { +// ---- Orchestration test (mock query + send): exactly-once (record-after-send) + pre-send re-check ---- +function makeScanMocks(opts: { receiptExists?: boolean; stillBreaching?: boolean } = {}) { + const receiptExists = opts.receiptExists ?? false; + const stillBreaching = opts.stillBreaching ?? true; const sendEmail = vi.fn().mockResolvedValue({ success: true, messageId: "msg-1" }); const calls: string[] = []; const q = vi.fn(async (sql: string) => { calls.push(sql); if (sql.includes("FROM public.offices")) return { rows: [{ id: "office-1", slug: "beta" }] }; - if (sql.includes(".deals")) { + if (sql.includes("FROM public.pipeline_stage_config")) return { rows: [{ id: "opp-stage", slug: "opportunity" }] }; + // The breach scan is the only .deals query that computes hours_pending. + if (sql.includes(".deals") && sql.includes("hours_pending")) { return { rows: [ { @@ -150,8 +170,16 @@ function makeScanMocks(opts: { claimReturnsRow: boolean }) { ], }; } - if (sql.includes("INSERT INTO public.rfp_pending_sla_email_receipts")) { - return { rows: opts.claimReturnsRow ? [{ deal_id: "deal-1" }] : [] }; + // The pre-send re-check re-reads the deal row (no hours_pending). + if (sql.includes(".deals") && sql.includes("WHERE d.id")) { + return { + rows: stillBreaching + ? [{ status: "pending", requested_at: "2026-01-01T00:00:00.000Z", bbo: false, active: true, test_data: false, on_hold: false, override_decision: "", override_state: "" }] + : [{ status: "approved", requested_at: "2026-01-01T00:00:00.000Z", bbo: false, active: true, test_data: false, on_hold: false, override_decision: "", override_state: "" }], + }; + } + if (sql.includes("SELECT 1") && sql.includes("rfp_pending_sla_email_receipts")) { + return { rows: receiptExists ? [{ "?column?": 1 }] : [] }; } return { rows: [] }; }); @@ -165,7 +193,7 @@ describe("runRfpPendingSlaScan orchestration", () => { beforeEach(() => vi.clearAllMocks()); it("is inert (no query, no send) when the flag is off", async () => { - const { sendEmail, q } = makeScanMocks({ claimReturnsRow: true }); + const { sendEmail, q } = makeScanMocks(); const summary = await runRfpPendingSlaScan({ query: q, sendEmail, env: { RFP_REJECTION_EMAIL_RECIPIENTS: "boss@trock.test" }, logger: silent }); expect(q).not.toHaveBeenCalled(); expect(sendEmail).not.toHaveBeenCalled(); @@ -173,33 +201,41 @@ describe("runRfpPendingSlaScan orchestration", () => { }); it("does not send (and logs) when no leadership recipients are configured", async () => { - const { sendEmail, q } = makeScanMocks({ claimReturnsRow: true }); + const { sendEmail, q } = makeScanMocks(); await runRfpPendingSlaScan({ query: q, sendEmail, env: { RFP_PENDING_SLA_ENABLED: "true" }, logger: silent }); expect(sendEmail).not.toHaveBeenCalled(); }); - it("claims the receipt then sends exactly once for a breaching deal", async () => { - const { sendEmail, q, calls } = makeScanMocks({ claimReturnsRow: true }); + it("sends exactly once for a breaching deal, recording the receipt only AFTER a durable send", async () => { + const { sendEmail, q, calls } = makeScanMocks(); const summary = await runRfpPendingSlaScan({ query: q, sendEmail, env: enabledEnv, logger: silent }); expect(sendEmail).toHaveBeenCalledTimes(1); expect(sendEmail.mock.calls[0][0]).toEqual(["boss@trock.test"]); expect(summary.sent).toBe(1); - // records the message id after a successful send - expect(calls.some((s) => s.includes("UPDATE public.rfp_pending_sla_email_receipts"))).toBe(true); + // The receipt INSERT happens after the send resolves. + const sendIdx = calls.findIndex((s) => s.includes("INSERT INTO public.rfp_pending_sla_email_receipts")); + expect(sendIdx).toBeGreaterThan(-1); + }); + + it("skips sending when a receipt already exists for this cycle", async () => { + const { sendEmail, q } = makeScanMocks({ receiptExists: true }); + const summary = await runRfpPendingSlaScan({ query: q, sendEmail, env: enabledEnv, logger: silent }); + expect(sendEmail).not.toHaveBeenCalled(); + expect(summary.skipped).toBe(1); }); - it("skips sending when the receipt claim conflicts (already sent this cycle)", async () => { - const { sendEmail, q } = makeScanMocks({ claimReturnsRow: false }); + it("skips (no send) when the deal is no longer awaiting by the time we go to send", async () => { + const { sendEmail, q } = makeScanMocks({ stillBreaching: false }); const summary = await runRfpPendingSlaScan({ query: q, sendEmail, env: enabledEnv, logger: silent }); expect(sendEmail).not.toHaveBeenCalled(); expect(summary.skipped).toBe(1); }); - it("rolls back the claim (delete) and does not throw when the send fails", async () => { - const { q, calls } = makeScanMocks({ claimReturnsRow: true }); + it("writes NO receipt (so the next scan retries) and does not throw when the send fails", async () => { + const { q, calls } = makeScanMocks(); const sendEmail = vi.fn().mockRejectedValue(new Error("provider down")); const summary = await runRfpPendingSlaScan({ query: q, sendEmail, env: enabledEnv, logger: silent }); expect(summary.failed).toBe(1); - expect(calls.some((s) => s.includes("DELETE FROM public.rfp_pending_sla_email_receipts"))).toBe(true); + expect(calls.some((s) => s.includes("INSERT INTO public.rfp_pending_sla_email_receipts"))).toBe(false); }); }); From d52aa203a1499820fe29acb18d95e381ac6333e3 Mon Sep 17 00:00:00 2001 From: Adnaan Iqbal <64221968+artificialadnaan@users.noreply.github.com> Date: Thu, 2 Jul 2026 20:50:31 -0500 Subject: [PATCH 3/8] fix(rfp): SLA scan single-flight + stage re-check + idempotent payload (Codex #864 r2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - P2 stage re-check: isStillBreaching now also verifies the deal's stage STILL canonicalizes to Opportunity. A deal moved out of Opportunity via POST /stage keeps its rfp_approval_status, so the status check alone would let a moved deal slip through and get a stale alert. - P2 concurrency: add a single-flight Postgres advisory lock (pg_try_advisory_lock over a dedicated pooled client) around the scan, so a still-running run or a second worker instance skips its tick instead of both submitting the same alert. - P2 idempotent retries: the email payload is now STABLE per cycle (">24h" + fixed "Pending since" instant, no drifting live hours). The Resend idempotencyKey is keyed on rfp_approval_requested_at, and Resend rejects a same-key/different-payload retry — so a crash-then-retry now re-sends cleanly instead of failing until the key TTL expires and then duplicating. - P2 test-gate stability: bump this PGlite beforeAll to a 30s hook timeout so it survives the concurrent full test:runtime gate (several PGlite suites cold-start together under maxWorkers 4). Tests: added single-flight (lock-held skip) + stage-moved-out skip cases; email assertions now check the stable ">24h" / "More than 24 hours" payload. Worker typecheck + full runtime suite green. Co-Authored-By: Claude Opus 4.8 (1M context) --- worker/src/jobs/rfp-pending-sla.ts | 124 +++++++++++++----- .../jobs/rfp-pending-sla.runtime.test.ts | 61 ++++++--- 2 files changed, 131 insertions(+), 54 deletions(-) diff --git a/worker/src/jobs/rfp-pending-sla.ts b/worker/src/jobs/rfp-pending-sla.ts index efdb30d4c..5f1be106e 100644 --- a/worker/src/jobs/rfp-pending-sla.ts +++ b/worker/src/jobs/rfp-pending-sla.ts @@ -33,6 +33,9 @@ interface RunDeps { env?: NodeJS.ProcessEnv; logger?: Pick; slaHours?: number; + /** Single-flight guard: resolves to a release fn if this run acquired the lock, or null if another run + * holds it (this run should then skip). Default = a Postgres session advisory lock over the pool. */ + acquireLock?: () => Promise Promise)>; } export interface RfpPendingSlaBreach { @@ -119,10 +122,13 @@ export async function findPendingRfpSlaBreaches( /** * Scan every active office for RFPs breaching the 24h pending SLA and email leadership once per pending - * cycle. Exactly-once + retry-safe: a claim row in public.rfp_pending_sla_email_receipts (keyed - * tenant_schema, deal_id, rfp_approval_requested_at) is inserted ON CONFLICT DO NOTHING BEFORE the send — - * a conflict means this cycle already alerted (skip), and a send failure DELETEs the claim so the next scan - * retries. A re-triggered RFP gets a fresh rfp_approval_requested_at -> a new cycle -> a new alert. + * cycle. Exactly-once + retry-safe: + * - a single-flight Postgres advisory lock serializes runs (a still-running or second-instance run skips), + * so two runs can't both submit the same alert; + * - the receipt row (keyed tenant_schema, deal_id, rfp_approval_requested_at) is written only AFTER a + * durable send, so a crash before the send leaves no receipt and the next scan retries; + * - the email payload is stable per cycle, so a crash-then-retry reuses the same Resend idempotencyKey. + * A re-triggered RFP gets a fresh rfp_approval_requested_at -> a new cycle -> a new alert. */ export async function runRfpPendingSlaScan(deps: RunDeps = {}): Promise { const env = deps.env ?? process.env; @@ -146,37 +152,74 @@ export async function runRfpPendingSlaScan(deps: RunDeps = {}): Promise undefined); } logger.log("[RfpPendingSla] Scan complete", summary); return summary; } +// Postgres session advisory lock over a dedicated pooled client (a session lock must be acquired + released +// on the SAME connection, so it can't go through pool.query which hands out arbitrary connections). Global +// across worker instances, so a second instance's tick skips too. Returns null if the lock is already held. +async function acquireScanAdvisoryLock(): Promise Promise)> { + const LOCK_KEY = 0x52_46_50_53; // "RFPS" — stable, arbitrary + const client = await pool.connect(); + try { + const res = await client.query("SELECT pg_try_advisory_lock($1) AS locked", [LOCK_KEY]); + if (res.rows[0]?.locked !== true) { + client.release(); + return null; + } + } catch (err) { + client.release(); + throw err; + } + return async () => { + try { + await client.query("SELECT pg_advisory_unlock($1)", [LOCK_KEY]); + } finally { + client.release(); + } + }; +} + async function processBreach( query: PgQuery, sendEmail: SendEmail, @@ -213,7 +256,6 @@ async function processBreach( dealName: breach.dealName, dealNumber: breach.dealNumber, pendingSinceLabel: formatPendingSince(breach.requestedAt), - hoursPending: breach.hoursPending, slaHours, officeId, frontendUrl, @@ -262,6 +304,7 @@ async function isStillBreaching( const result = await query( `SELECT d.rfp_approval_status AS status, d.rfp_approval_requested_at AS requested_at, + psc.slug AS stage_slug, COALESCE(d.is_bid_board_owned, false) AS bbo, COALESCE(d.is_active, false) AS active, COALESCE(d.is_test_data, false) AS test_data, @@ -269,13 +312,22 @@ async function isStillBreaching( COALESCE(d.rfp_override_decision, '') AS override_decision, COALESCE(d.rfp_override_state, '') AS override_state FROM ${schemaName}.deals d + LEFT JOIN public.pipeline_stage_config psc ON psc.id = d.stage_id WHERE d.id = $1::uuid LIMIT 1`, [breach.dealId], ); const row = result.rows[0]; if (!row) return false; + // The stage must STILL canonicalize to Opportunity — a deal moved out of Opportunity via POST /stage does + // not clear rfp_approval_status, so the status check alone would let a moved deal slip through. + const slug = row.stage_slug == null ? null : String(row.stage_slug); + const stageIsOpportunity = + slug != null && + (toCanonicalDealStageSlug(slug, "normal") === "opportunity" || + toCanonicalDealStageSlug(slug, "service") === "opportunity"); return ( + stageIsOpportunity && (PENDING_RFP_AWAITING_STATUSES as readonly string[]).includes(String(row.status)) && row.requested_at != null && new Date(row.requested_at).toISOString() === breach.requestedAt && @@ -305,7 +357,6 @@ export function buildRfpPendingSlaEmail(input: { dealName: string; dealNumber: string | null; pendingSinceLabel: string; - hoursPending: number; slaHours: number; officeId?: string | null; frontendUrl: string; @@ -318,17 +369,20 @@ export function buildRfpPendingSlaEmail(input: { const queueUrl = `${baseUrl}/deals/pending-rfp${officeParam}`; const safeDealUrl = escapeHtml(dealUrl); const safeQueueUrl = escapeHtml(queueUrl); - const hours = Math.floor(input.hoursPending); + // The payload is deliberately STABLE per pending cycle (no live "hours pending" count): the Resend + // idempotencyKey is keyed on rfp_approval_requested_at, and Resend rejects a same-key/different-payload + // retry. "more than {slaHours}h" + the fixed "Pending since" instant convey urgency without drifting each + // hour, so a crash-then-retry re-sends cleanly instead of being rejected until the key TTL expires. const subject = input.dealNumber - ? `RFP pending approval ${hours}h: ${input.dealNumber} (${input.dealName})` - : `RFP pending approval ${hours}h: ${input.dealName}`; + ? `RFP pending approval >${input.slaHours}h: ${input.dealNumber} (${input.dealName})` + : `RFP pending approval >${input.slaHours}h: ${input.dealName}`; const rows = [ ["Deal name", input.dealName], ["Project number", input.dealNumber ?? "Pending"], ["Pending since", input.pendingSinceLabel], - ["Time pending", `${hours} hours (SLA ${input.slaHours}h)`], + ["Time pending", `More than ${input.slaHours} hours`], ] as const; const htmlRows = rows diff --git a/worker/tests/jobs/rfp-pending-sla.runtime.test.ts b/worker/tests/jobs/rfp-pending-sla.runtime.test.ts index ab7346463..7fa264940 100644 --- a/worker/tests/jobs/rfp-pending-sla.runtime.test.ts +++ b/worker/tests/jobs/rfp-pending-sla.runtime.test.ts @@ -26,16 +26,16 @@ describe("buildRfpPendingSlaEmail", () => { dealName: "Sunrise ", dealNumber: "DFW-1-17326", pendingSinceLabel: "Jan 1, 2026, 9:00 AM CT", - hoursPending: 31, slaHours: 24, officeId: "office-uuid", frontendUrl: "https://trockcrm.com", }; - it("subject names the deal number + name", () => { + it("subject names the deal number + name, with a STABLE '>24h' (no live hours) for idempotent retries", () => { const email = buildRfpPendingSlaEmail(base); expect(email.subject).toContain("DFW-1-17326"); expect(email.subject.toLowerCase()).toContain("pending"); + expect(email.subject).toContain(">24h"); }); it("links to the deal with its officeId so a cross-office reviewer doesn't 404", () => { @@ -49,11 +49,10 @@ describe("buildRfpPendingSlaEmail", () => { expect(email.html).toContain("/deals/pending-rfp?officeId=office-uuid"); }); - it("states how long it's been pending and the SLA, and escapes the deal name", () => { + it("shows the stable pending-since instant + SLA (not a drifting live count) and escapes the deal name", () => { const email = buildRfpPendingSlaEmail(base); - expect(email.html).toContain("Jan 1, 2026, 9:00 AM CT"); - expect(email.html).toContain("31"); - expect(email.html).toContain("24"); + expect(email.html).toContain("Jan 1, 2026, 9:00 AM CT"); // stable per cycle + expect(email.html).toContain("More than 24 hours"); expect(email.html).toContain("Sunrise <Terraces>"); // escaped, not raw expect(email.html).not.toContain("Sunrise "); }); @@ -118,7 +117,9 @@ beforeAll(async () => { ('${U("db")}', 'OverrideApproving', '${OPP}', 'P-B', 'HS-B', false, 'pending', NOW() - INTERVAL '30 hours', true, false, false, NULL, 'approving'); `); query = (sql: string, params?: unknown[]) => pg.query(sql, params as never[]); -}); + // 30s (up from the 10s default): under the full `test:runtime` gate several PGlite suites set up + // concurrently (maxWorkers 4), and the shared cold-start can push this hook past 10s. +}, 30_000); afterAll(async () => { await pg?.close?.(); @@ -146,10 +147,11 @@ describe("findPendingRfpSlaBreaches (real SQL)", () => { }); }); -// ---- Orchestration test (mock query + send): exactly-once (record-after-send) + pre-send re-check ---- -function makeScanMocks(opts: { receiptExists?: boolean; stillBreaching?: boolean } = {}) { +// ---- Orchestration test (mock query + send): single-flight + exactly-once (record-after-send) + re-check ---- +function makeScanMocks(opts: { receiptExists?: boolean; stillBreaching?: boolean; stageSlug?: string } = {}) { const receiptExists = opts.receiptExists ?? false; const stillBreaching = opts.stillBreaching ?? true; + const stageSlug = opts.stageSlug ?? "opportunity"; const sendEmail = vi.fn().mockResolvedValue({ success: true, messageId: "msg-1" }); const calls: string[] = []; const q = vi.fn(async (sql: string) => { @@ -170,12 +172,16 @@ function makeScanMocks(opts: { receiptExists?: boolean; stillBreaching?: boolean ], }; } - // The pre-send re-check re-reads the deal row (no hours_pending). + // The pre-send re-check re-reads the deal row (no hours_pending) incl. the current stage slug. if (sql.includes(".deals") && sql.includes("WHERE d.id")) { return { - rows: stillBreaching - ? [{ status: "pending", requested_at: "2026-01-01T00:00:00.000Z", bbo: false, active: true, test_data: false, on_hold: false, override_decision: "", override_state: "" }] - : [{ status: "approved", requested_at: "2026-01-01T00:00:00.000Z", bbo: false, active: true, test_data: false, on_hold: false, override_decision: "", override_state: "" }], + rows: [{ + status: stillBreaching ? "pending" : "approved", + requested_at: "2026-01-01T00:00:00.000Z", + stage_slug: stageSlug, + bbo: false, active: true, test_data: false, on_hold: false, + override_decision: "", override_state: "", + }], }; } if (sql.includes("SELECT 1") && sql.includes("rfp_pending_sla_email_receipts")) { @@ -186,6 +192,8 @@ function makeScanMocks(opts: { receiptExists?: boolean; stillBreaching?: boolean return { sendEmail, q, calls }; } +// Always-acquires lock stub (no-op release) so tests exercise the scan body deterministically. +const noLock = async () => async () => {}; const enabledEnv = { RFP_PENDING_SLA_ENABLED: "true", RFP_REJECTION_EMAIL_RECIPIENTS: "boss@trock.test" }; const silent = { log: vi.fn(), warn: vi.fn(), error: vi.fn() }; @@ -194,7 +202,7 @@ describe("runRfpPendingSlaScan orchestration", () => { it("is inert (no query, no send) when the flag is off", async () => { const { sendEmail, q } = makeScanMocks(); - const summary = await runRfpPendingSlaScan({ query: q, sendEmail, env: { RFP_REJECTION_EMAIL_RECIPIENTS: "boss@trock.test" }, logger: silent }); + const summary = await runRfpPendingSlaScan({ query: q, sendEmail, acquireLock: noLock, env: { RFP_REJECTION_EMAIL_RECIPIENTS: "boss@trock.test" }, logger: silent }); expect(q).not.toHaveBeenCalled(); expect(sendEmail).not.toHaveBeenCalled(); expect(summary.sent).toBe(0); @@ -202,13 +210,21 @@ describe("runRfpPendingSlaScan orchestration", () => { it("does not send (and logs) when no leadership recipients are configured", async () => { const { sendEmail, q } = makeScanMocks(); - await runRfpPendingSlaScan({ query: q, sendEmail, env: { RFP_PENDING_SLA_ENABLED: "true" }, logger: silent }); + await runRfpPendingSlaScan({ query: q, sendEmail, acquireLock: noLock, env: { RFP_PENDING_SLA_ENABLED: "true" }, logger: silent }); expect(sendEmail).not.toHaveBeenCalled(); }); + it("skips the whole tick (no query, no send) when another scan already holds the lock", async () => { + const { sendEmail, q } = makeScanMocks(); + const summary = await runRfpPendingSlaScan({ query: q, sendEmail, acquireLock: async () => null, env: enabledEnv, logger: silent }); + expect(q).not.toHaveBeenCalled(); + expect(sendEmail).not.toHaveBeenCalled(); + expect(summary.sent).toBe(0); + }); + it("sends exactly once for a breaching deal, recording the receipt only AFTER a durable send", async () => { const { sendEmail, q, calls } = makeScanMocks(); - const summary = await runRfpPendingSlaScan({ query: q, sendEmail, env: enabledEnv, logger: silent }); + const summary = await runRfpPendingSlaScan({ query: q, sendEmail, acquireLock: noLock, env: enabledEnv, logger: silent }); expect(sendEmail).toHaveBeenCalledTimes(1); expect(sendEmail.mock.calls[0][0]).toEqual(["boss@trock.test"]); expect(summary.sent).toBe(1); @@ -219,14 +235,21 @@ describe("runRfpPendingSlaScan orchestration", () => { it("skips sending when a receipt already exists for this cycle", async () => { const { sendEmail, q } = makeScanMocks({ receiptExists: true }); - const summary = await runRfpPendingSlaScan({ query: q, sendEmail, env: enabledEnv, logger: silent }); + const summary = await runRfpPendingSlaScan({ query: q, sendEmail, acquireLock: noLock, env: enabledEnv, logger: silent }); expect(sendEmail).not.toHaveBeenCalled(); expect(summary.skipped).toBe(1); }); it("skips (no send) when the deal is no longer awaiting by the time we go to send", async () => { const { sendEmail, q } = makeScanMocks({ stillBreaching: false }); - const summary = await runRfpPendingSlaScan({ query: q, sendEmail, env: enabledEnv, logger: silent }); + const summary = await runRfpPendingSlaScan({ query: q, sendEmail, acquireLock: noLock, env: enabledEnv, logger: silent }); + expect(sendEmail).not.toHaveBeenCalled(); + expect(summary.skipped).toBe(1); + }); + + it("skips (no send) when the deal has moved out of Opportunity before the send", async () => { + const { sendEmail, q } = makeScanMocks({ stageSlug: "won" }); + const summary = await runRfpPendingSlaScan({ query: q, sendEmail, acquireLock: noLock, env: enabledEnv, logger: silent }); expect(sendEmail).not.toHaveBeenCalled(); expect(summary.skipped).toBe(1); }); @@ -234,7 +257,7 @@ describe("runRfpPendingSlaScan orchestration", () => { it("writes NO receipt (so the next scan retries) and does not throw when the send fails", async () => { const { q, calls } = makeScanMocks(); const sendEmail = vi.fn().mockRejectedValue(new Error("provider down")); - const summary = await runRfpPendingSlaScan({ query: q, sendEmail, env: enabledEnv, logger: silent }); + const summary = await runRfpPendingSlaScan({ query: q, sendEmail, acquireLock: noLock, env: enabledEnv, logger: silent }); expect(summary.failed).toBe(1); expect(calls.some((s) => s.includes("INSERT INTO public.rfp_pending_sla_email_receipts"))).toBe(false); }); From 8b6a76bc92c6cc0a4e6fa73fa8a49e70e893cfce Mon Sep 17 00:00:00 2001 From: Adnaan Iqbal <64221968+artificialadnaan@users.noreply.github.com> Date: Thu, 2 Jul 2026 21:16:33 -0500 Subject: [PATCH 4/8] fix(rfp): destroy the advisory-lock connection when unlock fails MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The SLA scan's single-flight advisory lock released its pooled client in a finally that always ran, so an unlock-query error returned a possibly-still- locked session to the pool. Release with the error (release(err)) on unlock failure so pg destroys the connection instead of reusing it — Postgres frees a session-level advisory lock when its backend disconnects. Co-Authored-By: Claude Opus 4.8 (1M context) --- worker/src/jobs/rfp-pending-sla.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/worker/src/jobs/rfp-pending-sla.ts b/worker/src/jobs/rfp-pending-sla.ts index 5f1be106e..70fe7f3b9 100644 --- a/worker/src/jobs/rfp-pending-sla.ts +++ b/worker/src/jobs/rfp-pending-sla.ts @@ -214,8 +214,13 @@ async function acquireScanAdvisoryLock(): Promise Promise)> return async () => { try { await client.query("SELECT pg_advisory_unlock($1)", [LOCK_KEY]); - } finally { client.release(); + } catch (err) { + // The unlock query failed — the session may still hold the advisory lock, so DESTROY the connection + // (release(err)) rather than returning a possibly-still-locked session to the pool. Postgres frees a + // session-level advisory lock when its backend disconnects, so dropping the client also drops the lock. + client.release(err as Error); + throw err; } }; } From 7c939e08a8559d36bf3ba68093b308364b4091a8 Mon Sep 17 00:00:00 2001 From: Adnaan Iqbal <64221968+artificialadnaan@users.noreply.github.com> Date: Thu, 2 Jul 2026 21:16:45 -0500 Subject: [PATCH 5/8] perf(rfp): resolve Opportunity stage ids once per SLA scan, not per office opportunityStageIds() reads office-invariant public.pipeline_stage_config but was called inside findPendingRfpSlaBreaches, which runs per office. Hoist the lookup to runRfpPendingSlaScan (once, before the office loop) and pass the ids into the finder via a new optional 4th param. Direct callers that omit it still resolve internally, so existing behavior + tests are unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) --- worker/src/jobs/rfp-pending-sla.ts | 18 +++++++--- .../jobs/rfp-pending-sla.runtime.test.ts | 36 ++++++++++++++++--- 2 files changed, 46 insertions(+), 8 deletions(-) diff --git a/worker/src/jobs/rfp-pending-sla.ts b/worker/src/jobs/rfp-pending-sla.ts index 70fe7f3b9..c09d6d01a 100644 --- a/worker/src/jobs/rfp-pending-sla.ts +++ b/worker/src/jobs/rfp-pending-sla.ts @@ -85,12 +85,16 @@ export async function findPendingRfpSlaBreaches( query: PgQuery, schemaName: string, slaHours: number, + oppStageIds?: string[], ): Promise { if (!TENANT_SCHEMA_REGEX.test(schemaName)) { throw new Error(`Unsafe tenant schema: ${schemaName}`); } - const oppStageIds = await opportunityStageIds(query); - if (oppStageIds.length === 0) return []; + // opportunityStageIds reads office-invariant public.pipeline_stage_config, so the caller resolves it ONCE + // per scan and passes it in (avoids one full-table read per office). Direct callers (tests) may omit it and + // we resolve internally. Either way the empty-set fast path stands. + const stageIds = oppStageIds ?? (await opportunityStageIds(query)); + if (stageIds.length === 0) return []; const result = await query( `SELECT d.id, d.name, @@ -109,7 +113,7 @@ export async function findPendingRfpSlaBreaches( AND COALESCE(d.rfp_override_decision, '') <> 'denial_reconfirmed' AND COALESCE(d.rfp_override_state, '') <> 'approving' ORDER BY d.rfp_approval_requested_at ASC`, - [oppStageIds, [...PENDING_RFP_AWAITING_STATUSES], String(slaHours)], + [stageIds, [...PENDING_RFP_AWAITING_STATUSES], String(slaHours)], ); return result.rows.map((row) => ({ dealId: String(row.id), @@ -161,6 +165,12 @@ export async function runRfpPendingSlaScan(deps: RunDeps = {}): Promise { const breaches = await findPendingRfpSlaBreaches(query, "office_test", 24); expect(breaches.every((b) => b.dealNumber?.startsWith("P-"))).toBe(true); }); + + it("uses caller-supplied oppStageIds when provided (hoisted once per scan), and fast-paths an empty set", async () => { + // Passing the full Opportunity-canonical set (as runRfpPendingSlaScan does) matches resolving internally. + const withIds = await findPendingRfpSlaBreaches(query, "office_test", 24, [OPP, DD]); + expect(withIds.map((b) => b.dealId)).toEqual([U("d2"), U("dd"), U("d1")]); + // The finder trusts the SUPPLIED ids: narrowing to just [OPP] drops the legacy dd-staged breach, proving + // the passed set (not an internal re-resolution) drives the stage filter. + const oppOnly = await findPendingRfpSlaBreaches(query, "office_test", 24, [OPP]); + expect(oppOnly.map((b) => b.dealId)).toEqual([U("d2"), U("d1")]); + // Empty set short-circuits before touching .deals. + expect(await findPendingRfpSlaBreaches(query, "office_test", 24, [])).toEqual([]); + }); }); -// ---- Orchestration test (mock query + send): single-flight + exactly-once (record-after-send) + re-check ---- -function makeScanMocks(opts: { receiptExists?: boolean; stillBreaching?: boolean; stageSlug?: string } = {}) { - const receiptExists = opts.receiptExists ?? false; +// ---- Orchestration test (mock query + send): single-flight + claim-then-send exactly-once + re-check ---- +// The receipts table is modeled as a real in-memory store keyed by cycle so the claim-before-send lifecycle +// (INSERT claim -> render from stored snapshot -> stamp sent_at on success) is exercised end-to-end, incl. +// the rename-stability guarantee. +function makeScanMocks( + opts: { stillBreaching?: boolean; stageSlug?: string; alreadySent?: boolean } = {}, +) { const stillBreaching = opts.stillBreaching ?? true; const stageSlug = opts.stageSlug ?? "opportunity"; + // Mutable so a test can rename the deal BETWEEN scans and assert the payload stays pinned to the snapshot. + const deal = { id: "deal-1", name: "Deal One", deal_number: "P-1", requestedAt: "2026-01-01T00:00:00.000Z" }; + // key = tenant_schema | deal_id | rfp_approval_requested_at (the first 3 bound params of every receipts query) + const receipts = new Map(); + const key = (p: unknown[]) => `${p[0]}|${p[1]}|${p[2]}`; + if (opts.alreadySent) { + receipts.set(`office_beta|${deal.id}|${deal.requestedAt}`, { + deal_name: "Deal One", + deal_number: "P-1", + sent_at: "2026-01-02T00:00:00.000Z", + }); + } const sendEmail = vi.fn().mockResolvedValue({ success: true, messageId: "msg-1" }); const calls: string[] = []; - const q = vi.fn(async (sql: string) => { + const q = vi.fn(async (sql: string, params?: unknown[]) => { calls.push(sql); if (sql.includes("FROM public.offices")) return { rows: [{ id: "office-1", slug: "beta" }] }; if (sql.includes("FROM public.pipeline_stage_config")) return { rows: [{ id: "opp-stage", slug: "opportunity" }] }; From 50049ebcb634913291861f6ac1fb9f58d883fe52 Mon Sep 17 00:00:00 2001 From: Adnaan Iqbal <64221968+artificialadnaan@users.noreply.github.com> Date: Thu, 2 Jul 2026 21:17:20 -0500 Subject: [PATCH 6/8] fix(rfp): persist a display snapshot before send so retries stay idempotent processBreach rendered the SLA email from the fresh-scan deal name/number and recorded the receipt only after send. A crash after a successful send but before the insert let the next scan re-read a renamed/renumbered deal and rebuild a different subject/body under the SAME Resend idempotencyKey (rejected or duplicated after TTL). Evolve the record-after-send into a retry-safe claim: insert a claim row (carrying deal_name/deal_number as first seen) BEFORE sending, render the email from that STORED snapshot, and stamp a nullable sent_at only after a durable send. The claim is never deleted; a failed send leaves sent_at NULL so the next scan retries with the same snapshot -> a stable payload. The single-flight advisory lock keeps the claim insert + sent_at check race-free. Migration 0174: add deal_name; make sent_at nullable (no default) so the row is a claim at insert and a receipt at completion. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../0174_rfp_pending_sla_email_receipts.sql | 6 +- worker/src/jobs/rfp-pending-sla.ts | 68 ++++++++++----- .../jobs/rfp-pending-sla.runtime.test.ts | 84 +++++++++++++++---- 3 files changed, 119 insertions(+), 39 deletions(-) diff --git a/migrations/0174_rfp_pending_sla_email_receipts.sql b/migrations/0174_rfp_pending_sla_email_receipts.sql index 13653cb60..4c67c97fb 100644 --- a/migrations/0174_rfp_pending_sla_email_receipts.sql +++ b/migrations/0174_rfp_pending_sla_email_receipts.sql @@ -20,10 +20,14 @@ CREATE TABLE IF NOT EXISTS public.rfp_pending_sla_email_receipts ( tenant_schema text NOT NULL, deal_id uuid NOT NULL, rfp_approval_requested_at timestamptz NOT NULL, + -- Display snapshot captured at claim time so a retry after a rename renders an identical Resend payload. + deal_name text, deal_number text, recipient_emails text, resend_message_id text, - sent_at timestamptz NOT NULL DEFAULT NOW(), + -- NULLABLE, no default: the row is a CLAIM at insert (sent_at NULL) and is stamped only after a durable + -- send. A failed send leaves it NULL so the next scan retries. + sent_at timestamptz, created_at timestamptz NOT NULL DEFAULT NOW(), updated_at timestamptz NOT NULL DEFAULT NOW(), PRIMARY KEY (tenant_schema, deal_id, rfp_approval_requested_at) diff --git a/worker/src/jobs/rfp-pending-sla.ts b/worker/src/jobs/rfp-pending-sla.ts index c09d6d01a..eae4d151b 100644 --- a/worker/src/jobs/rfp-pending-sla.ts +++ b/worker/src/jobs/rfp-pending-sla.ts @@ -129,9 +129,12 @@ export async function findPendingRfpSlaBreaches( * cycle. Exactly-once + retry-safe: * - a single-flight Postgres advisory lock serializes runs (a still-running or second-instance run skips), * so two runs can't both submit the same alert; - * - the receipt row (keyed tenant_schema, deal_id, rfp_approval_requested_at) is written only AFTER a - * durable send, so a crash before the send leaves no receipt and the next scan retries; - * - the email payload is stable per cycle, so a crash-then-retry reuses the same Resend idempotencyKey. + * - before sending, the scan writes a CLAIM row (keyed tenant_schema, deal_id, rfp_approval_requested_at) + * carrying a display snapshot (deal_name/deal_number as first seen); the claim is NEVER deleted and + * `sent_at` (nullable) marks completion. A failed send leaves sent_at NULL, so the next scan retries; + * - retries render the email from the STORED snapshot, so a rename/renumber between attempts can't change + * the payload — the same Resend idempotencyKey stays valid instead of being rejected as a key/payload + * mismatch. * A re-triggered RFP gets a fresh rfp_approval_requested_at -> a new cycle -> a new alert. */ export async function runRfpPendingSlaScan(deps: RunDeps = {}): Promise { @@ -250,26 +253,52 @@ async function processBreach( ): Promise<"sent" | "skipped" | "failed"> { const { tenantSchema, officeId, breach, recipients, frontendUrl, slaHours } = args; - // Already alerted for this pending cycle? The receipt is written only AFTER a durable send (below), so a - // crash between "decided to send" and "sent" leaves NO receipt and the next scan retries — the alert is - // never permanently suppressed. - const existing = await query( - `SELECT 1 FROM public.rfp_pending_sla_email_receipts + // The claim row (written BEFORE the send, never deleted) is the exactly-once ledger: sent_at IS NOT NULL + // means this cycle already alerted. sent_at NULL (or no row yet) means it still needs a send — a crash + // between claim and send, or a failed send, leaves sent_at NULL and the next scan retries. + const claimBefore = await query( + `SELECT deal_name, deal_number, sent_at + FROM public.rfp_pending_sla_email_receipts WHERE tenant_schema = $1 AND deal_id = $2::uuid AND rfp_approval_requested_at = $3 LIMIT 1`, [tenantSchema, breach.dealId, breach.requestedAt], ); - if (existing.rows.length > 0) return "skipped"; + if (claimBefore.rows[0]?.sent_at != null) return "skipped"; // Re-check the deal is STILL a breach right before sending: a reviewer may approve/decline (or an override // may resolve) between the batch scan and this send, and those transitions are async. Skip a stale alert. if (!(await isStillBreaching(query, tenantSchema, breach))) return "skipped"; + // Ensure a claim row exists carrying a STABLE display snapshot (deal_name/deal_number as first seen). On a + // crash-retry after a rename, the read-back below reuses these STORED values, so the rebuilt Resend payload + // is byte-identical and the idempotencyKey is honored rather than rejected as a key/payload mismatch. + await query( + `INSERT INTO public.rfp_pending_sla_email_receipts + (tenant_schema, deal_id, rfp_approval_requested_at, deal_name, deal_number, created_at, updated_at) + VALUES ($1, $2::uuid, $3, $4, $5, NOW(), NOW()) + ON CONFLICT (tenant_schema, deal_id, rfp_approval_requested_at) DO NOTHING`, + [tenantSchema, breach.dealId, breach.requestedAt, breach.dealName, breach.dealNumber], + ); + + // Read back the AUTHORITATIVE snapshot: the first-seen values win, even if THIS scan observed a renamed + // deal (its INSERT was a DO NOTHING no-op). Fall back to the fresh breach only if the row somehow vanished. + const claim = await query( + `SELECT deal_name, deal_number + FROM public.rfp_pending_sla_email_receipts + WHERE tenant_schema = $1 AND deal_id = $2::uuid AND rfp_approval_requested_at = $3 + LIMIT 1`, + [tenantSchema, breach.dealId, breach.requestedAt], + ); + const snapshotRow = claim.rows[0] ?? { deal_name: breach.dealName, deal_number: breach.dealNumber }; + const snapshotName = + typeof snapshotRow.deal_name === "string" && snapshotRow.deal_name.trim() ? snapshotRow.deal_name : "Deal"; + const snapshotNumber = snapshotRow.deal_number ?? null; + try { const email = buildRfpPendingSlaEmail({ dealId: breach.dealId, - dealName: breach.dealName, - dealNumber: breach.dealNumber, + dealName: snapshotName, + dealNumber: snapshotNumber, pendingSinceLabel: formatPendingSince(breach.requestedAt), slaHours, officeId, @@ -280,14 +309,13 @@ async function processBreach( idempotencyKey: `rfp-pending-sla-${tenantSchema}-${breach.dealId}-${breach.requestedAt}`, }); if (!result.success) throw new Error("Email provider returned unsuccessful result"); - // Record the receipt ONLY after the send is durable, so a failed send never suppresses a retry. ON - // CONFLICT DO NOTHING guards the (single-worker) case of an overlapping run that already recorded it. + // Mark the claim complete only AFTER a durable send. `sent_at IS NULL` guards against a double-complete + // (e.g. an overlapping run in the single-worker case) — the first completer wins. await query( - `INSERT INTO public.rfp_pending_sla_email_receipts - (tenant_schema, deal_id, rfp_approval_requested_at, deal_number, recipient_emails, resend_message_id, sent_at, created_at, updated_at) - VALUES ($1, $2::uuid, $3, $4, $5, $6, NOW(), NOW(), NOW()) - ON CONFLICT (tenant_schema, deal_id, rfp_approval_requested_at) DO NOTHING`, - [tenantSchema, breach.dealId, breach.requestedAt, breach.dealNumber, recipients.join(", "), result.messageId], + `UPDATE public.rfp_pending_sla_email_receipts + SET sent_at = NOW(), recipient_emails = $4, resend_message_id = $5, updated_at = NOW() + WHERE tenant_schema = $1 AND deal_id = $2::uuid AND rfp_approval_requested_at = $3 AND sent_at IS NULL`, + [tenantSchema, breach.dealId, breach.requestedAt, recipients.join(", "), result.messageId], ); logger.log("[RfpPendingSla] Sent SLA breach alert", { tenantSchema, @@ -297,8 +325,8 @@ async function processBreach( }); return "sent"; } catch (err) { - // No receipt was written, so the next hourly scan retries. One deal's failure must not abort the rest of - // the scan, so we swallow + log here. + // The claim stays with sent_at NULL (not deleted), so the next hourly scan retries with the SAME stored + // snapshot -> a stable Resend payload. One deal's failure must not abort the scan, so we swallow + log. logger.error("[RfpPendingSla] Failed to send SLA breach alert", { tenantSchema, dealId: breach.dealId, diff --git a/worker/tests/jobs/rfp-pending-sla.runtime.test.ts b/worker/tests/jobs/rfp-pending-sla.runtime.test.ts index a4c632c79..ccc0defa9 100644 --- a/worker/tests/jobs/rfp-pending-sla.runtime.test.ts +++ b/worker/tests/jobs/rfp-pending-sla.runtime.test.ts @@ -191,10 +191,10 @@ function makeScanMocks( return { rows: [ { - id: "deal-1", - name: "Deal One", - deal_number: "P-1", - rfp_approval_requested_at: "2026-01-01T00:00:00.000Z", + id: deal.id, + name: deal.name, + deal_number: deal.deal_number, + rfp_approval_requested_at: deal.requestedAt, hours_pending: 30, }, ], @@ -205,19 +205,35 @@ function makeScanMocks( return { rows: [{ status: stillBreaching ? "pending" : "approved", - requested_at: "2026-01-01T00:00:00.000Z", + requested_at: deal.requestedAt, stage_slug: stageSlug, bbo: false, active: true, test_data: false, on_hold: false, override_decision: "", override_state: "", }], }; } - if (sql.includes("SELECT 1") && sql.includes("rfp_pending_sla_email_receipts")) { - return { rows: receiptExists ? [{ "?column?": 1 }] : [] }; + if (sql.includes("rfp_pending_sla_email_receipts")) { + const k = key(params ?? []); + if (sql.includes("INSERT")) { + // (tenant, deal, requested_at, deal_name, deal_number) — ON CONFLICT DO NOTHING preserves first-seen. + if (!receipts.has(k)) { + receipts.set(k, { deal_name: params?.[3] as string, deal_number: (params?.[4] as string) ?? null, sent_at: null }); + } + return { rows: [] }; + } + if (sql.includes("UPDATE")) { + // SET sent_at = NOW() ... WHERE (pk) AND sent_at IS NULL — first completer wins. + const row = receipts.get(k); + if (row && row.sent_at == null) row.sent_at = new Date().toISOString(); + return { rows: [] }; + } + // SELECT (claim read-back / already-sent check) + const row = receipts.get(k); + return { rows: row ? [row] : [] }; } return { rows: [] }; }); - return { sendEmail, q, calls }; + return { sendEmail, q, calls, receipts, deal }; } // Always-acquires lock stub (no-op release) so tests exercise the scan body deterministically. @@ -250,19 +266,23 @@ describe("runRfpPendingSlaScan orchestration", () => { expect(summary.sent).toBe(0); }); - it("sends exactly once for a breaching deal, recording the receipt only AFTER a durable send", async () => { - const { sendEmail, q, calls } = makeScanMocks(); + it("claims BEFORE sending, then stamps sent_at AFTER a durable send (exactly-once)", async () => { + const { sendEmail, q, calls, receipts } = makeScanMocks(); const summary = await runRfpPendingSlaScan({ query: q, sendEmail, acquireLock: noLock, env: enabledEnv, logger: silent }); expect(sendEmail).toHaveBeenCalledTimes(1); expect(sendEmail.mock.calls[0][0]).toEqual(["boss@trock.test"]); expect(summary.sent).toBe(1); - // The receipt INSERT happens after the send resolves. - const sendIdx = calls.findIndex((s) => s.includes("INSERT INTO public.rfp_pending_sla_email_receipts")); - expect(sendIdx).toBeGreaterThan(-1); + // The claim INSERT precedes the send; the sent_at UPDATE follows it. + const insertIdx = calls.findIndex((s) => s.includes("INSERT INTO public.rfp_pending_sla_email_receipts")); + const updateIdx = calls.findIndex((s) => s.includes("UPDATE public.rfp_pending_sla_email_receipts")); + expect(insertIdx).toBeGreaterThan(-1); + expect(updateIdx).toBeGreaterThan(insertIdx); + // The claim row now records completion. + expect(receipts.get("office_beta|deal-1|2026-01-01T00:00:00.000Z")?.sent_at).not.toBeNull(); }); - it("skips sending when a receipt already exists for this cycle", async () => { - const { sendEmail, q } = makeScanMocks({ receiptExists: true }); + it("skips sending when the claim for this cycle is already sent (sent_at not null)", async () => { + const { sendEmail, q } = makeScanMocks({ alreadySent: true }); const summary = await runRfpPendingSlaScan({ query: q, sendEmail, acquireLock: noLock, env: enabledEnv, logger: silent }); expect(sendEmail).not.toHaveBeenCalled(); expect(summary.skipped).toBe(1); @@ -282,11 +302,39 @@ describe("runRfpPendingSlaScan orchestration", () => { expect(summary.skipped).toBe(1); }); - it("writes NO receipt (so the next scan retries) and does not throw when the send fails", async () => { - const { q, calls } = makeScanMocks(); + it("leaves the claim with sent_at NULL (retryable, never deleted) and does not throw when the send fails", async () => { + const { q, calls, receipts } = makeScanMocks(); const sendEmail = vi.fn().mockRejectedValue(new Error("provider down")); const summary = await runRfpPendingSlaScan({ query: q, sendEmail, acquireLock: noLock, env: enabledEnv, logger: silent }); expect(summary.failed).toBe(1); - expect(calls.some((s) => s.includes("INSERT INTO public.rfp_pending_sla_email_receipts"))).toBe(false); + // The claim WAS written (so the snapshot is pinned), but it is NOT marked sent and NOT deleted. + expect(calls.some((s) => s.includes("INSERT INTO public.rfp_pending_sla_email_receipts"))).toBe(true); + expect(calls.some((s) => s.includes("UPDATE public.rfp_pending_sla_email_receipts"))).toBe(false); + expect(receipts.get("office_beta|deal-1|2026-01-01T00:00:00.000Z")?.sent_at).toBeNull(); + }); + + it("retries render from the STORED snapshot, so a rename between attempts keeps the Resend payload stable", async () => { + const mocks = makeScanMocks(); + // First attempt: the send fails, so the claim persists with the FIRST-SEEN snapshot and sent_at NULL. + const failing = vi.fn().mockRejectedValue(new Error("provider down")); + await runRfpPendingSlaScan({ query: mocks.q, sendEmail: failing, acquireLock: noLock, env: enabledEnv, logger: silent }); + expect(failing).toHaveBeenCalledTimes(1); + + // The deal is renamed + renumbered between scans. + mocks.deal.name = "Renamed Deal"; + mocks.deal.deal_number = "P-999"; + + // Second attempt succeeds: it must render from the STORED snapshot, not the fresh (renamed) fields. + const sending = vi.fn().mockResolvedValue({ success: true, messageId: "msg-2" }); + const summary = await runRfpPendingSlaScan({ query: mocks.q, sendEmail: sending, acquireLock: noLock, env: enabledEnv, logger: silent }); + expect(summary.sent).toBe(1); + const [, subject, html, options] = sending.mock.calls[0]; + expect(subject).toContain("P-1"); // original number, pinned + expect(subject).toContain("Deal One"); // original name, pinned + expect(subject).not.toContain("P-999"); + expect(html).toContain("Deal One"); + expect(html).not.toContain("Renamed Deal"); + // Same idempotencyKey across both attempts (keyed on requested_at, which never changed). + expect(options.idempotencyKey).toBe(failing.mock.calls[0][3].idempotencyKey); }); }); From be274f21a74d067d130558f9f4bf5396bd08287a Mon Sep 17 00:00:00 2001 From: Adnaan Iqbal <64221968+artificialadnaan@users.noreply.github.com> Date: Thu, 2 Jul 2026 21:17:27 -0500 Subject: [PATCH 7/8] docs(rfp): rewrite the 0174 header to the claim-then-send protocol The header still described the removed protocol ("INSERT a claim ON CONFLICT DO NOTHING before sending, and a send failure DELETEs the claim"). Rewrite it to the actual protocol: a single-flight advisory lock serializes runs; a claim row (with a display snapshot) is inserted before send and is NEVER deleted; sent_at (nullable) marks completion, so a failed send leaves sent_at NULL and the next scan retries with the SAME snapshot for a stable Resend payload. Co-Authored-By: Claude Opus 4.8 (1M context) --- migrations/0174_rfp_pending_sla_email_receipts.sql | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/migrations/0174_rfp_pending_sla_email_receipts.sql b/migrations/0174_rfp_pending_sla_email_receipts.sql index 4c67c97fb..3f72bd904 100644 --- a/migrations/0174_rfp_pending_sla_email_receipts.sql +++ b/migrations/0174_rfp_pending_sla_email_receipts.sql @@ -6,8 +6,14 @@ -- rfp_approval_status in pending_outbox/pending) longer than 24h should alert leadership. Unlike the -- decline email (0148), an SLA breach is TIME-based, not a status transition, so no DB trigger fires it — -- the worker runs an hourly cron scan instead. Scanning hourly means the same breach is seen many times, so --- this ledger is the exactly-once guard: the scan INSERTs a claim row ON CONFLICT DO NOTHING BEFORE sending, --- and a send failure DELETEs the claim so the next scan retries. +-- this ledger is the exactly-once guard, and the protocol is: +-- 1. a single-flight Postgres advisory lock serializes runs cross-instance (no two scans race a send); +-- 2. BEFORE sending, the scan INSERTs a CLAIM row (ON CONFLICT DO NOTHING) carrying a display snapshot +-- (deal_name/deal_number as first seen). The claim is NEVER deleted. +-- 3. `sent_at` (nullable, no default) marks completion: it is set only AFTER a durable send. A crash +-- between claim and send, or a failed send, leaves sent_at NULL, so the next scan retries. +-- Because retries render the email from the STORED snapshot, a rename/renumber between attempts can't change +-- the payload — the same Resend idempotencyKey stays valid instead of being rejected as a key/payload mismatch. -- -- CYCLE IDENTITY: keyed (tenant_schema, deal_id, rfp_approval_requested_at). rfp_approval_requested_at is -- the instant the RFP entered pending; a re-triggered RFP gets a NEW requested_at -> a fresh cycle -> a new From 2ad9904312a7cf7d8c8dfb598083c6c53d9915e5 Mon Sep 17 00:00:00 2001 From: Adnaan Iqbal <64221968+artificialadnaan@users.noreply.github.com> Date: Thu, 2 Jul 2026 22:12:29 -0500 Subject: [PATCH 8/8] fix(rfp-sla): freeze recipient set in the claim snapshot for idempotent retries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The SLA breach email re-resolved RFP_REJECTION_EMAIL_RECIPIENTS on every attempt, so an env edit/reorder between a failed send and its retry changed the Resend payload's `to` while the idempotencyKey stayed the same — the provider then rejects it as a key/payload mismatch (delayed or dropped alert). Snapshot the recipient list into the claim row at claim time (the recipient_emails column already existed) alongside deal_name/deal_number, read it back, and send to the STORED set. The success UPDATE no longer re-writes recipient_emails, so the row records what was actually sent. Adds a retry-stability test covering a recipient-list change between attempts. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../0174_rfp_pending_sla_email_receipts.sql | 12 +++--- worker/src/jobs/rfp-pending-sla.ts | 39 ++++++++++++------- .../jobs/rfp-pending-sla.runtime.test.ts | 36 +++++++++++++++-- 3 files changed, 65 insertions(+), 22 deletions(-) diff --git a/migrations/0174_rfp_pending_sla_email_receipts.sql b/migrations/0174_rfp_pending_sla_email_receipts.sql index 3f72bd904..6d459f280 100644 --- a/migrations/0174_rfp_pending_sla_email_receipts.sql +++ b/migrations/0174_rfp_pending_sla_email_receipts.sql @@ -8,12 +8,13 @@ -- the worker runs an hourly cron scan instead. Scanning hourly means the same breach is seen many times, so -- this ledger is the exactly-once guard, and the protocol is: -- 1. a single-flight Postgres advisory lock serializes runs cross-instance (no two scans race a send); --- 2. BEFORE sending, the scan INSERTs a CLAIM row (ON CONFLICT DO NOTHING) carrying a display snapshot --- (deal_name/deal_number as first seen). The claim is NEVER deleted. +-- 2. BEFORE sending, the scan INSERTs a CLAIM row (ON CONFLICT DO NOTHING) carrying a full snapshot +-- (deal_name/deal_number AND recipient_emails as first seen). The claim is NEVER deleted. -- 3. `sent_at` (nullable, no default) marks completion: it is set only AFTER a durable send. A crash -- between claim and send, or a failed send, leaves sent_at NULL, so the next scan retries. --- Because retries render the email from the STORED snapshot, a rename/renumber between attempts can't change --- the payload — the same Resend idempotencyKey stays valid instead of being rejected as a key/payload mismatch. +-- Because retries render the email from the STORED snapshot — subject/body from the deal fields AND the `to` +-- from recipient_emails — a rename/renumber OR a RFP_REJECTION_EMAIL_RECIPIENTS edit between attempts can't +-- change the payload; the same Resend idempotencyKey stays valid instead of being rejected as a mismatch. -- -- CYCLE IDENTITY: keyed (tenant_schema, deal_id, rfp_approval_requested_at). rfp_approval_requested_at is -- the instant the RFP entered pending; a re-triggered RFP gets a NEW requested_at -> a fresh cycle -> a new @@ -26,7 +27,8 @@ CREATE TABLE IF NOT EXISTS public.rfp_pending_sla_email_receipts ( tenant_schema text NOT NULL, deal_id uuid NOT NULL, rfp_approval_requested_at timestamptz NOT NULL, - -- Display snapshot captured at claim time so a retry after a rename renders an identical Resend payload. + -- Snapshot captured at claim time so a retry (after a rename OR a recipient-list edit) renders an identical + -- Resend payload. deal_name/deal_number drive subject+body; recipient_emails (comma-joined) is the frozen `to`. deal_name text, deal_number text, recipient_emails text, diff --git a/worker/src/jobs/rfp-pending-sla.ts b/worker/src/jobs/rfp-pending-sla.ts index eae4d151b..8abfcf9d4 100644 --- a/worker/src/jobs/rfp-pending-sla.ts +++ b/worker/src/jobs/rfp-pending-sla.ts @@ -269,30 +269,40 @@ async function processBreach( // may resolve) between the batch scan and this send, and those transitions are async. Skip a stale alert. if (!(await isStillBreaching(query, tenantSchema, breach))) return "skipped"; - // Ensure a claim row exists carrying a STABLE display snapshot (deal_name/deal_number as first seen). On a - // crash-retry after a rename, the read-back below reuses these STORED values, so the rebuilt Resend payload - // is byte-identical and the idempotencyKey is honored rather than rejected as a key/payload mismatch. + // Ensure a claim row exists carrying a STABLE snapshot (deal_name/deal_number AND the recipient set, as first + // seen). On a crash-retry after a rename OR a RFP_REJECTION_EMAIL_RECIPIENTS edit/reorder, the read-back below + // reuses these STORED values, so the rebuilt Resend payload (subject/body AND `to`) is byte-identical and the + // idempotencyKey is honored rather than rejected as a key/payload mismatch (which would delay or duplicate). await query( `INSERT INTO public.rfp_pending_sla_email_receipts - (tenant_schema, deal_id, rfp_approval_requested_at, deal_name, deal_number, created_at, updated_at) - VALUES ($1, $2::uuid, $3, $4, $5, NOW(), NOW()) + (tenant_schema, deal_id, rfp_approval_requested_at, deal_name, deal_number, recipient_emails, created_at, updated_at) + VALUES ($1, $2::uuid, $3, $4, $5, $6, NOW(), NOW()) ON CONFLICT (tenant_schema, deal_id, rfp_approval_requested_at) DO NOTHING`, - [tenantSchema, breach.dealId, breach.requestedAt, breach.dealName, breach.dealNumber], + [tenantSchema, breach.dealId, breach.requestedAt, breach.dealName, breach.dealNumber, recipients.join(", ")], ); - // Read back the AUTHORITATIVE snapshot: the first-seen values win, even if THIS scan observed a renamed - // deal (its INSERT was a DO NOTHING no-op). Fall back to the fresh breach only if the row somehow vanished. + // Read back the AUTHORITATIVE snapshot: the first-seen values win, even if THIS scan observed a renamed deal + // or a changed recipient list (its INSERT was a DO NOTHING no-op). Fall back to the fresh scan only if the row + // somehow vanished. const claim = await query( - `SELECT deal_name, deal_number + `SELECT deal_name, deal_number, recipient_emails FROM public.rfp_pending_sla_email_receipts WHERE tenant_schema = $1 AND deal_id = $2::uuid AND rfp_approval_requested_at = $3 LIMIT 1`, [tenantSchema, breach.dealId, breach.requestedAt], ); - const snapshotRow = claim.rows[0] ?? { deal_name: breach.dealName, deal_number: breach.dealNumber }; + const snapshotRow = claim.rows[0] ?? { + deal_name: breach.dealName, + deal_number: breach.dealNumber, + recipient_emails: recipients.join(", "), + }; const snapshotName = typeof snapshotRow.deal_name === "string" && snapshotRow.deal_name.trim() ? snapshotRow.deal_name : "Deal"; const snapshotNumber = snapshotRow.deal_number ?? null; + const snapshotRecipients = + typeof snapshotRow.recipient_emails === "string" && snapshotRow.recipient_emails.trim() + ? snapshotRow.recipient_emails.split(",").map((e: string) => e.trim()).filter(Boolean) + : recipients; try { const email = buildRfpPendingSlaEmail({ @@ -304,18 +314,19 @@ async function processBreach( officeId, frontendUrl, }); - const result = await sendEmail(recipients, email.subject, email.html, { + const result = await sendEmail(snapshotRecipients, email.subject, email.html, { text: email.text, idempotencyKey: `rfp-pending-sla-${tenantSchema}-${breach.dealId}-${breach.requestedAt}`, }); if (!result.success) throw new Error("Email provider returned unsuccessful result"); // Mark the claim complete only AFTER a durable send. `sent_at IS NULL` guards against a double-complete - // (e.g. an overlapping run in the single-worker case) — the first completer wins. + // (e.g. an overlapping run in the single-worker case) — the first completer wins. recipient_emails is NOT + // re-written here: the claim's snapshot (set at claim time, and what we actually sent to) stays authoritative. await query( `UPDATE public.rfp_pending_sla_email_receipts - SET sent_at = NOW(), recipient_emails = $4, resend_message_id = $5, updated_at = NOW() + SET sent_at = NOW(), resend_message_id = $4, updated_at = NOW() WHERE tenant_schema = $1 AND deal_id = $2::uuid AND rfp_approval_requested_at = $3 AND sent_at IS NULL`, - [tenantSchema, breach.dealId, breach.requestedAt, recipients.join(", "), result.messageId], + [tenantSchema, breach.dealId, breach.requestedAt, result.messageId], ); logger.log("[RfpPendingSla] Sent SLA breach alert", { tenantSchema, diff --git a/worker/tests/jobs/rfp-pending-sla.runtime.test.ts b/worker/tests/jobs/rfp-pending-sla.runtime.test.ts index ccc0defa9..c2d932dd0 100644 --- a/worker/tests/jobs/rfp-pending-sla.runtime.test.ts +++ b/worker/tests/jobs/rfp-pending-sla.runtime.test.ts @@ -171,12 +171,16 @@ function makeScanMocks( // Mutable so a test can rename the deal BETWEEN scans and assert the payload stays pinned to the snapshot. const deal = { id: "deal-1", name: "Deal One", deal_number: "P-1", requestedAt: "2026-01-01T00:00:00.000Z" }; // key = tenant_schema | deal_id | rfp_approval_requested_at (the first 3 bound params of every receipts query) - const receipts = new Map(); + const receipts = new Map< + string, + { deal_name: string; deal_number: string | null; recipient_emails: string | null; sent_at: string | null } + >(); const key = (p: unknown[]) => `${p[0]}|${p[1]}|${p[2]}`; if (opts.alreadySent) { receipts.set(`office_beta|${deal.id}|${deal.requestedAt}`, { deal_name: "Deal One", deal_number: "P-1", + recipient_emails: "boss@trock.test", sent_at: "2026-01-02T00:00:00.000Z", }); } @@ -215,9 +219,15 @@ function makeScanMocks( if (sql.includes("rfp_pending_sla_email_receipts")) { const k = key(params ?? []); if (sql.includes("INSERT")) { - // (tenant, deal, requested_at, deal_name, deal_number) — ON CONFLICT DO NOTHING preserves first-seen. + // (tenant, deal, requested_at, deal_name, deal_number, recipient_emails) — ON CONFLICT DO NOTHING + // preserves first-seen (the full snapshot: display fields AND the recipient list). if (!receipts.has(k)) { - receipts.set(k, { deal_name: params?.[3] as string, deal_number: (params?.[4] as string) ?? null, sent_at: null }); + receipts.set(k, { + deal_name: params?.[3] as string, + deal_number: (params?.[4] as string) ?? null, + recipient_emails: (params?.[5] as string) ?? null, + sent_at: null, + }); } return { rows: [] }; } @@ -337,4 +347,24 @@ describe("runRfpPendingSlaScan orchestration", () => { // Same idempotencyKey across both attempts (keyed on requested_at, which never changed). expect(options.idempotencyKey).toBe(failing.mock.calls[0][3].idempotencyKey); }); + + it("retries send to the STORED recipient snapshot, so a recipient-list change between attempts keeps `to` stable", async () => { + const mocks = makeScanMocks(); + // First attempt: send fails, so the claim persists the FIRST-SEEN recipient set with sent_at NULL. + const failing = vi.fn().mockRejectedValue(new Error("provider down")); + await runRfpPendingSlaScan({ query: mocks.q, sendEmail: failing, acquireLock: noLock, env: enabledEnv, logger: silent }); + expect(failing).toHaveBeenCalledTimes(1); + expect(failing.mock.calls[0][0]).toEqual(["boss@trock.test"]); + + // RFP_REJECTION_EMAIL_RECIPIENTS is edited between scans (reordered + a new address). + const changedEnv = { RFP_PENDING_SLA_ENABLED: "true", RFP_REJECTION_EMAIL_RECIPIENTS: "newboss@trock.test, boss@trock.test" }; + + // Second attempt succeeds: it must send to the STORED recipient snapshot, not the fresh env list, so the + // Resend payload (`to`) is byte-identical to the first attempt and the idempotencyKey is honored. + const sending = vi.fn().mockResolvedValue({ success: true, messageId: "msg-2" }); + const summary = await runRfpPendingSlaScan({ query: mocks.q, sendEmail: sending, acquireLock: noLock, env: changedEnv, logger: silent }); + expect(summary.sent).toBe(1); + expect(sending.mock.calls[0][0]).toEqual(["boss@trock.test"]); // pinned to the snapshot, not newboss + expect(sending.mock.calls[0][3].idempotencyKey).toBe(failing.mock.calls[0][3].idempotencyKey); + }); });