From 86acc06b16cd4258eab4f9d6c805c18d3a986aef Mon Sep 17 00:00:00 2001 From: lorentz Date: Sun, 10 May 2026 07:28:21 -0400 Subject: [PATCH] feat(09-03): rewrite executeNotify with route_to_user branch and fallback semantics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Extract dispatchToGlobalChannel helper (backward-compat path unchanged) - Add dispatchUserRoute: field lookup, resolver dispatch, email→user_id resolution, mute check, personal channel lookup, send with fallback - Add fallbackToGlobal: annotates output with user_route_fallback reason - Mute path (enabled=false) returns success:true/notified:false, no fallback - Default channel-type order when omitted: ntfy then teams (ROUTE-06) - Export _INTERNALS test seam following link-discovery.ts precedent - All five fallback reasons: no_channel, send_failed, user_not_found, no_field_value, resolver_unknown (ROUTE-03..06) --- lib/services/pipeline-steps/notify.ts | 266 ++++++++++++++++++++++++-- 1 file changed, 253 insertions(+), 13 deletions(-) diff --git a/lib/services/pipeline-steps/notify.ts b/lib/services/pipeline-steps/notify.ts index 2310039..6c572d2 100644 --- a/lib/services/pipeline-steps/notify.ts +++ b/lib/services/pipeline-steps/notify.ts @@ -2,26 +2,63 @@ * Notify Step — send notification to a configured channel. * Config: { channel_id: 1, message: "...", card_template?: {...} } * Channel config is loaded from notification_channels table. + * + * Optional route_to_user block (Phase 9 ROUTE-01..06): + * When present, the step attempts delivery via the resolved Pulse user's + * personal channel before falling back to the step's channel_id. + * When absent, behavior is byte-identical to the pre-Phase-9 implementation. */ import { registerStepExecutor } from '../pipeline-engine'; import { postgresClient } from '../postgres-client'; -import { PipelineStep, PipelineContext, StepExecutorResult, NotificationChannel } from '../../types/pipeline'; +import { + PipelineStep, + PipelineContext, + StepExecutorResult, + NotificationChannel, + RouteToUser, + UserRouteFallback, +} from '../../types/pipeline'; +import { resolveRecipient } from './notify-resolvers'; + +// ============================================================================ +// Main executor +// ============================================================================ async function executeNotify( step: PipelineStep, - _context: PipelineContext, - _executionId: number + context: PipelineContext, + _executionId: number, ): Promise { const channelId = Number(step.config.channel_id); - if (!channelId || isNaN(channelId)) { return { success: false, error: 'Missing or invalid channel_id' }; } + const route = (step.config.route_to_user as RouteToUser | undefined) ?? null; + const message = step.config.message || ''; + + // BACKWARD COMPATIBLE PATH: no route_to_user → original behavior verbatim. + if (!route) { + return await dispatchToGlobalChannel(channelId, step.config, message); + } + + // USER-ROUTE PATH (ROUTE-01..06) + return await dispatchUserRoute({ step, context, message, channelId, route }); +} + +// ============================================================================ +// Global channel dispatch (original path — extracted for reuse as fallback) +// ============================================================================ + +async function dispatchToGlobalChannel( + channelId: number, + config: Record, + message: string, +): Promise { const result = await postgresClient.query( `SELECT * FROM notification_channels WHERE id = $1 AND is_active = true`, - [channelId] + [channelId], ); if (result.rows.length === 0) { @@ -29,26 +66,229 @@ async function executeNotify( } const channel = result.rows[0]; - const message = step.config.message || ''; switch (channel.channel_type) { case 'teams': - return await sendTeams(channel, step.config, message); + return await sendTeams(channel, config, message); case 'telegram': return await sendTelegram(channel, message); case 'ntfy': - return await sendNtfy(channel, step.config, message); + return await sendNtfy(channel, config, message); case 'webhook': - return await sendWebhook(channel, step.config, message); + return await sendWebhook(channel, config, message); default: return { success: false, error: `Unknown channel type: ${channel.channel_type}` }; } } +// ============================================================================ +// Per-user route dispatch (ROUTE-03..06, D-08..D-12) +// ============================================================================ + +async function dispatchUserRoute(args: { + step: PipelineStep; + context: PipelineContext; + message: string; + channelId: number; + route: RouteToUser; +}): Promise { + const { step, context, message, channelId, route } = args; + + // Step 5a — Read field value from context (single-level lookup, v1 limitation). + // v1: single-level only. Dotted paths like 'company.id' are NOT walked. + // Admins must surface nested values at top level via an upstream transform. + const sourceObj = context[route.source]; + const fieldValue = sourceObj?.[route.field]; + if (fieldValue === undefined || fieldValue === null || fieldValue === '') { + return await fallbackToGlobal({ + channelId, + config: step.config, + message, + fallback: { reason: 'no_field_value', channel_type: route.channel_type ?? 'auto' }, + }); + } + + // Step 5b — Resolve recipient via the named resolver. + const { recipient, resolverFound } = await resolveRecipient(route.resolve, fieldValue); + if (!resolverFound) { + return await fallbackToGlobal({ + channelId, + config: step.config, + message, + fallback: { + reason: 'resolver_unknown', + channel_type: route.channel_type ?? 'auto', + error: `Unknown resolver: ${route.resolve}`, + }, + }); + } + if (!recipient) { + return await fallbackToGlobal({ + channelId, + config: step.config, + message, + fallback: { reason: 'user_not_found', channel_type: route.channel_type ?? 'auto' }, + }); + } + + // Step 5c — Resolve recipient to a Pulse user.id. + let userId: string; + if ('user_id' in recipient) { + userId = recipient.user_id; + } else { + // Has email — look up Pulse user by email (case-insensitive; Autotask emails vary). + const userRes = await postgresClient.query<{ id: string }>( + `SELECT id FROM "user" WHERE LOWER(email) = LOWER($1)`, + [recipient.email], + ); + if (userRes.rows.length === 0) { + return await fallbackToGlobal({ + channelId, + config: step.config, + message, + fallback: { reason: 'user_not_found', channel_type: route.channel_type ?? 'auto' }, + }); + } + userId = userRes.rows[0].id; + } + + // Step 5d — Determine channel-type attempt order (ROUTE-06 / D-10). + // When channel_type is omitted: ntfy first (instant push), then teams. + const attemptOrder: Array<'ntfy' | 'teams'> = + route.channel_type ? [route.channel_type] : ['ntfy', 'teams']; + + // Step 5e — For each attempted type: check mute, look up personal channel, send. + for (const channelType of attemptOrder) { + // ROUTE-05 mute check — query user_event_subscriptions. + // Default-enabled (D-15): row absence = enabled = true. + const subRes = await postgresClient.query<{ enabled: boolean }>( + `SELECT enabled FROM user_event_subscriptions + WHERE user_id = $1 AND event_key = $2 AND channel_type = $3`, + [userId, route.event_key, channelType], + ); + const enabled = subRes.rows.length === 0 ? true : subRes.rows[0].enabled; + if (!enabled) { + // ROUTE-05 / D-12: skip silently — muting must actually mute. + // DO NOT call fallbackToGlobal here. Return without any fallback. + return { + success: true, + output: { + notified: false, + skipped_reason: 'user_muted', + user_id: userId, + event_key: route.event_key, + channel_type: channelType, + }, + }; + } + + // Look up the user's personal channel of this type. + const chanRes = await postgresClient.query( + `SELECT * FROM notification_channels + WHERE owner_user_id = $1 AND channel_type = $2 AND is_active = true + LIMIT 1`, + [userId, channelType], + ); + if (chanRes.rows.length === 0) { + // No personal channel of this type. + // If more types remain in attemptOrder, continue to the next. + // If this was the last type, fall back to global. + if (channelType === attemptOrder[attemptOrder.length - 1]) { + return await fallbackToGlobal({ + channelId, + config: step.config, + message, + fallback: { reason: 'no_channel', user_id: userId, channel_type: channelType }, + }); + } + continue; + } + + // Dispatch to the personal channel using existing send helpers. + const dispatchResult = + channelType === 'teams' + ? await sendTeams(chanRes.rows[0], step.config, message) + : await sendNtfy(chanRes.rows[0], step.config, message); + + if (dispatchResult.success) { + return { + success: true, + output: { + ...(dispatchResult.output ?? {}), + user_route: { user_id: userId, channel_type: channelType, event_key: route.event_key }, + }, + }; + } + + // Send failed — try next type if available, else fall back to global. + if (channelType === attemptOrder[attemptOrder.length - 1]) { + return await fallbackToGlobal({ + channelId, + config: step.config, + message, + fallback: { + reason: 'send_failed', + user_id: userId, + channel_type: channelType, + error: dispatchResult.error, + }, + }); + } + } + + // Defensive: the loop always returns or continues, so this should never run. + return await fallbackToGlobal({ + channelId, + config: step.config, + message, + fallback: { reason: 'no_channel', user_id: userId!, channel_type: 'auto' }, + }); +} + +// ============================================================================ +// Fallback to global channel (D-11) +// ============================================================================ + +async function fallbackToGlobal(args: { + channelId: number; + config: Record; + message: string; + fallback: UserRouteFallback; +}): Promise { + const result = await dispatchToGlobalChannel(args.channelId, args.config, args.message); + // Annotate output with the fallback reason regardless of global success/failure. + if (result.success) { + return { + success: true, + output: { + ...(result.output ?? {}), + user_route_fallback: args.fallback, + }, + }; + } + // Global also failed — return error with fallback context attached. + return { + success: false, + error: result.error, + output: { user_route_fallback: args.fallback }, + }; +} + +// ============================================================================ +// Test seam — exposes internals for vitest mocking without changing the API. +// Follows the _INTERNALS pattern from lib/services/analyzer/link-discovery.ts. +// ============================================================================ + +export const _INTERNALS = { dispatchToGlobalChannel, dispatchUserRoute, fallbackToGlobal }; + +// ============================================================================ +// Send helpers (preserved verbatim from pre-Phase-9) +// ============================================================================ + async function sendTeams( channel: NotificationChannel, config: Record, - message: string + message: string, ): Promise { const webhookUrl = channel.config.webhook_url; if (!webhookUrl) { @@ -88,7 +328,7 @@ async function sendTeams( async function sendTelegram( channel: NotificationChannel, - message: string + message: string, ): Promise { const botToken = channel.config.bot_token; const chatId = channel.config.chat_id; @@ -118,7 +358,7 @@ async function sendTelegram( async function sendNtfy( channel: NotificationChannel, config: Record, - message: string + message: string, ): Promise { const serverUrl = channel.config.server_url || 'https://ntfy.sh'; const topic = channel.config.topic; @@ -158,7 +398,7 @@ async function sendNtfy( async function sendWebhook( channel: NotificationChannel, config: Record, - message: string + message: string, ): Promise { const url = channel.config.url; if (!url) {