♻️ Refactor websockets subsystem (on backend)

- Refactor msgbus subsystem, simplifying many parts.
- Enable persistent websocket connection for the all session duration.

backend/deps.edn

@@ -19,7 +19,7 @@
   io.lettuce/lettuce-core {:mvn/version "6.1.6.RELEASE"}
   java-http-clj/java-http-clj {:mvn/version "0.4.3"}
 
-  funcool/yetti {:git/tag "v6.0" :git/sha "4c8690e"
+  funcool/yetti {:git/tag "v8.0" :git/sha "ea7162d"
                  :git/url "https://github.com/funcool/yetti.git"
                  :exclusions [org.slf4j/slf4j-api]}
 

backend/src/app/http/websocket.clj

@@ -22,51 +22,163 @@
 ;; WEBSOCKET HANDLER
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
-(declare send-presence!)
-
 (defmulti handle-message
-  (fn [_wsp message] (:type message)))
+  (fn [_ message]
+    (:type message)))
 
 (defmethod handle-message :connect
   [wsp _]
-  (let [{:keys [msgbus file-id team-id session-id ::ws/output-ch]} @wsp
+  (l/trace :fn "handle-message" :event :connect)
-        sub-ch (a/chan (a/dropping-buffer 32))]
 
-    (swap! wsp assoc :sub-ch sub-ch)
+  (let [msgbus-fn  (:msgbus @wsp)
+        profile-id (::profile-id @wsp)
+        session-id (::session-id @wsp)
+        output-ch  (::ws/output-ch @wsp)
 
-    ;; Start a subscription forwarding goroutine
+        xform      (remove #(= (:session-id %) session-id))
-    (a/go-loop []
+        channel    (a/chan (a/dropping-buffer 16) xform)]
-      (when-let [val (a/<! sub-ch)]
-        (when-not (= (:session-id val) session-id)
-          ;; If we receive a connect message of other user, we need
-          ;; to send an update presence to all participants.
-          (when (= :connect (:type val))
-            (a/<! (send-presence! @wsp :presence)))
 
-          ;; Then, just forward the message
+    (swap! wsp assoc ::profile-subs-channel channel)
-          (a/>! output-ch val))
+    (a/pipe channel output-ch false)
-        (recur)))
+    (msgbus-fn :cmd :sub :topic profile-id :chan channel)))
-
-    (a/go
-      (a/<! (msgbus :sub {:topics [file-id team-id] :chan sub-ch}))
-      (a/<! (send-presence! @wsp :connect)))))
 
 (defmethod handle-message :disconnect
   [wsp _]
-  (a/close! (:sub-ch @wsp))
+  (l/trace :fn "handle-message" :event :disconnect)
-  (send-presence! @wsp :disconnect))
+  (a/go
+    (let [msgbus-fn  (:msgbus @wsp)
+          profile-id (::profile-id @wsp)
+          session-id (::session-id @wsp)
+          profile-ch (::profile-subs-channel @wsp)
+          subs       (::subscriptions @wsp)]
+
+      ;; Close the main profile subscription
+      (a/close! profile-ch)
+      (a/<! (msgbus-fn :cmd :purge :chans [profile-ch]))
+
+      ;; Close all other active subscrption on this websocket context.
+      (doseq [{:keys [channel topic]} (map second subs)]
+        (a/close! channel)
+        (a/<! (msgbus-fn :cmd :pub :topic topic
+                         :message {:type :disconnect
+                                   :profile-id profile-id
+                                   :session-id session-id}))
+        (a/<! (msgbus-fn :cmd :purge :chans [channel]))))))
+
+(defmethod handle-message :subscribe-team
+  [wsp {:keys [team-id] :as params}]
+  (l/trace :fn "handle-message" :event :subscribe-team :team-id team-id)
+
+  (let [msgbus-fn  (:msgbus @wsp)
+        session-id (::session-id @wsp)
+        output-ch  (::ws/output-ch @wsp)
+        subs       (get-in @wsp [::subscriptions team-id])
+        xform      (comp
+                    (remove #(= (:session-id %) session-id))
+                    (map #(assoc % :subs-id team-id)))]
+
+    (a/go
+      (when (not= (:team-id subs) team-id)
+        ;; if it exists we just need to close that
+        (when-let [channel (:channel subs)]
+          (a/close! channel)
+          (a/<! (msgbus-fn :cmd :purge :chans [channel])))
+
+
+        (let [channel (a/chan (a/dropping-buffer 64) xform)]
+          ;; Message forwarding
+          (a/pipe channel output-ch false)
+
+          (let [state {:team-id team-id :channel channel :topic team-id}]
+            (swap! wsp update ::subscriptions assoc team-id state))
+
+          (a/<! (msgbus-fn :cmd :sub :topic team-id :chan channel)))))))
+
+(defmethod handle-message :subscribe-file
+  [wsp {:keys [subs-id file-id] :as params}]
+  (l/trace :fn "handle-message" :event :subscribe-file :subs-id subs-id :file-id file-id)
+  (let [msgbus-fn  (:msgbus @wsp)
+        profile-id (::profile-id @wsp)
+        session-id (::session-id @wsp)
+        output-ch  (::ws/output-ch @wsp)
+
+        xform      (comp
+                    (remove #(= (:session-id %) session-id))
+                    (map #(assoc % :subs-id subs-id)))
+
+        channel    (a/chan (a/dropping-buffer 64) xform)]
+
+    ;; Message forwarding
+    (a/go-loop []
+      (when-let [{:keys [type] :as message} (a/<! channel)]
+        (when (or (= :join-file type)
+                  (= :leave-file type)
+                  (= :disconnect type))
+          (let [message {:type :presence
+                         :file-id file-id
+                         :session-id session-id
+                         :profile-id profile-id}]
+            (a/<! (msgbus-fn :cmd :pub
+                             :topic file-id
+                             :message message))))
+        (a/>! output-ch message)
+        (recur)))
+
+    (let [state {:file-id file-id :channel channel :topic file-id}]
+      (swap! wsp update ::subscriptions assoc subs-id state))
+
+    (a/go
+      ;; Subscribe to file topic
+      (a/<! (msgbus-fn :cmd :sub :topic file-id :chan channel))
+
+      ;; Notifify the rest of participants of the new connection.
+      (let [message {:type :join-file
+                     :file-id file-id
+                     :session-id session-id
+                     :profile-id profile-id}]
+        (a/<! (msgbus-fn :cmd :pub
+                         :topic file-id
+                         :message message))))))
+
+(defmethod handle-message :unsubscribe-file
+  [wsp {:keys [subs-id] :as params}]
+  (l/trace :fn "handle-message" :event :unsubscribe-file :subs-id subs-id)
+  (let [msgbus-fn  (:msgbus @wsp)
+        session-id (::session-id @wsp)
+        profile-id (::profile-id @wsp)]
+    (a/go
+      (when-let [{:keys [file-id channel]} (get-in @wsp [::subscriptions subs-id])]
+        (let [message {:type :leave-file
+                       :file-id file-id
+                       :session-id session-id
+                       :profile-id profile-id}]
+          (a/close! channel)
+          (a/<! (msgbus-fn :cmd :pub :topic file-id :message message))
+          (a/<! (msgbus-fn :cmd :purge :chans [channel])))))))
 
 (defmethod handle-message :keepalive
   [_ _]
+  (l/trace :fn "handle-message" :event :keepalive)
   (a/go :nothing))
 
 (defmethod handle-message :pointer-update
-  [wsp message]
+  [wsp {:keys [subs-id] :as message}]
-  (let [{:keys [profile-id file-id session-id msgbus]} @wsp]
+  (a/go
-    (msgbus :pub {:topic file-id
+    ;; Only allow receive pointer updates when active subscription
-                  :message (assoc message
+    (when-let [{:keys [topic]} (get-in @wsp [::subscriptions subs-id])]
-                                  :profile-id profile-id
+      (l/trace :fn "handle-message" :event :pointer-update :message message)
-                                  :session-id session-id)})))
+      (let [msgbus-fn  (:msgbus @wsp)
+            profile-id (::profile-id @wsp)
+            session-id (::session-id @wsp)
+
+            message    (-> message
+                           (dissoc :subs-id)
+                           (assoc :profile-id profile-id)
+                           (assoc :session-id session-id))]
+
+        (a/<! (msgbus-fn :cmd :pub
+                         :topic topic
+                         :message message))))))
 
 (defmethod handle-message :default
   [_ message]
@@ -75,51 +187,33 @@
            :msg "received unexpected message"
            :message message)))
 
-;; --- IMPL
-
-(defn- send-presence!
-  ([ws] (send-presence! ws :presence))
-  ([{:keys [msgbus session-id profile-id file-id]} type]
-   (msgbus :pub {:topic file-id
-                 :message {:type type
-                           :session-id session-id
-                           :profile-id profile-id}})))
-
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; HTTP HANDLER
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
-(declare retrieve-file)
-
 (s/def ::msgbus fn?)
-(s/def ::file-id ::us/uuid)
 (s/def ::session-id ::us/uuid)
 
 (s/def ::handler-params
-  (s/keys :req-un [::file-id ::session-id]))
+  (s/keys :req-un [::session-id]))
 
 (defmethod ig/pre-init-spec ::handler [_]
   (s/keys :req-un [::msgbus ::db/pool ::mtx/metrics]))
 
 (defmethod ig/init-key ::handler
-  [_ {:keys [pool] :as cfg}]
+  [_ cfg]
   (fn [{:keys [profile-id params] :as req} respond raise]
-    (let [params (us/conform ::handler-params params)
+    (let [{:keys [session-id]} (us/conform ::handler-params params)
-          file   (retrieve-file pool (:file-id params))
+          cfg    (-> cfg
-          cfg    (-> (merge cfg params)
+                     (assoc ::profile-id profile-id)
-                     (assoc :profile-id profile-id)
+                     (assoc ::session-id session-id))]
-                     (assoc :team-id (:team-id file)))]
 
+      (l/trace :hint "http request to websocket" :profile-id profile-id :session-id session-id)
       (cond
         (not profile-id)
         (raise (ex/error :type :authentication
                          :hint "Authentication required."))
 
-        (not file)
-        (raise (ex/error :type :not-found
-                         :code :object-not-found))
-
-
         (not (yws/upgrade-request? req))
         (raise (ex/error :type :validation
                          :code :websocket-request-expected
@@ -129,16 +223,3 @@
         (->> (ws/handler handle-message cfg)
              (yws/upgrade req)
              (respond))))))
-
-(def ^:private
-  sql:retrieve-file
-  "select f.id as id,
-          p.team_id as team_id
-     from file as f
-     join project as p on (p.id = f.project_id)
-    where f.id = ?")
-
-(defn- retrieve-file
-  [conn id]
-  (db/exec-one! conn [sql:retrieve-file id]))
-

backend/src/app/main.clj

@@ -65,6 +65,7 @@
 
    :app.msgbus/msgbus
    {:backend   (cf/get :msgbus-backend :redis)
+    :executor  (ig/ref [::default :app.worker/executor])
     :redis-uri (cf/get :redis-uri)}
 
    :app.tokens/tokens

backend/src/app/msgbus.clj

@@ -7,12 +7,15 @@
 (ns app.msgbus
   "The msgbus abstraction implemented using redis as underlying backend."
   (:require
+   [app.common.data :as d]
    [app.common.exceptions :as ex]
    [app.common.logging :as l]
    [app.common.spec :as us]
+   [app.common.transit :as t]
    [app.config :as cfg]
-   [app.util.blob :as blob]
+   [app.util.async :as aa]
    [app.util.time :as dt]
+   [app.worker :as wrk]
    [clojure.core.async :as a]
    [clojure.spec.alpha :as s]
    [integrant.core :as ig]
@@ -28,120 +31,83 @@
    io.lettuce.core.codec.StringCodec
    io.lettuce.core.pubsub.RedisPubSubListener
    io.lettuce.core.pubsub.StatefulRedisPubSubConnection
-   io.lettuce.core.pubsub.api.async.RedisPubSubAsyncCommands
+   io.lettuce.core.pubsub.api.sync.RedisPubSubCommands
    io.lettuce.core.resource.ClientResources
    io.lettuce.core.resource.DefaultClientResources
    java.time.Duration))
 
+(set! *warn-on-reflection* true)
+
 (def ^:private prefix (cfg/get :tenant))
 
 (defn- prefix-topic
   [topic]
   (str prefix "." topic))
 
-(def xform-prefix (map prefix-topic))
+(def ^:private xform-prefix-topic
-(def xform-topics (map (fn [m] (update m :topics #(into #{} xform-prefix %)))))
+  (map (fn [obj] (update obj :topic prefix-topic))))
-(def xform-topic  (map (fn [m] (update m :topic prefix-topic))))
 
-(s/def ::redis-uri ::us/string)
+(declare ^:private redis-connect)
-(s/def ::buffer-size ::us/integer)
+(declare ^:private redis-disconnect)
-
+(declare ^:private start-io-loop)
-(defmulti init-backend :backend)
+(declare ^:private subscribe)
-(defmulti stop-backend :backend)
+(declare ^:private purge)
-(defmulti init-pub-loop :backend)
+(declare ^:private redis-pub)
-(defmulti init-sub-loop :backend)
+(declare ^:private redis-sub)
-
+(declare ^:private redis-unsub)
-(defmethod ig/pre-init-spec ::msgbus [_]
-  (s/keys :opt-un [::buffer-size ::redis-uri]))
 
 (defmethod ig/prep-key ::msgbus
   [_ cfg]
-  (merge {:buffer-size 128} cfg))
+  (merge {:buffer-size 128
+          :timeout (dt/duration {:seconds 30})}
+         (d/without-nils cfg)))
+
+(s/def ::timeout ::dt/duration)
+(s/def ::redis-uri ::us/string)
+(s/def ::buffer-size ::us/integer)
+
+(defmethod ig/pre-init-spec ::msgbus [_]
+  (s/keys :req-un [::buffer-size ::redis-uri ::timeout ::wrk/executor]))
 
 (defmethod ig/init-key ::msgbus
-  [_ {:keys [backend buffer-size] :as cfg}]
+  [_ {:keys [buffer-size redis-uri] :as cfg}]
-  (l/debug :action "initialize msgbus"
+  (l/info :hint "initialize msgbus"
-           :backend (name backend))
+          :buffer-size buffer-size
-  (let [cfg     (init-backend cfg)
+          :redis-uri redis-uri)
+  (let [cmd-ch (a/chan buffer-size)
+        rcv-ch (a/chan (a/dropping-buffer buffer-size))
+        pub-ch (a/chan (a/dropping-buffer buffer-size) xform-prefix-topic)
+        state  (agent {} :error-handler #(l/error :cause % :hint "unexpected error on agent" ::l/async false))
+        cfg    (-> (redis-connect cfg)
+                   (assoc ::cmd-ch cmd-ch)
+                   (assoc ::rcv-ch rcv-ch)
+                   (assoc ::pub-ch pub-ch)
+                   (assoc ::state state))]
 
-        ;; Channel used for receive publications from the application.
+    (start-io-loop cfg)
-        pub-ch  (-> (a/dropping-buffer buffer-size)
-                    (a/chan xform-topic))
-
-        ;; Channel used for receive subscription requests.
-        sub-ch  (a/chan 1 xform-topics)
-
-        cfg     (-> cfg
-                    (assoc ::pub-ch pub-ch)
-                    (assoc ::sub-ch sub-ch))]
-
-    (init-pub-loop cfg)
-    (init-sub-loop cfg)
 
     (with-meta
-      (fn run
+      (fn [& {:keys [cmd] :as params}]
-        ([command] (run command nil))
+        (a/go
-        ([command params]
+          (case cmd
-         (a/go
+            :pub   (a/>! pub-ch params)
-           (case command
+            :sub   (a/<! (subscribe cfg params))
-             :pub (a/>! pub-ch params)
+            :purge (a/<! (purge cfg params))
-             :sub (a/>! sub-ch params)))))
+            (l/error :hint "unexpeced error on msgbus command processing" :params params))))
       cfg)))
 
 (defmethod ig/halt-key! ::msgbus
   [_ f]
   (let [mdata (meta f)]
-    (stop-backend mdata)
+    (redis-disconnect mdata)
-    (a/close! (::pub-ch mdata))
+    (a/close! (::cmd-ch mdata))
-    (a/close! (::sub-ch mdata))))
+    (a/close! (::rcv-ch mdata))))
 
-;; --- IN-MEMORY BACKEND IMPL
+;; --- IMPL
 
-(defmethod init-backend :memory [cfg] cfg)
+(defn- redis-connect
-(defmethod stop-backend :memory [_])
+  [{:keys [redis-uri timeout] :as cfg}]
-(defmethod init-pub-loop :memory [_])
+  (let [codec     (RedisCodec/of StringCodec/UTF8 ByteArrayCodec/INSTANCE)
-
-(defmethod init-sub-loop :memory
-  [{:keys [::sub-ch ::pub-ch]}]
-  (a/go-loop [state {}]
-    (let [[val port] (a/alts! [pub-ch sub-ch])]
-      (cond
-        (and (= port sub-ch) (some? val))
-        (let [{:keys [topics chan]} val]
-          (recur (reduce #(update %1 %2 (fnil conj #{}) chan) state topics)))
-
-        (and (= port pub-ch) (some? val))
-        (let [topic   (:topic val)
-              message (:message val)
-              state   (loop [state state
-                             chans (get state topic)]
-                        (if-let [c (first chans)]
-                          (if (a/>! c message)
-                            (recur state (rest chans))
-                            (recur (update state topic disj c)
-                                   (rest chans)))
-                          state))]
-          (recur state))
-
-        :else
-        (->> (vals state)
-             (mapcat identity)
-             (run! a/close!))))))
-
-
-;; Add a unique listener to connection
-
-;; --- REDIS BACKEND IMPL
-
-(declare impl-redis-open?)
-(declare impl-redis-pub)
-(declare impl-redis-sub)
-(declare impl-redis-unsub)
-
-
-(defmethod init-backend :redis
-  [{:keys [redis-uri] :as cfg}]
-  (let [codec    (RedisCodec/of StringCodec/UTF8 ByteArrayCodec/INSTANCE)
 
         resources (.. (DefaultClientResources/builder)
                       (ioThreadPoolSize 4)
@@ -151,162 +117,181 @@
         uri       (RedisURI/create redis-uri)
         rclient   (RedisClient/create ^ClientResources resources ^RedisURI uri)
 
-        pub-conn  (.connect ^RedisClient rclient ^RedisCodec codec)
+        pconn     (.connect ^RedisClient rclient ^RedisCodec codec)
-        sub-conn  (.connectPubSub ^RedisClient rclient ^RedisCodec codec)]
+        sconn     (.connectPubSub ^RedisClient rclient ^RedisCodec codec)]
 
-    (.setTimeout ^StatefulRedisConnection pub-conn ^Duration (dt/duration {:seconds 10}))
+    (.setTimeout ^StatefulRedisConnection pconn ^Duration timeout)
-    (.setTimeout ^StatefulRedisPubSubConnection sub-conn ^Duration (dt/duration {:seconds 10}))
+    (.setTimeout ^StatefulRedisPubSubConnection sconn ^Duration timeout)
 
     (-> cfg
         (assoc ::resources resources)
-        (assoc ::pub-conn pub-conn)
+        (assoc ::pconn pconn)
-        (assoc ::sub-conn sub-conn))))
+        (assoc ::sconn sconn))))
 
-(defmethod stop-backend :redis
+(defn- redis-disconnect
-  [{:keys [::pub-conn ::sub-conn ::resources] :as cfg}]
+  [{:keys [::pconn ::sconn ::resources] :as cfg}]
-  (.close ^StatefulRedisConnection pub-conn)
+  (.. ^StatefulConnection pconn close)
-  (.close ^StatefulRedisPubSubConnection sub-conn)
+  (.. ^StatefulConnection sconn close)
   (.shutdown ^ClientResources resources))
 
-(defmethod init-pub-loop :redis
+(defn- conj-subscription
-  [{:keys [::pub-conn ::pub-ch]}]
+  "A low level function that is responsible to create on-demand
-  (let [rac (.async ^StatefulRedisConnection pub-conn)]
+  subscriptions on redis. It reuses the same subscription if it is
-    (a/go-loop []
+  already established. Intended to be executed in agent."
-      (when-let [val (a/<! pub-ch)]
+  [nsubs cfg topic chan]
-        (let [result (a/<! (impl-redis-pub rac val))]
+  (let [nsubs (if (nil? nsubs) #{chan} (conj nsubs chan))]
-          (when (and (impl-redis-open? pub-conn)
+    (when (= 1 (count nsubs))
-                     (ex/exception? result))
+      (l/trace :hint "open subscription" :topic topic ::l/async false)
-            (l/error :cause result
+      (redis-sub cfg topic))
-                     :hint "unexpected error on publish message to redis")))
+    nsubs))
-        (recur)))))
 
-(defmethod init-sub-loop :redis
+(defn- disj-subscription
-  [{:keys [::sub-conn ::sub-ch buffer-size]}]
+  "A low level function responsible on removing subscriptions. The
-  (let [rcv-ch  (a/chan (a/dropping-buffer buffer-size))
+  subscription is trully removed from redis once no single local
-        chans   (agent {} :error-handler #(l/error :cause % :hint "unexpected error on agent"))
+  subscription is look for it. Intended to be executed in agent."
-        rac     (.async ^StatefulRedisPubSubConnection sub-conn)]
+  [nsubs cfg topic chan]
+  (let [nsubs (disj nsubs chan)]
+    (when (empty? nsubs)
+      (l/trace :hint "close subscription" :topic topic ::l/async false)
+      (redis-unsub cfg topic))
+    nsubs))
 
-    ;; Add a unique listener to connection
+(defn- subscribe-to-topics
-    (.addListener sub-conn
+  "Function responsible to attach local subscription to the
-                  (reify RedisPubSubListener
+  state. Intended to be used in agent."
-                    (message [_ _pattern _topic _message])
+  [state cfg topics chan done-ch]
-                    (message [_ topic message]
+  (l/trace :hint "subscribe-to-topics" :topics topics ::l/async false)
-                      ;; There are no back pressure, so we use a slidding
+  (aa/with-closing done-ch
-                      ;; buffer for cases when the pubsub broker sends
+    (let [state (update state :chans assoc chan topics)]
-                      ;; more messages that we can process.
+      (reduce (fn [state topic]
-                      (let [val {:topic topic :message (blob/decode message)}]
+                (update-in state [:topics topic] conj-subscription cfg topic chan))
-                        (when-not (a/offer! rcv-ch val)
+              state
-                          (l/warn :msg "dropping message on subscription loop"))))
+              topics))))
-                    (psubscribed [_ _pattern _count])
-                    (punsubscribed [_ _pattern _count])
-                    (subscribed [_ _topic _count])
-                    (unsubscribed [_ _topic _count])))
 
-    (letfn [(subscribe-to-single-topic [nsubs topic chan]
+(defn- unsubscribe-single-channel
-              (let [nsubs (if (nil? nsubs) #{chan} (conj nsubs chan))]
+  "Auxiliar function responsible on removing a single local
-                (when (= 1 (count nsubs))
+  subscription from the state."
-                  (let [result (a/<!! (impl-redis-sub rac topic))]
+  [state cfg chan]
-                    (l/trace :action "open subscription"
+  (let [topics (get-in state [:chans chan])
-                             :topic topic)
+        state  (update state :chans dissoc chan)]
-                    (when (ex/exception? result)
+    (reduce (fn [state topic]
-                      (l/error :cause result
+              (update-in state [:topics topic] disj-subscription cfg topic chan))
-                               :hint "unexpected exception on subscribing"
+            state
-                               :topic topic))))
+            topics)))
-                nsubs))
 
-            (subscribe-to-topics [state topics chan]
+(defn- unsubscribe-channels
-              (let [state  (update state :chans assoc chan topics)]
+  "Function responsible from detach from state a seq of channels,
-                (reduce (fn [state topic]
+  useful when client disconnects or in-bulk unsubscribe
-                          (update-in state [:topics topic] subscribe-to-single-topic topic chan))
+  operations. Intended to be executed in agent."
-                        state
+  [state cfg channels done-ch]
-                        topics)))
+  (l/trace :hint "unsubscribe-channels" :chans (count channels) ::l/async false)
+  (aa/with-closing done-ch
+    (reduce #(unsubscribe-single-channel %1 cfg %2) state channels)))
 
-            (unsubscribe-from-single-topic [nsubs topic chan]
+(defn- subscribe
-              (let [nsubs (disj nsubs chan)]
+  [{:keys [::state executor] :as cfg} {:keys [topic topics chan]}]
-                (when (empty? nsubs)
+  (let [done-ch (a/chan)
-                  (let [result (a/<!! (impl-redis-unsub rac topic))]
+        topics  (into [] (map prefix-topic) (if topic [topic] topics))]
-                    (l/trace :action "close subscription"
+    (l/trace :hint "subscribe" :topics topics)
-                             :topic topic)
+    (send-via executor state subscribe-to-topics cfg topics chan done-ch)
-                    (when (and (impl-redis-open? sub-conn)
+    done-ch))
-                               (ex/exception? result))
-                      (l/error :cause result
-                               :hint "unexpected exception on unsubscribing"
-                               :topic topic))))
-                nsubs))
 
-            (unsubscribe-channels [state pending]
+(defn- purge
-              (reduce (fn [state ch]
+  [{:keys [::state executor] :as cfg} {:keys [chans]}]
-                        (let [topics (get-in state [:chans ch])
+  (l/trace :hint "purge" :chans (count chans))
-                              state  (update state :chans dissoc ch)]
+  (let [done-ch (a/chan)]
-                          (reduce (fn [state topic]
+    (send-via executor state unsubscribe-channels cfg chans done-ch)
-                                    (update-in state [:topics topic] unsubscribe-from-single-topic topic ch))
+    done-ch))
-                                  state
-                                  topics)))
-                      state
-                      pending))]
 
-      ;; Asynchronous subscription loop;
+(defn- create-listener
-      (a/go-loop []
+  [rcv-ch]
-        (if-let [{:keys [topics chan]} (a/<! sub-ch)]
+  (reify RedisPubSubListener
-          (do
+    (message [_ _pattern _topic _message])
-            (send-off chans subscribe-to-topics topics chan)
+    (message [_ topic message]
-            (recur))
+      ;; There are no back pressure, so we use a slidding
-          (a/close! rcv-ch)))
+      ;; buffer for cases when the pubsub broker sends
+      ;; more messages that we can process.
+      (let [val {:topic topic :message (t/decode message)}]
+        (when-not (a/offer! rcv-ch val)
+          (l/warn :msg "dropping message on subscription loop"))))
+    (psubscribed [_ _pattern _count])
+    (punsubscribed [_ _pattern _count])
+    (subscribed [_ _topic _count])
+    (unsubscribed [_ _topic _count])))
 
-      ;; Asynchronous message processing loop;x
+(defn start-io-loop
-      (a/go-loop []
+  [{:keys [::sconn ::rcv-ch ::pub-ch ::state executor] :as cfg}]
-        (if-let [{:keys [topic message]} (a/<! rcv-ch)]
-          ;; This means we receive data from redis and we need to
-          ;; forward it to the underlying subscriptions.
-          (let [pending (loop [chans   (seq (get-in @chans [:topics topic]))
-                               pending #{}]
-                          (if-let [ch (first chans)]
-                            (if (a/>! ch message)
-                              (recur (rest chans) pending)
-                              (recur (rest chans) (conj pending ch)))
-                            pending))]
-            (some->> (seq pending)
-                     (send-off chans unsubscribe-channels))
 
-            (recur))
+  ;; Add a single listener to the pubsub connection
+  (.addListener ^StatefulRedisPubSubConnection sconn
+                ^RedisPubSubListener (create-listener rcv-ch))
 
-          ;; Stop condition; close all underlying subscriptions and
+  (letfn [(send-to-topic [topic message]
-          ;; exit. The close operation is performed asynchronously.
+            (a/go-loop [chans  (seq (get-in @state [:topics topic]))
-          (send-off chans (fn [state]
+                        closed #{}]
-                            (->> (vals state)
+              (if-let [ch (first chans)]
-                                 (mapcat identity)
+                (if (a/>! ch message)
-                                 (filter some?)
+                  (recur (rest chans) closed)
-                                 (run! a/close!)))))))))
+                  (recur (rest chans) (conj closed ch)))
+                (seq closed))))
 
+          (process-incoming [{:keys [topic message]}]
+            (a/go
+              (when-let [closed (a/<! (send-to-topic topic message))]
+                (send-via executor state unsubscribe-channels cfg closed nil))))
+          ]
 
-(defn- impl-redis-open?
+  (a/go-loop []
-  [^StatefulConnection conn]
+    (let [[val port] (a/alts! [pub-ch rcv-ch])]
-  (.isOpen conn))
+      (cond
+        (nil? val)
+        (do
+          (l/trace :hint "stoping io-loop, nil received")
+          (send-via executor state (fn [state]
+                                     (->> (vals state)
+                                          (mapcat identity)
+                                          (filter some?)
+                                          (run! a/close!))
+                                     nil)))
 
-(defn- impl-redis-pub
+        (= port rcv-ch)
-  [^RedisAsyncCommands rac {:keys [topic message]}]
+        (do
-  (let [message (blob/encode message)
+          (a/<! (process-incoming val))
-        res     (a/chan 1)]
+          (recur))
-    (-> (.publish rac ^String topic ^bytes message)
+
-        (p/finally (fn [_ e]
+        (= port pub-ch)
-                     (when e (a/>!! res e))
+        (let [result (a/<! (redis-pub cfg val))]
+          (when (ex/exception? result)
+            (l/error :hint "unexpected error on publishing" :message val
+                     :cause result))
+          (recur)))))))
+
+(defn- redis-pub
+  "Publish a message to the redis server. Asynchronous operation,
+  intended to be used in core.async go blocks."
+  [{:keys [::pconn] :as cfg} {:keys [topic message]}]
+  (let [message (t/encode message)
+        res     (a/chan 1)
+        pcomm   (.async ^StatefulRedisConnection pconn)]
+    (-> (.publish ^RedisAsyncCommands pcomm ^String topic ^bytes message)
+        (p/finally (fn [_ cause]
+                     (when (and cause (.isOpen ^StatefulConnection pconn))
+                       (a/offer! res cause))
                      (a/close! res))))
     res))
 
-(defn impl-redis-sub
+(defn redis-sub
-  [^RedisPubSubAsyncCommands rac topic]
+  "Create redis subscription. Blocking operation, intended to be used
-  (let [res (a/chan 1)]
+  inside an agent."
-    (-> (.subscribe rac (into-array String [topic]))
+  [{:keys [::sconn] :as cfg} topic]
-        (p/finally (fn [_ e]
+  (let [topic (into-array String [topic])
-                     (when e (a/>!! res e))
+        scomm (.sync ^StatefulRedisPubSubConnection sconn)]
-                     (a/close! res))))
+    (.subscribe ^RedisPubSubCommands scomm topic)))
-    res))
 
-(defn impl-redis-unsub
+(defn redis-unsub
-  [rac topic]
+  "Removes redis subscription. Blocking operation, intended to be used
-  (let [res (a/chan 1)]
+  inside an agent."
-    (-> (.unsubscribe rac (into-array String [topic]))
+  [{:keys [::sconn] :as cfg} topic]
-        (p/finally (fn [_ e]
+  (let [topic (into-array String [topic])
-                     (when e (a/>!! res e))
+        scomm (.sync ^StatefulRedisPubSubConnection sconn)]
-                     (a/close! res))))
+    (.unsubscribe ^RedisPubSubCommands scomm topic)))
-    res))

backend/src/app/rpc/mutations/files.clj

@@ -386,31 +386,33 @@
                                (assoc :changes []))))))))
 
 (defn- send-notifications
-  [{:keys [msgbus conn] :as cfg} {:keys [file changes session-id] :as params}]
+  [{:keys [conn] :as cfg} {:keys [file changes session-id] :as params}]
-  (let [lchanges (filter library-change? changes)]
+  (let [lchanges  (filter library-change? changes)
+        msgbus-fn (:msgbus cfg)]
+
 
     ;; Asynchronously publish message to the msgbus
-    (msgbus :pub {:topic (:id file)
+    (msgbus-fn :cmd :pub
-                  :message
+               :topic (:id file)
-                  {:type :file-change
+               :message {:type :file-change
-                   :profile-id (:profile-id params)
+                         :profile-id (:profile-id params)
-                   :file-id (:id file)
+                         :file-id (:id file)
-                   :session-id (:session-id params)
+                         :session-id (:session-id params)
-                   :revn (:revn file)
+                         :revn (:revn file)
-                   :changes changes}})
+                         :changes changes})
 
     (when (and (:is-shared file) (seq lchanges))
       (let [team-id (retrieve-team-id conn (:project-id file))]
         ;; Asynchronously publish message to the msgbus
-        (msgbus :pub {:topic team-id
+        (msgbus-fn :cmd :pub
-                      :message
+                   :topic team-id
-                      {:type :library-change
+                   :message {:type :library-change
-                       :profile-id (:profile-id params)
+                             :profile-id (:profile-id params)
-                       :file-id (:id file)
+                             :file-id (:id file)
-                       :session-id session-id
+                             :session-id session-id
-                       :revn (:revn file)
+                             :revn (:revn file)
-                       :modified-at (dt/now)
+                             :modified-at (dt/now)
-                       :changes lchanges}})))))
+                             :changes lchanges})))))
 
 (defn- retrieve-team-id
   [conn project-id]

backend/src/app/util/async.clj

@@ -38,6 +38,13 @@
        (throw r#)
        r#)))
 
+(defmacro with-closing
+  [ch & body]
+  `(try
+     ~@body
+     (finally
+       (some-> ~ch a/close!))))
+
 (defn thread-call
   [^Executor executor f]
   (let [c (a/chan 1)]

backend/src/app/util/websocket.clj

@@ -54,16 +54,22 @@
            pong-ch    (a/chan (a/sliding-buffer 6))
            close-ch   (a/chan)
 
-           options    (-> options
+           options    (atom
-                          (assoc ::input-ch input-ch)
+                       (-> options
-                          (assoc ::output-ch output-ch)
+                           (assoc ::input-ch input-ch)
-                          (assoc ::close-ch close-ch)
+                           (assoc ::output-ch output-ch)
-                          (assoc ::channel channel)
+                           (assoc ::close-ch close-ch)
-                          (dissoc ::metrics))
+                           (assoc ::channel channel)
+                           (dissoc ::metrics)))
 
            terminated (atom false)
            created-at (dt/now)
 
+           on-open
+           (fn [channel]
+             (mtx/run! metrics {:id :websocket-active-connections :inc 1})
+             (yws/idle-timeout! channel (dt/duration idle-timeout)))
+
            on-terminate
            (fn [& _args]
              (when (compare-and-set! terminated false true)
@@ -79,7 +85,8 @@
            (fn [_ error]
              (on-terminate)
              ;; TODO: properly log timeout exceptions
-             (when-not (instance? java.nio.channels.ClosedChannelException error)
+             (when-not (or (instance? java.nio.channels.ClosedChannelException error)
+                           (instance? java.net.SocketException error))
                (l/error :hint (ex-message error) :cause error)))
 
            on-message
@@ -98,31 +105,28 @@
            (fn [_ buffers]
              (a/>!! pong-ch (yu/copy-many buffers)))]
 
-       (mtx/run! metrics {:id :websocket-active-connections :inc 1})
+       ;; launch heartbeat process
+       (-> @options
+           (assoc ::pong-ch pong-ch)
+           (assoc ::on-close on-terminate)
+           (process-heartbeat))
 
-       (let [wsp (atom options)]
+       ;; Forward all messages from output-ch to the websocket
-         ;; Handle heartbeat
+       ;; connection
-         (yws/idle-timeout! channel (dt/duration idle-timeout))
+       (a/go-loop []
-         (-> @wsp
+         (when-let [val (a/<! output-ch)]
-             (assoc ::pong-ch pong-ch)
+           (mtx/run! metrics {:id :websocket-messages-total :labels ["send"] :inc 1})
-             (assoc ::on-close on-terminate)
+           (a/<! (ws-send! channel (t/encode-str val)))
-             (process-heartbeat))
+           (recur)))
 
-         ;; Forward all messages from output-ch to the websocket
+       ;; React on messages received from the client
-         ;; connection
+       (process-input options handle-message)
-         (a/go-loop []
-           (when-let [val (a/<! output-ch)]
-             (mtx/run! metrics {:id :websocket-messages-total :labels ["send"] :inc 1})
-             (a/<! (ws-send! channel (t/encode-str val)))
-             (recur)))
 
-         ;; React on messages received from the client
+       {:on-open on-open
-         (process-input wsp handle-message)
+        :on-error on-error
-
+        :on-close on-terminate
-         {:on-error on-error
+        :on-text on-message
-          :on-close on-terminate
+        :on-pong on-pong}))))
-          :on-text on-message
-          :on-pong on-pong})))))
 
 (defn- ws-send!
   [channel s]
@@ -160,14 +164,21 @@
     (.rewind buffer)
     (.getLong buffer)))
 
+(defn- wrap-handler
+  [handler]
+  (fn [wsp message]
+    (locking wsp
+      (handler wsp message))))
+
 (defn- process-input
   [wsp handler]
-  (let [{:keys [::input-ch ::output-ch ::close-ch]} @wsp]
+  (let [{:keys [::input-ch ::output-ch ::close-ch]} @wsp
+        handler (wrap-handler handler)]
     (a/go
       (a/<! (handler wsp {:type :connect}))
       (a/<! (a/go-loop []
-              (when-let [request (a/<! input-ch)]
+              (when-let [message (a/<! input-ch)]
-                (let [[val port] (a/alts! [(handler wsp request) close-ch])]
+                (let [[val port] (a/alts! [(handler wsp message) close-ch])]
                   (when-not (= port close-ch)
                     (cond
                       (ex/ex-info? val)
@@ -177,8 +188,7 @@
                       (a/>! output-ch {:type :error :error {:message (ex-message val)}})
 
                       (map? val)
-                      (a/>! output-ch (cond-> val (:request-id request) (assoc :request-id (:request-id request)))))
+                      (a/>! output-ch (cond-> val (:request-id message) (assoc :request-id (:request-id message)))))
-
                     (recur))))))
       (a/<! (handler wsp {:type :disconnect})))))
 

common/deps.edn

@@ -6,11 +6,11 @@
   org.clojure/clojurescript {:mvn/version "1.11.4"}
 
   ;; Logging
-  org.apache.logging.log4j/log4j-api {:mvn/version "2.17.1"}
+  org.apache.logging.log4j/log4j-api {:mvn/version "2.17.2"}
-  org.apache.logging.log4j/log4j-core {:mvn/version "2.17.1"}
+  org.apache.logging.log4j/log4j-core {:mvn/version "2.17.2"}
-  org.apache.logging.log4j/log4j-web {:mvn/version "2.17.1"}
+  org.apache.logging.log4j/log4j-web {:mvn/version "2.17.2"}
-  org.apache.logging.log4j/log4j-jul {:mvn/version "2.17.1"}
+  org.apache.logging.log4j/log4j-jul {:mvn/version "2.17.2"}
-  org.apache.logging.log4j/log4j-slf4j18-impl {:mvn/version "2.17.1"}
+  org.apache.logging.log4j/log4j-slf4j18-impl {:mvn/version "2.17.2"}
   org.slf4j/slf4j-api {:mvn/version "2.0.0-alpha1"}
 
   selmer/selmer {:mvn/version "1.12.50"}

common/src/app/common/data/macros.cljc

@@ -132,7 +132,7 @@
 (defn- interpolate
   [s params]
   (loop [items  (->> (re-seq #"([^\%]+)*(\%(\d+)?)?" s)
-                     (remove (fn [[_ seg]] (nil? seg))))
+                     (remove (fn [[full seg]] (and (nil? seg) (not full)))))
          result []
          index  0]
     (if-let [[_ segment var? sidx] (first items)]
@@ -156,7 +156,8 @@
         (recur (rest items)
                (conj result segment)
                (inc index)))
-      result)))
+
+      (remove nil? result))))
 
 (defmacro fmt
   "String interpolation helper. Can only be used with strings known at

common/src/app/common/logging.cljc

@@ -181,16 +181,18 @@
              ~level-sym  (get-level ~level)]
          (when (enabled? ~logger-sym ~level-sym)
            ~(if async
-              `(send-off logging-agent
+              `(do
-                         (fn [_#]
+                 (send-off logging-agent
-                           (with-context (merge {:id (uuid/next)}
+                           (fn [_#]
-                                                (get-error-context ~cause)
+                             (with-context (into {:id (uuid/next)}
-                                                ~context)
+                                                 (get-error-context ~cause)
-                            (->> (or ~raw (build-map-message ~props))
+                                                 ~context)
-                                 (write-log! ~logger-sym ~level-sym ~cause)))))
+                               (->> (or ~raw (build-map-message ~props))
-
+                                    (write-log! ~logger-sym ~level-sym ~cause)))))
+                 nil)
               `(let [message# (or ~raw (build-map-message ~props))]
-                 (write-log! ~logger-sym ~level-sym ~cause message#))))))))
+                 (write-log! ~logger-sym ~level-sym ~cause message#)
+                 nil)))))))
 
 (defmacro info
   [& params]

common/src/app/common/spec.cljc

@@ -16,6 +16,7 @@
    ;; because of some strange interaction with cljs.spec.alpha and
    ;; modules splitting.
    [app.common.exceptions :as ex]
+   [app.common.uri :as u]
    [app.common.uuid :as uuid]
    [cuerdas.core :as str]
    [expound.alpha :as expound]))
@@ -96,6 +97,7 @@
     :else
     ::s/invalid))
 
+
 ;; --- Default Specs
 
 (s/def ::keyword (s/conformer keyword-conformer name))
@@ -192,6 +194,15 @@
    (fn [v]
      (str/join " " v))))
 
+(s/def ::uri
+  (s/conformer
+   (fn [s]
+     (cond
+       (u/uri? s) s
+       (string? s) (u/uri s)
+       :else ::s/invalid))
+   str))
+
 ;; --- SPEC: set-of-str
 
 (s/def ::set-of-str

common/src/app/common/uri.cljc

@@ -9,7 +9,9 @@
   (:require
    [app.common.data.macros :as dm]
    [lambdaisland.uri :as u]
-   [lambdaisland.uri.normalize :as un]))
+   [lambdaisland.uri.normalize :as un])
+  #?(:clj
+     (:import lambdaisland.uri.URI)))
 
 (dm/export u/uri)
 (dm/export u/join)
@@ -25,6 +27,11 @@
   [v]
   (if (keyword? v) (name v) v))
 
+(defn get-domain
+  [{:keys [host port] :as uri}]
+  (cond-> host
+    port (str ":" port)))
+
 (defn map->query-string
   ([params] (map->query-string params nil))
   ([params {:keys [value-fn key-fn]
@@ -35,3 +42,16 @@
                   (remove #(nil? (second %)))
                   (map (fn [[k v]] [(key-fn k) (value-fn v)]))))
         (u/map->query-string))))
+
+#?(:clj
+   (defmethod print-method lambdaisland.uri.URI [^URI this ^java.io.Writer writer]
+     (.write writer "#")
+     (.write writer (str u/edn-tag))
+     (.write writer " ")
+     (.write writer (pr-str (.toString this))))
+
+   :cljs
+   (extend-type u/URI
+     IPrintWithWriter
+     (-pr-writer [this writer _opts]
+       (write-all writer "#" (str u/edn-tag) " " (pr-str (.toString this))))))