konnosDashboard
UnifiedLogs.queries.ts406 lines · main
1import dayjs from 'dayjs'
2
3import { DEFAULT_LOG_TYPES } from './UnifiedLogs.constants'
4import { QuerySearchParamsType, SearchParamsType } from './UnifiedLogs.types'
5import {
6 joinSqlFragments,
7 analyticsLiteral as lit,
8 safeSql,
9 type SafeLogSqlFragment,
10} from '@/data/logs/safe-analytics-sql'
11
12// Pagination and control parameters
13const PAGINATION_PARAMS = ['sort', 'start', 'size', 'uuid', 'cursor', 'direction', 'live'] as const
14
15// Special filter parameters that need custom handling
16const SPECIAL_FILTER_PARAMS = ['date'] as const
17
18// Combined list of all parameters to exclude from standard filtering
19const EXCLUDED_QUERY_PARAMS = [...PAGINATION_PARAMS, ...SPECIAL_FILTER_PARAMS] as const
20
21// Facets the count query is allowed to be invoked for. Reject anything else
22// at the entry point rather than letting an unsupported value reach
23// `log_attributes[…]` lookups.
24const FACET_FIELDS = ['log_type', 'level', 'method', 'status', 'pathname'] as const
25
26// OTEL log_attributes keys for HTTP-style fields. Centralized so they can be
27// adjusted in one place if the backend conventions change.
28const ATTR = {
29 method: safeSql`log_attributes['request.method']`,
30 status: safeSql`log_attributes['response.status_code']`,
31 path: safeSql`log_attributes['request.path']`,
32} as const
33
34/**
35 * Predicate that matches rows belonging to a given log_type. Mirrors the
36 * shape of the original BigQuery unified-logs CTEs: edge gateway traffic
37 * (`source = 'edge_logs'`) is split between `edge`, `postgrest` and `storage`
38 * based on URL path. Other types map straight to a single source.
39 *
40 * The OTEL `postgrest_logs` and `storage_logs` sources contain process-level
41 * logs from postgREST / storage-api and are intentionally not part of unified
42 * logs; the UI surfaces gateway HTTP traffic for those buckets.
43 */
44const LOG_TYPE_PREDICATE: Record<string, SafeLogSqlFragment> = {
45 edge: safeSql`source = 'edge_logs' AND ${ATTR.path} NOT LIKE '%/rest/%' AND ${ATTR.path} NOT LIKE '%/storage/%'`,
46 postgrest: safeSql`source = 'edge_logs' AND ${ATTR.path} LIKE '%/rest/%'`,
47 storage: safeSql`source = 'edge_logs' AND ${ATTR.path} LIKE '%/storage/%'`,
48 postgres: safeSql`source = 'postgres_logs'`,
49 'edge function': safeSql`source = 'function_edge_logs'`,
50 auth: safeSql`source = 'auth_logs'`,
51}
52
53// Derived `log_type` column for SELECT / GROUP BY / countIf use.
54const LOG_TYPE_EXPR: SafeLogSqlFragment = safeSql`CASE
55 WHEN source = 'edge_logs' AND ${ATTR.path} LIKE '%/rest/%' THEN 'postgrest'
56 WHEN source = 'edge_logs' AND ${ATTR.path} LIKE '%/storage/%' THEN 'storage'
57 WHEN source = 'edge_logs' THEN 'edge'
58 WHEN source = 'postgres_logs' THEN 'postgres'
59 WHEN source = 'function_edge_logs' THEN 'edge function'
60 WHEN source = 'auth_logs' THEN 'auth'
61 ELSE source
62 END`
63
64// Status code is sourced from the HTTP response for gateway-style rows and
65// from the Postgres `parsed.sql_state_code` (e.g. `42P01`) for postgres rows.
66const STATUS_EXPR: SafeLogSqlFragment = safeSql`CASE
67 WHEN source = 'postgres_logs' THEN toString(log_attributes['parsed.sql_state_code'])
68 ELSE toString(${ATTR.status})
69 END`
70
71// SQL expression for derived `level`. Used inline (not as alias reference)
72// because the OTEL endpoint can't resolve aliases inside countIf when the
73// alias is not in GROUP BY.
74//
75// HTTP status is checked first so gateway rows (which always carry an
76// `severity_text` of `INFO` regardless of response code) bucket as
77// success/warning/error by status. Postgres-style severity is the
78// fallback for rows without a status code.
79const LEVEL_EXPR: SafeLogSqlFragment = safeSql`CASE
80 WHEN ${ATTR.status} != '' AND toInt32OrZero(${ATTR.status}) >= 500 THEN 'error'
81 WHEN ${ATTR.status} != '' AND toInt32OrZero(${ATTR.status}) BETWEEN 400 AND 499 THEN 'warning'
82 WHEN ${ATTR.status} != '' AND toInt32OrZero(${ATTR.status}) BETWEEN 200 AND 299 THEN 'success'
83 WHEN severity_text IN ('ERROR','FATAL','CRITICAL','ALERT','EMERGENCY') THEN 'error'
84 WHEN severity_text IN ('WARN','WARNING') THEN 'warning'
85 WHEN severity_text IN ('TRACE','DEBUG','INFO','LOG','NOTICE') THEN 'success'
86 ELSE 'success'
87 END`
88
89const logTypeWherePredicate = (logTypes: string[]): SafeLogSqlFragment => {
90 const effective = logTypes.filter((t) => t in LOG_TYPE_PREDICATE)
91 const types = effective.length ? effective : [...DEFAULT_LOG_TYPES]
92 const branches = types.map((t) => safeSql`(${LOG_TYPE_PREDICATE[t]})`)
93 return safeSql`(${joinSqlFragments(branches, ' OR ')})`
94}
95
96/**
97 * Translates a frontend filter key/value pair into an underlying SQL predicate.
98 * The OTEL endpoint won't accept queries that reference derived aliases like
99 * `log_type` or `level` in WHERE for some shapes, so we always emit raw-column
100 * predicates (source/severity_text/log_attributes[…]).
101 */
102const translateFilter = (key: string, value: unknown): SafeLogSqlFragment | null => {
103 if (value === null || value === undefined) return null
104
105 const arr = Array.isArray(value) ? (value.length > 0 ? value : null) : null
106 if (Array.isArray(value) && !arr) return null
107
108 const inList = (values: readonly unknown[]): SafeLogSqlFragment =>
109 safeSql`(${joinSqlFragments(
110 values.map((v) => lit(String(v))),
111 ','
112 )})`
113
114 switch (key) {
115 case 'log_type': {
116 const types = (arr ?? [value]).map((v) => String(v))
117 const branches = types.map(
118 (t) => safeSql`(${LOG_TYPE_PREDICATE[t] ?? safeSql`source = ${lit(t)}`})`
119 )
120 return safeSql`(${joinSqlFragments(branches, ' OR ')})`
121 }
122 case 'level': {
123 // No simple raw column for level; reference the inline CASE expression.
124 const levels = arr ?? [value]
125 return safeSql`(${LEVEL_EXPR}) IN ${inList(levels.map((v) => String(v)))}`
126 }
127 case 'method':
128 return arr
129 ? safeSql`${ATTR.method} IN ${inList(arr)}`
130 : safeSql`${ATTR.method} = ${lit(String(value))}`
131 case 'status': {
132 // Match the displayed status: HTTP response code for gateway rows,
133 // Postgres SQLSTATE for postgres rows. Inline STATUS_EXPR so e.g.
134 // filtering on '00000' picks up postgres success rows.
135 const statuses = arr ?? [value]
136 return safeSql`(${STATUS_EXPR}) IN ${inList(statuses.map((v) => String(v)))}`
137 }
138 case 'pathname':
139 return arr
140 ? safeSql`(${joinSqlFragments(
141 arr.map((v) => safeSql`${ATTR.path} LIKE ${lit('%' + String(v) + '%')}`),
142 ' OR '
143 )})`
144 : safeSql`${ATTR.path} LIKE ${lit('%' + String(value) + '%')}`
145 case 'host':
146 // Best-effort: use full request URL since `host` isn't a top-level field.
147 return arr
148 ? safeSql`(${joinSqlFragments(
149 arr.map(
150 (v) => safeSql`log_attributes['request.url'] LIKE ${lit('%' + String(v) + '%')}`
151 ),
152 ' OR '
153 )})`
154 : safeSql`log_attributes['request.url'] LIKE ${lit('%' + String(value) + '%')}`
155 default:
156 // Unknown filter key — fall back to a generic equality on log_attributes.
157 return arr
158 ? safeSql`log_attributes[${lit(key)}] IN ${inList(arr)}`
159 : safeSql`log_attributes[${lit(key)}] = ${lit(String(value))}`
160 }
161}
162
163/**
164 * Builds an array of WHERE predicate fragments from search params, optionally
165 * skipping a specific facet field (used when computing faceted counts).
166 * `log_type` is always handled separately (see `logTypeWherePredicate`).
167 */
168const buildPredicates = (
169 search: QuerySearchParamsType,
170 excludeField?: string
171): SafeLogSqlFragment[] => {
172 const predicates: SafeLogSqlFragment[] = []
173 Object.entries(search).forEach(([key, value]) => {
174 if (key === excludeField) return
175 if (key === 'log_type') return
176 if ((EXCLUDED_QUERY_PARAMS as readonly string[]).includes(key)) return
177 try {
178 const predicate = translateFilter(key, value)
179 if (predicate) predicates.push(predicate)
180 } catch {
181 // analyticsLiteral rejected an unsupported input — drop the predicate.
182 }
183 })
184 return predicates
185}
186
187const whereClause = (predicates: SafeLogSqlFragment[]): SafeLogSqlFragment =>
188 predicates.length > 0 ? safeSql`WHERE ${joinSqlFragments(predicates, ' AND ')}` : safeSql``
189
190/**
191 * Calculates the chart bucketing level (minute/hour/day) given the date range.
192 */
193const calculateChartBucketing = (
194 search: SearchParamsType | Record<string, unknown>
195): 'MINUTE' | 'HOUR' | 'DAY' => {
196 const dateRange = (search.date as Array<Date | string | number | null | undefined>) || []
197
198 const convertToMillis = (timestamp: Date | string | number | null | undefined) => {
199 if (!timestamp) return null
200 if (timestamp instanceof Date) return timestamp.getTime()
201 if (typeof timestamp === 'string') return dayjs(timestamp).valueOf()
202 if (typeof timestamp === 'number') {
203 const str = timestamp.toString()
204 if (str.length >= 16) return Math.floor(timestamp / 1000)
205 return timestamp
206 }
207 return null
208 }
209
210 let startMillis = convertToMillis(dateRange[0])
211 let endMillis = convertToMillis(dateRange[1])
212
213 if (!startMillis) startMillis = dayjs().subtract(1, 'hour').valueOf()
214 if (!endMillis) endMillis = dayjs().valueOf()
215
216 const startTime = dayjs(startMillis)
217 const endTime = dayjs(endMillis)
218
219 const hourDiff = endTime.diff(startTime, 'hour')
220 const dayDiff = endTime.diff(startTime, 'day')
221
222 if (dayDiff >= 2) return 'DAY'
223 if (hourDiff >= 12) return 'HOUR'
224 return 'MINUTE'
225}
226
227const truncationFunction = (level: 'MINUTE' | 'HOUR' | 'DAY'): SafeLogSqlFragment => {
228 switch (level) {
229 case 'DAY':
230 return safeSql`toStartOfDay`
231 case 'HOUR':
232 return safeSql`toStartOfHour`
233 case 'MINUTE':
234 default:
235 return safeSql`toStartOfMinute`
236 }
237}
238
239/**
240 * Returns the projection list for a unified-logs row. All derivations are
241 * inlined so the result can be referenced (or filtered) at the same query
242 * level — the OTEL endpoint rejects subqueries.
243 */
244const rowProjection = (): SafeLogSqlFragment => safeSql`
245 id,
246 null AS source_id,
247 timestamp,
248 ${LOG_TYPE_EXPR} AS log_type,
249 ${STATUS_EXPR} AS status,
250 ${LEVEL_EXPR} AS level,
251 ${ATTR.path} AS pathname,
252 event_message,
253 ${ATTR.method} AS method,
254 null AS log_count,
255 null AS logs
256`
257
258const buildBaseWhere = (
259 search: QuerySearchParamsType,
260 excludeField?: string
261): SafeLogSqlFragment[] => {
262 const effectiveLogTypes = search.log_type?.length ? search.log_type : [...DEFAULT_LOG_TYPES]
263 const parts: SafeLogSqlFragment[] = []
264 if (excludeField !== 'log_type') {
265 parts.push(logTypeWherePredicate(effectiveLogTypes))
266 }
267 parts.push(...buildPredicates(search, excludeField))
268 return parts
269}
270
271/**
272 * Unified logs row query — flat SELECT, no subquery wrapper.
273 */
274export const getUnifiedLogsQuery = (search: QuerySearchParamsType): SafeLogSqlFragment => {
275 const predicates = buildBaseWhere(search)
276 return safeSql`
277SELECT ${rowProjection()}
278FROM logs
279${whereClause(predicates)}
280`
281}
282
283/**
284 * Single-facet count query — a complete flat SELECT with GROUP BY.
285 */
286export const getFacetCountQuery = ({
287 search,
288 facet,
289 facetSearch,
290}: {
291 search: QuerySearchParamsType
292 facet: string
293 facetSearch?: string
294}): SafeLogSqlFragment => {
295 if (!(FACET_FIELDS as readonly string[]).includes(facet)) {
296 throw new Error('Invalid unified logs facet')
297 }
298
299 const MAX_FACETS_QUANTITY = 20
300
301 const facetExpr: SafeLogSqlFragment =
302 facet === 'log_type'
303 ? LOG_TYPE_EXPR
304 : facet === 'level'
305 ? LEVEL_EXPR
306 : facet === 'method'
307 ? ATTR.method
308 : facet === 'status'
309 ? STATUS_EXPR
310 : facet === 'pathname'
311 ? ATTR.path
312 : safeSql`log_attributes[${lit(facet)}]`
313
314 const predicates: SafeLogSqlFragment[] = [
315 ...buildBaseWhere(search, facet),
316 safeSql`(${facetExpr}) IS NOT NULL AND (${facetExpr}) != ''`,
317 ]
318 if (facetSearch) {
319 predicates.push(safeSql`(${facetExpr}) LIKE ${lit('%' + facetSearch + '%')}`)
320 }
321
322 return safeSql`
323SELECT ${lit(facet)} AS dimension, (${facetExpr}) AS value, count() AS count
324FROM logs
325${whereClause(predicates)}
326GROUP BY value
327LIMIT ${lit(MAX_FACETS_QUANTITY)}
328`
329}
330
331/**
332 * Bundled count query — UNION ALL of (dimension, value, count) rows so the
333 * frontend can render facet counts and total in one round trip.
334 */
335export const getLogsCountQuery = (search: QuerySearchParamsType): SafeLogSqlFragment => {
336 // When no predicates remain, fall back to `1` so we emit a valid
337 // tautology rather than a bare `WHERE`.
338 const baseFiltersFor = (excludeField?: string): SafeLogSqlFragment => {
339 const predicates = buildBaseWhere(search, excludeField)
340 return predicates.length > 0 ? joinSqlFragments(predicates, ' AND ') : safeSql`1`
341 }
342
343 // The "total" badge should reflect the user's *current* filter set,
344 // including any active log_type filter. Pass no excludeField so the
345 // log_type predicate is included.
346 const totalSql = safeSql`
347SELECT 'total' AS dimension, 'all' AS value, count() AS count
348FROM logs
349WHERE ${baseFiltersFor()}
350`
351
352 const logTypeBranches = joinSqlFragments(
353 Object.entries(LOG_TYPE_PREDICATE).map(
354 ([logType, predicate]) =>
355 safeSql`
356SELECT 'log_type' AS dimension, ${lit(logType)} AS value, countIf(${predicate}) AS count
357FROM logs
358WHERE ${baseFiltersFor('log_type')}
359`
360 ),
361 ' UNION ALL '
362 )
363
364 const levelBranches = joinSqlFragments(
365 (['success', 'warning', 'error'] as const).map(
366 (lvl) =>
367 safeSql`
368SELECT 'level' AS dimension, ${lit(lvl)} AS value, countIf((${LEVEL_EXPR}) = ${lit(lvl)}) AS count
369FROM logs
370WHERE ${baseFiltersFor('level')}
371`
372 ),
373 ' UNION ALL '
374 )
375
376 const facetBranches = joinSqlFragments(
377 (['method', 'status', 'pathname'] as const).map(
378 (facet) => safeSql`(${getFacetCountQuery({ search, facet })})`
379 ),
380 ' UNION ALL '
381 )
382
383 return joinSqlFragments([totalSql, logTypeBranches, levelBranches, facetBranches], ' UNION ALL ')
384}
385
386/**
387 * Logs chart query with dynamic bucketing based on time range.
388 */
389export const getLogsChartQuery = (search: QuerySearchParamsType): SafeLogSqlFragment => {
390 const truncationLevel = calculateChartBucketing(search)
391 const truncFn = truncationFunction(truncationLevel)
392 const predicates = buildBaseWhere(search)
393
394 return safeSql`
395SELECT
396 ${truncFn}(timestamp) AS time_bucket,
397 countIf((${LEVEL_EXPR}) = 'success') AS success,
398 countIf((${LEVEL_EXPR}) = 'warning') AS warning,
399 countIf((${LEVEL_EXPR}) = 'error') AS error,
400 count() AS total_per_bucket
401FROM logs
402${whereClause(predicates)}
403GROUP BY time_bucket
404ORDER BY time_bucket ASC
405`
406}