feat(rtc): connect to current user system

src/main/frontend/config.cljs

@@ -64,8 +64,8 @@
 
 (goog-define ENABLE-RTC-SYNC-PRODUCTION false)
 (if ENABLE-RTC-SYNC-PRODUCTION
-  (def RTC-WS-URL "wss://ws.logseq.com/rtc-sync?useruuid=%s")
-  (def RTC-WS-URL "wss://ws-dev.logseq.com/rtc-sync?useruuid=%s"))
+  (def RTC-WS-URL "wss://ws.logseq.com/rtc-sync?token=%s")
+  (def RTC-WS-URL "wss://ws-dev.logseq.com/rtc-sync?token=%s"))
 ;; Feature flags
 ;; =============
 

src/main/frontend/db/rtc/core.cljs

@@ -7,17 +7,19 @@
             [cljs.core.async :as async :refer [<! chan go go-loop]]
             [cljs.core.async.interop :refer [p->c]]
             [clojure.set :as set]
+            [cognitect.transit :as transit]
             [frontend.db :as db]
+            [frontend.db.rtc.const :as rtc-const]
             [frontend.db.rtc.op :as op]
             [frontend.db.rtc.ws :as ws]
-            [frontend.db.rtc.const :as rtc-const]
             [frontend.handler.page :as page-handler]
+            [frontend.handler.user :as user]
             [frontend.modules.outliner.core :as outliner-core]
             [frontend.modules.outliner.transaction :as outliner-tx]
+            [frontend.state :as state]
             [frontend.util :as util]
             [malli.core :as m]
-            [malli.util :as mu]
-            [cognitect.transit :as transit]))
+            [malli.util :as mu]))
 
 
 ;;                     +-------------+
@@ -42,7 +44,6 @@
 
 (def state-schema
   "
-  | :user-uuid             | string                                              |
   | :*graph-uuid           | atom of graph-uuid syncing now                      |
   | :*repo                 | atom of repo name syncing now                       |
   | :data-from-ws-chan     | channel for receive messages from server websocket  |
@@ -53,7 +54,6 @@
   | :*rtc-state            | atom of state of current rtc progress               |
 "
   [:map
-   [:user-uuid :string]
    [:*graph-uuid :any]
    [:*repo :any]
    [:data-from-ws-chan :any]
@@ -70,6 +70,9 @@
   [:enum :open :closed])
 (def rtc-state-validator (m/validator rtc-state-schema))
 
+(defn- guard-ex
+  [x]
+  (when (instance? ExceptionInfo x) x))
 
 (def transit-w (transit/writer :json))
 (def transit-r (transit/reader :json))
@@ -525,8 +528,12 @@
                 (do (<push-data-from-ws-handler repo push-data-from-ws)
                     (recur))
                 client-op-update
-                (do (<! (<client-op-update-handler state))
-                    (recur))
+                (let [maybe-exp (<! (user/<wrap-ensure-id&access-token
+                                     (<! (<client-op-update-handler state))))]
+                  (if (= :expired-token (:anom (ex-data maybe-exp)))
+                    (prn ::<loop-for-rtc "quiting loop" maybe-exp)
+                    (recur)))
+
                 stop
                 (do (ws/stop @(:*ws state))
                     (reset! (:*rtc-state state) :closed))
@@ -535,31 +542,24 @@
       (async/unsub data-from-ws-pub "push-updates" push-data-from-ws-ch))))
 
 (defn- init-state
-  [ws data-from-ws-chan user-uuid]
-  (m/parse state-schema
-           {:*rtc-state (atom :closed :validator rtc-state-validator)
-            :user-uuid user-uuid
-            :*graph-uuid (atom nil)
-            :*repo (atom nil)
-            :data-from-ws-chan data-from-ws-chan
-            :data-from-ws-pub (async/pub data-from-ws-chan :req-id)
-            :*stop-rtc-loop-chan (atom nil)
-            :client-op-update-chan (chan 1)
-            :*ws (atom ws)}))
+  [ws data-from-ws-chan]
+  {:post [(m/validate state-schema %)]}
+  {:*rtc-state (atom :closed :validator rtc-state-validator)
+   :*graph-uuid (atom nil)
+   :*repo (atom nil)
+   :data-from-ws-chan data-from-ws-chan
+   :data-from-ws-pub (async/pub data-from-ws-chan :req-id)
+   :*stop-rtc-loop-chan (atom nil)
+   :client-op-update-chan (chan 1)
+   :*ws (atom ws)})
 
 (defn <init-state
   []
   (go
     (let [data-from-ws-chan (chan (async/sliding-buffer 100))
-          ws-opened-ch (chan)
-          user-uuid "f92bb5b3-0f72-4a74-9ad8-1793e655c309"
-          ws (ws/ws-listen user-uuid data-from-ws-chan ws-opened-ch)]
-      (<! ws-opened-ch)
-      (init-state ws data-from-ws-chan user-uuid))))
-
-
-(comment
-  (go
-    (def global-state (<! (<init-state))))
-  (reset! (:*graph-uuid global-state) debug-graph-uuid)
-  (reset! (:*repo global-state) (state/get-current-repo)))
+          ws-opened-ch (chan)]
+      (<! (user/<wrap-ensure-id&access-token
+           (let [token (state/get-auth-id-token)
+                 ws (ws/ws-listen token data-from-ws-chan ws-opened-ch)]
+             (<! ws-opened-ch)
+             (init-state ws data-from-ws-chan)))))))

