import { createClient, InsertResult } from "npm:@clickhouse/client"; import { z } from "npm:zod"; import { format, getUnixTime } from 'npm:date-fns'; const schemaVersion = 3; function _client() { return createClient({ url: Deno.env.get("WIDE_CLICKHOUSE_URL"), username: "default", password: Deno.env.get("WIDE_CLICKHOUSE_PW"), }); } export async function runMigrations() { for ( const migration of [ ` CREATE TABLE IF NOT EXISTS logs_${schemaVersion} ( _tenantId LowCardinality(String) NOT NULL CODEC(ZSTD(1)), _timestamp DateTime64(3) NOT NULL CODEC(DoubleDelta, LZ4), _now DateTime64(3) NOT NULL CODEC(ZSTD(1)), _ttl DateTime NOT NULL, _traceId String CODEC(ZSTD(1)), _requestId String CODEC(ZSTD(1)), _spanId String CODEC(ZSTD(1)), _coldStart UInt32 Codec(T64, LZ4), _parentSpanId String CODEC(ZSTD(1)), _startTime DateTime64(3) CODEC(DoubleDelta, LZ4), _endTime DateTime64(3) CODEC(DoubleDelta, LZ4), _duration UInt32 Codec(T64, LZ4), _source String NOT NULL CODEC(ZSTD(1)), \`string.names\` Array(String) CODEC(ZSTD(2)), \`string.values\` Array(String) CODEC(ZSTD(2)), \`number.names\` Array(String) CODEC(ZSTD(2)), \`number.values\` Array(Float64) CODEC(ZSTD(2)), \`boolean.names\` Array(String) CODEC(ZSTD(2)), \`boolean.values\` Array(UInt8) CODEC(ZSTD(2)), INDEX idx_tenant_bloom _tenantId TYPE bloom_filter GRANULARITY 4 ) ENGINE = MergeTree() ORDER BY (_now)`, ] ) { try { await _client().command({ query: migration, }); } catch (e) { console.error(migration, e); throw new Error(`Migration failed: ${e}`); } } } export function parseRows(rows: any[]) { const processedRows = rows.map(row => { const paths: { string: Record; number: Record; boolean: Record; } = { string: {}, number: {}, boolean: {}, }; function processObject(obj: any, currentPath = "") { for (const [key, value] of Object.entries(obj)) { const newPath = currentPath ? `${currentPath}.${key}` : key; if (Array.isArray(value)) { continue; } if (typeof value === "object" && value !== null) { processObject(value, newPath); } else { const type = typeof value; if (type === "string" || type === "number" || type === "boolean") { paths[type][newPath] = value as any; } } } } processObject(row); return paths; }); return processedRows; } // const rows = parseRows([ // { // "name": "John", // "age": 30, // "isActive": true, // "contact": { // "email": "john@example.com", // "phone": { // "home": "555-1234", // "mobile": "555-5678", // }, // }, // "scores": [90, 85, 88], // This array will be ignored // "metadata": { // "lastLogin": "2023-01-01", // "visits": 42, // "preferences": { // "darkMode": true, // "notifications": false, // }, // }, // }, // { // "name": "Jane", // "age": 28, // "isActive": false, // "contact": { // "email": "jane@example.com", // "phone": { // "home": "555-4321", // "work": "555-8765", // }, // }, // "tags": ["developer", "designer"], // This array will be ignored // "metadata": { // "lastLogin": "2023-01-02", // "visits": 37, // "preferences": { // "darkMode": false, // "notifications": true, // }, // }, // }, // ]); const writeRowsToClickhouse = async (userId: string, rows: ReturnType): Promise => { // Write to ClickHouse const now = new Date(); const ttl = new Date(now.getTime() + (1 * 24 * 60 * 60 * 1000)); // 1 day TTL const clickhouseRows = rows.map(row => ({ _tenantId: userId, _timestamp: now, _now: now, _ttl: ttl, "string.names": Object.keys(row.string), "string.values": Object.values(row.string), "number.names": Object.keys(row.number), "number.values": Object.values(row.number), "boolean.names": Object.keys(row.boolean), "boolean.values": Object.values(row.boolean), })); return await _client().insert({ table: `logs_${schemaVersion}`, values: clickhouseRows, format: "JSONEachRow", }); }; // await writeRowsToClickhouse(rows); /** * write! */ export async function write(userId: string, data: any[]) { return await writeRowsToClickhouse(userId, parseRows(data)); } /** * fields types */ export enum FieldType { String = "string", Number = "number", Boolean = "boolean", } /** * values! */ export async function values( userId: string, { field_name, field_type }: { field_name: string; field_type: FieldType }, ) { if (field_type !== FieldType.Boolean && field_type !== FieldType.Number && field_type !== FieldType.String) { throw new Error(`Invalid field type: ${field_type}`); } const queryString = `SELECT DISTINCT \`${field_type}.values\`[arrayFirst(i -> \`${field_type}.names\`[i] = {field_name: String}, arrayEnumerate(\`${field_type}.names\`) )] as value, count(*) as count FROM logs_${schemaVersion} WHERE _tenantId = {userId: String} AND arrayExists(i -> \`${field_type}.names\`[i] = {field_name: String}, arrayEnumerate(\`${field_type}.names\`) ) GROUP BY value ORDER BY count DESC; `; console.log(queryString, { field_name, userId }); const query = await _client().query({ query: queryString, format: "JSONEachRow", query_params: { field_name, userId }, }); return ((await query.json()).map((r: any) => ({ ...r, count: Number(r.count) })) as { value: string; count: number; }[]); } /** * fields! */ export async function fields(userId: string, search: string = "") { return (await (await _client().query({ query: ` SELECT name AS field_name, field_type, count() AS count FROM ( -- String fields SELECT arrayJoin(\`string.names\`) AS name, 'string' AS field_type FROM logs_${schemaVersion} WHERE name ILIKE {name: String} AND _tenantId = {userId: String} UNION ALL -- Number fields SELECT arrayJoin(\`number.names\`) AS name, 'number' AS field_type FROM logs_${schemaVersion} WHERE name ILIKE {name: String} AND _tenantId = {userId: String} UNION ALL -- Boolean fields SELECT arrayJoin(\`boolean.names\`) AS name, 'boolean' AS field_type FROM logs_${schemaVersion} WHERE name ILIKE {name: String} AND _tenantId = {userId: String} ) GROUP BY field_name, field_type `, format: "JSONEachRow", query_params: { name: search, userId }, })).json() as any[]).map(r => ({ ...r, count: Number(r.count) })) as { field_name: string; field_type: FieldType; count: number; }[]; } export const StringOperator = z.enum(["equals", "contains", "startsWith", "endsWith"]); export const NumberOperator = z.enum(["equals", "greaterThan", "lessThan", "greaterThanOrEqual", "lessThanOrEqual"]); export const BooleanOperator = z.enum(["equals"]); export const SearchFilter = z.union([ z.object({ fieldName: z.string(), operator: StringOperator, value: z.string(), }), z.object({ fieldName: z.string(), operator: NumberOperator, value: z.number(), }), z.object({ fieldName: z.string(), operator: BooleanOperator, value: z.boolean(), }), ]); export type TStringOperator = z.infer; export type TNumberOperator = z.infer; export type TBooleanOperator = z.infer; export type TSearchFilter = z.infer; /** * Get the ClickHouse comparison expression for the given operator */ function getComparisonExpression( operator: TStringOperator | TNumberOperator | TBooleanOperator, valueRef: string, ): string { switch (operator) { // String operators case "equals": return `= ${valueRef}`; case "contains": return `LIKE concat('%', ${valueRef}, '%')`; case "startsWith": return `LIKE concat(${valueRef}, '%')`; case "endsWith": return `LIKE concat('%', ${valueRef})`; // Number operators case "greaterThan": return `> ${valueRef}`; case "lessThan": return `< ${valueRef}`; case "greaterThanOrEqual": return `>= ${valueRef}`; case "lessThanOrEqual": return `<= ${valueRef}`; default: throw new Error(`Unsupported operator: ${operator}`); } } /** * Enhanced search function supporting multiple conditions with different operators */ export async function search(userId: string, { filters, start, end, limit = 100, }: { filters: TSearchFilter[]; start: Date | string; end: Date | string; limit?: number; }) { // Build the WHERE clause for each condition const whereConditions = filters.map((filter, index) => { let typeCondition: string; let paramType: string; if (typeof filter.value === "string") { typeCondition = "string"; paramType = "String"; } else if (typeof filter.value === "number") { typeCondition = "number"; paramType = "Float64"; } else if (typeof filter.value === "boolean") { typeCondition = "boolean"; paramType = "UInt8"; } else { throw new Error(`Unsupported value type: ${typeof (filter as any).value}`); } const comparisonExpr = getComparisonExpression( filter.operator, `{value${index}: ${paramType}}`, ); return `arrayExists(i -> \`${typeCondition}.names\`[i] = {field_name${index}: String} AND \`${typeCondition}.values\`[i] ${comparisonExpr}, arrayEnumerate(\`${typeCondition}.names\`) )`; }); // Build the query const queryString = ` SELECT \`string.names\`, \`string.values\`, \`number.names\`, \`number.values\`, \`boolean.names\`, \`boolean.values\` FROM logs_${schemaVersion} WHERE _tenantId = {userId: String} ${whereConditions.length > 0 ? `AND ${whereConditions.join(" AND ")}` : ""} AND _timestamp >= {start: DateTime64} AND _timestamp < {end: DateTime64} limit {limit: UInt64} `; // Build query parameters const queryParams: Record = { userId }; filters.forEach((filter, index) => { queryParams[`field_name${index}`] = filter.fieldName; queryParams[`value${index}`] = typeof filter.value === "boolean" ? Number(filter.value) : filter.value; }); const query = await _client().query({ query: queryString, format: "JSONEachRow", query_params: { ...queryParams, limit, // Best effort does not seem to work so we do this format dance, generally the api // layer validates iso date formats so we are somewhat insulated. start: formatDate(typeof start === "string" ? new Date(start) : start), end: formatDate(typeof end === "string" ? new Date(end) : end), }, clickhouse_settings: { date_time_input_format: "best_effort", date_time_output_format: "iso", }, }); const results = await query.json() as LogRow[]; return reconstructObjects(results); } function formatDate(date: Date) { return String(getUnixTime(date) + (date.getMilliseconds() / 1000)) } interface LogRow { "string.names": string[]; "string.values": string[]; "number.names": string[]; "number.values": number[]; "boolean.names": string[]; "boolean.values": boolean[]; } function reconstructObjects( results: LogRow[], ) { // Reconstruct the original objects from the columnar data return results.map((row) => { const reconstructed: Record = {}; const setNestedValue = (obj: any, path: string, value: any) => { const parts = path.split("."); let current = obj; for (let i = 0; i < parts.length - 1; i++) { if (!(parts[i] in current)) { current[parts[i]] = {}; } current = current[parts[i]]; } current[parts[parts.length - 1]] = value; }; // Process string fields for (let i = 0; i < row["string.names"].length; i++) setNestedValue(reconstructed, row["string.names"][i], row["string.values"][i]); // Process number fields for (let i = 0; i < row["number.names"].length; i++) setNestedValue(reconstructed, row["number.names"][i], row["number.values"][i]); // Process boolean fields for (let i = 0; i < row["boolean.names"].length; i++) setNestedValue(reconstructed, row["boolean.names"][i], Boolean(row["boolean.values"][i])); return reconstructed; }); }

wideLib