enhance(rtc): stop rtc when user logout

src/main/frontend/flows.cljs

@@ -1,12 +1,24 @@
 (ns frontend.flows
   "This ns contains some event flows."
-  (:require [missionary.core :as m]))
+  (:require [malli.core :as ma]
+            [missionary.core :as m]))
 
 ;; Some Input Atoms
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
 (def *current-repo (atom nil))
 
+(def ^:private current-login-user-schema
+  [:or
+   [:= :logout]
+   [:map
+    [:email :string]
+    [:sub :string]
+    [:cognito:username :string]]])
+
+(def ^:private current-login-user-validator (ma/validator current-login-user-schema))
+(def *current-login-user (atom nil :validator current-login-user-validator))
+
 ;; Public Flows
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
@@ -15,3 +27,8 @@
   (m/eduction
    (dedupe)
    (m/watch *current-repo)))
+
+(def current-login-user-flow
+  (m/eduction
+   (dedupe)
+   (m/watch *current-login-user)))

src/main/frontend/handler/db_based/rtc.cljs

@@ -5,11 +5,13 @@
             [frontend.common.missionary :as c.m]
             [frontend.config :as config]
             [frontend.db :as db]
+            [frontend.flows :as flows]
             [frontend.handler.db-based.rtc-flows :as rtc-flows]
             [frontend.handler.notification :as notification]
             [frontend.handler.user :as user-handler]
             [frontend.state :as state]
             [frontend.util :as util]
+            [lambdaisland.glogi :as log]
             [logseq.common.util :as common-util]
             [logseq.db :as ldb]
             [logseq.db.common.sqlite :as sqlite-common-db]
@@ -198,21 +200,30 @@
           (prn :trying-to-restart-rtc graph-uuid (t/now))
           (c.m/<? (<rtc-start! (state/get-current-repo) :stop-before-start? false)))))))
 
-  (when-not config/publishing?
-    (c.m/run-background-task
-     ::notify-client-need-upgrade-when-larger-remote-schema-version-exists
-     (m/reduce
-      (constantly nil)
-      (m/ap
-        (let [{:keys [repo graph-uuid remote-schema-version sub-type]}
-              (m/?>
-               (m/eduction
-                (filter #(keyword-identical? :rtc.log/higher-remote-schema-version-exists (:type %)))
-                rtc-flows/rtc-log-flow))]
-          (case sub-type
-            :download
-            (notification-download-higher-schema-graph! repo graph-uuid remote-schema-version)
+  (c.m/run-background-task
+   ::notify-client-need-upgrade-when-larger-remote-schema-version-exists
+   (m/reduce
+    (constantly nil)
+    (m/ap
+      (let [{:keys [repo graph-uuid remote-schema-version sub-type]}
+            (m/?>
+             (m/eduction
+              (filter #(keyword-identical? :rtc.log/higher-remote-schema-version-exists (:type %)))
+              rtc-flows/rtc-log-flow))]
+        (case sub-type
+          :download
+          (notification-download-higher-schema-graph! repo graph-uuid remote-schema-version)
           ;; else
-            (notification/show!
-             "The server has a graph with a higher schema version, the client may need to upgrade."
-             :warning))))))))
+          (notification/show!
+           "The server has a graph with a higher schema version, the client may need to upgrade."
+           :warning))))))
+
+  (c.m/run-background-task
+   ::stop-rtc-when-user-logout
+   (m/reduce
+    (constantly nil)
+    (m/ap
+      (let [login-user (m/?> flows/current-login-user-flow)]
+        (when (= :logout login-user)
+          (log/info :try-to-stop-rtc-when-user-logout nil)
+          (c.m/<? (<rtc-stop!))))))))

src/main/frontend/handler/db_based/rtc_flows.cljs

@@ -1,6 +1,7 @@
 (ns frontend.handler.db-based.rtc-flows
   "Flows related to RTC"
   (:require [frontend.common.missionary :as c.m]
+            [frontend.flows :as flows]
             [frontend.state :as state]
             [logseq.common.util :as common-util]
             [missionary.core :as m]))
@@ -51,17 +52,22 @@
 (def rtc-try-restart-flow
   "emit an event when it's time to restart rtc loop.
 conditions:
-1. no rtc loop running now
-2. last rtc stop-reason is websocket message timeout
-3. current js/navigator.onLine=true
-5. throttle 5000ms"
+- user logged in
+- no rtc loop running now
+- last rtc stop-reason is websocket message timeout
+- current js/navigator.onLine=true
+- throttle 5000ms"
   (->> (m/latest
-        (fn [rtc-state _] rtc-state)
-        (c.m/continue-flow rtc-state-flow) (c.m/continue-flow network-online-change-flow))
+        (fn [rtc-state _ login-user]
+          (assoc rtc-state :login-user login-user))
+        (c.m/continue-flow rtc-state-flow)
+        (c.m/continue-flow network-online-change-flow)
+        flows/current-login-user-flow)
        (m/eduction
         (keep (fn [m]
-                (let [{:keys [rtc-lock last-stop-exception-ex-data graph-uuid]} m]
-                  (when (and (some? graph-uuid)
+                (let [{:keys [rtc-lock last-stop-exception-ex-data graph-uuid login-user]} m]
+                  (when (and (some? (:email login-user))
+                             (some? graph-uuid)
                              (not rtc-lock) ; no rtc loop now
                              (= :rtc.exception/ws-timeout (:type last-stop-exception-ex-data))
                              (true? js/navigator.onLine))

src/main/frontend/handler/user.cljs

@@ -9,6 +9,7 @@
             [frontend.common.missionary :as c.m]
             [frontend.config :as config]
             [frontend.debug :as debug]
+            [frontend.flows :as flows]
             [frontend.handler.config :as config-handler]
             [frontend.handler.notification :as notification]
             [frontend.state :as state]
@@ -200,6 +201,7 @@
    (:jwtToken (:idToken session))
    (:jwtToken (:accessToken session))
    (:token (:refreshToken session)))
+  (reset! flows/*current-login-user (parse-jwt (state/get-auth-id-token)))
   (state/pub-event! [:user/fetch-info-and-graphs]))
 
 (defn ^:export login-with-username-password-e2e
@@ -231,7 +233,8 @@
 (defn logout []
   (clear-tokens)
   (state/clear-user-info!)
-  (state/pub-event! [:user/logout]))
+  (state/pub-event! [:user/logout])
+  (reset! flows/*current-login-user :logout))
 
 (defn upgrade []
   (let [base-upgrade-url "https://logseqdemo.lemonsqueezy.com/checkout/buy/13e194b5-c927-41a8-af58-ed1a36d6000d"