use ilp to push activity updates to questdb instead of pg protocol api (#2633)

Author: themarolt

Diff for: services/apps/data_sink_worker/src/service/activity.service.ts

@@ -244,27 +244,62 @@ export default class ActivityService extends LoggerBase {
             platform: toUpdate.platform || (original.platform as PlatformType),
           })
 
-          await updateActivity(this.qdbStore.connection(), id, {
-            tenantId: tenantId,
-            segmentId: segmentId,
-            type: toUpdate.type || original.type,
-            isContribution: toUpdate.isContribution || original.isContribution,
-            score: toUpdate.score || original.score,
-            sourceId: toUpdate.sourceId || original.sourceId,
-            sourceParentId: toUpdate.sourceParentId || original.sourceParentId,
-            memberId: toUpdate.memberId || original.memberId,
-            username: toUpdate.username || original.username,
-            sentiment: toUpdate.sentiment || original.sentiment,
-            attributes: toUpdate.attributes || original.attributes,
-            body: escapeNullByte(toUpdate.body || original.body),
-            title: escapeNullByte(toUpdate.title || original.title),
-            channel: toUpdate.channel || original.channel,
-            url: toUpdate.url || original.url,
-            organizationId: toUpdate.organizationId || original.organizationId,
-            platform: toUpdate.platform || (original.platform as PlatformType),
-            isBotActivity: memberInfo.isBot,
-            isTeamMemberActivity: memberInfo.isTeamMember,
-          })
+          // use insert instead of update to avoid using pg protocol with questdb
+          try {
+            await insertActivities([
+              {
+                id,
+                memberId: toUpdate.memberId || original.memberId,
+                timestamp: original.timestamp,
+                platform: toUpdate.platform || (original.platform as PlatformType),
+                type: toUpdate.type || original.type,
+                isContribution: toUpdate.isContribution || original.isContribution,
+                score: toUpdate.score || original.score,
+                sourceId: toUpdate.sourceId || original.sourceId,
+                sourceParentId: toUpdate.sourceParentId || original.sourceParentId,
+                tenantId: tenantId,
+                attributes: toUpdate.attributes || original.attributes,
+                sentiment: toUpdate.sentiment || original.sentiment,
+                body: escapeNullByte(toUpdate.body || original.body),
+                title: escapeNullByte(toUpdate.title || original.title),
+                channel: toUpdate.channel || original.channel,
+                url: toUpdate.url || original.url,
+                username: toUpdate.username || original.username,
+                objectMemberId: activity.objectMemberId,
+                objectMemberUsername: activity.objectMemberUsername,
+                segmentId: segmentId,
+                organizationId: toUpdate.organizationId || original.organizationId,
+                isBotActivity: memberInfo.isBot,
+                isTeamMemberActivity: memberInfo.isTeamMember,
+                importHash: original.importHash,
+              },
+            ])
+          } catch (error) {
+            this.log.error('Error updating (by inserting) activity in QuestDB:', error)
+            throw error
+          }
+
+          // await updateActivity(this.qdbStore.connection(), id, {
+          //   tenantId: tenantId,
+          //   segmentId: segmentId,
+          //   type: toUpdate.type || original.type,
+          //   isContribution: toUpdate.isContribution || original.isContribution,
+          //   score: toUpdate.score || original.score,
+          //   sourceId: toUpdate.sourceId || original.sourceId,
+          //   sourceParentId: toUpdate.sourceParentId || original.sourceParentId,
+          //   memberId: toUpdate.memberId || original.memberId,
+          //   username: toUpdate.username || original.username,
+          //   sentiment: toUpdate.sentiment || original.sentiment,
+          //   attributes: toUpdate.attributes || original.attributes,
+          //   body: escapeNullByte(toUpdate.body || original.body),
+          //   title: escapeNullByte(toUpdate.title || original.title),
+          //   channel: toUpdate.channel || original.channel,
+          //   url: toUpdate.url || original.url,
+          //   organizationId: toUpdate.organizationId || original.organizationId,
+          //   platform: toUpdate.platform || (original.platform as PlatformType),
+          //   isBotActivity: memberInfo.isBot,
+          //   isTeamMemberActivity: memberInfo.isTeamMember,
+          // })
 
           return true
         } else {

Diff for: services/libs/data-access-layer/src/old/apps/data_sink_worker/repo/activity.data.ts

@@ -17,6 +17,7 @@ export interface IDbActivity {
   objectMemberId?: string
   objectMemberUsername?: string
   attributes: Record<string, unknown>
+  importHash?: string
   body?: string
   title?: string
   channel?: string

Diff for: services/libs/data-access-layer/src/old/apps/data_sink_worker/repo/activity.repo.ts

@@ -35,6 +35,7 @@ export default class ActivityRepository extends RepositoryBase<ActivityRepositor
             "objectMemberId",
             "objectMemberUsername",
             attributes,
+            "importHash",
             body,
             title,
             channel,
@@ -92,6 +93,7 @@ export default class ActivityRepository extends RepositoryBase<ActivityRepositor
               "objectMemberId",
               "objectMemberUsername",
               attributes,
+              "importHash",
               body,
               title,
               channel,

Diff for: services/libs/database/src/connection.ts

@@ -1,7 +1,7 @@
 import { getServiceChildLogger } from '@crowd/logging'
 import pgPromise from 'pg-promise'
 import { DbConnection, DbInstance, IDatabaseConfig } from './types'
-import { IS_CLOUD_ENV } from '@crowd/common'
+import { IS_CLOUD_ENV, IS_DEV_ENV } from '@crowd/common'
 
 const log = getServiceChildLogger('database.connection')
 
@@ -81,7 +81,7 @@ export const getDbConnection = async (
           rejectUnauthorized: false,
         }
       : false,
-    max: maxPoolSize || 20,
+    max: maxPoolSize || (IS_DEV_ENV ? 5 : 20),
     idleTimeoutMillis: idleTimeoutMillis !== undefined ? idleTimeoutMillis : 10000,
     // query_timeout: 30000,
     application_name: process.env.SERVICE || 'unknown-app',