refactor(alerts): add a notification dispatch seam

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();
+  });
+});

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<void>;
+
+/**
+ * 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<void>;
+  /** Flush anything pending, giving up after deadlineMs. */
+  shutdown(deadlineMs: number): Promise<void>;
+}
+
+/**
+ * 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<void> {
+    await this.deliver(job);
+  }
+
+  async shutdown(_deadlineMs: number): Promise<void> {
+    // Nothing buffered.
+  }
+}
+
+export const inlineNotificationDispatcher = new InlineNotificationDispatcher(
+  deliverNotification,
+);

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<string, IWebhook>;
+  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);
       }
     });
   };