src/main/frontend/db/rtc/debug_ui.cljs

@@ -26,8 +26,10 @@
        (<! (<start-rtc state)))))
   ([state]
    (go
-     (let [repo (state/get-current-repo)]
-       (<! (<start-rtc state repo)))))
+     (if (= :expired-token (:anom (ex-data state)))
+       (prn ::<start-rtc state)
+       (let [repo (state/get-current-repo)]
+         (<! (<start-rtc state repo))))))
   ([state repo]
    (go
      (if-let [graph-uuid (<! (p->c (op/<get-graph-uuid repo)))]

src/main/frontend/db/rtc/full_upload_download_graph.cljs

@@ -31,7 +31,9 @@
                       (when (and (contains? #{:block/parent :block/left} (:a datom))
                                  (not (pos-int? (:v datom))))
                         (throw (ex-info "invalid block data" {:datom datom})))
-                      (assoc r (:a datom) (:v datom)))
+                      (if (contains? db-schema/card-many-attributes (:a datom))
+                        (update r (:a datom) conj (:v datom))
+                        (assoc r (:a datom) (:v datom))))
                     {:db/id (:e (first datoms))}
                     datoms)))))))
 

src/main/frontend/db/rtc/ws.cljs

@@ -1,22 +1,23 @@
 (ns frontend.db.rtc.ws
-  "TODO"                                ; @zhiyuan
+  "Websocket related util-fns"
   (:require-macros
    [frontend.db.rtc.macro :refer [with-sub-data-from-ws get-req-id get-result-ch]])
   (:require [frontend.config :as config]
             [frontend.util :as util]
             [frontend.db.rtc.const :as rtc-const]
-            [cljs.core.async :as async :refer [<! chan go offer!]]))
+            [cljs.core.async :as async :refer [<! chan go offer!]]
+            [frontend.state :as state]))
 
 
 (def ws-addr config/RTC-WS-URL)
 
 (defn ws-listen
-  [user-uuid data-from-ws-chan ws-opened-ch]
-  (let [ws (js/WebSocket. (util/format ws-addr user-uuid))]
+  [token data-from-ws-chan ws-opened-ch]
+  (let [ws (js/WebSocket. (util/format ws-addr token))]
     (set! (.-onopen ws) (fn [_e] (async/close! ws-opened-ch)))
     (set! (.-onmessage ws) (fn [e]
-                                     (let [data (js->clj (js/JSON.parse (.-data e)) :keywordize-keys true)]
-                                       (offer! data-from-ws-chan data))))
+                             (let [data (js->clj (js/JSON.parse (.-data e)) :keywordize-keys true)]
+                               (offer! data-from-ws-chan data))))
 
     (set! (.-onclose ws) (fn [_e] (println :ws-stopped)))
     ws))
@@ -38,7 +39,8 @@
       (when (or (nil? ws)
                 (> (.-readyState ws) js/WebSocket.OPEN))
         (let [ws-opened-ch (chan)
-              ws* (ws-listen (:user-uuid state) (:data-from-ws-chan state) ws-opened-ch)]
+              token (state/get-auth-id-token)
+              ws* (ws-listen token (:data-from-ws-chan state) ws-opened-ch)]
           (<! ws-opened-ch)
           (reset! (:*ws state) ws*)
           (when-let [graph-uuid @(:*graph-uuid state)]