fix: ignore results that have no activity data (CM-1112) (#4033)

Signed-off-by: Uroš Marolt <uros@marolt.me>

Author: themarolt

services/apps/data_sink_worker/src/service/activity.service.ts

@@ -271,11 +271,8 @@ export default class ActivityService extends LoggerBase {
       // propagates out of prepareMemberData and crashes the entire batch, marking all other
       // results in the batch with the same error even though they are valid.
       if (!activity) {
-        this.log.error({ platform }, 'Activity data is missing.')
-        results.set(resultId, {
-          success: false,
-          err: new UnrepeatableError('Activity data is missing.'),
-        })
+        this.log.warn({ platform }, 'Activity data is missing, skipping and marking as processed.')
+        results.set(resultId, { success: true })
         continue
       }
 
@@ -435,7 +432,23 @@ export default class ActivityService extends LoggerBase {
   ): Promise<Map<string, IProcessActivityResult>> {
     const resultMap = new Map<string, IProcessActivityResult>()
 
-    let relevantPayloads = payloads
+    let relevantPayloads: IActivityProcessData[] = []
+    for (const payload of payloads) {
+      if (!payload.activity) {
+        this.log.warn(
+          {
+            resultId: payload.resultId,
+            integrationId: payload.integrationId,
+            segmentId: payload.segmentId,
+            platform: payload.platform,
+          },
+          'Activity data is missing, skipping and marking as processed.',
+        )
+        resultMap.set(payload.resultId, { success: true })
+        continue
+      }
+      relevantPayloads.push(payload)
+    }
     this.log.trace(`[ACTIVITY] Processing ${relevantPayloads.length} activities!`)
 
     const prepareMemberResults = this.prepareMemberData(relevantPayloads)