@baseworks/readmodel

Declarative read models over an event log — "PostgREST for your projections".

Define a projection as data (a _projections row), and the engine folds events into a queryable table with zero service code. Define a select policy (_policies) and a generic POST /v1/query/:name endpoint serves filtered, sorted, paged, column-projected reads — row-scoped to the caller. New table = new definition row, no new endpoints, no new code.

append(event) ──▶ applyEvent()  ──rules──▶  read_<name> table ──▶ execQuery() ──▶ rows
                  (mode A fold)                                  (policy AND-ed)

Storage-agnostic (SqlClient = any libsql-shaped client) and event-source- agnostic (EventLike = { type, streamId, tenant, payload } — structurally compatible with @baseworks/eventstore's StoredEvent).

The two modes (and what this package refuses to do)

• Mode A (this package): fold rules are data — upsert (payload paths / literals), inc counters, delete. Covers ~80% of read models. Rebuild is fully generic: the host can rebuild any projection from the log on its own.
• Mode B (upsertRow / deleteRow): real domain folds (cross-aggregate walks, invariants, a state machine). The owning service computes the row and pushes it through upsertRow — the rules engine still never learns domain logic, but the registry stays the single source of truth for the table shape (columns are validated, the (tenant, id) + updated_at spine is managed).

Quick start

import { createClient } from '@libsql/client'
import { Registry, applyEvent, upsertRow, execQuery, rebuildProjection } from '@baseworks/readmodel'

const db = createClient({ url: process.env.DB_URL! })
const registry = new Registry(db)
await registry.init()

// 1. Projection = data. Registering it creates read_notes AND whitelists "notes".
await registry.saveProjection({
  name: 'notes',
  columns: { owner: 'text', text: 'text', created_at: 'integer' },
  on: {
    NoteAdded:   { op: 'upsert', set: { owner: '$.owner', text: '$.text', created_at: '$.createdAt' } },
    NoteDeleted: { op: 'delete' },
  },
})

// 2. Policy = data. Deny-by-default; this grants "everyone sees their own rows".
await registry.savePolicy({ name: 'notes', role: '*', using: 'owner = :auth_uid' })

// 3a. Mode A — fold inline on every append (read-your-writes):
await applyEvent(db, registry, storedEvent)

// 3b. Mode B — the service computes the row itself and pushes it:
await upsertRow(db, registry, 'notes', tenant, id, { owner, text, created_at })

// 4. Serve the generic query endpoint:
const result = await execQuery(db, registry, 'notes', requestBody, {
  tenant: ctx.tenant, uid: ctx.userId, role: ctx.role,
})

Query body (POST /v1/query/notes — reads are POSTs by design: rich nested filters, no URL limits):

{
  "select": ["id", "text", "created_at"],
  "where":  { "and": [ { "text": { "like": "%demo%" } }, { "created_at": { "gt": 100 } } ] },
  "sort":   ["-created_at"],
  "limit":  50, "offset": 0
}

Operators: eq ne gt gte lt lte like in, nested and / or. eq: null → IS NULL.

Typed projections (Zod)

Instead of hand-writing a columns map, declare a projection from a Zod schema — one source of truth for the row shape, reusable by the service, the SDK, and clients. defineProjection derives the SQLite columns and returns typed serialize/parse helpers; jsonCol stores a structured value as JSON text (rich type via z.infer, JSON.stringify on write, JSON.parse on read).

import { z } from 'zod'
import { defineProjection, jsonCol } from '@baseworks/readmodel'

export const AssetRow = z.object({
  kind: z.string().nullable(),
  status: z.string(),
  unclaimed: z.boolean(),              // → integer (0/1)
  registered_at: z.number().int().nullable(),
  signals: jsonCol(z.array(SignalSchema)),   // → text (JSON)
  tags: jsonCol(z.array(z.string())),
})
export type AssetRow = z.infer<typeof AssetRow>

export const assets = defineProjection('assets', AssetRow, { on: {} })

registry.saveProjection(assets.def)                                  // plain columns (persisted)
upsertRow(db, registry, 'assets', tenant, id, assets.toColumns(row)) // rich → columns
const rich = assets.fromRow(queryRow)                                // columns → rich (typed)

Zod → SQLite: string/enum → text, number → real (.int() → integer), boolean → integer, jsonCol(...) → text. A bare object/array throws — wrap it in jsonCol. defineProjection is additive: spec.def is exactly the plain ProjectionDef the engine persists and queries.

Security invariants (each is tested)

1. Names resolve only through the registry — _projections, _policies, _rpc and the event log are structurally unreachable.
2. Deny-by-default — no policy for (name, role) (nor a * fallback) → 403.
3. tenant = ? is AND-ed structurally into every query, never via policy text.
4. Client JSON never reaches SQL as text: columns are whitelisted against the definition, values are bound parameters. Policy using fragments are operator-authored config (written in migrations, like schema) whose :auth_* placeholders bind from the verified auth context.
5. allow / deny column lists apply to select, where, and sort alike.

Rebuild

Read models are disposable; the log is the truth:

await rebuildProjection(db, registry, 'notes', tenant, eventsInGlobalSeqOrder)

Hosts typically expose this as POST /admin/rebuild.

Auth context

:auth_uid, :auth_role, :auth_email, :auth_tenant, :auth_claim_<key> — bound from the AuthCtx the host builds from its verified JWT. A policy that references a value the caller lacks denies (never widens).

What's deliberately out (for now)

• _rpc named parametric queries — table reserved, engine v2.
• A remote mode-B push endpoint (upsertRow is in-process today), lookup enrichment, catch-up checkpoints — the hosting service's concern or later iterations.