sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

feat(chord): example 59-chat keeps its messages in a table

Browse source
This commit is contained in:
Mario Zechner 2026-11-02 22:47:48 +00:00
parent d5c4b80ebf
commit 3c7a9ca0e7
6 changed files with 150 additions and 52 deletions
Download patch file Download diff file Expand all files Collapse all files

19
packages/chord/test/examples/59-chat/app/v1.ts
View file

@ -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<typeof Attachment>;
/** 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<typeof Message>;
/**
* 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()),
},
},

34
packages/chord/test/examples/59-chat/main.ts
View file

@ -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;
}

2
packages/chord/test/examples/59-chat/plugins/chat/server.ts
View file

@ -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",

83
packages/chord/test/examples/59-chat/plugins/chat/worker.ts
View file

@ -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<Message[]>([]);
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 };
}

46
packages/chord/test/examples/59-chat/tui/main.ts
View file

@ -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 <path>` 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 <path>` 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<string, { room: Held<typeof Room>; 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<typeof Room>;
live: () => readonly v1.Message[] | undefined;
older: readonly v1.Message[];
};
const rooms = new Map<string, Viewed>();
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<void> => {
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 = "";

18
packages/chord/test/examples/59-chat/web/main.ts
View file

@ -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<readonly v1.Message[]>([]);
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",