import dayjs from 'dayjs' import { DEFAULT_LOG_TYPES } from './UnifiedLogs.constants' import { QuerySearchParamsType, SearchParamsType } from './UnifiedLogs.types' import { joinSqlFragments, analyticsLiteral as lit, safeSql, type SafeLogSqlFragment, } from '@/data/logs/safe-analytics-sql' // Pagination and control parameters const PAGINATION_PARAMS = ['sort', 'start', 'size', 'uuid', 'cursor', 'direction', 'live'] as const // Special filter parameters that need custom handling const SPECIAL_FILTER_PARAMS = ['date'] as const // Combined list of all parameters to exclude from standard filtering const EXCLUDED_QUERY_PARAMS = [...PAGINATION_PARAMS, ...SPECIAL_FILTER_PARAMS] as const // Facets the count query is allowed to be invoked for. Reject anything else // at the entry point rather than letting an unsupported value reach // `log_attributes[…]` lookups. const FACET_FIELDS = ['log_type', 'level', 'method', 'status', 'pathname'] as const // OTEL log_attributes keys for HTTP-style fields. Centralized so they can be // adjusted in one place if the backend conventions change. const ATTR = { method: safeSql`log_attributes['request.method']`, status: safeSql`log_attributes['response.status_code']`, path: safeSql`log_attributes['request.path']`, } as const /** * Predicate that matches rows belonging to a given log_type. Mirrors the * shape of the original BigQuery unified-logs CTEs: edge gateway traffic * (`source = 'edge_logs'`) is split between `edge`, `postgrest` and `storage` * based on URL path. Other types map straight to a single source. * * The OTEL `postgrest_logs` and `storage_logs` sources contain process-level * logs from postgREST / storage-api and are intentionally not part of unified * logs; the UI surfaces gateway HTTP traffic for those buckets. */ const LOG_TYPE_PREDICATE: Record = { edge: safeSql`source = 'edge_logs' AND ${ATTR.path} NOT LIKE '%/rest/%' AND ${ATTR.path} NOT LIKE '%/storage/%'`, postgrest: safeSql`source = 'edge_logs' AND ${ATTR.path} LIKE '%/rest/%'`, storage: safeSql`source = 'edge_logs' AND ${ATTR.path} LIKE '%/storage/%'`, postgres: safeSql`source = 'postgres_logs'`, 'edge function': safeSql`source = 'function_edge_logs'`, auth: safeSql`source = 'auth_logs'`, } // Derived `log_type` column for SELECT / GROUP BY / countIf use. const LOG_TYPE_EXPR: SafeLogSqlFragment = safeSql`CASE WHEN source = 'edge_logs' AND ${ATTR.path} LIKE '%/rest/%' THEN 'postgrest' WHEN source = 'edge_logs' AND ${ATTR.path} LIKE '%/storage/%' THEN 'storage' WHEN source = 'edge_logs' THEN 'edge' WHEN source = 'postgres_logs' THEN 'postgres' WHEN source = 'function_edge_logs' THEN 'edge function' WHEN source = 'auth_logs' THEN 'auth' ELSE source END` // Status code is sourced from the HTTP response for gateway-style rows and // from the Postgres `parsed.sql_state_code` (e.g. `42P01`) for postgres rows. const STATUS_EXPR: SafeLogSqlFragment = safeSql`CASE WHEN source = 'postgres_logs' THEN toString(log_attributes['parsed.sql_state_code']) ELSE toString(${ATTR.status}) END` // SQL expression for derived `level`. Used inline (not as alias reference) // because the OTEL endpoint can't resolve aliases inside countIf when the // alias is not in GROUP BY. // // HTTP status is checked first so gateway rows (which always carry an // `severity_text` of `INFO` regardless of response code) bucket as // success/warning/error by status. Postgres-style severity is the // fallback for rows without a status code. const LEVEL_EXPR: SafeLogSqlFragment = safeSql`CASE WHEN ${ATTR.status} != '' AND toInt32OrZero(${ATTR.status}) >= 500 THEN 'error' WHEN ${ATTR.status} != '' AND toInt32OrZero(${ATTR.status}) BETWEEN 400 AND 499 THEN 'warning' WHEN ${ATTR.status} != '' AND toInt32OrZero(${ATTR.status}) BETWEEN 200 AND 299 THEN 'success' WHEN severity_text IN ('ERROR','FATAL','CRITICAL','ALERT','EMERGENCY') THEN 'error' WHEN severity_text IN ('WARN','WARNING') THEN 'warning' WHEN severity_text IN ('TRACE','DEBUG','INFO','LOG','NOTICE') THEN 'success' ELSE 'success' END` const logTypeWherePredicate = (logTypes: string[]): SafeLogSqlFragment => { const effective = logTypes.filter((t) => t in LOG_TYPE_PREDICATE) const types = effective.length ? effective : [...DEFAULT_LOG_TYPES] const branches = types.map((t) => safeSql`(${LOG_TYPE_PREDICATE[t]})`) return safeSql`(${joinSqlFragments(branches, ' OR ')})` } /** * Translates a frontend filter key/value pair into an underlying SQL predicate. * The OTEL endpoint won't accept queries that reference derived aliases like * `log_type` or `level` in WHERE for some shapes, so we always emit raw-column * predicates (source/severity_text/log_attributes[…]). */ const translateFilter = (key: string, value: unknown): SafeLogSqlFragment | null => { if (value === null || value === undefined) return null const arr = Array.isArray(value) ? (value.length > 0 ? value : null) : null if (Array.isArray(value) && !arr) return null const inList = (values: readonly unknown[]): SafeLogSqlFragment => safeSql`(${joinSqlFragments( values.map((v) => lit(String(v))), ',' )})` switch (key) { case 'log_type': { const types = (arr ?? [value]).map((v) => String(v)) const branches = types.map( (t) => safeSql`(${LOG_TYPE_PREDICATE[t] ?? safeSql`source = ${lit(t)}`})` ) return safeSql`(${joinSqlFragments(branches, ' OR ')})` } case 'level': { // No simple raw column for level; reference the inline CASE expression. const levels = arr ?? [value] return safeSql`(${LEVEL_EXPR}) IN ${inList(levels.map((v) => String(v)))}` } case 'method': return arr ? safeSql`${ATTR.method} IN ${inList(arr)}` : safeSql`${ATTR.method} = ${lit(String(value))}` case 'status': { // Match the displayed status: HTTP response code for gateway rows, // Postgres SQLSTATE for postgres rows. Inline STATUS_EXPR so e.g. // filtering on '00000' picks up postgres success rows. const statuses = arr ?? [value] return safeSql`(${STATUS_EXPR}) IN ${inList(statuses.map((v) => String(v)))}` } case 'pathname': return arr ? safeSql`(${joinSqlFragments( arr.map((v) => safeSql`${ATTR.path} LIKE ${lit('%' + String(v) + '%')}`), ' OR ' )})` : safeSql`${ATTR.path} LIKE ${lit('%' + String(value) + '%')}` case 'host': // Best-effort: use full request URL since `host` isn't a top-level field. return arr ? safeSql`(${joinSqlFragments( arr.map( (v) => safeSql`log_attributes['request.url'] LIKE ${lit('%' + String(v) + '%')}` ), ' OR ' )})` : safeSql`log_attributes['request.url'] LIKE ${lit('%' + String(value) + '%')}` default: // Unknown filter key — fall back to a generic equality on log_attributes. return arr ? safeSql`log_attributes[${lit(key)}] IN ${inList(arr)}` : safeSql`log_attributes[${lit(key)}] = ${lit(String(value))}` } } /** * Builds an array of WHERE predicate fragments from search params, optionally * skipping a specific facet field (used when computing faceted counts). * `log_type` is always handled separately (see `logTypeWherePredicate`). */ const buildPredicates = ( search: QuerySearchParamsType, excludeField?: string ): SafeLogSqlFragment[] => { const predicates: SafeLogSqlFragment[] = [] Object.entries(search).forEach(([key, value]) => { if (key === excludeField) return if (key === 'log_type') return if ((EXCLUDED_QUERY_PARAMS as readonly string[]).includes(key)) return try { const predicate = translateFilter(key, value) if (predicate) predicates.push(predicate) } catch { // analyticsLiteral rejected an unsupported input — drop the predicate. } }) return predicates } const whereClause = (predicates: SafeLogSqlFragment[]): SafeLogSqlFragment => predicates.length > 0 ? safeSql`WHERE ${joinSqlFragments(predicates, ' AND ')}` : safeSql`` /** * Calculates the chart bucketing level (minute/hour/day) given the date range. */ const calculateChartBucketing = ( search: SearchParamsType | Record ): 'MINUTE' | 'HOUR' | 'DAY' => { const dateRange = (search.date as Array) || [] const convertToMillis = (timestamp: Date | string | number | null | undefined) => { if (!timestamp) return null if (timestamp instanceof Date) return timestamp.getTime() if (typeof timestamp === 'string') return dayjs(timestamp).valueOf() if (typeof timestamp === 'number') { const str = timestamp.toString() if (str.length >= 16) return Math.floor(timestamp / 1000) return timestamp } return null } let startMillis = convertToMillis(dateRange[0]) let endMillis = convertToMillis(dateRange[1]) if (!startMillis) startMillis = dayjs().subtract(1, 'hour').valueOf() if (!endMillis) endMillis = dayjs().valueOf() const startTime = dayjs(startMillis) const endTime = dayjs(endMillis) const hourDiff = endTime.diff(startTime, 'hour') const dayDiff = endTime.diff(startTime, 'day') if (dayDiff >= 2) return 'DAY' if (hourDiff >= 12) return 'HOUR' return 'MINUTE' } const truncationFunction = (level: 'MINUTE' | 'HOUR' | 'DAY'): SafeLogSqlFragment => { switch (level) { case 'DAY': return safeSql`toStartOfDay` case 'HOUR': return safeSql`toStartOfHour` case 'MINUTE': default: return safeSql`toStartOfMinute` } } /** * Returns the projection list for a unified-logs row. All derivations are * inlined so the result can be referenced (or filtered) at the same query * level — the OTEL endpoint rejects subqueries. */ const rowProjection = (): SafeLogSqlFragment => safeSql` id, null AS source_id, timestamp, ${LOG_TYPE_EXPR} AS log_type, ${STATUS_EXPR} AS status, ${LEVEL_EXPR} AS level, ${ATTR.path} AS pathname, event_message, ${ATTR.method} AS method, null AS log_count, null AS logs ` const buildBaseWhere = ( search: QuerySearchParamsType, excludeField?: string ): SafeLogSqlFragment[] => { const effectiveLogTypes = search.log_type?.length ? search.log_type : [...DEFAULT_LOG_TYPES] const parts: SafeLogSqlFragment[] = [] if (excludeField !== 'log_type') { parts.push(logTypeWherePredicate(effectiveLogTypes)) } parts.push(...buildPredicates(search, excludeField)) return parts } /** * Unified logs row query — flat SELECT, no subquery wrapper. */ export const getUnifiedLogsQuery = (search: QuerySearchParamsType): SafeLogSqlFragment => { const predicates = buildBaseWhere(search) return safeSql` SELECT ${rowProjection()} FROM logs ${whereClause(predicates)} ` } /** * Single-facet count query — a complete flat SELECT with GROUP BY. */ export const getFacetCountQuery = ({ search, facet, facetSearch, }: { search: QuerySearchParamsType facet: string facetSearch?: string }): SafeLogSqlFragment => { if (!(FACET_FIELDS as readonly string[]).includes(facet)) { throw new Error('Invalid unified logs facet') } const MAX_FACETS_QUANTITY = 20 const facetExpr: SafeLogSqlFragment = facet === 'log_type' ? LOG_TYPE_EXPR : facet === 'level' ? LEVEL_EXPR : facet === 'method' ? ATTR.method : facet === 'status' ? STATUS_EXPR : facet === 'pathname' ? ATTR.path : safeSql`log_attributes[${lit(facet)}]` const predicates: SafeLogSqlFragment[] = [ ...buildBaseWhere(search, facet), safeSql`(${facetExpr}) IS NOT NULL AND (${facetExpr}) != ''`, ] if (facetSearch) { predicates.push(safeSql`(${facetExpr}) LIKE ${lit('%' + facetSearch + '%')}`) } return safeSql` SELECT ${lit(facet)} AS dimension, (${facetExpr}) AS value, count() AS count FROM logs ${whereClause(predicates)} GROUP BY value LIMIT ${lit(MAX_FACETS_QUANTITY)} ` } /** * Bundled count query — UNION ALL of (dimension, value, count) rows so the * frontend can render facet counts and total in one round trip. */ export const getLogsCountQuery = (search: QuerySearchParamsType): SafeLogSqlFragment => { // When no predicates remain, fall back to `1` so we emit a valid // tautology rather than a bare `WHERE`. const baseFiltersFor = (excludeField?: string): SafeLogSqlFragment => { const predicates = buildBaseWhere(search, excludeField) return predicates.length > 0 ? joinSqlFragments(predicates, ' AND ') : safeSql`1` } // The "total" badge should reflect the user's *current* filter set, // including any active log_type filter. Pass no excludeField so the // log_type predicate is included. const totalSql = safeSql` SELECT 'total' AS dimension, 'all' AS value, count() AS count FROM logs WHERE ${baseFiltersFor()} ` const logTypeBranches = joinSqlFragments( Object.entries(LOG_TYPE_PREDICATE).map( ([logType, predicate]) => safeSql` SELECT 'log_type' AS dimension, ${lit(logType)} AS value, countIf(${predicate}) AS count FROM logs WHERE ${baseFiltersFor('log_type')} ` ), ' UNION ALL ' ) const levelBranches = joinSqlFragments( (['success', 'warning', 'error'] as const).map( (lvl) => safeSql` SELECT 'level' AS dimension, ${lit(lvl)} AS value, countIf((${LEVEL_EXPR}) = ${lit(lvl)}) AS count FROM logs WHERE ${baseFiltersFor('level')} ` ), ' UNION ALL ' ) const facetBranches = joinSqlFragments( (['method', 'status', 'pathname'] as const).map( (facet) => safeSql`(${getFacetCountQuery({ search, facet })})` ), ' UNION ALL ' ) return joinSqlFragments([totalSql, logTypeBranches, levelBranches, facetBranches], ' UNION ALL ') } /** * Logs chart query with dynamic bucketing based on time range. */ export const getLogsChartQuery = (search: QuerySearchParamsType): SafeLogSqlFragment => { const truncationLevel = calculateChartBucketing(search) const truncFn = truncationFunction(truncationLevel) const predicates = buildBaseWhere(search) return safeSql` SELECT ${truncFn}(timestamp) AS time_bucket, countIf((${LEVEL_EXPR}) = 'success') AS success, countIf((${LEVEL_EXPR}) = 'warning') AS warning, countIf((${LEVEL_EXPR}) = 'error') AS error, count() AS total_per_bucket FROM logs ${whereClause(predicates)} GROUP BY time_bucket ORDER BY time_bucket ASC ` }