fix: use broadcast-to-clients! to postmessage to clients

deps/common/src/logseq/common/util.cljs

@@ -385,3 +385,11 @@ return: [{:id 3} {:id 2 :depend-on 3} {:id 1 :depend-on 2}]"
         (tc/to-long (f now (t/years 1)))
         nil)
       (tc/to-long (tc/to-date value)))))
+
+(defn keyword->string
+  [x]
+  (if (keyword? x)
+    (if-let [nn (namespace x)]
+      (str nn "/" (name x))
+      (name x))
+    x))

src/main/frontend/worker/db/validate.cljs

@@ -1,6 +1,6 @@
 (ns frontend.worker.db.validate
   "Validate db"
-  (:require [frontend.worker.util :as worker-util]
+  (:require [frontend.worker.shared-service :as shared-service]
             [logseq.db.frontend.validate :as db-validate]))
 
 (defn validate-db
@@ -8,16 +8,16 @@
   (let [{:keys [errors datom-count entities]} (db-validate/validate-db! db)]
     (if errors
       (do
-        (worker-util/post-message :log [:db-invalid :error
-                                        {:msg "Validation errors"
-                                         :errors errors}])
-        (worker-util/post-message :notification
-                                  [(str "Validation detected " (count errors) " invalid block(s). These blocks may be buggy. Attempting to fix invalid blocks. Run validation again to see if they were fixed.")
-                                   :warning false]))
+        (shared-service/broadcast-to-clients! :log [:db-invalid :error
+                                                    {:msg "Validation errors"
+                                                     :errors errors}])
+        (shared-service/broadcast-to-clients! :notification
+                                              [(str "Validation detected " (count errors) " invalid block(s). These blocks may be buggy. Attempting to fix invalid blocks. Run validation again to see if they were fixed.")
+                                               :warning false]))
 
-      (worker-util/post-message :notification
-                                [(str "Your graph is valid! " (assoc (db-validate/graph-counts db entities) :datoms datom-count))
-                                 :success false]))
+      (shared-service/broadcast-to-clients! :notification
+                                            [(str "Your graph is valid! " (assoc (db-validate/graph-counts db entities) :datoms datom-count))
+                                             :success false]))
     {:errors errors
      :datom-count datom-count
      :invalid-entity-ids (distinct (map (fn [e] (:db/id (:entity e))) errors))}))

src/main/frontend/worker/db_worker.cljs

@@ -688,7 +688,7 @@
               {:keys [type payload]} (when (map? data) data)]
           (case type
             :notification
-            (worker-util/post-message type [(:message payload) (:type payload)])
+            (shared-service/broadcast-to-clients! :notification [(:message payload) (:type payload)])
             (throw e)))))))
 
 (def-thread-api :thread-api/file-writes-finished?
@@ -849,6 +849,16 @@
      (c.m/<? (start-db! repo {}))
      (m/? (rtc.core/new-task--rtc-start true)))))
 
+(def broadcast-data-types
+  (set (map
+        common-util/keyword->string
+        [:sync-db-changes
+         :notification
+         :log
+         :add-repo
+         :rtc-log
+         :rtc-sync-state])))
+
 (defn- init-service!
   [graph]
   (when-let [prev-graph (first @*service)]
@@ -857,7 +867,7 @@
     (p/let [service (shared-service/<create-service graph
                                                     (bean/->js fns)
                                                     #(on-become-master graph)
-                                                    #{"sync-db-changes"})]
+                                                    broadcast-data-types)]
       (assert (p/promise? (get-in service [:status :ready])))
       (reset! *service [graph service])
       service)))

src/main/frontend/worker/pipeline.cljs

@@ -4,6 +4,7 @@
             [frontend.worker.commands :as commands]
             [frontend.worker.file :as file]
             [frontend.worker.react :as worker-react]
+            [frontend.worker.shared-service :as shared-service]
             [frontend.worker.state :as worker-state]
             [frontend.worker.util :as worker-util]
             [logseq.common.defkeywords :refer [defkeywords]]
