feat(schema): Enhance log schema with additional fields and improve data types for better logging structure

Author: abdotop

api/click-house-client.ts

@@ -6,22 +6,25 @@ import {
 } from './lib/env.ts'
 import { respond } from './lib/response.ts'
 import { log } from './lib/log.ts'
-import { ARR, NUM, OBJ, STR, UNION } from './lib/validator.ts'
+import { ARR, NUM, OBJ, optional, STR, UNION } from './lib/validator.ts'
 import { Asserted } from './lib/router.ts'
 
 const LogSchema = OBJ({
-  timestamp: STR(),
-  trace_id: STR(),
-  span_id: STR(),
-  severity_number: NUM(),
-  attributes: OBJ({}),
-  event_name: STR(),
-  context: OBJ({}),
-})
-
-const LogsInputSchema = UNION(LogSchema, ARR(LogSchema))
-
-export type Log = Asserted<typeof LogSchema>
+  timestamp: NUM('The timestamp of the log event'),
+  trace_id: NUM('A float64 representation of the trace ID'),
+  span_id: NUM('A float64 representation of the span ID'),
+  severity_number: NUM('The severity number of the log event'),
+  attributes: optional(OBJ({}, 'A map of attributes')),
+  event_name: STR('The name of the event'),
+  service_version: optional(STR('Service version')),
+  service_instance_id: optional(STR('Service instance ID')),
+}, 'A log event')
+const LogsInputSchema = UNION(
+  LogSchema,
+  ARR(LogSchema, 'An array of log events'),
+)
+
+type Log = Asserted<typeof LogSchema>
 type LogsInput = Asserted<typeof LogsInputSchema>
 
 const client = createClient({
@@ -34,26 +37,65 @@ const client = createClient({
   },
 })
 
+export function float64ToId128(
+  { trace, span }: { trace: number; span?: number },
+) {
+  const id128 = new Uint8Array(16)
+  const view = new DataView(id128.buffer)
+  view.setFloat64(0, trace, false)
+  if (span) {
+    view.setFloat64(8, span, false)
+  } else {
+    view.setFloat64(8, trace, false)
+  }
+  return id128
+}
+
+export function bytesToHex(bytes: Uint8Array) {
+  return Array.from(bytes).map((b) => b.toString(16).padStart(2, '0')).join('')
+}
+const escapeSql = (s: unknown) =>
+  String(s ?? '').replace(/\\/g, '\\\\').replace(/'/g, "''")
+
 async function insertLogs(
-  resource: string,
+  service_name: string,
   data: LogsInput,
 ) {
   const logsToInsert = Array.isArray(data) ? data : [data]
-  if (logsToInsert.length === 0) {
-    throw respond.NoContent()
-  }
-
-  const values = logsToInsert.map((log) => ({
-    ...log,
-    resource,
-  }))
+  if (logsToInsert.length === 0) throw respond.NoContent()
+
+  const rows = logsToInsert.map((log) => {
+    const traceHex = bytesToHex(
+      float64ToId128({ trace: log.trace_id }),
+    )
+    const spanHex = bytesToHex(
+      float64ToId128({
+        trace: log.trace_id,
+        span: log.span_id,
+      }),
+    )
+
+    const ts = log.timestamp
+    const attrs = escapeSql(JSON.stringify(log.attributes))
+    const event = escapeSql(log.event_name)
+    const svc = escapeSql(service_name)
+    const svcVer = escapeSql(log.service_version ?? 'unknown')
+    const svcInst = escapeSql(log.service_instance_id ?? 'unknown')
+    // trace_id / span_id inserted as binary via unhex('<32hex>')
+    return `('${svc}','${svcVer}','${svcInst}','${ts}',unhex('${traceHex}'),unhex('${spanHex}'),${log.severity_number},'${attrs}','${event}')`
+  })
+
+  const q = `
+    INSERT INTO logs
+      (service_name, service_version, service_instance_id,
+       timestamp, trace_id, span_id,
+       severity_number, attributes, event_name)
+    VALUES
+      ${rows.join(',')}
+  `.trim()
 
   try {
-    await client.insert({
-      table: 'logs',
-      values,
-      format: 'JSONEachRow',
-    })
+    await client.command({ query: q })
     return respond.OK()
   } catch (error) {
     log.error('Error inserting logs into ClickHouse:', { error })
@@ -89,9 +131,9 @@ async function getLogs({
   search?: Record<string, string>
 }) {
   const queryParts: string[] = []
-  const queryParams: Record<string, unknown> = { resource }
+  const queryParams: Record<string, unknown> = { service_name: resource }
 
-  queryParts.push('resource = {resource:String}')
+  queryParts.push('service_name = {service_name:String}')
 
   if (level) {
     queryParts.push('severity_number = {level:UInt8}')
@@ -134,11 +176,12 @@ async function getLogs({
   try {
     const resultSet = await client.query({
       query,
+
       query_params: queryParams,
-      format: 'JSONEachRow',
+      format: 'JSON',
     })
 
-    return resultSet.json<Log[]>()
+    return (await resultSet.json<Log>()).data
   } catch (error) {
     log.error('Error querying logs from ClickHouse:', { error })
     throw respond.InternalServerError()

api/lib/json_store.ts

@@ -32,10 +32,10 @@ const batch = async <T>(
   await Promise.all(pool)
 }
 
-export type BaseRecord = { createdAt?: number; updatedAt?: number }
+export type BaseRecord = { createdAt: number; updatedAt: number }
 
 export async function createCollection<
-  T extends Record<PropertyKey, unknown> & BaseRecord,
+  T extends Record<PropertyKey, unknown>,
   K extends keyof T,
 >({ name, primaryKey }: CollectionOptions<T, K>) {
   const dir = join(DB_DIR, name)
@@ -66,34 +66,34 @@ export async function createCollection<
       const id = data[primaryKey]
       if (!id) throw Error(`Missing primary key ${primaryKey}`)
       if (records.has(id)) throw Error(`${id} already exists`)
-      Object.assign(data, { createdAt: Date.now() })
+      Object.assign(data, { createdAt: Date.now(), updatedAt: Date.now() })
       records.set(id, data)
       await saveRecord(data)
 
-      return data
+      return data as T & BaseRecord
     },
 
     [Symbol.iterator]: () => records[Symbol.iterator],
     keys: () => records.keys(),
-    values: () => records.values(),
-    entries: () => records.entries(),
-    get: (id: T[K]) => records.get(id),
+    values: () => records.values() as MapIterator<T & BaseRecord>,
+    entries: () => records.entries() as MapIterator<[T[K], T & BaseRecord]>,
+    get: (id: T[K]) => records.get(id) as T & BaseRecord | undefined,
     assert: (id: T[K]) => {
       const match = records.get(id)
-      if (match) return match
+      if (match) return match as T & BaseRecord
       throw new Deno.errors.NotFound(`record ${id} not found`)
     },
     find: (predicate: (record: T) => unknown) =>
-      records.values().find(predicate),
+      records.values().find(predicate) as T & BaseRecord | undefined,
     filter: (predicate: (record: T) => unknown) =>
-      records.values().filter(predicate).toArray(),
+      records.values().filter(predicate).toArray() as (T & BaseRecord)[],
     async update(id: T[K], changes: Partial<Omit<T, K>>) {
       const record = records.get(id)
       if (!record) throw new Deno.errors.NotFound(`record ${id} not found`)
-      const updated = { ...record, ...changes, _updatedAt: Date.now() } as T
+      const updated = { ...record, ...changes, updatedAt: Date.now() } as T
       records.set(id, updated)
       await saveRecord(updated)
-      return updated
+      return updated as T & BaseRecord
     },
 
     async delete(id: T[K]) {

api/routes.ts

@@ -23,6 +23,7 @@ import {
   LogsInputSchema,
 } from './click-house-client.ts'
 import { decryptMessage, encryptMessage } from './user.ts'
+import { log } from './lib/log.ts'
 
 const withUserSession = ({ user }: RequestContext) => {
   if (!user) throw Error('Missing user session')
@@ -35,14 +36,19 @@ const withAdminSession = ({ user }: RequestContext) => {
 const withDeploymentSession = async (ctx: RequestContext) => {
   const token = ctx.req.headers.get('Authorization')?.replace(/^Bearer /i, '')
   if (!token) throw respond.Unauthorized({ message: 'Missing token' })
-  const message = await decryptMessage(token)
-  if (!message) throw respond.Unauthorized({ message: 'Invalid token' })
-  const data = JSON.parse(message)
-  const dep = DeploymentsCollection.get(data?.url)
-  if (!dep || dep.tokenSalt !== data?.tokenSalt) {
+  try {
+    const message = await decryptMessage(token)
+    if (!message) throw respond.Unauthorized({ message: 'Invalid token' })
+    const data = JSON.parse(message)
+    const dep = DeploymentsCollection.get(data?.url)
+    if (!dep || dep.tokenSalt !== data?.tokenSalt) {
+      throw respond.Unauthorized({ message: 'Invalid token' })
+    }
+    ctx.resource = dep?.url
+  } catch (error) {
+    log.error('Error validating deployment token:', { error })
     throw respond.Unauthorized({ message: 'Invalid token' })
   }
-  ctx.resource = dep?.url
 }
 
 const deploymentOutput = OBJ({
@@ -229,8 +235,6 @@ const defs = {
       return deployments.map(({ tokenSalt: _, ...d }) => {
         return {
           ...d,
-          createdAt: d.createdAt,
-          updatedAt: d.updatedAt,
           token: undefined,
           sqlToken: undefined,
           sqlEndpoint: undefined,
@@ -252,8 +256,6 @@ const defs = {
       )
       return {
         ...deployment,
-        createdAt: deployment.createdAt,
-        updatedAt: deployment.updatedAt,
         token,
       }
     },
@@ -275,8 +277,6 @@ const defs = {
       )
       return {
         ...deployment,
-        createdAt: deployment.createdAt,
-        updatedAt: deployment.updatedAt,
         token,
       }
     },
@@ -294,8 +294,6 @@ const defs = {
       )
       return {
         ...deployment,
-        createdAt: deployment.createdAt,
-        updatedAt: deployment.updatedAt,
         token,
       }
     },
@@ -319,8 +317,6 @@ const defs = {
       )
       return {
         ...deployment,
-        createdAt: deployment.createdAt,
-        updatedAt: deployment.updatedAt,
         token,
       }
     },
@@ -350,9 +346,22 @@ const defs = {
   }),
   'GET/api/logs': route({
     authorize: withUserSession,
-    fn: async (_ctx, params) => {
-      const logs = await getLogs(params)
-      return logs.flat()
+    fn: (ctx, params) => {
+      const deployments = DeploymentsCollection.filter((d) =>
+        d.projectId === params.resource
+      )
+      if (deployments.length === 0) {
+        throw respond.NotFound({ message: 'Deployments not found' })
+      }
+      const project = ProjectsCollection.get(deployments[0].projectId)
+      if (!project) throw respond.NotFound({ message: 'Project not found' })
+      if (!project.isPublic && !ctx.user?.isAdmin) {
+        const team = TeamsCollection.find((t) => t.teamId === project.teamId)
+        if (!team?.teamMembers.includes(ctx.user?.userEmail || '')) {
+          throw respond.Forbidden({ message: 'Access to project logs denied' })
+        }
+      }
+      return getLogs(params)
     },
     input: OBJ({
       resource: STR('The resource to fetch logs for'),
@@ -363,7 +372,7 @@ const defs = {
       sort_order: optional(
         LIST(['ASC', 'DESC'], 'The sort order (ASC or DESC)'),
       ),
-      search: optional(OBJ({}, 'A map of fields to search by')),
+      // search: optional(OBJ({}, 'A map of fields to search by')),
     }),
     output: ARR(LogSchema, 'List of logs'),
     description: 'Get logs from ClickHouse',

api/schema.ts

@@ -1,14 +1,14 @@
 import { ARR, BOOL, OBJ, optional, STR } from './lib/validator.ts'
 import { Asserted } from './lib/router.ts'
-import { BaseRecord, createCollection } from './lib/json_store.ts'
+import { createCollection } from './lib/json_store.ts'
 
 export const UserDef = OBJ({
   userEmail: STR('The user email address'),
   userFullName: STR('The user login name'),
   userPicture: optional(STR('The user profile picture URL')),
   isAdmin: BOOL('Is the user an admin?'),
 }, 'The user schema definition')
-export type User = Asserted<typeof UserDef> & BaseRecord
+export type User = Asserted<typeof UserDef>
 
 export const TeamDef = OBJ({
   teamId: STR('The unique identifier for the team'),
@@ -18,7 +18,7 @@ export const TeamDef = OBJ({
     'The list of user emails who are members of the team',
   ),
 }, 'The team schema definition')
-export type Team = Asserted<typeof TeamDef> & BaseRecord
+export type Team = Asserted<typeof TeamDef>
 
 export const ProjectDef = OBJ({
   slug: STR('The unique identifier for the project'),
@@ -27,7 +27,7 @@ export const ProjectDef = OBJ({
   isPublic: BOOL('Is the project public?'),
   repositoryUrl: optional(STR('The URL of the project repository')),
 }, 'The project schema definition')
-export type Project = Asserted<typeof ProjectDef> & BaseRecord
+export type Project = Asserted<typeof ProjectDef>
 
 export const DeploymentDef = OBJ({
   projectId: STR('The ID of the project this deployment belongs to'),
@@ -37,7 +37,7 @@ export const DeploymentDef = OBJ({
   sqlEndpoint: optional(STR('The SQL execution endpoint for the database')),
   sqlToken: optional(STR('The security token for the SQL endpoint')),
 }, 'The deployment schema definition')
-export type Deployment = Asserted<typeof DeploymentDef> & BaseRecord
+export type Deployment = Asserted<typeof DeploymentDef>
 
 export const UsersCollection = await createCollection<User, 'userEmail'>(
   { name: 'users', primaryKey: 'userEmail' },