feat(zero-throughput): adapt for CloudZero with custom queries and transform endpoint

apps/zero-throughput/README.md

@@ -6,7 +6,7 @@ The default run:
 
 1. Starts a dedicated PostgreSQL 16 Docker container on port `6436`.
 2. Resets the benchmark table and Zero metadata for app id `zero_throughput`.
-3. Deploys allow-read permissions for the benchmark table.
+3. Starts the Fastify app server on port `3000` for query transformations.
 4. Starts `zero-cache` on port `4848`.
 5. Runs analyze-query for each distinct live query shape in the selected profile.
 6. Starts synthetic Zero clients with live queries for the selected profile.
@@ -294,11 +294,12 @@ pnpm --filter zero-throughput run sweep:write-rates -- \
 
 ### Options Reference
 
-| CLI Option            | Environment Variable         | Default                 | Description                                                        |
-| :-------------------- | :--------------------------- | :---------------------- | :----------------------------------------------------------------- |
-| `--cache-url <url>`   | `ZERO_THROUGHPUT_CACHE_URL`  | `http://127.0.0.1:4848` | Primary Zero cache endpoint or load balancer (disables local Zero) |
-| `--cache-urls <urls>` | `ZERO_THROUGHPUT_CACHE_URLS` | `undefined`             | Comma-separated View-Syncer URLs for client partitioning           |
-| `--pg-url <url>`      | `ZERO_THROUGHPUT_PG_URL`     | `postgresql://...:6436` | Upstream database connection string (disables local Postgres)      |
-| `--reset <bool>`      | `ZERO_THROUGHPUT_RESET`      | `true`                  | When `false`, skips dropping/resetting the benchmark table         |
+| CLI Option                | Environment Variable              | Default                 | Description                                                        |
+| :------------------------ | :-------------------------------- | :---------------------- | :----------------------------------------------------------------- |
+| `--cache-url <url>`       | `ZERO_THROUGHPUT_CACHE_URL`       | `http://127.0.0.1:4848` | Primary Zero cache endpoint or load balancer (disables local Zero) |
+| `--cache-urls <urls>`     | `ZERO_THROUGHPUT_CACHE_URLS`      | `undefined`             | Comma-separated View-Syncer URLs for client partitioning           |
+| `--pg-url <url>`          | `ZERO_THROUGHPUT_PG_URL`          | `postgresql://...:6436` | Upstream database connection string (disables local Postgres)      |
+| `--app-server-port <num>` | `ZERO_THROUGHPUT_APP_SERVER_PORT` | `3000`                  | Local query-transform app server port                              |
+| `--reset <bool>`          | `ZERO_THROUGHPUT_RESET`           | `true`                  | When `false`, skips dropping/resetting the benchmark table         |
 
 Run `pnpm --filter zero-throughput start -- --help` for all options.

apps/zero-throughput/api/index.ts

@@ -0,0 +1,88 @@
+import {mustGetQuery, type ReadonlyJSONValue} from '@rocicorp/zero';
+import {
+  handleQueryRequest,
+  type QueryRequestHandler,
+} from '@rocicorp/zero/server';
+import Fastify, {type FastifyReply, type FastifyRequest} from 'fastify';
+import {queries} from '../src/queries.ts';
+import {schema} from '../src/schema.ts';
+
+export const fastify = Fastify({
+  logger: process.env.NODE_ENV !== 'test',
+});
+
+fastify.get('/health', (_req, reply) => {
+  reply.send({status: 'ok'});
+});
+
+fastify.get('/', (_req, reply) => {
+  reply.send({status: 'ok', service: 'zero-throughput-api'});
+});
+
+fastify.post('/api/push', mutateHandler);
+fastify.post('/api/mutate', mutateHandler);
+
+function mutateHandler(_request: FastifyRequest, reply: FastifyReply) {
+  reply.status(501).send({error: 'Mutations not supported'});
+}
+
+fastify.post<{
+  Querystring: Record<string, string>;
+  Body: ReadonlyJSONValue;
+}>('/api/get-queries', queryHandler);
+
+fastify.post<{
+  Querystring: Record<string, string>;
+  Body: ReadonlyJSONValue;
+}>('/api/query', queryHandler);
+
+type AnyQuery = ReturnType<QueryRequestHandler>;
+
+const queryTransformHandler: QueryRequestHandler = (name, args) => {
+  const query = mustGetQuery(queries, name);
+  return query.fn({args, ctx: undefined}) as unknown as AnyQuery;
+};
+
+function extractUserID(
+  headers: Record<string, string | string[] | undefined>,
+  query: Record<string, string>,
+): string | undefined {
+  const authHeader = headers['authorization'];
+  if (typeof authHeader === 'string') {
+    if (authHeader.startsWith('Bearer ')) {
+      return authHeader.slice('Bearer '.length).trim();
+    }
+    return authHeader.trim();
+  }
+  return (
+    (headers['x-user-id'] as string | undefined) ?? query.userID ?? undefined
+  );
+}
+
+async function queryHandler(
+  request: FastifyRequest<{
+    Querystring: Record<string, string>;
+    Body: ReadonlyJSONValue;
+  }>,
+  reply: FastifyReply,
+) {
+  const authUserID = extractUserID(request.headers, request.query);
+
+  const response = await handleQueryRequest({
+    handler: queryTransformHandler,
+    schema,
+    query: request.query,
+    body: request.body,
+    userID: authUserID,
+    logLevel: 'info',
+  });
+  reply.send(response);
+}
+
+export default async function handler(
+  req: FastifyRequest,
+  reply: FastifyReply,
+) {
+  await fastify.ready();
+  fastify.server.emit('request', req, reply);
+}

apps/zero-throughput/package.json

