Alerting_v2: Dispatcher notification policy

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

@@ -10,32 +10,103 @@ import type { DeeplyMockedApi } from '@kbn/core-elasticsearch-client-server-mock
 import type { ElasticsearchClient } from '@kbn/core/server';
 import moment from 'moment';
 import { ALERT_ACTIONS_DATA_STREAM, type AlertAction } from '../../resources/alert_actions';
+import type { RuleSavedObjectAttributes } from '../../saved_objects';
 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 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 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 {
   createAlertEpisodeSuppressionsResponse,
   createDispatchableAlertEventsResponse,
+  createLastNotifiedTimestampsResponse,
 } from './fixtures/dispatcher';
 import { getDispatchableAlertEventsQuery } from './queries';
 import type { AlertEpisode, AlertEpisodeSuppression } from './types';
 
+const createMockRuleSoAttributes = (
+  overrides: Partial<RuleSavedObjectAttributes> = {}
+): RuleSavedObjectAttributes => ({
+  kind: 'alert',
+  metadata: { name: 'Test rule' },
+  time_field: '@timestamp',
+  schedule: { every: '1m' },
+  evaluation: { query: { base: 'FROM logs-*' } },
+  notification_policies: [{ ref: 'policy_456' }],
+  enabled: true,
+  createdBy: null,
+  updatedBy: null,
+  createdAt: '2026-01-01T00:00:00.000Z',
+  updatedAt: '2026-01-01T00:00:00.000Z',
+  ...overrides,
+});
+
+const createMockRulesSoService = (
+  ruleIds: string[],
+  overrides?: Partial<RuleSavedObjectAttributes>
+): RulesSavedObjectServiceContract => ({
+  bulkGetByIds: jest.fn().mockResolvedValue(
+    ruleIds.map((id) => ({
+      id,
+      attributes: createMockRuleSoAttributes(overrides),
+    }))
+  ),
+  create: jest.fn(),
+  get: jest.fn(),
+  update: jest.fn(),
+  delete: jest.fn(),
+  find: jest.fn(),
+});
+
+const createMockNpSoService = (
+  policyIds: string[]
+): NotificationPolicySavedObjectServiceContract => ({
+  bulkGetByIds: jest.fn().mockResolvedValue(
+    policyIds.map((id) => ({
+      id,
+      attributes: {
+        name: `Policy ${id}`,
+        description: `Description for ${id}`,
+        workflow_id: 'workflow-test-id',
+        createdBy: null,
+        updatedBy: null,
+        createdAt: '2026-01-01T00:00:00.000Z',
+        updatedAt: '2026-01-01T00:00:00.000Z',
+      },
+    }))
+  ),
+  create: jest.fn(),
+  get: jest.fn(),
+  update: jest.fn(),
+  delete: jest.fn(),
+});
+
 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;
 
   beforeEach(() => {
     ({ queryService, mockEsClient: queryEsClient } = createQueryService());
     ({ storageService, mockEsClient: storageEsClient } = createStorageService());
     const { loggerService } = createLoggerService();
-    dispatcherService = new DispatcherService(queryService, loggerService, storageService);
+    rulesSoService = createMockRulesSoService(['rule-1', 'rule-2']);
+    npSoService = createMockNpSoService(['policy_456']);
+    dispatcherService = new DispatcherService(
+      queryService,
+      loggerService,
+      storageService,
+      rulesSoService,
+      npSoService
+    );
   });
 
   afterEach(() => {
@@ -78,7 +149,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 +169,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 +198,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 +261,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 +278,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 +319,23 @@ describe('DispatcherService', () => {
     });
 
     it('dispatches correct fire/suppress actions across 5 rules with ack, unack, snooze, and deactivate suppressions', async () => {
+      rulesSoService = createMockRulesSoService([
+        'rule-001',
+        'rule-002',
+        'rule-003',
+        'rule-004',
+        'rule-005',
+      ]);
+      npSoService = createMockNpSoService(['policy_456']);
+      const { loggerService } = createLoggerService();
+      dispatcherService = new DispatcherService(
+        queryService,
+        loggerService,
+        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 +457,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 +472,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(