feat: add Zod validation for query descriptors at API boundary

Replace unsafe `as` casts in query-do.ts parseQuery() with Zod schema
validation. All POST endpoints (/query, /query/count, /query/exists,
/query/first, /query/explain, /query/stream) now validate request bodies
and return 400 with clear error messages on malformed input.

Validated fields: table (required string), filters (op enum, column
non-empty), aggregates (fn enum), sortDirection (asc|desc), limit
(positive int), offset (non-negative int), cacheTTL (positive int),
vectorSearch (topK positive int).

Author: teamchong

package.json

@@ -40,5 +40,8 @@
     "src/wasm/querymode.wasm"
   ],
   "license": "MIT",
-  "packageManager": "pnpm@10.21.0"
+  "packageManager": "pnpm@10.21.0",
+  "dependencies": {
+    "zod": "^4.3.6"
+  }
 }

pnpm-lock.yaml

@@ -11,6 +11,7 @@ import { mergeQueryResults } from "./merge.js";
 import { coalesceRanges, fetchBounded, withRetry, withTimeout } from "./coalesce.js";
 import { VipCache } from "./vip-cache.js";
 import { parseLanceV2Columns } from "./lance-v2.js";
+import { parseAndValidateQuery } from "./query-schema.js";
 import wasmModule from "./wasm-module.js";
 
 const FRAGMENT_POOL_MAX = 20; // Max Fragment DO slots per datacenter (idle slots cost nothing)
