Kafka indexing (#60)

* Add Functionality To Enqueue Indexable Documents To Kafka For Processing

* Clean Up Scheduling of Indexing Jobs

* Send Batch Documents To Kafka Queue Service

packages/apps/asana/src/strategies/asana.ts

@@ -22,9 +22,7 @@ export default class AsanaStrategy extends AbstractBatchJobStrategy {
     const batchJob = await this.batchJobService_.retrieve(batchJobId)
     const stream = await this.asanaService_.getAsanaData(batchJob.context?.org as Organisation)
     stream.on('data', (documents) => {
-      for (const document of documents) {
-        this.queueService_.send(SEARCH_INDEXING_TOPIC, document)
-      }
+      this.queueService_.sendBatch(SEARCH_INDEXING_TOPIC, documents)
     });
     stream.on('end', () => {
       console.log('No more data');

packages/apps/confluence/src/strategies/confluence.ts

@@ -23,9 +23,7 @@ export default class ConfluenceStrategy extends AbstractBatchJobStrategy {
       batchJob.context?.org as Organisation
     );
     stream.on('data', (documents) => {
-      for (const document of documents) {
-        this.queueService_.send(SEARCH_INDEXING_TOPIC, document)
-      }
+      this.queueService_.sendBatch(SEARCH_INDEXING_TOPIC, documents)
     });
     stream.on('end', () => {
       console.log('No more data');

packages/apps/github/src/strategies/github.ts

@@ -27,10 +27,8 @@ class GithubStrategy extends AbstractBatchJobStrategy {
 
     const stream = await this.githubService_.getRepositoriesOcular(batchJob.context?.org as Organisation)
 
-    stream.on('data', (documents) => {
-      for (const document of documents) {
-        this.queueService_.send(SEARCH_INDEXING_TOPIC, document)
-      }
+   stream.on('data', (documents) => {
+      this.queueService_.sendBatch(SEARCH_INDEXING_TOPIC, documents)
     });
 
 

packages/apps/gmail/src/strategies/gmail.ts

@@ -24,9 +24,7 @@ class GmailStrategy extends AbstractBatchJobStrategy {
     const batchJob = await this.batchJobService_.retrieve(batchJobId)
     const stream = await this.gmailService_.getGmailData(batchJob.context?.org as Organisation)
     stream.on('data', (documents) => {
-      for (const document of documents) {
-        this.queueService_.send(SEARCH_INDEXING_TOPIC, document)
-      }
+      this.queueService_.sendBatch(SEARCH_INDEXING_TOPIC, documents)
     });
     stream.on('end', () => {
       console.log('No more data');

packages/apps/google-drive/src/strategies/google-drive.ts

@@ -11,24 +11,20 @@ class GoogleDriveStrategy extends AbstractBatchJobStrategy {
   static batchType = "google-drive"
   protected batchJobService_: BatchJobService
   protected googleDriveService_: GoogleDriveService
-  protected eventBusService_: EventBusService
   protected queueService_: QueueService
 
   constructor(container) {
     super(arguments[0])
     this.googleDriveService_ = container.googleDriveService
     this.batchJobService_ = container.batchJobService
-    this.eventBusService_ = container.eventBusService
     this.queueService_ = container.queueService
   }
 
   async processJob(batchJobId: string): Promise<void> {
     const batchJob = await this.batchJobService_.retrieve(batchJobId)
     const stream = await this.googleDriveService_.getGoogleDriveData(batchJob.context?.org as Organisation)
     stream.on('data', (documents) => {
-      for (const document of documents) {
-        this.queueService_.send(SEARCH_INDEXING_TOPIC, document)
-      }
+      this.queueService_.sendBatch(SEARCH_INDEXING_TOPIC, documents)
     });
     stream.on('end', () => {
       console.log('No more data');

packages/apps/jira/src/strategies/jira.ts

@@ -23,10 +23,8 @@ export default class JiraStrategy extends AbstractBatchJobStrategy {
       // Confluenec method need to be implemmented
       batchJob.context?.org as Organisation
     );
-    stream.on('data', (documents) => {
-      for (const document of documents) {
-        this.queueService_.send(SEARCH_INDEXING_TOPIC, document)
-      }
+   stream.on('data', (documents) => {
+      this.queueService_.sendBatch(SEARCH_INDEXING_TOPIC, documents)
     });
     stream.on('end', () => {
       console.log('No more data');

packages/apps/notion/src/strategies/notion.ts

@@ -22,10 +22,8 @@ export default class AsanaStrategy extends AbstractBatchJobStrategy {
     const stream = await this.asanaService_.getNotionPagesData(
       batchJob.context?.org as Organisation
     );
-    stream.on('data', (documents) => {
-      for (const document of documents) {
-        this.queueService_.send(SEARCH_INDEXING_TOPIC, document)
-      }
+   stream.on('data', (documents) => {
+      this.queueService_.sendBatch(SEARCH_INDEXING_TOPIC, documents)
     });
     stream.on('end', () => {
       console.log('No more data');

packages/apps/slack/src/strategies/slack.ts

@@ -23,9 +23,7 @@ export default class SlackStrategy extends AbstractBatchJobStrategy {
       batchJob.context?.org as Organisation
     );
     stream.on('data', (documents) => {
-      for (const document of documents) {
-        this.queueService_.send(SEARCH_INDEXING_TOPIC, document)
-      }
+      this.queueService_.sendBatch(SEARCH_INDEXING_TOPIC, documents)
     });
     stream.on('end', () => {
       console.log('No more data');

packages/ocular/src/services/__tests__/queue.ts

@@ -31,6 +31,7 @@ describe('queueService', () => {
         clientId: 'ocular',
         brokers: ['localhost:9092'],
       })
+
     const moduleDeps = {
       logger: loggerMock,
       eventBusRedisConnection: {},
@@ -51,4 +52,15 @@ describe('queueService', () => {
       done()
     }, {groupId: "ocular-group"});
   });
-})
+
+
+  it('should send batch messages to a topic', (done) => {
+    queueService.sendBatch("ocular", [{message: "Hello World 1"}, {message: "Hello World 2"}, {message: "Hello World 3"}, {message: "Hello World 4"}]);
+    queueService.subscribe("ocular", async (message, topic) => {
+      console.log("Message Received", message)
+      expect(message).toEqual("expectedMessage");
+      done()
+    }, {groupId: "ocular-group"});
+  });
+}
+)

packages/ocular/src/services/queue.ts

@@ -50,6 +50,31 @@ export default class QueueService extends AbstractQueueService {
     }
   }
 
+  async sendBatch<T>(
+    topicName: string,
+    data: T[],
+    options?: Record<string, unknown>
+  ): Promise<void> {
+    try {
+      await this.producer_.connect()
+      const messages = data.map((doc) => {
+        return { value: JSON.stringify(doc) }
+      })
+      const topicMessages: TopicMessages = {
+        topic: topicName,
+        messages: messages
+      }
+      const batch: ProducerBatch = {
+        topicMessages: [topicMessages]
+      }
+      await this.producer_.sendBatch(batch)
+      await this.producer_.disconnect()
+    } catch (error) {
+      this.logger_.error(`Error sending batch message to Kafka: ${error.message}`)
+      throw error
+    }
+  }
+
   async subscribe<T>(topicName: string, consumer: Consumer, context: ConsumerContext ): Promise<void> {
     // Check if the consumer is a function
     if (typeof consumer !== `function`) {

packages/types/src/interfaces/queue.ts

@@ -8,6 +8,11 @@ export interface IQueueService {
     data: T,
     options?: Record<string, unknown>
   ): Promise<void>
+  sendBatch<T>(
+    topicName: string,
+    data: T[],
+    options?: Record<string, unknown>
+  ) : Promise<void>
   subscribe<T>(topicName: string, consumer: Consumer, context:ConsumerContext): Promise<void>
 }
 
@@ -59,6 +64,12 @@ export abstract class AbstractQueueService implements IQueueService {
     options?: Record<string, unknown>
   ): Promise<void>
 
+  abstract sendBatch<T>(
+    topicName: string,
+    data: T[],
+    options?: Record<string, unknown>
+  ): Promise<void>
+
   abstract subscribe<T>(topicName: string, consumer:Consumer, context:ConsumerContext): Promise<void>
 }