Refactor streams to ensure they only have a single subscription

frontend/app/src/components/home/activity/ActivityFeed.svelte

@@ -37,7 +37,7 @@
     }
 
     function loadActivity() {
-        client.messageActivityFeed().subscribe((resp, final) => {
+        client.subscribeToMessageActivityFeed((resp, final) => {
             activityEvents.set(resp.events);
             if ($activityEvents.length > 0 && final) {
                 client.markActivityFeedRead($activityEvents[0].timestamp);

frontend/openchat-client/src/openchat.ts

@@ -380,7 +380,6 @@ import type {
     PaymentGateApproval,
     PaymentGateApprovals,
     MessageActivityFeedResponse,
-    Stream,
 } from "openchat-shared";
 import {
     AuthProvider,
@@ -1357,19 +1356,23 @@ export class OpenChat extends OpenChatAgentWorker {
         });
     }
 
-    messageActivityFeed(): Stream<MessageActivityFeedResponse> {
-        return this.sendStreamRequest({
+    subscribeToMessageActivityFeed(
+        subscribeFn: (value: MessageActivityFeedResponse, final: boolean) => void,
+    ) {
+        this.sendStreamRequest({
             kind: "messageActivityFeed",
             since: this._liveState.globalState.messageActivitySummary.readUpToTimestamp,
-        }).subscribe((response) => {
-            const userIds = new Set<string>();
-            for (const event of response.events) {
-                if (event.userId !== undefined) {
-                    userIds.add(event.userId);
+        }).subscribe({
+            onResult: (response, final) => {
+                const userIds = new Set<string>();
+                for (const event of response.events) {
+                    if (event.userId !== undefined) {
+                        userIds.add(event.userId);
+                    }
                 }
-            }
-            this.getMissingUsers(userIds);
-            return response;
+                this.getMissingUsers(userIds);
+                subscribeFn(response, final);
+            },
         });
     }
 
@@ -3553,8 +3556,8 @@ export class OpenChat extends OpenChatAgentWorker {
                 },
                 undefined,
                 isCryptoMessage ? 2 * DEFAULT_WORKER_TIMEOUT : undefined,
-            )
-                .subscribe((response) => {
+            ).subscribe({
+                onResult: (response) => {
                     if (response === "accepted") {
                         unconfirmed.markAccepted(messageContext, eventWrapper.event.messageId);
                         return;
@@ -3613,8 +3616,8 @@ export class OpenChat extends OpenChatAgentWorker {
                     }
 
                     resolve(resp);
-                })
-                .catch(() => {
+                },
+                onError: () => {
                     this.onSendMessageFailure(
                         chatId,
                         eventWrapper.event.messageId,
@@ -3625,7 +3628,8 @@ export class OpenChat extends OpenChatAgentWorker {
                     );
 
                     return resolve(CommonResponses.failure());
-                });
+                },
+            });
         });
     }
 
@@ -4587,8 +4591,8 @@ export class OpenChat extends OpenChatAgentWorker {
     getCurrentUser(): Promise<CurrentUserResponse> {
         return new Promise((resolve, reject) => {
             let resolved = false;
-            this.sendStreamRequest({ kind: "getCurrentUser" })
-                .subscribe((user) => {
+            this.sendStreamRequest({ kind: "getCurrentUser" }).subscribe({
+                onResult: (user) => {
                     if (user.kind === "created_user") {
                         userCreatedStore.set(true);
                         currentUser.set(user);
@@ -4600,8 +4604,9 @@ export class OpenChat extends OpenChatAgentWorker {
                         resolve(user);
                         resolved = true;
                     }
-                })
-                .catch(reject);
+                },
+                onError: reject,
+            });
         });
     }
 
@@ -5746,22 +5751,23 @@ export class OpenChat extends OpenChatAgentWorker {
             this.sendStreamRequest({
                 kind: "getUpdates",
                 initialLoad,
-            })
-                .subscribe(async (resp) => {
+            }).subscribe({
+                onResult: async (resp) => {
                     await this.handleChatsResponse(
                         updateRegistryTask,
                         !this._liveState.chatsInitialised,
                         resp as UpdatesResult,
                     );
                     chatsLoading.set(!this._liveState.chatsInitialised);
-                })
-                .catch((err) => {
+                },
+                onError: (err) => {
                     console.warn("getUpdates threw an error: ", err);
                     resolve();
-                })
-                .finally(() => {
+                },
+                onEnd: () => {
                     resolve();
-                });
+                },
+            });
         });
     }
 
@@ -6275,8 +6281,8 @@ export class OpenChat extends OpenChatAgentWorker {
         return new Promise((resolve) => {
             this.sendStreamRequest({
                 kind: "updateRegistry",
-            })
-                .subscribe(([registry, updated]) => {
+            }).subscribe({
+                onResult: ([registry, updated]) => {
                     if (updated || Object.keys(get(cryptoLookup)).length === 0) {
                         this.currentAirdropChannel = registry.currentAirdropChannel;
                         const cryptoRecord = toRecord(registry.tokenDetails, (t) => t.ledger);
@@ -6311,11 +6317,12 @@ export class OpenChat extends OpenChatAgentWorker {
                         resolved = true;
                         resolve();
                     }
-                })
-                .catch((err) => {
+                },
+                onError: (err) => {
                     console.warn(`Failed to update the registry: ${err}`);
                     resolve();
-                });
+                },
+            });
         });
     }
 

frontend/openchat-shared/src/domain/stream.ts

@@ -2,13 +2,20 @@ type OnStreamResult<T> = (result: T, final: boolean) => void;
 type OnStreamError = (reason?: unknown) => void;
 type OnStreamEnd = () => void;
 
+type Subscription<T> = {
+    onResult?: OnStreamResult<T>;
+    onError?: OnStreamError;
+    onEnd?: OnStreamEnd;
+};
+
 /**
  * This class offers a Promise-like interface but replaces `then` with `subscribe`.
  * The function passed into subscribe will then be called each time new data is available.
  * The onResult fn will also be given a `final` param in case the calling code needs to know
  * if the final chunk of data has been received.
  */
 export class Stream<T> {
+    private subscribed = false;
     private onResult?: OnStreamResult<T>;
     private onError?: OnStreamError;
     private onEnd?: OnStreamEnd;
@@ -36,18 +43,13 @@ export class Stream<T> {
         );
     }
 
-    subscribe(onResult: OnStreamResult<T>): Stream<T> {
-        this.onResult = onResult;
-        return this;
-    }
-
-    catch(onError: OnStreamError): Stream<T> {
-        this.onError = onError;
-        return this;
-    }
-
-    finally(onEnd: OnStreamEnd): Stream<T> {
-        this.onEnd = onEnd;
-        return this;
+    subscribe(subscription: Subscription<T>) {
+        if (this.subscribed) {
+            throw new Error("Already subscribed");
+        }
+        this.subscribed = true;
+        this.onResult = subscription.onResult;
+        this.onError = subscription.onError;
+        this.onEnd = subscription.onEnd;
     }
 }

frontend/openchat-worker/src/worker.ts

@@ -159,8 +159,8 @@ function streamReplies(
     chain: Stream<WorkerResponseInner>,
 ) {
     const start = Date.now();
-    chain
-        .subscribe((value, final) => {
+    chain.subscribe({
+        onResult: (value, final) => {
             console.debug(
                 `WORKER: sending streamed reply ${Date.now() - start}ms after subscribing`,
                 correlationId,
@@ -169,8 +169,9 @@ function streamReplies(
                 final,
             );
             sendResponse(correlationId, value, final);
-        })
-        .catch(sendError(correlationId, payload));
+        },
+        onError: sendError(correlationId, payload),
+    });
 }
 
 function executeThenReply(