From 8f91f7cd4e7ec7e88fb16424da2938ed5f2e4165 Mon Sep 17 00:00:00 2001 From: Andrey Antukh Date: Fri, 28 Aug 2026 07:23:58 +0000 Subject: [PATCH] :sparkles: 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 | 58 +++++++ backend/src/app/http/middleware.clj | 30 +--- backend/src/app/http/sse.clj | 58 ++++--- backend/src/app/http/websocket.clj | 54 +++++- backend/src/app/util/websocket.clj | 82 +++++---- .../http_content_negotiation_test.clj | 40 +++++ backend/test/backend_tests/http_sse_test.clj | 108 ++++++++++++ .../backend_tests/http_websocket_test.clj | 162 ++++++++++++++++-- frontend/src/app/main/repo.cljs | 13 +- 9 files changed, 501 insertions(+), 104 deletions(-) create mode 100644 backend/src/app/http/content_negotiation.clj create mode 100644 backend/test/backend_tests/http_content_negotiation_test.clj create mode 100644 backend/test/backend_tests/http_sse_test.clj diff --git a/backend/src/app/http/content_negotiation.clj b/backend/src/app/http/content_negotiation.clj new file mode 100644 index 0000000000..6bf76a7824 --- /dev/null +++ b/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)) diff --git a/backend/src/app/http/middleware.clj b/backend/src/app/http/middleware.clj index ad80bfe9c4..052e8fa92e 100644 --- a/backend/src/app/http/middleware.clj +++ b/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] diff --git a/backend/src/app/http/sse.clj b/backend/src/app/http/sse.clj index 4c1420876e..296c753d9a 100644 --- a/backend/src/app/http/sse.clj +++ b/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))))))}))) diff --git a/backend/src/app/http/websocket.clj b/backend/src/app/http/websocket.clj index 50bd64f115..8724ad4778 100644 --- a/backend/src/app/http/websocket.clj +++ b/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)})))) diff --git a/backend/src/app/util/websocket.clj b/backend/src/app/util/websocket.clj index 514de1178c..ee5af1329c 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) @@ -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)))))) diff --git a/backend/test/backend_tests/http_content_negotiation_test.clj b/backend/test/backend_tests/http_content_negotiation_test.clj new file mode 100644 index 0000000000..1cff5641f1 --- /dev/null +++ b/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 {}})))) diff --git a/backend/test/backend_tests/http_sse_test.clj b/backend/test/backend_tests/http_sse_test.clj new file mode 100644 index 0000000000..3215a9fcf8 --- /dev/null +++ b/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"}}))))) diff --git a/backend/test/backend_tests/http_websocket_test.clj b/backend/test/backend_tests/http_websocket_test.clj index 6ffc10e744..5bc3b4b323 100644 --- a/backend/test/backend_tests/http_websocket_test.clj +++ b/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)))))) diff --git a/frontend/src/app/main/repo.cljs b/frontend/src/app/main/repo.cljs index c3dca3b882..ccdaa02865 100644 --- a/frontend/src/app/main/repo.cljs +++ b/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)