[backend] now also send event_id for events sent to worker coming from stream (#9824)

opencti-platform/opencti-graphql/src/database/middleware.js

@@ -1870,6 +1870,10 @@
   };
 };
 
+const computeDateFromEventId = (context) => {
+  return utcDate(parseInt(context.eventId.split('-')[0], 10)).toISOString();
+};
+
 export const updateAttributeMetaResolved = async (context, user, initial, inputs, opts = {}) => {
   const { locks = [], impactStandardId = true } = opts;
   const updates = Array.isArray(inputs) ? inputs : [inputs];
@@ -2143,7 +2147,7 @@
         const uniqImpactKey = uniqImpactKeys[i];
         attributesMap.set(uniqImpactKey, {
           name: uniqImpactKey,
-          updated_at: now(),
+          updated_at: context?.eventId ? computeDateFromEventId(context) : now(),
           confidence: confidenceLevelToApply,
           user_id: user.internal_id,
         });
@@ -2509,7 +2513,7 @@
     const { updated_at: lastAttributeUpdateDate } = attributesMap.get(attributeKey) ?? {};
     if (lastAttributeUpdateDate) {
       try {
-        const eventDate = utcDate(parseInt(context.eventId.split('-')[0], 10)).toISOString();
+        const eventDate = computeDateFromEventId(context);
         return utcDate(lastAttributeUpdateDate).isAfter(eventDate);
       } catch (e) {
         logApp.error('Error evaluating event id', { key: attributeKey, event_id: context.eventId });

opencti-platform/opencti-graphql/src/database/redis.ts

@@ -454,12 +454,13 @@
 };
 const pushToStream = async (context: AuthContext, user: AuthUser, client: Cluster | Redis, event: BaseEvent, opts: EventOpts = {}) => {
   const draftContext = getDraftContext(context, user);
+  const eventToPush = { ...event, event_id: context.eventId };
   if (!draftContext && isStreamPublishable(opts)) {
     const pushToStreamFn = async () => {
       if (streamTrimming) {
-        await client.call('XADD', REDIS_STREAM_NAME, 'MAXLEN', '~', streamTrimming, '*', ...mapJSToStream(event));
+        await client.call('XADD', REDIS_STREAM_NAME, 'MAXLEN', '~', streamTrimming, '*', ...mapJSToStream(eventToPush));
       } else {
-        await client.call('XADD', REDIS_STREAM_NAME, '*', ...mapJSToStream(event));
+        await client.call('XADD', REDIS_STREAM_NAME, '*', ...mapJSToStream(eventToPush));
       }
     };
     telemetry(context, user, 'INSERT STREAM', {

opencti-platform/opencti-graphql/src/graphql/sseMiddleware.js

@@ -2,6 +2,7 @@
 import * as jsonpatch from 'fast-json-patch';
 import { Promise } from 'bluebird';
 import { LRUCache } from 'lru-cache';
+import { now } from 'moment';
 import conf, { basePath, logApp } from '../config/conf';
 import { AUTH_BEARER, authenticateUserFromRequest, TAXIIAPI } from '../domain/user';
 import { createStreamProcessor, EVENT_CURRENT_VERSION } from '../database/redis';
@@ -33,7 +34,7 @@
   KNOWLEDGE_ORGANIZATION_RESTRICT,
   SYSTEM_USER
 } from '../utils/access';
-import { FROM_START_STR, utcDate } from '../utils/format';
+import { streamEventId, FROM_START_STR, utcDate } from '../utils/format';
 import { stixRefsExtractor } from '../schema/stixEmbeddedRelationship';
 import { ABSTRACT_STIX_CORE_RELATIONSHIP, buildRefRelationKey, ENTITY_TYPE_CONTAINER, STIX_TYPE_RELATION, STIX_TYPE_SIGHTING } from '../schema/general';
 import { convertStoreToStix } from '../database/stix-converter';
@@ -577,7 +578,9 @@
           for (let index = 0; index < elements.length; index += 1) {
             const element = elements[index];
             const { id: eventId, event, data: eventData } = element;
-            const { type, data: stix, version: eventVersion, context: evenContext } = eventData;
+            const { type, data: stix, version: eventVersion, context: evenContext, event_id } = eventData;
+            const updateTime = stix.extensions[STIX_EXT_OCTI]?.updated_at ?? now();
+            eventData.event_id = event_id ?? streamEventId(updateTime, index);
             const isRelation = stix.type === 'relationship' || stix.type === 'sighting';
             // New stream support only v4+ events.
             const isCompatibleVersion = parseInt(eventVersion ?? '0', 10) >= 4;
@@ -672,7 +675,7 @@
             const instance = instances[index];
             const stixData = convertStoreToStix(instance);
             const stixUpdatedAt = stixData.extensions[STIX_EXT_OCTI].updated_at;
-            const eventId = `${utcDate(stixUpdatedAt).toDate().getTime()}-0`;
+            const eventId = streamEventId(stixUpdatedAt);
             if (channel.connected()) {
               // publish missing dependencies if needed
               const isValidResolution = await resolveAndPublishDependencies(context, noDependencies, cache, channel, req, eventId, stixData);

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

@@ -25,7 +25,7 @@
 import { executionContext, RETENTION_MANAGER_USER, SYSTEM_USER } from '../utils/access';
 import type { SseEvent, StreamDataEvent } from '../types/event';
 import type { StixBundle } from '../types/stix-common';
-import { utcDate } from '../utils/format';
+import { streamEventId, utcDate } from '../utils/format';
 import { findById } from '../modules/playbook/playbook-domain';
 import { type CronConfiguration, PLAYBOOK_INTERNAL_DATA_CRON, type StreamConfiguration } from '../modules/playbook/playbook-components';
 import { PLAYBOOK_COMPONENTS } from '../modules/playbook/playbook-components';
@@ -326,7 +326,7 @@
       const data = await stixLoadById(context, RETENTION_MANAGER_USER, entityId);
       if (data) {
         try {
-          const eventId = `${utcDate().toDate().getTime()}-0`;
+          const eventId = streamEventId();
           const nextStep = { component: connector, instance };
           const bundle: StixBundle = {
             id: uuidv4(),
@@ -465,7 +465,7 @@
               const data = await stixLoadById(context, RETENTION_MANAGER_USER, node.internal_id);
               if (data) {
                 try {
-                  const eventId = `${utcDate().toDate().getTime()}-${index}`;
+                  const eventId = streamEventId(null, index);
                   const nextStep = { component: connector, instance };
                   const bundle: StixBundle = {
                     id: uuidv4(),

opencti-platform/opencti-graphql/src/manager/syncManager.js

@@ -33,9 +33,9 @@
   let lastStateSaveTime;
   const handleEvent = (event) => {
     const { type, data, lastEventId } = event;
-    const { data: stixData, context, version } = JSON.parse(data);
+    const { data: stixData, context, version, event_id } = JSON.parse(data);
     if (version === EVENT_CURRENT_VERSION) {
-      eventsQueue.enqueue({ id: lastEventId, type, data: stixData, context });
+      eventsQueue.enqueue({ id: lastEventId, type, data: stixData, context, event_id });
     }
   };
   const startStreamListening = (sseUri, syncElement) => {
@@ -137,7 +137,7 @@
         if (event) {
           try {
             currentDelay = await manageBackPressure(httpClient, sync, currentDelay);
-            const { id: eventId, type: eventType, data, context: eventContext } = event;
+            const { id: eventId, type: eventType, data, context: eventContext, event_id } = event;
             if (eventType === 'heartbeat') {
               await saveCurrentState(context, eventType, sync, eventId);
             } else {
@@ -147,6 +147,7 @@
               // Applicant_id should be a userId coming from synchronizer
               await pushToWorkerForConnector(sync.internal_id, {
                 type: 'event',
+                event_id,
                 synchronized,
                 previous_standard,
                 update: true,

opencti-platform/opencti-graphql/src/types/user.d.ts

@@ -57,4 +57,5 @@ interface AuthContext {
   tracing: TracingContext
   user: AuthUser | undefined
   draft_context?: string | undefined
+  eventId?: string | undefined
 }

opencti-platform/opencti-graphql/src/utils/format.js

@@ -36,6 +36,8 @@ export const UNTIL_END_STR = '5138-11-16T09:46:40.000Z';
 const dateFormat = 'YYYY-MM-DDTHH:mm:ss.SSS';
 
 export const utcDate = (date) => (date ? moment(date).utc() : moment().utc());
+export const utcEpochTime = (date = null) => utcDate(date).toDate().getTime();
+export const streamEventId = (date = null, index = 0) => `${utcEpochTime(date)}-${index}`;
 export const now = () => utcDate().toISOString();
 export const nowTime = () => timeFormat(now());
 export const sinceNowInMinutes = (lastModified) => {