update some rtc related ns

deps/db/src/logseq/db/sqlite/db.cljs

@@ -12,6 +12,11 @@
 ;; Reference same sqlite default class in cljs + nbb without needing .cljc
 (def sqlite (if (find-ns 'nbb.core) (aget sqlite3 "default") sqlite3))
 
+(defn get-graphs-dir
+  []
+  (let [path (.getPath ^object app "home")]
+    (node-path/join path "logseq" "graphs")))
+
 (defn close!
   []
   (when @databases
@@ -106,7 +111,7 @@
 ;; other data such as config.edn, custom css/js
 ;; current page, sidebar blocks
 
-(defn- query
+(defn query
   [repo db sql]
   (let [stmt (prepare db sql repo)]
     (.all ^object stmt)))
@@ -134,4 +139,4 @@
   [repo journal-block-uuids]
   (when-let [db (get-db repo)]
     (query repo db (str "select * from blocks where type = 1 and uuid not in "
-                        (clj-list->sql journal-block-uuids)))))
+                        (clj-list->sql journal-block-uuids)))))

deps/db/src/logseq/db/sqlite/rtc.cljs

@@ -0,0 +1,46 @@
+(ns logseq.db.sqlite.rtc
+  "Entry point for using sqlite to store operations for rtc-sync"
+  (:require [logseq.db.sqlite.db :as sqlite-db]))
+
+(defn- ds-op->sqlite-op
+  [op]
+  {:v (js/JSON.stringify (clj->js op))})
+
+(defn create-op-table!
+  [db db-name]
+  (let [stmt (sqlite-db/prepare db "CREATE TABLE IF NOT EXISTS rtc_ops (
+                         id   INTEGER PRIMARY KEY AUTOINCREMENT,
+                         v    TEXT NOT NULL)"
+                                (str db-name))
+        init-stmt (sqlite-db/prepare db "INSERT INTO rtc_ops (v) VALUES (@v)" (str db-name))]
+    (.run ^object stmt)
+    (.run ^object init-stmt (clj->js (ds-op->sqlite-op {:local-tx 0})))))
+
+(defn init!
+  [repo]
+  (let [graphs-dir (sqlite-db/get-graphs-dir)
+        [_db-sanitized-name db-full-path] (sqlite-db/get-db-full-path graphs-dir repo)
+        db (new sqlite-db/sqlite db-full-path nil)]
+    (create-op-table! db repo)))
+
+
+(defn add-ops!
+  [repo ops]
+  (when (seq ops)
+    (when-let [db (sqlite-db/get-db repo)]
+      (let [insert      (sqlite-db/prepare db "INSERT INTO rtc_ops (v) VALUES (@v)" repo)
+            insert-many (.transaction ^object db
+                                      (fn [ops]
+                                        (doseq [op ops]
+                                          (.run ^object insert (clj->js (ds-op->sqlite-op op))))))]
+        (insert-many ops)))))
+
+(defn get-ops&local-tx
+  [repo]
+  (when-let [db (sqlite-db/get-db repo)]
+    (sqlite-db/query repo db "select * from rtc_ops")))
+
+(defn clean-ops!
+  [repo]
+  (when-let [db (sqlite-db/get-db repo)]
+    (.run (sqlite-db/prepare db "DELETE FROM rtc_ops WHERE id <> 1" repo))))

src/electron/electron/db.cljs

@@ -4,16 +4,10 @@
             ["fs-extra" :as fs]
             ["electron" :refer [app]]
             [electron.logger :as logger]
-            [logseq.db.sqlite.db :as sqlite-db]))
+            [logseq.db.sqlite.db :as sqlite-db :refer [get-graphs-dir]]))
 
 (def close! sqlite-db/close!)
 
-;; ~/logseq
-(defn get-graphs-dir
-  []
-  (let [path (.getPath ^object app "home")]
-    (node-path/join path "logseq" "graphs")))
-
 (defn ensure-graphs-dir!
   []
   (fs/ensureDirSync (get-graphs-dir)))

src/electron/electron/handler.cljs

@@ -29,13 +29,12 @@
             [electron.state :as state]
             [electron.utils :as utils]
             [electron.window :as win]
+            [electron.handler-interface :refer [handle]]
             [logseq.db.sqlite.db :as sqlite-db]
             [logseq.db.sqlite.util :as sqlite-util]
             [logseq.common.graph :as common-graph]
             [promesa.core :as p]))
 
-(defmulti handle (fn [_window args] (keyword (first args))))
-
 (defmethod handle :mkdir [_window [_ dir]]
   (fs/mkdirSync dir))
 

src/electron/electron/handler_interface.cljs

@@ -0,0 +1,3 @@
+(ns electron.handler-interface)
+
+(defmulti handle (fn [_window args] (keyword (first args))))

src/electron/electron/rtc_handler.cljs

@@ -0,0 +1,19 @@
+(ns electron.rtc-handler
+  (:require [electron.handler-interface :refer [handle]]
+            [cljs.reader :as reader]
+            [logseq.db.sqlite.rtc :as sqlite-rtc]))
+
+(defmethod handle :rtc/init [_window [_ repo]]
+  (sqlite-rtc/init! repo))
+
+(defmethod handle :rtc/add-ops [_window [_ repo data-str]]
+  (let [ops (reader/read-string data-str)]
+    (when (seq ops)
+      (sqlite-rtc/add-ops! repo ops))))
+
+
+(defmethod handle :rtc/get-ops&local-tx [_window [_ repo]]
+  (sqlite-rtc/get-ops&local-tx repo))
+
+(defmethod handle :rtc/clean-ops [_window [_ repo]]
+  (sqlite-rtc/clean-ops! repo))

