chore(integrations): improved logging and bugfixes

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

Author: themarolt

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

@@ -421,6 +421,7 @@ export default class ActivityService extends LoggerBase {
     const resultMap = new Map<string, { success: boolean; err?: any }>()
 
     let relevantPayloads = payloads
+    this.log.trace(`[ACTIVITY] Processing ${relevantPayloads.length} activities!`)
 
     const prepareMemberResults = this.prepareMemberData(relevantPayloads)
 
@@ -437,6 +438,10 @@ export default class ActivityService extends LoggerBase {
       return resultMap
     }
 
+    this.log.trace(
+      `[ACTIVITY] We still have ${relevantPayloads.length} activities left to process after member preparation!`,
+    )
+
     const allMemberIdentities = relevantPayloads
       .flatMap((a) => a.activity.member.identities)
       .concat(
@@ -598,6 +603,10 @@ export default class ActivityService extends LoggerBase {
 
     await Promise.all(promises)
 
+    this.log.trace(
+      `[ACTIVITY] We still have ${relevantPayloads.length} activities left to process after finding segments!`,
+    )
+
     const orConditions = relevantPayloads.map((r) => {
       return {
         and: [
@@ -823,7 +832,9 @@ export default class ActivityService extends LoggerBase {
 
     // we should have now all relevant dbActivity, dbMember and dbObjectMember set
     // we can now upsert activities and members
-
+    this.log.trace(
+      `[ACTIVITY] We still have ${relevantPayloads.length} activities left after mapping db activities and members!`,
+    )
     const preparedActivities: IActivityPrepareForUpsertResult[] = []
 
     const memberService = new MemberService(
@@ -1028,9 +1039,12 @@ export default class ActivityService extends LoggerBase {
       )
     }
 
+    this.log.trace(`[ACTIVITY] We have ${preparedActivities.length} intermediate results!`)
+
     const preparedForUpsert = preparedActivities.filter((a) => a.payload)
     const toUpsert = preparedForUpsert.map((a) => a.payload)
     if (toUpsert.length > 0) {
+      this.log.trace(`[ACTIVITY] Upserting ${toUpsert.length} activities!`)
       await insertActivities(this.client, toUpsert)
     }
 
@@ -1070,8 +1084,6 @@ export default class ActivityService extends LoggerBase {
         )
       }
 
-      resultMap.set(prepared.resultId, { success: true })
-
       await this.searchSyncWorkerEmitter.triggerMemberSync(
         prepared.payload.memberId,
         onboarding,
@@ -1094,6 +1106,11 @@ export default class ActivityService extends LoggerBase {
       }
     }
 
+    for (const prepared of preparedActivities) {
+      resultMap.set(prepared.resultId, { success: true })
+    }
+
+    this.log.trace(`[ACTIVITY] We have ${resultMap.size} results to return!`)
     return resultMap
   }
 

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

@@ -209,14 +209,17 @@ export default class DataSinkService extends LoggerBase {
 
     await Promise.all(promises)
 
-    this.log.info(`Prepared ${prepared.length} in memory results stored ${promises.length} in db!`)
+    this.log.info(
+      `[RESULTS] Prepared ${prepared.length} in memory results stored ${promises.length} in db!`,
+    )
 
     return prepared
   }
 
   public async processResults(
     batch: { resultId: string; data: IResultData | undefined; created: boolean }[],
   ): Promise<void> {
+    this.log.trace(`[RESULTS] Processing ${batch.length} results!`)
     const start = performance.now()
 
     const results: IResultData[] = []
@@ -233,6 +236,7 @@ export default class DataSinkService extends LoggerBase {
     }
 
     if (toLoadById.length > 0) {
+      this.log.info(`[RESULTS] Loading ${toLoadById.length} results from db!`)
       const infos = await this.repo.getResultInfos(toLoadById)
       results.push(...infos)
     }
@@ -269,10 +273,12 @@ export default class DataSinkService extends LoggerBase {
 
         while (toProcess.length > 0 && retry < 5) {
           if (retry > 0) {
+            this.log.trace(`[RESULTS] Retrying but sleeping first...`)
             await timeout(100)
           }
 
           try {
+            this.log.trace(`[RESULTS] Processing ${toProcess.length} activity results...`)
             const processResults = await service.processActivities(
               toProcess.map((r) => {
                 return {
@@ -296,6 +302,10 @@ export default class DataSinkService extends LoggerBase {
               }
             }
 
+            this.log.trace(
+              `[RESULTS] Processed ${processResults.size} activity results! We have total of ${resultMap.size} results for this batch and ${toProcess.length} to retry.`,
+            )
+
             // clear last error because we processed without unhandled error
             lastError = undefined
           } catch (err) {
@@ -310,6 +320,9 @@ export default class DataSinkService extends LoggerBase {
 
         // if lastError is still set and we have some left to process, we set the error for them cuz they were retried but failed
         if (lastError && toProcess.length > 0) {
+          this.log.trace(
+            `[RESULTS] Setting error for ${toProcess.length} activity results because we hit a retry limit!`,
+          )
           for (const leftToProcess of toProcess) {
             resultMap.set(leftToProcess.id, {
               success: false,
@@ -331,15 +344,18 @@ export default class DataSinkService extends LoggerBase {
             const memberData = entry.data.data as IMemberData
 
             await service.processMemberUpdate(entry.integrationId, entry.platform, memberData)
+            resultMap.set(entry.id, { success: true })
           } catch (err) {
             resultMap.set(entry.id, { success: false, err })
           }
         }
       } else {
-        this.log.error(`Unknown result type: ${type}!`)
+        this.log.error(`[RESULTS] Unknown result type: ${type}!`)
       }
     }
 
+    this.log.trace(`[RESULTS] Processing ${resultMap.size} process results!`)
+
     // handle results
     let errors = 0
     let successes = 0
@@ -371,10 +387,11 @@ export default class DataSinkService extends LoggerBase {
     }
 
     if (resultsToDelete.length > 0) {
+      this.log.trace(`[RESULTS] Deleting ${resultsToDelete.length} results from db!`)
       await this.repo.deleteResults(resultsToDelete)
     }
 
-    this.log.info(
+    this.log.trace(
       `Processed ${successes} results successfully, ${errors} with error, ${deletions} were deleted from db!`,
     )