diff --git a/packages/api/src/tasks/checkAlerts/__tests__/notifications.test.ts b/packages/api/src/tasks/checkAlerts/__tests__/notifications.test.ts new file mode 100644 index 0000000000..6988436010 --- /dev/null +++ b/packages/api/src/tasks/checkAlerts/__tests__/notifications.test.ts @@ -0,0 +1,31 @@ +import { InlineNotificationDispatcher } from '@/tasks/checkAlerts/notifications'; + +// The dispatcher forwards this opaque, so its shape doesn't matter for these +// tests — only that dispatch()/shutdown() behave per the documented contract. +const fakeJob: any = {}; + +describe('InlineNotificationDispatcher', () => { + it('resolves after the deliver fn resolves', async () => { + const order: string[] = []; + const dispatcher = new InlineNotificationDispatcher(async () => { + order.push('delivered'); + }); + await dispatcher.dispatch(fakeJob); + order.push('dispatch-returned'); + expect(order).toEqual(['delivered', 'dispatch-returned']); + }); + + it('propagates a delivery rejection to the caller', async () => { + const dispatcher = new InlineNotificationDispatcher(async () => { + throw new Error('webhook exploded'); + }); + await expect(dispatcher.dispatch(fakeJob)).rejects.toThrow( + 'webhook exploded', + ); + }); + + it('shutdown resolves immediately — nothing is buffered', async () => { + const dispatcher = new InlineNotificationDispatcher(async () => {}); + await expect(dispatcher.shutdown(1000)).resolves.toBeUndefined(); + }); +}); diff --git a/packages/api/src/tasks/checkAlerts/notifications.ts b/packages/api/src/tasks/checkAlerts/notifications.ts new file mode 100644 index 0000000000..47d157ed26 --- /dev/null +++ b/packages/api/src/tasks/checkAlerts/notifications.ts @@ -0,0 +1,70 @@ +import type { + Message, + PopulatedAlertChannel, +} from '@/tasks/checkAlerts/transports'; +import { deliverToChannel } from '@/tasks/checkAlerts/transports'; + +/** + * Alert notification dispatch seam. + * + * Evaluation produces fully rendered NotificationJobs; a NotificationDispatcher + * owns delivery. `eventId` is the idempotency key, so a queueing dispatcher can + * deduplicate without re-deriving it. + */ +export type NotificationJob = { + /** Idempotency key: objectHash(alertId, channel, group). */ + eventId: string; + alertId?: string; + teamId?: string; + group?: string; + /** + * Named `populatedChannel`, not `channel`, deliberately: a downstream build + * carries both a serializable channel *reference* and the resolved document, + * and reusing `channel` for the resolved one collides with that. + */ + populatedChannel: PopulatedAlertChannel; + message: Message; +}; + +export type NotificationDeliverFn = (job: NotificationJob) => Promise; + +/** + * dispatch() contract: the inline implementation resolves after delivery, so + * errors propagate to the caller. Queueing implementations resolve after + * enqueue; their errors surface in their own logs and metrics. + */ +export interface NotificationDispatcher { + dispatch(job: NotificationJob): Promise; + /** Flush anything pending, giving up after deadlineMs. */ + shutdown(deadlineMs: number): Promise; +} + +/** + * The default delivery implementation, wired to the transports registry. A + * downstream dispatcher composes this rather than calling `deliverToChannel` + * directly, so it stays swappable in one place. + * + * @public + */ +export const deliverNotification: NotificationDeliverFn = async job => { + await deliverToChannel(job.populatedChannel, job.message, { + group: job.group, + }); +}; + +/** Delivers synchronously so errors flow into the caller's executionErrors. */ +export class InlineNotificationDispatcher implements NotificationDispatcher { + constructor(private readonly deliver: NotificationDeliverFn) {} + + async dispatch(job: NotificationJob): Promise { + await this.deliver(job); + } + + async shutdown(_deadlineMs: number): Promise { + // Nothing buffered. + } +} + +export const inlineNotificationDispatcher = new InlineNotificationDispatcher( + deliverNotification, +); diff --git a/packages/api/src/tasks/checkAlerts/template.ts b/packages/api/src/tasks/checkAlerts/template.ts index 6fa1745f64..73cc62ee41 100644 --- a/packages/api/src/tasks/checkAlerts/template.ts +++ b/packages/api/src/tasks/checkAlerts/template.ts @@ -28,14 +28,16 @@ import { computeAliasWithClauses, doesExceedThreshold, } from '@/tasks/checkAlerts'; +import { + inlineNotificationDispatcher, + NotificationDispatcher, + NotificationJob, +} from '@/tasks/checkAlerts/notifications'; import { AlertProvider, PopulatedAlertChannel, } from '@/tasks/checkAlerts/providers'; -import { - createHandlebarsWithHelpers, - deliverToChannel, -} from '@/tasks/checkAlerts/transports'; +import { createHandlebarsWithHelpers } from '@/tasks/checkAlerts/transports'; import { unflattenObject } from '@/tasks/util'; import { truncateString } from '@/utils/common'; import logger from '@/utils/logger'; @@ -304,6 +306,7 @@ export const renderAlertTemplate = async ({ title, view: inputView, teamWebhooksById, + dispatcher = inlineNotificationDispatcher, }: { alertProvider: AlertProvider; clickhouseClient: ClickhouseClient; @@ -313,6 +316,7 @@ export const renderAlertTemplate = async ({ title: string; view: AlertMessageTemplateDefaultView; teamWebhooksById: Map; + dispatcher?: NotificationDispatcher; }) => { // Internal mutable view with __hdx_query_results__ populated on the // saved-search path. Untrusted values must flow through the view so @@ -399,9 +403,12 @@ export const renderAlertTemplate = async ({ ...(view.isGroupedAlert && group ? { groupId: group } : {}), }); - await deliverToChannel( - channel, - { + const job: NotificationJob = { + eventId, + alertId: alert.id, + group, + populatedChannel: channel, + message: { hdxLink: buildAlertMessageTemplateHdxLink(alertProvider, view), title, body: renderedBody, @@ -410,8 +417,9 @@ export const renderAlertTemplate = async ({ endTime, eventId, }, - { group }, - ); + }; + + await dispatcher.dispatch(job); } }); };