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"