chore: wait for redis connection before returning the connection inst…

…ance

fixes LATITUDE-LLM-APP-N

apps/web/src/actions/evaluations/runBatch.ts

@@ -95,7 +95,7 @@ export const runBatchEvaluationAction = withDataset
     const evaluations = await evaluationsRepo
       .filterById(input.evaluationIds)
       .then((r) => r.unwrap())
-    const queues = setupJobs()
+    const queues = await setupJobs()
     evaluations.forEach((evaluation) => {
       const batchId = `evaluation:${evaluation.id}:${nanoid(5)}`
       queues.defaultQueue.jobs.enqueueRunBatchEvaluationJob({

apps/workers/src/server.ts

@@ -3,7 +3,7 @@ import http from 'http'
 import { captureException, captureMessage } from './utils/sentry'
 import startWorkers from './workers'
 
-const workers = startWorkers()
+const workers = await startWorkers()
 
 console.log('Workers started')
 

apps/workers/src/workers/index.ts

@@ -14,12 +14,13 @@ const WORKER_OPTS = {
 }
 const WORKERS = [defaultWorker]
 
-export default function startWorkers() {
+export default async function startWorkers() {
+  const connection = await queues()
   return WORKERS.flatMap((w) =>
     w.queues.map((q) => {
       const worker = new Worker(q, w.processor, {
         ...WORKER_OPTS,
-        connection: queues(),
+        connection,
       })
 
       worker.on('error', (error: Error) => {

packages/core/src/cache/index.ts

@@ -5,10 +5,10 @@ import { buildRedisConnection } from '../redis'
 
 let connection: Redis
 
-export const cache = () => {
+export const cache = async () => {
   if (connection) return connection
 
-  connection = buildRedisConnection({
+  connection = await buildRedisConnection({
     host: env.CACHE_HOST,
     port: env.CACHE_PORT,
   })

packages/core/src/events/publisher.ts

@@ -11,8 +11,8 @@ export const publisher = {
       handlers.map((handler) => handler({ data: event })),
     )
   },
-  publishLater: (event: LatitudeEvent) => {
-    const queues = setupJobs()
+  publishLater: async (event: LatitudeEvent) => {
+    const queues = await setupJobs()
 
     queues.eventsQueue.jobs.enqueueCreateEventJob(event)
     queues.eventsQueue.jobs.enqueuePublishEventJob(event)

packages/core/src/queues/index.ts

@@ -5,10 +5,10 @@ import { buildRedisConnection } from '../redis'
 
 let connection: Redis
 
-export const queues = () => {
+export const queues = async () => {
   if (connection) return connection
 
-  connection = buildRedisConnection({
+  connection = await buildRedisConnection({
     host: env.QUEUE_HOST,
     port: env.QUEUE_PORT,
     password: env.QUEUE_PASSWORD,

packages/core/src/redis/index.ts

@@ -5,5 +5,11 @@ export function buildRedisConnection({
   host,
   ...opts
 }: Omit<RedisOptions, 'port' & 'host'> & { host: string; port: number }) {
-  return new Redis(port, host, opts)
+  return new Promise<Redis>((resolve) => {
+    const instance = new Redis(port, host, opts)
+
+    instance.on('connect', () => {
+      resolve(instance)
+    })
+  })
 }

packages/core/src/services/ai/index.ts

@@ -128,9 +128,8 @@ export async function ai({
       )
       providerLogUuid = providerLog.uuid
     } else {
-      await setupJobs().defaultQueue.jobs.enqueueCreateProviderLogJob(
-        payload.data,
-      )
+      const queues = await setupJobs()
+      queues.defaultQueue.jobs.enqueueCreateProviderLogJob(payload.data)
     }
 
     onFinish?.({ ...event, providerLogUuid })
@@ -175,7 +174,7 @@ const checkDefaultProviderUsage = async ({
   workspace: Workspace
 }) => {
   if (provider.token === env.DEFAULT_PROVIDER_API_KEY) {
-    const c = cache()
+    const c = await cache()
     const value = await c.incr(
       `workspace:${workspace.id}:defaultProviderRunCount`,
     )

packages/jobs/src/index.ts

@@ -2,10 +2,11 @@ import { setupQueues } from './queues'
 
 export { Worker } from 'bullmq'
 
-let queues: ReturnType<typeof setupQueues>
-export function setupJobs() {
+let queues: Awaited<ReturnType<typeof setupQueues>>
+
+export async function setupJobs() {
   if (queues) return queues
-  queues = setupQueues()
+  queues = await setupQueues()
 
   return queues
 }

packages/jobs/src/job-definitions/batchEvaluations/runBatchEvaluationJob.ts

@@ -66,15 +66,15 @@ export const runBatchEvaluationJob = async (
     )
   })
 
-  const progressTracker = new ProgressTracker(queues(), batchId)
+  const progressTracker = new ProgressTracker(await queues(), batchId)
   const firstAttempt = job.attemptsMade === 0
 
   if (firstAttempt) {
     await progressTracker.initializeProgress(parameters.length)
   }
 
   const progress = await progressTracker.getProgress()
-  const jobs = setupJobs()
+  const jobs = await setupJobs()
 
   if (firstAttempt && parameters.length > 0) {
     websockets.emit('evaluationStatus', {

packages/jobs/src/job-definitions/batchEvaluations/runDocumentJob.ts

@@ -34,7 +34,7 @@ export const runDocumentJob = async (job: Job<RunDocumentJobData>) => {
     batchId,
   } = job.data
 
-  const progressTracker = new ProgressTracker(queues(), batchId)
+  const progressTracker = new ProgressTracker(await queues(), batchId)
 
   try {
     const workspace = await unsafelyFindWorkspace(workspaceId)
@@ -58,7 +58,7 @@ export const runDocumentJob = async (job: Job<RunDocumentJobData>) => {
 
     await result.response
 
-    const queues = setupJobs()
+    const queues = await setupJobs()
 
     // Enqueue the evaluation job
     await queues.defaultQueue.jobs.enqueueRunEvaluationJob({

packages/jobs/src/job-definitions/batchEvaluations/runEvaluationJob.ts

@@ -29,7 +29,7 @@ export const runEvaluationJob = async (job: Job<RunEvaluationJobData>) => {
     evaluationId,
   } = job.data
 
-  const progressTracker = new ProgressTracker(queues(), batchId)
+  const progressTracker = new ProgressTracker(await queues(), batchId)
   const documentLogsScope = new DocumentLogsRepository(workspaceId)
   const evaluationsScope = new EvaluationsRepository(workspaceId)
   try {

packages/jobs/src/job-definitions/events/publishEventJob.ts

@@ -11,9 +11,8 @@ export const publishEventJob = async (job: Job<LatitudeEvent>) => {
   const handlers = EventHandlers[event.type]
   if (!handlers?.length) return
 
+  const queues = await setupJobs()
   handlers.forEach((handler) => {
-    const queues = setupJobs()
-
     queues.eventsQueue.queue.add(handler.name, event)
   })
 }

packages/jobs/src/queues/index.ts

@@ -1,5 +1,6 @@
 import { queues } from '@latitude-data/core/queues'
 import { Job, JobsOptions, Queue, QueueEvents } from 'bullmq'
+import Redis from 'ioredis'
 
 import { Jobs, Queues, QUEUES } from '../constants'
 import { JobDefinition } from '../job-definitions'
@@ -30,11 +31,12 @@ export const DEFAULT_JOB_OPTIONS: JobsOptions = {
 function setupQueue({
   name,
   jobs,
+  connection,
 }: {
   name: Queues
   jobs: readonly QueueJob[]
+  connection: Redis
 }) {
-  const connection = queues()
   const queue = new Queue(name, {
     connection,
     defaultJobOptions: DEFAULT_JOB_OPTIONS,
@@ -60,14 +62,16 @@ function setupQueue({
 
 type QueueJob = (typeof QUEUES)[keyof typeof QUEUES]['jobs'][number]
 
-export function setupQueues() {
+export async function setupQueues() {
+  const connection = await queues()
+
   return Object.entries(QUEUES).reduce<{
     [K in keyof typeof QUEUES]: ReturnType<typeof setupQueue>
   }>(
     (acc, [name, { jobs }]) => {
       return {
         ...acc,
-        [name]: setupQueue({ name: name as Queues, jobs }),
+        [name]: setupQueue({ name: name as Queues, jobs, connection }),
       }
     },
     {} as { [K in keyof typeof QUEUES]: ReturnType<typeof setupQueue> },