[backend] notif buffering implementation (#8685)

Author: JeremyCloarec

opencti-platform/opencti-graphql/config/default.json

@@ -302,7 +302,9 @@
   "publisher_manager": {
     "enabled": true,
     "lock_key": "publisher_manager_lock",
-    "interval": 10000
+    "interval": 10000,
+    "enable_buffering": true,
+    "buffering_seconds": 60
   },
   "sync_manager": {
     "enabled": true,

opencti-platform/opencti-graphql/package.json

@@ -8,7 +8,7 @@
     "build:types": "graphql-codegen --config graphql-codegen.yml",
     "build:schema": "yarn build:types && node builder/schema/schema.js && node build/script-generate-schema.js",
     "build:prod": "yarn build:types && yarn check-ts && node builder/prod/prod.js",
-    "build:dev": "node builder/dev/dev.js",
+    "build:dev": "yarn build:schema && node builder/dev/dev.js",
     "build:antlr": "java -jar ./builder/antlr/antlr-4.13.0-complete.jar -Dlanguage=JavaScript ./src/stixpattern/STIXPattern.g4 -o ./src/stixpattern/",
     "migrate:add": "migrate create --template-file src/utils/migration-template.js --migrations-dir=./src/migrations",
     "clean:relations": "yarn build:prod && node build/script-clean-relations.js",

opencti-platform/opencti-graphql/src/manager/publisherManager.ts

@@ -43,6 +43,8 @@ import { findById } from '../domain/user';
 
 const DOC_URI = 'https://docs.opencti.io';
 const PUBLISHER_ENGINE_KEY = conf.get('publisher_manager:lock_key');
+const PUBLISHER_ENABLE_BUFFERING = conf.get('publisher_manager:enable_buffering');
+const PUBLISHER_BUFFERING_SECONDS = conf.get('publisher_manager:buffering_seconds');
 const STREAM_SCHEDULE_TIME = 10000;
 
 export const internalProcessNotification = async (
@@ -52,11 +54,13 @@ export const internalProcessNotification = async (
   user: NotificationUser,
   notifier: BasicStoreEntityNotifier | NotifierTestInput,
   data: NotificationData[],
-  notification: BasicStoreEntityTrigger,
+  triggers: BasicStoreEntityTrigger[],
   // eslint-disable-next-line consistent-return
 ): Promise<{ error: string } | void> => {
   try {
-    const { name: notification_name, id: trigger_id, trigger_type } = notification;
+    const notification_name = triggers.map((t) => t?.name).join(';');
+    const trigger_type = triggers.length > 1 ? 'buffer' : triggers[0].trigger_type;
+    const trigger_id = triggers.map((t) => t?.id).filter((t) => t);
     const { notifier_connector_id, notifier_configuration: configuration } = notifier;
     const generatedContent: Record<string, Array<NotificationContentEvent>> = {};
     for (let index = 0; index < data.length; index += 1) {
@@ -78,7 +82,7 @@ export const internalProcessNotification = async (
     // region data generation
     const background_color = (settings.platform_theme_dark_background ?? '#0a1929').substring(1);
     const platformOpts = { doc_uri: DOC_URI, platform_uri: getBaseUrl(), background_color };
-    const templateData = { content, notification_content: content, notification, settings, user, data, ...platformOpts };
+    const templateData = { content, notification_content: content, notification: triggers[0], settings, user, data, ...platformOpts };
     // endregion
     if (notifier_connector_id === NOTIFIER_CONNECTOR_UI) {
       const createNotification = {
@@ -171,7 +175,7 @@ const processNotificationEvent = async (
     const notifier = userNotifiers[notifierIndex];
 
     // There is no await in purpose, the goal is to send notification and continue without waiting result.
-    internalProcessNotification(context, settings, notificationMap, user, notifierMap.get(notifier) ?? {} as BasicStoreEntityNotifier, data, notification).catch((reason) => logApp.error('[OPENCTI-MODULE] Publisher manager unknown error.', { cause: reason }));
+    internalProcessNotification(context, settings, notificationMap, user, notifierMap.get(notifier) ?? {} as BasicStoreEntityNotifier, data, [notification]).catch((reason) => logApp.error('[OPENCTI-MODULE] Publisher manager unknown error.', { cause: reason }));
   }
 };
 
@@ -216,6 +220,95 @@ const processDigestNotificationEvent = async (context: AuthContext, notification
   await processNotificationEvent(context, notificationMap, event.notification_id, user, dataWithFullMessage);
 };
 
+const liveNotificationBufferPerEntity: Record<string, { timestamp: number, events: SseEvent<KnowledgeNotificationEvent>[] }> = {};
+
+const processBufferedEvents = async (
+  context: AuthContext,
+  triggerMap: Map<string, BasicStoreEntityTrigger>,
+  events: KnowledgeNotificationEvent[]
+) => {
+  const usersFromCache = await getEntitiesMapFromCache<AuthUser>(context, SYSTEM_USER, ENTITY_TYPE_USER);
+  const notifDataPerUser: Record<string, { user: NotificationUser, data: NotificationData }[]> = {};
+  // We process all events to transform them into notification data per user
+  for (let i = 0; i < events.length; i += 1) {
+    const event = events[i];
+    const { targets, data: instance, origin } = event;
+    // For each event, transform it into NotificationData for all targets
+    for (let index = 0; index < targets.length; index += 1) {
+      const { user, type, message } = targets[index];
+      const notificationMessage = createFullNotificationMessage(message, usersFromCache, event.streamMessage, origin);
+      const currentData = { notification_id: event.notification_id, instance, type, message: notificationMessage };
+      const currentNotifDataForUser = notifDataPerUser[user.user_id];
+      if (currentNotifDataForUser) {
+        currentNotifDataForUser.push({ user, data: currentData });
+      } else {
+        notifDataPerUser[user.user_id] = [{ user, data: currentData }];
+      }
+    }
+  }
+  const settings = await getEntityFromCache<BasicStoreSettings>(context, SYSTEM_USER, ENTITY_TYPE_SETTINGS);
+  const allNotifiers = await getEntitiesListFromCache<BasicStoreEntityNotifier>(context, SYSTEM_USER, ENTITY_TYPE_NOTIFIER);
+  const allNotifiersMap = new Map(allNotifiers.map((n) => [n.internal_id, n]));
+
+  const notifUsers = Object.keys(notifDataPerUser);
+  // Handle notification data for each user
+  for (let i = 0; i < notifUsers.length; i += 1) {
+    const currentUserId = notifUsers[i];
+    const userNotificationData = notifDataPerUser[currentUserId];
+    const userNotifiers = [...new Set(userNotificationData.map((d) => d.user.notifiers).flat())];
+
+    // For each notifier of the user, filter the relevant notification data, and send it
+    for (let notifierIndex = 0; notifierIndex < userNotifiers.length; notifierIndex += 1) {
+      const notifier = userNotifiers[notifierIndex];
+
+      // Only include the notificationData that has the current notifier included in its trigger config
+      const impactedData = userNotificationData.filter((d) => d.user.notifiers.includes(notifier));
+      if (impactedData.length > 0) {
+        const currentUser = impactedData[0].user;
+        const dataToSend = impactedData.map((d) => d.data);
+        const triggersInDataToSend = [...new Set(dataToSend.map((d) => triggerMap.get(d.notification_id)).filter((t) => t))];
+        // If triggers can't be found, no need to send the data
+        if (triggersInDataToSend.length >= 1) {
+          // There is no await in purpose, the goal is to send notification and continue without waiting result.
+          internalProcessNotification(
+            context,
+            settings,
+            triggerMap,
+            currentUser,
+            allNotifiersMap.get(notifier) ?? {} as BasicStoreEntityNotifier,
+            dataToSend,
+            triggersInDataToSend as BasicStoreEntityTrigger[]
+          ).catch((reason) => logApp.error('[OPENCTI-MODULE] Publisher manager unknown error.', { cause: reason }));
+        }
+      }
+    }
+  }
+};
+
+const handleEntityNotificationBuffer = async (forceSend = false) => {
+  const dateNow = Date.now();
+  const context = executionContext('publisher_manager');
+  const bufferKeys = Object.keys(liveNotificationBufferPerEntity);
+  // Iterate on all buffers to check if they need to be sent
+  for (let i = 0; i < bufferKeys.length; i += 1) {
+    const key = bufferKeys[i];
+    const value = liveNotificationBufferPerEntity[key];
+    if (value) {
+      const isBufferingTimeElapsed = (dateNow - value.timestamp) > PUBLISHER_BUFFERING_SECONDS * 1000;
+      // If buffer is older than configured buffering time length OR we want to forceSend, it needs to be sent
+      if (forceSend || isBufferingTimeElapsed) {
+        const bufferEvents = value.events.map((e) => e.data);
+        // We remove current buffer from buffers map before processing buffer events, otherwise some new events coming in might be lost
+        // This way, if new events are coming in from the stream, they will initiate a new buffer that will be handled later
+        delete liveNotificationBufferPerEntity[key];
+        const allExistingTriggers = await getNotifications(context);
+        const allExistingTriggersMap = new Map(allExistingTriggers.map((n) => [n.trigger.internal_id, n.trigger]));
+        await processBufferedEvents(context, allExistingTriggersMap, bufferEvents);
+      }
+    }
+  }
+};
+
 const publisherStreamHandler = async (streamEvents: Array<SseEvent<StreamNotifEvent>>) => {
   try {
     if (streamEvents.length === 0) {
@@ -229,7 +322,19 @@ const publisherStreamHandler = async (streamEvents: Array<SseEvent<StreamNotifEv
       const { data: { notification_id, type } } = streamEvent;
       if (type === 'live' || type === 'action') {
         const liveEvent = streamEvent as SseEvent<KnowledgeNotificationEvent>;
-        await processLiveNotificationEvent(context, notificationMap, liveEvent.data);
+        // If buffering is enabled, we store the event in local buffer instead of handling it directly
+        if (PUBLISHER_ENABLE_BUFFERING) {
+          const liveEventEntityId = liveEvent.data.data.id;
+          const currentEntityBuffer = liveNotificationBufferPerEntity[liveEventEntityId];
+          // If there are buffered events already, simply add current event to array of buffered events
+          if (currentEntityBuffer) {
+            currentEntityBuffer.events.push(liveEvent);
+          } else { // If there are currently no buffered events for this entity, initialize them using current time as timestamp
+            liveNotificationBufferPerEntity[liveEventEntityId] = { timestamp: Date.now(), events: [liveEvent] };
+          }
+        } else { // If no buffering is enabled, we handle the notification directly
+          await processLiveNotificationEvent(context, notificationMap, liveEvent.data);
+        }
       }
       if (type === 'digest') {
         const digestEvent = streamEvent as SseEvent<DigestEvent>;
@@ -269,6 +374,9 @@ const initPublisherManager = () => {
       await streamProcessor.start('live');
       while (!shutdown && streamProcessor.running()) {
         lock.signal.throwIfAborted();
+        if (PUBLISHER_ENABLE_BUFFERING) {
+          await handleEntityNotificationBuffer();
+        }
         await wait(WAIT_TIME_ACTION);
       }
       logApp.info('[OPENCTI-MODULE] End of publisher manager processing');
@@ -280,6 +388,9 @@ const initPublisherManager = () => {
       }
     } finally {
       if (streamProcessor) await streamProcessor.shutdown();
+      if (PUBLISHER_ENABLE_BUFFERING) {
+        await handleEntityNotificationBuffer(true);
+      }
       if (lock) await lock.unlock();
     }
   };

opencti-platform/opencti-graphql/src/modules/notification/notification-types.ts

@@ -73,7 +73,7 @@ export interface NotificationAddInput {
     title: string,
     events: Array<NotificationContentEvent>
   }>
-  trigger_id?: string
+  trigger_id?: string | string[]
   user_id: string
 }
 

opencti-platform/opencti-graphql/src/modules/notification/notification.ts

@@ -73,7 +73,7 @@ const NOTIFICATION_DEFINITION: ModuleDefinition<StoreEntityNotification, StixNot
   },
   attributes: [
     { name: 'name', label: 'Trigger name', type: 'string', format: 'short', mandatoryType: 'internal', editDefault: false, multiple: false, upsert: false, isFilterable: true },
-    { name: 'trigger_id', label: 'Trigger', type: 'string', format: 'id', entityTypes: [ENTITY_TYPE_TRIGGER], mandatoryType: 'no', editDefault: false, multiple: false, upsert: false, isFilterable: true },
+    { name: 'trigger_id', label: 'Trigger', type: 'string', format: 'id', entityTypes: [ENTITY_TYPE_TRIGGER], mandatoryType: 'no', editDefault: false, multiple: true, upsert: false, isFilterable: true },
     { name: 'notification_type', label: 'Notification type', type: 'string', format: 'enum', values: TRIGGER_TYPE_VALUES, mandatoryType: 'internal', editDefault: false, multiple: false, upsert: false, isFilterable: true },
     {
       name: 'notification_content',

opencti-platform/opencti-graphql/src/modules/notifier/notifier-domain.ts

@@ -189,6 +189,6 @@ export const testNotifier = async (context: AuthContext, user: AuthUser, notifie
     user_id: user.id,
     user_email: user.user_email,
     notifiers: [],
-  }, notifier, MOCK_NOTIFICATIONS[notifier.notifier_test_id], { created: (new Date()).toISOString() } as unknown as BasicStoreEntityTrigger);
+  }, notifier, MOCK_NOTIFICATIONS[notifier.notifier_test_id], [{ created: (new Date()).toISOString() }] as unknown as BasicStoreEntityTrigger[]);
   return result?.error;
 };