@@ -105,8 +106,8 @@
                    true
                    (db-validate/validate-tx-report! tx-report (:validate-db-options context)))]
       (when (and (get-in context [:validate-db-options :fail-invalid?]) (not valid?))
-        (worker-util/post-message :notification
-                                  [["Invalid DB!"] :error]))))
+        (shared-service/broadcast-to-clients! :notification
+                                              [["Invalid DB!"] :error]))))
 
   ;; Ensure :block/order is unique for any block that has :block/parent
   (when (or (:dev? context) (exists? js/process))

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

@@ -17,6 +17,7 @@
             [frontend.worker.rtc.skeleton]
             [frontend.worker.rtc.ws :as ws]
             [frontend.worker.rtc.ws-util :as ws-util :refer [gen-get-ws-create-map--memoized]]
+            [frontend.worker.shared-service :as shared-service]
             [frontend.worker.state :as worker-state]
             [frontend.worker.util :as worker-util]
             [lambdaisland.glogi :as log]
@@ -635,7 +636,7 @@
   (c.m/run-background-task
    ::subscribe-state
    (m/reduce
-    (fn [_ v] (worker-util/post-message :rtc-sync-state v))
+    (fn [_ v] (shared-service/broadcast-to-clients! :rtc-sync-state v))
     create-get-state-flow)))
 
 (comment

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

@@ -13,6 +13,7 @@
             [frontend.worker.rtc.const :as rtc-const]
             [frontend.worker.rtc.log-and-state :as rtc-log-and-state]
             [frontend.worker.rtc.ws-util :as ws-util]
+            [frontend.worker.shared-service :as shared-service]
             [frontend.worker.state :as worker-state]
             [frontend.worker.util :as worker-util]
             [logseq.db :as ldb]
@@ -395,7 +396,7 @@
                          :persist-op? false} (worker-state/get-context))
           (transact-remote-schema-version! repo)
           (transact-block-refs! repo))))
-      (worker-util/post-message :add-repo {:repo repo}))))
+      (shared-service/broadcast-to-clients! :add-repo {:repo repo}))))
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; async download-graph ;;
@@ -412,12 +413,13 @@
                                                  :graph-uuid graph-uuid
                                                  :schema-version (str schema-version)})))
 
-(defn new-task--download-info-list
-  [get-ws-create-task graph-uuid schema-version]
-  (m/join :download-info-list
-          (ws-util/send&recv get-ws-create-task {:action "download-info-list"
-                                                 :graph-uuid graph-uuid
-                                                 :schema-version (str schema-version)})))
+(comment
+  (defn new-task--download-info-list
+    [get-ws-create-task graph-uuid schema-version]
+    (m/join :download-info-list
+            (ws-util/send&recv get-ws-create-task {:action "download-info-list"
+                                                   :graph-uuid graph-uuid
+                                                   :schema-version (str schema-version)}))))
 
 (defn new-task--wait-download-info-ready
   [get-ws-create-task download-info-uuid graph-uuid schema-version timeout-ms]

src/main/frontend/worker/rtc/log_and_state.cljs

@@ -1,7 +1,7 @@
 (ns frontend.worker.rtc.log-and-state
   "Fns to generate rtc related logs"
   (:require [frontend.common.missionary :as c.m]
-            [frontend.worker.util :as worker-util]
+            [frontend.worker.shared-service :as shared-service]
             [lambdaisland.glogi :as log]
             [logseq.common.defkeywords :refer [defkeywords]]
             [malli.core :as ma]
@@ -86,9 +86,10 @@
   (swap! *graph-uuid->remote-t assoc (ensure-uuid graph-uuid) remote-t))
 
 ;;; subscribe-logs, push to frontend
+;;; TODO: refactor by using c.m/run-background-task
 (defn- subscribe-logs
   []
   (remove-watch *rtc-log :subscribe-logs)
   (add-watch *rtc-log :subscribe-logs
-             (fn [_ _ _ n] (when n (worker-util/post-message :rtc-log n)))))
+             (fn [_ _ _ n] (when n (shared-service/broadcast-to-clients! :rtc-log n)))))
 (subscribe-logs)

src/main/frontend/worker/rtc/skeleton.cljs

