From 2332be1e2caa124f858d105ac2bd47d127fa6026 Mon Sep 17 00:00:00 2001 From: Andrey Antukh Date: Fri, 28 Aug 2026 09:39:27 +0000 Subject: [PATCH] :sparkles: 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 | 49 +++++++++++- backend/src/app/util/websocket.clj | 20 +++-- .../backend_tests/http_websocket_test.clj | 78 +++++++++++++++++++ 3 files changed, 139 insertions(+), 8 deletions(-) create mode 100644 backend/test/backend_tests/http_websocket_test.clj diff --git a/backend/src/app/http/websocket.clj b/backend/src/app/http/websocket.clj index a9631be1d3..aaf29fdc1f 100644 --- a/backend/src/app/http/websocket.clj +++ b/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)})))) diff --git a/backend/src/app/util/websocket.clj b/backend/src/app/util/websocket.clj index 514de1178c..6e7dd9a3a1 100644 --- a/backend/src/app/util/websocket.clj +++ b/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)))))) diff --git a/backend/test/backend_tests/http_websocket_test.clj b/backend/test/backend_tests/http_websocket_test.clj new file mode 100644 index 0000000000..16a2738898 --- /dev/null +++ b/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)))))