Data: optional per-collection JSON Schema validation

README.md

@@ -224,6 +224,23 @@ expression index) instead of a scan — `PUT /v1/data/:collection/indexes { fiel
 is declarative, so it reconciles to exactly the set you send. It's key-gated and
 never anonymous.
 
+Collections are **schemaless by default**. Opt into validation per collection by
+declaring a JSON Schema; writes that violate it then fail with `422` and a list of
+problems (PATCH validates the merged result, and bulk is all-or-nothing):
+
+```bash
+curl -X PUT localhost:3737/v1/data/members/schema \
+  -H 'content-type: application/json' \
+  -d '{"type":"object","properties":{"name":{"type":"string"},"age":{"type":"integer","minimum":0}},"required":["name"]}'
+
+curl -X POST localhost:3737/v1/data/members -H 'content-type: application/json' -d '{"age":-1}'
+#→ 422 { "error":"validation_failed", "errors":[{"field":"name","message":"..."},{"field":"age","message":"..."}] }
+
+curl -X DELETE localhost:3737/v1/data/members/schema   # back to schemaless
+```
+
+Via the SDK: `api.data('members').setSchema(schema)` / `getSchema()` / `clearSchema()`.
+
 ### Access control — per-collection rules
 
 By default every collection is private: the API key (if set) governs all access.

package.json

@@ -59,6 +59,7 @@
     "@fastify/multipart": "^10.0.0",
     "@fastify/static": "^9.1.3",
     "@fastify/websocket": "^11.2.0",
+    "ajv": "^8.20.0",
     "better-sqlite3": "^12.10.0",
     "fastify": "^5.8.5",
     "ioredis": "^5.11.1",

public/docs.html

@@ -194,6 +194,7 @@ <h2 id="rest-data">REST API — Data</h2>
               <tr><td class="ep"><span class="m">GET</span>/v1/data</td><td>List collections + counts.</td></tr>
               <tr><td class="ep"><span class="m">GET·PUT</span>/v1/data/:c/acl</td><td>Access rules <code>{ read, write: public|private, hidden: [fields] }</code> (key-gated). With <code>API_KEY</code> set, a <code>read:public</code> collection is readable anonymously; <code>hidden</code> fields are stripped from those reads.</td></tr>
               <tr><td class="ep"><span class="m">GET·PUT</span>/v1/data/:c/indexes</td><td>Declared indexed fields <code>{ fields: [...] }</code> (key-gated). Backs hot filter/sort fields with a real DB index for large collections; declarative — PUT the full desired set.</td></tr>
+              <tr><td class="ep"><span class="m">GET·PUT·DELETE</span>/v1/data/:c/schema</td><td>Optional JSON Schema (key-gated). Schemaless until set; then writes that violate it return <code>422</code> with an <code>errors</code> list (PATCH checks the merged result; bulk is all-or-nothing). DELETE to go back to schemaless.</td></tr>
             </tbody>
           </table>
           <p class="muted small" style="margin:.6rem 0 .2rem">Examples</p>

public/sdk.d.ts

@@ -169,6 +169,16 @@ declare namespace Zero {
       write?: 'public' | 'private';
       hidden?: string[];
     }): Promise<Acl>;
+    /** The collection's JSON Schema, or null if it's schemaless. */
+    getSchema(): Promise<Record<string, unknown> | null>;
+    /** Set a JSON Schema; subsequent writes that violate it reject with a 422. */
+    setSchema(schema: Record<string, unknown>): Promise<{ collection: string; schema: Record<string, unknown> }>;
+    /** Remove the schema (writes become unconstrained again). */
+    clearSchema(): Promise<{ collection: string; schema: null }>;
+    /** The fields declared as indexed for this collection. */
+    getIndexes(): Promise<string[]>;
+    /** Declare the indexed fields (declarative — pass the full desired set). */
+    setIndexes(fields: string[]): Promise<string[]>;
     /** Subscribe to live changes for this collection. Returns the room (`.close()`). */
     subscribe(cb: (change: ChangeEvent) => void): Room;
   }

public/sdk.js

@@ -181,6 +181,24 @@
         setAcl: function (acl) {
           return request('PUT', root + '/acl', acl || {});
         },
+        // Optional JSON Schema validation for this collection. Writes that violate
+        // a set schema reject with a 422 (err.body.errors lists the problems).
+        getSchema: function () {
+          return request('GET', root + '/schema').then(function (r) { return r.schema; });
+        },
+        setSchema: function (schema) {
+          return request('PUT', root + '/schema', schema || {});
+        },
+        clearSchema: function () {
+          return request('DELETE', root + '/schema');
+        },
+        // Declared indexed fields for fast filter/sort on large collections.
+        getIndexes: function () {
+          return request('GET', root + '/indexes').then(function (r) { return r.fields; });
+        },
+        setIndexes: function (fields) {
+          return request('PUT', root + '/indexes', { fields: fields || [] }).then(function (r) { return r.fields; });
+        },
         // Live updates: cb receives { type: 'created'|'updated'|'deleted', collection, id, document? }
         // for every write to this collection. Returns the room (call .close() to stop).
         subscribe: function (cb) {

src/db/postgres.ts

@@ -165,5 +165,12 @@ async function bootstrap(pool: pg.Pool): Promise<void> {
       field      text NOT NULL,
       PRIMARY KEY (project, collection, field)
     );
+
+    CREATE TABLE IF NOT EXISTS collection_schemas (
+      project    text NOT NULL DEFAULT 'default',
+      collection text NOT NULL,
+      schema     text NOT NULL,
+      PRIMARY KEY (project, collection)
+    );
   `);
 }

