diff --git a/apps/api/src/automation/automation.service.ts b/apps/api/src/automation/automation.service.ts index cea8d326..983ded73 100644 --- a/apps/api/src/automation/automation.service.ts +++ b/apps/api/src/automation/automation.service.ts @@ -114,6 +114,21 @@ export class AutomationService implements OnModuleInit { ) this.logger.log('Scheduled recurring automation gate nudge job') + // refs #1466 P3c — hourly insurance-prompt autopilot. Nudges the advisor + // first when a booked trip still has pending insurance past the grace + // window; on a later sweep (after the nudge grace) it auto-sends the + // client insurance proposal via the EXISTING queueProposalEmails path + // (zero new send code, inherits the #1048 cutover + MIGRATION_SILENT_MODE + // guards). Idempotent per traveler via the deterministic insurance jobId. + await this.scheduleRecurring( + QUEUES.CLIENT_CARE, + 'recurring.insurance_prompt_autopilot', + { type: 'recurring.insurance_prompt_autopilot' }, + '0 * * * *', // Every hour + { jobId: 'recurring:insurance_prompt_autopilot' }, + ) + this.logger.log('Scheduled recurring insurance prompt autopilot job') + // Contact lifecycle: returned → awaiting_next after 30 days await this.scheduleRecurring( QUEUES.CLIENT_CARE, diff --git a/apps/api/src/automation/automation.types.ts b/apps/api/src/automation/automation.types.ts index 98bd6ba0..dd1734a3 100644 --- a/apps/api/src/automation/automation.types.ts +++ b/apps/api/src/automation/automation.types.ts @@ -80,6 +80,10 @@ export const JOB_TYPES = { // refs #1466 P2 — hourly sweep nudging advisors about gate automations // (proposed/paused) nearing their scheduled send time. RECURRING_AUTOMATION_GATE_NUDGE: 'recurring.automation_gate_nudge', + // refs #1466 P3c — hourly autopilot fallback: nudges the advisor first, then + // (on a later sweep) auto-sends the insurance proposal for still-pending + // travelers past the booking grace window. + RECURRING_INSURANCE_PROMPT_AUTOPILOT: 'recurring.insurance_prompt_autopilot', // OCR processing jobs OCR_EXTRACT: 'ocr.extract', @@ -305,6 +309,7 @@ export interface RecurringJobData { | 'recurring.task_assignment_digest' | 'recurring.task_due_reminder' | 'recurring.automation_gate_nudge' + | 'recurring.insurance_prompt_autopilot' agencyId?: string // Optional - if not provided, runs for all agencies } diff --git a/apps/api/src/automation/insurance-autopilot.spec.ts b/apps/api/src/automation/insurance-autopilot.spec.ts new file mode 100644 index 00000000..c11a9cf9 --- /dev/null +++ b/apps/api/src/automation/insurance-autopilot.spec.ts @@ -0,0 +1,115 @@ +/** + * refs #1466 P3c — pure selection/grouping/staging helpers for the insurance + * autopilot. No DB, no clock: proves the idempotency fence (existing job → + * excluded), the advisor-required filter, deterministic grouping, the two grace + * windows, and the stable advisor-scoped dedupe key. + */ +import { + getInsurancePromptGraceHours, + getInsuranceNudgeGraceHours, + buildAutopilotNudgeDedupeKey, + buildAutopilotNudgeCopy, + groupAutopilotCandidates, + isNudgeGraceElapsed, + type AutopilotCandidateRow, +} from './insurance-autopilot' +import { getInsuranceProposalJobId } from './automation.types' + +const jobIdFor = getInsuranceProposalJobId + +function row( + tripId: string, + tripTravelerId: string, + advisorId: string | null = 'advisor-1', +): AutopilotCandidateRow { + return { tripId, tripTravelerId, advisorId, agencyId: 'agency-1' } +} + +describe('insurance-autopilot — grace resolution', () => { + it('uses the env override when positive', () => { + expect(getInsurancePromptGraceHours({ INSURANCE_PROMPT_GRACE_HOURS: '12' } as never)).toBe(12) + expect(getInsuranceNudgeGraceHours({ INSURANCE_PROMPT_NUDGE_GRACE_HOURS: '6' } as never)).toBe(6) + }) + + it('falls back to the default for missing / zero / non-numeric env', () => { + expect(getInsurancePromptGraceHours({} as never)).toBe(72) + expect(getInsurancePromptGraceHours({ INSURANCE_PROMPT_GRACE_HOURS: '0' } as never)).toBe(72) + expect(getInsurancePromptGraceHours({ INSURANCE_PROMPT_GRACE_HOURS: 'abc' } as never)).toBe(72) + expect(getInsuranceNudgeGraceHours({} as never)).toBe(24) + expect(getInsuranceNudgeGraceHours({ INSURANCE_PROMPT_NUDGE_GRACE_HOURS: '-4' } as never)).toBe(24) + }) +}) + +describe('insurance-autopilot — dedupe key', () => { + it('is stable and embeds the advisor id (global dedupe index)', () => { + expect(buildAutopilotNudgeDedupeKey('advisor-1', 'trip-1')).toBe( + 'insurance-autopilot:advisor-1:trip-1', + ) + // Two advisors on the same trip never collide. + expect(buildAutopilotNudgeDedupeKey('advisor-2', 'trip-1')).not.toBe( + buildAutopilotNudgeDedupeKey('advisor-1', 'trip-1'), + ) + }) +}) + +describe('insurance-autopilot — groupAutopilotCandidates', () => { + it('groups pending travelers by trip (deterministic order)', () => { + const groups = groupAutopilotCandidates( + [row('trip-b', 't2'), row('trip-a', 't1'), row('trip-a', 't0')], + new Set(), + jobIdFor, + ) + expect(groups.map((g) => g.tripId)).toEqual(['trip-a', 'trip-b']) + expect(groups[0]!.travelerIds).toEqual(['t0', 't1']) // sorted + }) + + it('excludes a traveler whose insurance proposal job already exists (idempotency)', () => { + const existing = new Set([jobIdFor('trip-a', 't1')]) + const groups = groupAutopilotCandidates([row('trip-a', 't1'), row('trip-a', 't2')], existing, jobIdFor) + expect(groups).toHaveLength(1) + expect(groups[0]!.travelerIds).toEqual(['t2']) + }) + + it('drops a trip entirely when every traveler already has a job', () => { + const existing = new Set([jobIdFor('trip-a', 't1'), jobIdFor('trip-a', 't2')]) + const groups = groupAutopilotCandidates([row('trip-a', 't1'), row('trip-a', 't2')], existing, jobIdFor) + expect(groups).toHaveLength(0) + }) + + it('skips a trip with no advisor/owner to nudge', () => { + const groups = groupAutopilotCandidates([row('trip-a', 't1', null)], new Set(), jobIdFor) + expect(groups).toHaveLength(0) + }) +}) + +describe('insurance-autopilot — isNudgeGraceElapsed', () => { + const now = new Date('2026-07-05T12:00:00Z') + + it('is false when no marker exists (advisor must be nudged first)', () => { + expect(isNudgeGraceElapsed(null, now, 24)).toBe(false) + expect(isNudgeGraceElapsed(undefined, now, 24)).toBe(false) + }) + + it('is false before the grace window elapses', () => { + const nudgedAt = new Date(now.getTime() - 23 * 60 * 60 * 1000) // 23h ago + expect(isNudgeGraceElapsed(nudgedAt, now, 24)).toBe(false) + }) + + it('is true once the grace window has elapsed (inclusive boundary)', () => { + const exactly = new Date(now.getTime() - 24 * 60 * 60 * 1000) + const past = new Date(now.getTime() - 48 * 60 * 60 * 1000) + expect(isNudgeGraceElapsed(exactly, now, 24)).toBe(true) + expect(isNudgeGraceElapsed(past, now, 24)).toBe(true) + }) +}) + +describe('insurance-autopilot — nudge copy', () => { + it('pluralizes on traveler count', () => { + const one = buildAutopilotNudgeCopy({ tripId: 't', advisorId: 'a', agencyId: 'g', travelerIds: ['x'] }) + const two = buildAutopilotNudgeCopy({ tripId: 't', advisorId: 'a', agencyId: 'g', travelerIds: ['x', 'y'] }) + expect(one.body).toContain('1 traveler ') + expect(one.body).toContain('has no insurance') + expect(two.body).toContain('2 travelers ') + expect(two.body).toContain('have no insurance') + }) +}) diff --git a/apps/api/src/automation/insurance-autopilot.ts b/apps/api/src/automation/insurance-autopilot.ts new file mode 100644 index 00000000..7c7ea1d1 --- /dev/null +++ b/apps/api/src/automation/insurance-autopilot.ts @@ -0,0 +1,159 @@ +/** + * refs #1466 P3c — Autopilot fallback for the insurance prompt (the ONLY net-new + * client send in the Automation Gate epic). + * + * Pure selection / grouping / staging helpers for the hourly + * `recurring.insurance_prompt_autopilot` sweep. The processor owns the DB reads, + * the advisor nudge, and the (verbatim) call into + * `InsuranceAutomationService.queueProposalEmails`; these functions decide WHICH + * travelers qualify and — given the advisor-nudge marker — WHETHER a trip is + * ready to auto-send. Kept pure so the two-stage (advisor-first) logic and the + * grace math are unit-testable with no DB and no clock. + * + * Two-stage, advisor-first (safety): + * 1. First qualification → NUDGE the advisor (creates a durable marker). + * 2. Only after the marker exists AND a second grace window has elapsed does + * the sweep SEND. A nudge and a send NEVER happen in the same pass. + */ + +/** Booking → nudge eligibility grace (hours since trips.lastStatusChangeAt). */ +const DEFAULT_PROMPT_GRACE_HOURS = 72 + +/** Nudge → autopilot-send grace (hours the advisor has after being nudged). */ +const DEFAULT_NUDGE_GRACE_HOURS = 24 + +const HOUR_MS = 60 * 60 * 1000 + +/** + * A pending-insurance candidate: one still-pending traveler on an active trip + * that is already past the booking grace window (the DB query applies the + * status + grace filter; these rows are the survivors). + */ +export interface AutopilotCandidateRow { + tripId: string + tripTravelerId: string + /** Trip owner = the advisor to nudge. Null when the trip has no owner. */ + advisorId: string | null + agencyId: string +} + +/** One trip ready for the autopilot pipeline, with its qualifying travelers. */ +export interface AutopilotTripGroup { + tripId: string + advisorId: string + agencyId: string + /** Pending travelers with NO existing insurance proposal job (send targets). */ + travelerIds: string[] +} + +/** + * Resolve the booking-grace window (hours). Env override + * `INSURANCE_PROMPT_GRACE_HOURS`, else {@link DEFAULT_PROMPT_GRACE_HOURS}. A + * non-positive / unparseable value falls back to the default so a bad env can + * never make the window 0 (which would fire the instant a trip is booked). + */ +export function getInsurancePromptGraceHours( + env: NodeJS.ProcessEnv = process.env, +): number { + const parsed = Number(env.INSURANCE_PROMPT_GRACE_HOURS) + return Number.isFinite(parsed) && parsed > 0 ? parsed : DEFAULT_PROMPT_GRACE_HOURS +} + +/** + * Resolve the nudge-grace window (hours) — how long the advisor has to act after + * being nudged before the autopilot sends. Env override + * `INSURANCE_PROMPT_NUDGE_GRACE_HOURS`, else {@link DEFAULT_NUDGE_GRACE_HOURS}. + */ +export function getInsuranceNudgeGraceHours( + env: NodeJS.ProcessEnv = process.env, +): number { + const parsed = Number(env.INSURANCE_PROMPT_NUDGE_GRACE_HOURS) + return Number.isFinite(parsed) && parsed > 0 ? parsed : DEFAULT_NUDGE_GRACE_HOURS +} + +/** + * Stable per-(advisor, trip) dedupe key for the advisor nudge. Unlike the P2 + * gate nudge (day-bucketed so it re-fires daily), this key is STABLE — the + * notification is created exactly once and its row is the durable "nudged at T" + * marker that stage 2 reads. The advisor id MUST be embedded because the + * platform dedupe index is global (not per-user). + */ +export function buildAutopilotNudgeDedupeKey(advisorId: string, tripId: string): string { + return `insurance-autopilot:${advisorId}:${tripId}` +} + +/** + * Group qualifying candidates by trip, excluding: + * - travelers whose insurance proposal job already exists (idempotency fence — + * `existingJobIds` holds the canonical getInsuranceProposalJobId ids found in + * automation_job_history), and + * - trips with no advisor/owner to nudge. + * + * `jobIdFor(tripId, travelerId)` produces the canonical job id so the exclusion + * compares like-for-like with the history rows. + * + * Deterministic ordering (trips + travelers sorted) so the sweep is stable. + */ +export function groupAutopilotCandidates( + rows: AutopilotCandidateRow[], + existingJobIds: ReadonlySet, + jobIdFor: (tripId: string, travelerId: string) => string, +): AutopilotTripGroup[] { + const byTrip = new Map() + + for (const row of rows) { + if (!row.advisorId) continue + if (existingJobIds.has(jobIdFor(row.tripId, row.tripTravelerId))) continue + + const existing = byTrip.get(row.tripId) + if (!existing) { + byTrip.set(row.tripId, { + tripId: row.tripId, + advisorId: row.advisorId, + agencyId: row.agencyId, + travelerIds: [row.tripTravelerId], + }) + } else if (!existing.travelerIds.includes(row.tripTravelerId)) { + existing.travelerIds.push(row.tripTravelerId) + } + } + + const groups = [...byTrip.values()] + for (const g of groups) g.travelerIds.sort((a, b) => a.localeCompare(b)) + return groups.sort((a, b) => a.tripId.localeCompare(b.tripId)) +} + +/** + * Stage-2 gate: given the marker's creation instant, is the nudge grace window + * elapsed (so the autopilot may send now)? A missing marker (`null`) is never + * ready — the advisor must be nudged first. + */ +export function isNudgeGraceElapsed( + nudgeCreatedAt: Date | null | undefined, + now: Date, + nudgeGraceHours: number, +): boolean { + if (!nudgeCreatedAt) return false + const elapsed = now.getTime() - nudgeCreatedAt.getTime() + return elapsed >= nudgeGraceHours * HOUR_MS +} + +/** + * Advisor-facing copy for the autopilot nudge. Neutral, action-oriented: the + * advisor can resolve insurance manually, or let the autopilot send the client + * proposal once the grace window elapses. + */ +export function buildAutopilotNudgeCopy(group: AutopilotTripGroup): { + title: string + body: string +} { + const n = group.travelerIds.length + const plural = n === 1 ? '' : 's' + return { + title: 'Insurance still needs attention', + body: + `${n} traveler${plural} on this trip still ${n === 1 ? 'has' : 'have'} no insurance ` + + `decision on file. Resolve it in the Insurance tab, or the autopilot will email ` + + `the client${n === 1 ? '' : 's'} an insurance proposal shortly.`, + } +} diff --git a/apps/api/src/automation/processors/client-care.processor.insurance-autopilot.spec.ts b/apps/api/src/automation/processors/client-care.processor.insurance-autopilot.spec.ts new file mode 100644 index 00000000..e6fef20a --- /dev/null +++ b/apps/api/src/automation/processors/client-care.processor.insurance-autopilot.spec.ts @@ -0,0 +1,303 @@ +/** + * refs #1466 P3c — ClientCareProcessor.handleInsurancePromptAutopilot. + * + * Focus: the client-send blast radius. Proves — + * - advisor-first: first qualification NUDGES the advisor and does NOT send; + * - stage 2: only after the nudge marker exists AND the nudge grace elapses + * does it call queueProposalEmails VERBATIM ({ userId:'system' }); + * - within nudge grace → neither nudge nor send; + * - idempotent: a traveler whose proposal job already exists is dropped, so a + * 2nd sweep is a no-op; + * - MIGRATION_SILENT_MODE hard-stops the entire sweep (no nudge, no send); + * - the inherited send path tags audience:'client' + automated:true (the + * #1048 cutover guard the autopilot relies on). + */ + +// Lazy-resolved service — mock the module so the real dependency graph is never +// loaded and moduleRef.get() returns our stub regardless of the token. +jest.mock('../../trips/insurance-automation.service', () => ({ + InsuranceAutomationService: class InsuranceAutomationService {}, +})) + +import { ClientCareProcessor } from './client-care.processor' +import { getInsuranceProposalJobId } from '../automation.types' + +const HOUR = 60 * 60 * 1000 + +type Candidate = { tripId: string; tripTravelerId: string; advisorId: string | null; agencyId: string } + +/** + * Chainable db mock. Routes by the sentinel table passed to .from(): + * tripTravelerInsurance → candidate rows + * automationJobHistory → existing-job rows + * platformNotifications → the nudge marker (or none) + * `.where()` is awaitable AND exposes `.limit()` (marker read). + */ +function makeDb(opts: { + candidates: Candidate[] + existingJobIds?: string[] + marker?: { createdAt: Date } | null + captureWhere?: (arg: unknown) => void +}) { + const schema = { + tripTravelerInsurance: { __t: 'tti', tripId: 'tti.trip', tripTravelerId: 'tti.traveler', status: 'tti.status' }, + trips: { + __t: 'trips', + id: 'trips.id', + ownerId: 'trips.owner', + agencyId: 'trips.agency', + status: 'trips.status', + lastStatusChangeAt: 'trips.lsc', + deletedAt: 'trips.deleted', + }, + automationJobHistory: { __t: 'history', jobId: 'h.jobId', queueName: 'h.queue' }, + platformNotifications: { __t: 'notif', dedupeKey: 'n.key', createdAt: 'n.created' }, + } as never + + let currentTable = '' + let whereCalls = 0 + const rowsFor = (): unknown[] => { + if (currentTable === 'tti') return opts.candidates + if (currentTable === 'history') return (opts.existingJobIds ?? []).map((jobId) => ({ jobId })) + if (currentTable === 'notif') return opts.marker ? [opts.marker] : [] + return [] + } + + const client = { + select: () => client, + from: (t: { __t: string }) => { + currentTable = t.__t + return client + }, + innerJoin: () => client, + where: (arg: unknown) => { + whereCalls += 1 + if (whereCalls === 1) opts.captureWhere?.(arg) + const rows = rowsFor() + const p = Promise.resolve(rows) as Promise & { limit: () => Promise } + p.limit = async () => rows + return p + }, + } + return { client, schema } +} + +function makeProcessor(overrides: { + db?: unknown + notificationService?: unknown + moduleRef?: unknown + emailService?: unknown + contentService?: unknown + documentTemplatesService?: unknown +}): ClientCareProcessor { + return new ClientCareProcessor( + (overrides.db ?? {}) as never, + {} as never, // eventEmitter + {} as never, // automationService + (overrides.contentService ?? {}) as never, + (overrides.notificationService ?? { send: jest.fn() }) as never, + (overrides.emailService ?? {}) as never, + (overrides.documentTemplatesService ?? {}) as never, + {} as never, // contactLifecycleService + (overrides.moduleRef ?? undefined) as never, + ) +} + +const call = (p: ClientCareProcessor) => + (p as unknown as { handleInsurancePromptAutopilot: () => Promise }).handleInsurancePromptAutopilot() + +function makeInsuranceAutomation() { + return { + queueProposalEmails: jest.fn( + (_tripId: string, _params: { travelerIds: string[]; agencyId: string; userId: string }) => + Promise.resolve({ queuedCount: 1, jobIds: ['insurance:trip-1:t1'] }), + ), + } +} + +describe('ClientCareProcessor.handleInsurancePromptAutopilot', () => { + const origSilent = process.env.MIGRATION_SILENT_MODE + afterEach(() => { + if (origSilent === undefined) delete process.env.MIGRATION_SILENT_MODE + else process.env.MIGRATION_SILENT_MODE = origSilent + jest.restoreAllMocks() + }) + + it('STAGE 1 — nudges the advisor and does NOT send on first qualification (no marker)', async () => { + const notificationService = { + send: jest.fn((_params: Record) => + Promise.resolve({ channels: ['platform'], results: {} }), + ), + } + const insurance = makeInsuranceAutomation() + const processor = makeProcessor({ + db: makeDb({ candidates: [{ tripId: 'trip-1', tripTravelerId: 't1', advisorId: 'advisor-1', agencyId: 'ag-1' }], marker: null }), + notificationService, + moduleRef: { get: () => insurance }, + }) + + await call(processor) + + expect(notificationService.send).toHaveBeenCalledTimes(1) + const arg = notificationService.send.mock.calls[0]![0] as Record + expect(arg.userId).toBe('advisor-1') + expect(arg.category).toBe('automation_gate') + expect(arg.actionUrl).toBe('/trips/trip-1?tab=insurance') + expect(arg.dedupeKey).toBe('insurance-autopilot:advisor-1:trip-1') + // Advisor nudge STRICTLY precedes any send. + expect(insurance.queueProposalEmails).not.toHaveBeenCalled() + }) + + it('STAGE 2 — sends verbatim once the marker exists AND the nudge grace elapsed', async () => { + const notificationService = { send: jest.fn() } + const insurance = makeInsuranceAutomation() + const processor = makeProcessor({ + db: makeDb({ + candidates: [ + { tripId: 'trip-1', tripTravelerId: 't1', advisorId: 'advisor-1', agencyId: 'ag-1' }, + { tripId: 'trip-1', tripTravelerId: 't2', advisorId: 'advisor-1', agencyId: 'ag-1' }, + ], + marker: { createdAt: new Date(Date.now() - 25 * HOUR) }, // > 24h default grace + }), + notificationService, + moduleRef: { get: () => insurance }, + }) + + await call(processor) + + // No re-nudge on the send pass. + expect(notificationService.send).not.toHaveBeenCalled() + expect(insurance.queueProposalEmails).toHaveBeenCalledTimes(1) + const [tripId, params] = insurance.queueProposalEmails.mock.calls[0]! + expect(tripId).toBe('trip-1') + expect(params).toEqual({ travelerIds: ['t1', 't2'], agencyId: 'ag-1', userId: 'system' }) + }) + + it('STAGE 2 — defers (no nudge, no send) while still inside the nudge grace window', async () => { + const notificationService = { send: jest.fn() } + const insurance = makeInsuranceAutomation() + const processor = makeProcessor({ + db: makeDb({ + candidates: [{ tripId: 'trip-1', tripTravelerId: 't1', advisorId: 'advisor-1', agencyId: 'ag-1' }], + marker: { createdAt: new Date(Date.now() - 1 * HOUR) }, // nudged 1h ago (< 24h) + }), + notificationService, + moduleRef: { get: () => insurance }, + }) + + await call(processor) + + expect(notificationService.send).not.toHaveBeenCalled() + expect(insurance.queueProposalEmails).not.toHaveBeenCalled() + }) + + it('is idempotent — a traveler whose proposal job already exists is dropped (2nd sweep no-op)', async () => { + const notificationService = { send: jest.fn() } + const insurance = makeInsuranceAutomation() + const processor = makeProcessor({ + db: makeDb({ + candidates: [{ tripId: 'trip-1', tripTravelerId: 't1', advisorId: 'advisor-1', agencyId: 'ag-1' }], + existingJobIds: [getInsuranceProposalJobId('trip-1', 't1')], + marker: { createdAt: new Date(Date.now() - 25 * HOUR) }, + }), + notificationService, + moduleRef: { get: () => insurance }, + }) + + await call(processor) + + // Traveler already has a job → excluded → nothing to nudge or send. + expect(notificationService.send).not.toHaveBeenCalled() + expect(insurance.queueProposalEmails).not.toHaveBeenCalled() + }) + + it('MIGRATION_SILENT_MODE hard-stops the sweep — no nudge, no send, no db read', async () => { + process.env.MIGRATION_SILENT_MODE = 'true' + const notificationService = { send: jest.fn() } + const insurance = makeInsuranceAutomation() + const select = jest.fn() + const processor = makeProcessor({ + db: { client: { select }, schema: {} }, + notificationService, + moduleRef: { get: () => insurance }, + }) + + await call(processor) + + expect(select).not.toHaveBeenCalled() + expect(notificationService.send).not.toHaveBeenCalled() + expect(insurance.queueProposalEmails).not.toHaveBeenCalled() + }) + + it('candidate query filters to pending insurance on active, non-deleted trips', async () => { + let captured: unknown + const processor = makeProcessor({ + db: makeDb({ + candidates: [], + captureWhere: (arg) => { + captured = arg + }, + }), + notificationService: { send: jest.fn() }, + moduleRef: { get: () => makeInsuranceAutomation() }, + }) + + await call(processor) + + // The pending-only + active + soft-delete filter is a raw SQL guarantee; + // assert the predicate encodes both status literals. + const serialized = JSON.stringify(captured) + expect(serialized).toContain('pending') + expect(serialized).toContain('active') + }) +}) + +describe('ClientCareProcessor.handleInsuranceProposalEmail — inherited client-send guard', () => { + it('tags the send audience:"client" + automated:true (autopilot inherits the #1048 cutover guard)', async () => { + const emailService = { + sendEmail: jest.fn((_arg: Record) => Promise.resolve(undefined)), + } + const contentService = { + mergeCustomVariables: (vars: unknown) => vars, + applyEmailOverride: (content: unknown) => content, + } + const documentTemplatesService = { renderEmail: jest.fn(async () => null) } // fall back to hardcoded HTML + const db = { + client: { + select: () => ({ + from: () => ({ + where: () => ({ limit: async () => [{ name: 'Acme Travel' }] }), + }), + }), + }, + schema: { agencies: { id: 'a.id', name: 'a.name' } }, + } + const processor = makeProcessor({ db, emailService, contentService, documentTemplatesService }) + + await ( + processor as unknown as { + handleInsuranceProposalEmail: (d: unknown, o: unknown) => Promise + } + ).handleInsuranceProposalEmail( + { + type: 'insurance.proposal.email', + tripId: 'trip-1', + recipientTravelerId: 't1', + recipientName: 'Jane Doe', + recipientEmail: 'jane@example.com', + dependentTravelerIds: [], + formToken: 'tok-123', + agencyId: 'ag-1', + tripName: 'Paris 2026', + }, + null, + ) + + expect(emailService.sendEmail).toHaveBeenCalledTimes(1) + const arg = emailService.sendEmail.mock.calls[0]![0] as Record + expect(arg.audience).toBe('client') + expect(arg.automated).toBe(true) + expect(arg.templateSlug).toBe('insurance-proposal') + expect(arg.to).toEqual(['jane@example.com']) + }) +}) diff --git a/apps/api/src/automation/processors/client-care.processor.ts b/apps/api/src/automation/processors/client-care.processor.ts index 247af64c..f8f6ace1 100644 --- a/apps/api/src/automation/processors/client-care.processor.ts +++ b/apps/api/src/automation/processors/client-care.processor.ts @@ -11,7 +11,8 @@ */ import { Processor, WorkerHost, OnWorkerEvent } from '@nestjs/bullmq' -import { Logger, Injectable } from '@nestjs/common' +import { Logger, Injectable, Optional } from '@nestjs/common' +import { ModuleRef } from '@nestjs/core' import { Job } from 'bullmq' import { EventEmitter2 } from '@nestjs/event-emitter' import { eq, and, or, lte, gte, gt, sql, inArray, isNotNull, ne } from 'drizzle-orm' @@ -28,6 +29,7 @@ import { getClientFollowUpTemplate } from '../../email/templates/client-follow-u import { getClientPostTripTemplate } from '../../email/templates/client-post-trip.template' import { QUEUES, + getInsuranceProposalJobId, type ClientCareJobData, type PaymentReminderJobData, type DepartureReminderJobData, @@ -41,6 +43,18 @@ import { buildNudgeDedupeKey, buildNudgeCopy, } from '../automation-gate-nudge' +import { + getInsurancePromptGraceHours, + getInsuranceNudgeGraceHours, + buildAutopilotNudgeDedupeKey, + buildAutopilotNudgeCopy, + groupAutopilotCandidates, + isNudgeGraceElapsed, + type AutopilotCandidateRow, +} from '../insurance-autopilot' +// Type-only import — the concrete service is resolved lazily via ModuleRef at +// call time to avoid a TripsModule ↔ AutomationModule constructor DI cycle. +import type { InsuranceAutomationService } from '../../trips/insurance-automation.service' /** Escape HTML special characters to prevent markup breakage in email templates */ function escapeHtml(str: string): string { @@ -71,6 +85,10 @@ export class ClientCareProcessor extends WorkerHost { private readonly emailService: EmailService, private readonly documentTemplatesService: DocumentTemplatesService, private readonly contactLifecycleService: ContactLifecycleService, + // refs #1466 P3c — resolves InsuranceAutomationService lazily (avoids the + // TripsModule ↔ AutomationModule DI cycle). Optional so existing unit + // constructions that don't exercise the autopilot path stay valid. + @Optional() private readonly moduleRef?: ModuleRef, ) { super() } @@ -174,6 +192,11 @@ export class ClientCareProcessor extends WorkerHost { break } + case 'recurring.insurance_prompt_autopilot': { + await this.handleInsurancePromptAutopilot() + break + } + case 'recurring.automation_gate_nudge': { await this.handleAutomationGateNudge() break @@ -1146,6 +1169,195 @@ ${removedTasks.map((t) => ` { + // Safety 2 — during the TES→prod cutover the entire pipeline is inert. Bail + // BEFORE any read/nudge/token so silent mode can't leak orphan form tokens + // (queueProposalEmails mints a token before schedule() would no-op). + if (process.env.MIGRATION_SILENT_MODE === 'true') { + this.logger.warn('[insurance-autopilot] MIGRATION_SILENT_MODE — sweep suppressed') + return + } + + const { tripTravelerInsurance, trips, automationJobHistory, platformNotifications } = + this.db.schema + const now = new Date() + const promptGraceMs = getInsurancePromptGraceHours() * 60 * 60 * 1000 + const nudgeGraceHours = getInsuranceNudgeGraceHours() + const bookingCutoff = new Date(now.getTime() - promptGraceMs) + + // 1. Candidates: still-pending travelers on BOOKED (active) trips whose + // booking anchor (lastStatusChangeAt, set at planning→active) is past the + // grace window. status='active' is the only status where + // lastStatusChangeAt reliably equals the booking instant, so restricting + // to it keeps the grace math honest (travelling/travelled would anchor on + // a later transition). Soft-deleted trips are excluded. + const rows: AutopilotCandidateRow[] = await this.db.client + .select({ + tripId: tripTravelerInsurance.tripId, + tripTravelerId: tripTravelerInsurance.tripTravelerId, + advisorId: trips.ownerId, + agencyId: trips.agencyId, + }) + .from(tripTravelerInsurance) + .innerJoin(trips, eq(trips.id, tripTravelerInsurance.tripId)) + .where( + and( + eq(tripTravelerInsurance.status, 'pending'), + eq(trips.status, 'active'), + isNotNull(trips.lastStatusChangeAt), + lte(trips.lastStatusChangeAt, bookingCutoff), + sql`${trips.deletedAt} IS NULL`, + ), + ) + + if (rows.length === 0) { + this.logger.debug('[insurance-autopilot] no pending travelers past the booking grace window') + return + } + + // 2. Idempotency fence — drop any traveler whose insurance proposal job + // already exists in automation_job_history (queued OR terminal). The + // canonical jobId is the identity; a matched row means this traveler has + // already been queued. A leftover pending MINOR whose guardian was sent is + // harmless: queueProposalEmails only builds recipients from adults WITH an + // email, so a minor passed alone yields zero recipients (no double-send). + const candidateJobIds = rows.map((r) => getInsuranceProposalJobId(r.tripId, r.tripTravelerId)) + const existingRows = await this.db.client + .select({ jobId: automationJobHistory.jobId }) + .from(automationJobHistory) + .where( + and( + eq(automationJobHistory.queueName, QUEUES.CLIENT_CARE), + inArray(automationJobHistory.jobId, candidateJobIds), + ), + ) + const existingJobIds = new Set(existingRows.map((r) => r.jobId)) + + const groups = groupAutopilotCandidates(rows, existingJobIds, getInsuranceProposalJobId) + if (groups.length === 0) { + this.logger.debug('[insurance-autopilot] all candidates already have a proposal job or no advisor') + return + } + + this.logger.log(`[insurance-autopilot] ${groups.length} trip(s) with unresolved insurance past grace`) + + for (const group of groups) { + try { + const dedupeKey = buildAutopilotNudgeDedupeKey(group.advisorId, group.tripId) + + // Read the durable advisor-nudge marker (stable dedupe key → at most one + // row). Its createdAt is the "nudged at T" instant for the stage gate. + const [marker] = await this.db.client + .select({ createdAt: platformNotifications.createdAt }) + .from(platformNotifications) + .where(eq(platformNotifications.dedupeKey, dedupeKey)) + .limit(1) + + if (!marker) { + // STAGE 1 — first qualification: nudge the advisor. This creates the + // marker (atomic dedupe claim). NO client send this pass. + const { title, body } = buildAutopilotNudgeCopy(group) + await this.notificationService.send({ + userId: group.advisorId, + category: 'automation_gate', + title, + body, + actionUrl: `/trips/${group.tripId}?tab=insurance`, + dedupeKey, + data: { + tripId: group.tripId, + travelerCount: group.travelerIds.length, + notificationType: 'insurance_autopilot.nudge', + }, + }) + this.logger.log(`[insurance-autopilot] nudged advisor ${group.advisorId} for trip ${group.tripId}`) + continue + } + + // STAGE 2 — the advisor was nudged; only send once the nudge grace has + // elapsed. Otherwise wait for a later sweep. + if (!isNudgeGraceElapsed(marker.createdAt as Date, now, nudgeGraceHours)) { + this.logger.debug( + `[insurance-autopilot] trip ${group.tripId} within nudge grace — deferring send`, + ) + continue + } + + const insuranceAutomation = this.resolveInsuranceAutomationService() + if (!insuranceAutomation) { + this.logger.warn('[insurance-autopilot] InsuranceAutomationService unavailable — skipping send') + continue + } + + // SEND — verbatim into the existing proposal pipeline. Inherits the + // #1048 cutover + audience:'client' guards; P3b evidence populates on the + // resulting client_esign waiver. + const { queuedCount } = await insuranceAutomation.queueProposalEmails(group.tripId, { + travelerIds: group.travelerIds, + agencyId: group.agencyId, + userId: 'system', + }) + this.logger.log( + `[insurance-autopilot] auto-sent ${queuedCount} insurance proposal(s) for trip ${group.tripId}`, + ) + } catch (error) { + // Best-effort — one trip's failure must not stop the sweep. + this.logger.error(`[insurance-autopilot] failed for trip ${group.tripId}: ${String(error)}`) + } + } + } + + /** + * Lazily resolve InsuranceAutomationService from the app container. Kept out of + * the constructor to avoid a TripsModule ↔ AutomationModule DI cycle (the same + * pattern trips.service.ts uses for ShoreEx). Returns null if ModuleRef isn't + * wired (unit constructions) or the provider can't be found. + */ + private resolveInsuranceAutomationService(): InsuranceAutomationService | null { + if (!this.moduleRef) return null + try { + // eslint-disable-next-line @typescript-eslint/no-require-imports -- lazy runtime resolution to break the TripsModule ↔ AutomationModule DI cycle + const { InsuranceAutomationService } = require('../../trips/insurance-automation.service') + return this.moduleRef.get(InsuranceAutomationService, { strict: false }) + } catch (error) { + this.logger.warn(`[insurance-autopilot] could not resolve InsuranceAutomationService: ${String(error)}`) + return null + } + } + // ============================================================================ // Helpers // ============================================================================