feat: Mutator example

backend/package.json

@@ -20,7 +20,8 @@
     "mssql": "^12.1.0",
     "mysql2": "^3.11.3",
     "pg": "^8.11.3",
-    "winston": "^3.11.0"
+    "winston": "^3.11.0",
+    "zod": "^4.4.3"
   },
   "devDependencies": {
     "@types/express": "^5.0.6",

backend/src/api/index.ts

@@ -1,10 +1,12 @@
 import express from 'express';
 import { authRouter } from './auth.js';
 import { dataRouter } from './data.js';
+import { mutatorsRouter } from '../mutators/router.js';
 
 const router = express.Router();
 
 router.use('/auth', authRouter);
 router.use('/data', dataRouter);
+router.use('/mutators', mutatorsRouter);
 
 export { router as apiRouter };

backend/src/generated/api.ts

@@ -21,10 +21,44 @@ export interface paths {
         patch?: never;
         trace?: never;
     };
+    "/api/mutators/invoke": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        get?: never;
+        put?: never;
+        /** Invoke a server-side mutator by name */
+        post: operations["invokeMutator"];
+        delete?: never;
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
 }
 export type webhooks = Record<string, never>;
 export interface components {
     schemas: {
+        MutatorInvokeRequest: {
+            /** @description Registered mutator name */
+            name: string;
+            /** @description Mutator arguments. Validated server-side via the mutator's own zod schema. */
+            args: {
+                [key: string]: unknown;
+            };
+            /** @description Client-generated id for this invocation */
+            call_id: string;
+            /**
+             * Format: int64
+             * @description ps_crud transaction_id this invocation came from, for logging/idempotency.
+             */
+            transaction_id?: number;
+            /** @description Acting user id. Trusted from the body for v1 — replace with JWT-extracted id once auth is wired. */
+            user_id?: string;
+        };
         CrudTransaction: {
             crud: components["schemas"]["CrudEntry"][];
             /**
@@ -120,13 +154,28 @@ export interface operations {
                     "application/json": components["schemas"]["TransactionResponse"];
                 };
             };
-            /** @description Unexpected server error */
-            500: {
+        };
+    };
+    invokeMutator: {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        requestBody: {
+            content: {
+                "application/json": components["schemas"]["MutatorInvokeRequest"];
+            };
+        };
+        responses: {
+            /** @description Mutator result. Always returns 200 — the outcome is determined by the status field in the response body. */
+            200: {
                 headers: {
                     [name: string]: unknown;
                 };
                 content: {
-                    "application/json": components["schemas"]["MessageResponse"];
+                    "application/json": components["schemas"]["TransactionResponse"];
                 };
             };
         };

backend/src/mutators/mutators.ts

@@ -0,0 +1,82 @@
+import { z } from 'zod';
+import type { ServerMutator } from './types.js';
+
+const listCreateArgs = z.object({
+  id: z.string().uuid(),
+  name: z.string().min(1)
+});
+
+const listCreate: ServerMutator<typeof listCreateArgs> = {
+  args: listCreateArgs,
+  run: async (args, ctx) => {
+    await ctx.pg.query(
+      `INSERT INTO lists (id, name, owner_id) VALUES ($1, $2, $3)`,
+      [args.id, args.name, ctx.userId]
+    );
+  }
+};
+
+const listDeleteArgs = z.object({ id: z.string().uuid() });
+
+const listDelete: ServerMutator<typeof listDeleteArgs> = {
+  args: listDeleteArgs,
+  run: async (args, ctx) => {
+    await ctx.pg.query(`DELETE FROM todos WHERE list_id = $1`, [args.id]);
+    await ctx.pg.query(`DELETE FROM lists WHERE id = $1`, [args.id]);
+  }
+};
+
+const todoCreateArgs = z.object({
+  id: z.string().uuid(),
+  list_id: z.string().uuid(),
+  description: z.string().min(1)
+});
+
+const todoCreate: ServerMutator<typeof todoCreateArgs> = {
+  args: todoCreateArgs,
+  run: async (args, ctx) => {
+    await ctx.pg.query(
+      `INSERT INTO todos (id, list_id, created_by, description, completed) VALUES ($1, $2, $3, $4, false)`,
+      [args.id, args.list_id, ctx.userId, args.description]
+    );
+  }
+};
+
+const todoToggleArgs = z.object({
+  id: z.string().uuid(),
+  completed: z.boolean()
+});
+
+const todoToggle: ServerMutator<typeof todoToggleArgs> = {
+  args: todoToggleArgs,
+  run: async (args, ctx) => {
+    if (args.completed) {
+      await ctx.pg.query(
+        `UPDATE todos SET completed = true, completed_at = now(), completed_by = $2 WHERE id = $1`,
+        [args.id, ctx.userId]
+      );
+    } else {
+      await ctx.pg.query(
+        `UPDATE todos SET completed = false, completed_at = NULL, completed_by = NULL WHERE id = $1`,
+        [args.id]
+      );
+    }
+  }
+};
+
+const todoDeleteArgs = z.object({ id: z.string().uuid() });
+
+const todoDelete: ServerMutator<typeof todoDeleteArgs> = {
+  args: todoDeleteArgs,
+  run: async (args, ctx) => {
+    await ctx.pg.query(`DELETE FROM todos WHERE id = $1`, [args.id]);
+  }
+};
+
+export const serverMutators = {
+  listCreate,
+  listDelete,
+  todoCreate,
+  todoToggle,
+  todoDelete
+} as const satisfies Record<string, ServerMutator>;

backend/src/mutators/router.ts

@@ -0,0 +1,134 @@
+import express, { type Request, type Response } from 'express';
+import PG from 'pg';
+import { z } from 'zod';
+import { URL } from 'url';
+import config from '../../config.js';
+import { FatalOperationError, RetryableError } from '../errors.js';
+import type { OpBody, OpResponse } from '../types.js';
+import { serverMutators } from './mutators.js';
+import type { ServerMutator } from './types.js';
+
+const { Pool } = PG;
+
+const router = express.Router();
+
+if (config.database.type !== 'postgres') {
+  router.post('/invoke', (_req, res) => {
+    res.status(200).send({
+      status: 'fatal_error',
+      message: `Mutators are only implemented for the postgres driver (got ${config.database.type})`,
+      failed_operation: {
+        error_code: 'UNSUPPORTED_DRIVER',
+        message: `Configured driver "${config.database.type}" does not support mutators yet.`
+      }
+    });
+  });
+} else {
+  if (!config.database.uri) {
+    throw new Error('DATABASE_URI environment variable is required');
+  }
+  const url = new URL(config.database.uri);
+  const pool = new Pool({
+    host: url.hostname,
+    database: url.pathname.split('/')[1],
+    user: url.username,
+    password: url.password,
+    port: parseInt(url.port)
+  });
+  pool.on('error', (err) => {
+    console.error('Pool connection failure to postgres (mutators):', err);
+  });
+
+  router.post(
+    '/invoke',
+    async (
+      req: Request<{}, OpResponse<'invokeMutator'>, OpBody<'invokeMutator'>>,
+      res: Response<OpResponse<'invokeMutator'>>
+    ) => {
+      const { name, args: rawArgs, user_id } = req.body;
+      const mutator = (serverMutators as Record<string, ServerMutator>)[name];
+
+      if (!mutator) {
+        res.status(200).send({
+          status: 'fatal_error',
+          message: `Unknown mutator "${name}"`,
+          failed_operation: { error_code: 'UNKNOWN_MUTATOR', message: `No server-side handler registered for mutator "${name}".` }
+        });
+        return;
+      }
+
+      let parsedArgs: unknown;
+      try {
+        parsedArgs = mutator.args.parse(rawArgs);
+      } catch (e) {
+        const message = e instanceof z.ZodError ? JSON.stringify(e.issues) : String(e);
+        res.status(200).send({
+          status: 'fatal_error',
+          message: `Invalid args for mutator "${name}"`,
+          failed_operation: { error_code: 'VALIDATION_ERROR', message }
+        });
+        return;
+      }
+
+      // TODO: replace user_id from body with JWT-extracted id once auth lands.
+      const userId = user_id;
+      if (!userId) {
+        res.status(200).send({
+          status: 'fatal_error',
+          message: 'Missing user_id',
+          failed_operation: { error_code: 'VALIDATION_ERROR', message: 'user_id is required for v1 (until JWT auth lands).' }
+        });
+        return;
+      }
+
+      const client = await pool.connect();
+      try {
+        await client.query('BEGIN');
+        await mutator.run(parsedArgs, { userId, pg: client });
+        await client.query('COMMIT');
+        res.status(200).send({ status: 'success', message: `Mutator "${name}" applied` });
+      } catch (e) {
+        await client.query('ROLLBACK');
+        if (e instanceof FatalOperationError) {
+          res.status(200).send({
+            status: 'fatal_error',
+            message: e.message,
+            failed_operation: { error_code: e.errorCode, message: e.message }
+          });
+          return;
+        }
+        if (e instanceof RetryableError) {
+          res.status(200).send({ status: 'retryable_error', message: e.message });
+          return;
+        }
+        const err = e as Error & { code?: string };
+        const code = err.code ?? '';
+        if (code === '23505') {
+          res.status(200).send({
+            status: 'fatal_error',
+            message: err.message,
+            failed_operation: { error_code: 'UNIQUE_VIOLATION', message: err.message }
+          });
+        } else if (code === '23503') {
+          res.status(200).send({
+            status: 'fatal_error',
+            message: err.message,
+            failed_operation: { error_code: 'FOREIGN_KEY_VIOLATION', message: err.message }
+          });
+        } else if (code.startsWith('42')) {
+          res.status(200).send({
+            status: 'fatal_error',
+            message: err.message,
+            failed_operation: { error_code: 'SCHEMA_MISMATCH', message: err.message }
+          });
+        } else {
+          res.status(200).send({ status: 'retryable_error', message: err.message });
+        }
+      } finally {
+        client.release();
+      }
+    }
+  );
+}
+
+export { router as mutatorsRouter };

backend/src/mutators/types.ts

@@ -0,0 +1,12 @@
+import { z } from 'zod';
+import type { PoolClient } from 'pg';
+
+export interface ServerCtx {
+  userId: string;
+  pg: PoolClient;
+}
+
+export interface ServerMutator<S extends z.ZodTypeAny = z.ZodTypeAny> {
+  args: S;
+  run: (args: z.infer<S>, ctx: ServerCtx) => Promise<void>;
+}

frontend/package.json

@@ -13,20 +13,21 @@
     "@emotion/react": "^11.11.4",
     "@emotion/styled": "^11.11.5",
     "@esbuild-plugins/node-globals-polyfill": "^0.2.3",
-    "@journeyapps/wa-sqlite": "^1.2.0",
+    "@journeyapps/wa-sqlite": "latest",
     "@mui/icons-material": "^5.15.15",
     "@mui/material": "^5.15.12",
     "@mui/x-data-grid": "^6.19.6",
-    "@powersync/react": "^1.5.1",
-    "@powersync/web": "^1.12.1",
+    "@powersync/react": "latest",
+    "@powersync/web": "latest",
     "@supabase/supabase-js": "^2.39.7",
     "js-logger": "^1.6.1",
     "lodash": "^4.17.21",
     "openapi-fetch": "^0.17.0",
     "react": "^18.2.0",
     "react-dom": "^18.2.0",
     "react-router-dom": "^6.22.3",
-    "uuid": "^9.0.1"
+    "uuid": "^9.0.1",
+    "zod": "^4.4.3"
   },
   "devDependencies": {
     "@types/lodash": "^4.14.202",