Alerting_v2: Dispatcher notification policy

x-pack/platform/plugins/shared/alerting_v2/server/index.ts

@@ -14,6 +14,7 @@ import { bindOnStart } from './setup/bind_on_start';
 import { bindRoutes } from './setup/bind_routes';
 import { bindServices } from './setup/bind_services';
 import { bindRuleExecutionServices } from './setup/bind_rule_executor';
+import { bindDispatcherExecutionServices } from './setup/bind_dispatcher_executor';
 import { bindTasks } from './setup/bind_tasks';
 
 export const config: PluginConfigDescriptor<PluginConfig> = {
@@ -31,6 +32,7 @@ export const module = new ContainerModule((options) => {
   bindRoutes(options);
   bindServices(options);
   bindRuleExecutionServices(options);
+  bindDispatcherExecutionServices(options);
   bindTasks(options);
 });
 

x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/dispatcher.test.ts

@@ -7,35 +7,134 @@
 
 import type { BulkResponse } from '@elastic/elasticsearch/lib/api/types';
 import type { DeeplyMockedApi } from '@kbn/core-elasticsearch-client-server-mocks';
-import type { ElasticsearchClient } from '@kbn/core/server';
+import type { ElasticsearchClient, SavedObjectsClientContract } from '@kbn/core/server';
 import moment from 'moment';
 import { ALERT_ACTIONS_DATA_STREAM, type AlertAction } from '../../resources/alert_actions';
+import { RULE_SAVED_OBJECT_TYPE, NOTIFICATION_POLICY_SAVED_OBJECT_TYPE } from '../../saved_objects';
+import type { RuleSavedObjectAttributes } from '../../saved_objects';
+import { createRuleSoAttributes } from '../test_utils';
 import { createLoggerService } from '../services/logger_service/logger_service.mock';
+import type { NotificationPolicySavedObjectServiceContract } from '../services/notification_policy_saved_object_service/notification_policy_saved_object_service';
+import { createNotificationPolicySavedObjectService } from '../services/notification_policy_saved_object_service/notification_policy_saved_object_service.mock';
 import type { QueryServiceContract } from '../services/query_service/query_service';
 import { createQueryService } from '../services/query_service/query_service.mock';
+import type { RulesSavedObjectServiceContract } from '../services/rules_saved_object_service/rules_saved_object_service';
+import { createRulesSavedObjectService } from '../services/rules_saved_object_service/rules_saved_object_service.mock';
 import type { StorageServiceContract } from '../services/storage_service/storage_service';
 import { createStorageService } from '../services/storage_service/storage_service.mock';
 import { LOOKBACK_WINDOW_MINUTES } from './constants';
 import { DispatcherService } from './dispatcher';
+import { DispatcherPipeline } from './execution_pipeline';
 import {
   createAlertEpisodeSuppressionsResponse,
   createDispatchableAlertEventsResponse,
+  createLastNotifiedTimestampsResponse,
 } from './fixtures/dispatcher';
 import { getDispatchableAlertEventsQuery } from './queries';
+import {
+  FetchEpisodesStep,
+  FetchSuppressionsStep,
+  ApplySuppressionStep,
+  FetchRulesStep,
+  FetchPoliciesStep,
+  EvaluateMatchersStep,
+  BuildGroupsStep,
+  ApplyThrottlingStep,
+  DispatchStep,
+  StoreActionsStep,
+} from './steps';
 import type { AlertEpisode, AlertEpisodeSuppression } from './types';
 
+function mockRulesBulkGet(
+  mockSoClient: jest.Mocked<SavedObjectsClientContract>,
+  ruleIds: string[],
+  overrides?: Partial<RuleSavedObjectAttributes>
+) {
+  mockSoClient.bulkGet.mockResolvedValue({
+    saved_objects: ruleIds.map((id) => ({
+      id,
+      type: RULE_SAVED_OBJECT_TYPE,
+      attributes: createRuleSoAttributes({
+        notification_policies: [{ ref: 'policy_456' }],
+        ...overrides,
+      }),
+      references: [],
+    })),
+  });
+}
+
+function mockNpBulkGet(mockSoClient: jest.Mocked<SavedObjectsClientContract>, policyIds: string[]) {
+  mockSoClient.bulkGet.mockResolvedValue({
+    saved_objects: policyIds.map((id) => ({
+      id,
+      type: NOTIFICATION_POLICY_SAVED_OBJECT_TYPE,
+      attributes: {
+        name: `Policy ${id}`,
+        description: `Description for ${id}`,
+        destinations: [{ type: 'workflow', id: 'workflow-test-id' }],
+        createdBy: null,
+        updatedBy: null,
+        createdAt: '2026-01-01T00:00:00.000Z',
+        updatedAt: '2026-01-01T00:00:00.000Z',
+      },
+      references: [],
+    })),
+  });
+}
+
+function buildDispatcherService(deps: {
+  queryService: QueryServiceContract;
+  storageService: StorageServiceContract;
+  rulesSoService: RulesSavedObjectServiceContract;
+  npSoService: NotificationPolicySavedObjectServiceContract;
+}): DispatcherService {
+  const { loggerService } = createLoggerService();
+  const pipeline = new DispatcherPipeline(loggerService, [
+    new FetchEpisodesStep(deps.queryService),
+    new FetchSuppressionsStep(deps.queryService),
+    new ApplySuppressionStep(),
+    new FetchRulesStep(deps.rulesSoService),
+    new FetchPoliciesStep(deps.npSoService),
+    new EvaluateMatchersStep(),
+    new BuildGroupsStep(),
+    new ApplyThrottlingStep(deps.queryService, loggerService),
+    new DispatchStep(loggerService),
+    new StoreActionsStep(deps.storageService),
+  ]);
+  return new DispatcherService(pipeline);
+}
+
 describe('DispatcherService', () => {
   let dispatcherService: DispatcherService;
   let queryService: QueryServiceContract;
   let storageService: StorageServiceContract;
   let queryEsClient: DeeplyMockedApi<ElasticsearchClient>;
   let storageEsClient: jest.Mocked<ElasticsearchClient>;
+  let rulesSoService: RulesSavedObjectServiceContract;
+  let npSoService: NotificationPolicySavedObjectServiceContract;
+  let rulesMockSoClient: jest.Mocked<SavedObjectsClientContract>;
+  let npMockSoClient: jest.Mocked<SavedObjectsClientContract>;
 
   beforeEach(() => {
     ({ queryService, mockEsClient: queryEsClient } = createQueryService());
     ({ storageService, mockEsClient: storageEsClient } = createStorageService());
-    const { loggerService } = createLoggerService();
-    dispatcherService = new DispatcherService(queryService, loggerService, storageService);
+
+    const rulesMock = createRulesSavedObjectService();
+    rulesSoService = rulesMock.rulesSavedObjectService;
+    rulesMockSoClient = rulesMock.mockSavedObjectsClient;
+    mockRulesBulkGet(rulesMockSoClient, ['rule-1', 'rule-2']);
+
+    const npMock = createNotificationPolicySavedObjectService();
+    npSoService = npMock.notificationPolicySavedObjectService;
+    npMockSoClient = npMock.mockSavedObjectsClient;
+    mockNpBulkGet(npMockSoClient, ['policy_456']);
+
+    dispatcherService = buildDispatcherService({
+      queryService,
+      storageService,
+      rulesSoService,
+      npSoService,
+    });
   });
 
   afterEach(() => {
@@ -78,7 +177,8 @@ describe('DispatcherService', () => {
 
       queryEsClient.esql.query
         .mockResolvedValueOnce(createDispatchableAlertEventsResponse(alertEpisodes))
-        .mockResolvedValueOnce(createAlertEpisodeSuppressionsResponse(suppressions));
+        .mockResolvedValueOnce(createAlertEpisodeSuppressionsResponse(suppressions))
+        .mockResolvedValueOnce(createLastNotifiedTimestampsResponse());
 
       storageEsClient.bulk.mockResolvedValue({
         items: [{ create: { _id: '1', status: 201 } }, { create: { _id: '2', status: 201 } }],
@@ -97,7 +197,7 @@ describe('DispatcherService', () => {
         .subtract(LOOKBACK_WINDOW_MINUTES, 'minutes')
         .toISOString();
 
-      expect(queryEsClient.esql.query).toHaveBeenCalledTimes(2);
+      expect(queryEsClient.esql.query).toHaveBeenCalledTimes(3);
       expect(queryEsClient.esql.query).toHaveBeenCalledWith(
         {
           query: getDispatchableAlertEventsQuery().query,
@@ -126,7 +226,11 @@ describe('DispatcherService', () => {
       expect(createOperations).toEqual(
         expect.arrayContaining([{ create: { _index: ALERT_ACTIONS_DATA_STREAM } }])
       );
-      expect(docs).toHaveLength(alertEpisodes.length);
+
+      const fireActions = docs.filter((d: any) => d.action_type === 'fire');
+      const notifiedActions = docs.filter((d: any) => d.action_type === 'notified');
+      expect(fireActions).toHaveLength(alertEpisodes.length);
+      expect(notifiedActions.length).toBeGreaterThan(0);
 
       expect(docs).toEqual(
         expect.arrayContaining([
@@ -185,7 +289,8 @@ describe('DispatcherService', () => {
 
       queryEsClient.esql.query
         .mockResolvedValueOnce(createDispatchableAlertEventsResponse(alertEpisodes))
-        .mockResolvedValueOnce(createAlertEpisodeSuppressionsResponse(suppressions));
+        .mockResolvedValueOnce(createAlertEpisodeSuppressionsResponse(suppressions))
+        .mockResolvedValueOnce(createLastNotifiedTimestampsResponse());
 
       storageEsClient.bulk.mockResolvedValue({
         items: [{ create: { _id: '1', status: 201 } }, { create: { _id: '2', status: 201 } }],
@@ -201,7 +306,11 @@ describe('DispatcherService', () => {
       const [{ operations }] = storageEsClient.bulk.mock.calls[0];
       const safeOperations = operations ?? [];
       const docs = safeOperations.filter((_, index) => index % 2 === 1);
-      expect(docs).toHaveLength(2);
+
+      const suppressDocs = docs.filter((d: any) => d.action_type === 'suppress');
+      const fireDocs = docs.filter((d: any) => d.action_type === 'fire');
+      expect(suppressDocs).toHaveLength(1);
+      expect(fireDocs).toHaveLength(1);
 
       expect(docs).toEqual(
         expect.arrayContaining([
@@ -238,6 +347,29 @@ describe('DispatcherService', () => {
     });
 
     it('dispatches correct fire/suppress actions across 5 rules with ack, unack, snooze, and deactivate suppressions', async () => {
+      const rulesMock = createRulesSavedObjectService();
+      rulesSoService = rulesMock.rulesSavedObjectService;
+      rulesMockSoClient = rulesMock.mockSavedObjectsClient;
+      mockRulesBulkGet(rulesMockSoClient, [
+        'rule-001',
+        'rule-002',
+        'rule-003',
+        'rule-004',
+        'rule-005',
+      ]);
+
+      const npMock = createNotificationPolicySavedObjectService();
+      npSoService = npMock.notificationPolicySavedObjectService;
+      npMockSoClient = npMock.mockSavedObjectsClient;
+      mockNpBulkGet(npMockSoClient, ['policy_456']);
+
+      dispatcherService = buildDispatcherService({
+        queryService,
+        storageService,
+        rulesSoService,
+        npSoService,
+      });
+
       // Dataset: 5 rules, 9 episodes total
       // rule-001: single series, ack then unack → fire
       // rule-002: single series, ack with no unack → suppress
@@ -359,7 +491,8 @@ describe('DispatcherService', () => {
 
       queryEsClient.esql.query
         .mockResolvedValueOnce(createDispatchableAlertEventsResponse(alertEpisodes))
-        .mockResolvedValueOnce(createAlertEpisodeSuppressionsResponse(suppressions));
+        .mockResolvedValueOnce(createAlertEpisodeSuppressionsResponse(suppressions))
+        .mockResolvedValueOnce(createLastNotifiedTimestampsResponse());
 
       storageEsClient.bulk.mockResolvedValue({
         items: Array.from({ length: 10 }, (_, i) => ({
@@ -373,17 +506,18 @@ describe('DispatcherService', () => {
       });
 
       expect(result.startedAt).toBeInstanceOf(Date);
-      expect(queryEsClient.esql.query).toHaveBeenCalledTimes(2);
+      expect(queryEsClient.esql.query).toHaveBeenCalledTimes(3);
 
       const [{ operations }] = storageEsClient.bulk.mock.calls[0];
 
       const docs = (operations ?? []).filter((_, index) => index % 2 === 1) as AlertAction[];
-      expect(docs).toHaveLength(10);
 
       const fireActions = docs.filter((doc) => doc.action_type === 'fire');
       const suppressActions = docs.filter((doc) => doc.action_type === 'suppress');
+      const notifiedActions = docs.filter((doc) => doc.action_type === 'notified');
       expect(fireActions).toHaveLength(6);
       expect(suppressActions).toHaveLength(4);
+      expect(notifiedActions.length).toBeGreaterThan(0);
 
       // rule-001: fire (ack then unack cancels suppression)
       expect(docs).toEqual(