From e0efa3dca1b18100dc59060dbb528bc338a4c4b7 Mon Sep 17 00:00:00 2001 From: Kevin Delemme Date: Fri, 6 Mar 2026 14:05:18 -0500 Subject: [PATCH 1/5] dispatch to workflow --- .../plugins/shared/alerting_v2/kibana.jsonc | 3 +- .../server/lib/dispatcher/dispatcher.test.ts | 82 +++++---- .../integration_tests/dispatcher.test.ts | 58 +++++-- .../dispatcher/steps/dispatch_step.test.ts | 161 ++++++++++++++++-- .../lib/dispatcher/steps/dispatch_step.ts | 108 +++++++++++- .../dispatcher/steps/dispatch_step_tokens.ts | 18 ++ .../steps/fetch_policies_step.test.ts | 151 ++++++++++------ .../dispatcher/steps/fetch_policies_step.ts | 58 +++++-- .../server/lib/dispatcher/types.ts | 2 + .../alerting_v2/server/setup/bind_services.ts | 28 ++- .../shared/alerting_v2/server/types.ts | 2 + 11 files changed, 533 insertions(+), 138 deletions(-) create mode 100644 x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step_tokens.ts diff --git a/x-pack/platform/plugins/shared/alerting_v2/kibana.jsonc b/x-pack/platform/plugins/shared/alerting_v2/kibana.jsonc index df977627b73fa..38d7d7f0cf279 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/kibana.jsonc +++ b/x-pack/platform/plugins/shared/alerting_v2/kibana.jsonc @@ -15,7 +15,8 @@ "spaces", "data", "security", - "encryptedSavedObjects" + "encryptedSavedObjects", + "workflowsManagement" ], "optionalPlugins": ["management"], "extraPublicDirs": [] diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/dispatcher.test.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/dispatcher.test.ts index f93a70a6be65d..ecc4da6ae63ce 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/dispatcher.test.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/dispatcher.test.ts @@ -8,14 +8,14 @@ import type { BulkResponse } from '@elastic/elasticsearch/lib/api/types'; import type { DeeplyMockedApi } from '@kbn/core-elasticsearch-client-server-mocks'; import type { ElasticsearchClient, SavedObjectsClientContract } from '@kbn/core/server'; +import type { EncryptedSavedObjectsClient } from '@kbn/encrypted-saved-objects-plugin/server'; +import type { WorkflowsServerPluginSetup } from '@kbn/workflows-management-plugin/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'; @@ -63,30 +63,46 @@ function mockRulesBulkGet( }); } -function mockNpBulkGet(mockSoClient: jest.Mocked, 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 createMockEsoClient(policyIds: string[]): jest.Mocked { + const savedObjects = 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' }], + auth: { apiKey: 'test-api-key', owner: 'elastic', createdByUser: false }, + createdBy: null, + updatedBy: null, + createdAt: '2026-01-01T00:00:00.000Z', + updatedAt: '2026-01-01T00:00:00.000Z', + }, + references: [], + })); + + const client = { + createPointInTimeFinderDecryptedAsInternalUser: jest.fn().mockResolvedValue({ + find: jest.fn().mockImplementation(async function* () { + yield { saved_objects: savedObjects }; + }), + close: jest.fn(), + }), + } as unknown as jest.Mocked; + return client; } +const createMockWorkflowsManagement = (): jest.Mocked => + ({ + getWorkflow: jest.fn().mockResolvedValue(null), + runWorkflow: jest.fn().mockResolvedValue('exec-1'), + } as unknown as jest.Mocked); + function buildDispatcherService(deps: { queryService: QueryServiceContract; storageService: StorageServiceContract; rulesSoService: RulesSavedObjectServiceContract; - npSoService: NotificationPolicySavedObjectServiceContract; + esoClient: EncryptedSavedObjectsClient; + workflowsManagement: WorkflowsServerPluginSetup['management']; }): DispatcherService { const { loggerService } = createLoggerService(); const pipeline = new DispatcherPipeline(loggerService, [ @@ -94,11 +110,11 @@ function buildDispatcherService(deps: { new FetchSuppressionsStep(deps.queryService), new ApplySuppressionStep(), new FetchRulesStep(deps.rulesSoService), - new FetchPoliciesStep(deps.npSoService), + new FetchPoliciesStep(deps.esoClient), new EvaluateMatchersStep(), new BuildGroupsStep(), new ApplyThrottlingStep(deps.queryService, loggerService), - new DispatchStep(loggerService), + new DispatchStep(loggerService, deps.workflowsManagement), new StoreActionsStep(deps.storageService), ]); return new DispatcherService(pipeline); @@ -111,9 +127,9 @@ describe('DispatcherService', () => { let queryEsClient: DeeplyMockedApi; let storageEsClient: jest.Mocked; let rulesSoService: RulesSavedObjectServiceContract; - let npSoService: NotificationPolicySavedObjectServiceContract; let rulesMockSoClient: jest.Mocked; - let npMockSoClient: jest.Mocked; + let mockEsoClient: jest.Mocked; + let mockWfm: jest.Mocked; beforeEach(() => { ({ queryService, mockEsClient: queryEsClient } = createQueryService()); @@ -124,16 +140,15 @@ describe('DispatcherService', () => { rulesMockSoClient = rulesMock.mockSavedObjectsClient; mockRulesBulkGet(rulesMockSoClient, ['rule-1', 'rule-2']); - const npMock = createNotificationPolicySavedObjectService(); - npSoService = npMock.notificationPolicySavedObjectService; - npMockSoClient = npMock.mockSavedObjectsClient; - mockNpBulkGet(npMockSoClient, ['policy_456']); + mockEsoClient = createMockEsoClient(['policy_456']); + mockWfm = createMockWorkflowsManagement(); dispatcherService = buildDispatcherService({ queryService, storageService, rulesSoService, - npSoService, + esoClient: mockEsoClient, + workflowsManagement: mockWfm, }); }); @@ -358,16 +373,15 @@ describe('DispatcherService', () => { 'rule-005', ]); - const npMock = createNotificationPolicySavedObjectService(); - npSoService = npMock.notificationPolicySavedObjectService; - npMockSoClient = npMock.mockSavedObjectsClient; - mockNpBulkGet(npMockSoClient, ['policy_456']); + mockEsoClient = createMockEsoClient(['policy_456']); + mockWfm = createMockWorkflowsManagement(); dispatcherService = buildDispatcherService({ queryService, storageService, rulesSoService, - npSoService, + esoClient: mockEsoClient, + workflowsManagement: mockWfm, }); // Dataset: 5 rules, 9 episodes total diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/integration_tests/dispatcher.test.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/integration_tests/dispatcher.test.ts index 45252dfd104ad..2a5fb87a66d6b 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/integration_tests/dispatcher.test.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/integration_tests/dispatcher.test.ts @@ -7,27 +7,30 @@ import type { TestElasticsearchUtils, TestKibanaUtils } from '@kbn/core-test-helpers-kbn-server'; import type { ElasticsearchClient } from '@kbn/core/server'; +import type { EncryptedSavedObjectsClient } from '@kbn/encrypted-saved-objects-plugin/server'; +import type { WorkflowsServerPluginSetup } from '@kbn/workflows-management-plugin/server'; import { ALERT_ACTIONS_DATA_STREAM, type AlertAction } from '../../../resources/alert_actions'; import { ALERT_EVENTS_DATA_STREAM, type AlertEvent } from '../../../resources/alert_events'; +import { NOTIFICATION_POLICY_SAVED_OBJECT_TYPE } from '../../../saved_objects'; +import type { + RuleSavedObjectAttributes, + NotificationPolicySavedObjectAttributes, +} from '../../../saved_objects'; import type { LoggerServiceContract } from '../../services/logger_service/logger_service'; import { createLoggerService } from '../../services/logger_service/logger_service.mock'; +import { NotificationPolicySavedObjectService } from '../../services/notification_policy_saved_object_service/notification_policy_saved_object_service'; +import type { NotificationPolicySavedObjectServiceContract } from '../../services/notification_policy_saved_object_service/notification_policy_saved_object_service'; import { QueryService, type QueryServiceContract, } from '../../services/query_service/query_service'; +import { RulesSavedObjectService } from '../../services/rules_saved_object_service/rules_saved_object_service'; +import type { RulesSavedObjectServiceContract } from '../../services/rules_saved_object_service/rules_saved_object_service'; import { StorageService, type StorageServiceContract, } from '../../services/storage_service/storage_service'; import type { SpacesPluginStart } from '@kbn/spaces-plugin/server'; -import { NotificationPolicySavedObjectService } from '../../services/notification_policy_saved_object_service/notification_policy_saved_object_service'; -import type { NotificationPolicySavedObjectServiceContract } from '../../services/notification_policy_saved_object_service/notification_policy_saved_object_service'; -import { RulesSavedObjectService } from '../../services/rules_saved_object_service/rules_saved_object_service'; -import type { RulesSavedObjectServiceContract } from '../../services/rules_saved_object_service/rules_saved_object_service'; -import type { - RuleSavedObjectAttributes, - NotificationPolicySavedObjectAttributes, -} from '../../../saved_objects'; import { DispatcherService, type DispatcherServiceContract } from '../dispatcher'; import { DispatcherPipeline } from '../execution_pipeline'; import { @@ -333,6 +336,37 @@ const SUPPRESSION_USER_ACTIONS: AlertAction[] = [ }, ]; +const createMockEsoClient = ( + npSoService: NotificationPolicySavedObjectServiceContract +): EncryptedSavedObjectsClient => + ({ + createPointInTimeFinderDecryptedAsInternalUser: jest.fn().mockImplementation(async () => { + const { saved_objects: allPolicies } = await npSoService.find({ + page: 1, + perPage: 1000, + }); + return { + async *find() { + yield { + saved_objects: allPolicies.map((doc) => ({ + id: doc.id, + type: NOTIFICATION_POLICY_SAVED_OBJECT_TYPE, + attributes: doc.attributes, + references: [], + })), + }; + }, + close: jest.fn(), + }; + }), + } as unknown as EncryptedSavedObjectsClient); + +const createMockWorkflowsManagement = (): WorkflowsServerPluginSetup['management'] => + ({ + getWorkflow: jest.fn().mockResolvedValue(null), + runWorkflow: jest.fn().mockResolvedValue('exec-1'), + } as unknown as WorkflowsServerPluginSetup['management']); + describe('DispatcherService integration tests', () => { let esServer: TestElasticsearchUtils; let kibanaServer: TestKibanaUtils; @@ -343,6 +377,8 @@ describe('DispatcherService integration tests', () => { let mockLoggerService: LoggerServiceContract; let rulesSoService: RulesSavedObjectServiceContract; let npSoService: NotificationPolicySavedObjectServiceContract; + let mockEsoClient: EncryptedSavedObjectsClient; + let mockWfm: WorkflowsServerPluginSetup['management']; beforeAll(async () => { const servers = await setupTestServers(); @@ -380,17 +416,19 @@ describe('DispatcherService integration tests', () => { queryService = new QueryService(esClient, mockLoggerService); storageService = new StorageService(esClient, mockLoggerService); + mockEsoClient = createMockEsoClient(npSoService); + mockWfm = createMockWorkflowsManagement(); const pipeline = new DispatcherPipeline(mockLoggerService, [ new FetchEpisodesStep(queryService), new FetchSuppressionsStep(queryService), new ApplySuppressionStep(), new FetchRulesStep(rulesSoService), - new FetchPoliciesStep(npSoService), + new FetchPoliciesStep(mockEsoClient), new EvaluateMatchersStep(), new BuildGroupsStep(), new ApplyThrottlingStep(queryService, mockLoggerService), - new DispatchStep(mockLoggerService), + new DispatchStep(mockLoggerService, mockWfm), new StoreActionsStep(storageService), ]); dispatcherService = new DispatcherService(pipeline); diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step.test.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step.test.ts index de5d8baab322c..d1822103081a4 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step.test.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step.test.ts @@ -5,6 +5,8 @@ * 2.0. */ +import type { WorkflowDetailDto } from '@kbn/workflows'; +import type { WorkflowsServerPluginSetup } from '@kbn/workflows-management-plugin/server'; import { createLoggerService } from '../../services/logger_service/logger_service.mock'; import { createDispatcherPipelineState, @@ -13,41 +15,164 @@ import { } from '../fixtures/test_utils'; import { DispatchStep } from './dispatch_step'; +const createMockWorkflowsManagement = (): jest.Mocked => + ({ + getWorkflow: jest.fn(), + runWorkflow: jest.fn(), + } as unknown as jest.Mocked); + +const createWorkflowDetailDto = ( + overrides: Partial = {} +): WorkflowDetailDto => ({ + id: 'workflow-1', + name: 'Test Workflow', + description: 'A test workflow', + enabled: true, + createdAt: '2026-01-01T00:00:00.000Z', + createdBy: 'elastic', + lastUpdatedAt: '2026-01-01T00:00:00.000Z', + lastUpdatedBy: 'elastic', + definition: null, + yaml: 'name: Test Workflow', + valid: true, + ...overrides, +}); + describe('DispatchStep', () => { + let mockWfm: jest.Mocked; + + beforeEach(() => { + mockWfm = createMockWorkflowsManagement(); + }); + afterEach(() => jest.clearAllMocks()); - it('logs debug message for each dispatch group', async () => { - const { loggerService, mockLogger } = createLoggerService(); - const step = new DispatchStep(loggerService); + it('dispatches each group to its workflow destinations', async () => { + const { loggerService } = createLoggerService(); + const step = new DispatchStep(loggerService, mockWfm); - const group1 = createNotificationGroup({ id: 'g1', policyId: 'p1' }); - const group2 = createNotificationGroup({ id: 'g2', policyId: 'p2' }); - const policy1 = createNotificationPolicy({ - id: 'p1', + mockWfm.getWorkflow.mockResolvedValue(createWorkflowDetailDto()); + mockWfm.runWorkflow.mockResolvedValue('exec-1'); + + const group = createNotificationGroup({ + id: 'g1', + policyId: 'p1', destinations: [{ type: 'workflow', id: 'workflow-1' }], }); - const policy2 = createNotificationPolicy({ - id: 'p2', - destinations: [{ type: 'workflow', id: 'workflow-2' }], + const policy = createNotificationPolicy({ + id: 'p1', + apiKey: 'dGVzdC1pZDp0ZXN0LWtleQ==', + }); + + const state = createDispatcherPipelineState({ + dispatch: [group], + policies: new Map([['p1', policy]]), + }); + + const result = await step.execute(state); + + expect(result.type).toBe('continue'); + expect(mockWfm.getWorkflow).toHaveBeenCalledWith('workflow-1', 'default'); + expect(mockWfm.runWorkflow).toHaveBeenCalledWith( + expect.objectContaining({ id: 'workflow-1', name: 'Test Workflow' }), + 'default', + expect.objectContaining({ + id: 'g1', + ruleId: group.ruleId, + policyId: 'p1', + groupKey: group.groupKey, + episodes: group.episodes, + }), + expect.objectContaining({ + headers: expect.objectContaining({ + authorization: 'ApiKey dGVzdC1pZDp0ZXN0LWtleQ==', + }), + }) + ); + }); + + it('skips dispatch when policy has no API key', async () => { + const { loggerService, mockLogger } = createLoggerService(); + const step = new DispatchStep(loggerService, mockWfm); + + const group = createNotificationGroup({ id: 'g1', policyId: 'p1' }); + const policy = createNotificationPolicy({ id: 'p1' }); + + const state = createDispatcherPipelineState({ + dispatch: [group], + policies: new Map([['p1', policy]]), + }); + + const result = await step.execute(state); + + expect(result.type).toBe('continue'); + expect(mockLogger.warn).toHaveBeenCalledTimes(1); + expect(mockWfm.getWorkflow).not.toHaveBeenCalled(); + expect(mockWfm.runWorkflow).not.toHaveBeenCalled(); + }); + + it('skips dispatch when workflow is not found', async () => { + const { loggerService, mockLogger } = createLoggerService(); + const step = new DispatchStep(loggerService, mockWfm); + + mockWfm.getWorkflow.mockResolvedValue(null); + + const group = createNotificationGroup({ + id: 'g1', + policyId: 'p1', + destinations: [{ type: 'workflow', id: 'missing-workflow' }], + }); + const policy = createNotificationPolicy({ + id: 'p1', + apiKey: 'dGVzdC1pZDp0ZXN0LWtleQ==', }); const state = createDispatcherPipelineState({ - dispatch: [group1, group2], - policies: new Map([ - ['p1', policy1], - ['p2', policy2], - ]), + dispatch: [group], + policies: new Map([['p1', policy]]), }); const result = await step.execute(state); expect(result.type).toBe('continue'); - expect(mockLogger.debug).toHaveBeenCalledTimes(2); + expect(mockLogger.warn).toHaveBeenCalledTimes(1); + expect(mockWfm.runWorkflow).not.toHaveBeenCalled(); + }); + + it('dispatches to multiple workflow destinations', async () => { + const { loggerService } = createLoggerService(); + const step = new DispatchStep(loggerService, mockWfm); + + mockWfm.getWorkflow.mockResolvedValue(createWorkflowDetailDto()); + mockWfm.runWorkflow.mockResolvedValue('exec-1'); + + const group = createNotificationGroup({ + id: 'g1', + policyId: 'p1', + destinations: [ + { type: 'workflow', id: 'workflow-1' }, + { type: 'workflow', id: 'workflow-2' }, + ], + }); + const policy = createNotificationPolicy({ + id: 'p1', + apiKey: 'dGVzdC1pZDp0ZXN0LWtleQ==', + }); + + const state = createDispatcherPipelineState({ + dispatch: [group], + policies: new Map([['p1', policy]]), + }); + + await step.execute(state); + + expect(mockWfm.getWorkflow).toHaveBeenCalledTimes(2); + expect(mockWfm.runWorkflow).toHaveBeenCalledTimes(2); }); it('continues with no-op when dispatch is empty', async () => { const { loggerService, mockLogger } = createLoggerService(); - const step = new DispatchStep(loggerService); + const step = new DispatchStep(loggerService, mockWfm); const state = createDispatcherPipelineState({ dispatch: [] }); const result = await step.execute(state); @@ -58,7 +183,7 @@ describe('DispatchStep', () => { it('continues when dispatch is undefined', async () => { const { loggerService, mockLogger } = createLoggerService(); - const step = new DispatchStep(loggerService); + const step = new DispatchStep(loggerService, mockWfm); const state = createDispatcherPipelineState({}); const result = await step.execute(state); diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step.ts index 0ac8dd438932a..4bdb64569d330 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step.ts @@ -5,29 +5,123 @@ * 2.0. */ +import type { Headers, FakeRawRequest } from '@kbn/core-http-server'; +import { kibanaRequestFactory } from '@kbn/core-http-server-utils'; +import type { KibanaRequest } from '@kbn/core/server'; +import type { WorkflowExecutionEngineModel } from '@kbn/workflows'; +import type { WorkflowsServerPluginSetup } from '@kbn/workflows-management-plugin/server'; import { inject, injectable } from 'inversify'; import { LoggerServiceToken, type LoggerServiceContract, } from '../../services/logger_service/logger_service'; -import type { DispatcherPipelineState, DispatcherStep, DispatcherStepOutput } from '../types'; +import type { + DispatcherPipelineState, + DispatcherStep, + DispatcherStepOutput, + NotificationGroup, +} from '../types'; +import { WorkflowsManagementApiToken } from './dispatch_step_tokens'; + +const DEFAULT_SPACE_ID = 'default'; @injectable() export class DispatchStep implements DispatcherStep { public readonly name = 'dispatch'; - constructor(@inject(LoggerServiceToken) private readonly logger: LoggerServiceContract) {} + constructor( + @inject(LoggerServiceToken) private readonly logger: LoggerServiceContract, + @inject(WorkflowsManagementApiToken) + private readonly workflowsManagement: WorkflowsServerPluginSetup['management'] + ) {} public async execute(state: Readonly): Promise { - const { dispatch = [] } = state; + const { dispatch = [], policies } = state; for (const group of dispatch) { - this.logger.debug({ - message: () => - `Dispatching notification group ${group.id} for policy ${group.policyId} with ${group.destinations.length} destination(s)`, - }); + const policy = policies?.get(group.policyId); + const apiKey = policy?.apiKey; + + if (!apiKey) { + this.logger.warn({ + message: () => + `No API key found for policy ${group.policyId}, skipping dispatch of group ${group.id}`, + }); + continue; + } + + const fakeRequest = this.craftFakeRequest(apiKey); + + for (const destination of group.destinations) { + if (destination.type !== 'workflow') { + continue; + } + + await this.dispatchWorkflow(group, destination.id, fakeRequest); + } } return { type: 'continue' }; } + + private craftFakeRequest(apiKey: string): KibanaRequest { + const requestHeaders: Headers = { + authorization: `ApiKey ${apiKey}`, + }; + + const fakeRawRequest: FakeRawRequest = { + headers: requestHeaders, + path: '/', + }; + + return kibanaRequestFactory(fakeRawRequest); + } + + private async dispatchWorkflow( + group: NotificationGroup, + workflowId: string, + request: KibanaRequest + ): Promise { + const workflow = await this.workflowsManagement.getWorkflow(workflowId, DEFAULT_SPACE_ID); + + if (!workflow) { + this.logger.warn({ + message: () => `Workflow ${workflowId} not found, skipping dispatch for group ${group.id}`, + }); + return; + } + + const model: WorkflowExecutionEngineModel = { + id: workflow.id, + name: workflow.name, + enabled: workflow.enabled, + definition: workflow.definition ?? undefined, + yaml: workflow.yaml, + }; + + const payload = { + id: group.id, + ruleId: group.ruleId, + policyId: group.policyId, + groupKey: group.groupKey, + episodes: group.episodes, + }; + + this.logger.debug({ + message: () => + `Dispatching notification group ${group.id} to workflow ${workflowId} for policy ${group.policyId}`, + }); + + const executionId = await this.workflowsManagement.runWorkflow( + model, + DEFAULT_SPACE_ID, + payload, + request + ); + + this.logger.debug({ + message: () => + `Workflow ${workflowId} execution started with id ${executionId} for group ${group.id}`, + }); + } } diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step_tokens.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step_tokens.ts new file mode 100644 index 0000000000000..ae5ce23a7a66e --- /dev/null +++ b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step_tokens.ts @@ -0,0 +1,18 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the Elastic License + * 2.0; you may not use this file except in compliance with the Elastic License + * 2.0. + */ + +import type { EncryptedSavedObjectsClient } from '@kbn/encrypted-saved-objects-plugin/server'; +import type { WorkflowsServerPluginSetup } from '@kbn/workflows-management-plugin/server'; +import type { ServiceIdentifier } from 'inversify'; + +export const EncryptedSavedObjectsClientToken = Symbol.for( + 'alerting_v2.EncryptedSavedObjectsClient' +) as ServiceIdentifier; + +export const WorkflowsManagementApiToken = Symbol.for( + 'alerting_v2.WorkflowsManagementApi' +) as ServiceIdentifier; diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/fetch_policies_step.test.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/fetch_policies_step.test.ts index 81168add89d5d..965550823c175 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/fetch_policies_step.test.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/fetch_policies_step.test.ts @@ -5,43 +5,61 @@ * 2.0. */ -import type { SavedObjectsClientContract } from '@kbn/core/server'; -import { FetchPoliciesStep } from './fetch_policies_step'; -import type { NotificationPolicySavedObjectService } 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 { SavedObject } from '@kbn/core/server'; +import type { EncryptedSavedObjectsClient } from '@kbn/encrypted-saved-objects-plugin/server'; +import type { NotificationPolicySavedObjectAttributes } from '../../../saved_objects'; import { NOTIFICATION_POLICY_SAVED_OBJECT_TYPE } from '../../../saved_objects'; import { createDispatcherPipelineState, createRule } from '../fixtures/test_utils'; +import { FetchPoliciesStep } from './fetch_policies_step'; + +const createPolicySavedObject = ( + id: string, + overrides: Partial = {} +): SavedObject => ({ + id, + type: NOTIFICATION_POLICY_SAVED_OBJECT_TYPE, + attributes: { + name: `Policy ${id}`, + description: 'Test', + destinations: [{ type: 'workflow' as const, id: 'w1' }], + auth: { apiKey: `key-${id}`, owner: 'elastic', createdByUser: false }, + createdBy: null, + updatedBy: null, + createdAt: '2026-01-01T00:00:00.000Z', + updatedAt: '2026-01-01T00:00:00.000Z', + ...overrides, + }, + references: [], +}); + +const createMockFinder = ( + savedObjects: Array> +) => ({ + find: jest.fn().mockImplementation(async function* () { + yield { saved_objects: savedObjects }; + }), + close: jest.fn(), +}); + +const createMockEncryptedSavedObjectsClient = (): jest.Mocked => + ({ + createPointInTimeFinderDecryptedAsInternalUser: jest.fn(), + } as unknown as jest.Mocked); describe('FetchPoliciesStep', () => { - let npSoService: NotificationPolicySavedObjectService; - let mockSavedObjectsClient: jest.Mocked; + let mockEsoClient: jest.Mocked; beforeEach(() => { - ({ notificationPolicySavedObjectService: npSoService, mockSavedObjectsClient } = - createNotificationPolicySavedObjectService()); + mockEsoClient = createMockEncryptedSavedObjectsClient(); }); - it('fetches unique policies from rules', async () => { - mockSavedObjectsClient.bulkGet.mockResolvedValue({ - saved_objects: [ - { - id: 'p1', - type: NOTIFICATION_POLICY_SAVED_OBJECT_TYPE, - attributes: { - name: 'Policy 1', - description: 'Test', - destinations: [{ type: 'workflow' as const, id: 'w1' }], - createdBy: null, - updatedBy: null, - createdAt: '2026-01-01T00:00:00.000Z', - updatedAt: '2026-01-01T00:00:00.000Z', - }, - references: [], - }, - ], - }); + it('fetches and decrypts unique policies via PIT finder', async () => { + const mockFinderInstance = createMockFinder([createPolicySavedObject('p1')]); + mockEsoClient.createPointInTimeFinderDecryptedAsInternalUser.mockResolvedValue( + mockFinderInstance as any + ); - const step = new FetchPoliciesStep(npSoService); + const step = new FetchPoliciesStep(mockEsoClient); const state = createDispatcherPipelineState({ rules: new Map([ ['r1', createRule({ id: 'r1', notificationPolicyIds: ['p1'] })], @@ -54,15 +72,23 @@ describe('FetchPoliciesStep', () => { expect(result.type).toBe('continue'); if (result.type !== 'continue') return; expect(result.data?.policies?.size).toBe(1); - expect(result.data?.policies?.get('p1')?.name).toBe('Policy 1'); - expect(mockSavedObjectsClient.bulkGet).toHaveBeenCalledWith( - [{ type: NOTIFICATION_POLICY_SAVED_OBJECT_TYPE, id: 'p1' }], - undefined + + const policy = result.data?.policies?.get('p1'); + expect(policy?.name).toBe('Policy p1'); + expect(policy?.apiKey).toBe('key-p1'); + + expect(mockEsoClient.createPointInTimeFinderDecryptedAsInternalUser).toHaveBeenCalledTimes(1); + expect(mockEsoClient.createPointInTimeFinderDecryptedAsInternalUser).toHaveBeenCalledWith( + expect.objectContaining({ + type: NOTIFICATION_POLICY_SAVED_OBJECT_TYPE, + filter: expect.anything(), + }) ); + expect(mockFinderInstance.close).toHaveBeenCalled(); }); it('returns empty map when rules is empty', async () => { - const step = new FetchPoliciesStep(npSoService); + const step = new FetchPoliciesStep(mockEsoClient); const state = createDispatcherPipelineState({ rules: new Map() }); const result = await step.execute(state); @@ -70,11 +96,11 @@ describe('FetchPoliciesStep', () => { expect(result.type).toBe('continue'); if (result.type !== 'continue') return; expect(result.data?.policies?.size).toBe(0); - expect(mockSavedObjectsClient.bulkGet).not.toHaveBeenCalled(); + expect(mockEsoClient.createPointInTimeFinderDecryptedAsInternalUser).not.toHaveBeenCalled(); }); it('returns empty map when rules have no policy IDs', async () => { - const step = new FetchPoliciesStep(npSoService); + const step = new FetchPoliciesStep(mockEsoClient); const state = createDispatcherPipelineState({ rules: new Map([['r1', createRule({ id: 'r1', notificationPolicyIds: [] })]]), @@ -85,23 +111,20 @@ describe('FetchPoliciesStep', () => { expect(result.type).toBe('continue'); if (result.type !== 'continue') return; expect(result.data?.policies?.size).toBe(0); - expect(mockSavedObjectsClient.bulkGet).not.toHaveBeenCalled(); + expect(mockEsoClient.createPointInTimeFinderDecryptedAsInternalUser).not.toHaveBeenCalled(); }); - it('skips documents with errors', async () => { - mockSavedObjectsClient.bulkGet.mockResolvedValue({ - saved_objects: [ - { - id: 'p1', - type: NOTIFICATION_POLICY_SAVED_OBJECT_TYPE, - attributes: {}, - references: [], - error: { statusCode: 404, message: 'Not found', error: 'Not Found' }, - }, - ], - } as any); - - const step = new FetchPoliciesStep(npSoService); + it('skips saved objects with errors', async () => { + const errorSo = { + ...createPolicySavedObject('p1'), + error: { statusCode: 500, message: 'Decryption failed', error: 'Internal Server Error' }, + }; + const mockFinderInstance = createMockFinder([errorSo as any]); + mockEsoClient.createPointInTimeFinderDecryptedAsInternalUser.mockResolvedValue( + mockFinderInstance as any + ); + + const step = new FetchPoliciesStep(mockEsoClient); const state = createDispatcherPipelineState({ rules: new Map([['r1', createRule({ id: 'r1', notificationPolicyIds: ['p1'] })]]), }); @@ -112,4 +135,32 @@ describe('FetchPoliciesStep', () => { if (result.type !== 'continue') return; expect(result.data?.policies?.size).toBe(0); }); + + it('fetches multiple policies in a single PIT query', async () => { + const mockFinderInstance = createMockFinder([ + createPolicySavedObject('p1'), + createPolicySavedObject('p2'), + ]); + mockEsoClient.createPointInTimeFinderDecryptedAsInternalUser.mockResolvedValue( + mockFinderInstance as any + ); + + const step = new FetchPoliciesStep(mockEsoClient); + const state = createDispatcherPipelineState({ + rules: new Map([ + ['r1', createRule({ id: 'r1', notificationPolicyIds: ['p1'] })], + ['r2', createRule({ id: 'r2', notificationPolicyIds: ['p2'] })], + ]), + }); + + const result = await step.execute(state); + + expect(result.type).toBe('continue'); + if (result.type !== 'continue') return; + expect(result.data?.policies?.size).toBe(2); + expect(result.data?.policies?.get('p1')?.apiKey).toBe('key-p1'); + expect(result.data?.policies?.get('p2')?.apiKey).toBe('key-p2'); + expect(mockEsoClient.createPointInTimeFinderDecryptedAsInternalUser).toHaveBeenCalledTimes(1); + expect(mockFinderInstance.close).toHaveBeenCalled(); + }); }); diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/fetch_policies_step.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/fetch_policies_step.ts index 1e800148fcba0..5175e75f02f9c 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/fetch_policies_step.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/fetch_policies_step.ts @@ -5,7 +5,11 @@ * 2.0. */ +import type { EncryptedSavedObjectsClient } from '@kbn/encrypted-saved-objects-plugin/server'; +import { nodeBuilder } from '@kbn/es-query'; import { inject, injectable } from 'inversify'; +import type { NotificationPolicySavedObjectAttributes } from '../../../saved_objects'; +import { NOTIFICATION_POLICY_SAVED_OBJECT_TYPE } from '../../../saved_objects'; import type { NotificationPolicy, NotificationPolicyId, @@ -13,16 +17,15 @@ import type { DispatcherPipelineState, DispatcherStepOutput, } from '../types'; -import type { NotificationPolicySavedObjectServiceContract } from '../../services/notification_policy_saved_object_service/notification_policy_saved_object_service'; -import { NotificationPolicySavedObjectServiceInternalToken } from '../../services/notification_policy_saved_object_service/tokens'; +import { EncryptedSavedObjectsClientToken } from './dispatch_step_tokens'; @injectable() export class FetchPoliciesStep implements DispatcherStep { public readonly name = 'fetch_policies'; constructor( - @inject(NotificationPolicySavedObjectServiceInternalToken) - private readonly notificationPolicySavedObjectService: NotificationPolicySavedObjectServiceContract + @inject(EncryptedSavedObjectsClientToken) + private readonly encryptedSavedObjectsClient: EncryptedSavedObjectsClient ) {} public async execute(state: Readonly): Promise { @@ -38,22 +41,45 @@ export class FetchPoliciesStep implements DispatcherStep { return { type: 'continue', data: { policies: new Map() } }; } - const result = await this.notificationPolicySavedObjectService.bulkGetByIds(uniquePolicyIds); + const filter = nodeBuilder.or( + uniquePolicyIds.map((id) => + nodeBuilder.is( + `${NOTIFICATION_POLICY_SAVED_OBJECT_TYPE}.id`, + `${NOTIFICATION_POLICY_SAVED_OBJECT_TYPE}:${id}` + ) + ) + ); + + const finder = + await this.encryptedSavedObjectsClient.createPointInTimeFinderDecryptedAsInternalUser( + { + type: NOTIFICATION_POLICY_SAVED_OBJECT_TYPE, + filter, + } + ); + const policies = new Map(); - for (const doc of result) { - if ('error' in doc) continue; - - policies.set(doc.id, { - id: doc.id, - name: doc.attributes.name, - destinations: doc.attributes.destinations ?? [], - matcher: doc.attributes.matcher, - groupBy: doc.attributes.group_by ?? [], - throttle: doc.attributes.throttle, - }); + for await (const response of finder.find()) { + for (const doc of response.saved_objects) { + if (doc.error) { + continue; + } + + policies.set(doc.id, { + id: doc.id, + name: doc.attributes.name, + destinations: doc.attributes.destinations ?? [], + matcher: doc.attributes.matcher, + groupBy: doc.attributes.group_by ?? [], + throttle: doc.attributes.throttle, + apiKey: doc.attributes.auth.apiKey, + }); + } } + await finder.close(); + return { type: 'continue', data: { policies } }; } } diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/types.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/types.ts index ac04f7903f54b..164d709a75523 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/types.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/types.ts @@ -69,6 +69,8 @@ export interface NotificationPolicy { }; /** Target destinations to dispatch matched episodes to */ destinations: NotificationPolicyDestination[]; + /** Decrypted base64-encoded API key (id:key) for authenticated workflow dispatch */ + apiKey?: string; } export interface MatchedPair { diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/setup/bind_services.ts b/x-pack/platform/plugins/shared/alerting_v2/server/setup/bind_services.ts index 339fae13e1a1f..9238c3bf47e25 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/setup/bind_services.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/setup/bind_services.ts @@ -5,7 +5,7 @@ * 2.0. */ -import { PluginStart } from '@kbn/core-di'; +import { PluginSetup, PluginStart } from '@kbn/core-di'; import { CoreStart, Request } from '@kbn/core-di-server'; import type { ContainerModuleLoadOptions } from 'inversify'; import { AlertActionsClient } from '../lib/alert_actions_client'; @@ -49,8 +49,12 @@ import { TaskRunnerFactoryToken, } from '../lib/services/task_run_scope_service/create_task_runner'; import { UserService } from '../lib/services/user_service/user_service'; +import { + EncryptedSavedObjectsClientToken, + WorkflowsManagementApiToken, +} from '../lib/dispatcher/steps/dispatch_step_tokens'; import { NOTIFICATION_POLICY_SAVED_OBJECT_TYPE, RULE_SAVED_OBJECT_TYPE } from '../saved_objects'; -import type { AlertingServerStartDependencies } from '../types'; +import type { AlertingServerSetupDependencies, AlertingServerStartDependencies } from '../types'; export function bindServices({ bind }: ContainerModuleLoadOptions) { bind(AlertActionsClient).toSelf().inRequestScope(); @@ -144,6 +148,26 @@ export function bindServices({ bind }: ContainerModuleLoadOptions) { }) .inSingletonScope(); + bind(EncryptedSavedObjectsClientToken) + .toDynamicValue(({ get }) => { + const eso = get( + PluginStart( + 'encryptedSavedObjects' + ) + ); + return eso.getClient({ includedHiddenTypes: [NOTIFICATION_POLICY_SAVED_OBJECT_TYPE] }); + }) + .inSingletonScope(); + + bind(WorkflowsManagementApiToken) + .toDynamicValue(({ get }) => { + const wfm = get( + PluginSetup('workflowsManagement') + ); + return wfm.management; + }) + .inSingletonScope(); + bind(DispatcherService).toSelf().inSingletonScope(); bind(DispatcherServiceInternalToken).toService(DispatcherService); diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/types.ts b/x-pack/platform/plugins/shared/alerting_v2/server/types.ts index c59a1cd3cb5cf..cdcd87fe7591e 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/types.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/types.ts @@ -18,6 +18,7 @@ import type { EncryptedSavedObjectsPluginSetup, EncryptedSavedObjectsPluginStart, } from '@kbn/encrypted-saved-objects-plugin/server'; +import type { WorkflowsServerPluginSetup } from '@kbn/workflows-management-plugin/server'; export type AlertingServerSetup = void; export type AlertingServerStart = void; @@ -27,6 +28,7 @@ export interface AlertingServerSetupDependencies { features: FeaturesPluginSetup; spaces: SpacesPluginSetup; encryptedSavedObjects: EncryptedSavedObjectsPluginSetup; + workflowsManagement: WorkflowsServerPluginSetup; } export interface AlertingServerStartDependencies { From 3ec9bd9cca767ad10133fd212b81c60229fa7382 Mon Sep 17 00:00:00 2001 From: kibanamachine <42973632+kibanamachine@users.noreply.github.com> Date: Fri, 6 Mar 2026 19:22:49 +0000 Subject: [PATCH 2/5] Changes from node scripts/lint_ts_projects --fix --- x-pack/platform/plugins/shared/alerting_v2/tsconfig.json | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/x-pack/platform/plugins/shared/alerting_v2/tsconfig.json b/x-pack/platform/plugins/shared/alerting_v2/tsconfig.json index a4b0200e6e488..a8d73ad57b633 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/tsconfig.json +++ b/x-pack/platform/plugins/shared/alerting_v2/tsconfig.json @@ -58,7 +58,11 @@ "@kbn/eval-kql", "@kbn/react-query", "@kbn/core-security-server", - "@kbn/encrypted-saved-objects-plugin" + "@kbn/encrypted-saved-objects-plugin", + "@kbn/workflows-management-plugin", + "@kbn/workflows", + "@kbn/core-http-server-utils", + "@kbn/es-query" ], "exclude": ["target/**/*"] } From 79290b22dafdc9f97b0b4819afa949daa31eca10 Mon Sep 17 00:00:00 2001 From: Kevin Delemme Date: Mon, 9 Mar 2026 09:44:33 -0400 Subject: [PATCH 3/5] Consolidate into NP saved object client --- .../server/lib/dispatcher/dispatcher.test.ts | 70 ++++---- .../integration_tests/dispatcher.test.ts | 39 ++--- .../dispatcher/steps/dispatch_step_tokens.ts | 5 - .../steps/fetch_policies_step.test.ts | 152 ++++++++---------- .../dispatcher/steps/fetch_policies_step.ts | 58 +++---- ...cation_policy_saved_object_service.mock.ts | 11 +- ...otification_policy_saved_object_service.ts | 50 +++++- .../alerting_v2/server/setup/bind_services.ts | 26 ++- 8 files changed, 199 insertions(+), 212 deletions(-) diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/dispatcher.test.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/dispatcher.test.ts index ecc4da6ae63ce..e161330716b94 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/dispatcher.test.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/dispatcher.test.ts @@ -8,14 +8,15 @@ import type { BulkResponse } from '@elastic/elasticsearch/lib/api/types'; import type { DeeplyMockedApi } from '@kbn/core-elasticsearch-client-server-mocks'; import type { ElasticsearchClient, SavedObjectsClientContract } from '@kbn/core/server'; -import type { EncryptedSavedObjectsClient } from '@kbn/encrypted-saved-objects-plugin/server'; import type { WorkflowsServerPluginSetup } from '@kbn/workflows-management-plugin/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 { RULE_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'; @@ -63,32 +64,22 @@ function mockRulesBulkGet( }); } -function createMockEsoClient(policyIds: string[]): jest.Mocked { - const savedObjects = 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' }], - auth: { apiKey: 'test-api-key', owner: 'elastic', createdByUser: false }, - createdBy: null, - updatedBy: null, - createdAt: '2026-01-01T00:00:00.000Z', - updatedAt: '2026-01-01T00:00:00.000Z', - }, - references: [], - })); - - const client = { - createPointInTimeFinderDecryptedAsInternalUser: jest.fn().mockResolvedValue({ - find: jest.fn().mockImplementation(async function* () { - yield { saved_objects: savedObjects }; - }), - close: jest.fn(), - }), - } as unknown as jest.Mocked; - return client; +function mockNpBulkGetDecrypted(spy: jest.SpyInstance, policyIds: string[]) { + spy.mockResolvedValue( + policyIds.map((id) => ({ + id, + attributes: { + name: `Policy ${id}`, + description: `Description for ${id}`, + destinations: [{ type: 'workflow', id: 'workflow-test-id' }], + auth: { apiKey: 'test-api-key', owner: 'elastic', createdByUser: false }, + createdBy: null, + updatedBy: null, + createdAt: '2026-01-01T00:00:00.000Z', + updatedAt: '2026-01-01T00:00:00.000Z', + }, + })) + ); } const createMockWorkflowsManagement = (): jest.Mocked => @@ -101,7 +92,7 @@ function buildDispatcherService(deps: { queryService: QueryServiceContract; storageService: StorageServiceContract; rulesSoService: RulesSavedObjectServiceContract; - esoClient: EncryptedSavedObjectsClient; + npSoService: NotificationPolicySavedObjectServiceContract; workflowsManagement: WorkflowsServerPluginSetup['management']; }): DispatcherService { const { loggerService } = createLoggerService(); @@ -110,7 +101,7 @@ function buildDispatcherService(deps: { new FetchSuppressionsStep(deps.queryService), new ApplySuppressionStep(), new FetchRulesStep(deps.rulesSoService), - new FetchPoliciesStep(deps.esoClient), + new FetchPoliciesStep(deps.npSoService), new EvaluateMatchersStep(), new BuildGroupsStep(), new ApplyThrottlingStep(deps.queryService, loggerService), @@ -127,8 +118,9 @@ describe('DispatcherService', () => { let queryEsClient: DeeplyMockedApi; let storageEsClient: jest.Mocked; let rulesSoService: RulesSavedObjectServiceContract; + let npSoService: NotificationPolicySavedObjectServiceContract; let rulesMockSoClient: jest.Mocked; - let mockEsoClient: jest.Mocked; + let mockBulkGetDecryptedByIds: jest.SpyInstance; let mockWfm: jest.Mocked; beforeEach(() => { @@ -140,14 +132,18 @@ describe('DispatcherService', () => { rulesMockSoClient = rulesMock.mockSavedObjectsClient; mockRulesBulkGet(rulesMockSoClient, ['rule-1', 'rule-2']); - mockEsoClient = createMockEsoClient(['policy_456']); + const npMock = createNotificationPolicySavedObjectService(); + npSoService = npMock.notificationPolicySavedObjectService; + mockBulkGetDecryptedByIds = npMock.mockBulkGetDecryptedByIds; + mockNpBulkGetDecrypted(mockBulkGetDecryptedByIds, ['policy_456']); + mockWfm = createMockWorkflowsManagement(); dispatcherService = buildDispatcherService({ queryService, storageService, rulesSoService, - esoClient: mockEsoClient, + npSoService, workflowsManagement: mockWfm, }); }); @@ -373,14 +369,18 @@ describe('DispatcherService', () => { 'rule-005', ]); - mockEsoClient = createMockEsoClient(['policy_456']); + const npMock = createNotificationPolicySavedObjectService(); + npSoService = npMock.notificationPolicySavedObjectService; + mockBulkGetDecryptedByIds = npMock.mockBulkGetDecryptedByIds; + mockNpBulkGetDecrypted(mockBulkGetDecryptedByIds, ['policy_456']); + mockWfm = createMockWorkflowsManagement(); dispatcherService = buildDispatcherService({ queryService, storageService, rulesSoService, - esoClient: mockEsoClient, + npSoService, workflowsManagement: mockWfm, }); diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/integration_tests/dispatcher.test.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/integration_tests/dispatcher.test.ts index 2a5fb87a66d6b..569bf107a3ce1 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/integration_tests/dispatcher.test.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/integration_tests/dispatcher.test.ts @@ -7,11 +7,9 @@ import type { TestElasticsearchUtils, TestKibanaUtils } from '@kbn/core-test-helpers-kbn-server'; import type { ElasticsearchClient } from '@kbn/core/server'; -import type { EncryptedSavedObjectsClient } from '@kbn/encrypted-saved-objects-plugin/server'; import type { WorkflowsServerPluginSetup } from '@kbn/workflows-management-plugin/server'; import { ALERT_ACTIONS_DATA_STREAM, type AlertAction } from '../../../resources/alert_actions'; import { ALERT_EVENTS_DATA_STREAM, type AlertEvent } from '../../../resources/alert_events'; -import { NOTIFICATION_POLICY_SAVED_OBJECT_TYPE } from '../../../saved_objects'; import type { RuleSavedObjectAttributes, NotificationPolicySavedObjectAttributes, @@ -336,31 +334,6 @@ const SUPPRESSION_USER_ACTIONS: AlertAction[] = [ }, ]; -const createMockEsoClient = ( - npSoService: NotificationPolicySavedObjectServiceContract -): EncryptedSavedObjectsClient => - ({ - createPointInTimeFinderDecryptedAsInternalUser: jest.fn().mockImplementation(async () => { - const { saved_objects: allPolicies } = await npSoService.find({ - page: 1, - perPage: 1000, - }); - return { - async *find() { - yield { - saved_objects: allPolicies.map((doc) => ({ - id: doc.id, - type: NOTIFICATION_POLICY_SAVED_OBJECT_TYPE, - attributes: doc.attributes, - references: [], - })), - }; - }, - close: jest.fn(), - }; - }), - } as unknown as EncryptedSavedObjectsClient); - const createMockWorkflowsManagement = (): WorkflowsServerPluginSetup['management'] => ({ getWorkflow: jest.fn().mockResolvedValue(null), @@ -377,7 +350,6 @@ describe('DispatcherService integration tests', () => { let mockLoggerService: LoggerServiceContract; let rulesSoService: RulesSavedObjectServiceContract; let npSoService: NotificationPolicySavedObjectServiceContract; - let mockEsoClient: EncryptedSavedObjectsClient; let mockWfm: WorkflowsServerPluginSetup['management']; beforeAll(async () => { @@ -416,15 +388,22 @@ describe('DispatcherService integration tests', () => { queryService = new QueryService(esClient, mockLoggerService); storageService = new StorageService(esClient, mockLoggerService); - mockEsoClient = createMockEsoClient(npSoService); mockWfm = createMockWorkflowsManagement(); + jest.spyOn(npSoService, 'bulkGetDecryptedByIds').mockImplementation(async () => { + const { saved_objects: allPolicies } = await npSoService.find({ + page: 1, + perPage: 1000, + }); + return allPolicies.map((doc) => ({ id: doc.id, attributes: doc.attributes })); + }); + const pipeline = new DispatcherPipeline(mockLoggerService, [ new FetchEpisodesStep(queryService), new FetchSuppressionsStep(queryService), new ApplySuppressionStep(), new FetchRulesStep(rulesSoService), - new FetchPoliciesStep(mockEsoClient), + new FetchPoliciesStep(npSoService), new EvaluateMatchersStep(), new BuildGroupsStep(), new ApplyThrottlingStep(queryService, mockLoggerService), diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step_tokens.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step_tokens.ts index ae5ce23a7a66e..1a6f8cb56fdf3 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step_tokens.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step_tokens.ts @@ -5,14 +5,9 @@ * 2.0. */ -import type { EncryptedSavedObjectsClient } from '@kbn/encrypted-saved-objects-plugin/server'; import type { WorkflowsServerPluginSetup } from '@kbn/workflows-management-plugin/server'; import type { ServiceIdentifier } from 'inversify'; -export const EncryptedSavedObjectsClientToken = Symbol.for( - 'alerting_v2.EncryptedSavedObjectsClient' -) as ServiceIdentifier; - export const WorkflowsManagementApiToken = Symbol.for( 'alerting_v2.WorkflowsManagementApi' ) as ServiceIdentifier; diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/fetch_policies_step.test.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/fetch_policies_step.test.ts index 965550823c175..4cb6ec992d5b2 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/fetch_policies_step.test.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/fetch_policies_step.test.ts @@ -5,61 +5,38 @@ * 2.0. */ -import type { SavedObject } from '@kbn/core/server'; -import type { EncryptedSavedObjectsClient } from '@kbn/encrypted-saved-objects-plugin/server'; -import type { NotificationPolicySavedObjectAttributes } from '../../../saved_objects'; -import { NOTIFICATION_POLICY_SAVED_OBJECT_TYPE } from '../../../saved_objects'; +import type { NotificationPolicySavedObjectService } 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 { createDispatcherPipelineState, createRule } from '../fixtures/test_utils'; import { FetchPoliciesStep } from './fetch_policies_step'; -const createPolicySavedObject = ( - id: string, - overrides: Partial = {} -): SavedObject => ({ - id, - type: NOTIFICATION_POLICY_SAVED_OBJECT_TYPE, - attributes: { - name: `Policy ${id}`, - description: 'Test', - destinations: [{ type: 'workflow' as const, id: 'w1' }], - auth: { apiKey: `key-${id}`, owner: 'elastic', createdByUser: false }, - createdBy: null, - updatedBy: null, - createdAt: '2026-01-01T00:00:00.000Z', - updatedAt: '2026-01-01T00:00:00.000Z', - ...overrides, - }, - references: [], -}); - -const createMockFinder = ( - savedObjects: Array> -) => ({ - find: jest.fn().mockImplementation(async function* () { - yield { saved_objects: savedObjects }; - }), - close: jest.fn(), -}); - -const createMockEncryptedSavedObjectsClient = (): jest.Mocked => - ({ - createPointInTimeFinderDecryptedAsInternalUser: jest.fn(), - } as unknown as jest.Mocked); - describe('FetchPoliciesStep', () => { - let mockEsoClient: jest.Mocked; + let npSoService: NotificationPolicySavedObjectService; + let mockBulkGetDecryptedByIds: jest.SpyInstance; beforeEach(() => { - mockEsoClient = createMockEncryptedSavedObjectsClient(); + ({ notificationPolicySavedObjectService: npSoService, mockBulkGetDecryptedByIds } = + createNotificationPolicySavedObjectService()); }); - it('fetches and decrypts unique policies via PIT finder', async () => { - const mockFinderInstance = createMockFinder([createPolicySavedObject('p1')]); - mockEsoClient.createPointInTimeFinderDecryptedAsInternalUser.mockResolvedValue( - mockFinderInstance as any - ); + it('fetches and decrypts unique policies via bulkGetDecryptedByIds', async () => { + mockBulkGetDecryptedByIds.mockResolvedValue([ + { + id: 'p1', + attributes: { + name: 'Policy 1', + description: 'Test', + destinations: [{ type: 'workflow' as const, id: 'w1' }], + auth: { apiKey: 'decrypted-key', owner: 'elastic', createdByUser: false }, + createdBy: null, + updatedBy: null, + createdAt: '2026-01-01T00:00:00.000Z', + updatedAt: '2026-01-01T00:00:00.000Z', + }, + }, + ]); - const step = new FetchPoliciesStep(mockEsoClient); + const step = new FetchPoliciesStep(npSoService); const state = createDispatcherPipelineState({ rules: new Map([ ['r1', createRule({ id: 'r1', notificationPolicyIds: ['p1'] })], @@ -74,21 +51,15 @@ describe('FetchPoliciesStep', () => { expect(result.data?.policies?.size).toBe(1); const policy = result.data?.policies?.get('p1'); - expect(policy?.name).toBe('Policy p1'); - expect(policy?.apiKey).toBe('key-p1'); - - expect(mockEsoClient.createPointInTimeFinderDecryptedAsInternalUser).toHaveBeenCalledTimes(1); - expect(mockEsoClient.createPointInTimeFinderDecryptedAsInternalUser).toHaveBeenCalledWith( - expect.objectContaining({ - type: NOTIFICATION_POLICY_SAVED_OBJECT_TYPE, - filter: expect.anything(), - }) - ); - expect(mockFinderInstance.close).toHaveBeenCalled(); + expect(policy?.name).toBe('Policy 1'); + expect(policy?.apiKey).toBe('decrypted-key'); + + expect(mockBulkGetDecryptedByIds).toHaveBeenCalledTimes(1); + expect(mockBulkGetDecryptedByIds).toHaveBeenCalledWith(['p1']); }); it('returns empty map when rules is empty', async () => { - const step = new FetchPoliciesStep(mockEsoClient); + const step = new FetchPoliciesStep(npSoService); const state = createDispatcherPipelineState({ rules: new Map() }); const result = await step.execute(state); @@ -96,11 +67,11 @@ describe('FetchPoliciesStep', () => { expect(result.type).toBe('continue'); if (result.type !== 'continue') return; expect(result.data?.policies?.size).toBe(0); - expect(mockEsoClient.createPointInTimeFinderDecryptedAsInternalUser).not.toHaveBeenCalled(); + expect(mockBulkGetDecryptedByIds).not.toHaveBeenCalled(); }); it('returns empty map when rules have no policy IDs', async () => { - const step = new FetchPoliciesStep(mockEsoClient); + const step = new FetchPoliciesStep(npSoService); const state = createDispatcherPipelineState({ rules: new Map([['r1', createRule({ id: 'r1', notificationPolicyIds: [] })]]), @@ -111,20 +82,15 @@ describe('FetchPoliciesStep', () => { expect(result.type).toBe('continue'); if (result.type !== 'continue') return; expect(result.data?.policies?.size).toBe(0); - expect(mockEsoClient.createPointInTimeFinderDecryptedAsInternalUser).not.toHaveBeenCalled(); + expect(mockBulkGetDecryptedByIds).not.toHaveBeenCalled(); }); - it('skips saved objects with errors', async () => { - const errorSo = { - ...createPolicySavedObject('p1'), - error: { statusCode: 500, message: 'Decryption failed', error: 'Internal Server Error' }, - }; - const mockFinderInstance = createMockFinder([errorSo as any]); - mockEsoClient.createPointInTimeFinderDecryptedAsInternalUser.mockResolvedValue( - mockFinderInstance as any - ); - - const step = new FetchPoliciesStep(mockEsoClient); + it('skips documents with errors', async () => { + mockBulkGetDecryptedByIds.mockResolvedValue([ + { id: 'p1', error: { statusCode: 500, message: 'Decryption failed', error: 'Error' } }, + ]); + + const step = new FetchPoliciesStep(npSoService); const state = createDispatcherPipelineState({ rules: new Map([['r1', createRule({ id: 'r1', notificationPolicyIds: ['p1'] })]]), }); @@ -136,16 +102,35 @@ describe('FetchPoliciesStep', () => { expect(result.data?.policies?.size).toBe(0); }); - it('fetches multiple policies in a single PIT query', async () => { - const mockFinderInstance = createMockFinder([ - createPolicySavedObject('p1'), - createPolicySavedObject('p2'), + it('fetches multiple policies', async () => { + mockBulkGetDecryptedByIds.mockResolvedValue([ + { + id: 'p1', + attributes: { + name: 'Policy 1', + destinations: [{ type: 'workflow' as const, id: 'w1' }], + auth: { apiKey: 'key-1', owner: 'elastic', createdByUser: false }, + createdBy: null, + updatedBy: null, + createdAt: '2026-01-01T00:00:00.000Z', + updatedAt: '2026-01-01T00:00:00.000Z', + }, + }, + { + id: 'p2', + attributes: { + name: 'Policy 2', + destinations: [], + auth: { apiKey: 'key-2', owner: 'elastic', createdByUser: false }, + createdBy: null, + updatedBy: null, + createdAt: '2026-01-01T00:00:00.000Z', + updatedAt: '2026-01-01T00:00:00.000Z', + }, + }, ]); - mockEsoClient.createPointInTimeFinderDecryptedAsInternalUser.mockResolvedValue( - mockFinderInstance as any - ); - const step = new FetchPoliciesStep(mockEsoClient); + const step = new FetchPoliciesStep(npSoService); const state = createDispatcherPipelineState({ rules: new Map([ ['r1', createRule({ id: 'r1', notificationPolicyIds: ['p1'] })], @@ -158,9 +143,8 @@ describe('FetchPoliciesStep', () => { expect(result.type).toBe('continue'); if (result.type !== 'continue') return; expect(result.data?.policies?.size).toBe(2); - expect(result.data?.policies?.get('p1')?.apiKey).toBe('key-p1'); - expect(result.data?.policies?.get('p2')?.apiKey).toBe('key-p2'); - expect(mockEsoClient.createPointInTimeFinderDecryptedAsInternalUser).toHaveBeenCalledTimes(1); - expect(mockFinderInstance.close).toHaveBeenCalled(); + expect(result.data?.policies?.get('p1')?.apiKey).toBe('key-1'); + expect(result.data?.policies?.get('p2')?.apiKey).toBe('key-2'); + expect(mockBulkGetDecryptedByIds).toHaveBeenCalledTimes(1); }); }); diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/fetch_policies_step.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/fetch_policies_step.ts index 5175e75f02f9c..db09adb7d9361 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/fetch_policies_step.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/fetch_policies_step.ts @@ -5,11 +5,9 @@ * 2.0. */ -import type { EncryptedSavedObjectsClient } from '@kbn/encrypted-saved-objects-plugin/server'; -import { nodeBuilder } from '@kbn/es-query'; import { inject, injectable } from 'inversify'; -import type { NotificationPolicySavedObjectAttributes } from '../../../saved_objects'; -import { NOTIFICATION_POLICY_SAVED_OBJECT_TYPE } from '../../../saved_objects'; +import type { NotificationPolicySavedObjectServiceContract } from '../../services/notification_policy_saved_object_service/notification_policy_saved_object_service'; +import { NotificationPolicySavedObjectServiceInternalToken } from '../../services/notification_policy_saved_object_service/tokens'; import type { NotificationPolicy, NotificationPolicyId, @@ -17,15 +15,14 @@ import type { DispatcherPipelineState, DispatcherStepOutput, } from '../types'; -import { EncryptedSavedObjectsClientToken } from './dispatch_step_tokens'; @injectable() export class FetchPoliciesStep implements DispatcherStep { public readonly name = 'fetch_policies'; constructor( - @inject(EncryptedSavedObjectsClientToken) - private readonly encryptedSavedObjectsClient: EncryptedSavedObjectsClient + @inject(NotificationPolicySavedObjectServiceInternalToken) + private readonly notificationPolicySavedObjectService: NotificationPolicySavedObjectServiceContract ) {} public async execute(state: Readonly): Promise { @@ -41,44 +38,27 @@ export class FetchPoliciesStep implements DispatcherStep { return { type: 'continue', data: { policies: new Map() } }; } - const filter = nodeBuilder.or( - uniquePolicyIds.map((id) => - nodeBuilder.is( - `${NOTIFICATION_POLICY_SAVED_OBJECT_TYPE}.id`, - `${NOTIFICATION_POLICY_SAVED_OBJECT_TYPE}:${id}` - ) - ) + const result = await this.notificationPolicySavedObjectService.bulkGetDecryptedByIds( + uniquePolicyIds ); - const finder = - await this.encryptedSavedObjectsClient.createPointInTimeFinderDecryptedAsInternalUser( - { - type: NOTIFICATION_POLICY_SAVED_OBJECT_TYPE, - filter, - } - ); - const policies = new Map(); - for await (const response of finder.find()) { - for (const doc of response.saved_objects) { - if (doc.error) { - continue; - } - - policies.set(doc.id, { - id: doc.id, - name: doc.attributes.name, - destinations: doc.attributes.destinations ?? [], - matcher: doc.attributes.matcher, - groupBy: doc.attributes.group_by ?? [], - throttle: doc.attributes.throttle, - apiKey: doc.attributes.auth.apiKey, - }); + for (const doc of result) { + if ('error' in doc) { + continue; } - } - await finder.close(); + policies.set(doc.id, { + id: doc.id, + name: doc.attributes.name, + destinations: doc.attributes.destinations ?? [], + matcher: doc.attributes.matcher, + groupBy: doc.attributes.group_by ?? [], + throttle: doc.attributes.throttle, + apiKey: doc.attributes.auth.apiKey, + }); + } return { type: 'continue', data: { policies } }; } diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.mock.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.mock.ts index 672b46a299a73..11f4672033d24 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.mock.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.mock.ts @@ -13,6 +13,7 @@ import { NotificationPolicySavedObjectService } from './notification_policy_save export function createNotificationPolicySavedObjectService(): { notificationPolicySavedObjectService: NotificationPolicySavedObjectService; mockSavedObjectsClient: jest.Mocked; + mockBulkGetDecryptedByIds: jest.SpyInstance; } { const mockSavedObjectsClient = savedObjectsClientMock.create(); const mockSavedObjectsClientFactory = jest.fn().mockReturnValue(mockSavedObjectsClient); @@ -23,5 +24,13 @@ export function createNotificationPolicySavedObjectService(): { mockSpaces ); - return { notificationPolicySavedObjectService, mockSavedObjectsClient }; + const mockBulkGetDecryptedByIds = jest + .spyOn(notificationPolicySavedObjectService, 'bulkGetDecryptedByIds') + .mockResolvedValue([]); + + return { + notificationPolicySavedObjectService, + mockSavedObjectsClient, + mockBulkGetDecryptedByIds, + }; } diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.ts index 43ba392677e76..4d5c43327dddf 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.ts @@ -11,8 +11,10 @@ import { SavedObjectsClientFactory } from '@kbn/core-di-server'; import type { SavedObjectsClientContract } from '@kbn/core/server'; import { SavedObjectsUtils } from '@kbn/core/server'; import type { SavedObjectError } from '@kbn/core/types'; +import type { EncryptedSavedObjectsClient } from '@kbn/encrypted-saved-objects-plugin/server'; +import { nodeBuilder } from '@kbn/es-query'; import type { SpacesPluginStart } from '@kbn/spaces-plugin/server'; -import { inject, injectable } from 'inversify'; +import { inject, injectable, optional } from 'inversify'; import type { NotificationPolicySavedObjectAttributes } from '../../../saved_objects'; import { NOTIFICATION_POLICY_SAVED_OBJECT_TYPE } from '../../../saved_objects'; import type { AlertingServerStartDependencies } from '../../../types'; @@ -47,6 +49,7 @@ export interface NotificationPolicySavedObjectServiceContract { attrs: NotificationPolicySavedObjectAttributes; version: string; }): Promise<{ id: string; version?: string }>; + bulkGetDecryptedByIds(ids: string[]): Promise; delete(params: { id: string }): Promise; find(params: { page: number; perPage: number }): Promise<{ saved_objects: Array<{ @@ -68,7 +71,8 @@ export class NotificationPolicySavedObjectService @inject(SavedObjectsClientFactory) private readonly savedObjectsClientFactory: ISavedObjectsClientFactory, @inject(PluginStart('spaces')) - private readonly spaces: SpacesPluginStart + private readonly spaces: SpacesPluginStart, + @optional() private readonly encryptedSavedObjectsClient?: EncryptedSavedObjectsClient ) { this.client = this.savedObjectsClientFactory({ includedHiddenTypes: [NOTIFICATION_POLICY_SAVED_OBJECT_TYPE], @@ -158,6 +162,48 @@ export class NotificationPolicySavedObjectService }); } + public async bulkGetDecryptedByIds( + ids: string[] + ): Promise { + if (ids.length === 0) { + return []; + } + + if (!this.encryptedSavedObjectsClient) { + throw new Error('bulkGetDecryptedByIds requires an EncryptedSavedObjectsClient'); + } + + const filter = nodeBuilder.or( + ids.map((id) => + nodeBuilder.is( + `${NOTIFICATION_POLICY_SAVED_OBJECT_TYPE}.id`, + `${NOTIFICATION_POLICY_SAVED_OBJECT_TYPE}:${id}` + ) + ) + ); + + const finder = + await this.encryptedSavedObjectsClient.createPointInTimeFinderDecryptedAsInternalUser( + { type: NOTIFICATION_POLICY_SAVED_OBJECT_TYPE, filter } + ); + + const results: NotificationPolicySavedObjectBulkGetItem[] = []; + + for await (const response of finder.find()) { + for (const doc of response.saved_objects) { + if (doc.error) { + results.push({ id: doc.id, error: doc.error }); + } else { + results.push({ id: doc.id, attributes: doc.attributes }); + } + } + } + + await finder.close(); + + return results; + } + public async delete({ id }: { id: string }): Promise { await this.client.delete(NOTIFICATION_POLICY_SAVED_OBJECT_TYPE, id); } diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/setup/bind_services.ts b/x-pack/platform/plugins/shared/alerting_v2/server/setup/bind_services.ts index 9238c3bf47e25..182839f1aac7e 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/setup/bind_services.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/setup/bind_services.ts @@ -49,10 +49,7 @@ import { TaskRunnerFactoryToken, } from '../lib/services/task_run_scope_service/create_task_runner'; import { UserService } from '../lib/services/user_service/user_service'; -import { - EncryptedSavedObjectsClientToken, - WorkflowsManagementApiToken, -} from '../lib/dispatcher/steps/dispatch_step_tokens'; +import { WorkflowsManagementApiToken } from '../lib/dispatcher/steps/dispatch_step_tokens'; import { NOTIFICATION_POLICY_SAVED_OBJECT_TYPE, RULE_SAVED_OBJECT_TYPE } from '../saved_objects'; import type { AlertingServerSetupDependencies, AlertingServerStartDependencies } from '../types'; @@ -112,7 +109,15 @@ export function bindServices({ bind }: ContainerModuleLoadOptions) { const internalClient = savedObjects.createInternalRepository([ NOTIFICATION_POLICY_SAVED_OBJECT_TYPE, ]); - return new NotificationPolicySavedObjectService(() => internalClient, spaces); + const eso = get( + PluginStart( + 'encryptedSavedObjects' + ) + ); + const esoClient = eso.getClient({ + includedHiddenTypes: [NOTIFICATION_POLICY_SAVED_OBJECT_TYPE], + }); + return new NotificationPolicySavedObjectService(() => internalClient, spaces, esoClient); }) .inSingletonScope(); @@ -148,17 +153,6 @@ export function bindServices({ bind }: ContainerModuleLoadOptions) { }) .inSingletonScope(); - bind(EncryptedSavedObjectsClientToken) - .toDynamicValue(({ get }) => { - const eso = get( - PluginStart( - 'encryptedSavedObjects' - ) - ); - return eso.getClient({ includedHiddenTypes: [NOTIFICATION_POLICY_SAVED_OBJECT_TYPE] }); - }) - .inSingletonScope(); - bind(WorkflowsManagementApiToken) .toDynamicValue(({ get }) => { const wfm = get( From 24e03a6049d929234a1d186f41183930beda3cfe Mon Sep 17 00:00:00 2001 From: Kevin Delemme Date: Mon, 9 Mar 2026 09:46:43 -0400 Subject: [PATCH 4/5] Fix inject --- .../notification_policy_saved_object_service.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.ts index 4d5c43327dddf..33e11877da564 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.ts @@ -14,7 +14,7 @@ import type { SavedObjectError } from '@kbn/core/types'; import type { EncryptedSavedObjectsClient } from '@kbn/encrypted-saved-objects-plugin/server'; import { nodeBuilder } from '@kbn/es-query'; import type { SpacesPluginStart } from '@kbn/spaces-plugin/server'; -import { inject, injectable, optional } from 'inversify'; +import { inject, injectable, unmanaged } from 'inversify'; import type { NotificationPolicySavedObjectAttributes } from '../../../saved_objects'; import { NOTIFICATION_POLICY_SAVED_OBJECT_TYPE } from '../../../saved_objects'; import type { AlertingServerStartDependencies } from '../../../types'; @@ -72,7 +72,7 @@ export class NotificationPolicySavedObjectService private readonly savedObjectsClientFactory: ISavedObjectsClientFactory, @inject(PluginStart('spaces')) private readonly spaces: SpacesPluginStart, - @optional() private readonly encryptedSavedObjectsClient?: EncryptedSavedObjectsClient + @unmanaged() private readonly encryptedSavedObjectsClient?: EncryptedSavedObjectsClient ) { this.client = this.savedObjectsClientFactory({ includedHiddenTypes: [NOTIFICATION_POLICY_SAVED_OBJECT_TYPE], From af9715a07babd83ea547baaa0b5acd4a81b20469 Mon Sep 17 00:00:00 2001 From: Kevin Delemme Date: Mon, 9 Mar 2026 09:53:11 -0400 Subject: [PATCH 5/5] Simplify injection --- .../dispatcher/steps/dispatch_step_tokens.ts | 5 ++++ ...cation_policy_saved_object_service.mock.ts | 13 +++++++++- ...otification_policy_saved_object_service.ts | 10 +++----- .../alerting_v2/server/setup/bind_services.ts | 25 ++++++++++++------- 4 files changed, 37 insertions(+), 16 deletions(-) diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step_tokens.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step_tokens.ts index 1a6f8cb56fdf3..ae5ce23a7a66e 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step_tokens.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/lib/dispatcher/steps/dispatch_step_tokens.ts @@ -5,9 +5,14 @@ * 2.0. */ +import type { EncryptedSavedObjectsClient } from '@kbn/encrypted-saved-objects-plugin/server'; import type { WorkflowsServerPluginSetup } from '@kbn/workflows-management-plugin/server'; import type { ServiceIdentifier } from 'inversify'; +export const EncryptedSavedObjectsClientToken = Symbol.for( + 'alerting_v2.EncryptedSavedObjectsClient' +) as ServiceIdentifier; + export const WorkflowsManagementApiToken = Symbol.for( 'alerting_v2.WorkflowsManagementApi' ) as ServiceIdentifier; diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.mock.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.mock.ts index 11f4672033d24..d22626870b986 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.mock.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.mock.ts @@ -7,21 +7,31 @@ import type { SavedObjectsClientContract } from '@kbn/core/server'; import { savedObjectsClientMock } from '@kbn/core-saved-objects-api-server-mocks'; +import type { EncryptedSavedObjectsClient } from '@kbn/encrypted-saved-objects-plugin/server'; import { spacesMock } from '@kbn/spaces-plugin/server/mocks'; import { NotificationPolicySavedObjectService } from './notification_policy_saved_object_service'; +const createMockEncryptedSavedObjectsClient = (): jest.Mocked => + ({ + getDecryptedAsInternalUser: jest.fn(), + createPointInTimeFinderDecryptedAsInternalUser: jest.fn(), + } as unknown as jest.Mocked); + export function createNotificationPolicySavedObjectService(): { notificationPolicySavedObjectService: NotificationPolicySavedObjectService; mockSavedObjectsClient: jest.Mocked; + mockEncryptedSavedObjectsClient: jest.Mocked; mockBulkGetDecryptedByIds: jest.SpyInstance; } { const mockSavedObjectsClient = savedObjectsClientMock.create(); const mockSavedObjectsClientFactory = jest.fn().mockReturnValue(mockSavedObjectsClient); const mockSpaces = spacesMock.createStart(); + const mockEncryptedSavedObjectsClient = createMockEncryptedSavedObjectsClient(); const notificationPolicySavedObjectService = new NotificationPolicySavedObjectService( mockSavedObjectsClientFactory, - mockSpaces + mockSpaces, + mockEncryptedSavedObjectsClient ); const mockBulkGetDecryptedByIds = jest @@ -31,6 +41,7 @@ export function createNotificationPolicySavedObjectService(): { return { notificationPolicySavedObjectService, mockSavedObjectsClient, + mockEncryptedSavedObjectsClient, mockBulkGetDecryptedByIds, }; } diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.ts b/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.ts index 33e11877da564..efcf198ae594e 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/lib/services/notification_policy_saved_object_service/notification_policy_saved_object_service.ts @@ -14,7 +14,8 @@ import type { SavedObjectError } from '@kbn/core/types'; import type { EncryptedSavedObjectsClient } from '@kbn/encrypted-saved-objects-plugin/server'; import { nodeBuilder } from '@kbn/es-query'; import type { SpacesPluginStart } from '@kbn/spaces-plugin/server'; -import { inject, injectable, unmanaged } from 'inversify'; +import { inject, injectable } from 'inversify'; +import { EncryptedSavedObjectsClientToken } from '../../dispatcher/steps/dispatch_step_tokens'; import type { NotificationPolicySavedObjectAttributes } from '../../../saved_objects'; import { NOTIFICATION_POLICY_SAVED_OBJECT_TYPE } from '../../../saved_objects'; import type { AlertingServerStartDependencies } from '../../../types'; @@ -72,7 +73,8 @@ export class NotificationPolicySavedObjectService private readonly savedObjectsClientFactory: ISavedObjectsClientFactory, @inject(PluginStart('spaces')) private readonly spaces: SpacesPluginStart, - @unmanaged() private readonly encryptedSavedObjectsClient?: EncryptedSavedObjectsClient + @inject(EncryptedSavedObjectsClientToken) + private readonly encryptedSavedObjectsClient: EncryptedSavedObjectsClient ) { this.client = this.savedObjectsClientFactory({ includedHiddenTypes: [NOTIFICATION_POLICY_SAVED_OBJECT_TYPE], @@ -169,10 +171,6 @@ export class NotificationPolicySavedObjectService return []; } - if (!this.encryptedSavedObjectsClient) { - throw new Error('bulkGetDecryptedByIds requires an EncryptedSavedObjectsClient'); - } - const filter = nodeBuilder.or( ids.map((id) => nodeBuilder.is( diff --git a/x-pack/platform/plugins/shared/alerting_v2/server/setup/bind_services.ts b/x-pack/platform/plugins/shared/alerting_v2/server/setup/bind_services.ts index 182839f1aac7e..08946c1a8f80b 100644 --- a/x-pack/platform/plugins/shared/alerting_v2/server/setup/bind_services.ts +++ b/x-pack/platform/plugins/shared/alerting_v2/server/setup/bind_services.ts @@ -49,7 +49,10 @@ import { TaskRunnerFactoryToken, } from '../lib/services/task_run_scope_service/create_task_runner'; import { UserService } from '../lib/services/user_service/user_service'; -import { WorkflowsManagementApiToken } from '../lib/dispatcher/steps/dispatch_step_tokens'; +import { + EncryptedSavedObjectsClientToken, + WorkflowsManagementApiToken, +} from '../lib/dispatcher/steps/dispatch_step_tokens'; import { NOTIFICATION_POLICY_SAVED_OBJECT_TYPE, RULE_SAVED_OBJECT_TYPE } from '../saved_objects'; import type { AlertingServerSetupDependencies, AlertingServerStartDependencies } from '../types'; @@ -98,6 +101,17 @@ export function bindServices({ bind }: ContainerModuleLoadOptions) { }) .inSingletonScope(); + bind(EncryptedSavedObjectsClientToken) + .toDynamicValue(({ get }) => { + const eso = get( + PluginStart( + 'encryptedSavedObjects' + ) + ); + return eso.getClient({ includedHiddenTypes: [NOTIFICATION_POLICY_SAVED_OBJECT_TYPE] }); + }) + .inSingletonScope(); + bind(NotificationPolicySavedObjectService).toSelf().inRequestScope(); bind(NotificationPolicySavedObjectServiceScopedToken).toService( NotificationPolicySavedObjectService @@ -109,14 +123,7 @@ export function bindServices({ bind }: ContainerModuleLoadOptions) { const internalClient = savedObjects.createInternalRepository([ NOTIFICATION_POLICY_SAVED_OBJECT_TYPE, ]); - const eso = get( - PluginStart( - 'encryptedSavedObjects' - ) - ); - const esoClient = eso.getClient({ - includedHiddenTypes: [NOTIFICATION_POLICY_SAVED_OBJECT_TYPE], - }); + const esoClient = get(EncryptedSavedObjectsClientToken); return new NotificationPolicySavedObjectService(() => internalClient, spaces, esoClient); }) .inSingletonScope();