@@ -246,11 +247,13 @@ export class QueryDO implements DurableObject {
 
   private async handleQuery(request: Request): Promise<Response> {
     const requestId = request.headers.get("x-querymode-request-id") ?? crypto.randomUUID();
-    const body = await request.json() as Record<string, unknown>;
-    if (!body.table || typeof body.table !== "string") {
-      return this.json({ error: "Missing or invalid 'table' field" }, 400);
+    const body = await request.json();
+    let query: QueryDescriptor;
+    try {
+      query = this.parseQuery(body);
+    } catch (err) {
+      return this.json({ error: (err as Error).message }, 400);
     }
-    const query = this.parseQuery(body);
     try {
       const result = await this.executeQuery(query);
       result.requestId = requestId;
@@ -291,25 +294,14 @@ export class QueryDO implements DurableObject {
     return `qr:${query.table}:${(h >>> 0).toString(36)}`;
   }
 
-  private parseQuery(request_body: Record<string, unknown>): QueryDescriptor {
-    return {
-      table: request_body.table as string,
-      filters: (request_body.filters ?? []) as QueryDescriptor["filters"],
-      projections: (request_body.projections ?? request_body.select ?? []) as string[],
-      sortColumn: request_body.sortColumn as string | undefined,
-      sortDirection: request_body.sortDirection as "asc" | "desc" | undefined,
-      limit: request_body.limit as number | undefined,
-      vectorSearch: request_body.vectorSearch as QueryDescriptor["vectorSearch"],
-      aggregates: request_body.aggregates as QueryDescriptor["aggregates"],
-      groupBy: request_body.groupBy as string[] | undefined,
-      cacheTTL: request_body.cacheTTL as number | undefined,
-    };
+  private parseQuery(body: unknown): QueryDescriptor {
+    return parseAndValidateQuery(body) as QueryDescriptor;
   }
 
   private async handleCount(request: Request): Promise<Response> {
-    const body = await request.json() as Record<string, unknown>;
-    if (!body.table || typeof body.table !== "string") return this.json({ error: "Missing 'table'" }, 400);
-    const query = this.parseQuery(body);
+    const body = await request.json();
+    let query: QueryDescriptor;
+    try { query = this.parseQuery(body); } catch (err) { return this.json({ error: (err as Error).message }, 400); }
     try {
       // Fast path: no filters — sum page rowCounts from cached metadata
       if (query.filters.length === 0) {
@@ -331,9 +323,9 @@ export class QueryDO implements DurableObject {
   }
 
   private async handleExists(request: Request): Promise<Response> {
-    const body = await request.json() as Record<string, unknown>;
-    if (!body.table || typeof body.table !== "string") return this.json({ error: "Missing 'table'" }, 400);
-    const query = this.parseQuery(body);
+    const body = await request.json();
+    let query: QueryDescriptor;
+    try { query = this.parseQuery(body); } catch (err) { return this.json({ error: (err as Error).message }, 400); }
     query.limit = 1;
     try {
       const result = await this.executeQuery(query);
@@ -344,9 +336,9 @@ export class QueryDO implements DurableObject {
   }
 
   private async handleFirst(request: Request): Promise<Response> {
-    const body = await request.json() as Record<string, unknown>;
-    if (!body.table || typeof body.table !== "string") return this.json({ error: "Missing 'table'" }, 400);
-    const query = this.parseQuery(body);
+    const body = await request.json();
+    let query: QueryDescriptor;
+    try { query = this.parseQuery(body); } catch (err) { return this.json({ error: (err as Error).message }, 400); }
     query.limit = 1;
     try {
       const result = await this.executeQuery(query);
@@ -357,9 +349,9 @@ export class QueryDO implements DurableObject {
   }
 
   private async handleExplain(request: Request): Promise<Response> {
-    const body = await request.json() as Record<string, unknown>;
-    if (!body.table || typeof body.table !== "string") return this.json({ error: "Missing 'table'" }, 400);
-    const query = this.parseQuery(body);
+    const body = await request.json();
+    let query: QueryDescriptor;
+    try { query = this.parseQuery(body); } catch (err) { return this.json({ error: (err as Error).message }, 400); }
     try {
       let meta: TableMeta | undefined = this.footerCache.get(query.table);
       const metaCached = !!meta;
@@ -1358,7 +1350,9 @@ export class QueryDO implements DurableObject {
 
   /** Stream query results as NDJSON. */
   private async handleQueryStream(request: Request): Promise<Response> {
-    const query = (await request.json()) as QueryDescriptor;
+    const body = await request.json();
+    let query: QueryDescriptor;
+    try { query = this.parseQuery(body); } catch (err) { return this.json({ error: (err as Error).message }, 400); }
     const result = await this.executeQuery(query);
 
     const { readable, writable } = new TransformStream<Uint8Array>();

src/query-do.ts

@@ -0,0 +1,96 @@
+/**
+ * Zod schemas for query descriptor validation.
+ *
+ * Validates POST bodies at the API boundary (query-do.ts) before they
+ * reach the execution engine. Catches malformed filters, invalid ops,
+ * bad aggregate fns, etc. with clear error messages.
+ */
+import { z } from "zod/v4";
+
+const filterOpSchema = z.object({
+  column: z.string().min(1, "Filter column name cannot be empty"),
+  op: z.enum(["eq", "neq", "gt", "gte", "lt", "lte", "in"]),
+  value: z.union([
+    z.number(),
+    z.string(),
+    z.array(z.union([z.number(), z.string()])),
+  ]),
+});
+
+const aggregateOpSchema = z.object({
+  fn: z.enum(["sum", "avg", "min", "max", "count"]),
+  column: z.string().min(1, "Aggregate column name cannot be empty"),
+  alias: z.string().optional(),
+});
+
+const vectorSearchSchema = z.object({
+  column: z.string().min(1),
+  queryVector: z.union([
+    z.array(z.number()),
+    z.instanceof(Float32Array),
+  ]),
+  topK: z.number().int().positive(),
+});
+
+export const queryDescriptorSchema = z.object({
+  table: z.string().min(1, "Table name is required"),
+  filters: z.array(filterOpSchema).default([]),
+  projections: z.array(z.string()).default([]),
+  select: z.array(z.string()).optional(), // alias for projections
+  sortColumn: z.string().optional(),
+  sortDirection: z.enum(["asc", "desc"]).optional(),
+  limit: z.number().int().positive().optional(),
+  offset: z.number().int().nonnegative().optional(),
+  vectorSearch: vectorSearchSchema.optional(),
+  aggregates: z.array(aggregateOpSchema).optional(),
+  groupBy: z.array(z.string()).optional(),
+  cacheTTL: z.number().int().positive().optional(),
+});
+
+export type ValidatedQuery = z.infer<typeof queryDescriptorSchema>;
+
+/**
+ * Parse and validate a raw request body into a QueryDescriptor.
+ * Throws a formatted error string on validation failure.
+ */
+export function parseAndValidateQuery(body: unknown): {
+  table: string;
+  filters: { column: string; op: "eq" | "neq" | "gt" | "gte" | "lt" | "lte" | "in"; value: number | string | (number | string)[] }[];
+  projections: string[];
+  sortColumn?: string;
+  sortDirection?: "asc" | "desc";
+  limit?: number;
+  offset?: number;
+  vectorSearch?: { column: string; queryVector: number[] | Float32Array; topK: number };
+  aggregates?: { fn: "sum" | "avg" | "min" | "max" | "count"; column: string; alias?: string }[];
+  groupBy?: string[];
+  cacheTTL?: number;
+} {
+  const result = queryDescriptorSchema.safeParse(body);
+  if (!result.success) {
+    const issues = result.error.issues.map(i =>
+      `${i.path.join(".")}: ${i.message}`
+    ).join("; ");
+    throw new Error(`Invalid query: ${issues}`);
+  }
+
+  const data = result.data;
+  // Merge `select` alias into `projections`
+  const projections = data.projections.length > 0
+    ? data.projections
+    : (data.select ?? []);
+
+  return {
+    table: data.table,
+    filters: data.filters,
+    projections,
+    sortColumn: data.sortColumn,
+    sortDirection: data.sortDirection,
+    limit: data.limit,
+    offset: data.offset,
+    vectorSearch: data.vectorSearch,
+    aggregates: data.aggregates,
+    groupBy: data.groupBy,
+    cacheTTL: data.cacheTTL,
+  };
+}