@@ -12,6 +12,8 @@
     "sweep:num-view-syncers": "node src/linear-sweep.ts --topology distributed --num-view-syncers 1,2,3 --sync-workers 2 --write-rates 300 --users 12 --profiles forum --duration-ms 15000",
     "sweep:users": "node src/linear-sweep.ts --users 5,10,20,50 --sync-workers 2 --write-rates 200 --profiles forum --duration-ms 15000",
     "go": "node src/main.ts",
+    "server": "node src/server.ts",
+    "start:server": "node src/server.ts",
     "db-up": "cd docker && docker compose up",
     "db-down": "cd docker && docker compose down",
     "check-types": "tsc",
@@ -24,8 +26,10 @@
   "dependencies": {
     "@dotenvx/dotenvx": "^1.39.0",
     "@rocicorp/zero": "workspace:*",
+    "fastify": "^5.0.0",
     "postgres": "3.4.7",
-    "ws": "^8.18.1"
+    "ws": "^8.18.1",
+    "zod": "^4.1.11"
   },
   "devDependencies": {
     "@types/node": "^22.10.5",

apps/zero-throughput/src/analyze.ts

@@ -1,7 +1,6 @@
 import {mapAST, type AST} from '../../../packages/zero-protocol/src/ast.ts';
 import {clientToServer} from '../../../packages/zero-schema/src/name-mapper.ts';
 import {runAnalyzeCLI} from '../../../packages/zero/src/analyze.ts';
-import {createBuilder} from '../../../packages/zql/src/query/create-builder.ts';
 import type {BenchmarkModel, BenchmarkProfile} from './config.ts';
 import {
   buildProfileQuery,
@@ -53,7 +52,6 @@ if (config.help) {
 } else {
   const {profile, queryIndex} = resolveProfileQuery(config);
   const {name, query} = buildProfileQuery(
-    createBuilder(schema),
     profile,
     config.model,
     queryIndex,
@@ -176,10 +174,23 @@ function parseArgs(argv: readonly string[]): AnalyzeConfig {
 }
 
 function queryAST(query: unknown): AST {
-  if (query === null || typeof query !== 'object' || !('ast' in query)) {
-    throw new Error('Profile query did not expose an AST');
+  if (query !== null && typeof query === 'object') {
+    if ('ast' in query) {
+      return (query as {readonly ast: AST}).ast;
+    }
+    if (
+      'query' in query &&
+      typeof (query as {query: unknown}).query === 'function' &&
+      'fn' in (query as {query: {fn?: unknown}}).query
+    ) {
+      const q = query as {
+        args: unknown;
+        query: {fn: (opts: {args: unknown; ctx: unknown}) => {ast: AST}};
+      };
+      return q.query.fn({args: q.args, ctx: undefined}).ast;
+    }
   }
-  return (query as {readonly ast: AST}).ast;
+  throw new Error('Profile query did not expose an AST');
 }
 
 function parseOption(arg: string): {

apps/zero-throughput/src/client.ts

@@ -62,6 +62,7 @@ export class SyntheticClient {
       schema,
       cacheURL: targetCacheURL,
       userID,
+      auth: userID,
       storageKey: `${config.runID}-${clientIndex}`,
       kvStore: 'mem',
       logLevel: 'error',
@@ -81,7 +82,6 @@ export class SyntheticClient {
 
   #registerProfileQuery(config: BenchmarkConfig, queryIndex: number): void {
     const {name, query} = buildProfileQuery(
-      this.#zero.query,
       config.profile,
       config.model,
       queryIndex,

apps/zero-throughput/src/config.ts

@@ -39,6 +39,7 @@ const options = {
   reset: v.boolean().default(true),
   cacheURL: v.string().optional(),
   cacheURLs: v.string().optional(),
+  appServerPort: v.number().default(3_000),
 
   topology: v.literalUnion('single', 'distributed').default('single'),
   numViewSyncers: v.number().default(1),
@@ -97,6 +98,7 @@ export type BenchmarkConfig = {
   readonly profileVS: boolean;
   readonly processLogMode: 'file' | 'inherit' | 'ignore';
   readonly reset: boolean;
+  readonly appServerPort: number;
   readonly cacheURL: string;
   readonly cacheURLs: readonly string[];
   readonly pg: {
@@ -196,6 +198,7 @@ export function loadConfig(): BenchmarkConfig {
     profileVS: parsed.profileVS,
     processLogMode: parsed.processLogMode,
     reset: parsed.reset,
+    appServerPort: parsed.appServerPort,
     cacheURL: cacheURLs[0],
     cacheURLs,
     pg: {

apps/zero-throughput/src/main.ts

@@ -9,12 +9,13 @@ import {
 import {OTelMetricsCollector} from './metrics.ts';
 import {
   analyzeProfileQueries,
-  deployPermissions,
   queryPlanAnalysisLogPath,
   removeReplicaFiles,
+  startAppServer,
   startPostgres,
   startZeroTopology,
   stopPostgres,
+  waitForAppServer,
   waitForZeroCache,
   type ProcessCommand,
 } from './processes.ts';
@@ -76,8 +77,16 @@ async function main(): Promise<void> {
       await removeReplicaFiles(config.zero.replicaFile);
     }
 
-    log('Deploying benchmark permissions...');
-    processes.push(await deployPermissions(config));
+    if (config.zero.start) {
+      log(`Starting app server on port ${config.appServerPort}...`);
+      const appServer = startAppServer(config);
+      processes.push(appServer);
+      cleanup.push(() => appServer.stop());
+      if (appServer.logPath !== undefined) {
+        log(`app-server logs: ${appServer.logPath}`);
+      }
+      await waitForAppServer(config.appServerPort, 30_000, appServer);
+    }
 
     if (config.zero.start) {
       log(