Feature/sentry in gateway and workers (#207)

* chore: add sentry to workers

* chore: added sentry to gateway

apps/gateway/package.json

@@ -22,6 +22,7 @@
     "@latitude-data/env": "workspace:^",
     "@latitude-data/jobs": "workspace:^",
     "@latitude-data/mailers": "workspace:^",
+    "@sentry/node": "^8.30.0",
     "@t3-oss/env-core": "^0.10.1",
     "drizzle-orm": "^0.33.0",
     "hono": "^4.5.3",

apps/gateway/src/common/sentry.ts

@@ -0,0 +1,24 @@
+import { env } from '@latitude-data/env'
+import * as Sentry from '@sentry/node'
+
+Sentry.init({
+  dsn: env.SENTRY_DSN,
+
+  tracesSampleRate: 1.0,
+})
+
+export const captureException = (error: Error) => {
+  if (env.NODE_ENV === 'production') {
+    Sentry.captureException(error)
+  } else {
+    console.error(error)
+  }
+}
+
+export const captureMessage = (message: string) => {
+  if (env.NODE_ENV === 'production') {
+    Sentry.captureMessage(message)
+  } else {
+    console.log(message)
+  }
+}

apps/gateway/src/middlewares/errorHandler.ts

@@ -2,6 +2,7 @@ import {
   LatitudeError,
   UnprocessableEntityError,
 } from '@latitude-data/core/lib/errors'
+import { captureException } from '$/common/sentry'
 import { createMiddleware } from 'hono/factory'
 
 import HttpStatusCodes from '../common/httpStatusCodes'
@@ -12,7 +13,7 @@ const errorHandlerMiddleware = () =>
     if (!err) return next()
 
     if (process.env.NODE_ENV !== 'test') {
-      console.error(err.message)
+      captureException(err)
     }
 
     if (err instanceof UnprocessableEntityError) {

apps/gateway/src/server.ts

@@ -14,13 +14,10 @@ serve(
   },
 )
 
-// Add graceful shutdown handler
 function gracefulShutdown() {
   console.log('Received termination signal. Shutting down gracefully...')
-  // Perform any cleanup operations here
   process.exit(0)
 }
 
-// Register signal handlers
 process.on('SIGTERM', gracefulShutdown)
 process.on('SIGINT', gracefulShutdown)

apps/workers/package.json

@@ -18,6 +18,7 @@
     "@latitude-data/env": "workspace:^",
     "@latitude-data/jobs": "workspace:^",
     "@latitude-data/mailers": "workspace:^",
+    "@sentry/node": "^8.30.0",
     "@t3-oss/env-core": "*",
     "zod": "^3.23.8"
   },

apps/workers/src/server.ts

@@ -1,8 +1,9 @@
 import http from 'http'
 
-import { setupWorkers } from '@latitude-data/jobs'
+import { captureException, captureMessage } from './utils/sentry'
+import startWorkers from './workers'
 
-const workers = setupWorkers()
+const workers = startWorkers()
 
 console.log('Workers started')
 
@@ -33,13 +34,13 @@ process.on('SIGINT', () => gracefulShutdown('SIGINT'))
 process.on('SIGTERM', () => gracefulShutdown('SIGTERM'))
 
 process.on('uncaughtException', function (err) {
-  // TODO: Sentry.captureException(err)
+  captureException(err)
 
   console.error(err, 'Uncaught exception')
 })
 
