Eureka
7adc7315a3
A send queued at the room gate passed the brake check at entry only, so a brake placed while it waited still let the PUT through. The check now also runs inside the gate, just before the PUT, and answers a typed Brake refusal. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
556 lines
30 KiB
JavaScript
556 lines
30 KiB
JavaScript
// The Matrix ear and mouth of every Persona. One process: one account (token) and
|
|
// one /sync loop per Persona, one CultNet endpoint for all of them.
|
|
//
|
|
// Ear: /sync the configured rooms (and the account's direct rooms); every message
|
|
// from another sender becomes a delivery kept in memory, carrying its sender id,
|
|
// its m.mentions user ids and whether the room is a DM. The /sync position
|
|
// (next_batch) is persisted only when an ack covers it, so a restart before the
|
|
// ack re-delivers instead of losing. Reading wakes nothing: the heartbeat polls
|
|
// with a non-consuming chat_pending_request and decides who runs.
|
|
// Mouth: a send request becomes one PUT with txnId = request_id (a retry is idempotent), and a room
|
|
// is joined, only while that Persona's brake is released. A v2 send carries the basis of her turn (the
|
|
// newest room event she saw): inside the room gate, one step per room across every Persona, the mouth
|
|
// reads the room's head from Matrix and answers RoomMoved, naming the messages that moved it, instead
|
|
// of the PUT. A v1 send, and a v2 send with no basis, are not checked; both still pass through the gate.
|
|
//
|
|
// Context: Matrix is the record. A delivery's `context` is read from the homeserver when the delivery
|
|
// is built (GET /messages, or the thread's relations for a threaded message), as this Persona's
|
|
// account, and holds m.room.message events from everyone, her own included, oldest first, ending
|
|
// just before the delivered event. Nothing about a room's past is kept in this process, so a restart
|
|
// loses none of it. A failed fetch gives empty context and a log line; the delivery still goes out.
|
|
//
|
|
// Each Matrix access token is read here, from its own systemd credential, and
|
|
// nowhere else. It is never put in a document, an environment passed to a child,
|
|
// or a log, and it is only ever presented in its own Persona's Authorization header.
|
|
//
|
|
// Everything outside this process is a port (fetch, state, brake, clock), so the
|
|
// core runs against a fake homeserver with no CultLib and no daemon.
|
|
import { createHash } from "node:crypto";
|
|
import { mkdirSync, readFileSync } from "node:fs";
|
|
import { createRequire } from "node:module";
|
|
import { createSocket } from "node:dgram";
|
|
import { resolve, join } from "node:path";
|
|
import { pathToFileURL } from "node:url";
|
|
import * as docs from "./persona-chat-documents.mjs";
|
|
|
|
const API = "/_matrix/client/v3";
|
|
const BRAKE_TYPE = "eureka.control.brake.v1";
|
|
const CONTEXT = 20; // /sync timeline limit per round
|
|
const CONTEXT_LINES = 20; // the most context lines the delivery tuple carries (persona-chat-documents maxContext)
|
|
const CONTEXT_BYTES = 16 * 1024; // text and sender bytes over all lines of one delivery: under CONTEXT_LINES x limits.contextBytes
|
|
const FETCH_LIMIT = 100; // events asked of the homeserver per context fetch
|
|
const V1 = "/_matrix/client/v1"; // /relations lives only under v1
|
|
const MAX_PENDING = 500;
|
|
const SLUG = /^[A-Za-z0-9_-]{1,64}$/;
|
|
const SEND_TIMEOUT_MS = 10000; // the head read and the PUT of one gated send
|
|
const HEAD_PAGE = 50; // the room head a gated send reads; also the most ids a RoomMoved receipt names
|
|
|
|
// A message line: a text event with a sender, not an edit.
|
|
const isLine = e => e?.type === "m.room.message" && typeof e.content?.body === "string" && typeof e.sender === "string" && e.sender !== "" && e.content?.["m.relates_to"]?.rel_type !== "m.replace";
|
|
// Whether `event` moves the room under a Persona whose account is `selfId`: someone else's line that is
|
|
// not an automated notice. Threaded replies count; the ear delivers them to her.
|
|
export const movesRoom = (event, selfId) => isLine(event) && event.sender !== selfId && event.content?.msgtype !== "m.notice";
|
|
|
|
// The one gate every mouth shares: run(roomId, fn) runs fn after every earlier fn on that room, and a
|
|
// failure of one does not stop the next. It holds ordering only, nothing across a restart.
|
|
export function createRoomGate() {
|
|
const tails = new Map();
|
|
return {
|
|
run(roomId, fn) {
|
|
const result = (tails.get(roomId) ?? Promise.resolve()).then(fn);
|
|
const tail = result.catch(() => {});
|
|
tails.set(roomId, tail);
|
|
tail.then(() => { if (tails.get(roomId) === tail) tails.delete(roomId); });
|
|
return result;
|
|
},
|
|
};
|
|
}
|
|
|
|
export function createMouth({
|
|
persona, homeserver, serverName, rooms, token, fetch: httpFetch = globalThis.fetch, state, brake,
|
|
now = Date.now, log = () => {}, syncTimeoutMs = 30000, maxPending = MAX_PENDING, gate = createRoomGate(), sendTimeoutMs = SEND_TIMEOUT_MS,
|
|
}) {
|
|
if (!persona || !homeserver || !token || !rooms?.length) throw new Error("persona, homeserver, rooms and token are required");
|
|
const say = message => log(redact(String(message), token));
|
|
const roomIdByAlias = new Map();
|
|
const aliasByRoomId = new Map();
|
|
const joinedAliases = new Set();
|
|
const freshRooms = new Set(); // joined by this process since the last sync: their history is context, not mail
|
|
const directRooms = new Set(); // room ids the account marks as DMs (m.direct)
|
|
const displayNames = new Map();
|
|
let selfId = null;
|
|
let cursor = null; // the /sync position we are reading from
|
|
let seqCounter = 0;
|
|
let pending = []; // deliveries not yet acked
|
|
let batches = []; // { token, maxSeq }: positions awaiting an ack
|
|
let committed = null; // the position that is persisted
|
|
let started = false;
|
|
const directContent = {}; // the account's m.direct content: user id -> room ids
|
|
const invites = new Map(); // room id -> { inviter, direct }: invitations not yet answered
|
|
const processed = new Set(); // event ids read since the position last advanced
|
|
|
|
async function matrix(method, path, body, prefix = API, signal) {
|
|
let response;
|
|
try {
|
|
response = await httpFetch(`${homeserver}${prefix}${path}`, {
|
|
method,
|
|
headers: { authorization: `Bearer ${token}`, "content-type": "application/json" },
|
|
body: body === undefined ? undefined : JSON.stringify(body),
|
|
signal,
|
|
});
|
|
} catch (error) {
|
|
throw new Error(`homeserver unreachable: ${redact(error?.message ?? String(error), token)}`);
|
|
}
|
|
if (!response.ok) throw new Error(`homeserver answered ${response.status} for ${method} ${path.split("?")[0]}`);
|
|
return response.json();
|
|
}
|
|
|
|
const released = async () => { try { return (await brake()) === true; } catch { return false; } };
|
|
const noteDirect = content => {
|
|
for (const [user, ids] of Object.entries(content ?? {})) {
|
|
if (!Array.isArray(ids)) continue;
|
|
directContent[user] = ids.filter(id => typeof id === "string");
|
|
directContent[user].forEach(id => directRooms.add(id));
|
|
}
|
|
};
|
|
|
|
function noteInvites(invited) {
|
|
for (const [roomId, room] of Object.entries(invited ?? {})) {
|
|
const event = (room.invite_state?.events ?? []).find(e => e.type === "m.room.member" && e.state_key === selfId && typeof e.sender === "string");
|
|
if (event) invites.set(roomId, { inviter: event.sender, direct: event.content?.is_direct === true });
|
|
}
|
|
}
|
|
|
|
// A DM invitation from an account on our own homeserver is accepted, and only while the brake is
|
|
// released; one received while braked waits for the release. An invitation from any other server is
|
|
// declined. A same-server invitation to a room that is not a DM is not ours to accept and is dropped.
|
|
async function answerInvites() {
|
|
let free = null;
|
|
for (const [roomId, invite] of [...invites]) {
|
|
try {
|
|
const room = encodeURIComponent(roomId);
|
|
if (!serverName || invite.inviter.slice(invite.inviter.indexOf(":") + 1) !== serverName) {
|
|
await matrix("POST", `/rooms/${room}/leave`, {});
|
|
invites.delete(roomId);
|
|
} else if (!invite.direct) {
|
|
invites.delete(roomId);
|
|
} else {
|
|
free ??= await released();
|
|
if (!free) continue;
|
|
await matrix("POST", `/join/${room}`, {});
|
|
invites.delete(roomId);
|
|
directRooms.add(roomId);
|
|
directContent[invite.inviter] = [...new Set([...(directContent[invite.inviter] ?? []), roomId])];
|
|
await matrix("PUT", `/user/${encodeURIComponent(selfId)}/account_data/m.direct`, directContent).catch(error => say(`m.direct not saved: ${error.message}`));
|
|
}
|
|
} catch (error) {
|
|
say(`invite to ${roomId} not answered: ${error.message}`);
|
|
}
|
|
}
|
|
}
|
|
const unjoined = () => joinedAliases.size < rooms.length;
|
|
|
|
// Join the configured rooms the account is not in, only while the brake is released.
|
|
// One room failing is logged and retried on the next round; it never stops the others.
|
|
async function ensureRooms() {
|
|
const member = new Set((await matrix("GET", "/joined_rooms")).joined_rooms ?? []);
|
|
let free = null;
|
|
for (const alias of rooms.filter(a => !joinedAliases.has(a))) {
|
|
try {
|
|
let roomId = roomIdByAlias.get(alias);
|
|
if (!roomId) ({ room_id: roomId } = await matrix("GET", `/directory/room/${encodeURIComponent(alias)}`));
|
|
roomIdByAlias.set(alias, roomId);
|
|
if (!member.has(roomId)) {
|
|
free ??= await released();
|
|
if (!free) continue;
|
|
await matrix("POST", `/join/${encodeURIComponent(roomId)}`, {});
|
|
freshRooms.add(roomId);
|
|
}
|
|
aliasByRoomId.set(roomId, alias);
|
|
joinedAliases.add(alias);
|
|
} catch (error) {
|
|
say(`room ${alias} not joined: ${error.message}`);
|
|
}
|
|
}
|
|
}
|
|
|
|
async function start() {
|
|
if (!started) {
|
|
const saved = await state.load();
|
|
cursor = committed = saved?.nextBatch ?? null;
|
|
seqCounter = saved?.seq ?? 0;
|
|
selfId = (await matrix("GET", "/account/whoami")).user_id;
|
|
try { noteDirect(await matrix("GET", `/user/${encodeURIComponent(selfId)}/account_data/m.direct`)); } catch { /* no m.direct: no DMs yet */ }
|
|
started = true;
|
|
}
|
|
if (unjoined()) await ensureRooms();
|
|
}
|
|
|
|
const threadRootOf = e => { const r = e?.content?.["m.relates_to"]; return r?.rel_type === "m.thread" && typeof r.event_id === "string" && r.event_id !== "" ? r.event_id : null; };
|
|
const lineOf = e => ({ sender: e.sender, text: docs.clip(e.content.body, docs.limits.contextBytes), atUnixMs: Number.isSafeInteger(e.origin_server_ts) && e.origin_server_ts >= 0 ? e.origin_server_ts : 0 });
|
|
|
|
// The room's recent messages outside threads, oldest first.
|
|
async function roomEvents(roomId) {
|
|
const filter = encodeURIComponent(JSON.stringify({ types: ["m.room.message"] }));
|
|
const reply = await matrix("GET", `/rooms/${encodeURIComponent(roomId)}/messages?dir=b&limit=${FETCH_LIMIT}&filter=${filter}`);
|
|
return (reply.chunk ?? []).filter(e => isLine(e) && threadRootOf(e) === null).reverse();
|
|
}
|
|
// One thread's messages, root first.
|
|
async function threadEvents(roomId, rootId) {
|
|
const room = encodeURIComponent(roomId);
|
|
const root = await matrix("GET", `/rooms/${room}/event/${encodeURIComponent(rootId)}`);
|
|
const replies = await matrix("GET", `/rooms/${room}/relations/${encodeURIComponent(rootId)}/m.thread?dir=b&limit=${FETCH_LIMIT}`, undefined, V1);
|
|
return [root, ...(replies.chunk ?? []).reverse()].filter(isLine);
|
|
}
|
|
|
|
// The context of one delivered event: the messages before it, the newest kept to the byte budget, each line clipped.
|
|
// `round` caches the fetches of one /sync round, so a room or thread is read once however many deliveries it made.
|
|
async function contextFor(roomId, event, round) {
|
|
const root = threadRootOf(event);
|
|
const key = root ? `${roomId}|${root}` : roomId;
|
|
if (!round.has(key)) {
|
|
round.set(key, (root ? threadEvents(roomId, root) : roomEvents(roomId)).catch(error => {
|
|
say(`context for ${roomId}${root ? ` thread ${root}` : ""} not fetched: ${error.message}`);
|
|
return [];
|
|
}));
|
|
}
|
|
const events = await round.get(key);
|
|
let cut = events.findIndex(e => e.event_id === event.event_id);
|
|
if (cut < 0) cut = events.filter(e => e.origin_server_ts < event.origin_server_ts).length;
|
|
const lines = [];
|
|
let bytes = 0;
|
|
for (const e of events.slice(0, cut).reverse()) {
|
|
const l = lineOf(e);
|
|
const size = Buffer.byteLength(l.text) + Buffer.byteLength(l.sender);
|
|
if (lines.length >= CONTEXT_LINES || bytes + size > CONTEXT_BYTES) break;
|
|
bytes += size;
|
|
lines.unshift(l);
|
|
}
|
|
return lines;
|
|
}
|
|
|
|
// One /sync round. Returns the deliveries it made.
|
|
async function syncOnce() {
|
|
if (!started || unjoined()) await start();
|
|
if (pending.length >= maxPending) return []; // unanswered mail is not outrun: the position stays where it is
|
|
const filter = encodeURIComponent(JSON.stringify({ room: { timeline: { limit: CONTEXT } } }));
|
|
const since = cursor === null ? "" : `&since=${encodeURIComponent(cursor)}`;
|
|
const reply = await matrix("GET", `/sync?timeout=${cursor === null ? 0 : syncTimeoutMs}&filter=${filter}${since}`);
|
|
const baseline = cursor === null; // first ever sync: history is context, not mail
|
|
for (const event of reply.account_data?.events ?? []) if (event.type === "m.direct") noteDirect(event.content);
|
|
noteInvites(reply.rooms?.invite);
|
|
await answerInvites();
|
|
const made = [];
|
|
const round = new Map();
|
|
let truncated = false; // the pending cap was reached inside this round
|
|
rounds: for (const [roomId, room] of Object.entries(reply.rooms?.join ?? {})) {
|
|
const alias = aliasByRoomId.get(roomId) ?? (directRooms.has(roomId) ? roomId : undefined);
|
|
if (!alias) continue;
|
|
for (const event of [...(room.state?.events ?? []), ...(room.timeline?.events ?? [])]) {
|
|
const shown = event.content?.displayname;
|
|
if (event.type === "m.room.member" && typeof event.state_key === "string" && typeof shown === "string" && shown !== "") displayNames.set(event.state_key, docs.clip(shown, docs.limits.displayBytes));
|
|
}
|
|
for (const event of room.timeline?.events ?? []) {
|
|
if (event.type !== "m.room.message" || typeof event.content?.body !== "string") continue;
|
|
if (typeof event.sender !== "string" || event.sender === "" || typeof event.event_id !== "string" || event.event_id === "") continue;
|
|
if (processed.has(event.event_id)) continue; // read already in a round the cap kept from advancing
|
|
const ts = event.origin_server_ts;
|
|
const entry = { sender: event.sender, text: docs.clip(event.content.body, docs.limits.textBytes), atUnixMs: Number.isSafeInteger(ts) && ts >= 0 ? ts : now() };
|
|
const deliverable = !baseline && !freshRooms.has(roomId) && event.sender !== selfId;
|
|
if (deliverable && pending.length + made.length >= maxPending) { truncated = true; break rounds; }
|
|
processed.add(event.event_id);
|
|
if (deliverable) {
|
|
const context = await contextFor(roomId, event, round);
|
|
const mentioned = event.content["m.mentions"]?.user_ids;
|
|
made.push({
|
|
seq: ++seqCounter, roomAlias: alias, roomId, eventId: event.event_id, sender: event.sender,
|
|
senderDisplay: displayNames.get(event.sender) ?? localpart(event.sender), text: entry.text, atUnixMs: entry.atUnixMs,
|
|
context,
|
|
mentions: Array.isArray(mentioned) ? mentioned.filter(id => typeof id === "string" && id !== "" && id.length <= 255).slice(0, docs.limits.mentions) : [],
|
|
isDirect: directRooms.has(roomId),
|
|
});
|
|
}
|
|
}
|
|
}
|
|
pending.push(...made);
|
|
if (!truncated) { // a truncated round leaves the position where it was: the rest stays on the server
|
|
cursor = reply.next_batch;
|
|
batches.push({ token: reply.next_batch, maxSeq: seqCounter });
|
|
processed.clear();
|
|
freshRooms.clear();
|
|
}
|
|
await commit();
|
|
if (made.length > 0) await state.save({ nextBatch: committed, seq: seqCounter }); // sequence numbers never repeat across a restart
|
|
return made;
|
|
}
|
|
|
|
// The persisted position is the newest batch whose deliveries are all acked.
|
|
async function commit(throughSeq = pending.length ? pending[0].seq - 1 : seqCounter) {
|
|
let covered = null;
|
|
for (const b of batches) { if (b.maxSeq <= throughSeq) covered = b; else break; }
|
|
if (covered && covered.token !== committed) {
|
|
committed = covered.token;
|
|
await state.save({ nextBatch: committed, seq: seqCounter });
|
|
}
|
|
batches = batches.filter(b => b.maxSeq > throughSeq);
|
|
}
|
|
|
|
async function ack(throughSeq) {
|
|
pending = pending.filter(d => d.seq > throughSeq);
|
|
await commit(pending.length ? pending[0].seq - 1 : seqCounter);
|
|
}
|
|
|
|
function pendingAfter(afterSeq) {
|
|
return pending.filter(d => d.seq > afterSeq).map(d => ({ ...d, context: d.context.map(c => ({ ...c })), mentions: [...d.mentions] }));
|
|
}
|
|
|
|
// The ids of the messages that moved the room since `basis`, oldest first. Walks the head newest to oldest
|
|
// and stops at the event she saw; with none, at her turn's start time; if she saw an event the page does
|
|
// not reach, the whole page counts.
|
|
async function newerThan(roomId, basis) {
|
|
const filter = encodeURIComponent(JSON.stringify({ types: ["m.room.message"] }));
|
|
const reply = await matrix("GET", `/rooms/${encodeURIComponent(roomId)}/messages?dir=b&limit=${HEAD_PAGE}&filter=${filter}`, undefined, API, AbortSignal.timeout(sendTimeoutMs));
|
|
const moved = [];
|
|
for (const e of reply.chunk ?? []) {
|
|
if (basis.seenEventId !== null ? e.event_id === basis.seenEventId : e.origin_server_ts <= basis.sinceUnixMs) break;
|
|
if (movesRoom(e, selfId)) moved.push(e.event_id);
|
|
}
|
|
return moved.reverse();
|
|
}
|
|
|
|
async function send({ requestId, persona: asked, roomAlias, text, basis }) {
|
|
const refuse = (refusal, newer = []) => ({ requestId, eventId: null, refusal, newer });
|
|
if (asked !== persona) return refuse("PersonaMismatch");
|
|
if (!(await released())) return refuse("Brake");
|
|
if (typeof text !== "string" || text.trim() === "") return refuse("EmptyText");
|
|
try {
|
|
if (!started || unjoined()) await start();
|
|
const roomId = joinedAliases.has(roomAlias) ? roomIdByAlias.get(roomAlias) : directRooms.has(roomAlias) ? roomAlias : null;
|
|
if (!roomId) return refuse("UnknownRoom");
|
|
return await gate.run(roomId, async () => {
|
|
if (basis) {
|
|
const newer = await newerThan(roomId, basis);
|
|
if (newer.length > 0) return refuse("RoomMoved", newer);
|
|
}
|
|
if (!(await released())) return refuse("Brake"); // a brake placed during the gate wait or the head read stops this send
|
|
const sent = await matrix("PUT", `/rooms/${encodeURIComponent(roomId)}/send/m.room.message/${encodeURIComponent(requestId)}`, { msgtype: "m.text", body: text }, API, AbortSignal.timeout(sendTimeoutMs));
|
|
return { requestId, eventId: sent.event_id, refusal: null, newer: [] };
|
|
});
|
|
} catch (error) {
|
|
say(`send failed: ${error.message}`);
|
|
return refuse("SendFailed");
|
|
}
|
|
}
|
|
|
|
async function run(signal) {
|
|
const pause = ms => new Promise(r => { const t = setTimeout(r, ms); signal?.addEventListener("abort", () => { clearTimeout(t); r(); }, { once: true }); });
|
|
while (!signal?.aborted) {
|
|
try {
|
|
await syncOnce();
|
|
if (pending.length >= maxPending) await pause(1000);
|
|
} catch (error) {
|
|
say(`sync failed: ${error.message}`);
|
|
await pause(5000);
|
|
}
|
|
}
|
|
}
|
|
|
|
return { start, syncOnce, ack, pendingAfter, send, run, get pendingCount() { return pending.length; } };
|
|
}
|
|
|
|
// Replace every secret in a message. `secrets` is one string or a list.
|
|
export function redact(message, secrets) {
|
|
let out = String(message);
|
|
for (const secret of [secrets].flat()) if (secret) out = out.split(secret).join("[redacted]");
|
|
return out;
|
|
}
|
|
|
|
// Run `fn`; whatever it throws leaves with every secret redacted.
|
|
export async function guarded(secrets, fn) {
|
|
try { return await fn(); } catch (error) { throw new Error(redact(error?.message ?? String(error), secrets)); }
|
|
}
|
|
|
|
function localpart(userId) {
|
|
return userId.replace(/^@/, "").split(":")[0];
|
|
}
|
|
|
|
// Brake: released only when the instance's control store decodes to a released
|
|
// brake. Absent, undecodable or held reads are all "not released".
|
|
export function createBrakeReader({ controlPath, readStore, decode }) {
|
|
return async () => {
|
|
try {
|
|
const read = readStore(controlPath);
|
|
if (read.absent) return false;
|
|
const envelope = read.envelopes.find(e => e.type === BRAKE_TYPE && e.key === "brake");
|
|
if (!envelope) return false;
|
|
const value = decode(new Uint8Array(envelope.payload));
|
|
const released = Array.isArray(value) ? value[0] : value instanceof Map ? value.get(0) ?? value.get("released") : value?.[0] ?? value?.["0"] ?? value?.released;
|
|
return released === true;
|
|
} catch {
|
|
return false;
|
|
}
|
|
};
|
|
}
|
|
|
|
// The RUDP connection id is derived from the credential, so a peer without it cannot open the session.
|
|
export function connectionIdFor(connectionKey) {
|
|
return createHash("sha256").update(connectionKey).digest().readUInt32BE(0);
|
|
}
|
|
|
|
// The reply to a chat_pending_request: deliveries oldest first, each encoded on its own so one that
|
|
// will not encode is skipped and logged without silencing the rest, up to the reply byte budget.
|
|
// The caller acks what it has read and asks again.
|
|
export function pageOf(persona, deliveries, msgpack, log = () => {}) {
|
|
const page = [];
|
|
let bytes = 0;
|
|
for (const delivery of deliveries) {
|
|
let tuple, size;
|
|
try { tuple = docs.deliveryTuple(delivery); size = msgpack.encode(tuple).length; } catch (error) {
|
|
log(`delivery ${delivery?.seq} skipped: ${error?.message ?? error}`);
|
|
continue;
|
|
}
|
|
if (page.length > 0 && bytes + size > docs.limits.replyBytes) break;
|
|
page.push(tuple);
|
|
bytes += size;
|
|
}
|
|
return [persona, page];
|
|
}
|
|
|
|
// CultNet RUDP endpoint on loopback, one for every Persona in `mouths` (persona -> mouth).
|
|
// Every client (each Persona process and heartbeat dials with its own socket) gets its own
|
|
// session, and a request is answered on the session it arrived on. Requests arrive as
|
|
// document_put_raw messages. A request naming a Persona this
|
|
// process does not serve is answered with a typed UnknownPersona refusal.
|
|
export async function startChatEndpoint({ mouths, runtime, port, connectionKey, runtimeId, socketFactory = createSocket, host = "127.0.0.1", log = () => {} }) {
|
|
if (!/^127\.\d+\.\d+\.\d+$/.test(host)) throw new Error("the chat endpoint listens on loopback only");
|
|
const { msgpack, cultnet } = runtime;
|
|
const socket = socketFactory("udp4");
|
|
await new Promise((ok, fail) => { socket.once("error", fail); socket.once("listening", ok); socket.bind(port, host); });
|
|
const connectionId = connectionIdFor(connectionKey);
|
|
const server = new cultnet.CultNetRudpSocketTransportServer({ runtimeId, socket, connectionId, initialSequence: 100, resendDelayMs: 50, resendPollMs: 10, maxFragmentBytes: 1200,
|
|
// A client keeps its link between requests and does not redial, so only a client gone for an hour (a crash that sent no goodbye) is reaped.
|
|
peerTimeoutMs: 3_600_000 });
|
|
const respond = (peer, persona, kind, recordKey, tuple) => server.send(peer, "schema", msgpack.encode(cultnet.encodeCultNetMessageForWire(
|
|
docs.envelope({ kind, recordKey, payload: msgpack.encode(tuple), sourceRuntimeId: runtimeId, sourceAgentId: persona, sourceRole: "bifrost-persona-matrix-mouth" }), "cultnet.schema.v0")));
|
|
const refuseUnknown = (peer, persona) => respond(peer, persona, docs.kinds.refusal, persona, docs.refusalTuple({ persona, refusal: "UnknownPersona" }));
|
|
const handle = async ({ peer, channelId, payload }) => {
|
|
try {
|
|
if (channelId !== "schema") throw new Error("chat request arrived on a foreign channel");
|
|
const message = cultnet.parseCultNetMessage(msgpack.decode(payload), "cultnet.schema.v0");
|
|
const document = docs.checkEnvelope(message, [docs.kinds.pendingRequest, docs.kinds.ack, docs.kinds.sendRequest, docs.kinds.sendRequestV2]);
|
|
const tuple = msgpack.decode(document.payload);
|
|
if (document.schemaId === docs.kinds.pendingRequest) {
|
|
const request = docs.parsePendingRequest(tuple);
|
|
const mouth = mouths.get(request.persona);
|
|
if (!mouth) return refuseUnknown(peer, request.persona);
|
|
respond(peer, request.persona, docs.kinds.pending, request.persona, pageOf(request.persona, mouth.pendingAfter(request.afterSeq), msgpack, log));
|
|
} else if (document.schemaId === docs.kinds.ack) {
|
|
const request = docs.parseAck(tuple);
|
|
const mouth = mouths.get(request.persona);
|
|
if (!mouth) return refuseUnknown(peer, request.persona);
|
|
await mouth.ack(request.throughSeq);
|
|
} else {
|
|
const v2 = document.schemaId === docs.kinds.sendRequestV2;
|
|
const request = v2 ? docs.parseSendRequestV2(tuple) : docs.parseSendRequest(tuple);
|
|
const mouth = mouths.get(request.persona);
|
|
const receipt = mouth ? await mouth.send(request) : { requestId: request.requestId, eventId: null, refusal: "UnknownPersona", newer: [] };
|
|
if (v2) respond(peer, request.persona, docs.kinds.sendReceiptV2, receipt.requestId, docs.sendReceiptV2Tuple(receipt));
|
|
else respond(peer, request.persona, docs.kinds.sendReceipt, receipt.requestId, docs.sendReceiptTuple(receipt));
|
|
}
|
|
} catch (error) {
|
|
log(`chat endpoint rejected a frame: ${error?.message ?? error}`);
|
|
}
|
|
};
|
|
// A peer's requests run in order; peers do not wait on one another.
|
|
const busy = new WeakMap();
|
|
const frame = value => { busy.set(value.peer, (busy.get(value.peer) ?? Promise.resolve()).then(() => handle(value))); };
|
|
server.on("frame", frame);
|
|
// A socket error is logged; the endpoint keeps serving.
|
|
server.on("error", error => log(`chat endpoint socket error: ${error?.message ?? error}`));
|
|
return { address: socket.address(), close() { server.off("frame", frame); server.close(); } };
|
|
}
|
|
|
|
export function cultcacheState({ runtime, path }) {
|
|
const def = runtime.defineDocumentType({
|
|
type: "gamecult.persona.chat_cursor.v1", schemaName: "gamecult.persona.chat_cursor.v1", schemaId: "gamecult.persona.chat_cursor.v1",
|
|
schemaVersion: "gamecult.persona.chat_cursor.v1", contentHash: "gamecult.persona.chat_cursor.v1", global: false, name: "key",
|
|
schema: { parse: v => ({ key: String(v.key), nextBatch: v.nextBatch ?? null, seq: Number(v.seq) }) },
|
|
});
|
|
const cache = runtime.CultCache.builder().withDocumentType(def).withGenericStore(new runtime.SingleFileMessagePackBackingStore(path)).build();
|
|
const unwrap = v => (Array.isArray(v) ? v[0] : v?.value ?? v);
|
|
return {
|
|
async load() { await cache.pullAllBackingStores(); return unwrap(cache.get(def, "cursor")) ?? null; },
|
|
async save({ nextBatch, seq }) { await cache.put(def, "cursor", { key: "cursor", nextBatch, seq }); },
|
|
};
|
|
}
|
|
|
|
export function loadRuntime(cultRoot) {
|
|
const at = name => createRequire(resolve(cultRoot, "packages", name, "package.json"));
|
|
const cc = at("cultcache-ts")("@gamecult/cultcache-ts"), cn = at("cultnet-ts")("cultnet-ts"), mp = at("cultnet-ts")("@msgpack/msgpack");
|
|
return { CultCache: cc.CultCache, SingleFileMessagePackBackingStore: cc.SingleFileMessagePackBackingStore, defineDocumentType: cc.defineDocumentType, readSingleFileStore: cc.readSingleFileStore, cultnet: cn, msgpack: mp };
|
|
}
|
|
|
|
function required(env, name) {
|
|
if (!env[name]) throw new Error(`${name} is required`);
|
|
return env[name];
|
|
}
|
|
function credential(env, name) {
|
|
return readFileSync(join(required(env, "CREDENTIALS_DIRECTORY"), name), "utf8").trim();
|
|
}
|
|
|
|
const ROOM_ALIAS = /^#[^\s:,]+:[^\s,]+$/;
|
|
|
|
// The env variable holding one Persona's project rooms: MATRIX_ROOMS_<SLUG>, the slug
|
|
// upper-cased with "." and "-" turned into "_".
|
|
export function projectRoomsVariable(persona) {
|
|
return `MATRIX_ROOMS_${persona.toUpperCase().replace(/[.-]/g, "_")}`;
|
|
}
|
|
|
|
// One Persona's rooms: the shared MATRIX_ROOMS plus her own project rooms, deduplicated.
|
|
// Another Persona's project variable is never read. A malformed project alias is refused.
|
|
export function roomsFor(env, persona) {
|
|
const list = text => text.split(",").map(s => s.trim()).filter(Boolean);
|
|
const name = projectRoomsVariable(persona);
|
|
const project = list(env[name] ?? "");
|
|
const bad = project.find(alias => !ROOM_ALIAS.test(alias));
|
|
if (bad) throw new Error(`${name} has a malformed room alias: ${JSON.stringify(bad)} (expected #name:server)`);
|
|
return [...new Set([...list(required(env, "MATRIX_ROOMS")), ...project])];
|
|
}
|
|
|
|
// PERSONAS is the comma-separated list of Persona slugs this process speaks for.
|
|
// Each slug names its credential (matrix-token-<slug>), its cursor directory under
|
|
// STATE_DIRECTORY and its brake at CONTROL_ROOT/<slug>/control.cc.
|
|
export async function main(env = process.env) {
|
|
const personas = required(env, "PERSONAS").split(",").map(s => s.trim()).filter(Boolean);
|
|
if (personas.length === 0 || new Set(personas).size !== personas.length || !personas.every(p => SLUG.test(p))) throw new Error("PERSONAS must be distinct slugs");
|
|
if (new Set(personas.map(projectRoomsVariable)).size !== personas.length) throw new Error("PERSONAS must map to distinct MATRIX_ROOMS_<SLUG> variables");
|
|
const tokens = new Map(personas.map(p => [p, credential(env, `matrix-token-${p}`)]));
|
|
const chatKey = credential(env, "chat-key");
|
|
const secrets = [...tokens.values(), chatKey];
|
|
return guarded(secrets, async () => {
|
|
const runtime = loadRuntime(resolve(env.CULTLIB_ROOT ?? resolve(import.meta.dirname, "..", "..", "CultLib")));
|
|
const log = line => console.error(redact(line, secrets));
|
|
const homeserver = required(env, "MATRIX_HOMESERVER").replace(/\/$/, "");
|
|
const serverName = required(env, "MATRIX_SERVER_NAME");
|
|
const roomsByPersona = new Map(personas.map(persona => [persona, roomsFor(env, persona)]));
|
|
const stateRoot = required(env, "STATE_DIRECTORY");
|
|
const controlRoot = env.CONTROL_ROOT ?? "/etc/gamecult/minds";
|
|
const gate = createRoomGate(); // one for every Persona: two Minds in one room take turns
|
|
const mouths = new Map(personas.map(persona => {
|
|
mkdirSync(join(stateRoot, persona), { recursive: true });
|
|
return [persona, createMouth({
|
|
persona, homeserver, serverName, rooms: roomsByPersona.get(persona), token: tokens.get(persona), gate, log: line => log(`${persona}: ${line}`),
|
|
state: cultcacheState({ runtime, path: join(stateRoot, persona, "cursor.cc") }),
|
|
brake: createBrakeReader({ controlPath: join(controlRoot, persona, "control.cc"), readStore: runtime.readSingleFileStore, decode: runtime.msgpack.decode }),
|
|
})];
|
|
}));
|
|
const listen = required(env, "CHAT_RUDP_LISTEN"); // host:port, loopback
|
|
const colon = listen.lastIndexOf(":");
|
|
await startChatEndpoint({ mouths, runtime, host: listen.slice(0, colon), port: Number(listen.slice(colon + 1)), connectionKey: chatKey, runtimeId: "bifrost-persona-mouth", log });
|
|
await Promise.all([...mouths.values()].map(mouth => mouth.run()));
|
|
});
|
|
}
|
|
|
|
if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {
|
|
main().catch(error => { console.error(error?.message ?? String(error)); process.exitCode = 1; });
|
|
}
|