✨ Negotiate transit/JSON payload format on SSE and WebSocket

SSE endpoints always encoded event payloads in transit, ignoring
request content negotiation. WebSocket notifications always used
transit as well, with no way to request plain JSON.

Extract the response format negotiation (Accept header and _fmt
query param) into the shared app.http.content-negotiation helper
and reuse it from the response formatting middleware, the SSE
response builder and the WebSocket notifications endpoint. The
_fmt=json parameter takes precedence over Accept, transit stays
the default, and unknown formats fall back to transit.

SSE event payloads are now encoded as transit (default) or plain
JSON depending on the request, while the text/event-stream
transport remains unchanged. The WebSocket format is negotiated
once on the upgrade request and applies bidirectionally: outbound
notification messages use the negotiated encoder and inbound
client messages the matching decoder, with a normalization step
converting JSON messages (:type to keyword, entity ids to UUID)
into the same shape produced by transit.

The generic websocket listener now accepts encode/decode functions
as options (defaulting to the historical transit codecs) and
exposes the resolved options in the handler metadata under
::options for testing and introspection. The notifications
http-handler is now public so tests can cover the wiring.

On the frontend, stream requests accept a :response-format :json
option that switches the Accept header and decodes each event
with the plain JSON decoder.

Add unit tests for negotiate-format, SSE response encoding and
the WebSocket codec wiring (JSON via Accept header or _fmt param,
transit default for unsignaled requests).

Closes #11409

AI-assisted-by: glm-5.3-flash
AI-assisted-by: muse-spark-1.3-contributor

backend/src/app/http/content_negotiation.clj

@@ -0,0 +1,58 @@
+;; 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 app.http.content-negotiation
+  "Content format negotiation helpers shared between the regular
+  response formatting middleware and the SSE streaming endpoints.
+
+  The negotiated format affects only the encoding of the response
+  payload; the transport (regular http body or `text/event-stream`)
+  is not affected."
+  (:require
+   [app.common.json :as json]
+   [app.util.pointer-map :as pmap]
+   [cuerdas.core :as str]
+   [yetti.request :as yreq]))
+
+(defn- format-from-params
+  [{:keys [query-params]}]
+  (and (= "json" (get query-params :_fmt))
+       :json))
+
+(defn negotiate-format
+  "Determine the response payload format for the request. Returns
+  `:json` or `:transit`.
+
+  The `:_fmt=json` query parameter takes precedence over the
+  `Accept` header; when no explicit signal is present, transit is
+  the default."
+  [request]
+  (or (format-from-params request)
+      (let [accept (yreq/get-header request "accept")]
+        (cond
+          (or (= accept "application/transit+json")
+              (str/includes? accept "application/transit+json"))
+          :transit
+
+          (or (= accept "application/json")
+              (str/includes? accept "application/json"))
+          :json
+
+          :else
+          :transit))))
+
+(defn write-json-value
+  [_ val]
+  (if (pmap/pointer-map? val)
+    [(pmap/get-id val) (meta val)]
+    val))
+
+(defn json-encode-str
+  "Encode value as a JSON string using the same conventions as the
+  regular JSON API responses: camelCase keys, keywords/UUIDs and
+  instants serialized as plain JSON scalars."
+  [v]
+  (json/encode v :key-fn json/write-camel-key :value-fn write-json-value))

backend/src/app/http/middleware.clj

@@ -13,11 +13,11 @@
    [app.common.transit :as t]
    [app.config :as cf]
    [app.http :as-alias http]
+   [app.http.content-negotiation :as cnegot]
    [app.http.errors :as errors]
    [app.metrics :as mtx]
    [app.tokens :as tokens]
    [app.util.inet :as inet]
-   [app.util.pointer-map :as pmap]
    [cuerdas.core :as str]
    [yetti.adapter :as yt]
    [yetti.middleware :as ymw]