src/db/sqlite.ts

@@ -187,6 +187,16 @@ export function bootstrap(db: Database.Database): void {
     );
   `);
 
+  // --- optional per-collection JSON Schema (schemaless unless declared) ---
+  db.exec(`
+    CREATE TABLE IF NOT EXISTS collection_schemas (
+      project    TEXT NOT NULL DEFAULT 'default',
+      collection TEXT NOT NULL,
+      schema     TEXT NOT NULL,
+      PRIMARY KEY (project, collection)
+    );
+  `);
+
   // --- full-text index (rebuilt if it predates the project column) ---
   if (tableExists(db, 'documents_fts') && !hasColumn(db, 'documents_fts', 'project')) {
     db.exec(`DROP TABLE documents_fts;`);

src/modules/data/routes.ts

@@ -7,6 +7,7 @@ import {
   type Operator,
 } from './store.js';
 import type { AccessLevel } from '../acl/store.js';
+import { InvalidSchemaError } from './schema.js';
 
 const ACCESS_LEVELS = new Set(['public', 'private']);
 const AGGREGATE_OPS = new Set<AggregateOp>(['count', 'sum', 'avg', 'min', 'max']);
@@ -92,7 +93,7 @@ function parseFilters(query: Record<string, unknown>, prefix: string): Filter[]
  * name and it just works — no setup, no schema, no migration.
  */
 export async function dataRoutes(app: FastifyInstance): Promise<void> {
-  const store = new DocumentStore(app.db);
+  const store = new DocumentStore(app.db, app.schemas);
 
   // Broadcast a change so clients subscribed to `data:<collection>` get live
   // updates. Channels are namespaced per project by the realtime routes.
@@ -231,6 +232,36 @@ export async function dataRoutes(app: FastifyInstance): Promise<void> {
     return { collection, fields: await store.setIndexes(req.projectId, collection, body.fields as string[]) };
   });
 
+  // Optional per-collection JSON Schema. Collections are schemaless until one is
+  // set; then writes that violate it fail with 422. Key-gated (never anonymous).
+  app.get('/v1/data/:collection/schema', async (req) => {
+    const { collection } = req.params as { collection: string };
+    return { collection, schema: await app.schemas.get(req.projectId, collection) };
+  });
+
+  app.put('/v1/data/:collection/schema', async (req, reply) => {
+    const { collection } = req.params as { collection: string };
+    const body = req.body;
+    if (!body || typeof body !== 'object' || Array.isArray(body)) {
+      return reply.code(400).send({ error: 'bad_request', message: 'Body must be a JSON Schema object.' });
+    }
+    try {
+      await app.schemas.set(req.projectId, collection, body as Record<string, unknown>);
+    } catch (err) {
+      if (err instanceof InvalidSchemaError) {
+        return reply.code(400).send({ error: 'bad_request', message: err.message });
+      }
+      throw err;
+    }
+    return { collection, schema: body };
+  });
+
+  app.delete('/v1/data/:collection/schema', async (req) => {
+    const { collection } = req.params as { collection: string };
+    await app.schemas.clear(req.projectId, collection);
+    return { collection, schema: null };
+  });
+
   // Count matching documents without transferring them: same f./or. filters as
   // list. The static "count" segment takes precedence over the :id route.
   app.get('/v1/data/:collection/count', async (req) => {

src/modules/meta/routes.ts

@@ -79,6 +79,7 @@ export async function metaRoutes(app: FastifyInstance): Promise<void> {
         collections: 'GET /v1/data',
         acl: 'GET/PUT /v1/data/:collection/acl { read, write: public|private, hidden: [fields] } (key-gated)',
         indexes: 'GET/PUT /v1/data/:collection/indexes { fields: [...] } — declare indexed fields for fast queries (key-gated)',
+        schema: 'GET/PUT/DELETE /v1/data/:collection/schema — optional JSON Schema; violating writes 422 (key-gated)',
         subscribe: 'WS /v1/realtime/data:<collection> (live created/updated/deleted events)',
       },
       files: {

src/server.ts

@@ -19,6 +19,7 @@ import { metaRoutes } from './modules/meta/routes.js';
 import { backupRoutes } from './modules/backup/routes.js';
 import { registerRateLimit, type RateLimitOptions } from './rate-limit.js';
 import { AclStore } from './modules/acl/store.js';
+import { SchemaStore, SchemaValidationError } from './modules/data/schema.js';
 import { getBlobStore } from './modules/files/blob.js';
 import { newId } from './lib/id.js';
 
@@ -30,8 +31,8 @@ import { newId } from './lib/id.js';
 function dataAccess(method: string, url: string): { collection: string; op: 'read' | 'write' } | null {
   const parts = url.split('/').filter(Boolean); // url is already query-stripped
   if (parts[0] !== 'v1' || parts[1] !== 'data' || parts.length < 3) return null;
-  // ACL and index management are never anonymously accessible.
-  if (parts[3] === 'acl' || parts[3] === 'indexes') return null;
+  // ACL, index, and schema management are never anonymously accessible.
+  if (parts[3] === 'acl' || parts[3] === 'indexes' || parts[3] === 'schema') return null;
   return { collection: parts[2]!, op: method === 'GET' ? 'read' : 'write' };
 }
 
@@ -124,6 +125,7 @@ export async function buildServer(opts: BuildOptions = {}): Promise<FastifyInsta
   app.decorate('blobs', getBlobStore());
   app.decorate('projects', new ProjectStore(app.db, app.blobs));
   app.decorate('acl', new AclStore(app.db));
+  app.decorate('schemas', new SchemaStore(app.db));
   app.decorateRequest('projectId', DEFAULT_PROJECT);
   // True unless the request reached a collection anonymously via a public ACL;
   // read routes use this to redact `hidden` fields from anonymous callers.
@@ -235,6 +237,14 @@ export async function buildServer(opts: BuildOptions = {}): Promise<FastifyInsta
     reply.code(404).send({ error: 'not_found', message: `Cannot ${req.method} ${req.url}` });
   });
   app.setErrorHandler((err: FastifyError, req, reply) => {
+    // A document that fails its collection's schema is a 422 with the details.
+    if (err instanceof SchemaValidationError) {
+      return reply.code(422).send({
+        error: 'validation_failed',
+        message: 'Document does not match the collection schema.',
+        errors: err.errors,
+      });
+    }
     const status = err.statusCode ?? 500;
     if (status >= 500) req.log.error(err);
     reply.code(status).send({

src/types.d.ts

@@ -3,6 +3,7 @@ import type { Db } from './db.js';
 import type { RealtimeHub } from './modules/realtime/hub.js';
 import type { ProjectStore } from './modules/projects/store.js';
 import type { AclStore } from './modules/acl/store.js';
+import type { SchemaStore } from './modules/data/schema.js';
 import type { BlobStore } from './modules/files/blob.js';
 
 declare module 'fastify' {
@@ -12,6 +13,8 @@ declare module 'fastify' {
     projects: ProjectStore;
     /** Per-collection access rules. */
     acl: AclStore;
+    /** Optional per-collection JSON Schema validation. */
+    schemas: SchemaStore;
     /** Raw blob backend for file uploads (disk or S3). */
     blobs: BlobStore;
     /** Effective bearer key for this instance (null = open). */

test/api.test.ts

@@ -501,6 +501,56 @@ test('data: declared indexes are created, listed, reconciled, and keep queries c
   assert.equal(bad.status, 400);
 });
 
+test('data: optional per-collection JSON Schema validates writes (422)', async () => {
+  const api = Zero(base);
+  const col = api.data('members');
+
+  // Schemaless to start: anything goes.
+  assert.equal(await col.getSchema(), null);
+  await col.create({ anything: true });
+
+  // Declare a schema: name (string, required), age (integer ≥ 0), no extras.
+  const schema = {
+    type: 'object',
+    properties: { name: { type: 'string' }, age: { type: 'integer', minimum: 0 } },
+    required: ['name'],
+    additionalProperties: false,
+  };
+  await col.setSchema(schema);
+  assert.deepEqual(await col.getSchema(), schema);
+
+  // Valid write succeeds.
+  const ok = await col.create({ name: 'Ada', age: 36 });
+  assert.equal(ok.name, 'Ada');
+
+  // Invalid writes reject with 422 + a structured errors list.
+  await assert.rejects(
+    col.create({ age: -1 }), // missing name + negative age
+    (e: any) => e.status === 422 && Array.isArray(e.body.errors) && e.body.errors.length >= 1,
+  );
+  // PATCH that would violate the schema is also rejected (the merged result is checked).
+  await assert.rejects(col.update(ok.id, { age: 'old' }), (e: any) => e.status === 422);
+  // Bulk is all-or-nothing: one bad doc rejects the batch.
+  await assert.rejects(
+    col.createMany([{ name: 'Grace' }, { age: 5 }]),
+    (e: any) => e.status === 422,
+  );
+  assert.equal(await col.count({ where: { name: 'Grace' } }), 0, 'no partial bulk insert');
+
+  // A malformed schema declaration is a 400, not a 422.
+  const badSchema = await fetch(`${base}/v1/data/members/schema`, {
+    method: 'PUT',
+    headers: { 'content-type': 'application/json' },
+    body: JSON.stringify({ type: 'not-a-real-type' }),
+  });
+  assert.equal(badSchema.status, 400);
+
+  // Clearing the schema makes writes unconstrained again.
+  await col.clearSchema();
+  assert.equal(await col.getSchema(), null);
+  await col.create({ totally: 'freeform' });
+});
+
 test('hardening: bad limit/offset query params are clamped, not crashed', async () => {
   await fetch(`${base}/v1/data/clamp_probe`, {
     method: 'POST',