minor: move latte to its own worker and queue (#1714)

Author: geclos

apps/workers/src/server.ts

@@ -94,7 +94,7 @@ const gracefulShutdown = async (signal: string) => {
 process.on('SIGINT', () => gracefulShutdown('SIGINT'))
 process.on('SIGTERM', () => gracefulShutdown('SIGTERM'))
 
-process.on('uncaughtException', function(err) {
+process.on('uncaughtException', function (err) {
   captureException(err)
 })
 

apps/workers/src/workers/index.ts

@@ -6,6 +6,7 @@ import {
   startEventHandlersWorker,
   startEventsWorker,
 } from './worker-definitions/eventsWorker'
+import { startLatteWorker } from './worker-definitions/latteWorker'
 import { startMaintenanceWorker } from './worker-definitions/maintenanceWorker'
 import { startTracingWorker } from './worker-definitions/tracingWorker'
 import { startWebhooksWorker } from './worker-definitions/webhooksWorker'
@@ -15,6 +16,7 @@ export async function startWorkers() {
   const evaluationsWorker = startEvaluationsWorker()
   const eventsWorker = startEventsWorker()
   const eventHandlersWorker = startEventHandlersWorker()
+  const latteWorker = startLatteWorker()
   const maintenanceWorker = startMaintenanceWorker()
   const webhooksWorker = startWebhooksWorker()
   const documentsWorker = startDocumentsWorker()
@@ -26,6 +28,7 @@ export async function startWorkers() {
     evaluationsWorker,
     eventsWorker,
     eventHandlersWorker,
+    latteWorker,
     maintenanceWorker,
     webhooksWorker,
     documentsWorker,

apps/workers/src/workers/worker-definitions/documentsWorker.ts

@@ -6,7 +6,6 @@ import { WORKER_CONNECTION_CONFIG } from '../utils/connectionConfig'
 const jobMappings = {
   runDocumentJob: jobs.runDocumentJob,
   runDocumentForExperimentJob: jobs.runDocumentForExperimentJob,
-  runLatteJob: jobs.runLatteJob,
   runDocumentTriggerEventJob: jobs.runDocumentTriggerEventJob,
 }
 

apps/workers/src/workers/worker-definitions/latteWorker.ts

@@ -0,0 +1,15 @@
+import { Queues } from '@latitude-data/core/queues/types'
+import * as jobs from '@latitude-data/core/jobs/definitions'
+import { createWorker } from '../utils/createWorker'
+import { WORKER_CONNECTION_CONFIG } from '../utils/connectionConfig'
+
+const jobMappings = {
+  runLatteJob: jobs.runLatteJob,
+}
+
+export function startLatteWorker() {
+  return createWorker(Queues.latteQueue, jobMappings, {
+    concurrency: 25,
+    connection: WORKER_CONNECTION_CONFIG,
+  })
+}

apps/workers/src/workers/worker-definitions/maintenanceWorker.ts

@@ -19,7 +19,8 @@ const jobMappings = {
 
   // Migrate provider logs to object storage
   scheduleProviderLogsMigrationJobs: jobs.scheduleProviderLogsMigrationJobs,
-  migrateProviderLogsToObjectStorageJob: jobs.migrateProviderLogsToObjectStorageJob,
+  migrateProviderLogsToObjectStorageJob:
+    jobs.migrateProviderLogsToObjectStorageJob,
 
   // Migrate document logs workspace ids
   scheduleWorkspaceLogsMigrationJobs: jobs.scheduleWorkspaceLogsMigrationJobs,

packages/core/src/jobs/job-definitions/copilot/chat.ts

@@ -118,6 +118,7 @@ export const runLatteJob = async (job: Job<RunLatteJobData>) => {
         abortSignal: controller.signal,
         debugVersionUuid,
       })
+
       if (!runResult.ok) {
         await emitError({
           workspaceId,

packages/core/src/jobs/queues/index.ts

@@ -14,6 +14,7 @@ let _queues:
       maintenanceQueue: Queue
       tracingQueue: Queue
       webhooksQueue: Queue
+      latteQueue: Queue
     }
   | undefined
 
@@ -33,8 +34,8 @@ export async function queues() {
         type: 'exponential',
         delay: 1000,
       },
-      removeOnFail: true,
-      removeOnComplete: true,
+      removeOnFail: 100,
+      removeOnComplete: 100,
     },
   }
 
@@ -48,6 +49,7 @@ export async function queues() {
     maintenanceQueue: new Queue(Queues.maintenanceQueue, options),
     tracingQueue: new Queue(Queues.tracingQueue, options),
     webhooksQueue: new Queue(Queues.webhooksQueue, options),
+    latteQueue: new Queue(Queues.latteQueue, options),
   }
 
   return _queues

packages/core/src/jobs/queues/types.ts

@@ -8,4 +8,5 @@ export enum Queues {
   documentsQueue = 'documentsQueue',
   documentSuggestionsQueue = 'documentSuggestionsQueue',
   tracingQueue = 'tracing',
+  latteQueue = 'latteQueue',
 }

packages/core/src/repositories/documentLogsRepository/index.ts

@@ -31,10 +31,7 @@ export class DocumentLogsRepository extends Repository<DocumentLog> {
         commits,
         and(isNull(commits.deletedAt), eq(commits.id, documentLogs.commitId)),
       )
-      .innerJoin(
-        projects,
-        eq(projects.id, commits.projectId),
-      )
+      .innerJoin(projects, eq(projects.id, commits.projectId))
       .leftJoin(
         runErrors,
         and(
@@ -51,17 +48,23 @@ export class DocumentLogsRepository extends Repository<DocumentLog> {
       return Result.error(new NotFoundError('DocumentLog not found'))
     }
 
-    const result = await this.scope.where(
-      and(this.scopeFilter, eq(documentLogs.uuid, uuid)),
-    )
-
-    if (!result.length) {
-      return Result.error(
-        new NotFoundError(`DocumentLog not found with uuid ${uuid}`),
+    try {
+      const result = await this.scope.where(
+        and(this.scopeFilter, eq(documentLogs.uuid, uuid)),
       )
-    }
 
-    return Result.ok(result[0]!)
+      if (!result.length) {
+        return Result.error(
+          new NotFoundError(`DocumentLog not found with uuid ${uuid}`),
+        )
+      }
+
+      return Result.ok(result[0]!)
+    } catch (err) {
+      console.log(err)
+
+      return Result.error(err as Error)
+    }
   }
 
   async totalCountSinceDate(minDate: Date) {

packages/core/src/schema/models/documentLogs.ts

@@ -21,7 +21,10 @@ export const documentLogs = latitudeSchema.table(
   {
     id: bigserial('id', { mode: 'number' }).notNull().primaryKey(),
     uuid: uuid('uuid').notNull().unique(),
-    workspaceId: bigint('workspace_id', { mode: 'number' }).references(() => workspaces.id, { onDelete: 'cascade' }),
+    workspaceId: bigint('workspace_id', { mode: 'number' }).references(
+      () => workspaces.id,
+      { onDelete: 'cascade' },
+    ),
     documentUuid: uuid('document_uuid').notNull(),
     commitId: bigint('commit_id', { mode: 'number' })
       .notNull()