87c21f9 · 9b3a459
@@ -26,7 +26,7 @@
decide nothing themselves. `AGENTS.md` says how to add to it.
| `src/actions.ts`                   | propose, list, read, withdraw, health — and who may          |
| `src/who.ts`                       | who is asking, from the forge's signed headers               |
| `src/db.ts`                        | SQLite under `FORGE_APP_DATA`, WAL, forward-only migrations  |
| `src/live.ts`                      | the in-process bus and `GET /events`                         |
| `src/live.ts`                      | the event log, the bus, and `GET /events`                    |
| `src/mcp.ts`                       | the actions as MCP tools at `POST /mcp`                      |
| `src/routes.ts`                    | every path, in one table                                     |
| `src/pages/`                       | the HTML, rendered on the server: one layout, two pages      |
@@ -102,7 +102,12 @@
claude mcp add --transport http __TITLE__ https://<app>/mcp
```
The page stays current over `/events`: a proposal someone else opens or
withdraws appears without a reload. Without JavaScript it is the same page, one
withdraws appears without a reload — including one written at the terminal,
which is another process on the same file. Every event is a row in `event`,
so the server follows what it did not write itself while anyone is
watching, and a browser that reconnects with `Last-Event-ID` is replayed
from those rows even across a restart. The rows are trimmed to the last
hour and the last thousand. Without JavaScript it is the same page, one
reload behind.
The terminal opens with the exact command in `FORGE_APP_SSH` when the app
@@ -12,9 +12,12 @@
import {
  type AdminHealth,
} from "./actions.ts";
import { migrations, openDb } from "./db.ts";
import { useDb } from "./live.ts";
import type { Who } from "./who.ts";
const db = openDb(mkdtempSync(join(tmpdir(), "app-actions-")));
// What an action emits belongs in this test's database, not the app's.
useDb(db);
const anon: Who = { role: "anon" };
const alice: Who = { role: "user", id: "u1", name: "alice" };
const bob: Who = { role: "user", id: "u2", name: "bob" };
@@ -49,6 +49,15 @@
export const migrations: string[] = [
  // The template's note is gone: a proposal is what this app is about.
  // Migration 1 stays exactly as it ran; this is how history is undone.
  `DROP TABLE note`,
  // The live log (`live.ts`): every emitted event, so a write by another
  // process on this file reaches the pages this one is serving. Trimmed to
  // the last hour and the last thousand rows; the rowid is the event id.
  `CREATE TABLE event (
    id INTEGER PRIMARY KEY,
    type TEXT NOT NULL,
    data TEXT NOT NULL,
    created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
  )`,
];
export function openDb(dir = dataDir): Database {
@@ -0,0 +1,237 @@
import { afterAll, expect, test } from "bun:test";
import { mkdtempSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { openDb } from "./db.ts";
import { emit, events, following, subscribe, trim } from "./live.ts";
import { serve } from "./server.ts";
const dir = mkdtempSync(join(tmpdir(), "app-live-"));
const server = serve({ port: 0, dataDir: dir, identityKey: "" });
const base = `http://${server.hostname}:${server.port}`;
afterAll(() => server.stop(true));
/** A second connection to the same file, which is all another process is. */
const other = openDb(dir);
const write = (type: string, data: unknown): number =>
  other
    .query<
      { id: number },
      [string, string]
    >("INSERT INTO event (type, data) VALUES (?1, ?2) RETURNING id")
    .get(type, JSON.stringify(data))!.id;
/**
 * A stream, read frame by frame, closed the way a browser closes one. The
 * route is `events(request)`; opening it here rather than over HTTP is what
 * lets a test close it and see the server let go — Bun holds a cancelled
 * fetch's socket in its pool, so the server never learns it was dropped.
 */
