From 120d15d01cf780447d57c6fa906a8e546631739a Mon Sep 17 00:00:00 2001
From: Igor Kotua <36304232+garrrikkotua@users.noreply.github.com>
Date: Wed, 15 Nov 2023 15:47:17 +0300
Subject: [PATCH] Fix data-sink-worker when activity marked deleted (#1846)

---
 .../apps/data_sink_worker/src/repo/activity.data.ts |  1 +
 .../apps/data_sink_worker/src/repo/activity.repo.ts |  4 ++--
 .../src/service/activity.service.ts                 | 13 ++++++++++++-
 3 files changed, 15 insertions(+), 3 deletions(-)

diff --git a/services/apps/data_sink_worker/src/repo/activity.data.ts b/services/apps/data_sink_worker/src/repo/activity.data.ts
index 927890871e..cd65dee2d5 100644
--- a/services/apps/data_sink_worker/src/repo/activity.data.ts
+++ b/services/apps/data_sink_worker/src/repo/activity.data.ts
@@ -22,6 +22,7 @@ export interface IDbActivity {
   url?: string
   sentiment: ISentimentAnalysisResult
   organizationId?: string
+  deletedAt?: string
 }
 
 export interface IDbActivityCreateData {
diff --git a/services/apps/data_sink_worker/src/repo/activity.repo.ts b/services/apps/data_sink_worker/src/repo/activity.repo.ts
index 1b4effe68a..d15ec7388b 100644
--- a/services/apps/data_sink_worker/src/repo/activity.repo.ts
+++ b/services/apps/data_sink_worker/src/repo/activity.repo.ts
@@ -39,12 +39,12 @@ export default class ActivityRepository extends RepositoryBase<ActivityRepositor
             title,
             channel,
             url,
-            sentiment
+            sentiment,
+            "deletedAt"
     from activities
     where "tenantId" = $(tenantId)
       and "segmentId" = $(segmentId)
       and "sourceId" = $(sourceId)
-      and "deletedAt" is null
   `
   public async findExisting(
     tenantId: string,
diff --git a/services/apps/data_sink_worker/src/service/activity.service.ts b/services/apps/data_sink_worker/src/service/activity.service.ts
index ee9fee021f..6abff28c59 100644
--- a/services/apps/data_sink_worker/src/service/activity.service.ts
+++ b/services/apps/data_sink_worker/src/service/activity.service.ts
@@ -453,6 +453,15 @@ export default class ActivityService extends LoggerBase {
           // find existing activity
           const dbActivity = await txRepo.findExisting(tenantId, segmentId, activity.sourceId)
 
+          if (dbActivity && dbActivity?.deletedAt) {
+            // we found an existing activity but it's deleted - nothing to do here
+            this.log.trace(
+              { activityId: dbActivity.id },
+              'Found existing activity but it is deleted, nothing to do here.',
+            )
+            return
+          }
+
           let createActivity = false
 
           if (dbActivity) {
@@ -846,7 +855,9 @@ export default class ActivityService extends LoggerBase {
         }
       })
 
-      await this.searchSyncWorkerEmitter.triggerMemberSync(tenantId, memberId)
+      if (memberId) {
+        await this.searchSyncWorkerEmitter.triggerMemberSync(tenantId, memberId)
+      }
       if (objectMemberId) {
         await this.searchSyncWorkerEmitter.triggerMemberSync(tenantId, objectMemberId)
       }