Updated Wrapper component, added queue for all ttp requests, updated some desktop app configuration,

packages/ttp/src/context.tsx

@@ -189,14 +189,14 @@ export default function Provider(props: {
    * Sends typed protocol messages through the active transport client.
    * @param type Protocol message type.
    * @param data Optional request payload.
-   * @param options Optional request id and response mode.
-   * @returns A promise for either void (no response) or typed message payload.
+    * @param options Optional request id.
+    * @returns A promise for the typed message payload.
    */
   const send = useCallback<BoundSendFn<Schemas>>(
     ((
       type: string,
       data?: Record<string, unknown>,
-      options?: { id?: number; noResponse?: boolean },
+      options?: { id?: number },
     ) => {
       const client = clientRef.current;
 
@@ -204,16 +204,8 @@ export default function Provider(props: {
         return Promise.reject(new Error("ttp is not connected"));
       }
 
-      if (options?.noResponse) {
-        return client.send(type as keyof Schemas & string, data as never, {
-          ...options,
-          noResponse: true,
-        });
-      }
-
       return client.send(type as keyof Schemas & string, data as never, {
         ...options,
-        noResponse: false,
       });
     }) as BoundSendFn<Schemas>,
     [],

packages/ttp/src/core.test.ts

@@ -27,9 +27,7 @@ type MockTransportInstance = {
   ready: Promise<void>;
   closed: Promise<void>;
   createUnidirectionalStream: () => Promise<MockStream>;
-  incomingUnidirectionalStreams: {
-    getReader: () => MockReader;
-  };
+  incomingUnidirectionalStreams: ReadableStream<ReadableStream<Uint8Array>>;
   close: () => void;
 };
 
@@ -69,6 +67,10 @@ function createMockWebTransport() {
 
   let readyResolve!: () => void;
   let closedResolve!: () => void;
+  let incomingController:
+    | ReadableStreamDefaultController<ReadableStream<Uint8Array>>
+    | null = null;
+  const outgoingMessages: TypedMessage[] = [];
 
   const ready = new Promise<void>((resolve) => {
     readyResolve = resolve;
@@ -87,17 +89,19 @@ function createMockWebTransport() {
     getWriter: () => writer,
   };
 
+  const incomingUnidirectionalStreams = new ReadableStream<
+    ReadableStream<Uint8Array>
+  >({
+    start(controller) {
+      incomingController = controller;
+    },
+  });
+
   const transport: MockTransportInstance = {
     ready,
     closed,
     createUnidirectionalStream: async () => stream,
-    incomingUnidirectionalStreams: {
-      getReader: () => ({
-        read: async () => ({ done: true }),
-        releaseLock: () => {},
-        cancel: async () => {},
-      }),
-    },
+    incomingUnidirectionalStreams,
     close: () => {},
   };
 
@@ -109,10 +113,32 @@ function createMockWebTransport() {
     close = transport.close;
   }
 
+  const pushIncomingMessage = (message: TypedMessage) => {
+    if (!incomingController) {
+      throw new Error("Incoming stream controller is not ready");
+    }
+
+    const frame = wrapForDecode(encodeCommunicationMessage(message));
+    const incomingStream = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(frame);
+        controller.close();
+      },
+    });
+
+    incomingController.enqueue(incomingStream);
+  };
+
+  writer.write = async (chunk: Uint8Array) => {
+    outgoingMessages.push(decodeCommunicationMessage(chunk));
+  };
+
   return {
     readyResolve,
     closedResolve,
     MockWebTransport,
+    outgoingMessages,
+    pushIncomingMessage,
   };
 }
 
@@ -378,6 +404,61 @@ describe("Core Protocol", () => {
 
       await expectRejection(sendPromise);
     });
+
+    it("queues sends until the active request receives a response", async () => {
+      const {
+        readyResolve,
+        pushIncomingMessage,
+        outgoingMessages,
+        MockWebTransport,
+      } = createMockWebTransport();
+      const globalWithWebTransport = globalThis as unknown as {
+        WebTransport?: new (url: string) => MockTransportInstance;
+      };
+      globalWithWebTransport.WebTransport = MockWebTransport;
+
+      const schemas: SchemaMap = {
+        ping: {
+          request: z.object({}),
+          response: z.object({}),
+        },
+      };
+
+      const client = createTransportClient(schemas);
+      const connectPromise = client.connect("http://localhost:8000");
+      readyResolve();
+      await connectPromise;
+
+      const firstSend = client.send("ping", {});
+      const secondSend = client.send("ping", {});
+
+      for (let attempt = 0; attempt < 10 && outgoingMessages.length === 0; attempt += 1) {
+        await Promise.resolve();
+      }
+
+      expect(outgoingMessages.length).toBe(1);
+
+      pushIncomingMessage({
+        id: outgoingMessages[0].id,
+        type: "ping",
+        data: {},
+      });
+
+      await firstSend;
+      for (let attempt = 0; attempt < 10 && outgoingMessages.length === 1; attempt += 1) {
+        await Promise.resolve();
+      }
+
+      expect(outgoingMessages.length).toBe(2);
+
+      pushIncomingMessage({
+        id: outgoingMessages[1].id,
+        type: "ping",
+        data: {},
+      });
+
+      await secondSend;
+    });
   });
 
   describe("Push subscriptions", () => {

packages/ttp/src/core.ts

@@ -69,20 +69,13 @@ export type SchemaMap = Record<
 
 type SendOptions = {
   id?: number;
-  noResponse?: boolean;
 };
 
 export type BoundSendFn<T extends SchemaMap> = {
   <K extends keyof T & string>(
     type: K,
     data: z.input<T[K]["request"]>,
-    options: { id?: number; noResponse: true },
-  ): Promise<void>;
-
-  <K extends keyof T & string>(
-    type: K,
-    data: z.input<T[K]["request"]>,
-    options?: { id?: number; noResponse?: false },
+    options?: { id?: number },
   ): Promise<TypedMessage<z.output<T[K]["response"]>>>;
 };
 
@@ -125,6 +118,7 @@ export function createTransportClient<T extends SchemaMap>(
   let currentConnection: ActiveConnection | null = null;
   let currentReadyState: number = READY_STATE.CLOSED;
   let nextRequestId = 1;
+  let sendQueueTail: Promise<void> = Promise.resolve();
   let configuredUrl = options.url;
 
   /**
@@ -192,6 +186,20 @@ export function createTransportClient<T extends SchemaMap>(
     notifyClosed(connection, error);
   };
 
+  /**
+   * Serializes outbound send work so only one request is active at a time.
+   * @param task Request task to run in queue order.
+   * @returns Promise for the task result.
+   */
+  const enqueueSend = <T>(task: () => Promise<T>) => {
+    const queuedTask = sendQueueTail.then(task, task);
+    sendQueueTail = queuedTask.then(
+      () => undefined,
+      () => undefined,
+    );
+    return queuedTask;
+  };
+
   /**
    * Handles decoded incoming messages and resolves request promises or push listeners.
    * @param message Decoded incoming message.
@@ -515,8 +523,8 @@ export function createTransportClient<T extends SchemaMap>(
    * Sends a typed protocol request over the current connection.
    * @param type Protocol message type.
    * @param input Optional request payload.
-   * @param options Optional id and response behavior.
-   * @returns Promise for response message or void when no response is expected.
+    * @param options Optional request id.
+    * @returns Promise for the typed response message.
    */
   const send: BoundSendFn<T> = ((
     type: string,
@@ -527,8 +535,6 @@ export function createTransportClient<T extends SchemaMap>(
       return Promise.reject(new Error("Transport is not connected"));
     }
 
-    const connection = currentConnection;
-
     try {
       const schema = schemas[type];
       let payload: Record<string, unknown>;
@@ -555,69 +561,66 @@ export function createTransportClient<T extends SchemaMap>(
         payload = coercePayload(input ?? {});
       }
 
-      if (!options?.id && !options?.noResponse) {
-        const array = new Uint32Array(1);
-        crypto.getRandomValues(array);
-        options = options ?? {};
-        options.id = array[0];
-      }
+      const requestOptions = options ? { ...options } : {};
+      const enqueuedConnection = currentConnection;
 
-      if (type !== "ping") {
-        log(2, "ttp", "gray", "Sent:", type, payload, { id: options.id });
-      }
+      return enqueueSend(() => {
+        if (
+          !enqueuedConnection ||
+          currentConnection !== enqueuedConnection ||
+          currentReadyState !== READY_STATE.OPEN ||
+          enqueuedConnection.closeNotified
+        ) {
+          return Promise.reject(new Error("Transport is not connected"));
+        }
 
-      const expectsResponse = !options?.noResponse;
-      const requestId = resolveRequestId(
-        options.id,
-        expectsResponse,
-        pending,
-        () => {
-          const current = nextRequestId;
-          nextRequestId = current >= MAX_REQUEST_ID ? 1 : current + 1;
-          return current;
-        },
-      );
-
-      const messageBytes = encodeCommunicationMessage({
-        id: requestId,
-        type,
-        data: payload,
-      });
-
-      if (!expectsResponse) {
-        return writeMessageOnPersistentStream(connection, messageBytes).catch(
-          (error) => {
-            handleConnectionFailure(connection, error);
-            throw error;
+        const requestId = resolveRequestId(
+          requestOptions.id,
+          true,
+          pending,
+          () => {
+            const current = nextRequestId;
+            nextRequestId = current >= MAX_REQUEST_ID ? 1 : current + 1;
+            return current;
           },
         );
-      }
 
-      return new Promise<TypedMessage>((resolve, reject) => {
-        const timeoutId = setTimeout(() => {
-          pending.delete(requestId);
-          reject(
-            new Error(
-              `Request "${type}" timed out after ${RESPONSE_TIMEOUT}ms`,
-            ),
-          );
-        }, RESPONSE_TIMEOUT);
+        if (type !== "ping") {
+          log(2, "ttp", "gray", "Sent:", type, payload, { id: requestId });
+        }
 
-        pending.set(requestId, {
-          requestType: type,
-          resolve,
-          reject,
-          timeoutId,
+        const messageBytes = encodeCommunicationMessage({
+          id: requestId,
+          type,
+          data: payload,
         });
 
-        void writeMessageOnPersistentStream(connection, messageBytes).catch(
-          (error) => {
-            handleConnectionFailure(connection, error);
-            clearTimeout(timeoutId);
+        return new Promise<TypedMessage>((resolve, reject) => {
+          const timeoutId = setTimeout(() => {
             pending.delete(requestId);
-            reject(error);
-          },
-        );
+            reject(
+              new Error(
+                `Request "${type}" timed out after ${RESPONSE_TIMEOUT}ms`,
+              ),
+            );
+          }, RESPONSE_TIMEOUT);
+
+          pending.set(requestId, {
+            requestType: type,
+            resolve,
+            reject,
+            timeoutId,
+          });
+
+          void writeMessageOnPersistentStream(enqueuedConnection, messageBytes).catch(
+            (error) => {
+              handleConnectionFailure(enqueuedConnection, error);
+              clearTimeout(timeoutId);
+              pending.delete(requestId);
+              reject(error);
+            },
+          );
+        });
       });
     } catch (error) {
       return Promise.reject(error);

packages/user/package.json

@@ -16,6 +16,7 @@
   "dependencies": {
     "@tensamin/ttp": "workspace:*",
     "@tensamin/shared": "workspace:*",
+    "@tensamin/storage": "workspace:*",
     "react": "^19.2.0",
     "react-dom": "^19.2.0",
     "zod": "^4.3.6"

packages/user/src/wrapper.tsx

@@ -2,14 +2,16 @@ import { useEffect, useState } from "react";
 import { useUser, type User } from "./context";
 
 import { failedUser } from "@tensamin/shared/data";
+import { useStorage } from "@tensamin/storage/context";
 
 // Wrapper function to pass user data to some component
 export default function Wrapper(props: {
-  userId?: number;
+  userId?: number | "own";
   loading: React.ReactNode;
   component: (user: User) => React.ReactNode;
 }) {
   const { get } = useUser();
+  const { load } = useStorage();
   const [user, setUser] = useState<User | null>(null);
 
   useEffect(() => {
@@ -20,6 +22,17 @@ export default function Wrapper(props: {
 
     let active = true;
 
+    if (props.userId === "own") {
+      load("user_id")
+        .then((id) => {
+          get(id)
+            .then((user) => (active ? setUser(user) : null))
+            .catch(() => setUser(failedUser));
+        })
+        .catch(() => setUser(failedUser));
+      return;
+    }
+
     get(props.userId)
       .then((value) => {
         if (active) {
@@ -35,7 +48,7 @@ export default function Wrapper(props: {
     return () => {
       active = false;
     };
-  }, [get, props.userId]);
+  }, [load, get, props.userId]);
 
   return <>{user ? props.component(user) : props.loading}</>;
 }