Updated some webtransport stuff

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(),

packages/ttp/backend

@@ -0,0 +1 @@
+Subproject commit bafbf13f43a9f7092341ec102621e234438be163

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<Record<string, unknown>> {
   };
 }
 
+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<void>;
+
+    private resolveClosed!: () => void;
+
+    readonly incomingUnidirectionalStreams: ReadableStream<ReadableStream<Uint8Array>>;
+
+    constructor(_url: string) {
+      this.closed = new Promise<void>((resolve) => {
+        this.resolveClosed = resolve;
+      });
+
+      const incomingStream = new ReadableStream<ReadableStream<Uint8Array>>({
+        start: (controller) => {
+          controller.enqueue(
+            new ReadableStream<Uint8Array>({
+              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<Uint8Array>({
+        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<string, string>();
+
+  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<string, unknown>).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<never>((_, 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();
+    }
+  });
 });

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<Uint8Array>) {
-  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<Uint8Array>) {
   const reader = stream.getReader();
   const chunks: Uint8Array[] = [];
   let totalLength = 0;
@@ -1244,11 +1218,58 @@ async function readAll(stream: ReadableStream<Uint8Array>) {
 
       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];
 }
 
 /**