feat(09-03): rewrite executeNotify with route_to_user branch and fallback semantics
- 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)
This commit is contained in:
parent
d27462f713
commit
86acc06b16
1 changed files with 253 additions and 13 deletions
|
|
@ -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<StepExecutorResult> {
|
||||
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<string, any>,
|
||||
message: string,
|
||||
): Promise<StepExecutorResult> {
|
||||
const result = await postgresClient.query<NotificationChannel>(
|
||||
`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<StepExecutorResult> {
|
||||
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<NotificationChannel>(
|
||||
`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<string, any>;
|
||||
message: string;
|
||||
fallback: UserRouteFallback;
|
||||
}): Promise<StepExecutorResult> {
|
||||
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<string, any>,
|
||||
message: string
|
||||
message: string,
|
||||
): Promise<StepExecutorResult> {
|
||||
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<StepExecutorResult> {
|
||||
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<string, any>,
|
||||
message: string
|
||||
message: string,
|
||||
): Promise<StepExecutorResult> {
|
||||
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<string, any>,
|
||||
message: string
|
||||
message: string,
|
||||
): Promise<StepExecutorResult> {
|
||||
const url = channel.config.url;
|
||||
if (!url) {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue