feat(studio): add bounded request correlation queries

apps/studio/components/interfaces/UnifiedLogs/UnifiedLogs.queries.test.ts

@@ -4,6 +4,8 @@ import {
   getFacetCountQuery,
   getLogsChartQuery,
   getLogsCountQuery,
+  getRequestCorrelationIds,
+  getRequestTimelineQuery,
   getUnifiedLogsQuery,
   isUserFilterUnreachable,
 } from './UnifiedLogs.queries'
@@ -582,3 +584,55 @@ describe('pathname ILIKE prefix matching (cross-builder)', () => {
     expect(bqSql).not.toContain("LOWER('%foo%%')")
   })
 })
+
+describe('request timeline', () => {
+  it('reads request and execution IDs from the keys each source sets', () => {
+    expect(
+      getRequestCorrelationIds([
+        { source: 'edge_logs', attributes: { 'request.headers.cf_ray': '8f00aa-IAD' } },
+        { source: 'auth_logs', attributes: { request_id: '8f00aa-IAD' } },
+        {
+          source: 'function_edge_logs',
+          attributes: { execution_id: 'exec-1', 'request.headers.cf_ray': '8f00bb-IAD' },
+        },
+        // Keys are only read for the sources that set them
+        { source: 'postgres_logs', attributes: { request_id: 'ignored' } },
+      ])
+    ).toEqual({ requestIds: ['8f00aa-IAD', '8f00bb-IAD'], executionIds: ['exec-1'] })
+  })
+
+  it('ignores IDs that are not plain identifiers', () => {
+    expect(
+      getRequestCorrelationIds([
+        { source: 'edge_logs', attributes: { 'request.headers.cf_ray': "x' OR 1=1 --" } },
+        { source: 'auth_logs', attributes: { request_id: 42 } },
+        { source: 'storage_logs', attributes: null },
+      ])
+    ).toEqual({ requestIds: [], executionIds: [] })
+  })
+
+  it('matches each source on its own key and orders oldest first', () => {
+    const sql = getRequestTimelineQuery({ requestIds: ['8f00aa-IAD'], executionIds: ['exec-1'] })
+    expect(sql).toContain(
+      `(source = 'edge_logs' AND (log_attributes['request.headers.cf_ray'] IN ('8f00aa-IAD') OR log_attributes['response.headers.cf_ray'] IN ('8f00aa-IAD')))`
+    )
+    expect(sql).toContain(
+      `(source = 'auth_logs' AND (log_attributes['request_id'] IN ('8f00aa-IAD')))`
+    )
+    expect(sql).toContain(
+      `(source = 'function_logs' AND (log_attributes['execution_id'] IN ('exec-1')))`
+    )
+    expect(sql).toContain('log_attributes AS metadata')
+    expect(sql).toContain('ORDER BY timestamp ASC')
+    expect(sql).toMatch(/LIMIT 200\s*$/)
+  })
+
+  it('skips execution ID conditions when there are none', () => {
+    const sql = getRequestTimelineQuery({ requestIds: ['8f00aa-IAD'], executionIds: [] })
+    expect(sql).not.toContain('execution_id')
+  })
+
+  it('refuses to build an unbounded query', () => {
+    expect(() => getRequestTimelineQuery({ requestIds: [], executionIds: [] })).toThrow()
+  })
+})

apps/studio/components/interfaces/UnifiedLogs/UnifiedLogs.queries.ts

@@ -591,6 +591,67 @@ ORDER BY time_bucket ASC
 `
 }
 
+/**
+ * `log_attributes` keys that carry the identifiers linking logs from one
+ * request. The Cloudflare ray ID follows a request through the API Gateway and
+ * into Auth and Storage (Auth logs it as `request_id`). Edge Functions add an
+ * `execution_id` shared by the invocation and its console output.
+ */
+export const REQUEST_CORRELATION_KEYS = {
+  requestId: {
+    edge_logs: ['request.headers.cf_ray', 'response.headers.cf_ray'],
+    auth_logs: ['request_id'],
+    storage_logs: ['req.headers.cf_ray'],
+    function_edge_logs: ['request.headers.cf_ray'],
+  },
+  executionId: {
+    function_edge_logs: ['execution_id'],
+    function_logs: ['execution_id'],
+  },
+} as const satisfies Record<'requestId' | 'executionId', Record<string, readonly string[]>>
+
+export type RequestCorrelationIds = { requestIds: string[]; executionIds: string[] }
+
+export const REQUEST_TIMELINE_LIMIT = 200
+
+// Ray and execution IDs are hex/uuid-shaped with an optional `-COLO` suffix.
+const CORRELATION_ID_PATTERN = /^[0-9A-Za-z-]{1,64}$/
+
+/** Reads the request and execution IDs a log carries, per its source. */
+export const getRequestCorrelationIds = (
+  logs: { source: string; attributes: Record<string, unknown> | null | undefined }[]
+): RequestCorrelationIds => {
+  const collect = (keysBySource: Record<string, readonly string[]>) => {
+    const ids = new Set<string>()
+    for (const { source, attributes } of logs) {
+      for (const key of keysBySource[source] ?? []) {
+        const value = attributes?.[key]
+        if (typeof value === 'string' && CORRELATION_ID_PATTERN.test(value)) ids.add(value)
+      }
+    }
+    return [...ids].sort()
+  }
+  return {
+    requestIds: collect(REQUEST_CORRELATION_KEYS.requestId),
+    executionIds: collect(REQUEST_CORRELATION_KEYS.executionId),
+  }
+}
+
+const correlationConditions = (
+  keysBySource: Record<string, readonly string[]>,
+  ids: string[]
+): SafeLogSqlFragment[] => {
+  if (ids.length === 0) return []
+  const idList = joinSqlFragments(
+    ids.map((id) => lit(id)),
+    ','
+  )
+  return Object.entries(keysBySource).map(([source, keys]) => {
+    const matches = keys.map((key) => safeSql`log_attributes[${lit(key)}] IN (${idList})`)
+    return safeSql`(source = ${lit(source)} AND (${joinSqlFragments(matches, ' OR ')}))`
+  })
+}
+
 /** One log's `log_attributes`, looked up by id within its source. */
 export const getLogAttributesQuery = ({
   logId,
@@ -604,3 +665,40 @@ FROM logs
 WHERE id = ${lit(logId)} AND source = ${lit(source)}
 LIMIT 1
 `
+
+/**
+ * Every log that shares a request or execution ID, oldest first. Selects the
+ * same columns as the row list, with `log_attributes` as `metadata` so each
+ * log can be shown in full without another lookup.
+ */
+export const getRequestTimelineQuery = ({
+  requestIds,
+  executionIds,
+}: RequestCorrelationIds): SafeLogSqlFragment => {
+  const conditions = [
+    ...correlationConditions(REQUEST_CORRELATION_KEYS.requestId, requestIds),
+    ...correlationConditions(REQUEST_CORRELATION_KEYS.executionId, executionIds),
+  ]
+  if (conditions.length === 0) throw new Error('A request or execution ID is required')
+
+  return safeSql`-- unified logs: request timeline
+SELECT
+    id,
+    timestamp,
+    source,
+    ${LOG_TYPE_EXPR} AS log_type,
+    ${STATUS_EXPR} AS status,
+    ${LEVEL_EXPR} AS level,
+    ${PATHNAME_EXPR} AS pathname,
+    event_message,
+    ${METHOD_EXPR} AS method,
+    ${AUTH_USER_EXPR} AS auth_user,
+    log_attributes AS metadata,
+    null AS log_count,
+    null AS logs
+FROM logs
+WHERE ${joinSqlFragments(conditions, ' OR ')}
+ORDER BY timestamp ASC, id ASC
+LIMIT ${lit(REQUEST_TIMELINE_LIMIT)}
+`
+}

