✨ Negotiate transit/JSON payload format on WebSocket

Extend the transit/JSON content negotiation to the WebSocket
notifications endpoint (/ws/notifications). The format is
negotiated once on the upgrade request (Accept header or
_fmt=json query parameter, same rules as the rest of the RPC
API) and applies bidirectionally: outbound notification
messages are encoded with the negotiated encoder and inbound
client messages are decoded accordingly.

The generic websocket protocol implementation now accepts
encode/decode functions as listener options, defaulting to
the historical transit encoding. A small normalization step
converts JSON decoded messages (:type to keyword and entity
ids to UUID objects) into the same shape produced by transit.

Closes #11409

AI-assisted-by: glm-5.3-flash

backend/src/app/http/websocket.clj

@@ -9,12 +9,15 @@
   (:require
    [app.binfile.common :as bfc]
    [app.common.exceptions :as ex]
+   [app.common.json :as json]
    [app.common.logging :as l]
    [app.common.pprint :as pp]
    [app.common.schema :as sm]
    [app.common.time :as ct]
+   [app.common.transit :as t]
    [app.common.uuid :as uuid]
    [app.db :as db]
+   [app.http.content-negotiation :as cnegot]
    [app.http.session :as session]
    [app.metrics :as mtx]
    [app.msgbus :as mbus]
@@ -84,6 +87,39 @@
 ;; WEBSOCKET HANDLER
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
+(defn resolve-encoder
+  "Resolve the outbound message encoder for the negotiated format.
+  The transit encoding is the default for unexpected format values."
+  [format]
+  (case format
+    :transit #(t/encode-str % {:type :json-verbose})
+    :json    cnegot/json-encode-str
+    #(t/encode-str % {:type :json-verbose})))
+
+(defn resolve-decoder
+  "Resolve the inbound message decoder for the negotiated format.
+  The transit decoding is the default for unexpected format values."
+  [format]
+  (case format
+    :transit t/decode-str
+    :json    #(json/decode % {:key-fn json/read-kebab-key})
+    t/decode-str))
+
+(defn normalize-message
+  "Normalize the inbound message into the same shape produced by the
+  transit decoding: the `:type` value is a keyword and the entity ids
+  are UUID objects. This is a no-op for transit decoded messages."
+  [message]
+  (cond-> message
+    (string? (:type message))
+    (assoc :type (keyword (:type message)))
+
+    (string? (:team-id message))
+    (assoc :team-id (uuid/parse* (:team-id message)))
+
+    (string? (:file-id message))
+    (assoc :file-id (uuid/parse* (:file-id message)))))
+
 (defmulti handle-message
   (fn [_ _ message]
     (:type message)))
@@ -267,7 +303,8 @@
             :id :websocket-messages-total
             :labels recv-labels
             :inc 1)
-  (assoc message :profile-id profile-id :session-id session-id))
+  (let [message (normalize-message message)]
+    (assoc message :profile-id profile-id :session-id session-id)))
 
 (defn- on-snd-message
   [{:keys [::mtx/metrics]} message]
@@ -279,7 +316,8 @@
 
 (defn- http-handler
   [cfg {:keys [params ::session/profile-id] :as request}]
-  (let [session-id (some-> params :session-id uuid/parse*)]
+  (let [session-id (some-> params :session-id uuid/parse*)
+        format     (cnegot/negotiate-format request)]
     (when-not (uuid? session-id)
       (ex/raise :type :validation
                 :code :missing-session-id
@@ -301,12 +339,17 @@
 
       :else
       (do
-        (l/trace :hint "websocket request" :profile-id profile-id :session-id session-id)
+        (l/trace :hint "websocket request"
+                 :profile-id profile-id
+                 :session-id session-id
+                 :format format)
         {::yws/listener (ws/listener request
                                      ::ws/on-rcv-message (partial on-rcv-message cfg)
                                      ::ws/on-snd-message (partial on-snd-message cfg)
                                      ::ws/on-connect (partial on-connect cfg)
                                      ::ws/handler (partial handle-message cfg)
+                                     ::ws/encode-fn (resolve-encoder format)
+                                     ::ws/decode-fn (resolve-decoder format)
                                      ::profile-id profile-id
                                      ::session-id session-id)}))))
 

backend/src/app/util/websocket.clj

@@ -26,6 +26,9 @@
 (def max-missed-heartbeats 3)
 (def heartbeat-interval 5000)
 
+(def default-encode-fn #(t/encode-str % {:type :json-verbose}))
+(def default-decode-fn t/decode-str)
+
 (defn- encode-beat
   [n]
   (doto (ByteBuffer/allocate 8)
@@ -64,6 +67,8 @@
   [request & {:keys [::on-rcv-message
                      ::on-snd-message
                      ::on-connect
+                     ::encode-fn
+                     ::decode-fn
                      ::input-buff-size
                      ::output-buff-size
                      ::idle-timeout]
@@ -72,7 +77,9 @@
                    idle-timeout 60000
                    on-connect identity
                    on-snd-message identity-3
-                   on-rcv-message identity-3}
+                   on-rcv-message identity-3
+                   encode-fn default-encode-fn
+                   decode-fn default-decode-fn}
               :as options}]
 
   (assert (fn? on-rcv-message) "'on-rcv-message' should be a function")
@@ -99,7 +106,9 @@
                        (assoc ::output-ch output-ch)
                        (assoc ::close-ch close-ch)
                        (assoc ::remote-addr ip-addr)
-                       (assoc ::user-agent uagent))]
+                       (assoc ::user-agent uagent)
+                       (assoc ::encode-fn encode-fn)
+                       (assoc ::decode-fn decode-fn))]
 
     {:on-open
      (fn on-open [channel]
@@ -143,7 +152,8 @@
 
 (defn- start-io-loop!
   [{:keys [::id ::close-ch ::input-ch ::output-ch ::heartbeat-ch
-           ::channel ::handler ::beats ::on-rcv-message ::on-snd-message]
+           ::channel ::handler ::beats ::on-rcv-message ::on-snd-message
+           ::encode-fn ::decode-fn]
     :as wsp}]
   (try
     (handler wsp {:type :open})
@@ -169,7 +179,7 @@
               (recur i))
 
             (identical? p input-ch)
-            (let [message (t/decode-str msg)
+            (let [message (decode-fn msg)
                   message (on-rcv-message message)
                   {:keys [request-id] :as response} (handler wsp message)]
               (when (map? response)
@@ -181,7 +191,7 @@
 
             (identical? p output-ch)
             (let [message (on-snd-message msg)
-                  message (t/encode-str message {:type :json-verbose})]
+                  message (encode-fn message)]
               (yws/send channel message)
               (recur i))))))
 

backend/test/backend_tests/http_websocket_test.clj

@@ -0,0 +1,78 @@
+;; This Source Code Form is subject to the terms of the Mozilla Public
+;; License, v. 2.0. If a copy of the MPL was not distributed with this
+;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
+;;
+;; Copyright (c) KALEIDOS SUBSIDIARY SL
+
+(ns backend-tests.http-websocket-test
+  (:require
+   [app.common.json :as json]
+   [app.common.transit :as tr]
+   [app.common.uuid :as uuid]
+   [app.http.websocket :as http.ws]
+   [clojure.test :as t]))
+
+(t/deftest resolve-encoder-defaults-to-transit
+  (let [encode (http.ws/resolve-encoder :transit)
+        msg    {:type :join-file
+                :file-id (uuid/next)
+                :profile-id (uuid/next)}]
+    (t/is (= msg (tr/decode-str (encode msg))))))
+
+(t/deftest resolve-encoder-unexpected-format-falls-back-to-transit
+  (let [encode (http.ws/resolve-encoder :foo)
+        msg    {:type :join-file :file-id (uuid/next)}]
+    (t/is (= msg (tr/decode-str (encode msg))))))
+
+(t/deftest resolve-encoder-json
+  (let [encode  (http.ws/resolve-encoder :json)
+        file-id (uuid/next)
+        encoded (encode {:type :join-file :file-id file-id})]
+    (t/is (string? encoded))
+    (t/is (= "join-file" (get (json/decode encoded) "type")))
+    (t/is (= (str file-id) (get (json/decode encoded) "fileId")))))
+
+(t/deftest resolve-decoder-defaults-to-transit
+  (let [decode (http.ws/resolve-decoder :transit)
+        msg    {:type :subscribe-file :file-id (uuid/next)}]
+    (t/is (= msg (decode (tr/encode-str msg {:type :json-verbose}))))))
+
+(t/deftest resolve-decoder-unexpected-format-falls-back-to-transit
+  (let [decode (http.ws/resolve-decoder :foo)
+        msg    {:type :subscribe-file :file-id (uuid/next)}]
+    (t/is (= msg (decode (tr/encode-str msg {:type :json-verbose}))))))
+
+(t/deftest resolve-decoder-json
+  (let [decode (http.ws/resolve-decoder :json)
+        file-id (uuid/next)
+        decoded (decode (str "{\"type\":\"subscribe-file\","
+                             "\"fileId\":\"" file-id "\"}"))]
+    (t/is (= "subscribe-file" (:type decoded)))
+    (t/is (= (str file-id) (:file-id decoded)))))
+
+(t/deftest normalize-message-from-json-decoding
+  (let [file-id (uuid/next)
+        decoded {:type "subscribe-file"
+                 :file-id (str file-id)}
+        message (http.ws/normalize-message decoded)]
+    (t/is (= :subscribe-file (:type message)))
+    (t/is (uuid? (:file-id message)))
+    (t/is (= file-id (:file-id message)))))
+
+(t/deftest normalize-message-noop-for-transit-decoding
+  (let [file-id (uuid/next)
+        decoded {:type :subscribe-file :file-id file-id}]
+    (t/is (= decoded (http.ws/normalize-message decoded)))))
+
+(t/deftest normalize-message-with-invalid-uuid
+  (let [message (http.ws/normalize-message
+                 {:type "subscribe-file" :file-id "invalid"})]
+    (t/is (= :subscribe-file (:type message)))
+    (t/is (nil? (:file-id message)))))
+
+(t/deftest normalize-message-team-id
+  (let [team-id (uuid/next)
+        message (http.ws/normalize-message
+                 {:type "subscribe-team" :team-id (str team-id)})]
+    (t/is (= :subscribe-team (:type message)))
+    (t/is (= team-id (:team-id message)))))