-process.on('unhandledRejection', (reason, promise) => {
-  // TODO: Sentry.captureException(reason)
+process.on('unhandledRejection', (reason: string, promise) => {
+  captureMessage(reason)
 
   console.error({ promise, reason }, 'Unhandled Rejection at: Promise')
 })

apps/workers/src/utils/sentry.ts

@@ -0,0 +1,24 @@
+import { env } from '@latitude-data/env'
+import * as Sentry from '@sentry/node'
+
+Sentry.init({
+  dsn: env.SENTRY_DSN,
+
+  tracesSampleRate: 1.0,
+})
+
+export const captureException = (error: Error) => {
+  if (env.NODE_ENV === 'production') {
+    Sentry.captureException(error)
+  } else {
+    console.error(error)
+  }
+}
+
+export const captureMessage = (message: string) => {
+  if (env.NODE_ENV === 'production') {
+    Sentry.captureMessage(message)
+  } else {
+    console.log(message)
+  }
+}

packages/jobs/src/workers/_shared.ts → apps/workers/src/workers/_shared.ts

@@ -1,8 +1,7 @@
+import { Queues, QUEUES } from '@latitude-data/jobs/constants'
+import { captureException } from '$/utils/sentry'
 import { Processor } from 'bullmq'
 
-import { Queues } from '../constants'
-import { QUEUES } from '../queues'
-
 export const buildProcessor =
   (queues: Queues[]): Processor =>
   async (job) => {
@@ -13,10 +12,10 @@ export const buildProcessor =
             if (j.name === job.name) {
               try {
                 await j(job)
-              } catch (err) {
-                console.error(err)
+              } catch (error) {
+                captureException(error as Error)
 
-                throw err
+                throw error
               }
             }
           }),

packages/jobs/src/workers/index.ts → apps/workers/src/workers/index.ts

@@ -1,6 +1,7 @@
+import { queues } from '@latitude-data/core/queues'
+import { captureException } from '$/utils/sentry'
 import { Worker } from 'bullmq'
 
-import { connection } from '../utils/connection'
 import { defaultWorker } from './worker-definitions/defaultWorker'
 
 const WORKER_OPTS = {
@@ -18,11 +19,11 @@ export default function startWorkers() {
     w.queues.map((q) => {
       const worker = new Worker(q, w.processor, {
         ...WORKER_OPTS,
-        connection,
+        connection: queues(),
       })
 
       worker.on('error', (error: Error) => {
-        console.error(error)
+        captureException(error)
       })
 
       return worker

packages/jobs/src/workers/worker-definitions/defaultWorker.ts → apps/workers/src/workers/worker-definitions/defaultWorker.ts

@@ -1,10 +1,11 @@
+import { Queues } from '@latitude-data/jobs/constants'
+
 import { buildProcessor } from '../_shared'
-import { Queues } from '../../constants'
 
 const defaultWorkerQueues = [
   Queues.defaultQueue,
-  Queues.eventsQueue,
   Queues.eventHandlersQueue,
+  Queues.eventsQueue,
 ]
 
 export const defaultWorker = {

packages/core/package.json

@@ -5,17 +5,19 @@
   "type": "module",
   "exports": {
     "./browser": "./src/browser.ts",
-    "./data-access": "./src/data-access/index.ts",
+    "./cache": "./src/cache/index.ts",
     "./client": "./src/client/index.ts",
+    "./data-access": "./src/data-access/index.ts",
     "./events/*": "./src/events/*.ts",
     "./lib/*": "./src/lib/*.ts",
+    "./queues": "./src/queues/index.ts",
     "./redis": "./src/redis/index.ts",
-    "./websockets/*": "./src/websockets/*.ts",
+    "./repositories": "./src/repositories/index.ts",
+    "./repositories/*": "./src/repositories/*.ts",
     "./schema": "./src/schema/index.ts",
     "./schema/*": "./src/schema/*.ts",
     "./services/*": "./src/services/*.ts",
-    "./repositories": "./src/repositories/index.ts",
-    "./repositories/*": "./src/repositories/*.ts",
+    "./websockets/*": "./src/websockets/*.ts",
     "./factories": "./src/tests/factories/index.ts",
     "./test": "./src/tests/useTestDatabase.ts"
   },

packages/core/src/queues/index.ts

@@ -0,0 +1,20 @@
+import { env } from '@latitude-data/env'
+import Redis from 'ioredis'
+
+import { buildRedisConnection } from '../redis'
+
+let connection: Redis
+
+export const queues = () => {
+  if (connection) return connection
+
+  connection = buildRedisConnection({
+    host: env.QUEUE_HOST,
+    port: env.QUEUE_PORT,
+    password: env.QUEUE_PASSWORD,
+    enableOfflineQueue: false,
+    maxRetriesPerRequest: null,
+  })
+
+  return connection
+}

packages/jobs/package.json

@@ -2,7 +2,10 @@
   "name": "@latitude-data/jobs",
   "version": "0.0.1",
   "description": "Latitude jobs from Latitude llm",
-  "main": "./src/index.ts",
+  "exports": {
+    ".": "./src/index.ts",
+    "./constants": "./src/constants.ts"
+  },
   "type": "module",
   "scripts": {
     "lint": "eslint src",

packages/jobs/src/constants.ts

@@ -1,3 +1,12 @@
+import { createProviderLogJob } from '@latitude-data/core/events/handlers/createProviderLogJob'
+import { EventHandlers } from '@latitude-data/core/events/handlers/index'
+
+import { runBatchEvaluationJob } from './job-definitions/batchEvaluations/runBatchEvaluationJob'
+import { runDocumentJob } from './job-definitions/batchEvaluations/runDocumentJob'
+import { runEvaluationJob } from './job-definitions/batchEvaluations/runEvaluationJob'
+import { createEventJob } from './job-definitions/events/createEventJob'
+import { publishEventJob } from './job-definitions/events/publishEventJob'
+
 // TODO: Review if we can remove this declarations
 export enum Queues {
   defaultQueue = 'defaultQueue',
@@ -12,3 +21,23 @@ export enum Jobs {
   createEventJob = 'createEventJob',
   publishEventJob = 'publishEventJob',
 }
+
+export const QUEUES = {
+  [Queues.defaultQueue]: {
+    name: Queues.defaultQueue,
+    jobs: [
+      runBatchEvaluationJob,
+      runDocumentJob,
+      runEvaluationJob,
+      createProviderLogJob,
+    ],
+  },
+  [Queues.eventsQueue]: {
+    name: Queues.eventsQueue,
+    jobs: [publishEventJob, createEventJob],
+  },
+  [Queues.eventHandlersQueue]: {
+    name: Queues.eventHandlersQueue,
+    jobs: Object.values(EventHandlers).flat(),
+  },
+} as const

packages/jobs/src/index.ts

@@ -1,5 +1,4 @@
 import { setupQueues } from './queues'
-import startWorkers from './workers'
 
 export { Worker } from 'bullmq'
 
@@ -10,7 +9,3 @@ export function setupJobs() {
 
   return queues
 }
-
-export function setupWorkers() {
-  return startWorkers()
-}

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

@@ -7,13 +7,13 @@ import {
 } from '@latitude-data/core/browser'
 import { findWorkspaceFromDocument } from '@latitude-data/core/data-access'
 import { NotFoundError } from '@latitude-data/core/lib/errors'
+import { queues } from '@latitude-data/core/queues'
 import { CommitsRepository } from '@latitude-data/core/repositories'
 import { previewDataset } from '@latitude-data/core/services/datasets/preview'
 import { WebsocketClient } from '@latitude-data/core/websockets/workers'
 import { Job } from 'bullmq'
 
 import { setupJobs } from '../..'
-import { connection } from '../../utils/connection'
 import { ProgressTracker } from '../../utils/progressTracker'
 
 type RunBatchEvaluationJobParams = {
@@ -66,15 +66,15 @@ export const runBatchEvaluationJob = async (
     )
   })
 
-  const progressTracker = new ProgressTracker(connection, batchId)
+  const progressTracker = new ProgressTracker(queues(), batchId)
   const firstAttempt = job.attemptsMade === 0
 
   if (firstAttempt) {
     await progressTracker.initializeProgress(parameters.length)
   }
 
   const progress = await progressTracker.getProgress()
-  const queues = setupJobs()
+  const jobs = setupJobs()
 
   if (firstAttempt && parameters.length > 0) {
     websockets.emit('evaluationStatus', {
@@ -95,7 +95,7 @@ export const runBatchEvaluationJob = async (
   // enqueued job. This allows us to resume the batch if the job fails.
   for (let i = progress.enqueued; i < parameters.length; i++) {
     const isFirstEnqueued = progress.enqueued === 0
-    await queues.defaultQueue.jobs.enqueueRunDocumentJob({
+    await jobs.defaultQueue.jobs.enqueueRunDocumentJob({
       workspaceId: workspace.id,
       documentUuid: document.documentUuid,
       commitUuid: commit.uuid,