Vex · guide

A reactive read

A read that answers again whenever a table it reads is written, for as long as somebody follows it.

Install it

pnpm add @niscorp/vex @niscorp/prism @niscorp/strata zod @electric-sql/pglite

Step 1 of 5

The database and the engine

Two bookings, and an engine that has read the schema. The read in this file keeps answering after the first call, so the file ends by stopping it.

The database and the engine

import { PGlite } from '@electric-sql/pglite';
import { createPostgresAdapter, createQueryEngine, handleQuery, seedCache } from '@niscorp/vex';
import { createPglitePool } from '@niscorp/vex/pglite';

const db = new PGlite();
await db.exec(`
  CREATE TABLE bookings (id text PRIMARY KEY, member text, seats integer);
  INSERT INTO bookings VALUES ('B-2040', 'Ada', 1), ('B-2041', 'Grace', 2);
`);
const engine = createQueryEngine({ adapter: createPostgresAdapter({ pool: createPglitePool(db) }) });
await engine.introspect();

Step 2 of 5

A read that keeps answering, and a write

A read is a snapshot unless its entry says otherwise. refresh: 'reactive' is authored on the seeded entry, never generated, and it means: answer again whenever a table this read reads is written, for as long as somebody follows.

The write is a mutation, stored beside the read under its own name. There is no generation path for writes, and this one is the only way a row gets into bookings from here on.

A read that keeps answering, and a write

await seedCache(engine.cache, [
  {
    fingerprint: 'boxoffice/seats',
    dsl: {
      from: ['bookings'],
      fields: ['bookings.member', 'bookings.seats'],
      sort: [{ field: 'bookings.member', dir: 'asc' }],
    },
    refresh: 'reactive',
  },
  {
    fingerprint: 'boxoffice/book',
    mutation: {
      op: 'insert',
      table: 'bookings',
      values: { id: { $context: 'id' }, member: { $context: 'member' }, seats: { $context: 'seats' } },
    },
  },
]);

Step 3 of 5

Follow it

Following is { signal, onChange }. The returned promise is the first answer, as for any read. Every later answer that differs from the last arrives through onChange, until the signal aborts.

Follow it

const abortController = new AbortController();
const response = await engine.execute({ fingerprint: 'boxoffice/seats', context: {} }, {
  signal: abortController.signal,
  onChange: (update) => console.log('changed:', JSON.stringify(update.result)),
});
console.log('first:', JSON.stringify(response.result));

Step 4 of 5

Write, and hear about it

The write goes through the handler, with a client to write through and a policy that allows it. After the commit the handler tells the engine which table changed. The followed read is fetched again, once, and its follower hears from it if the answer changed.

A write Vex did not make, raw SQL or a trigger, is the host’s to announce with engine.invalidate.

Write, and hear about it

const policy = { default: 'allow' as const, entities: {} };
const config = { engine, locked: true, mutations: { client: db, policy } };
const booking = { fingerprint: 'boxoffice/book', context: { id: 'B-2046', member: 'Ruth', seats: 1 } };
await handleQuery(config, booking, {});
await new Promise((resolve) => setTimeout(resolve, 200));

Step 5 of 5

Stop following

Aborting the signal ends the following. The rows cache keeps a timer for its TTL, so engine.rows.stop() is what lets the process end.

Stop following

abortController.abort();
engine.rows.stop();

Then

Run it.

The first answer came back from the call. The second arrived through onChange, after the write, without anybody asking again.

Run it

npx tsx reactive.mts

It prints

first: [{"member":"Ada","seats":1},{"member":"Grace","seats":2}]
changed: [{"member":"Ada","seats":1},{"member":"Grace","seats":2},{"member":"Ruth","seats":1}]