function open(since?: number | string) {
  const aborter = new AbortController();
  const url = since === undefined ? "/events" : `/events?since=${since}`;
  const response = events(
    new Request(`http://live.test${url}`, { signal: aborter.signal }),
  );
  const reader = response.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = "";
  let reading: Promise<void> | undefined;
  // One read at a time, and a chunk that arrives after we stopped waiting
  // is still kept: waiting is what times out, never the stream.
  const pump = async (within: number): Promise<void> => {
    reading ??= reader.read().then(({ value }) => {
      reading = undefined;
      if (value !== undefined) buffer += decoder.decode(value);
    });
    await Promise.race([reading, Bun.sleep(within)]);
  };
  return {
    response,
    async next(within = 3000): Promise<{ id: number; type: string; data: any }> {
      const deadline = Date.now() + within;
      for (;;) {
        const frame = /id: (\d+)\ndata: (.*)\n\n/.exec(buffer);
        if (frame) {
          buffer = buffer.slice(frame.index + frame[0].length);
          const { type, data } = JSON.parse(frame[2]!);
          return { id: Number(frame[1]), type, data };
        }
        if (Date.now() > deadline) throw new Error(`no event: ${buffer}`);
        await pump(100);
      }
    },
    /** Everything the stream has to say right now. */
    async drain(ms = 100): Promise<string> {
      for (;;) {
        const before = buffer.length;
        await pump(ms);
        if (buffer.length === before) return buffer;
      }
    },
    seen: () => buffer,
    /** Whether the server has closed its end. */
    ended: async (): Promise<boolean> => {
      for (;;) {
        const { done, value } = await reader.read();
        if (done) return true;
        buffer += decoder.decode(value);
      }
    },
    close: () => aborter.abort(),
  };
}
test("an event emitted in this process reaches a subscriber at once", () => {
  const heard: { id: number; type: string; data: unknown }[] = [];
  const unsubscribe = subscribe((event) => heard.push(event));
  const event = emit("test.ping", { n: 1 });
  unsubscribe();
  expect(heard).toEqual([{ id: event.id, type: "test.ping", data: { n: 1 } }]);
  expect(event.id).toBeGreaterThan(0);
  // And it is a row, so another process can read it back.
  expect(
    other
      .query<
        { type: string; data: string },
        [number]
      >("SELECT type, data FROM event WHERE id = ?1")
      .get(event.id),
  ).toEqual({ type: "test.ping", data: '{"n":1}' });
  // Nobody is listening now, and the ids still climb.
  expect(emit("test.ping", null).id).toBe(event.id + 1);
});
test("an event another connection writes reaches the stream within a tick", async () => {
  const stream = open();
  const id = write("proposal.opened", { id: 7, title: "from the terminal" });
  expect(await stream.next()).toEqual({
    id,
    type: "proposal.opened",
    data: { id: 7, title: "from the terminal" },
  });
  // A row that will not parse is skipped, never thrown, and does not stall
  // the rows behind it.
  other
    .query("INSERT INTO event (type, data) VALUES ('proposal.opened', '{')")
    .run();
  const next = write("proposal.withdrawn", { id: 7 });
  expect((await stream.next()).id).toBe(next);
  stream.close();
});
test("a stream with no id starts from now; `since` replays only what follows", async () => {
  const first = emit("test.one", 1);
  const second = emit("test.two", 2);
  const fresh = open();
  const replay = open(first.id);
  expect(await replay.next()).toEqual({
    id: second.id,
    type: "test.two",
    data: 2,
  });
  // What came before `since` is not replayed, and the stream that asked
  // for nothing is given nothing at all.
  expect(await replay.drain()).not.toContain("test.one");
  const opening = await fresh.drain();
  expect(opening).toContain("retry:");
  expect(opening).not.toContain("test.one");
  expect(opening).not.toContain("test.two");
  // Both hear what comes next.
  const third = emit("test.three", 3);
  expect((await fresh.next()).id).toBe(third.id);
  expect((await replay.next()).id).toBe(third.id);
  fresh.close();
  replay.close();
  // A `since` that is not a whole number is no `since` at all.
  const nonsense = open("yesterday");
  expect(await nonsense.drain()).not.toContain("test.three");
  nonsense.close();
});
test("the trim keeps the newest and drops the oldest", () => {
  const count = (): number =>
    other.query<{ n: number }, []>("SELECT count(*) AS n FROM event").get()!.n;
  const top = (): number =>
    other.query<{ id: number }, []>("SELECT max(id) AS id FROM event").get()!
      .id;
  const ancient = write("test.ancient", 0);
  other
    .query("UPDATE event SET created_at = '2001-01-01T00:00:00.000Z' WHERE id = ?1")
    .run(ancient);
  other.transaction(() => {
    for (let i = 0; i < 1100; i++) write("test.bulk", i);
  })();
  const newest = top();
  trim();
  expect(count()).toBeLessThanOrEqual(1000);
  expect(top()).toBe(newest);
  expect(
    other
      .query<{ id: number }, [number]>("SELECT id FROM event WHERE id = ?1")
      .get(ancient),
  ).toBeNull();
  // The newest row survives even when everything is older than an hour, so
  // the next id is still higher than any a client holds.
  other.query("UPDATE event SET created_at = '2001-01-01T00:00:00.000Z'").run();
  trim();
  expect(count()).toBe(1);
  expect(top()).toBe(newest);
  expect(emit("test.after-trim", null).id).toBe(newest + 1);
});
test("a closed stream unsubscribes, and the last one stops the tick", async () => {
  expect(following()).toBe(false);
  const one = open();
  const two = open();
  expect(following()).toBe(true);
  const id = write("test.watched", null);
  expect((await one.next()).id).toBe(id);
  // One of two closing leaves the timer where it is; the second stops it.
  one.close();
  expect(following()).toBe(true);
  two.close();
  expect(following()).toBe(false);
  // Both ends are done, and what is written now reaches neither of them.
  const after = write("test.unwatched", null);
  expect(await one.ended()).toBe(true);
  expect(await two.ended()).toBe(true);
  expect(two.seen()).not.toContain(`id: ${after}`);
  expect(following()).toBe(false);
});
test("a write by another process reaches an open page, through the same action", async () => {
  const stream = await fetch(base + "/events");
  const reader = stream.body!.getReader();
  await reader.read(); // the retry line: the stream is open and following
  // What the terminal is: another process, the same actions, the same
  // `emit`, over the same file.
  const child = Bun.spawnSync({
    cmd: [
      "bun",
      "-e",
      `import { propose } from ${JSON.stringify(join(import.meta.dir, "actions.ts"))};
       import { openDb } from ${JSON.stringify(join(import.meta.dir, "db.ts"))};
       const opened = propose(openDb(), { role: "user", id: "u1", name: "tty" }, {
         title: "typed at the terminal",
         body: "Made by another process entirely.",
         options: ["Yes", "No"],
       });
       if (!opened.ok) throw new Error(opened.message);`,
    ],
    env: { ...process.env, FORGE_APP_DATA: dir },
  });
  expect(child.stderr.toString()).toBe("");
  expect(child.exitCode).toBe(0);
  let buffer = "";
  const deadline = Date.now() + 3000;
  while (!buffer.includes("proposal.opened") && Date.now() < deadline) {
    const read = await Promise.race([
      reader.read(),
      Bun.sleep(3000).then(() => ({ value: undefined })),
    ]);
    buffer += new TextDecoder().decode(read.value);
  }
  expect(buffer).toContain("typed at the terminal");
  await reader.cancel();
});
@@ -1,44 +1,194 @@
/**
 * Realtime without a framework. Actions `emit` what changed; `GET /events`
 * streams it to every open page as server-sent events, numbered, with the
 * last hundred kept so a client that reconnects with `Last-Event-ID` (or
 * `?since=`) hears what it missed. `static/live.js` turns the stream into
 * DOM events. The bus is this process's: the terminal is another process,
 * and its writes show on the next load rather than live.
 * Realtime without a framework. An action `emit`s what changed: the event
 * is appended to the `event` table and handed to this process's listeners
 * in the same breath, so a page served here waits on nothing. `GET /events`
 * streams it to every open page as server-sent events, numbered by the
 * row's id.
 *
 * The table is what makes the terminal live: `src/tty.ts` is another
 * process on the same file, and while anyone is watching, the server
 * follows the rows it did not write itself — one timer for the whole
 * process, a primary-key read every quarter second that asks for nothing
 * when the highest id has not moved. The page still polls nothing.
 *
 * Because the ids are the table's, they keep climbing across a restart: a
 * client that reconnects with `Last-Event-ID` (or `?since=`) is replayed
 * from the rows, never from a ring this process forgot.
 */
