}
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];
}
/**