Fix live_patch bug caused by returning early from handleMsg

- returning early meant active was never reset so got into bad state
- also add optional onError to tap into errors from server config
- make errors in handleMsg and send routed through onError
- fix import

.changeset/clean-brooms-clap.md

@@ -0,0 +1,8 @@
+---
+"@liveviewjs/lambda-examples": patch
+"liveviewjs": patch
+"@liveviewjs/examples": patch
+"@liveviewjs/express": patch
+---
+
+Fix live_patch early return bug and add option onError server config

packages/core/dist/liveview.d.ts

@@ -1329,6 +1329,7 @@ interface WsHandlerConfig {
     wrapperTemplate?: LiveViewWrapperTemplate;
     flashAdaptor: FlashAdaptor;
     pubSub: PubSub;
+    onError?: (err: any) => void;
 }
 declare class WsHandlerContext {
     #private;
@@ -1357,6 +1358,7 @@ declare class WsHandler {
     handleMsg(msg: Phx.Msg<unknown>): Promise<void>;
     close(): Promise<void>;
     send(reply: PhxReply.Reply): void;
+    private maybeHandleError;
     private cleanupPostReply;
     private viewToDiff;
     private viewToRendered;

packages/core/dist/liveview.js

@@ -3190,16 +3190,17 @@ class WsHandler {
         this.#ws.subscribeToClose(() => this.close);
     }
     async handleMsg(msg) {
-        // attempt to prevent race conditions by queuing messages
-        // if we are already processing a message
-        if (this.#activeMsg) {
-            this.#msgQueue.push(msg);
-            return;
-        }
-        this.#activeMsg = true;
-        const event = msg[exports.Phx.MsgIdx.event];
-        const topic = msg[exports.Phx.MsgIdx.topic];
+        console.log("dispatching message", msg);
         try {
+            // attempt to prevent race conditions by queuing messages
+            // if we are already processing a message
+            if (this.#activeMsg) {
+                this.#msgQueue.push(msg);
+                return;
+            }
+            this.#activeMsg = true;
+            const event = msg[exports.Phx.MsgIdx.event];
+            const topic = msg[exports.Phx.MsgIdx.topic];
             switch (event) {
                 case "phx_join":
                     // phx_join event used for both LiveView joins and LiveUpload joins
@@ -3303,6 +3304,7 @@ class WsHandler {
                     this.send(msg);
                     break;
                 case "live_patch":
+                    console.log("live_patch", msg);
                     // two cases of live_patch: server-side (pushPatch) or client-side (click on link)
                     try {
                         const payload = msg[exports.Phx.MsgIdx.payload];
@@ -3315,16 +3317,17 @@ class WsHandler {
                             const diff = await this.viewToDiff(view);
                             this.send(PhxReply.diffReply(msg, diff));
                             this.cleanupPostReply();
-                            return;
                         }
-                        // case 2: server-side live_patch
-                        const { to } = payload;
-                        // to is relative so need to provide the urlBase determined on initial join
-                        this.#ctx.url = new URL(to, this.#ctx.url);
-                        // let the `LiveView` udpate its context based on the new url
-                        await this.#ctx.liveView.handleParams(this.#ctx.url, this.#ctx.socket);
-                        // send the message on to the client
-                        this.send(msg);
+                        else {
+                            // case 2: server-side live_patch
+                            const { to } = payload;
+                            // to is relative so need to provide the urlBase determined on initial join
+                            this.#ctx.url = new URL(to, this.#ctx.url);
+                            // let the `LiveView` udpate its context based on the new url
+                            await this.#ctx.liveView.handleParams(this.#ctx.url, this.#ctx.socket);
+                            // send the message on to the client
+                            this.send(msg);
+                        }
                     }
                     catch (e) {
                         /* istanbul ignore next */
@@ -3405,15 +3408,15 @@ class WsHandler {
                 default:
                     throw new Error(`unexpected phx protocol event ${event}`);
             }
+            // we're done with this message, so we can process the next one if there is one
+            this.#activeMsg = false;
+            const nextMsg = this.#msgQueue.pop();
+            if (nextMsg) {
+                this.handleMsg(nextMsg);
+            }
         }
         catch (e) {
-            console.error("error handling phx message", e);
-        }
-        // we're done with this message, so we can process the next one if there is one
-        this.#activeMsg = false;
-        const nextMsg = this.#msgQueue.pop();
-        if (nextMsg) {
-            this.handleMsg(nextMsg);
+            this.maybeHandleError(e);
         }
     }
     async close() {
@@ -3422,7 +3425,17 @@ class WsHandler {
         this.handleMsg([null, null, joinId, "phx_leave", null]);
     }
     send(reply) {
-        this.#ws.send(PhxReply.serialize(reply));
+        try {
+            this.#ws.send(PhxReply.serialize(reply), this.maybeHandleError);
+        }
+        catch (e) {
+            this.maybeHandleError(e);
+        }
+    }
+    maybeHandleError(err) {
+        if (err && this.#config.onError) {
+            this.#config.onError(err);
+        }
     }
     async cleanupPostReply() {
         // do post-send lifecycle step

packages/core/dist/liveview.mjs

@@ -3183,16 +3183,17 @@ class WsHandler {
         this.#ws.subscribeToClose(() => this.close);
     }
     async handleMsg(msg) {
-        // attempt to prevent race conditions by queuing messages
-        // if we are already processing a message
-        if (this.#activeMsg) {
-            this.#msgQueue.push(msg);
-            return;
-        }
-        this.#activeMsg = true;
-        const event = msg[Phx.MsgIdx.event];
-        const topic = msg[Phx.MsgIdx.topic];
+        console.log("dispatching message", msg);
         try {
+            // attempt to prevent race conditions by queuing messages
+            // if we are already processing a message
+            if (this.#activeMsg) {
+                this.#msgQueue.push(msg);
+                return;
+            }
+            this.#activeMsg = true;
+            const event = msg[Phx.MsgIdx.event];
+            const topic = msg[Phx.MsgIdx.topic];
             switch (event) {
                 case "phx_join":
                     // phx_join event used for both LiveView joins and LiveUpload joins
@@ -3296,6 +3297,7 @@ class WsHandler {
                     this.send(msg);
                     break;
                 case "live_patch":
+                    console.log("live_patch", msg);
                     // two cases of live_patch: server-side (pushPatch) or client-side (click on link)
                     try {
                         const payload = msg[Phx.MsgIdx.payload];
@@ -3308,16 +3310,17 @@ class WsHandler {
                             const diff = await this.viewToDiff(view);
                             this.send(PhxReply.diffReply(msg, diff));
                             this.cleanupPostReply();
-                            return;
                         }
-                        // case 2: server-side live_patch
-                        const { to } = payload;
-                        // to is relative so need to provide the urlBase determined on initial join
-                        this.#ctx.url = new URL(to, this.#ctx.url);
-                        // let the `LiveView` udpate its context based on the new url
-                        await this.#ctx.liveView.handleParams(this.#ctx.url, this.#ctx.socket);
-                        // send the message on to the client
-                        this.send(msg);
+                        else {
+                            // case 2: server-side live_patch
+                            const { to } = payload;
+                            // to is relative so need to provide the urlBase determined on initial join
+                            this.#ctx.url = new URL(to, this.#ctx.url);
+                            // let the `LiveView` udpate its context based on the new url
+                            await this.#ctx.liveView.handleParams(this.#ctx.url, this.#ctx.socket);
+                            // send the message on to the client
+                            this.send(msg);
+                        }
                     }
                     catch (e) {
                         /* istanbul ignore next */
@@ -3398,15 +3401,15 @@ class WsHandler {
                 default:
                     throw new Error(`unexpected phx protocol event ${event}`);
             }
+            // we're done with this message, so we can process the next one if there is one
+            this.#activeMsg = false;
+            const nextMsg = this.#msgQueue.pop();
+            if (nextMsg) {
+                this.handleMsg(nextMsg);
+            }
         }
         catch (e) {
-            console.error("error handling phx message", e);
-        }
-        // we're done with this message, so we can process the next one if there is one
-        this.#activeMsg = false;
-        const nextMsg = this.#msgQueue.pop();
-        if (nextMsg) {
-            this.handleMsg(nextMsg);
+            this.maybeHandleError(e);
         }
     }
     async close() {
@@ -3415,7 +3418,17 @@ class WsHandler {
         this.handleMsg([null, null, joinId, "phx_leave", null]);
     }
     send(reply) {
-        this.#ws.send(PhxReply.serialize(reply));
+        try {
+            this.#ws.send(PhxReply.serialize(reply), this.maybeHandleError);
+        }
+        catch (e) {
+            this.maybeHandleError(e);
+        }
+    }
+    maybeHandleError(err) {
+        if (err && this.#config.onError) {
+            this.#config.onError(err);
+        }
     }
     async cleanupPostReply() {
         // do post-send lifecycle step

packages/core/src/server/socket/ws/wsHandler.ts

@@ -32,6 +32,7 @@ export interface WsHandlerConfig {
   wrapperTemplate?: LiveViewWrapperTemplate;
   flashAdaptor: FlashAdaptor;
   pubSub: PubSub;
+  onError?: (err: any) => void;
 }
 
 export class WsHandlerContext {
@@ -151,16 +152,16 @@ export class WsHandler {
   }
 
   async handleMsg(msg: Phx.Msg<unknown>) {
-    // attempt to prevent race conditions by queuing messages
-    // if we are already processing a message
-    if (this.#activeMsg) {
-      this.#msgQueue.push(msg);
-      return;
-    }
-    this.#activeMsg = true;
-    const event = msg[Phx.MsgIdx.event];
-    const topic = msg[Phx.MsgIdx.topic];
     try {
+      // attempt to prevent race conditions by queuing messages
+      // if we are already processing a message
+      if (this.#activeMsg) {
+        this.#msgQueue.push(msg);
+        return;
+      }
+      this.#activeMsg = true;
+      const event = msg[Phx.MsgIdx.event];
+      const topic = msg[Phx.MsgIdx.topic];
       switch (event) {
         case "phx_join":
           // phx_join event used for both LiveView joins and LiveUpload joins
@@ -283,8 +284,8 @@ export class WsHandler {
           // two cases of live_patch: server-side (pushPatch) or client-side (click on link)
           try {
             const payload = msg[Phx.MsgIdx.payload] as Phx.LivePatchPayload | Phx.LiveNavPushPayload;
-            // case 1: client-side live_patch
             if (payload.hasOwnProperty("url")) {
+              // case 1: client-side live_patch
               const url = new URL((payload as Phx.LivePatchPayload).url);
               this.#ctx!.url = url;
               await this.#ctx!.liveView.handleParams(url, this.#ctx!.socket);
@@ -295,16 +296,16 @@ export class WsHandler {
               const diff = await this.viewToDiff(view);
               this.send(PhxReply.diffReply(msg, diff));
               this.cleanupPostReply();
-              return;
+            } else {
+              // case 2: server-side live_patch
+              const { to } = payload as Phx.LiveNavPushPayload;
+              // to is relative so need to provide the urlBase determined on initial join
+              this.#ctx!.url = new URL(to, this.#ctx!.url);
+              // let the `LiveView` udpate its context based on the new url
+              await this.#ctx!.liveView.handleParams(this.#ctx!.url, this.#ctx!.socket);
+              // send the message on to the client
+              this.send(msg as PhxReply.Reply);
             }
-            // case 2: server-side live_patch
-            const { to } = payload as Phx.LiveNavPushPayload;
-            // to is relative so need to provide the urlBase determined on initial join
-            this.#ctx!.url = new URL(to, this.#ctx!.url);
-            // let the `LiveView` udpate its context based on the new url
-            await this.#ctx!.liveView.handleParams(this.#ctx!.url, this.#ctx!.socket);
-            // send the message on to the client
-            this.send(msg as PhxReply.Reply);
           } catch (e) {
             /* istanbul ignore next */
             console.error("Error handling live_patch", e);
@@ -379,15 +380,15 @@ export class WsHandler {
         default:
           throw new Error(`unexpected phx protocol event ${event}`);
       }
-    } catch (e) {
-      console.error("error handling phx message", e);
-    }
 
-    // we're done with this message, so we can process the next one if there is one
-    this.#activeMsg = false;
-    const nextMsg = this.#msgQueue.pop();
-    if (nextMsg) {
-      this.handleMsg(nextMsg);
+      // we're done with this message, so we can process the next one if there is one
+      this.#activeMsg = false;
+      const nextMsg = this.#msgQueue.pop();
+      if (nextMsg) {
+        this.handleMsg(nextMsg);
+      }
+    } catch (e) {
+      this.maybeHandleError(e);
     }
   }
 
@@ -398,7 +399,17 @@ export class WsHandler {
   }
 
   send(reply: PhxReply.Reply) {
-    this.#ws.send(PhxReply.serialize(reply));
+    try {
+      this.#ws.send(PhxReply.serialize(reply), this.maybeHandleError);
+    } catch (e) {
+      this.maybeHandleError(e);
+    }
+  }
+
+  private maybeHandleError(err: any) {
+    if (err && this.#config.onError) {
+      this.#config.onError(err);
+    }
   }
 
   private async cleanupPostReply() {

packages/core/src/server/socket/ws/wsUploadHandler.ts

@@ -1,6 +1,6 @@
 import { nanoid } from "nanoid";
-import { LiveViewTemplate } from "src/server/live";
 import { FileSystemAdaptor } from "../../../server/adaptor";
+import { LiveViewTemplate } from "../../../server/live";
 import { Phx } from "../../../server/protocol/phx";
 import { PhxReply } from "../../../server/protocol/reply";
 import { UploadConfig } from "../../../server/upload";

packages/deno/src/deno/server.ts

@@ -30,6 +30,7 @@ interface DenoOakLiveViewServerOptions {
   flashAdaptor?: FlashAdaptor;
   fileSystemAdaptor?: FileSystemAdaptor;
   wrapperTemplate?: LiveViewWrapperTemplate;
+  onError?: (err: any) => void;
 }
 
 type DenoMiddleware = (
@@ -87,6 +88,7 @@ export class DenoOakLiveViewServer implements LiveViewServerAdaptor<DenoMiddlewa
       wrapperTemplate: this.wrapperTemplate,
       flashAdaptor: this.flashAdapter,
       pubSub: this.pubSub,
+      onError: options?.onError,
     };
   }
 

packages/express/dist/liveviewjs-express.d.ts

@@ -42,6 +42,7 @@ interface NodeExpressLiveViewServerOptions {
     flashAdaptor?: FlashAdaptor;
     fileSystemAdaptor?: FileSystemAdaptor;
     wrapperTemplate?: LiveViewWrapperTemplate;
+    onError?: (err: any) => void;
 }
 declare class NodeExpressLiveViewServer implements LiveViewServerAdaptor<RequestHandler, (wsServer: WebSocketServer) => Promise<void>> {
     #private;

packages/express/dist/liveviewjs-express.js

@@ -161,6 +161,7 @@ class NodeExpressLiveViewServer {
             wrapperTemplate: this.wrapperTemplate,
             flashAdaptor: this.flashAdapter,
             pubSub: this.pubSub,
+            onError: options === null || options === void 0 ? void 0 : options.onError,
         }, "f");
     }
     wsMiddleware() {

packages/express/dist/liveviewjs-express.mjs

@@ -151,6 +151,7 @@ class NodeExpressLiveViewServer {
             wrapperTemplate: this.wrapperTemplate,
             flashAdaptor: this.flashAdapter,
             pubSub: this.pubSub,
+            onError: options === null || options === void 0 ? void 0 : options.onError,
         }, "f");
     }
     wsMiddleware() {