fix: some refactoring and better control on logExecutionTimeV2 duration logging

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

Author: themarolt

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

@@ -1581,20 +1581,6 @@ export default class ActivityService extends LoggerBase {
         )
       }
 
-      await this.searchSyncWorkerEmitter.triggerMemberSync(
-        prepared.payload.memberId,
-        onboarding,
-        prepared.payload.segmentId,
-      )
-
-      if (prepared.payload.objectMemberId) {
-        await this.searchSyncWorkerEmitter.triggerMemberSync(
-          prepared.payload.objectMemberId,
-          onboarding,
-          prepared.payload.segmentId,
-        )
-      }
-
       if (prepared.payload.organizationId) {
         await this.redisClient.sAdd(
           'organizationIdsForAggComputation',
@@ -1603,6 +1589,28 @@ export default class ActivityService extends LoggerBase {
       }
     }
 
+    // Deduplicate member sync triggers — a member may appear in many activities in the
+    // same batch. Emit once per unique (memberId, segmentId) pair.
+    const memberSyncKeys = new Set<string>()
+    const memberSyncPromises: Promise<void>[] = []
+    for (const prepared of preparedForUpsert) {
+      for (const memberId of [prepared.payload.memberId, prepared.payload.objectMemberId]) {
+        if (!memberId) continue
+        const key = `${memberId}:${prepared.payload.segmentId}`
+        if (!memberSyncKeys.has(key)) {
+          memberSyncKeys.add(key)
+          memberSyncPromises.push(
+            this.searchSyncWorkerEmitter.triggerMemberSync(
+              memberId,
+              onboarding,
+              prepared.payload.segmentId,
+            ),
+          )
+        }
+      }
+    }
+    await Promise.all(memberSyncPromises)
+
     for (const prepared of preparedActivities) {
       resultMap.set(prepared.resultId, { success: true })
     }

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

@@ -229,7 +229,6 @@ export default class MemberService extends LoggerBase {
 
           if (organizations.length > 0) {
             const uniqOrgs = uniqby(organizations, 'id')
-            const orgService = new OrganizationService(this.store, this.log)
 
             const orgsToAdd = (
               await Promise.all(
@@ -451,7 +450,6 @@ export default class MemberService extends LoggerBase {
 
           if (organizations.length > 0) {
             const uniqOrgs = uniqby(organizations, 'id')
-            const orgService = new OrganizationService(this.store, this.log)
 
             this.log.trace({ memberId: id }, 'Finding member organizations!')
             const orgsToAdd = (

services/libs/database/src/connection.ts

@@ -129,7 +129,9 @@ export const getDbConnection = async (
   })
 
   const profile = process.env['CROWD_POSTGRESQL_PROFILE_QUERIES'] !== undefined
-  const minQueryDuration = Number(process.env['CROWD_POSTGRESQL_PROFILE_QUERIES_MIN_DURATION'] || 0)
+  const minQueryDurationMs = Number(
+    process.env['CROWD_POSTGRESQL_PROFILE_QUERIES_MIN_DURATION'] || 0,
+  )
 
   const oldQuery = client.query
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
@@ -141,7 +143,7 @@ export const getDbConnection = async (
       return result
     } finally {
       const duration = performance.now() - start
-      if (profile && duration >= minQueryDuration) {
+      if (profile && duration >= minQueryDurationMs) {
         const durationSeconds = duration / 1000.0
         log.warn(
           { durationSeconds: durationSeconds.toFixed(2), query, values: options },

services/libs/logging/src/utility.ts

@@ -27,24 +27,29 @@ export const logExecutionTimeV2 = async <T>(
   if (!process.env.CROWD_LOG_EXECUTION_TIME) {
     return toProcess()
   }
+
+  const minDurationMs = Number(process.env['CROWD_LOG_EXECUTION_TIME_MIN_DURATION'] || 0)
+
   const start = performance.now()
 
-  const end = () => {
-    const end = performance.now()
-    const duration = end - start
-    const durationInSeconds = duration / 1000
-    return durationInSeconds.toFixed(2)
-  }
+  const formatDuration = (durationMs: number) => (durationMs / 1000).toFixed(2)
+
   try {
     if (process.env.CROWD_LOG_EXECUTION_START) {
       log.info(`Starting process ${name}...`)
     }
 
     const result = await toProcess()
-    log.info(`Process ${name} took ${end()} seconds!`)
+    const durationMs = performance.now() - start
+    if (durationMs >= minDurationMs) {
+      log.info(`Process ${name} took ${formatDuration(durationMs)} seconds!`)
+    }
     return result
   } catch (e) {
-    log.info(`Process ${name} failed after ${end()} seconds!`)
+    const durationMs = performance.now() - start
+    if (durationMs >= minDurationMs) {
+      log.info(`Process ${name} failed after ${formatDuration(durationMs)} seconds!`)
+    }
     throw e
   }
 }