fix: previous period calc, tracing

Author: izadoesdev

apps/api/src/query/batch-executor.ts

@@ -1,4 +1,5 @@
 import { chQuery } from "@databuddy/db";
+import { record, setAttributes } from "../lib/tracing";
 import { QueryBuilders } from "./builders";
 import { SimpleQueryBuilder } from "./simple-builder";
 import type { QueryRequest, SimpleQueryConfig } from "./types";
@@ -13,53 +14,68 @@ function getSchemaSignature(config: SimpleQueryConfig): string | null {
     return fields?.length ? fields.map((f) => `${f.name}:${f.type}`).join(",") : null;
 }
 
-async function runSingle(req: BatchRequest, opts?: BatchOptions): Promise<BatchResult> {
+function runSingle(req: BatchRequest, opts?: BatchOptions): Promise<BatchResult> {
     const config = QueryBuilders[req.type];
     if (!config) {
-        return { type: req.type, data: [], error: `Unknown query type: ${req.type}` };
+        return Promise.resolve({ type: req.type, data: [], error: `Unknown query type: ${req.type}` });
     }
 
-    try {
-        const builder = new SimpleQueryBuilder(
-            config,
-            { ...req, timezone: opts?.timezone ?? req.timezone },
-            opts?.websiteDomain
-        );
-        return { type: req.type, data: await builder.execute() };
-    } catch (e) {
-        return { type: req.type, data: [], error: e instanceof Error ? e.message : "Query failed" };
-    }
+    return record(`query.single.${req.type}`, async () => {
+        const startTime = performance.now();
+        try {
+            const builder = new SimpleQueryBuilder(
+                config,
+                { ...req, timezone: opts?.timezone ?? req.timezone },
+                opts?.websiteDomain
+            );
+            const data = await builder.execute();
+
+            setAttributes({
+                "query.type": req.type,
+                "query.from": req.from,
+                "query.to": req.to,
+                "query.rows": data.length,
+                "query.duration_ms": Math.round(performance.now() - startTime),
+            });
+
+            return { type: req.type, data };
+        } catch (e) {
+            const error = e instanceof Error ? e.message : "Query failed";
+            setAttributes({ "query.error": error });
+            return { type: req.type, data: [], error };
+        }
+    });
 }
 
-function groupBySchema(requests: BatchRequest[]): Map<string, BatchRequest[]> {
-    const groups = new Map<string, BatchRequest[]>();
+function groupBySchema(requests: BatchRequest[]): Map<string, { index: number; req: BatchRequest }[]> {
+    const groups = new Map<string, { index: number; req: BatchRequest }[]>();
+
+    for (let i = 0; i < requests.length; i++) {
+        const req = requests[i];
+        if (!req) {
+            continue;
+        }
 
-    for (const req of requests) {
         const config = QueryBuilders[req.type];
         if (!config) {
             continue;
         }
 
         const sig = getSchemaSignature(config) || `__solo_${req.type}`;
         const list = groups.get(sig) || [];
-        list.push(req);
+        list.push({ index: i, req });
         groups.set(sig, list);
     }
 
     return groups;
 }
 
-function buildUnionQuery(requests: BatchRequest[], opts?: BatchOptions) {
+function buildUnionQuery(items: { index: number; req: BatchRequest }[], opts?: BatchOptions) {
     const queries: string[] = [];
     const params: Record<string, unknown> = {};
-    const types: string[] = [];
-
-    for (let i = 0; i < requests.length; i++) {
-        const req = requests[i];
-        if (!req) {
-            continue;
-        }
+    const indices: number[] = [];
 
+    for (const { index, req } of items) {
         const config = QueryBuilders[req.type];
         if (!config) {
             continue;
@@ -74,74 +90,104 @@ function buildUnionQuery(requests: BatchRequest[], opts?: BatchOptions) {
         let { sql, params: queryParams } = builder.compile();
 
         for (const [key, value] of Object.entries(queryParams)) {
-            const prefixedKey = `q${i}_${key}`;
+            const prefixedKey = `q${index}_${key}`;
             params[prefixedKey] = value;
             sql = sql.replaceAll(`{${key}:`, `{${prefixedKey}:`);
         }
 
-        types.push(req.type);
-        queries.push(`SELECT '${req.type}' as __query_type, * FROM (${sql})`);
+        indices.push(index);
+        queries.push(`SELECT ${index} as __query_idx, * FROM (${sql})`);
     }
 
-    return { sql: queries.join("\nUNION ALL\n"), params, types };
+    return { sql: queries.join("\nUNION ALL\n"), params, indices };
 }
 
 function splitResults(
-    rows: Array<Record<string, unknown> & { __query_type: string }>,
-    types: string[]
-): Map<string, Record<string, unknown>[]> {
-    const byType = new Map<string, Record<string, unknown>[]>(types.map((t) => [t, []]));
+    rows: Array<Record<string, unknown> & { __query_idx: number }>,
+    indices: number[]
+): Map<number, Record<string, unknown>[]> {
+    const byIndex = new Map<number, Record<string, unknown>[]>(indices.map((i) => [i, []]));
 
-    for (const { __query_type, ...rest } of rows) {
-        byType.get(__query_type)?.push(rest);
+    for (const { __query_idx, ...rest } of rows) {
+        byIndex.get(__query_idx)?.push(rest);
     }
 
-    return byType;
+    return byIndex;
 }
 
-export async function executeBatch(requests: BatchRequest[], opts?: BatchOptions): Promise<BatchResult[]> {
+export function executeBatch(requests: BatchRequest[], opts?: BatchOptions): Promise<BatchResult[]> {
     if (requests.length === 0) {
-        return [];
-    }
-    if (requests.length === 1 && requests[0]) {
-        return [await runSingle(requests[0], opts)];
+        return Promise.resolve([]);
     }
 
-    const groups = groupBySchema(requests);
-    const results: BatchResult[] = [];
+    return record("query.batch", async () => {
+        const startTime = performance.now();
 
-    for (const groupReqs of groups.values()) {
-        if (groupReqs.length === 0) {
-            continue;
-        }
+        setAttributes({
+            "batch.size": requests.length,
+            "batch.types": requests.map((r) => r.type).join(","),
+        });
 
-        if (groupReqs.length === 1 && groupReqs[0]) {
-            results.push(await runSingle(groupReqs[0], opts));
-            continue;
+        if (requests.length === 1 && requests[0]) {
+            return [await runSingle(requests[0], opts)];
         }
 
-        try {
-            const { sql, params, types } = buildUnionQuery(groupReqs, opts);
-            const rawRows = await chQuery(sql, params);
-            const split = splitResults(rawRows as Array<Record<string, unknown> & { __query_type: string }>, types);
-
-            for (const type of types) {
-                const config = QueryBuilders[type];
-                const raw = split.get(type) || [];
-                results.push({
-                    type,
-                    data: config ? applyPlugins(raw, config, opts?.websiteDomain) : raw,
-                });
+        const groups = groupBySchema(requests);
+        const results: BatchResult[] = Array.from({ length: requests.length });
+        let unionCount = 0;
+        let singleCount = 0;
+
+        for (const groupItems of groups.values()) {
+            if (groupItems.length === 0) {
+                continue;
             }
-        } catch {
-            for (const req of groupReqs) {
-                results.push(await runSingle(req, opts));
+
+            if (groupItems.length === 1 && groupItems[0]) {
+                const { index, req } = groupItems[0];
+                results[index] = await runSingle(req, opts);
+                singleCount += 1;
+                continue;
+            }
+
+            try {
+                const { sql, params, indices } = buildUnionQuery(groupItems, opts);
+                const queryStart = performance.now();
+                const rawRows = await chQuery(sql, params);
+                const queryDuration = Math.round(performance.now() - queryStart);
+
+                setAttributes({
+                    "batch.union.query_count": indices.length,
+                    "batch.union.rows": rawRows.length,
+                    "batch.union.duration_ms": queryDuration,
+                });
+
+                const split = splitResults(rawRows as Array<Record<string, unknown> & { __query_idx: number }>, indices);
+
+                for (const { index, req } of groupItems) {
+                    const config = QueryBuilders[req.type];
+                    const raw = split.get(index) || [];
+                    results[index] = {
+                        type: req.type,
+                        data: config ? applyPlugins(raw, config, opts?.websiteDomain) : raw,
+                    };
+                }
+                unionCount += 1;
+            } catch {
+                for (const { index, req } of groupItems) {
+                    results[index] = await runSingle(req, opts);
+                    singleCount += 1;
+                }
             }
         }
-    }
 
-    const resultMap = new Map(results.map((r) => [r.type, r]));
-    return requests.map((req) => resultMap.get(req.type) || { type: req.type, data: [] });
+        setAttributes({
+            "batch.union_groups": unionCount,
+            "batch.single_queries": singleCount,
+            "batch.duration_ms": Math.round(performance.now() - startTime),
+        });
+
+        return results.map((r, i) => r || { type: requests[i]?.type || "unknown", data: [] });
+    });
 }
 
 export function areQueriesCompatible(type1: string, type2: string): boolean {