Insert activities via Kafka Connect

backend/src/database/repositories/activityRepository.ts

@@ -14,6 +14,8 @@ import { findManyLfxMemberships } from '@crowd/data-access-layer/src/lfx_members
 import { ActivityDisplayService } from '@crowd/integrations'
 import { IIntegrationResult, IntegrationResultState } from '@crowd/types'
 
+import { QUEUE_CLIENT } from '@/serverless/utils/queueService'
+
 import { AttributeData } from '../attributes/attribute'
 import SequelizeFilterUtils from '../utils/sequelizeFilterUtils'
 
@@ -58,7 +60,7 @@ class ActivityRepository {
       data.platform = data.platform.toLowerCase()
     }
 
-    const ids = await insertActivities([
+    const ids = await insertActivities(QUEUE_CLIENT(), [
       {
         type: data.type,
         timestamp: data.timestamp,

backend/src/services/activityService.ts

@@ -23,7 +23,7 @@ import { IMemberIdentity, IntegrationResultType, PlatformType, SegmentData } fro
 
 import { IRepositoryOptions } from '@/database/repositories/IRepositoryOptions'
 import OrganizationRepository from '@/database/repositories/organizationRepository'
-import { getDataSinkWorkerEmitter } from '@/serverless/utils/queueService'
+import { QUEUE_CLIENT, getDataSinkWorkerEmitter } from '@/serverless/utils/queueService'
 
 import { GITHUB_CONFIG, IS_DEV_ENV, IS_TEST_ENV } from '../conf'
 import ActivityRepository from '../database/repositories/activityRepository'
@@ -174,7 +174,7 @@ export default class ActivityService extends LoggerBase {
         )
 
         record = await ActivityRepository.create(data, repositoryOptions)
-        await insertActivities([{ ...data, id: record.id }], true)
+        await insertActivities(QUEUE_CLIENT(), [{ ...data, id: record.id }], true)
 
         // Only track activity's platform and timestamp and memberId. It is completely annonymous.
         telemetryTrack(

scripts/scaffold.yaml

@@ -161,6 +161,23 @@ services:
     networks:
       - crowd-bridge
 
+  kafka-connect:
+    build:
+      context: scaffold/kafka-connect
+    restart: unless-stopped
+    entrypoint:
+      - connect-standalone
+      - /etc/kafka-connect/worker-local.properties
+      - /etc/kafka-connect/console-local-sink.properties
+      - /etc/kafka-connect/questdb-local-sink.properties
+    volumes:
+      - kafka-connect-dev:/storage
+      - ./scaffold/kafka-connect/worker-local.properties:/etc/kafka-connect/worker-local.properties
+      - ./scaffold/kafka-connect/console-local-sink.properties:/etc/kafka-connect/console-local-sink.properties
+      - ./scaffold/kafka-connect/questdb-local-sink.properties:/etc/kafka-connect/questdb-local-sink.properties
+    networks:
+      - crowd-bridge
+
   temporal:
     build:
       context: scaffold/temporal
@@ -182,3 +199,4 @@ volumes:
   opensearch-dev:
   s3-dev:
   redis-dev:
+  kafka-connect-dev:

scripts/scaffold/kafka-connect/Dockerfile

@@ -0,0 +1,13 @@
+FROM confluentinc/cp-kafka-connect:7.8.0-2-ubi8
+
+USER root
+
+RUN yum install -y jq findutils unzip
+
+RUN confluent-hub install snowflakeinc/snowflake-kafka-connector:2.5.0 --no-prompt
+RUN confluent-hub install questdb/kafka-questdb-connector:0.12 --no-prompt
+
+VOLUME /storage
+
+USER appuser
+

scripts/scaffold/kafka-connect/build-docker-image.sh

@@ -0,0 +1,21 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+TAG="sjc.ocir.io/axbydjxa5zuh/kafka-connect:$(date +%s)"
+readonly TAG
+
+docker build -t "${TAG}" .
+
+echo "----------------------------------------"
+echo "Image built with tag: ${TAG}"
+echo "----------------------------------------"
+echo -n "Type 'y' and press Enter to push the image to the registry. Ctrl+C to cancel: "
+read -r PUSH
+if [ "${PUSH}" = "y" ]; then
+    echo "Pushing image to the registry..."
+    echo "----------------------------------------"
+    docker push "${TAG}"
+else
+    echo "Skipping push"
+fi

scripts/scaffold/kafka-connect/console-local-sink.properties

@@ -0,0 +1,6 @@
+name=console-sink
+connector.class=FileStreamSinkConnector
+tasks.max=1
+topics=activities
+value.converter=org.apache.kafka.connect.json.JsonConverter
+value.converter.schemas.enable=false

scripts/scaffold/kafka-connect/questdb-local-sink.properties

@@ -0,0 +1,12 @@
+name=questdb-sink
+client.conf.string=http::addr=questdb:9000;
+topics=activities
+table=activities
+connector.class=io.questdb.kafka.QuestDBSinkConnector
+value.converter=org.apache.kafka.connect.json.JsonConverter
+include.key=false
+key.converter=org.apache.kafka.connect.storage.StringConverter
+timestamp.field.name=timestamp
+timestamp.string.fields=createdAt,updatedAt
+timestamp.string.format=yyyy-MM-ddTHH:mm:ss.SSSZ
+value.converter.schemas.enable=false

scripts/scaffold/kafka-connect/worker-local.properties

@@ -0,0 +1,13 @@
+bootstrap.servers=kafka:9092
+group.id=kafka-connect
+
+key.converter=org.apache.kafka.connect.storage.StringConverter
+value.converter=org.apache.kafka.connect.json.JsonConverter
+internal.key.converter=org.apache.kafka.connect.storage.StringConverter
+internal.value.converter=org.apache.kafka.connect.json.JsonConverter
+key.converter.schemas.enable=true
+value.converter.schemas.enable=true
+
+offset.storage.file.filename=/storage/connect.offsets
+offset.flush.interval.ms=10000
+plugin.path=/usr/share/java,/usr/share/filestream-connectors,/usr/share/confluent-hub-components

services/apps/activities_worker/src/activities/createConversations.ts

@@ -182,7 +182,7 @@ export async function createConversations(): Promise<ICreateConversationsResult>
   if (toUpdate.length > 0) {
     for (const batch of partition(toUpdate, 100)) {
       try {
-        const results = await insertActivities(batch, true)
+        const results = await insertActivities(svc.queue, batch, true)
         activitiesAddedToConversations += results.length
       } catch (err) {
         svc.log.error(err, 'Error linking activities to conversations')
@@ -209,7 +209,7 @@ async function getRows(qdbConn: DbConnOrTx, current: Date): Promise<any[]> {
   activities_to_check_for_parentId AS (
     SELECT *
     FROM activities child
-    WHERE deletedAt IS NULL 
+    WHERE deletedAt IS NULL
     AND sourceParentId IS NOT NULL
     AND conversationId IS NULL
     AND timestamp > dateadd('w', -1, $(limit))
@@ -236,7 +236,7 @@ async function getMinActivityTimestamp(qdbConn: DbConnOrTx): Promise<string | nu
   const query = `
   select first(timestamp) as minTimestamp
   from activities
-  where 
+  where
     deletedAt is null and
     sourceParentId is not null and
     conversationId is null;

services/apps/activities_worker/src/main.ts

@@ -26,6 +26,9 @@ const options: Options = {
   opensearch: {
     enabled: false,
   },
+  queue: {
+    enabled: true,
+  },
 }
 
 export const svc = new ServiceWorker(config, options)

services/apps/data_sink_worker/src/bin/process-results.ts

@@ -57,6 +57,7 @@ setImmediate(async () => {
     dataSinkWorkerEmitter,
     redis,
     temporal,
+    queueClient,
     log,
   )
 

services/apps/data_sink_worker/src/queue/index.ts

@@ -17,7 +17,7 @@ import DataSinkService from '../service/dataSink.service'
 export class WorkerQueueReceiver extends PrioritizedQueueReciever {
   constructor(
     level: QueuePriorityLevel,
-    client: IQueue,
+    private readonly client: IQueue,
     private readonly pgConn: DbConnection,
     private readonly qdbConn: DbConnection,
     private readonly searchSyncWorkerEmitter: SearchSyncWorkerEmitter,
@@ -49,6 +49,7 @@ export class WorkerQueueReceiver extends PrioritizedQueueReciever {
         this.dataSinkWorkerEmitter,
         this.redisClient,
         this.temporal,
+        this.client,
         this.log,
       )
 

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

@@ -34,6 +34,7 @@ import SettingsRepository from '@crowd/data-access-layer/src/old/apps/data_sink_
 import { DEFAULT_ACTIVITY_TYPE_SETTINGS, GithubActivityType } from '@crowd/integrations'
 import { GitActivityType } from '@crowd/integrations/src/integrations/git/types'
 import { Logger, LoggerBase, getChildLogger } from '@crowd/logging'
+import { IQueue } from '@crowd/queue'
 import { RedisClient } from '@crowd/redis'
 import { Client as TemporalClient } from '@crowd/temporal'
 import {
@@ -60,6 +61,7 @@ export default class ActivityService extends LoggerBase {
     private readonly searchSyncWorkerEmitter: SearchSyncWorkerEmitter,
     private readonly redisClient: RedisClient,
     private readonly temporal: TemporalClient,
+    private readonly client: IQueue,
     parentLog: Logger,
   ) {
     super(parentLog)
@@ -102,7 +104,7 @@ export default class ActivityService extends LoggerBase {
 
         this.log.debug('Creating an activity in QuestDB!')
         try {
-          await insertActivities([
+          await insertActivities(this.client, [
             {
               id: activity.id,
               timestamp: activity.timestamp.toISOString(),
@@ -185,7 +187,7 @@ export default class ActivityService extends LoggerBase {
 
           // use insert instead of update to avoid using pg protocol with questdb
           try {
-            await insertActivities([
+            await insertActivities(this.client, [
               {
                 id,
                 memberId: toUpdate.memberId || original.memberId,
@@ -592,6 +594,7 @@ export default class ActivityService extends LoggerBase {
           this.searchSyncWorkerEmitter,
           this.redisClient,
           this.temporal,
+          this.client,
           this.log,
         )
         const txIntegrationRepo = new IntegrationRepository(txStore, this.log)
@@ -1303,6 +1306,7 @@ export default class ActivityService extends LoggerBase {
     }) => {
       await updateActivities(
         this.qdbStore.connection(),
+        this.client,
         async (activity) => ({
           attributes: {
             ...gitAttributes,
@@ -1352,6 +1356,7 @@ export default class ActivityService extends LoggerBase {
 
     await updateActivities(
       this.qdbStore.connection(),
+      this.client,
       async () => ({
         sourceParentId: activity.sourceId,
       }),

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

@@ -7,6 +7,7 @@ import {
 } from '@crowd/data-access-layer/src/old/apps/data_sink_worker/repo/dataSink.data'
 import DataSinkRepository from '@crowd/data-access-layer/src/old/apps/data_sink_worker/repo/dataSink.repo'
 import { Logger, LoggerBase, getChildLogger } from '@crowd/logging'
+import { IQueue } from '@crowd/queue'
 import { RedisClient } from '@crowd/redis'
 import telemetry from '@crowd/telemetry'
 import { Client as TemporalClient } from '@crowd/temporal'
@@ -36,6 +37,7 @@ export default class DataSinkService extends LoggerBase {
     private readonly dataSinkWorkerEmitter: DataSinkWorkerEmitter,
     private readonly redisClient: RedisClient,
     private readonly temporal: TemporalClient,
+    private readonly client: IQueue,
     parentLog: Logger,
   ) {
     super(parentLog)
@@ -218,6 +220,7 @@ export default class DataSinkService extends LoggerBase {
                 this.searchSyncWorkerEmitter,
                 this.redisClient,
                 this.temporal,
+                this.client,
                 this.log,
               )
               const activityData = data.data as IActivityData

services/apps/entity_merging_worker/src/activities/members.ts

@@ -38,7 +38,7 @@ export async function moveActivitiesBetweenMembers(
   if (!memberExists) {
     return
   }
-  await moveActivitiesToNewMember(svc.questdbSQL, primaryId, secondaryId, tenantId)
+  await moveActivitiesToNewMember(svc.questdbSQL, svc.queue, primaryId, secondaryId, tenantId)
 }
 
 export async function moveActivitiesWithIdentityToAnotherMember(
@@ -67,6 +67,7 @@ export async function moveActivitiesWithIdentityToAnotherMember(
   )) {
     await moveIdentityActivitiesToNewMember(
       svc.questdbSQL,
+      svc.queue,
       tenantId,
       fromId,
       toId,

services/apps/entity_merging_worker/src/activities/organizations.ts

@@ -29,7 +29,7 @@ export async function moveActivitiesBetweenOrgs(
   secondaryId: string,
   tenantId: string,
 ): Promise<void> {
-  await moveActivitiesToNewOrg(svc.questdbSQL, primaryId, secondaryId, tenantId)
+  await moveActivitiesToNewOrg(svc.questdbSQL, svc.queue, primaryId, secondaryId, tenantId)
 }
 
 export async function recalculateActivityAffiliationsOfOrganizationSynchronous(