src/main/frontend/config.cljs

@@ -58,6 +58,12 @@
       (def IDENTITY-POOL-ID "us-east-2:cc7d2ad3-84d0-4faf-98fe-628f6b52c0a5")
       (def OAUTH-DOMAIN "logseq-test2.auth.us-east-2.amazoncognito.com")))
 
+
+(goog-define ENABLE-RTC-SYNC-PRODUCTION false)
+(if ENABLE-RTC-SYNC-PRODUCTION
+  (do (def RTC-WS-URL "wss://ws.logseq.com/rtc-sync?graphuuid=%"))
+  (do (def RTC-WS-URL "wss://ws-dev.logseq.com/rtc-sync?graphuuid=%"))
+  )
 ;; Feature flags
 ;; =============
 

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

@@ -0,0 +1,33 @@
+(ns frontend.db.rtc.core
+  (:require [datascript.core :as d]
+            [frontend.db.conn :as conn]
+            [frontend.util :as util]
+            [frontend.config :as config]
+            [cljs.core.async :as async :refer [<! >! chan go go-loop offer!
+                                               poll! timeout]]))
+
+(def ws-addr config/RTC-WS-URL)
+
+(defn export-as-blocks
+  [repo]
+  (let [db (conn/get-db repo)
+        datoms (d/datoms db :eavt)]
+    (->> datoms
+         (partition-by :e)
+         (keep (fn [datoms]
+                 (when (seq datoms)
+                   (reduce
+                    (fn [r datom] (assoc r (:a datom) (:v datom)))
+                    {:db/id (:e (first datoms))}
+                    datoms)))))))
+
+(def data-from-ws-chan (chan 1000))
+
+(defn ws-listen!
+  [graph-uuid *ws data-from-ws-chan]
+  (reset! *ws {:ws (js/WebSocket. (util/format ws-addr graph-uuid)) :stop false})
+  (set! (.-onmessage (:ws @*ws)) (fn [e]
+                                   (let [data (js->clj (js/JSON.parse (.-data e)) :keywordize-keys true)]
+                                     (>! data-from-ws-chan data))))
+
+  (set! (.-onclose (:ws @*ws)) (fn [_e] (println :ws-stopped))))

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

@@ -0,0 +1,60 @@
+(ns frontend.db.rtc.op
+  (:require [electron.ipc :as ipc]
+            [malli.core :as m]
+            [cljs.core.async :as async]
+            [cljs.core.async.interop :refer [p->c]]
+            [frontend.components.select :as select]))
+
+(def op-schema
+  [:or
+   [:catn
+    [:op [:= "move"]]
+    [:value [:map
+             [:block-uuid :string]
+             [:target-uuid :string]
+             [:sibling? :boolean]]]]
+   [:catn
+    [:op [:= "remove"]]
+    [:value [:map
+             [:block-uuids [:sequential :string]]]]]
+   [:catn
+    [:op [:= "update"]]
+    [:value [:map
+             [:block-uuid :string]
+             [:content {:optional true} :string]]]]])
+
+(def op-validator (m/validator op-schema))
+
+
+(defn <move-block-op!
+  [repo block-uuid target-uuid sibling?]
+  (let [op ["move" {:block-uuid (str block-uuid)
+                    :target-uuid (str target-uuid)
+                    :sibling? sibling?}]]
+    (assert (op-validator op) "illegal op")
+    (p->c (ipc/ipc :rtc/add-ops repo (pr-str op)))))
+
+(defn <remove-blocks-op!
+  [repo block-uuids]
+  (let [op ["remove" {:block-uuids (mapv str block-uuids)}]]
+    (assert (op-validator op) "illegal op")
+    (p->c (ipc/ipc :rtc/add-ops repo (pr-str op)))))
+
+(defn <update-block-op!
+  [repo block-uuid attrs-map]
+  (let [op ["update" (merge {:block-uuid (str block-uuid)}
+                            (select-keys attrs-map [:content]))]]
+    (assert (op-validator op) "illegal op")
+    (p->c (ipc/ipc :rtc/add-ops repo (pr-str op)))))
+
+(defn <get-ops&local-tx
+  [repo]
+  (p->c (ipc/ipc :rtc/get-ops&local-tx repo)))
+
+(defn <clean-ops
+  [repo]
+  (p->c (ipc/ipc :rtc/clean-ops repo)))
+
+(defn <init
+  [repo]
+  (p->c (ipc/ipc :rtc/init repo)))

src/main/frontend/db/sync.cljs

@@ -1,17 +0,0 @@
-(ns ^:no-doc frontend.db.sync
-  (:require [datascript.core :as d]
-            [frontend.db.conn :as conn]))
-
-
-(defn export-as-blocks
-  [repo]
-  (let [db (conn/get-db repo)
-        datoms (d/datoms db :eavt)]
-    (->> datoms
-         (partition-by :e)
-         (keep (fn [datoms]
-                 (when (seq datoms)
-                   (reduce
-                    (fn [r datom] (assoc r (:a datom) (:v datom)))
-                    {:db/id (:e (first datoms))}
-                    datoms)))))))