@@ -3,7 +3,7 @@
   (:require [clojure.data :as data]
             [datascript.core :as d]
             [frontend.worker.rtc.ws-util :as ws-util]
-            [frontend.worker.util :as worker-util]
+            [frontend.worker.shared-service :as shared-service]
             [lambdaisland.glogi :as log]
             [logseq.db :as ldb]
             [logseq.db.frontend.schema :as db-schema]
@@ -37,19 +37,19 @@
               client-builtin-db-idents (set (get-builtin-db-idents db))
               client-schema-version (ldb/get-graph-schema-version db)]
           (when-not (zero? (db-schema/compare-schema-version client-schema-version server-schema-version))
-            (worker-util/post-message :notification
-                                      [[:div
-                                        [:p (str :client-schema-version client-schema-version)]
-                                        [:p (str :server-schema-version server-schema-version)]]
-                                       :error]))
+            (shared-service/broadcast-to-clients! :notification
+                                                  [[:div
+                                                    [:p (str :client-schema-version client-schema-version)]
+                                                    [:p (str :server-schema-version server-schema-version)]]
+                                                   :error]))
           (let [[client-only server-only _]
                 (data/diff client-builtin-db-idents server-builtin-db-idents)]
             (when (or (seq client-only) (seq server-only))
-              (worker-util/post-message :notification
-                                        [(cond-> [:div]
-                                           (seq client-only)
-                                           (conj [:p (str :client-only-db-idents client-only)])
-                                           (seq server-only)
-                                           (conj [:p (str :server-only-db-idents server-only)]))
-                                         :error])))
+              (shared-service/broadcast-to-clients! :notification
+                                                    [(cond-> [:div]
+                                                       (seq client-only)
+                                                       (conj [:p (str :client-only-db-idents client-only)])
+                                                       (seq server-only)
+                                                       (conj [:p (str :server-only-db-idents server-only)]))
+                                                     :error])))
           r)))))

src/main/frontend/worker/shared_service.cljs

@@ -3,6 +3,7 @@
   (:require [cljs-bean.core :as bean]
             [goog.object :as gobj]
             [lambdaisland.glogi :as log]
+            [logseq.common.util :as common-util]
             [logseq.db :as ldb]
             [promesa.core :as p]))
 
@@ -342,10 +343,8 @@
 (defn broadcast-to-clients!
   [type' data]
   (let [transit-payload (ldb/write-transit-str [type' data])]
-    (.postMessage js/self transit-payload)
+    (when (exists? js/self) (.postMessage js/self transit-payload))
     (when-let [common-channel @*common-channel]
-      (let [str-type' (if-let [ns (namespace type')]
-                        (str ns "/" (name type'))
-                        (name type'))]
+      (let [str-type' (common-util/keyword->string type')]
         (.postMessage common-channel #js {:type str-type'
                                           :data transit-payload})))))

src/rtc_e2e_test/client_steps.cljs

@@ -35,12 +35,12 @@
   client2: start rtc, wait page1, remote->client2"
   {:client1
    (m/sp
-     (let [r (m/? (rtc-core/new-task--rtc-start const/downloaded-test-repo const/test-token))]
+     (let [r (m/? (rtc-core/new-task--rtc-start false))]
        (is (nil? r))
        (m/? (helper/new-task--wait-all-client-ops-sent))))
    :client2
    (m/sp
-     (let [r (m/? (rtc-core/new-task--rtc-start const/downloaded-test-repo const/test-token))]
+     (let [r (m/? (rtc-core/new-task--rtc-start false))]
        (is (nil? r)))
      (m/?
       (c.m/backoff
@@ -162,7 +162,7 @@
        (m/? (helper/new-task--client1-sync-barrier-2->1 "move-blocks-concurrently-signal"))
        (m/? helper/new-task--stop-rtc)
        (helper/transact! conn tx-data2)
-       (is (nil? (m/? (rtc-core/new-task--rtc-start const/downloaded-test-repo const/test-token))))
+       (is (nil? (m/? (rtc-core/new-task--rtc-start false))))
        (m/? (helper/new-task--wait-all-client-ops-sent))
        (m/? (helper/new-task--client1-sync-barrier-2->1 "step5"))
        (let [message (m/? (helper/new-task--wait-message-from-other-client
@@ -189,7 +189,7 @@
        (m/? (helper/new-task--client2-sync-barrier-2->1 "move-blocks-concurrently-signal"))
        (m/? helper/new-task--stop-rtc)
        (helper/transact! conn (const/tx-data-map :move-blocks-concurrently-client2))
-       (is (nil? (m/? (rtc-core/new-task--rtc-start const/downloaded-test-repo const/test-token))))
+       (is (nil? (m/? (rtc-core/new-task--rtc-start false))))
        (m/? (helper/new-task--wait-all-client-ops-sent))
        (m/? (helper/new-task--client2-sync-barrier-2->1 "step5"))
        (m/? (helper/new-task--send-message-to-other-client
@@ -222,7 +222,7 @@ client2:
        (m/? (helper/new-task--client1-sync-barrier-1->2 "step6"))
        (m/? helper/new-task--stop-rtc)
        (helper/transact! conn tx-data2)
-       (let [r (m/? (rtc-core/new-task--rtc-start const/downloaded-test-repo const/test-token))]
+       (let [r (m/? (rtc-core/new-task--rtc-start false))]
          (is (nil? r))
          (m/? (helper/new-task--wait-all-client-ops-sent)))))
    :client2