Fix data-sink-worker when activity marked deleted

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 {

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,

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)
       }