From 3c7a9ca0e7f1305c5b3a2606179a944a53ca97fd Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Mon, 2 Nov 2026 22:47:48 +0000 Subject: [PATCH] feat(chord): example 59-chat keeps its messages in a table --- .../chord/test/examples/59-chat/app/v1.ts | 19 ++++- packages/chord/test/examples/59-chat/main.ts | 34 ++++++-- .../examples/59-chat/plugins/chat/server.ts | 2 +- .../examples/59-chat/plugins/chat/worker.ts | 83 +++++++++++++------ .../chord/test/examples/59-chat/tui/main.ts | 46 +++++++--- .../chord/test/examples/59-chat/web/main.ts | 18 +++- 6 files changed, 150 insertions(+), 52 deletions(-) diff --git a/packages/chord/test/examples/59-chat/app/v1.ts b/packages/chord/test/examples/59-chat/app/v1.ts index 53c3cbeed..56a13b090 100644 --- a/packages/chord/test/examples/59-chat/app/v1.ts +++ b/packages/chord/test/examples/59-chat/app/v1.ts @@ -1,6 +1,7 @@ // The chat's API, version 1, shared by every host kind (`@chat/app/v1`): the part of Chord its plugins and clients use, // and the chat's own contracts. A channel is a worker of its own (`Channel({ id })`); its `Room` lives there, with the -// messages in the worker's store and the attachments in its files. The server keeps the channel list (`Channels`). +// messages in a table of the worker's store (the newest live, older ones by page) and the attachments in its files. +// The server keeps the channel list (`Channels`). import { arr, bytes, hostKind, int, method, num, obj, opt, remote, type Static, str } from "../../../../src/index.ts"; @@ -9,12 +10,16 @@ export { arr, constant, defineFacet, - defineState, + defineTable, + int, + json, local, + lt, method, num, obj, opt, + query, str, } from "../../../../src/index.ts"; @@ -24,6 +29,9 @@ export const Channel = hostKind("channel", { id: str() }); /** A file shared in a message: its id in the channel's files, its name, type and size. */ export const Attachment = obj({ id: str(), name: str(), mime: str(), size: int() }); +/** An attachment (see `Attachment`). */ +export type Attachment = Static; + /** A message: its number, author, text, time, and maybe a file. */ export const Message = obj({ id: int(), author: str(), text: str(), at: num(), file: opt(Attachment) }); @@ -31,8 +39,9 @@ export const Message = obj({ id: int(), author: str(), text: str(), at: num(), f export type Message = Static; /** - * A channel's room, on its worker: its messages, posting, and sharing a file in chunks (`begin`, `append` at the - * stored offset, `share` posts it) and reading one back (`read`, at most 4 MiB a call). + * A channel's room, on its worker: its newest 100 messages (live), the 100 before a message (`history`, the newest + * without `before`), posting, sharing a file in chunks (`begin`, `append` at the stored offset, `share` posts it), and + * a shared file's details (`attachment`) and bytes (`read`, at most 4 MiB a call). */ export const Room = remote("chat.room", { version: 1, @@ -42,10 +51,12 @@ export const Room = remote("chat.room", { main: { state: { messages: arr(Message) }, methods: { + history: method(obj({ before: opt(int()) }), arr(Message)), post: method(str()), begin: method(obj({ name: str(), mime: str() }), str()), append: method(obj({ upload: str(), offset: int(), bytes: bytes() })), share: method(obj({ upload: str(), text: str() })), + attachment: method(str(), Attachment), read: method(obj({ id: str(), offset: int(), length: int() }), bytes()), }, }, diff --git a/packages/chord/test/examples/59-chat/main.ts b/packages/chord/test/examples/59-chat/main.ts index 96abed77b..7e977239b 100644 --- a/packages/chord/test/examples/59-chat/main.ts +++ b/packages/chord/test/examples/59-chat/main.ts @@ -1,8 +1,9 @@ // A chat (slice 11 with the app kit): one worker per channel (`Channel({ id })`), each channel's `Room` on its worker -// with the messages in the worker's store and the attachments in its files (the kit's `Files`, sent in chunks); the -// server keeps the channel list and serves the page (`/chat`, after picking a name at `/`) and the terminal client. A -// browser and a terminal are live in the same room. Here a user who picked a name ("ada") joins over the same Chord -// link a page uses, and with Node 26.5.1 or later the owner's terminal client joins too. +// with every message in a table of the worker's store (the newest 100 live, older ones by page: `history`) and the +// attachments in its files (the kit's `Files`, sent in chunks); the server keeps the channel list and serves the page +// (`/chat`, after picking a name at `/`) and the terminal client. A browser and a terminal are live in the same room. +// Here a user who picked a name ("ada") joins over the same Chord link a page uses, and with Node 26.5.1 or later the +// owner's terminal client joins too. // Run from packages/chord (`--serve` keeps it running: open the printed URL, and run the terminal with 57-tui's // `terminal.ts` and the owner token file): // node --conditions=source --experimental-strip-types test/examples/59-chat/main.ts @@ -58,14 +59,23 @@ console.log( ); console.log(" one worker per channel:", Object.keys(app.workers.get("channel")!.state.value).join(", ")); -// 4. The owner's terminal client in the same room (Node 26.5.1 or later): it shows ada's messages; its post and an -// attached file (the launcher asks before it reads the file) reach ada at once. +// 4. Every message is kept in the channel's table: the room shows the newest 100 live and pages older ones. +for (let n = 1; n <= 120; n++) await random.post(`message ${n}`, context); +while (random.messages.value.at(-1)?.text !== "message 120") await new Promise((resolve) => setTimeout(resolve, 20)); +const live = random.messages.value; +const older = await random.history({ before: live[0]!.id }, context); +console.log( + `4. #random keeps ${older.length + live.length} messages: ${live.length} live (from "${live[0]!.text}"), ${older.length} older by history (from "${older[0]!.text}")`, +); + +// 5. The owner's terminal client in the same room (Node 26.5.1 or later): it shows ada's messages; its post and an +// attached file (the launcher asks before it reads the file) reach ada at once; PgUp in #random loads older ones. let sandboxed = true; try { requireSandboxNode(); } catch (error) { sandboxed = false; - console.log(`4. (${(error as Error).message})`); + console.log(`5. (${(error as Error).message})`); } if (sandboxed) { const input = new PassThrough(); @@ -86,7 +96,7 @@ if (sandboxed) { }; await shows("[image: dot.png]"); console.log( - "4. the terminal shows:", + "5. the terminal shows:", JSON.stringify( screen .split("\x1b[2J\x1b[H") @@ -113,6 +123,14 @@ if (sandboxed) { .slice(-2) .map((message) => `${message.author}: ${message.text || message.file?.name}`); console.log(" ada sees:", last.join(" | ")); + // #random, then PgUp to its first message: the terminal loads the page before the live window. + input.write("\t"); + for (let n = 0; n < 8; n++) { + input.write("\x1b[5~"); + await new Promise((resolve) => setTimeout(resolve, 50)); + } + await shows("ada anyone here?"); + console.log(" PgUp in #random reaches its first message: anyone here?"); input.write("\x03"); await terminal.exited; } diff --git a/packages/chord/test/examples/59-chat/plugins/chat/server.ts b/packages/chord/test/examples/59-chat/plugins/chat/server.ts index 6f3cf89d7..968a0f399 100644 --- a/packages/chord/test/examples/59-chat/plugins/chat/server.ts +++ b/packages/chord/test/examples/59-chat/plugins/chat/server.ts @@ -28,7 +28,7 @@ export default defineFacet((host, self) => { "GET /": { public: true, handler: async (request) => ({ kind: "html", body: signIn(request.nonce) }) }, "GET /files/:channel/:id": async (request, ctx) => { const room = await ctx.get(Room, { id: request.params.channel! }); - const file = room.messages.value.find((message) => message.file?.id === request.params.id)?.file; + const file = await room.attachment(request.params.id!, BACKGROUND_CONTEXT).catch(() => undefined); if (file === undefined) return { kind: "error", status: 404 }; return { kind: "bytes", diff --git a/packages/chord/test/examples/59-chat/plugins/chat/worker.ts b/packages/chord/test/examples/59-chat/plugins/chat/worker.ts index 7f78cf8cb..90d3e02a6 100644 --- a/packages/chord/test/examples/59-chat/plugins/chat/worker.ts +++ b/packages/chord/test/examples/59-chat/plugins/chat/worker.ts @@ -1,58 +1,79 @@ -// The chat plugin on a channel's worker: the room's messages in a document of the worker's store (kept by a local -// object: a persisted state needs a stable identity), and each principal's `Room` posting as them and sharing files -// through the kit's `Files` of this worker. +// The chat plugin on a channel's worker: the room's messages in a table of the worker's store (every message kept), +// written and read by one local object. Tables have no watch, so the object keeps the live window itself: the newest +// 100 messages, read at activation and appended to after each post (it is the only writer); older ones are read by +// page on request. Each principal's `Room` posts as them and shares files through the kit's `Files` of this worker. import { Attachment, arr, defineFacet, - defineState, + defineTable, + int, + json, local, + lt, Message, method, num, obj, opt, + query, Room, str, } from "@chat/app/v1"; import { Files } from "@chat/app/v1/worker"; -/** The newest messages a channel keeps. */ -const KEPT = 500; - -const Log = defineState("chat.log", { - schema: obj({ next: num(), messages: arr(Message) }), +/** Every message of the channel, by its number. */ +const Messages = defineTable("chat.messages", { version: 1, - initial: () => ({ next: 1, messages: [] }), - persist: true, - checkpointWhen: (_value, _ops, { deltasSinceBase }) => deltasSinceBase >= 50, + columns: { id: int(), author: str(), text: str(), at: num(), file: opt(json(Attachment, { version: 1 })) }, + key: ["id"], }); +/** The messages, newest first. */ +const NEWEST = query(Messages).orderBy((message) => message.id, "desc"); + +/** How many messages the live window and a page of history hold. */ +const PAGE = 100; + const Store = local("chat.store", { version: 1, params: {}, - members: { messages: arr(Message), add: method(obj({ author: str(), text: str(), file: opt(Attachment) })) }, + members: { + messages: arr(Message), + add: method(obj({ author: str(), text: str(), file: opt(Attachment) })), + history: method(obj({ before: opt(int()) }), arr(Message)), + }, }); export default defineFacet((host) => { host.serve(Store, (_params, self) => { - const log = self.state(Log); + const messages = self.table(Messages); + // The newest messages (oldest first) and the next message's number: read once, then kept here. + const window = self.writable([]); + let next = 1; + self.open( + async (ctx) => { + const { items } = await messages.page(NEWEST, PAGE, ctx); + window.set([...items].reverse()); + next = (items[0]?.id ?? 0) + 1; + }, + () => {}, + ); return { - messages: self.derive([log], ([value]) => value.messages), + messages: self.derive([window], ([value]) => value), add: async ({ author, text, file }, ctx) => { - await log.change((draft) => { - draft.messages.push({ - id: draft.next++, - author, - text, - at: Date.now(), - ...(file === undefined ? {} : { file }), - }); - if (draft.messages.length > KEPT) draft.messages.splice(0, draft.messages.length - KEPT); - }, ctx); + const message = { id: next++, author, text, at: Date.now(), ...(file === undefined ? {} : { file }) }; + await self.tx(async (tx) => tx.put(messages, message), ctx); + // In number order: two posts may commit in either order. + window.set([...window.value, message].sort((a, b) => a.id - b.id).slice(-PAGE)); return null; }, + history: async ({ before }, ctx) => { + const older = before === undefined ? NEWEST : NEWEST.where((message) => lt(message.id, before)); + const { items } = await messages.page(older, PAGE, ctx); + return [...items].reverse(); + }, }; }); host.serve(Room, (_params, principal, self) => { @@ -63,16 +84,24 @@ export default defineFacet((host) => { variant: "main", members: { messages: self.derive([store.messages], ([messages]) => messages), + history: (page, ctx) => store.history(page, ctx), post: (text, ctx) => store.add({ author, text }, ctx), begin: (file, ctx) => files.begin(file, ctx), append: (chunk, ctx) => files.append(chunk, ctx), share: async ({ upload, text }, ctx) => { const id = await files.finish(upload, ctx); - const { name, mime, size } = files.list.value.find((file) => file.id === id)!; - return store.add({ author, text, file: { id, name, mime, size } }, ctx); + return store.add({ author, text, file: attachment(files.list.value, id) }, ctx); }, + attachment: async (id) => attachment(files.list.value, id), read: (range, ctx) => files.read(range, ctx), }, }; }); }); + +/** The attachment `id` of the channel's files; throws when there is none. */ +function attachment(list: readonly Attachment[], id: string): Attachment { + const file = list.find((each) => each.id === id); + if (file === undefined) throw new Error(`no file ${id}`); + return { id, name: file.name, mime: file.mime, size: file.size }; +} diff --git a/packages/chord/test/examples/59-chat/tui/main.ts b/packages/chord/test/examples/59-chat/tui/main.ts index 6d70cd439..2e997f151 100644 --- a/packages/chord/test/examples/59-chat/tui/main.ts +++ b/packages/chord/test/examples/59-chat/tui/main.ts @@ -1,7 +1,7 @@ // The chat in a terminal (the terminal base bundle, run in the launcher's sandbox): the channels (Tab switches), the -// current room's messages (an image as a placeholder, other files as their link), and a line to type into. Enter -// posts; `/attach ` asks the launcher for a file of this machine (it asks the user) and shares it in chunks; -// Ctrl-C quits. +// current room's messages (an image as a placeholder, other files as their link; PgUp and PgDn scroll, loading older +// messages a page at a time), and a line to type into. Enter posts; `/attach ` asks the launcher for a file of +// this machine (it asks the user) and shares it in chunks; Ctrl-C quits. import type { Held } from "../../../../src/index.ts"; import { type AttachedFile, startTerminalRuntime, TuiScope } from "../../../../src/tui/index.ts"; @@ -42,28 +42,51 @@ void startTerminalRuntime({ const { terminal, theme } = runtime; const scope = new TuiScope(runtime.host, () => draw()); const channels = scope.watch(scope.hold(Channels, {}).names); - // Each channel's room held once, on first view. - const rooms = new Map; messages: () => readonly v1.Message[] | undefined }>(); - const roomOf = (id: string) => { + // Each channel's room held once, on first view, with the older messages loaded so far. + type Viewed = { + room: Held; + live: () => readonly v1.Message[] | undefined; + older: readonly v1.Message[]; + }; + const rooms = new Map(); + const roomOf = (id: string): Viewed => { let held = rooms.get(id); if (held === undefined) { const room = scope.hold(Room, { id }); - held = { room, messages: scope.watch(room.messages) }; + held = { room, live: scope.watch(room.messages), older: [] }; rooms.set(id, held); } return held; }; + const messagesOf = ({ live, older }: Viewed): readonly v1.Message[] => [ + ...older, + ...(live() ?? []).filter((message) => message.id > (older.at(-1)?.id ?? 0)), + ]; let channel = "general"; let draft = ""; + // How many messages up from the newest the view is scrolled. + let scrolled = 0; + const height = () => terminal.rows - 3; const draw = (): void => { const tabs = (channels() ?? []).map((name) => name === channel ? theme.fg("accent", `#${name}`) : `#${name}`, ); - const shown = (roomOf(channel).messages() ?? []) - .slice(-(terminal.rows - 3)) - .map((message) => line(message, channel, theme)); + const messages = messagesOf(roomOf(channel)); + const end = messages.length - scrolled; + const shown = messages.slice(Math.max(0, end - height()), end).map((message) => line(message, channel, theme)); terminal.write(`\x1b[2J\x1b[H${tabs.join(" ")}\r\n${shown.join("\r\n")}\r\n> ${draft}`); }; + // PgUp: a screen further back, first loading the page before the oldest message when the screen would pass it. + const scrollUp = async (): Promise => { + const viewed = roomOf(channel); + const messages = messagesOf(viewed); + if (scrolled + 2 * height() > messages.length && messages.length > 0) { + const page = await viewed.room.history({ before: messages[0]!.id }, context); + viewed.older = [...page, ...viewed.older]; + } + scrolled = Math.min(scrolled + height(), Math.max(0, messagesOf(viewed).length - height())); + draw(); + }; const send = (text: string): void => { if (text.startsWith("/attach ")) { const { room } = roomOf(channel); @@ -74,9 +97,12 @@ void startTerminalRuntime({ }; terminal.onInput((data) => { if (data === "\x03") runtime.exit(0); + else if (data === "\x1b[5~") void scrollUp(); + else if (data === "\x1b[6~") scrolled = Math.max(0, scrolled - height()); else if (data === "\t") { const names = channels() ?? []; channel = names[(names.indexOf(channel) + 1) % names.length] ?? channel; + scrolled = 0; } else if (data === "\r") { send(draft); draft = ""; diff --git a/packages/chord/test/examples/59-chat/web/main.ts b/packages/chord/test/examples/59-chat/web/main.ts index fe21d9d9d..71f48310c 100644 --- a/packages/chord/test/examples/59-chat/web/main.ts +++ b/packages/chord/test/examples/59-chat/web/main.ts @@ -1,5 +1,6 @@ // The chat's page (the web base bundle): the channels as tabs, the current room's messages (images inline, other files -// as links), a message box and a file picker. A file goes up in chunks through the room (into its worker's files). +// as links; older ones a page at a time with "Load older"), a message box and a file picker. A file goes up in chunks +// through the room (into its worker's files). import * as React from "react"; import { createElement as h, useState } from "react"; @@ -40,7 +41,17 @@ function Message({ channel, message }: { channel: string; message: v1.Message }) function RoomView({ channel }: { channel: string }) { const room = useHold(Room, { id: channel }); - const messages = useWatch(room?.messages) ?? []; + const live = useWatch(room?.messages) ?? []; + // Pages of history before the live window, and whether there may be more. + const [older, setOlder] = useState([]); + const [more, setMore] = useState(true); + const messages = [...older, ...live.filter((message) => message.id > (older.at(-1)?.id ?? 0))]; + const loadOlder = async () => { + if (room === undefined || messages.length === 0) return; + const page = await room.history({ before: messages[0]!.id }, context); + setOlder([...page, ...older]); + setMore(page.length === 100); + }; const [text, setText] = useState(""); const send = (event: { preventDefault(): void }) => { event.preventDefault(); @@ -51,6 +62,9 @@ function RoomView({ channel }: { channel: string }) { return h( "section", null, + more && live.length === 100 + ? h("button", { type: "button", onClick: () => void loadOlder() }, "Load older") + : null, h("ul", null, ...messages.map((message) => h(Message, { key: message.id, channel, message }))), h( "form",