feat(cruncher): disk-composed queries with standalone server package

.changeset/disk-composed-queries.md

@@ -0,0 +1,8 @@
+---
+"cruncher": minor
+"@cruncher/qql": patch
+"@cruncher/adapter-docker": patch
+"@cruncher/adapter-grafana-loki-browser": patch
+---
+
+Engine revamped with disk-composed queries: the server process is now a standalone package (`cruncher-server`) with shared types in `@cruncher/server-shared`, enabling better separation of concerns and disk-based query composition.

.claude/skills/create-adapter/SKILL.md

@@ -244,7 +244,7 @@ Fix any TypeScript errors before moving on.
 "@cruncher/adapter-<name>": "workspace:*"
 ```
 
-**`apps/cruncher/src/processes/server/main.ts`** — add import and registration:
+**`apps/cruncher-server/src/main.ts`** — add import and registration:
 ```ts
 import myAdapter from "@cruncher/adapter-<name>";
 // ...inside initializeServer():
@@ -297,7 +297,7 @@ Stage only the adapter-related files:
 ```bash
 git add packages/adapters/<name>/ \
         apps/cruncher/package.json \
-        apps/cruncher/src/processes/server/main.ts \
+        apps/cruncher-server/src/main.ts \
         docs/src/content/docs/adapters/<name>.mdx \
         docs/src/content/docs/adapters/examples/<name>*.yaml \
         .changeset/<changeset-file>.md \

.github/workflows/checks.yml

@@ -39,3 +39,14 @@ jobs:
 
       - name: Run lint
         run: pnpm lint
+
+  knip:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: pnpm/action-setup@v4
+      - name: Install dependencies
+        run: pnpm install --frozen-lockfile
+
+      - name: Run knip
+        run: pnpm knip

.mise.toml

@@ -1,4 +1,11 @@
 [tools]
+ast-grep = "latest"
+difftastic = "latest"
+fd = "latest"
+hyperfine = "latest"
 java = "latest"
 node = "lts"
 pnpm = "latest"
+shellcheck = "latest"
+watchexec = "latest"
+yq = "latest"

.npmrc

@@ -6,9 +6,10 @@ auto-install-peers = true
 strict-peer-dependencies = false
 prefer-workspace-packages = true
 link-workspace-packages = deep
+inject-workspace-packages = true
 hoist = true
 shamefully-hoist = true
 hoist-workspace-packages = false
 package-manager-strict=false
 # https://github.com/pnpm/pnpm/issues/7024
-package-import-method=clone-or-copy
+package-import-method=clone-or-copy

apps/cruncher-server/drizzle.config.ts

@@ -0,0 +1,7 @@
+import { defineConfig } from "drizzle-kit";
+
+export default defineConfig({
+  schema: "./src/engineV2/localstate/schema.ts",
+  out: "./src/engineV2/localstate/migrations",
+  dialect: "sqlite",
+});

apps/cruncher-server/package.json

@@ -0,0 +1,60 @@
+{
+  "name": "cruncher-server",
+  "private": true,
+  "version": "0.0.0",
+  "main": "dist/server.js",
+  "files": [
+    "dist"
+  ],
+  "exports": {
+    ".": {
+      "default": "./dist/server.js"
+    },
+    "./router_messages": {
+      "types": "./src/router_messages.d.ts"
+    }
+  },
+  "scripts": {
+    "build": "vite build --config vite.server.config.mts",
+    "build:server": "vite build --config vite.server.config.mts",
+    "build:watch": "vite build --config vite.server.config.mts --watch",
+    "typecheck": "tsc -p tsconfig.node.json --noEmit"
+  },
+  "dependencies": {
+    "@cruncher/adapter-coralogix": "workspace:*",
+    "@cruncher/adapter-datadog": "workspace:*",
+    "@cruncher/adapter-docker": "workspace:*",
+    "@cruncher/adapter-grafana-loki-browser": "workspace:*",
+    "@cruncher/adapter-k8s": "workspace:*",
+    "@cruncher/adapter-loki": "workspace:*",
+    "@cruncher/adapter-mock": "workspace:*",
+    "@cruncher/adapter-utils": "workspace:*",
+    "@cruncher/qql": "workspace:*",
+    "@cruncher/server-shared": "workspace:*",
+    "@cruncher/utils": "workspace:*",
+    "@duckdb/node-api": "^1.5.0-r.1",
+    "@trpc/server": "^11.12.0",
+    "apache-arrow": "^18.0.0",
+    "async-mutex": "^0.5.0",
+    "better-sqlite3": "^12.6.2",
+    "date-fns": "^4.1.0",
+    "drizzle-orm": "^0.45.1",
+    "electron-log": "^5.4.3",
+    "export-to-csv": "^1.4.0",
+    "merge-k-sorted-arrays": "^2.1.0",
+    "object-hash": "^3.0.0",
+    "uuid": "^13.0.0",
+    "winston": "^3.19.0",
+    "winston-daily-rotate-file": "^5.0.0",
+    "ws": "^8.19.0",
+    "yaml": "^2.8.2",
+    "yauzl": "^3.2.0",
+    "zod": "^4.3.6"
+  },
+  "devDependencies": {
+    "electron": "^40.8.0",
+    "typescript": "^5.9.3",
+    "vite": "^7.3.1",
+    "vite-tsconfig-paths": "^6.1.1"
+  }
+}

apps/cruncher-server/src/engineV2/duckdb/IQueryBackend.ts

@@ -0,0 +1,55 @@
+export interface IQueryBackend {
+  // SubTask — per-adapter in-memory DuckDB (write buffer for Parquet chunks)
+  openSubtask(subtaskId: string, dataDir: string): Promise<void>;
+  closeSubtask(subtaskId: string): Promise<void>;
+  writeBatch(subtaskId: string, batch: Uint8Array): Promise<{ chunkPath: string; rowCount: number; diskBytes: number; minTime: number; maxTime: number }>;
+
+  // Read raw chunks (time-range filter only)
+  readChunks(
+    chunkPaths: string[],
+    fromTime: number,
+    toTime: number,
+  ): Promise<Record<string, unknown>[]>;
+
+  // Write pipeline-processed results to a dedicated Parquet file
+  writeTaskResult(
+    arrowBytes: Uint8Array,
+    outputPath: string,
+  ): Promise<{ rowCount: number; diskBytes: number }>;
+
+  // Paginate a result Parquet file
+  paginateFile(
+    filePath: string,
+    offset: number,
+    limit: number,
+  ): Promise<{ rows: Record<string, unknown>[]; total: number }>;
+
+  // Find the closest event by timestamp in a result Parquet file
+  getClosestEvent(
+    filePath: string,
+    refTime: number,
+  ): Promise<{ closest: number; rowIndex: number } | null>;
+
+  // Compute a time-based histogram over chunk files (pure SQL aggregation, no row transfer)
+  computeHistogram(
+    chunkPaths: string[],
+    fromTime: number,
+    toTime: number,
+    numBuckets: number,
+  ): Promise<{ timestamp: number; count: number }[]>;
+
+  // Collect all field names present in chunk files (fixed schema + json_keys of fields blob)
+  getColumnNames(
+    chunkPaths: string[],
+    fromTime: number,
+    toTime: number,
+  ): Promise<string[]>;
+
+
+  /** Wait for all in-flight worker operations to complete. */
+  drain(): Promise<void>;
+  terminate(): void;
+
+  /** Returns current worker health and load stats. */
+  getStats(): { inflightOps: number; pendingOps: number; isDead: boolean; totalOpsDispatched: number };
+}

apps/cruncher-server/src/engineV2/duckdb/LocalDuckDBBackend.ts

@@ -0,0 +1,173 @@
+import { Worker, Transferable } from "node:worker_threads";
+import path from "node:path";
+import { IQueryBackend } from "./IQueryBackend";
+
+type PendingEntry = {
+  resolve: (v: unknown) => void;
+  reject: (e: unknown) => void;
+};
+
+type WorkerReply =
+  | { id: string; type: "ok" }
+  | { id: string; type: "chunk"; chunkPath: string; rowCount: number; diskBytes: number; minTime: number; maxTime: number }
+  | { id: string; type: "rows"; rows: Record<string, unknown>[] }
+  | { id: string; type: "taskResultWritten"; rowCount: number; diskBytes: number }
+  | { id: string; type: "paginated"; rows: Record<string, unknown>[]; total: number }
+  | { id: string; type: "closestEvent"; result: { closest: number; rowIndex: number } | null }
+  | { id: string; type: "histogram"; buckets: { timestamp: number; count: number }[] }
+  | { id: string; type: "columnNames"; columns: string[] }
+  | { id: string; type: "error"; message: string };
+
+export class LocalDuckDBBackend implements IQueryBackend {
+  private worker: Worker;
+  private pending = new Map<string, PendingEntry>();
+  private nextId = 0;
+  private dead = false;
+  private _inflightOps = 0;
+  private _totalOpsDispatched = 0;
+  private _drainCallbacks: Array<() => void> = [];
+
+  constructor(workerScriptPath?: string) {
+    const workerPath = workerScriptPath ?? resolveDefaultWorkerPath();
+    const workerOpts = workerPath.endsWith(".ts")
+      ? { execArgv: ["--import", "tsx"] }
+      : {};
+    this.worker = new Worker(workerPath, workerOpts);
+
+    this.worker.on("message", (msg: WorkerReply) => {
+      const entry = this.pending.get(msg.id);
+      if (!entry) return;
+      this.pending.delete(msg.id);
+      this._inflightOps--;
+      this._notifyDrain();
+      if (msg.type === "error") {
+        entry.reject(new Error(msg.message));
+      } else {
+        entry.resolve(msg);
+      }
+    });
+
+    this.worker.on("error", (err) => {
+      this.dead = true;
+      this._inflightOps = 0;
+      this._notifyDrain();
+      for (const entry of this.pending.values()) {
+        entry.reject(err);
+      }
+      this.pending.clear();
+    });
+  }
+
+  private send<T>(payload: Record<string, unknown>, transfer?: Transferable[]): Promise<T> {
+    if (this.dead) return Promise.reject(new Error("DuckDB worker has crashed"));
+    this._inflightOps++;
+    this._totalOpsDispatched++;
+    const id = String(++this.nextId);
+    return new Promise<T>((resolve, reject) => {
+      this.pending.set(id, { resolve: resolve as (v: unknown) => void, reject });
+      this.worker.postMessage({ ...payload, id }, transfer ?? []);
+    });
+  }
+
+  private _notifyDrain(): void {
+    if (this._inflightOps === 0) {
+      const callbacks = this._drainCallbacks.splice(0);
+      for (const cb of callbacks) cb();
+    }
+  }
+
+  /** Waits for all in-flight worker operations to complete. */
+  drain(): Promise<void> {
+    if (this._inflightOps === 0) return Promise.resolve();
+    return new Promise<void>((resolve) => this._drainCallbacks.push(resolve));
+  }
+
+  async openSubtask(subtaskId: string, dataDir: string): Promise<void> {
+    await this.send({ type: "openSubtask", subtaskId, dataDir });
+  }
+
+  async closeSubtask(subtaskId: string): Promise<void> {
+    await this.send({ type: "closeSubtask", subtaskId });
+  }
+
+  async writeBatch(subtaskId: string, batch: Uint8Array): Promise<{ chunkPath: string; rowCount: number; diskBytes: number; minTime: number; maxTime: number }> {
+    const reply = await this.send<{ type: "chunk"; chunkPath: string; rowCount: number; diskBytes: number; minTime: number; maxTime: number }>(
+      { type: "writeBatch", subtaskId, batch },
+      [batch.buffer as ArrayBuffer],
+    );
+    return { chunkPath: reply.chunkPath, rowCount: reply.rowCount, diskBytes: reply.diskBytes, minTime: reply.minTime, maxTime: reply.maxTime };
+  }
+
+  async readChunks(chunkPaths: string[], fromTime: number, toTime: number): Promise<Record<string, unknown>[]> {
+    if (chunkPaths.length === 0) return [];
+    const reply = await this.send<{ type: "rows"; rows: Record<string, unknown>[] }>({
+      type: "readChunks", chunkPaths, fromTime, toTime,
+    });
+    return reply.rows;
+  }
+
+  async writeTaskResult(arrowBytes: Uint8Array, outputPath: string): Promise<{ rowCount: number; diskBytes: number }> {
+    const reply = await this.send<{ type: "taskResultWritten"; rowCount: number; diskBytes: number }>(
+      { type: "writeTaskResult", arrowBytes, outputPath },
+      [arrowBytes.buffer as ArrayBuffer],
+    );
+    return { rowCount: reply.rowCount, diskBytes: reply.diskBytes };
+  }
+
+  async paginateFile(filePath: string, offset: number, limit: number): Promise<{ rows: Record<string, unknown>[]; total: number }> {
+    const reply = await this.send<{ type: "paginated"; rows: Record<string, unknown>[]; total: number }>({
+      type: "paginateFile", filePath, offset, limit,
+    });
+    return { rows: reply.rows, total: reply.total };
+  }
+
+  async computeHistogram(chunkPaths: string[], fromTime: number, toTime: number, numBuckets: number): Promise<{ timestamp: number; count: number }[]> {
+    if (chunkPaths.length === 0) return [];
+    const reply = await this.send<{ type: "histogram"; buckets: { timestamp: number; count: number }[] }>({
+      type: "computeHistogram", chunkPaths, fromTime, toTime, numBuckets,
+    });
+    return reply.buckets;
+  }
+
+  async getColumnNames(chunkPaths: string[], fromTime: number, toTime: number): Promise<string[]> {
+    if (chunkPaths.length === 0) return ["_time", "message"];
+    const reply = await this.send<{ type: "columnNames"; columns: string[] }>({
+      type: "getColumnNames", chunkPaths, fromTime, toTime,
+    });
+    return reply.columns;
+  }
+
+
+  async getClosestEvent(filePath: string, refTime: number): Promise<{ closest: number; rowIndex: number } | null> {
+    const reply = await this.send<{ type: "closestEvent"; result: { closest: number; rowIndex: number } | null }>({
+      type: "getClosestEvent", filePath, refTime,
+    });
+    return reply.result;
+  }
+
+  getStats(): { inflightOps: number; pendingOps: number; isDead: boolean; totalOpsDispatched: number } {
+    return { inflightOps: this._inflightOps, pendingOps: this.pending.size, isDead: this.dead, totalOpsDispatched: this._totalOpsDispatched };
+  }
+
+  terminate(): void {
+    this.dead = true;
+    this._inflightOps = 0;
+    this._notifyDrain();
+    this.worker.terminate();
+    for (const entry of this.pending.values()) {
+      entry.reject(new Error("DuckDB worker terminated"));
+    }
+    this.pending.clear();
+  }
+}
+
+function resolveDefaultWorkerPath(): string {
+  // In Electron asar bundles, the worker script is in the unpacked directory.
+  if (__dirname.includes(".asar")) {
+    return path.join(
+      __dirname.replace(/\.asar([/\\])/, ".asar.unpacked$1"),
+      "duckdb-worker.js",
+    );
+  }
+  return path.join(__dirname, "duckdb-worker.js");
+}