import type { Database } from "bun:sqlite";
import { openDb } from "./db.ts";
export interface LiveEvent {
  readonly id: number;
  readonly type: string;
  readonly data: unknown;
}
const KEEP = 100;
/** How often the server looks for rows another process wrote. */
const TICK_MS = 250;
const HEARTBEAT_MS = 25_000;
const kept: LiveEvent[] = [];
const listeners = new Set<(event: LiveEvent) => void>();
let last = 0;
/** The most a reconnecting client is replayed in one go. */
const REPLAY_MAX = 1000;
/** What survives a trim: an hour, and a thousand rows. */
const KEEP_ROWS = 1000;
const KEEP_MS = 60 * 60 * 1000;
/** A burst of writes trims once, after the responses, not inside them. */
const TRIM_AFTER_MS = 5_000;
interface Row {
  readonly id: number;
  readonly type: string;
  readonly data: string;
}
const listeners = new Set<(event: LiveEvent) => void>();
let store: Database | undefined;
/** The highest id handed to the listeners. Only ever climbs. */
let sent = 0;
let tick: ReturnType<typeof setInterval> | undefined;
let trimming: ReturnType<typeof setTimeout> | undefined;
/**
 * The database the log is kept in. A door that opened one says so at boot
 * (`server.ts`); a process that did not — the terminal — falls back to the
 * app's own file, which is the same file.
 */
export function useDb(db: Database): void {
  store = db;
  sent = highest();
}
const db = (): Database => (store ??= openDb());
const highest = (): number =>
  db()
    .query<{ id: number | null }, []>("SELECT max(id) AS id FROM event")
    .get()?.id ?? 0;
/** Append the event, then tell this process's listeners at once. */
export function emit(type: string, data: unknown): LiveEvent {
  const event: LiveEvent = { id: ++last, type, data };
  kept.push(event);
  if (kept.length > KEEP) kept.shift();
  for (const listen of listeners) listen(event);
  const row = db()
    .query<
      { id: number },
      [string, string]
    >("INSERT INTO event (type, data) VALUES (?1, ?2) RETURNING id")
    .get(type, JSON.stringify(data) ?? "null")!;
  const event: LiveEvent = { id: row.id, type, data };
  deliver(event);
  scheduleTrim();
  return event;
}
/** Every event goes out exactly once, in id order, whoever wrote it. */
function deliver(event: LiveEvent): void {
  if (event.id <= sent) return;
  sent = event.id;
  for (const listen of listeners) listen(event);
}
/** A row as an event, or nothing at all: data that will not parse is skipped, never thrown. */
function parse(row: Row): LiveEvent | undefined {
  try {
    return { id: row.id, type: row.type, data: JSON.parse(row.data) };
  } catch {
    return undefined;
  }
}
const after = (id: number): Row[] =>
  db()
    .query<
      Row,
      [number, number]
    >("SELECT id, type, data FROM event WHERE id > ?1 ORDER BY id LIMIT ?2")
    .all(id, REPLAY_MAX);
