From f1ad6f915ef076c6efafb27901e2b9b047d2a168 Mon Sep 17 00:00:00 2001 From: Alois Date: Wed, 18 Mar 2026 23:27:35 +0100 Subject: [PATCH] Updated some webtransport stuff --- apps/web/src/components/sidebar.tsx | 22 +--- packages/shared/src/data.ts | 2 +- packages/ttp/backend | 1 + packages/ttp/src/core.test.ts | 169 ++++++++++++++++++++++++++++ packages/ttp/src/core.ts | 86 ++++++++------ 5 files changed, 226 insertions(+), 54 deletions(-) create mode 160000 packages/ttp/backend diff --git a/apps/web/src/components/sidebar.tsx b/apps/web/src/components/sidebar.tsx index b48cffb..ff6da7f 100644 --- a/apps/web/src/components/sidebar.tsx +++ b/apps/web/src/components/sidebar.tsx @@ -1,27 +1,9 @@ import { useStorage } from "@tensamin/storage/context"; import Wrapper from "@tensamin/user/wrapper"; -import { type User } from "@tensamin/user/context"; import * as React from "react"; import { Basic, Loading } from "./modals/basic"; import List from "@/features/conversation/list/body"; -/** - * Renders sidebar user summary content for the current user. - * @param user Loaded user data. - * @returns Sidebar user card JSX. - */ -function renderSidebarUser(user: User): React.ReactNode { - return ; -} - -/** - * Renders sidebar user summary content skeleton while loading user data. - * @returns Sidebar user card skeleton JSX. - */ -function renderSidebarUserLoading(): React.ReactNode { - return ; -} - /** * Renders the conversation sidebar with account summary and conversation list. * @returns Sidebar JSX. @@ -37,9 +19,9 @@ export default function Sidebar() { return (
} userId={userId} - component={renderSidebarUser} + component={(user) => } />
diff --git a/packages/shared/src/data.ts b/packages/shared/src/data.ts index eb578a5..c39e795 100644 --- a/packages/shared/src/data.ts +++ b/packages/shared/src/data.ts @@ -100,7 +100,7 @@ export const socket = { communities: z.array(community), }), }, - live_message: { + message_live: { request: z.object({}), response: z.object({ sender_id: z.number(), diff --git a/packages/ttp/backend b/packages/ttp/backend new file mode 160000 index 0000000..bafbf13 --- /dev/null +++ b/packages/ttp/backend @@ -0,0 +1 @@ +Subproject commit bafbf13f43a9f7092341ec102621e234438be163 diff --git a/packages/ttp/src/core.test.ts b/packages/ttp/src/core.test.ts index 70c1285..3c1ad3c 100644 --- a/packages/ttp/src/core.test.ts +++ b/packages/ttp/src/core.test.ts @@ -1,5 +1,7 @@ import { describe, expect, test } from "bun:test"; +import { socket } from "@tensamin/shared/data"; import { + createTransportClient, decodeCommunicationMessage, encodeCommunicationMessage, type TypedMessage, @@ -35,6 +37,101 @@ function createRoundTripMessage(): TypedMessage> { }; } +function installFakeWebTransport(frameBytes: Uint8Array) { + const globalScope = globalThis as typeof globalThis & { + WebTransport?: unknown; + }; + const originalWebTransport = globalScope.WebTransport; + + class FakeWebTransport { + readonly ready = Promise.resolve(); + + readonly closed: Promise; + + private resolveClosed!: () => void; + + readonly incomingUnidirectionalStreams: ReadableStream>; + + constructor(_url: string) { + this.closed = new Promise((resolve) => { + this.resolveClosed = resolve; + }); + + const incomingStream = new ReadableStream>({ + start: (controller) => { + controller.enqueue( + new ReadableStream({ + start(innerController) { + const splitIndex = 3; + innerController.enqueue(frameBytes.subarray(0, splitIndex)); + setTimeout(() => { + innerController.enqueue(frameBytes.subarray(splitIndex)); + }, 10); + }, + }), + ); + }, + }); + + this.incomingUnidirectionalStreams = incomingStream; + } + + createUnidirectionalStream() { + return new WritableStream({ + write() { + return undefined; + }, + close() { + return undefined; + }, + }); + } + + close() { + this.resolveClosed(); + } + } + + globalScope.WebTransport = FakeWebTransport as never; + + return () => { + globalScope.WebTransport = originalWebTransport; + }; +} + +function installFakeLocalStorage() { + const globalScope = globalThis as typeof globalThis & { + localStorage?: Storage; + }; + const originalLocalStorage = globalScope.localStorage; + const store = new Map(); + + globalScope.localStorage = { + getItem(key: string) { + return store.has(key) ? store.get(key)! : null; + }, + setItem(key: string, value: string) { + store.set(key, value); + }, + removeItem(key: string) { + store.delete(key); + }, + clear() { + store.clear(); + }, + key(index: number) { + return [...store.keys()][index] ?? null; + }, + get length() { + return store.size; + }, + } as Storage; + + return () => { + globalScope.localStorage = originalLocalStorage; + }; +} + describe("TTP communication codec", () => { test("encodes and decodes a mixed payload message", () => { const input = createRoundTripMessage(); @@ -84,4 +181,76 @@ describe("TTP communication codec", () => { }), ).toThrow("Unknown data type"); }); + + test("skips unknown container keys without failing decode", () => { + const encoded = encodeCommunicationMessage({ + id: 9, + type: "error", + data: { + accepted: true, + }, + }); + + const corrupted = encoded.slice(); + corrupted[corrupted.length - 1] = 215; + + const decoded = decodeCommunicationMessage(corrupted); + + expect(decoded.id).toBe(9); + expect(decoded.type).toBe("error"); + expect(decoded.data).toEqual({}); + }); + + test("aligns live message schema with backend protocol name", () => { + expect(socket.message_live !== undefined).toBe(true); + expect((socket as Record).live_message).toEqual(undefined); + }); + + test("resolves an identification response before the stream closes", async () => { + const frame = encodeCommunicationMessage({ + id: 7, + type: "identification", + data: { + challenge: "Zm9v", + public_key: "Zm9v", + }, + }); + + const restoreWebTransport = installFakeWebTransport(frame); + const restoreLocalStorage = installFakeLocalStorage(); + + try { + const client = createTransportClient(socket, { url: "https://example.test" }); + + await client.connect("https://example.test"); + + const response = client.send( + "identification", + { user_id: 1 }, + { id: 7 }, + ); + + const timeout = new Promise((_, reject) => { + setTimeout(() => { + reject(new Error("identification response timed out")); + }, 250); + }); + + const result = await Promise.race([response, timeout]); + + expect(result).toEqual({ + id: 7, + type: "identification", + data: { + challenge: "Zm9v", + public_key: "Zm9v", + }, + }); + + await client.close("test-complete"); + } finally { + restoreLocalStorage(); + restoreWebTransport(); + } + }); }); diff --git a/packages/ttp/src/core.ts b/packages/ttp/src/core.ts index c3507a8..6fbf787 100644 --- a/packages/ttp/src/core.ts +++ b/packages/ttp/src/core.ts @@ -1205,32 +1205,6 @@ async function writeCloseFrame(transport: WebTransportLike) { * @returns Decoded typed message or null for close sentinel frames. */ async function readFrame(stream: ReadableStream) { - const payload = await readAll(stream); - if (payload.byteLength < 4) { - throw new Error("Received truncated transport frame"); - } - - const declaredLength = readU32(payload, 0); - if (declaredLength === CLOSE_FRAME_LEN) { - return null; - } - - const actualLength = payload.byteLength - 4; - if (actualLength !== declaredLength) { - throw new Error( - `Transport frame length mismatch: expected ${declaredLength}, received ${actualLength}`, - ); - } - - return decodeCommunicationMessage(payload.subarray(4)); -} - -/** - * Reads all chunks from a stream into a contiguous byte array. - * @param stream Stream providing Uint8Array chunks. - * @returns Concatenated stream bytes. - */ -async function readAll(stream: ReadableStream) { const reader = stream.getReader(); const chunks: Uint8Array[] = []; let totalLength = 0; @@ -1244,11 +1218,58 @@ async function readAll(stream: ReadableStream) { chunks.push(value); totalLength += value.byteLength; + + if (totalLength >= 4) { + const payload = concatChunks(chunks, totalLength); + const declaredLength = readU32(payload, 0); + + if (declaredLength === CLOSE_FRAME_LEN) { + return null; + } + + const expectedLength = 4 + declaredLength; + if (totalLength >= expectedLength) { + if (totalLength !== expectedLength) { + throw new Error( + `Transport frame length mismatch: expected ${declaredLength}, received ${totalLength - 4}`, + ); + } + + return decodeCommunicationMessage(payload); + } + } } } finally { reader.releaseLock(); } + if (totalLength < 4) { + throw new Error("Received truncated transport frame"); + } + + const payload = concatChunks(chunks, totalLength); + const declaredLength = readU32(payload, 0); + if (declaredLength === CLOSE_FRAME_LEN) { + return null; + } + + const actualLength = payload.byteLength - 4; + if (actualLength !== declaredLength) { + throw new Error( + `Transport frame length mismatch: expected ${declaredLength}, received ${actualLength}`, + ); + } + + return decodeCommunicationMessage(payload); +} + +/** + * Concatenates stream chunks into a single byte array. + * @param chunks Stream chunks to concatenate. + * @param totalLength Total byte length of the concatenated chunks. + * @returns Concatenated stream bytes. + */ +function concatChunks(chunks: Uint8Array[], totalLength: number) { const buffer = new Uint8Array(totalLength); let offset = 0; @@ -1729,11 +1750,15 @@ function decodeContainerPayload(reader: ByteReader) { const marker = reader.readU8(); const payloadLength = isBoolKindMarker(marker) ? 0 : reader.readU16(); const keyIndex = reader.readU8(); - const key = getDataTypeNameByIndex(keyIndex); const payload = isBoolKindMarker(marker) ? new Uint8Array(0) : reader.readBytes(payloadLength); + const key = getDataTypeNameByIndex(keyIndex); + if (!key) { + continue; + } + const expectedKind = getExpectedKind(key); if (!isMarkerCompatibleWithKey(marker, expectedKind, key)) { throw new Error( @@ -1756,12 +1781,7 @@ function decodeContainerPayload(reader: ByteReader) { * @returns Canonical protocol data key name. */ function getDataTypeNameByIndex(index: number) { - const key = DATA_TYPES[index]; - if (!key) { - throw new Error(`Unknown data type index ${index}`); - } - - return key; + return DATA_TYPES[index]; } /**