[WIP] Ingest spans and process segments (#1436)

* Ingest spans and process segments

* wip

Author: neoxelox

apps/gateway/src/api.routes.ts

@@ -49,5 +49,8 @@ export const API_ROUTES = {
       annotate:
         '/api/v3/conversations/:conversationUuid/evaluations/:evaluationUuid/annotate',
     },
+    traces: {
+      ingest: '/api/v3/traces',
+    },
   },
 }

apps/gateway/src/routes/api/index.ts

@@ -1,8 +1,9 @@
+import { default as v1Routes } from '$/routes/api/v1'
+import { default as v2Routes } from '$/routes/api/v2'
 import { conversationsRouter } from '$/routes/api/v3/conversations'
 import { projectsRouter } from '$/routes/api/v3/projects'
+import { tracesRouter } from '$/routes/api/v3/traces'
 import { OpenAPIHono } from '@hono/zod-openapi'
-import { default as v1Routes } from '$/routes/api/v1'
-import { default as v2Routes } from '$/routes/api/v2'
 
 export function configureApiRoutes(app: OpenAPIHono) {
   // Deprecated
@@ -12,4 +13,5 @@ export function configureApiRoutes(app: OpenAPIHono) {
   // V3
   app.route('/', conversationsRouter)
   app.route('/', projectsRouter)
+  app.route('/', tracesRouter)
 }

apps/gateway/src/routes/api/v1/run/run.handler.ts

@@ -42,7 +42,7 @@ export const runHandler: AppRouteHandler<RunRoute> = async (c) => {
       }
 
       const { stream: newStream } = await runDocumentAtCommit({
-        context: BACKGROUND(),
+        context: BACKGROUND({ workspaceId: workspace.id }),
         workspace,
         document,
         commit,

apps/gateway/src/routes/api/v2/documents/run/run.handler.ts

@@ -51,7 +51,7 @@ export const runHandler: AppRouteHandler<RunRoute> = async (c) => {
     error,
     trace,
   } = await runDocumentAtCommit({
-    context: BACKGROUND(),
+    context: BACKGROUND({ workspaceId: workspace.id }),
     workspace,
     document,
     commit,

apps/gateway/src/routes/api/v3/conversations/chat/chat.handler.ts

@@ -15,7 +15,7 @@ export const chatHandler: AppRouteHandler<ChatRoute> = async (c) => {
   const { messages, stream: useSSE, trace, __internal } = c.req.valid('json')
   const workspace = c.get('workspace')
 
-  let context = BACKGROUND()
+  let context = BACKGROUND({ workspaceId: workspace.id })
   if (trace) context = telemetry.resume(trace)
 
   const result = (

apps/gateway/src/routes/api/v3/projects/versions/documents/run/run.handler.ts

@@ -44,7 +44,7 @@ export const runHandler: AppRouteHandler<RunRoute> = async (c) => {
   }
 
   const result = await runDocumentAtCommit({
-    context: BACKGROUND(),
+    context: BACKGROUND({ workspaceId: workspace.id }),
     workspace,
     document,
     commit,

apps/gateway/src/routes/api/v3/traces/index.ts

@@ -0,0 +1,5 @@
+import { createRouter } from '$/openApi/createApp'
+import { ingestHandler } from './ingest/ingest.handler'
+import { ingestRoute } from './ingest/ingest.route'
+
+export const tracesRouter = createRouter().openapi(ingestRoute, ingestHandler)

apps/gateway/src/routes/api/v3/traces/ingest/ingest.handler.ts

@@ -0,0 +1,22 @@
+import { AppRouteHandler } from '$/openApi/types'
+import { TRACING_JOBS_MAX_ATTEMPTS } from '@latitude-data/core/browser'
+import { tracingQueue } from '@latitude-data/core/queues'
+import { IngestRoute } from './ingest.route'
+
+export const ingestHandler: AppRouteHandler<IngestRoute> = async (ctx) => {
+  const workspace = ctx.get('workspace')
+  const apiKey = ctx.get('apiKey')
+  const request = ctx.req.valid('json')
+
+  await tracingQueue.add(
+    'ingestSpansJob',
+    {
+      spans: request.resourceSpans,
+      apiKeyId: apiKey.id,
+      workspaceId: workspace.id,
+    },
+    { attempts: TRACING_JOBS_MAX_ATTEMPTS },
+  )
+
+  return ctx.body(null, 200)
+}

apps/gateway/src/routes/api/v3/traces/ingest/ingest.route.ts

@@ -0,0 +1,27 @@
+import http from '$/common/http'
+import { GENERIC_ERROR_RESPONSES } from '$/openApi/responses/errorResponses'
+import { ROUTES } from '$/routes'
+import { createRoute } from '@hono/zod-openapi'
+import { Otlp } from '@latitude-data/core/browser'
+
+export const ingestRoute = createRoute({
+  method: http.Methods.POST,
+  path: ROUTES.api.v3.traces.ingest,
+  tags: ['Traces'],
+  description: 'Ingest OTLP spans',
+  request: {
+    body: {
+      content: {
+        [http.MediaTypes.JSON]: { schema: Otlp.serviceRequestSchema },
+      },
+    },
+  },
+  responses: {
+    ...GENERIC_ERROR_RESPONSES,
+    [http.Status.OK]: {
+      description: 'Spans ingested successfully',
+    },
+  },
+})
+
+export type IngestRoute = typeof ingestRoute

apps/web/src/app/api/documents/logs/[documentLogUuid]/chat/route.ts

@@ -3,6 +3,8 @@ import {
   LogSources,
   toolCallSchema,
   traceContextSchema,
+  User,
+  Workspace,
 } from '@latitude-data/core/browser'
 import {
   BACKGROUND,
@@ -16,7 +18,6 @@ import { z } from 'zod'
 import { createSdk } from '$/app/(private)/_lib/createSdk'
 import { authHandler } from '$/middlewares/authHandler'
 import { errorHandler } from '$/middlewares/errorHandler'
-import { getCurrentUserOrError } from '$/services/auth/getCurrentUser'
 import { publisher } from '@latitude-data/core/events/publisher'
 
 const inputSchema = z.object({
@@ -36,12 +37,14 @@ export const POST = errorHandler(
       req: NextRequest,
       {
         params,
+        workspace,
+        user,
       }: {
         params: {
           documentLogUuid: string
         }
-        workspace: any
-        user: any
+        workspace: Workspace
+        user: User
       },
     ) => {
       const body = await req.json()
@@ -50,7 +53,9 @@ export const POST = errorHandler(
         const { messages: response, toolCalls, trace } = inputSchema.parse(body)
         const documentLogUuid = params.documentLogUuid
         const messages = response as Message[]
-        const context = trace ? telemetry.resume(trace) : BACKGROUND()
+        const context = trace
+          ? telemetry.resume(trace)
+          : BACKGROUND({ workspaceId: workspace.id })
 
         // Note: faking playground tool spans so they show in the trace
         await fakeToolSpans(context, messages, toolCalls)
@@ -61,14 +66,14 @@ export const POST = errorHandler(
           data: {
             documentLogUuid,
             messages,
-            workspaceId: (await getCurrentUserOrError()).workspace.id,
-            userEmail: (await getCurrentUserOrError()).user.email,
+            workspaceId: workspace.id,
+            userEmail: user.email,
           },
         })
 
         // Create SDK
         const sdkResult = await createSdk({
-          workspace: (await getCurrentUserOrError()).workspace,
+          workspace: workspace,
           __internal: { source: LogSources.Playground },
         })