lorentz
2ff2dc9904
- NTFY_TOPIC_RE tightened to ^pulse-me-[A-Za-z0-9-]{6,64}$ (rejects noc-*, soc-*, bare pulse-)
- mintNtfyTopic() now returns pulse-me-XXXXXXXX (8 hex chars, same entropy)
- sendChannelTest (ntfy): forced to NTFY_BASE_URL + NTFY_PULSE_TOKEN; drops channel.config.auth_token path
- sendNtfy (notify.ts): personal/global branch on owner_user_id; personal -> company server + bearer NTFY_PULSE_TOKEN
- approval.ts ntfy branch: same personal/global split (soft fallback when token missing)
- ticket-digest-service.ts: deliver() + getAvailableChannels() SELECTs now include owner_user_id; ntfy branch applies same split
446 lines
15 KiB
TypeScript
446 lines
15 KiB
TypeScript
/**
|
|
* 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,
|
|
RouteToUser,
|
|
UserRouteFallback,
|
|
} from '../../types/pipeline';
|
|
import { resolveRecipient } from './notify-resolvers';
|
|
|
|
// ============================================================================
|
|
// Main executor
|
|
// ============================================================================
|
|
|
|
async function executeNotify(
|
|
step: PipelineStep,
|
|
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],
|
|
);
|
|
|
|
if (result.rows.length === 0) {
|
|
return { success: false, error: `Notification channel #${channelId} not found or inactive` };
|
|
}
|
|
|
|
const channel = result.rows[0];
|
|
|
|
switch (channel.channel_type) {
|
|
case 'teams':
|
|
return await sendTeams(channel, config, message);
|
|
case 'telegram':
|
|
return await sendTelegram(channel, message);
|
|
case 'ntfy':
|
|
return await sendNtfy(channel, config, message);
|
|
case 'webhook':
|
|
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,
|
|
): Promise<StepExecutorResult> {
|
|
const webhookUrl = channel.config.webhook_url;
|
|
if (!webhookUrl) {
|
|
return { success: false, error: 'Teams channel missing webhook_url' };
|
|
}
|
|
|
|
// If a card_template is provided, use it as Adaptive Card
|
|
const body = config.card_template || {
|
|
type: 'message',
|
|
attachments: [{
|
|
contentType: 'application/vnd.microsoft.card.adaptive',
|
|
content: {
|
|
type: 'AdaptiveCard',
|
|
$schema: 'http://adaptivecards.io/schemas/adaptive-card.json',
|
|
version: '1.4',
|
|
body: [
|
|
{ type: 'TextBlock', text: config.title || 'Pulse Notification', weight: 'bolder', size: 'medium' },
|
|
{ type: 'TextBlock', text: message, wrap: true },
|
|
],
|
|
},
|
|
}],
|
|
};
|
|
|
|
const resp = await fetch(webhookUrl, {
|
|
method: 'POST',
|
|
headers: { 'Content-Type': 'application/json' },
|
|
body: JSON.stringify(body),
|
|
});
|
|
|
|
if (!resp.ok) {
|
|
const errText = await resp.text();
|
|
return { success: false, error: `Teams webhook failed (${resp.status}): ${errText.substring(0, 200)}` };
|
|
}
|
|
|
|
return { success: true, output: { notified: true, channel: 'teams' } };
|
|
}
|
|
|
|
async function sendTelegram(
|
|
channel: NotificationChannel,
|
|
message: string,
|
|
): Promise<StepExecutorResult> {
|
|
const botToken = channel.config.bot_token;
|
|
const chatId = channel.config.chat_id;
|
|
|
|
if (!botToken || !chatId) {
|
|
return { success: false, error: 'Telegram channel missing bot_token or chat_id' };
|
|
}
|
|
|
|
const resp = await fetch(`https://api.telegram.org/bot${botToken}/sendMessage`, {
|
|
method: 'POST',
|
|
headers: { 'Content-Type': 'application/json' },
|
|
body: JSON.stringify({
|
|
chat_id: chatId,
|
|
text: message,
|
|
parse_mode: channel.config.parse_mode || 'HTML',
|
|
}),
|
|
});
|
|
|
|
if (!resp.ok) {
|
|
const errText = await resp.text();
|
|
return { success: false, error: `Telegram API failed (${resp.status}): ${errText.substring(0, 200)}` };
|
|
}
|
|
|
|
return { success: true, output: { notified: true, channel: 'telegram' } };
|
|
}
|
|
|
|
async function sendNtfy(
|
|
channel: NotificationChannel,
|
|
config: Record<string, any>,
|
|
message: string,
|
|
): Promise<StepExecutorResult> {
|
|
const topic = channel.config.topic;
|
|
if (!topic) {
|
|
return { success: false, error: 'ntfy channel missing topic' };
|
|
}
|
|
|
|
// Personal channels (owner_user_id set) are forced to the company ntfy
|
|
// server with the company bearer token (UAT-FIX-01). Global / admin rows
|
|
// (owner_user_id NULL) retain their existing config-driven behavior so
|
|
// legacy ntfy.sh deployments and custom self-hosted instances keep working.
|
|
const isPersonal = !!channel.owner_user_id;
|
|
const serverUrl = isPersonal
|
|
? (process.env.NTFY_BASE_URL || 'https://ntfy.wulfconsulting.cloud')
|
|
: (channel.config.server_url || 'https://ntfy.sh');
|
|
|
|
const headers: Record<string, string> = {
|
|
'Content-Type': 'text/plain',
|
|
};
|
|
|
|
if (config.title || channel.config.default_title) {
|
|
headers['Title'] = config.title || channel.config.default_title;
|
|
}
|
|
if (config.priority || channel.config.default_priority) {
|
|
headers['Priority'] = config.priority || channel.config.default_priority;
|
|
}
|
|
|
|
if (isPersonal) {
|
|
const token = process.env.NTFY_PULSE_TOKEN;
|
|
if (!token) {
|
|
return { success: false, error: 'NTFY_PULSE_TOKEN not configured' };
|
|
}
|
|
headers['Authorization'] = `Bearer ${token}`;
|
|
} else if (channel.config.auth_token) {
|
|
headers['Authorization'] = `Bearer ${channel.config.auth_token}`;
|
|
}
|
|
|
|
const resp = await fetch(`${serverUrl}/${topic}`, {
|
|
method: 'POST',
|
|
headers,
|
|
body: message,
|
|
});
|
|
|
|
if (!resp.ok) {
|
|
const errText = await resp.text();
|
|
return { success: false, error: `ntfy failed (${resp.status}): ${errText.substring(0, 200)}` };
|
|
}
|
|
|
|
return { success: true, output: { notified: true, channel: 'ntfy' } };
|
|
}
|
|
|
|
async function sendWebhook(
|
|
channel: NotificationChannel,
|
|
config: Record<string, any>,
|
|
message: string,
|
|
): Promise<StepExecutorResult> {
|
|
const url = channel.config.url;
|
|
if (!url) {
|
|
return { success: false, error: 'Webhook channel missing url' };
|
|
}
|
|
|
|
const method = channel.config.method || 'POST';
|
|
const customHeaders = channel.config.headers || {};
|
|
|
|
const body = config.body_template
|
|
? config.body_template
|
|
: { message, timestamp: new Date().toISOString() };
|
|
|
|
const resp = await fetch(url, {
|
|
method,
|
|
headers: {
|
|
'Content-Type': 'application/json',
|
|
...customHeaders,
|
|
},
|
|
body: JSON.stringify(body),
|
|
});
|
|
|
|
if (!resp.ok) {
|
|
const errText = await resp.text();
|
|
return { success: false, error: `Webhook failed (${resp.status}): ${errText.substring(0, 200)}` };
|
|
}
|
|
|
|
return { success: true, output: { notified: true, channel: 'webhook' } };
|
|
}
|
|
|
|
registerStepExecutor('notify', executeNotify);
|