/** One pass of the follower: what another process wrote since we last looked. */
function follow(): void {
  try {
    for (const row of after(sent)) {
      const event = parse(row);
      if (event === undefined) sent = Math.max(sent, row.id);
      else deliver(event);
    }
  } catch (error) {
    console.error(error);
  }
}
export function subscribe(listen: (event: LiveEvent) => void): () => void {
  if (listeners.size === 0) startFollowing();
  listeners.add(listen);
  return () => listeners.delete(listen);
  return () => {
    listeners.delete(listen);
    if (listeners.size === 0) stopFollowing();
  };
}
/**
 * The follower runs while someone is listening and not a moment longer:
 * one timer for the process, started by the first stream and stopped by
 * the last. It starts from the highest id there is — what happened while
 * nobody was watching is history, and a client that wants it asks with
 * `since`.
 */
export function startFollowing(): void {
  if (tick !== undefined) return;
  sent = Math.max(sent, highest());
  tick = setInterval(follow, TICK_MS);
  tick.unref?.();
}
export function stopFollowing(): void {
  if (tick === undefined) return;
  clearInterval(tick);
  tick = undefined;
}
/** Whether the follower's timer is running. */
export const following = (): boolean => tick !== undefined;
/**
 * Keep an hour, keep a thousand, and always keep the newest row — with the
 * table empty SQLite would hand out id 1 again, and ids a client holds
 * must never come back around.
 */
