✨ Negotiate transit/JSON payload format on SSE endpoints

SSE endpoints always encoded event payloads in transit, ignoring
request content negotiation.

Extract the response format negotiation (Accept header and _fmt
query param) into a shared app.http.content-negotiation helper and
use it from both the response formatting middleware and the SSE
response builder. SSE event payloads are now encoded as transit
(default) or plain JSON depending on the request, while the
text/event-stream transport remains unchanged.

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

AI-assisted-by: glm-5.3-flash

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,9 +13,9 @@
    [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.tokens :as tokens]
-   [app.util.pointer-map :as pmap]
    [cuerdas.core :as str]
    [yetti.adapter :as yt]
    [yetti.middleware :as ymw]
@@ -126,12 +126,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]
@@ -153,7 +147,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}]
@@ -183,24 +177,10 @@
                       (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
-                (format-response-with-transit response request))))
+            (case (cnegot/negotiate-format request)
+              :transit (format-response-with-transit response request)
+              :json    (format-response-with-json response request)))
 
           (process-response [response request]
             (cond-> response

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,12 @@
              :cause cause)
       nil)))
 
+(defn- resolve-encoder
+  [format]
+  (case format
+    :transit #(t/encode-str % {:type :json-verbose})
+    :json    cnegot/json-encode-str))
+
 ;; ---- PUBLIC API
 
 (def default-headers
@@ -47,27 +54,32 @@
    "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 [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 encode))
-                          listener (events/spawn-listener
-                                    channel
-                                    (partial write! output)
-                                    (partial pu/close! output))]
-                      (try
-                        (binding [events/*channel* channel]
-                          (let [result (handler)]
-                            (events/tap :end result)))
+                      (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/*channel* channel]
+                            (let [result (handler)]
+                              (events/tap :end result)))
 
-                        (catch Throwable cause
-                          (let [result (errors/handle' cause request)]
-                            (events/tap channel :error result)))
+                          (catch Throwable cause
+                            (let [result (errors/handle' cause request)]
+                              (events/tap channel :error result)))
 
-                        (finally
-                          (sp/close! channel)
-                          (px/await! listener))))))}))
+                          (finally
+                            (sp/close! channel)
+                            (px/await! listener))))))})))

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"}})))))

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]
@@ -158,6 +159,7 @@
   [id params options]
   (let [{:keys [response-type
                 stream?
+                response-format
                 form-data?
                 raw-transit?
                 query-params
@@ -180,11 +182,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))}
@@ -225,7 +232,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)