apps/studio/data/logs/keys.ts

@@ -65,6 +65,20 @@ export const logsKeys = {
       logId,
       { source, logTimestampMs },
     ] as const,
+  requestTimeline: (
+    projectRef: string | undefined,
+    logId: string | undefined,
+    source: string | undefined,
+    logTimestampMs: number | null | undefined
+  ) =>
+    [
+      'projects',
+      projectRef,
+      'unified-logs',
+      'request-timeline',
+      logId,
+      { source, logTimestampMs },
+    ] as const,
   serviceFlow: (
     projectRef: string | undefined,
     searchParams: QuerySearchParamsType | undefined,

apps/studio/data/logs/unified-log-attributes-query.ts

@@ -27,13 +27,13 @@ export type UnifiedLogAttributesVariables = {
 
 export type UnifiedLogAttributesError = ResponseError
 
-const getLogTimeWindow = (timestampMs: number, windowMs: number) => ({
+export const getLogTimeWindow = (timestampMs: number, windowMs: number) => ({
   iso_timestamp_start: new Date(timestampMs - windowMs).toISOString(),
   iso_timestamp_end: new Date(timestampMs + windowMs).toISOString(),
 })
 
 /** Runs SQL against the ClickHouse logs endpoint, which can answer 200 with an `error` body. */
-async function runOtelLogsSql(variables: Omit<ExecuteAnalyticsSqlVariables, 'endpoint'>) {
+export async function runOtelLogsSql(variables: Omit<ExecuteAnalyticsSqlVariables, 'endpoint'>) {
   const data = await executeAnalyticsSql({
     ...variables,
     endpoint: '/platform/projects/{ref}/analytics/endpoints/logs.all.otel',
@@ -43,7 +43,7 @@ async function runOtelLogsSql(variables: Omit<ExecuteAnalyticsSqlVariables, 'end
 }
 
 /** One log's source and attributes, or undefined when it can't be found. */
-async function getLogAttributesRow(
+export async function getLogAttributesRow(
   { projectRef, logId, source, logTimestampMs }: UnifiedLogAttributesVariables,
   signal?: AbortSignal
 ) {

apps/studio/data/logs/unified-log-request-timeline-query.test.ts

@@ -0,0 +1,94 @@
+import { QueryClient } from '@tanstack/react-query'
+import type { platformComponents } from 'api-types'
+import { HttpResponse } from 'msw'
+import { describe, expect, it } from 'vitest'
+
+import { unifiedLogRequestTimelineQueryOptions } from './unified-log-request-timeline-query'
+import { addAPIMock, type APIErrorBody } from '@/tests/lib/msw'
+
+type LogsResponse = platformComponents['schemas']['AnalyticsResponse_Output']
+const variables = {
+  projectRef: 'default',
+  logId: 'gateway',
+  source: 'edge_logs',
+  logTimestampMs: Date.parse('2026-01-01T10:00:00Z'),
+}
+const fetchTimeline = () =>
+  new QueryClient({ defaultOptions: { queries: { retry: false } } }).fetchQuery(
+    unifiedLogRequestTimelineQueryOptions(variables)
+  )
+const step = (id: string, source: string, metadata: Record<string, string>) => ({
+  id,
+  source,
+  metadata,
+  timestamp: '2026-01-01T10:00:00.000000',
+  log_type: 'edge function',
+  status: '200',
+  level: 'success',
+  pathname: '/functions/v1/test',
+  event_message: 'Request',
+  method: 'POST',
+  auth_user: null,
+  log_count: null,
+  logs: null,
+})
+
+describe('request timeline query', () => {
+  it('follows an execution ID from a request and stops after two lookups', async () => {
+    const requests: { sql: string; iso_timestamp_start: string; iso_timestamp_end: string }[] = []
+    const invocation = step('invocation', 'function_edge_logs', { execution_id: 'exec-1' })
+    const consoleLog = step('console', 'function_logs', { execution_id: 'exec-2' })
+    addAPIMock({
+      method: 'post',
+      path: '/platform/projects/:ref/analytics/endpoints/logs.all.otel',
+      response: async ({ request }) => {
+        const body = (await request.json()) as (typeof requests)[number]
+        requests.push(body)
+        if (body.sql.includes('log attributes')) {
+          return HttpResponse.json<LogsResponse>({
+            result: [
+              { source: 'edge_logs', log_attributes: { 'request.headers.cf_ray': 'ray-1' } },
+            ],
+          })
+        }
+        return HttpResponse.json<LogsResponse>({
+          result: requests.length === 2 ? [invocation] : [invocation, consoleLog],
+        })
+      },
+    })
+    const result = await fetchTimeline()
+    expect(result.logs.map((log) => log.id)).toEqual(['invocation', 'console'])
+    expect(requests).toHaveLength(3)
+    expect(requests[2].sql).toContain("log_attributes['execution_id'] IN ('exec-1')")
+    expect(requests[2]).toMatchObject({
+      iso_timestamp_start: '2026-01-01T09:55:00.000Z',
+      iso_timestamp_end: '2026-01-01T10:05:00.000Z',
+    })
+  })
+
+  it.each([
+    { result: [] },
+    { result: [{ source: 'postgres_logs', log_attributes: { request_id: 'unsupported' } }] },
+  ])('returns no related logs without correlation IDs: %j', async ({ result }) => {
+    let requests = 0
+    addAPIMock({
+      method: 'post',
+      path: '/platform/projects/:ref/analytics/endpoints/logs.all.otel',
+      response: () => {
+        requests++
+        return HttpResponse.json<LogsResponse>({ result })
+      },
+    })
+    await expect(fetchTimeline()).resolves.toEqual({ requestIds: [], executionIds: [], logs: [] })
+    expect(requests).toBe(1)
+  })
+
+  it('propagates failed lookups', async () => {
+    addAPIMock({
+      method: 'post',
+      path: '/platform/projects/:ref/analytics/endpoints/logs.all.otel',
+      response: () => HttpResponse.json<APIErrorBody>({ message: 'Unavailable' }, { status: 500 }),
+    })
+    await expect(fetchTimeline()).rejects.toThrow('Unavailable')
+  })
+})

apps/studio/data/logs/unified-log-request-timeline-query.ts

@@ -0,0 +1,113 @@
+import { queryOptions } from '@tanstack/react-query'
+import { z } from 'zod'
+
+import { logsKeys } from './keys'
+import {
+  getLogAttributesRow,
+  getLogTimeWindow,
+  runOtelLogsSql,
+} from './unified-log-attributes-query'
+import { UNIFIED_LOGS_QUERY_OPTIONS } from './unified-logs-infinite-query'
+import { mapUnifiedLogRow, unifiedLogsQueryRowSchema } from './unified-logs.utils'
+import {
+  getRequestCorrelationIds,
+  getRequestTimelineQuery,
+  type RequestCorrelationIds,
+} from '@/components/interfaces/UnifiedLogs/UnifiedLogs.queries'
+import type { ColumnSchema } from '@/components/interfaces/UnifiedLogs/UnifiedLogs.schema'
+import { IS_PLATFORM } from '@/lib/constants'
+import type { ResponseError } from '@/types'
+
+// Related logs can trail the request (e.g. a slow Edge Function), so they get
+// a wider window than the selected log's own lookup.
+const TIMELINE_WINDOW_MS = 5 * 60 * 1000
+
+// Following IDs found on related logs (e.g. an Edge Function's execution ID)
+// can surface more logs; stop after one extra hop.
+const MAX_LOOKUPS = 2
+
+const timelineRowSchema = unifiedLogsQueryRowSchema.extend({ source: z.string() })
+
+export type UnifiedLogRequestTimelineVariables = {
+  projectRef?: string
+  logId?: string
+  /** Source of the selected log, e.g. `edge_logs`. */
+  source?: string
+  logTimestampMs?: number | null
+}
+
+export type UnifiedLogRequestTimelineError = ResponseError
+
+const hasNewIds = (next: RequestCorrelationIds, prev: RequestCorrelationIds) =>
+  next.requestIds.some((id) => !prev.requestIds.includes(id)) ||
+  next.executionIds.some((id) => !prev.executionIds.includes(id))
+
+async function getUnifiedLogRequestTimeline(
+  { projectRef, logId, source, logTimestampMs }: UnifiedLogRequestTimelineVariables,
+  signal?: AbortSignal
+) {
+  if (!projectRef) throw new Error('projectRef is required')
+  if (!logId) throw new Error('logId is required')
+  if (!source) throw new Error('source is required')
+  if (typeof logTimestampMs !== 'number' || !Number.isFinite(logTimestampMs)) {
+    throw new Error('logTimestampMs is required')
+  }
+
+  const root = await getLogAttributesRow({ projectRef, logId, source, logTimestampMs }, signal)
+
+  let ids = getRequestCorrelationIds(
+    root ? [{ source: root.source, attributes: root.log_attributes }] : []
+  )
+  let rows: z.infer<typeof timelineRowSchema>[] = []
+
+  for (let lookup = 0; lookup < MAX_LOOKUPS; lookup++) {
+    if (ids.requestIds.length === 0 && ids.executionIds.length === 0) break
+
+    const result = await runOtelLogsSql({
+      projectRef,
+      sql: getRequestTimelineQuery(ids),
+      ...getLogTimeWindow(logTimestampMs, TIMELINE_WINDOW_MS),
+      signal,
+    })
+    rows = z.array(timelineRowSchema).parse(result)
+
+    const found = getRequestCorrelationIds(
+      rows.map((row) => ({ source: row.source, attributes: row.metadata }))
+    )
+    if (!hasNewIds(found, ids)) break
+    ids = {
+      requestIds: [...new Set([...ids.requestIds, ...found.requestIds])],
+      executionIds: [...new Set([...ids.executionIds, ...found.executionIds])],
+    }
+  }
+
+  // Related logs can come from sources the row list doesn't show (e.g.
+  // `function_logs`), so `log_type` may fall outside the list's log types.
+  const logs = rows.map(
+    (row) => ({ ...mapUnifiedLogRow(row), metadata: row.metadata ?? null }) as ColumnSchema
+  )
+
+  return { requestIds: ids.requestIds, executionIds: ids.executionIds, logs }
+}
+
+export type UnifiedLogRequestTimelineData = Awaited<ReturnType<typeof getUnifiedLogRequestTimeline>>
+
+export const unifiedLogRequestTimelineQueryOptions = ({
+  projectRef,
+  logId,
+  source,
+  logTimestampMs,
+}: UnifiedLogRequestTimelineVariables) =>
+  queryOptions<UnifiedLogRequestTimelineData, UnifiedLogRequestTimelineError>({
+    queryKey: logsKeys.requestTimeline(projectRef, logId, source, logTimestampMs),
+    queryFn: ({ signal }) =>
+      getUnifiedLogRequestTimeline({ projectRef, logId, source, logTimestampMs }, signal),
+    enabled:
+      IS_PLATFORM &&
+      typeof projectRef !== 'undefined' &&
+      typeof logId !== 'undefined' &&
+      typeof source !== 'undefined' &&
+      typeof logTimestampMs === 'number' &&
+      Number.isFinite(logTimestampMs),
+    ...UNIFIED_LOGS_QUERY_OPTIONS,
+  })

apps/studio/data/logs/unified-logs.utils.ts

@@ -12,7 +12,7 @@ type UnifiedLogMetadataRow = {
   event_message?: string | null
 }
 
-const unifiedLogsQueryRowSchema = z.object({
+export const unifiedLogsQueryRowSchema = z.object({
   id: z.string(),
   timestamp: z.union([z.string(), z.number()]),
   log_type: z.string(),