export function trim(now = Date.now()): void {
  const top = highest();
  if (top === 0) return;
  const cutoff = new Date(now - KEEP_MS).toISOString();
  db().query("DELETE FROM event WHERE id <= ?1").run(top - KEEP_ROWS);
  db()
    .query("DELETE FROM event WHERE id < ?1 AND created_at < ?2")
    .run(top, cutoff);
}
function scheduleTrim(): void {
  if (trimming !== undefined) return;
  trimming = setTimeout(() => {
    trimming = undefined;
    try {
      trim();
    } catch (error) {
      console.error(error);
    }
  }, TRIM_AFTER_MS);
  trimming.unref?.();
}
/** `GET /events`: a stream that replays what `since` missed, then follows. */
export function events(request: Request): Response {
  const since = Number(
  // A client says where it got to, or says nothing and starts from now —
  // the page it is on was rendered a moment ago. Anything but a whole
  // number is nothing.
  const asked =
    request.headers.get("last-event-id") ??
      new URL(request.url).searchParams.get("since") ??
      0,
  );
    new URL(request.url).searchParams.get("since");
  const since =
    asked === null || !/^\d+$/.test(asked.trim())
      ? undefined
      : Number(asked.trim());
  const encoder = new TextEncoder();
  let stop = (): void => {};
  const stream = new ReadableStream<Uint8Array>({
@@ -65,7 +215,13 @@
export function events(request: Request): Response {
      };
      request.signal.addEventListener("abort", stop);
      write("retry: 2000\n\n");
      for (const event of kept) if (event.id > since) send(event);
      // Nothing can arrive between subscribing and replaying: this runs to
      // the end before any other turn of the loop.
      if (since !== undefined)
        for (const row of after(since)) {
          const event = parse(row);
          if (event !== undefined) send(event);
        }
    },
    cancel() {
      stop();
@@ -16,6 +16,7 @@
import {
  where,
} from "./env.ts";
import { allow, html, sameOrigin, text } from "./http.ts";
import { useDb } from "./live.ts";
import { match } from "./routes.ts";
import { who } from "./who.ts";
@@ -28,6 +29,11 @@
export interface ServeOptions {
export function serve(options: ServeOptions) {
  const db = openDb(options.dataDir);
  // The live log is kept in this database: `emit` appends to it, and while
  // a stream is open the follower reads back what another process — the
  // terminal — wrote. The first stream starts that timer and the last one
  // stops it, so an idle server does nothing at all.
  useDb(db);
  // The fact that identities are checked travels with the request; the key
  // itself stays here, where `who` reads it and nothing else can.
  const identity = identityState(options.identityKey);
		@@ -26,7 +26,7 @@ decide nothing themselves. `AGENTS.md` says how to add to it.
26	26	\| `src/actions.ts` \| propose, list, read, withdraw, health — and who may \|
27	27	\| `src/who.ts` \| who is asking, from the forge's signed headers \|
28	28	\| `src/db.ts` \| SQLite under `FORGE_APP_DATA`, WAL, forward-only migrations \|
29		\| `src/live.ts` \| the in-process bus and `GET /events` \|
	29	\| `src/live.ts` \| the event log, the bus, and `GET /events` \|
30	30	\| `src/mcp.ts` \| the actions as MCP tools at `POST /mcp` \|
31	31	\| `src/routes.ts` \| every path, in one table \|
32	32	\| `src/pages/` \| the HTML, rendered on the server: one layout, two pages \|
		@@ -102,7 +102,12 @@ claude mcp add --transport http __TITLE__ https://<app>/mcp
102	102	```
103	103
104	104	The page stays current over `/events`: a proposal someone else opens or
105		withdraws appears without a reload. Without JavaScript it is the same page, one
	105	withdraws appears without a reload — including one written at the terminal,
	106	which is another process on the same file. Every event is a row in `event`,
	107	so the server follows what it did not write itself while anyone is
	108	watching, and a browser that reconnects with `Last-Event-ID` is replayed
	109	from those rows even across a restart. The rows are trimmed to the last
	110	hour and the last thousand. Without JavaScript it is the same page, one
106	111	reload behind.
107	112
108	113	The terminal opens with the exact command in `FORGE_APP_SSH` when the app
		@@ -12,9 +12,12 @@ import {
12	12	type AdminHealth,
13	13	} from "./actions.ts";
14	14	import { migrations, openDb } from "./db.ts";
	15	import { useDb } from "./live.ts";
15	16	import type { Who } from "./who.ts";
16	17
17	18	const db = openDb(mkdtempSync(join(tmpdir(), "app-actions-")));
	19	// What an action emits belongs in this test's database, not the app's.
	20	useDb(db);
18	21	const anon: Who = { role: "anon" };
19	22	const alice: Who = { role: "user", id: "u1", name: "alice" };
20	23	const bob: Who = { role: "user", id: "u2", name: "bob" };
		@@ -49,6 +49,15 @@ export const migrations: string[] = [
49	49	// The template's note is gone: a proposal is what this app is about.
50	50	// Migration 1 stays exactly as it ran; this is how history is undone.
51	51	`DROP TABLE note`,
	52	// The live log (`live.ts`): every emitted event, so a write by another
	53	// process on this file reaches the pages this one is serving. Trimmed to
	54	// the last hour and the last thousand rows; the rowid is the event id.
	55	`CREATE TABLE event (
	56	id INTEGER PRIMARY KEY,
	57	type TEXT NOT NULL,
	58	data TEXT NOT NULL,
	59	created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
	60	)`,
52	61	];
53	62
54	63	export function openDb(dir = dataDir): Database {
		@@ -0,0 +1,237 @@
	1	import { afterAll, expect, test } from "bun:test";
	2	import { mkdtempSync } from "node:fs";
	3	import { tmpdir } from "node:os";
	4	import { join } from "node:path";
	5	import { openDb } from "./db.ts";
	6	import { emit, events, following, subscribe, trim } from "./live.ts";
	7	import { serve } from "./server.ts";
	8
	9	const dir = mkdtempSync(join(tmpdir(), "app-live-"));
	10	const server = serve({ port: 0, dataDir: dir, identityKey: "" });
	11	const base = `http://${server.hostname}:${server.port}`;
	12	afterAll(() => server.stop(true));
	13
	14	/** A second connection to the same file, which is all another process is. */
	15	const other = openDb(dir);
	16	const write = (type: string, data: unknown): number =>
	17	other
	18	.query<
	19	{ id: number },
	20	[string, string]
	21	>("INSERT INTO event (type, data) VALUES (?1, ?2) RETURNING id")
	22	.get(type, JSON.stringify(data))!.id;
	23
	24	/**
	25	* A stream, read frame by frame, closed the way a browser closes one. The
	26	* route is `events(request)`; opening it here rather than over HTTP is what
	27	* lets a test close it and see the server let go — Bun holds a cancelled
	28	* fetch's socket in its pool, so the server never learns it was dropped.
	29	*/
	30	function open(since?: number \| string) {
	31	const aborter = new AbortController();
	32	const url = since === undefined ? "/events" : `/events?since=${since}`;
	33	const response = events(
	34	new Request(`http://live.test${url}`, { signal: aborter.signal }),
	35	);
	36	const reader = response.body!.getReader();
	37	const decoder = new TextDecoder();
	38	let buffer = "";
	39	let reading: Promise<void> \| undefined;
	40	// One read at a time, and a chunk that arrives after we stopped waiting
	41	// is still kept: waiting is what times out, never the stream.
	42	const pump = async (within: number): Promise<void> => {
	43	reading ??= reader.read().then(({ value }) => {
	44	reading = undefined;
	45	if (value !== undefined) buffer += decoder.decode(value);
	46	});
	47	await Promise.race([reading, Bun.sleep(within)]);
	48	};
	49	return {
	50	response,
	51	async next(within = 3000): Promise<{ id: number; type: string; data: any }> {
	52	const deadline = Date.now() + within;
	53	for (;;) {
	54	const frame = /id: (\d+)\ndata: (.*)\n\n/.exec(buffer);
	55	if (frame) {
	56	buffer = buffer.slice(frame.index + frame[0].length);
	57	const { type, data } = JSON.parse(frame[2]!);
	58	return { id: Number(frame[1]), type, data };
	59	}
	60	if (Date.now() > deadline) throw new Error(`no event: ${buffer}`);
	61	await pump(100);
	62	}
	63	},
	64	/** Everything the stream has to say right now. */
	65	async drain(ms = 100): Promise<string> {
	66	for (;;) {
	67	const before = buffer.length;
	68	await pump(ms);
	69	if (buffer.length === before) return buffer;
	70	}
	71	},
	72	seen: () => buffer,
	73	/** Whether the server has closed its end. */
	74	ended: async (): Promise<boolean> => {
	75	for (;;) {
	76	const { done, value } = await reader.read();
	77	if (done) return true;
	78	buffer += decoder.decode(value);
	79	}
	80	},
	81	close: () => aborter.abort(),
	82	};
	83	}
	84
	85	test("an event emitted in this process reaches a subscriber at once", () => {
	86	const heard: { id: number; type: string; data: unknown }[] = [];
	87	const unsubscribe = subscribe((event) => heard.push(event));
	88	const event = emit("test.ping", { n: 1 });
	89	unsubscribe();
	90	expect(heard).toEqual([{ id: event.id, type: "test.ping", data: { n: 1 } }]);
	91	expect(event.id).toBeGreaterThan(0);
	92	// And it is a row, so another process can read it back.
	93	expect(
	94	other
	95	.query<
	96	{ type: string; data: string },
	97	[number]
	98	>("SELECT type, data FROM event WHERE id = ?1")
	99	.get(event.id),
	100	).toEqual({ type: "test.ping", data: '{"n":1}' });
	101	// Nobody is listening now, and the ids still climb.
	102	expect(emit("test.ping", null).id).toBe(event.id + 1);
	103	});
	104
	105	test("an event another connection writes reaches the stream within a tick", async () => {
	106	const stream = open();
	107	const id = write("proposal.opened", { id: 7, title: "from the terminal" });
	108	expect(await stream.next()).toEqual({
	109	id,
	110	type: "proposal.opened",
	111	data: { id: 7, title: "from the terminal" },
	112	});
	113	// A row that will not parse is skipped, never thrown, and does not stall
	114	// the rows behind it.
	115	other
	116	.query("INSERT INTO event (type, data) VALUES ('proposal.opened', '{')")
	117	.run();
	118	const next = write("proposal.withdrawn", { id: 7 });
	119	expect((await stream.next()).id).toBe(next);
	120	stream.close();
	121	});
	122
	123	test("a stream with no id starts from now; `since` replays only what follows", async () => {
	124	const first = emit("test.one", 1);
	125	const second = emit("test.two", 2);
	126	const fresh = open();
	127	const replay = open(first.id);
	128	expect(await replay.next()).toEqual({
	129	id: second.id,
	130	type: "test.two",
	131	data: 2,
	132	});
	133	// What came before `since` is not replayed, and the stream that asked
	134	// for nothing is given nothing at all.
	135	expect(await replay.drain()).not.toContain("test.one");
	136	const opening = await fresh.drain();
	137	expect(opening).toContain("retry:");
	138	expect(opening).not.toContain("test.one");
	139	expect(opening).not.toContain("test.two");
	140	// Both hear what comes next.
	141	const third = emit("test.three", 3);
	142	expect((await fresh.next()).id).toBe(third.id);
	143	expect((await replay.next()).id).toBe(third.id);
	144	fresh.close();
	145	replay.close();
	146	// A `since` that is not a whole number is no `since` at all.
	147	const nonsense = open("yesterday");
	148	expect(await nonsense.drain()).not.toContain("test.three");
	149	nonsense.close();
	150	});
	151
	152	test("the trim keeps the newest and drops the oldest", () => {
	153	const count = (): number =>
	154	other.query<{ n: number }, []>("SELECT count(*) AS n FROM event").get()!.n;
	155	const top = (): number =>
	156	other.query<{ id: number }, []>("SELECT max(id) AS id FROM event").get()!
	157	.id;
	158	const ancient = write("test.ancient", 0);
	159	other
	160	.query("UPDATE event SET created_at = '2001-01-01T00:00:00.000Z' WHERE id = ?1")
	161	.run(ancient);
	162	other.transaction(() => {
	163	for (let i = 0; i < 1100; i++) write("test.bulk", i);
	164	})();
	165	const newest = top();
	166	trim();
	167	expect(count()).toBeLessThanOrEqual(1000);
	168	expect(top()).toBe(newest);
	169	expect(
	170	other
	171	.query<{ id: number }, [number]>("SELECT id FROM event WHERE id = ?1")
	172	.get(ancient),
	173	).toBeNull();
	174	// The newest row survives even when everything is older than an hour, so
	175	// the next id is still higher than any a client holds.
	176	other.query("UPDATE event SET created_at = '2001-01-01T00:00:00.000Z'").run();
	177	trim();
	178	expect(count()).toBe(1);
	179	expect(top()).toBe(newest);
	180	expect(emit("test.after-trim", null).id).toBe(newest + 1);
	181	});
	182
	183	test("a closed stream unsubscribes, and the last one stops the tick", async () => {
	184	expect(following()).toBe(false);
	185	const one = open();
	186	const two = open();
	187	expect(following()).toBe(true);
	188	const id = write("test.watched", null);
	189	expect((await one.next()).id).toBe(id);
	190	// One of two closing leaves the timer where it is; the second stops it.
	191	one.close();
	192	expect(following()).toBe(true);
	193	two.close();
	194	expect(following()).toBe(false);
	195	// Both ends are done, and what is written now reaches neither of them.
	196	const after = write("test.unwatched", null);
	197	expect(await one.ended()).toBe(true);
	198	expect(await two.ended()).toBe(true);
	199	expect(two.seen()).not.toContain(`id: ${after}`);
	200	expect(following()).toBe(false);
	201	});
	202
	203	test("a write by another process reaches an open page, through the same action", async () => {
	204	const stream = await fetch(base + "/events");
	205	const reader = stream.body!.getReader();
	206	await reader.read(); // the retry line: the stream is open and following
	207	// What the terminal is: another process, the same actions, the same
	208	// `emit`, over the same file.
	209	const child = Bun.spawnSync({
	210	cmd: [
	211	"bun",
	212	"-e",
	213	`import { propose } from ${JSON.stringify(join(import.meta.dir, "actions.ts"))};
	214	import { openDb } from ${JSON.stringify(join(import.meta.dir, "db.ts"))};
	215	const opened = propose(openDb(), { role: "user", id: "u1", name: "tty" }, {
	216	title: "typed at the terminal",
	217	body: "Made by another process entirely.",
	218	options: ["Yes", "No"],
	219	});
	220	if (!opened.ok) throw new Error(opened.message);`,
	221	],
	222	env: { ...process.env, FORGE_APP_DATA: dir },
	223	});
	224	expect(child.stderr.toString()).toBe("");
	225	expect(child.exitCode).toBe(0);
	226	let buffer = "";
	227	const deadline = Date.now() + 3000;
	228	while (!buffer.includes("proposal.opened") && Date.now() < deadline) {
	229	const read = await Promise.race([
	230	reader.read(),
	231	Bun.sleep(3000).then(() => ({ value: undefined })),
	232	]);
	233	buffer += new TextDecoder().decode(read.value);
	234	}
	235	expect(buffer).toContain("typed at the terminal");
	236	await reader.cancel();
	237	});
		@@ -1,44 +1,194 @@
1	1	/**
2		* Realtime without a framework. Actions `emit` what changed; `GET /events`
3		* streams it to every open page as server-sent events, numbered, with the
4		* last hundred kept so a client that reconnects with `Last-Event-ID` (or
5		* `?since=`) hears what it missed. `static/live.js` turns the stream into
6		* DOM events. The bus is this process's: the terminal is another process,
7		* and its writes show on the next load rather than live.
	2	* Realtime without a framework. An action `emit`s what changed: the event
	3	* is appended to the `event` table and handed to this process's listeners
	4	* in the same breath, so a page served here waits on nothing. `GET /events`
	5	* streams it to every open page as server-sent events, numbered by the
	6	* row's id.
	7	*
	8	* The table is what makes the terminal live: `src/tty.ts` is another
	9	* process on the same file, and while anyone is watching, the server
	10	* follows the rows it did not write itself — one timer for the whole
	11	* process, a primary-key read every quarter second that asks for nothing
	12	* when the highest id has not moved. The page still polls nothing.
	13	*
	14	* Because the ids are the table's, they keep climbing across a restart: a
	15	* client that reconnects with `Last-Event-ID` (or `?since=`) is replayed
	16	* from the rows, never from a ring this process forgot.
8	17	*/
9	18
	19	import type { Database } from "bun:sqlite";
	20	import { openDb } from "./db.ts";
	21
10	22	export interface LiveEvent {
11	23	readonly id: number;
12	24	readonly type: string;
13	25	readonly data: unknown;
14	26	}
15	27
16		const KEEP = 100;
	28	/** How often the server looks for rows another process wrote. */
	29	const TICK_MS = 250;
17	30	const HEARTBEAT_MS = 25_000;
18		const kept: LiveEvent[] = [];
19		const listeners = new Set<(event: LiveEvent) => void>();
20		let last = 0;
	31	/** The most a reconnecting client is replayed in one go. */
	32	const REPLAY_MAX = 1000;
	33	/** What survives a trim: an hour, and a thousand rows. */
	34	const KEEP_ROWS = 1000;
	35	const KEEP_MS = 60 * 60 * 1000;
	36	/** A burst of writes trims once, after the responses, not inside them. */
	37	const TRIM_AFTER_MS = 5_000;
21	38
	39	interface Row {
	40	readonly id: number;
	41	readonly type: string;
	42	readonly data: string;
	43	}
	44
	45	const listeners = new Set<(event: LiveEvent) => void>();
	46	let store: Database \| undefined;
	47	/** The highest id handed to the listeners. Only ever climbs. */
	48	let sent = 0;
	49	let tick: ReturnType<typeof setInterval> \| undefined;
	50	let trimming: ReturnType<typeof setTimeout> \| undefined;
	51
	52	/**
	53	* The database the log is kept in. A door that opened one says so at boot
	54	* (`server.ts`); a process that did not — the terminal — falls back to the
	55	* app's own file, which is the same file.
	56	*/
	57	export function useDb(db: Database): void {
	58	store = db;
	59	sent = highest();
	60	}
	61
	62	const db = (): Database => (store ??= openDb());
	63
	64	const highest = (): number =>
	65	db()
	66	.query<{ id: number \| null }, []>("SELECT max(id) AS id FROM event")
	67	.get()?.id ?? 0;
	68
	69	/** Append the event, then tell this process's listeners at once. */
22	70	export function emit(type: string, data: unknown): LiveEvent {
23		const event: LiveEvent = { id: ++last, type, data };
24		kept.push(event);
25		if (kept.length > KEEP) kept.shift();
26		for (const listen of listeners) listen(event);
	71	const row = db()
	72	.query<
	73	{ id: number },
	74	[string, string]
	75	>("INSERT INTO event (type, data) VALUES (?1, ?2) RETURNING id")
	76	.get(type, JSON.stringify(data) ?? "null")!;
	77	const event: LiveEvent = { id: row.id, type, data };
	78	deliver(event);
	79	scheduleTrim();
27	80	return event;
28	81	}
29	82
	83	/** Every event goes out exactly once, in id order, whoever wrote it. */
	84	function deliver(event: LiveEvent): void {
	85	if (event.id <= sent) return;
	86	sent = event.id;
	87	for (const listen of listeners) listen(event);
	88	}
	89
	90	/** A row as an event, or nothing at all: data that will not parse is skipped, never thrown. */
	91	function parse(row: Row): LiveEvent \| undefined {
	92	try {
	93	return { id: row.id, type: row.type, data: JSON.parse(row.data) };
	94	} catch {
	95	return undefined;
	96	}
	97	}
	98
	99	const after = (id: number): Row[] =>
	100	db()
	101	.query<
	102	Row,
	103	[number, number]
	104	>("SELECT id, type, data FROM event WHERE id > ?1 ORDER BY id LIMIT ?2")
	105	.all(id, REPLAY_MAX);
	106
	107	/** One pass of the follower: what another process wrote since we last looked. */
	108	function follow(): void {
	109	try {
	110	for (const row of after(sent)) {
	111	const event = parse(row);
	112	if (event === undefined) sent = Math.max(sent, row.id);
	113	else deliver(event);
	114	}
	115	} catch (error) {
	116	console.error(error);
	117	}
	118	}
	119
30	120	export function subscribe(listen: (event: LiveEvent) => void): () => void {
	121	if (listeners.size === 0) startFollowing();
31	122	listeners.add(listen);
32		return () => listeners.delete(listen);
	123	return () => {
	124	listeners.delete(listen);
	125	if (listeners.size === 0) stopFollowing();
	126	};
	127	}
	128
	129	/**
	130	* The follower runs while someone is listening and not a moment longer:
	131	* one timer for the process, started by the first stream and stopped by
	132	* the last. It starts from the highest id there is — what happened while
	133	* nobody was watching is history, and a client that wants it asks with
	134	* `since`.
	135	*/
	136	export function startFollowing(): void {
	137	if (tick !== undefined) return;
	138	sent = Math.max(sent, highest());
	139	tick = setInterval(follow, TICK_MS);
	140	tick.unref?.();
	141	}
	142
	143	export function stopFollowing(): void {
	144	if (tick === undefined) return;
	145	clearInterval(tick);
	146	tick = undefined;
	147	}
	148
	149	/** Whether the follower's timer is running. */
	150	export const following = (): boolean => tick !== undefined;
	151
	152	/**
	153	* Keep an hour, keep a thousand, and always keep the newest row — with the
	154	* table empty SQLite would hand out id 1 again, and ids a client holds
	155	* must never come back around.
	156	*/
	157	export function trim(now = Date.now()): void {
	158	const top = highest();
	159	if (top === 0) return;
	160	const cutoff = new Date(now - KEEP_MS).toISOString();
	161	db().query("DELETE FROM event WHERE id <= ?1").run(top - KEEP_ROWS);
	162	db()
	163	.query("DELETE FROM event WHERE id < ?1 AND created_at < ?2")
	164	.run(top, cutoff);
	165	}
	166
	167	function scheduleTrim(): void {
	168	if (trimming !== undefined) return;
	169	trimming = setTimeout(() => {
	170	trimming = undefined;
	171	try {
	172	trim();
	173	} catch (error) {
	174	console.error(error);
	175	}
	176	}, TRIM_AFTER_MS);
	177	trimming.unref?.();
33	178	}
34	179
35	180	/** `GET /events`: a stream that replays what `since` missed, then follows. */
36	181	export function events(request: Request): Response {
37		const since = Number(
	182	// A client says where it got to, or says nothing and starts from now —
	183	// the page it is on was rendered a moment ago. Anything but a whole
	184	// number is nothing.
	185	const asked =
38	186	request.headers.get("last-event-id") ??
39		new URL(request.url).searchParams.get("since") ??
40		0,
41		);
	187	new URL(request.url).searchParams.get("since");
	188	const since =
	189	asked === null \|\| !/^\d+$/.test(asked.trim())
	190	? undefined
	191	: Number(asked.trim());
42	192	const encoder = new TextEncoder();
43	193	let stop = (): void => {};
44	194	const stream = new ReadableStream<Uint8Array>({
		@@ -65,7 +215,13 @@ export function events(request: Request): Response {
65	215	};
66	216	request.signal.addEventListener("abort", stop);
67	217	write("retry: 2000\n\n");
68		for (const event of kept) if (event.id > since) send(event);
	218	// Nothing can arrive between subscribing and replaying: this runs to
	219	// the end before any other turn of the loop.
	220	if (since !== undefined)
	221	for (const row of after(since)) {
	222	const event = parse(row);
	223	if (event !== undefined) send(event);
	224	}
69	225	},
70	226	cancel() {
71	227	stop();
		@@ -16,6 +16,7 @@ import {
16	16	where,
17	17	} from "./env.ts";
18	18	import { allow, html, sameOrigin, text } from "./http.ts";
	19	import { useDb } from "./live.ts";
19	20	import { match } from "./routes.ts";
20	21	import { who } from "./who.ts";
21	22
		@@ -28,6 +29,11 @@ export interface ServeOptions {
28	29
29	30	export function serve(options: ServeOptions) {
30	31	const db = openDb(options.dataDir);
	32	// The live log is kept in this database: `emit` appends to it, and while
	33	// a stream is open the follower reads back what another process — the
	34	// terminal — wrote. The first stream starts that timer and the last one
	35	// stops it, so an idle server does nothing at all.
	36	useDb(db);
31	37	// The fact that identities are checked travels with the request; the key
32	38	// itself stays here, where `who` reads it and nothing else can.
33	39	const identity = identityState(options.identityKey);