@@ -186,12 +186,6 @@
 
 (def ^:const buffer-size (:xnio/buffer-size yt/defaults))
 
-(defn- write-json-value
-  [_ val]
-  (if (pmap/pointer-map? val)
-    [(pmap/get-id val) (meta val)]
-    val))
-
 (defn wrap-format-response
   [handler]
   (letfn [(transit-streamable-body [data opts _ output-stream]
@@ -213,7 +207,7 @@
                     data   (encode data)]
                 (with-open [^OutputStream bos (buffered-output-stream output-stream buffer-size)]
                   (with-open [^java.io.OutputStreamWriter writer (java.io.OutputStreamWriter. bos)]
-                    (json/write writer data :key-fn json/write-camel-key :value-fn write-json-value))))
+                    (json/write writer data :key-fn json/write-camel-key :value-fn cnegot/write-json-value))))
               (catch java.io.IOException _)
               (catch Throwable cause
                 (binding [l/*context* {:value data}]
@@ -243,23 +237,11 @@
                       (assoc ::yres/body (yres/stream-body (partial transit-streamable-body body opts)))))
                 response)))
 
-          (format-from-params [{:keys [query-params] :as request}]
-            (and (= "json" (get query-params :_fmt))
-                 "application/json"))
-
           (format-response [response request]
-            (let [accept (or (format-from-params request)
-                             (yreq/get-header request "accept"))]
-              (cond
-                (or (= accept "application/transit+json")
-                    (str/includes? accept "application/transit+json"))
-                (format-response-with-transit response request)
-
-                (or (= accept "application/json")
-                    (str/includes? accept "application/json"))
-                (format-response-with-json response request)
-
-                :else
+            (let [format (cnegot/negotiate-format request)]
+              (case format
+                :transit (format-response-with-transit response request)
+                :json    (format-response-with-json response request)
                 (format-response-with-transit response request))))
 
           (process-response [response request]

backend/src/app/http/sse.clj

@@ -10,6 +10,7 @@
    [app.common.data :as d]
    [app.common.logging :as l]
    [app.common.transit :as t]
+   [app.http.content-negotiation :as cnegot]
    [app.http.errors :as errors]
    [app.util.events :as events]
    [promesa.exec :as px]
@@ -26,11 +27,11 @@
   (.flush output))
 
 (defn- encode
-  [[name data]]
+  [encode-fn [name data]]
   (try
     (let [data (with-out-str
                  (println "event:" (d/name name))
-                 (println "data:" (t/encode-str data {:type :json-verbose}))
+                 (println "data:" (encode-fn data))
                  (println))]
       (.getBytes ^String data "UTF-8"))
     (catch Throwable cause
@@ -38,6 +39,13 @@
              :cause cause)
       nil)))
 
+(defn- resolve-encoder
+  [format]
+  (case format
+    :transit #(t/encode-str % {:type :json-verbose})
+    :json    cnegot/json-encode-str
+    #(t/encode-str % {:type :json-verbose})))
+
 ;; ---- PUBLIC API
 
 (def default-headers
@@ -47,26 +55,30 @@
    "X-Accel-Buffering" "no"})
 
 (defn response
+  "Create a streaming SSE response. The payload of each event is
+  encoded in transit or plain JSON depending on the request content
+  negotiation (see `app.http.content-negotiation/negotiate-format`),
+  transit being the default."
   [handler & {:keys [buf] :or {buf 32} :as opts}]
   (fn [request]
-    {::yres/headers default-headers
-     ::yres/status 200
-     ::yres/body (yres/stream-body
-                  (fn [_ output]
-
-                    (let [channel  (sp/chan :buf buf :xf (keep encode))
-                          listener (events/spawn-listener
-                                    channel
-                                    (partial write! output)
-                                    (partial pu/close! output))]
-                      (try
-                        (binding [events/*sink* channel]
-                          (try
-                            (let [result (handler)]
-                              (events/tap :end result))
-                            (catch Throwable cause
-                              (let [result (errors/handle' cause request)]
-                                (events/tap :error result)))))
-                        (finally
-                          (sp/close! channel)
-                          (px/await! listener))))))}))
+    (let [encode-fn (resolve-encoder (cnegot/negotiate-format request))]
+      {::yres/headers default-headers
+       ::yres/status 200
+       ::yres/body (yres/stream-body
+                    (fn [_ output]
+                      (let [channel  (sp/chan :buf buf :xf (keep (partial encode encode-fn)))
+                            listener (events/spawn-listener
+                                      channel
+                                      (partial write! output)
+                                      (partial pu/close! output))]
+                        (try
+                          (binding [events/*sink* channel]
+                            (try
+                              (let [result (handler)]
+                                (events/tap :end result))
+                              (catch Throwable cause
+                                (let [result (errors/handle' cause request)]
+                                  (events/tap :error result)))))
+                          (finally
+                            (sp/close! channel)
+                            (px/await! listener))))))})))

backend/src/app/http/websocket.clj

@@ -10,12 +10,15 @@
    [app.binfile.common :as bfc]
    [app.common.data.macros :as dm]
    [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]
@@ -88,6 +91,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)))
@@ -293,7 +329,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]
@@ -303,9 +340,13 @@
             :inc 1)
   message)
 
-(defn- http-handler
+(defn http-handler
+  "The websocket upgrade request handler. Returns the upgrade
+  response containing the websocket listener built with the
+  codecs of the negotiated payload format."
   [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
@@ -327,12 +368,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)
@@ -60,10 +63,16 @@
 
   It also accepts some options that allows you parametrize the
   protocol behavior. The options map will be used as-as for the
-  initial data of the `ws` data structure"
+  initial data of the `ws` data structure.
+
+  The returned map has the resolved options attached in its
+  metadata under the `::options` key; it is intended for testing
+  and introspection purposes only."
   [request & {:keys [::on-rcv-message
                      ::on-snd-message
                      ::on-connect
+                     ::encode-fn
+                     ::decode-fn
                      ::input-buff-size
                      ::output-buff-size
                      ::idle-timeout]
@@ -72,7 +81,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,40 +110,44 @@
                        (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]
-       (l/dbg :fn "on-open" :conn-id (str id))
-       (let [options (-> options
-                         (assoc ::channel channel)
-                         (on-connect))
-             timeout (ct/duration idle-timeout)]
+    (with-meta
+      {:on-open
+       (fn on-open [channel]
+         (l/dbg :fn "on-open" :conn-id (str id))
+         (let [options (-> options
+                           (assoc ::channel channel)
+                           (on-connect))
+               timeout (ct/duration idle-timeout)]
 
-         (yws/set-idle-timeout! channel timeout)
-         (px/submit! :vthread (partial start-io-loop! options))))
+           (yws/set-idle-timeout! channel timeout)
+           (px/submit! :vthread (partial start-io-loop! options))))
 
-     :on-close
-     (fn on-close [_channel code reason]
-       (l/dbg :fn "on-close"
-              :conn-id (str id)
-              :code code
-              :reason reason)
-       (sp/close! close-ch))
+       :on-close
+       (fn on-close [_channel code reason]
+         (l/dbg :fn "on-close"
+                :conn-id (str id)
+                :code code
+                :reason reason)
+         (sp/close! close-ch))
 
-     :on-error
-     (fn on-error [_channel cause]
-       (sp/close! close-ch cause))
+       :on-error
+       (fn on-error [_channel cause]
+         (sp/close! close-ch cause))
 
-     :on-message
-     (fn on-message [_channel message]
-       (when (string? message)
-         (sp/offer! input-ch message)
-         (swap! state assoc ::last-activity-at (ct/now))))
+       :on-message
+       (fn on-message [_channel message]
+         (when (string? message)
+           (sp/offer! input-ch message)
+           (swap! state assoc ::last-activity-at (ct/now))))
 
-     :on-pong
-     (fn on-pong [_channel data]
-       (sp/put! hbeat-ch data))}))
+       :on-pong
+       (fn on-pong [_channel data]
+         (sp/put! hbeat-ch data))}
+      {::options options})))
 
 (defn- handle-ping!
   [{:keys [::id ::beats ::channel] :as wsp} beat-id]
@@ -143,7 +158,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 +185,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 +197,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_content_negotiation_test.clj

@@ -0,0 +1,40 @@
+;; 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-content-negotiation-test
+  (:require
+   [app.http.content-negotiation :as cnegot]
+   [clojure.test :as t]))
+
+(t/deftest negotiate-format-defaults-to-transit
+  (t/is (= :transit (cnegot/negotiate-format {})))
+  (t/is (= :transit (cnegot/negotiate-format {:headers {"accept" "*/*"}})))
+  (t/is (= :transit (cnegot/negotiate-format {:headers {"accept" "text/event-stream"}}))))
+
+(t/deftest negotiate-format-from-accept-header
+  (t/is (= :transit (cnegot/negotiate-format
+                     {:headers {"accept" "application/transit+json"}})))
+  (t/is (= :transit (cnegot/negotiate-format
+                     {:headers {"accept" "application/transit+json,text/event-stream,*/*"}})))
+  (t/is (= :json (cnegot/negotiate-format
+                  {:headers {"accept" "application/json"}})))
+  (t/is (= :json (cnegot/negotiate-format
+                  {:headers {"accept" "application/json, text/event-stream"}}))))
+
+(t/deftest negotiate-format-from-query-param
+  (t/is (= :json (cnegot/negotiate-format {:query-params {:_fmt "json"}})))
+  (t/is (= :json (cnegot/negotiate-format
+                  {:query-params {:_fmt "json"}
+                   :headers {"accept" "application/transit+json"}}))))
+
+(t/deftest negotiate-format-only-json-param-is-recognized
+  (t/is (= :json (cnegot/negotiate-format
+                  {:query-params {:_fmt "transit"}
+                   :headers {"accept" "application/json"}})))
+  (t/is (= :transit (cnegot/negotiate-format {:query-params {:_fmt "transit"}}))))
+
+(t/deftest negotiate-format-without-accept-header
+  (t/is (= :transit (cnegot/negotiate-format {:headers {}}))))

backend/test/backend_tests/http_sse_test.clj

@@ -0,0 +1,108 @@
+;; 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-sse-test
+  (:require
+   [app.common.transit :as tr]
+   [app.http.sse :as sse]
+   [app.util.events :as events]
+   [clojure.string :as str]
+   [clojure.test :as t]
+   [yetti.response :as yres])
+  (:import
+   java.io.ByteArrayOutputStream))
+
+(def ^:private test-file-id #uuid "00000000-0000-0000-0000-000000000001")
+
+(def ^:private progress-json
+  (str "{\"fileId\":\"" test-file-id "\",\"index\":1,\"total\":2}"))
+
+(defn- make-handler
+  []
+  (fn []
+    (events/tap :progress {:file-id test-file-id :index 1 :total 2})
+    {:status "ok"}))
+
+(defn- run-sse
+  "Runs the sse response with the provided fake request and returns
+  the raw stream content as a string."
+  [request]
+  (let [response ((sse/response (make-handler)) request)
+        output   (ByteArrayOutputStream.)]
+    (yres/write-body-to-stream (::yres/body response) response output)
+    (.toString output "UTF-8")))
+
+(defn- data-events
+  [stream]
+  (->> (str/split stream #"\n\n")
+       (keep (fn [block]
+               (let [[_ event] (re-find #"event: (.+)" block)
+                     [_ data]  (re-find #"data: (.+)" block)]
+                 (when (and event data)
+                   {:event (str/trim event) :data (str/trim data)}))))
+       (vec)))
+
+(t/deftest default-format-is-transit
+  (let [[progress end] (data-events (run-sse {}))]
+    (t/is (= "progress" (:event progress)))
+    (t/is (= {:file-id test-file-id :index 1 :total 2}
+             (tr/decode-str (:data progress))))
+    (t/is (= "end" (:event end)))
+    (t/is (= {:status "ok"} (tr/decode-str (:data end))))))
+
+(t/deftest accept-transit-header-uses-transit
+  (let [[progress] (data-events (run-sse {:headers {"accept" "application/transit+json"}}))]
+    (t/is (= {:file-id test-file-id :index 1 :total 2}
+             (tr/decode-str (:data progress))))))
+
+(t/deftest accept-frontend-combo-header-uses-transit
+  (let [[progress] (data-events (run-sse {:headers {"accept" "application/transit+json,text/event-stream,*/*"}}))]
+    (t/is (= {:file-id test-file-id :index 1 :total 2}
+             (tr/decode-str (:data progress))))))
+
+(t/deftest accept-json-header-uses-json
+  (let [stream (run-sse {:headers {"accept" "application/json"}})
+        [progress end] (data-events stream)]
+    (t/is (= "progress" (:event progress)))
+    (t/is (= progress-json (:data progress)))
+    (t/is (= "end" (:event end)))
+    (t/is (= "{\"status\":\"ok\"}" (:data end)))))
+
+(t/deftest fmt-json-query-param-uses-json
+  (let [[progress] (data-events (run-sse {:query-params {:_fmt "json"}}))]
+    (t/is (= progress-json (:data progress)))))
+
+(t/deftest fmt-json-query-param-precedence-over-accept
+  (let [[progress] (data-events (run-sse {:query-params {:_fmt "json"}
+                                          :headers {"accept" "application/transit+json"}}))]
+    (t/is (= progress-json (:data progress)))))
+
+(t/deftest wildcard-accept-defaults-to-transit
+  (let [[progress] (data-events (run-sse {:headers {"accept" "*/*"}}))]
+    (t/is (= {:file-id test-file-id :index 1 :total 2}
+             (tr/decode-str (:data progress))))))
+
+(t/deftest error-events-are-negotiated
+  (let [response ((sse/response (fn []
+                                  (throw (ex-info "boom"
+                                                  {:type :validation
+                                                   :code :generic
+                                                   :hint "boom"}))))
+                  {:headers {"accept" "application/json"}})
+        output   (ByteArrayOutputStream.)]
+    (yres/write-body-to-stream (::yres/body response) response output)
+    (let [[error] (data-events (.toString output "UTF-8"))]
+      (t/is (= "error" (:event error)))
+      (t/is (= "{\"type\":\"validation\",\"code\":\"generic\",\"hint\":\"boom\"}"
+               (:data error))))))
+
+(t/deftest content-type-is-event-stream-on-both-formats
+  (let [check (fn [request]
+                (let [response ((sse/response (make-handler)) request)]
+                  (get (::yres/headers response) "Content-Type")))]
+    (t/is (= "text/event-stream;charset=UTF-8" (check {})))
+    (t/is (= "text/event-stream;charset=UTF-8"
+             (check {:headers {"accept" "application/json"}})))))

backend/test/backend_tests/http_websocket_test.clj

@@ -6,29 +6,33 @@
 
 (ns backend-tests.http-websocket-test
   (:require
+   [app.common.json :as json]
+   [app.common.transit :as tr]
    [app.common.uuid :as uuid]
    [app.db :as db]
-   [app.http.websocket :as ws]
+   [app.http.session :as-alias session]
+   [app.http.websocket :as http.ws]
    [app.msgbus :as mbus]
    [app.nitrate :as nitrate]
    [app.rpc :as-alias rpc]
    [app.rpc.commands.files :as files]
    [app.rpc.commands.teams :as teams]
-   [app.util.websocket :as util-ws]
+   [app.util.websocket :as ws]
    [backend-tests.helpers :as th]
    [clojure.test :as t]
-   [promesa.exec.csp :as sp]))
+   [promesa.exec.csp :as sp]
+   [yetti.websocket :as yws]))
 
 (t/use-fixtures :once th/state-init)
 (t/use-fixtures :each th/database-reset)
 
 (defn make-wsp
   [profile-id state output-ch]
-  {::util-ws/id (uuid/next)
-   ::util-ws/state state
-   ::util-ws/output-ch output-ch
-   ::ws/profile-id profile-id
-   ::ws/session-id (uuid/next)})
+  {::ws/id (uuid/next)
+   ::ws/state state
+   ::ws/output-ch output-ch
+   ::http.ws/profile-id profile-id
+   ::http.ws/session-id (uuid/next)})
 
 (t/deftest subscribe-file-permission-check
   (let [profile1 (th/create-profile* 1 {:is-active true})
@@ -44,7 +48,7 @@
         (t/is (thrown-with-msg?
                clojure.lang.ExceptionInfo
                #"not found"
-               ((get-method ws/handle-message :subscribe-file)
+               ((get-method http.ws/handle-message :subscribe-file)
                 cfg wsp {:file-id (:id file)})))))
 
     (t/testing "permission check passes for authorized user"
@@ -63,7 +67,7 @@
         (t/is (thrown-with-msg?
                clojure.lang.ExceptionInfo
                #"not found"
-               ((get-method ws/handle-message :subscribe-team)
+               ((get-method http.ws/handle-message :subscribe-team)
                 cfg wsp {:team-id (:id team)})))))
 
     (t/testing "permission check passes for authorized user"
@@ -80,11 +84,11 @@
     (with-redefs [nitrate/call nitrate-call
                   mbus/sub!    (fn [_ & {:keys [topics]}]
                                  (swap! calls conj topics))]
-      ((get-method ws/handle-message :subscribe-team)
+      ((get-method http.ws/handle-message :subscribe-team)
        th/*system* wsp {:team-id team-id}))
-    (some-> @state ::ws/team-subscription :channel sp/close!)
+    (some-> @state ::http.ws/team-subscription :channel sp/close!)
     {:topics @calls
-     :subscription (::ws/team-subscription @state)}))
+     :subscription (::http.ws/team-subscription @state)}))
 
 (t/deftest subscribe-team-subscribes-to-team-organization
   (let [profile (th/create-profile* 1 {:is-active true})
@@ -120,9 +124,9 @@
         cfg      th/*system*
         file-id  (:id file)
         sub-ch   (sp/chan :buf (sp/dropping-buffer 64))
-        state    (atom {::ws/file-subscription {:file-id file-id
-                                                :channel sub-ch
-                                                :topic file-id}})
+        state    (atom {::http.ws/file-subscription {:file-id file-id
+                                                     :channel sub-ch
+                                                     :topic file-id}})
         output-ch (sp/chan :buf (sp/dropping-buffer 64))
         wsp      (make-wsp (:id profile) state output-ch)]
 
@@ -132,7 +136,7 @@
                        :position {:x 10 :y 20}
                        :zoom 1.0}]
         (t/is (nil?
-               ((get-method ws/handle-message :pointer-update)
+               ((get-method http.ws/handle-message :pointer-update)
                 cfg wsp wrong-msg)))))
 
     (t/testing "does nothing when no file subscription exists"
@@ -143,5 +147,127 @@
                          :position {:x 10 :y 20}
                          :zoom 1.0}]
         (t/is (nil?
-               ((get-method ws/handle-message :pointer-update)
+               ((get-method http.ws/handle-message :pointer-update)
                 cfg empty-wsp msg)))))))
+
+(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)))))
+
+(defn- ws-request
+  [{:keys [headers params]}]
+  {:params (merge {:session-id (str (uuid/next))} params)
+   ::session/profile-id (uuid/next)
+   :headers (merge {"upgrade" "websocket"
+                    "connection" "Upgrade"}
+                   headers)})
+
+(defn- listener-options
+  [request]
+  (let [response (http.ws/http-handler {} request)
+        listener (::yws/listener response)]
+    (t/is (map? listener))
+    (t/is (fn? (:on-open listener)))
+    (t/is (fn? (:on-close listener)))
+    (t/is (fn? (:on-error listener)))
+    (t/is (fn? (:on-message listener)))
+    (t/is (fn? (:on-pong listener)))
+    (::ws/options (meta listener))))
+
+(t/deftest http-handler-negotiates-json-codecs
+  (let [request (ws-request {:headers {"accept" "application/json"}})
+        options (listener-options request)
+        encode  (::ws/encode-fn options)
+        decode  (::ws/decode-fn options)
+        file-id (uuid/next)
+        raw     (str "{\"type\":\"subscribe-file\",\"requestId\":42,"
+                     "\"fileId\":\"" file-id "\"}")]
+    (let [decoded (decode raw)]
+      (t/is (= "subscribe-file" (:type decoded)))
+      (t/is (= 42 (:request-id decoded)))
+      (t/is (= (str file-id) (:file-id decoded))))
+    (let [encoded (encode {:type :join-file
+                           :file-id file-id
+                           :request-id 42})]
+      (t/is (string? encoded))
+      (t/is (= "join-file" (get (json/decode encoded) "type")))
+      (t/is (= 42 (get (json/decode encoded) "requestId")))
+      (t/is (= (str file-id) (get (json/decode encoded) "fileId"))))))
+
+(t/deftest http-handler-negotiates-json-codecs-from-fmt-param
+  (let [request (-> (ws-request {:headers {"accept" "application/transit+json"}})
+                    (assoc :query-params {:_fmt "json"})
+                    (assoc-in [:params :_fmt] "json"))
+        options (listener-options request)
+        encoded ((::ws/encode-fn options) {:type :join-file})]
+    (t/is (string? encoded))
+    (t/is (= "join-file" (get (json/decode encoded) "type")))))
+
+(t/deftest http-handler-defaults-to-transit-codecs
+  (let [request (ws-request {})
+        options (listener-options request)
+        encode  (::ws/encode-fn options)
+        decode  (::ws/decode-fn options)
+        msg     {:type :join-file :file-id (uuid/next)}]
+    (t/is (= msg (decode (encode msg))))))

frontend/src/app/main/repo.cljs

@@ -8,6 +8,7 @@
   (:require
    [app.common.data :as d]
    [app.common.exceptions :as ex]
+   [app.common.json :as json]
    [app.common.logging :as log]
    [app.common.time :as ct]
    [app.common.transit :as t]
@@ -218,6 +219,7 @@
   [id params options]
   (let [{:keys [response-type
                 stream?
+                response-format
                 form-data?
                 raw-transit?
                 query-params
@@ -240,11 +242,16 @@
         response-type
         (d/nilv response-type :text)
 
+        accept
+        (if (and stream? (= response-format :json))
+          "application/json,text/event-stream,*/*"
+          "application/transit+json,text/event-stream,*/*")
+
         request
         {:method method
          :uri (u/join cf/public-uri "api/main/methods/" nid)
          :credentials "include"
-         :headers {"accept" "application/transit+json,text/event-stream,*/*"
+         :headers {"accept" accept
                    "x-external-session-id" (cf/external-session-id)
                    "x-session-id" (str cf/session-id)
                    "x-event-origin" (::ev/origin (meta params))}
@@ -285,7 +292,9 @@
 
                                 (if response-stream?
                                   (-> (sse/create-stream body)
-                                      (sse/read-stream t/decode-str))
+                                      (sse/read-stream (if (= response-format :json)
+                                                         json/decode
+                                                         t/decode-str)))
 
                                   (->> response
                                        (http/process-response-type response-type)