From 13545d2d0a5f5ffb0d5575e2a1b3c8724d56a472 Mon Sep 17 00:00:00 2001 From: m1ngsama Date: Wed, 14 Oct 2026 19:44:37 +0800 Subject: [PATCH] feat(gateway): carry private messages both ways, list app users in TNT and search history App users can message each other and SSH users privately. Messages for SSH users go to TNT 1.6 as private.post, and TNT's private.created reaches the app user it names. TNT learns the set of app users from presence.remote. A search frame finds up to 50 recent messages by sender or text. --- gateway/src/gateway.ts | 63 ++++++++++++++++++++++++-- gateway/src/history.ts | 12 +++++ gateway/src/main.ts | 13 +++++- gateway/src/module.ts | 18 +++++++- gateway/src/presence.ts | 15 +++++++ gateway/test/gateway.test.ts | 86 ++++++++++++++++++++++++++++++++++++ gateway/test/module.test.ts | 14 +++++- gateway/tnt-module.json | 5 ++- 8 files changed, 217 insertions(+), 9 deletions(-) diff --git a/gateway/src/gateway.ts b/gateway/src/gateway.ts index 50e14da..f6ffe9d 100644 --- a/gateway/src/gateway.ts +++ b/gateway/src/gateway.ts @@ -1,7 +1,7 @@ import type { Server, ServerWebSocket } from "bun"; import type { Verify } from "./auth"; import { type Ledger, type Message, ROOM } from "./history"; -import type { RoomEvent } from "./module"; +import type { PrivateEvent, RoomEvent } from "./module"; import { Presence } from "./presence"; import { RateLimiter } from "./ratelimit"; import { normalizeText, validateText } from "./text"; @@ -13,6 +13,8 @@ const HISTORY_PAGE = 50; const HISTORY_PAGE_MAX = 100; const NONCE_TTL_MS = 10 * 60_000; const NONCE_MAX_LENGTH = 64; +const SEARCH_LIMIT = 50; +const QUERY_MAX_LENGTH = 64; const OPEN = 1; type Session = { @@ -29,6 +31,10 @@ export type GatewayOptions = { verify: Verify; post: (sender: string, text: string) => void; signedIn?: (handle: string) => void; + /** Hands a private message to TNT for an SSH user; false when TNT cannot take one. */ + privatePost?: (from: string, to: string, text: string) => boolean; + /** Tells TNT which app users are online, whenever that set changes. */ + remotePresence?: (handles: string[]) => void; now?: () => Date; authTimeoutMs?: number; }; @@ -52,13 +58,27 @@ export function createGateway(options: GatewayOptions) { const typingAt = new Map(); const sent = new Map(); let online = "[]"; + let remote = "[]"; const broadcast = (frame: object) => { const data = JSON.stringify(frame); for (const ws of clients) ws.send(data); }; const deliver = (message: Message, nonce?: string) => broadcast(nonce ? { type: "message", ...message, nonce } : { type: "message", ...message }); + const syncRemote = () => { + const list = presence.web(); + const key = JSON.stringify(list); + if (key === remote) return; + remote = key; + options.remotePresence?.(list); + }; + const toHandle = (handle: string, frame: object) => { + const data = JSON.stringify(frame); + for (const ws of clients) if (ws.data.handle === handle) ws.send(data); + }; + const stamp = () => now().toISOString().replace(/\.\d{3}Z$/, "Z"); const syncPresence = () => { + syncRemote(); const list = presence.online(); const key = JSON.stringify(list); if (key === online) return; @@ -104,10 +124,35 @@ export function createGateway(options: GatewayOptions) { for (const client of clients) if (client.data.handle !== handle) client.send(data); } + function sendPrivate(ws: Socket, frame: Record) { + const text = validateText(frame.text); + if (typeof frame.to !== "string" || frame.to === "") return fail(ws, "invalid_frame"); + if (text === null) return fail(ws, "invalid_text"); + const from = ws.data.handle; + const to = frame.to; + const toApp = presence.isWeb(to); + if (!toApp && !presence.isSsh(to)) return fail(ws, "not_online"); + if (!limiter.take(from)) return fail(ws, "rate_limited"); + if (!toApp && !options.privatePost?.(from, to, text)) return fail(ws, "not_online"); + const message = { type: "private", id: crypto.randomUUID(), from, to, text, timestamp: stamp(), verified: true }; + if (toApp && to !== from) toHandle(to, message); + const nonce = typeof frame.nonce === "string" && frame.nonce.length <= NONCE_MAX_LENGTH ? frame.nonce : undefined; + toHandle(from, nonce ? { ...message, nonce } : message); + } + + function search(ws: Socket, frame: Record) { + const query = typeof frame.query === "string" ? frame.query.trim() : ""; + if (query === "" || query.length > QUERY_MAX_LENGTH) return fail(ws, "invalid_frame"); + if (!historyLimiter.take(ws.data.handle)) return fail(ws, "rate_limited"); + ws.send(JSON.stringify({ type: "search", query, messages: options.ledger.search(query, SEARCH_LIMIT) })); + } + function receive(ws: Socket, frame: Frame) { if (frame?.type === "ping") return ws.send('{"type":"pong"}'); if (frame?.type === "history") return sendHistory(ws, frame); if (frame?.type === "typing") return forwardTyping(ws); + if (frame?.type === "private") return sendPrivate(ws, frame); + if (frame?.type === "search") return search(ws, frame); if (frame?.type !== "send") return fail(ws, "invalid_frame"); const text = validateText(frame.text); if (text === null) return fail(ws, "invalid_text"); @@ -118,8 +163,7 @@ export function createGateway(options: GatewayOptions) { if (earlier) return ws.send(JSON.stringify({ type: "message", ...earlier.message, nonce })); if (!limiter.take(sender)) return fail(ws, "rate_limited"); options.post(sender, text); - const timestamp = now().toISOString().replace(/\.\d{3}Z$/, "Z"); - const message = options.ledger.add({ timestamp, sender, text, verified: true }); + const message = options.ledger.add({ timestamp: stamp(), sender, text, verified: true }); if (key !== undefined) sent.set(key, { message, at: Date.now() }); deliver(message, nonce); } @@ -153,6 +197,19 @@ export function createGateway(options: GatewayOptions) { for (const [key, entry] of sent) if ((at ?? Date.now()) - entry.at > NONCE_TTL_MS) sent.delete(key); for (const [handle, time] of typingAt) if ((at ?? Date.now()) - time > TYPING_INTERVAL_MS) typingAt.delete(handle); }, + handlePrivate(event: PrivateEvent) { + const to = [...clients].find((ws) => ws.data.handle.toLowerCase() === event.to.toLowerCase())?.data.handle; + if (!to) return; + toHandle(to, { + type: "private", + id: crypto.randomUUID(), + from: event.from, + to, + text: normalizeText(event.plain_text), + timestamp: event.timestamp, + verified: event.verified, + }); + }, handleEvent(event: RoomEvent) { if (event.type === "message.created") { const { timestamp, sender, plain_text, verified } = event.message; diff --git a/gateway/src/history.ts b/gateway/src/history.ts index 927804e..d597519 100644 --- a/gateway/src/history.ts +++ b/gateway/src/history.ts @@ -164,6 +164,18 @@ export class Ledger { return this.#messages.slice(-RECENT); } + /** Newest first, matching sender or text without regard to case, like :search. */ + search(query: string, limit: number): Message[] { + const wanted = query.toLowerCase(); + const found: Message[] = []; + for (let i = this.#messages.length - 1; i >= 0 && found.length < limit; i--) { + const message = this.#messages[i]!; + if (message.kind === "system") continue; + if (message.sender.toLowerCase().includes(wanted) || message.text.toLowerCase().includes(wanted)) found.push(message); + } + return found; + } + before(id: string, limit: number): { messages: Message[]; more: boolean } | null { const index = this.#messages.findIndex((m) => m.id === id); if (index < 0) return null; diff --git a/gateway/src/main.ts b/gateway/src/main.ts index 65eccd6..fc63256 100644 --- a/gateway/src/main.ts +++ b/gateway/src/main.ts @@ -2,7 +2,7 @@ import { createVerifier, remoteKeys } from "./auth"; import { createGateway } from "./gateway"; import { Ledger, markReserved, readHistory, readReserved } from "./history"; import { signIn } from "./identity"; -import { encodePost, encodeRecord, parseTntLine, supportsIdentity } from "./module"; +import { encodePost, encodeRecord, parseTntLine, supportsIdentity, supportsPrivate } from "./module"; const HISTORY_LIMIT = 5000; const env = process.env; @@ -20,12 +20,20 @@ for (const entry of await readHistory(stateDir, HISTORY_LIMIT)) ledger.add(markR const verify = createVerifier(issuer, remoteKeys(env.TNT_GATEWAY_JWKS_URL || `${issuer}/oidc/jwks`)); let identity = false; +let direct = false; const signingIn = new Set(); const gateway = createGateway({ ledger, verify, post: (sender, text) => write(encodePost(sender, text)), + privatePost: (from, to, plain_text) => { + if (direct) write(encodeRecord({ type: "private.post", from, to, plain_text })); + return direct; + }, + remotePresence: (nicknames) => { + if (direct) write(encodeRecord({ type: "presence.remote", nicknames })); + }, signedIn: (handle) => { if (!identity || reserved.has(handle)) return; reserved.add(handle); @@ -49,6 +57,9 @@ for await (const line of console) { if (!event) continue; if (event.type === "handshake") { identity = supportsIdentity(event.version); + direct = supportsPrivate(event.version); + } else if (event.type === "private.created") { + gateway.handlePrivate(event); } else if (event.type === "identity.requested") { const { session } = event; if (!identity || signingIn.has(session)) continue; diff --git a/gateway/src/module.ts b/gateway/src/module.ts index 520072e..fc35ddd 100644 --- a/gateway/src/module.ts +++ b/gateway/src/module.ts @@ -2,10 +2,12 @@ export type TntEvent = | { type: "handshake"; version: string } | { type: "message.created"; message: { timestamp: string; sender: string; plain_text: string; verified: boolean } } | { type: "identity.requested"; session: string } + | { type: "private.created"; from: string; to: string; plain_text: string; timestamp: string; verified: boolean } | { type: "presence.snapshot"; nicknames: string[] } | { type: "presence.joined" | "presence.left"; nickname: string }; -export type RoomEvent = Exclude; +export type RoomEvent = Exclude; +export type PrivateEvent = Extract; export const HANDSHAKE_OK = `${JSON.stringify({ type: "handshake.ok", @@ -20,7 +22,7 @@ export function encodePost(sender: string, text: string): string { return `${JSON.stringify({ type: "message.post", sender, plain_text: text })}\n`; } -export function encodeRecord(record: Record): string { +export function encodeRecord(record: Record): string { return `${JSON.stringify(record)}\n`; } @@ -30,6 +32,12 @@ export function supportsIdentity(version: string): boolean { return major > 1 || (major === 1 && minor >= 5); } +/** Private messages and app presence arrived in TNT 1.6.0. */ +export function supportsPrivate(version: string): boolean { + const [major = 0, minor = 0] = version.split(".").map(Number); + return major > 1 || (major === 1 && minor >= 6); +} + export function parseTntLine(line: string): { reply: string | null; event: TntEvent | null } { let record: Record; try { @@ -54,6 +62,12 @@ export function parseTntLine(line: string): { reply: string | null; event: TntEv event: valid ? { type: "message.created", message: { timestamp, sender, plain_text, verified } } : null, }; } + case "private.created": { + const { from, to, plain_text, timestamp } = record; + return isString(from) && isString(to) && isString(plain_text) && isString(timestamp) + ? { reply: null, event: { type: "private.created", from, to, plain_text, timestamp, verified: record.verified === true } } + : NOTHING; + } case "identity.requested": return isString(record.session) ? { reply: null, event: { type: "identity.requested", session: record.session } } : NOTHING; case "presence.snapshot": diff --git a/gateway/src/presence.ts b/gateway/src/presence.ts index 9330947..9ea7060 100644 --- a/gateway/src/presence.ts +++ b/gateway/src/presence.ts @@ -30,6 +30,21 @@ export class Presence { if (connections.size === 0) this.#web.delete(nickname); } + /** App users, for TNT to list beside its SSH sessions. */ + web(): string[] { + return [...this.#web.keys()].sort(); + } + + isWeb(nickname: string): boolean { + return this.#web.has(nickname); + } + + isSsh(nickname: string): boolean { + const wanted = nickname.toLowerCase(); + for (const name of this.#ssh) if (name.toLowerCase() === wanted) return true; + return false; + } + online(): string[] { return [...new Set([...this.#ssh, ...this.#web.keys()])].sort(); } diff --git a/gateway/test/gateway.test.ts b/gateway/test/gateway.test.ts index a052387..b62cdba 100644 --- a/gateway/test/gateway.test.ts +++ b/gateway/test/gateway.test.ts @@ -211,3 +211,89 @@ describe("history and typing", () => { await expect(alice.next((f) => f.type === "typing", 300)).rejects.toThrow(); }); }); + +describe("private messages", () => { + let delivered: [string, string, string][]; + let remote: string[][]; + let direct: ReturnType; + let directServer: ReturnType; + let directUrl: string; + + beforeEach(() => { + delivered = []; + remote = []; + direct = createGateway({ + ledger: new Ledger(100), + verify: async (token) => (["alice", "bob"].includes(token) ? token : null), + post: () => {}, + privatePost: (from, to, text) => { + delivered.push([from, to, text]); + return true; + }, + remotePresence: (handles) => remote.push(handles), + now: () => new Date("2026-09-22T10:00:00.900Z"), + }); + directServer = Bun.serve({ hostname: "127.0.0.1", port: 0, fetch: direct.fetch, websocket: direct.websocket }); + directUrl = `ws://127.0.0.1:${directServer.port}/ws`; + }); + + afterEach(() => { + void directServer.stop(true); + }); + + test("tells TNT which app users are online", async () => { + const alice = connect(directUrl, "alice"); + await alice.next((f) => f.type === "ready"); + const bob = connect(directUrl, "bob"); + await bob.next((f) => f.type === "ready"); + bob.ws.close(); + await bob.closed; + await alice.next((f) => f.type === "presence" && f.online.length === 1); + expect(remote).toEqual([["alice"], ["alice", "bob"], ["alice"]]); + }); + + test("delivers between app users and echoes to the sender", async () => { + const alice = connect(directUrl, "alice"); + const bob = connect(directUrl, "bob"); + await alice.next((f) => f.type === "ready"); + await bob.next((f) => f.type === "ready"); + alice.send({ type: "private", to: "bob", text: "psst", nonce: "n1" }); + expect(await bob.next((f) => f.type === "private")).toMatchObject({ from: "alice", to: "bob", text: "psst", verified: true }); + expect(await alice.next((f) => f.type === "private")).toMatchObject({ from: "alice", to: "bob", nonce: "n1" }); + expect(delivered).toEqual([]); + }); + + test("hands messages for SSH users to TNT and refuses names nobody uses", async () => { + direct.handleEvent({ type: "presence.snapshot", nicknames: ["Carol"] }); + const alice = connect(directUrl, "alice"); + await alice.next((f) => f.type === "ready"); + alice.send({ type: "private", to: "carol", text: "hi carol" }); + expect(await alice.next((f) => f.type === "private")).toMatchObject({ to: "carol", text: "hi carol" }); + expect(delivered).toEqual([["alice", "carol", "hi carol"]]); + alice.send({ type: "private", to: "nobody", text: "hello?" }); + expect(await alice.next((f) => f.type === "error")).toEqual({ type: "error", code: "not_online" }); + }); + + test("passes TNT's private messages to the app user they name", async () => { + const bob = connect(directUrl, "bob"); + await bob.next((f) => f.type === "ready"); + direct.handlePrivate({ type: "private.created", from: "carol", to: "Bob", plain_text: "back at you", timestamp: "2026-09-22T10:00:00Z", verified: false }); + expect(await bob.next((f) => f.type === "private")).toMatchObject({ from: "carol", to: "bob", text: "back at you", verified: false }); + }); +}); + +describe("search", () => { + test("finds senders and text, newest first, without notices", async () => { + gateway.handleEvent({ type: "message.created", message: { timestamp: "2026-09-22T10:00:01Z", sender: "system", plain_text: "Release joined", verified: false } }); + gateway.handleEvent({ type: "message.created", message: { timestamp: "2026-09-22T10:00:02Z", sender: "dave", plain_text: "the release is out", verified: false } }); + gateway.handleEvent({ type: "message.created", message: { timestamp: "2026-09-22T10:00:03Z", sender: "release_bot", plain_text: "ok", verified: false } }); + const alice = connect(url, "alice"); + await alice.next((f) => f.type === "ready"); + alice.send({ type: "search", query: " RELEASE " }); + const result = await alice.next((f) => f.type === "search"); + expect(result.query).toBe("RELEASE"); + expect(result.messages.map((m: { sender: string }) => m.sender)).toEqual(["release_bot", "dave"]); + alice.send({ type: "search", query: "" }); + expect(await alice.next((f) => f.type === "error")).toEqual({ type: "error", code: "invalid_frame" }); + }); +}); diff --git a/gateway/test/module.test.ts b/gateway/test/module.test.ts index e431afd..613ff14 100644 --- a/gateway/test/module.test.ts +++ b/gateway/test/module.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "bun:test"; -import { EVENT_OK, HANDSHAKE_OK, encodePost, parseTntLine, supportsIdentity } from "../src/module"; +import { EVENT_OK, HANDSHAKE_OK, encodePost, parseTntLine, supportsIdentity, supportsPrivate } from "../src/module"; describe("parseTntLine", () => { test("answers the handshake", () => { @@ -37,6 +37,18 @@ describe("parseTntLine", () => { expect(parseTntLine('{"type":"identity.requested"}')).toEqual({ reply: null, event: null }); }); + test("parses private messages from TNT and knows which servers take them", () => { + expect( + parseTntLine('{"type":"private.created","from":"carol","to":"bob","plain_text":"hi","timestamp":"2026-09-22T10:00:00Z","verified":true}'), + ).toEqual({ + reply: null, + event: { type: "private.created", from: "carol", to: "bob", plain_text: "hi", timestamp: "2026-09-22T10:00:00Z", verified: true }, + }); + expect(parseTntLine('{"type":"private.created","from":"carol"}')).toEqual({ reply: null, event: null }); + expect(supportsPrivate("1.5.0")).toBe(false); + expect(supportsPrivate("1.6.0")).toBe(true); + }); + test("parses presence records without answering them", () => { expect(parseTntLine('{"type":"presence.snapshot","nicknames":["a",1,"b"]}')).toEqual({ reply: null, diff --git a/gateway/tnt-module.json b/gateway/tnt-module.json index 2761e0d..c2ab5d4 100644 --- a/gateway/tnt-module.json +++ b/gateway/tnt-module.json @@ -2,7 +2,7 @@ "protocol": "tnt.module.v1", "name": "tnt-gateway", "version": "0.1.0", - "tnt_min_version": "1.5.0", + "tnt_min_version": "1.6.0", "description": "WebSocket gateway for the TNT public room.", "entrypoint": "./tnt-gateway", "permissions": [ @@ -10,7 +10,8 @@ "message:create", "message:post", "presence:read", - "identity:write" + "identity:write", + "message:private" ], "events": [ "message.created"