chore: integration events service (#7614)

https://linear.app/unleash/issue/2-2438/create-new-integration-event-service

https://linear.app/unleash/issue/2-2442/automatically-clean-up-old-integration-events

Adds a new `IntegrationEventsService`.

src/lib/features/integration-events/integration-events-service.ts

@@ -0,0 +1,51 @@
+import type { Logger } from '../../logger';
+import type { IFlagResolver, IUnleashConfig } from '../../types';
+import type {
+    IntegrationEventsStore,
+    IntegrationEventWriteModel,
+} from './integration-events-store';
+import type { IntegrationEventSchema } from '../../openapi/spec/integration-event-schema';
+
+export class IntegrationEventsService {
+    private readonly logger: Logger;
+    private integrationEventsStore: IntegrationEventsStore;
+    private flagResolver: IFlagResolver;
+
+    constructor(
+        {
+            integrationEventsStore,
+        }: { integrationEventsStore: IntegrationEventsStore },
+        {
+            getLogger,
+            flagResolver,
+        }: Pick<IUnleashConfig, 'getLogger' | 'flagResolver'>,
+    ) {
+        this.integrationEventsStore = integrationEventsStore;
+        this.flagResolver = flagResolver;
+        this.logger = getLogger('integration-events-service');
+    }
+
+    async getPaginatedEvents(
+        id: number,
+        limit: number,
+        offset: number,
+    ): Promise<IntegrationEventSchema[]> {
+        return this.integrationEventsStore.getPaginatedEvents(
+            id,
+            limit,
+            offset,
+        );
+    }
+
+    async registerEvent(
+        integrationEvent: IntegrationEventWriteModel,
+    ): Promise<IntegrationEventSchema> {
+        return this.integrationEventsStore.insert(integrationEvent);
+    }
+
+    async cleanUpEvents(): Promise<void> {
+        if (!this.flagResolver.isEnabled('integrationEvents')) return;
+
+        await this.integrationEventsStore.cleanUpEvents();
+    }
+}

src/lib/features/integration-events/integration-events.e2e.test.ts

@@ -5,14 +5,12 @@ import {
 } from '../../../test/e2e/helpers/test-helper';
 import getLogger from '../../../test/fixtures/no-logger';
 import { TEST_AUDIT_USER } from '../../types';
-import type {
+import type { IntegrationEventsService } from './integration-events-service';
-    IntegrationEventsStore,
+import type { IntegrationEventWriteModel } from './integration-events-store';
-    IntegrationEventWriteModel,
-} from './integration-events-store';
 
 let app: IUnleashTest;
 let db: ITestDb;
-let integrationEventsStore: IntegrationEventsStore;
+let integrationEventsService: IntegrationEventsService;
 let integrationId: number;
 
 const EVENT_SUCCESS: IntegrationEventWriteModel = {
@@ -50,7 +48,7 @@ beforeAll(async () => {
         },
         db.rawDatabase,
     );
-    integrationEventsStore = db.stores.integrationEventsStore;
+    integrationEventsService = app.services.integrationEventsService;
 });
 
 beforeEach(async () => {
@@ -80,7 +78,7 @@ const insertPastEvent = async (
     event: IntegrationEventWriteModel,
     date: Date,
 ): Promise<void> => {
-    const { id } = await integrationEventsStore.insert(event);
+    const { id } = await integrationEventsService.registerEvent(event);
 
     await db.rawDatabase.raw(
         `UPDATE integration_events SET created_at = ? WHERE id = ?`,
@@ -99,10 +97,10 @@ const getTestEventFailed = () => ({
 });
 
 test('insert and fetch integration events', async () => {
-    await integrationEventsStore.insert(getTestEventSuccess());
+    await integrationEventsService.registerEvent(getTestEventSuccess());
-    await integrationEventsStore.insert(getTestEventFailed());
+    await integrationEventsService.registerEvent(getTestEventFailed());
 
-    const events = await integrationEventsStore.getPaginatedEvents(
+    const events = await integrationEventsService.getPaginatedEvents(
         integrationId,
         10,
         0,
@@ -114,10 +112,10 @@ test('insert and fetch integration events', async () => {
 });
 
 test('paginate to latest event', async () => {
-    await integrationEventsStore.insert(getTestEventSuccess());
+    await integrationEventsService.registerEvent(getTestEventSuccess());
-    await integrationEventsStore.insert(getTestEventFailed());
+    await integrationEventsService.registerEvent(getTestEventFailed());
 
-    const events = await integrationEventsStore.getPaginatedEvents(
+    const events = await integrationEventsService.getPaginatedEvents(
         integrationId,
         1,
         0,
@@ -128,10 +126,10 @@ test('paginate to latest event', async () => {
 });
 
 test('paginate to second most recent event', async () => {
-    await integrationEventsStore.insert(getTestEventSuccess());
+    await integrationEventsService.registerEvent(getTestEventSuccess());
-    await integrationEventsStore.insert(getTestEventFailed());
+    await integrationEventsService.registerEvent(getTestEventFailed());
 
-    const events = await integrationEventsStore.getPaginatedEvents(
+    const events = await integrationEventsService.getPaginatedEvents(
         integrationId,
         1,
         1,
@@ -142,10 +140,10 @@ test('paginate to second most recent event', async () => {
 });
 
 test('paginate to non-existing event, returning empty array', async () => {
-    await integrationEventsStore.insert(getTestEventSuccess());
+    await integrationEventsService.registerEvent(getTestEventSuccess());
-    await integrationEventsStore.insert(getTestEventFailed());
+    await integrationEventsService.registerEvent(getTestEventFailed());
 
-    const events = await integrationEventsStore.getPaginatedEvents(
+    const events = await integrationEventsService.getPaginatedEvents(
         integrationId,
         1,
         999,
@@ -164,11 +162,11 @@ test('clean up events, keeping events from the last 2 hours', async () => {
     await insertPastEvent(getTestEventFailed(), twoDaysAgo);
     await insertPastEvent(getTestEventSuccess(), longTimeAgo);
     await insertPastEvent(getTestEventFailed(), oneHourAgo);
-    await integrationEventsStore.insert(getTestEventSuccess());
+    await integrationEventsService.registerEvent(getTestEventSuccess());
 
-    await integrationEventsStore.cleanUpEvents();
+    await integrationEventsService.cleanUpEvents();
 
-    const events = await integrationEventsStore.getPaginatedEvents(
+    const events = await integrationEventsService.getPaginatedEvents(
         integrationId,
         10,
         0,
@@ -181,12 +179,12 @@ test('clean up events, keeping events from the last 2 hours', async () => {
 
 test('clean up events, keeping the last 100 events', async () => {
     for (let i = 0; i < 200; i++) {
-        await integrationEventsStore.insert(getTestEventSuccess());
+        await integrationEventsService.registerEvent(getTestEventSuccess());
     }
 
-    await integrationEventsStore.cleanUpEvents();
+    await integrationEventsService.cleanUpEvents();
 
-    const events = await integrationEventsStore.getPaginatedEvents(
+    const events = await integrationEventsService.getPaginatedEvents(
         integrationId,
         200,
         0,

src/lib/features/scheduler/schedule-services.ts

@@ -31,6 +31,7 @@ export const scheduleServices = async (
         lastSeenService,
         frontendApiService,
         clientMetricsServiceV2,
+        integrationEventsService,
     } = services;
 
     schedulerService.schedule(
@@ -165,4 +166,10 @@ export const scheduleServices = async (
             'setFeatureCreatedByUserIdFromEvents',
         );
     }
+
+    schedulerService.schedule(
+        integrationEventsService.cleanUpEvents.bind(integrationEventsService),
+        minutesToMilliseconds(15),
+        'cleanUpIntegrationEvents',
+    );
 };

src/lib/services/index.ts

@@ -134,6 +134,7 @@ import {
     createApiTokenService,
     createFakeApiTokenService,
 } from '../features/api-tokens/createApiTokenService';
+import { IntegrationEventsService } from '../features/integration-events/integration-events-service';
 
 export const createServices = (
     stores: IUnleashStores,
@@ -191,6 +192,10 @@ export const createServices = (
         ? withTransactional(createTagTypeService(config), db)
         : withFakeTransactional(createFakeTagTypeService(config));
     const tagTypeService = transactionalTagTypeService;
+    const integrationEventsService = new IntegrationEventsService(
+        stores,
+        config,
+    );
     const addonService = new AddonService(
         stores,
         config,
@@ -436,6 +441,7 @@ export const createServices = (
         jobService,
         featureLifecycleService,
         transactionalFeatureLifecycleService,
+        integrationEventsService,
     };
 };
 
@@ -484,4 +490,5 @@ export {
     ProjectInsightsService,
     JobService,
     FeatureLifecycleService,
+    IntegrationEventsService,
 };

src/lib/types/services.ts

@@ -54,6 +54,7 @@ import type { InactiveUsersService } from '../users/inactive/inactive-users-serv
 import type { ProjectInsightsService } from '../features/project-insights/project-insights-service';
 import type { JobService } from '../features/scheduler/job-service';
 import type { FeatureLifecycleService } from '../features/feature-lifecycle/feature-lifecycle-service';
+import type { IntegrationEventsService } from '../features/integration-events/integration-events-service';
 
 export interface IUnleashServices {
     accessService: AccessService;
@@ -118,4 +119,5 @@ export interface IUnleashServices {
     jobService: JobService;
     featureLifecycleService: FeatureLifecycleService;
     transactionalFeatureLifecycleService: WithTransactional<FeatureLifecycleService>;
+    integrationEventsService: IntegrationEventsService;
 }