wip: transact to worker directly

Previously, the transactions will go though main thread db -> worker
db, this commit changes it to worker db first and then send the
updates to the main thread db.

This change does introduce some bugs because all the db transactions
are async now.

deps/db/src/logseq/db.cljs

@@ -48,18 +48,23 @@
     {:block/page [:db/id :block/name :block/original-name :block/journal-day]}
     {:block/_parent ...}])
 
+(defonce *transact-fn (atom nil))
+(defn register-transact-fn!
+  [f]
+  (when f (reset! *transact-fn f)))
 
 (defn transact!
   ([conn tx-data]
    (transact! conn tx-data nil))
   ([conn tx-data tx-meta]
    (let [tx-data (common-util/fast-remove-nils tx-data)]
-    (when (seq tx-data)
+     (when (seq tx-data)
 
       ;; (prn :debug :transact)
       ;; (cljs.pprint/pprint tx-data)
 
-      (d/transact! conn tx-data tx-meta)))))
+       (let [f (or @*transact-fn d/transact!)]
+         (f conn tx-data tx-meta))))))
 
 (defn create-default-pages!
   "Creates default pages if one of the default pages does not exist. This

src/main/frontend/db/conn.cljs

@@ -10,7 +10,6 @@
             [logseq.db :as ldb]
             [logseq.db.frontend.schema :as db-schema]
             [logseq.common.util :as common-util]
-            [datascript.core :as d]
             [logseq.db.sqlite.util :as sqlite-util]))
 
 (defonce conns (atom {}))

src/main/frontend/handler.cljs

@@ -46,9 +46,6 @@
             [frontend.persist-db.browser :as db-browser]
             [frontend.persist-db :as persist-db]))
 
-;; TODO: remove this after transact directly to worker db
-(reset! db-listener/*db-listener persist-db/transact-db->worker!)
-
 (defn- set-global-error-notification!
   []
   (when-not config/dev?

src/main/frontend/handler/editor.cljs

@@ -821,15 +821,15 @@
                              (delete-block-fn prev-block)
                              (save-block! repo block new-content {:editor/op :delete})
                              (outliner-save-block! {:db/id (:db/id block)
-                                                         :block/parent (:db/id (:block/parent prev-block))
-                                                         :block/left (or (:db/id (:block/left prev-block))
-                                                                         (:db/id (:block/parent prev-block)))})
+                                                    :block/parent (:db/id (:block/parent prev-block))
+                                                    :block/left (or (:db/id (:block/left prev-block))
+                                                                    (:db/id (:block/parent prev-block)))})
 
                              ;; block->right needs to point its `left` to block->left
                              (when (and block-right (not= (:db/id (:block/parent prev-block))
                                                           (:db/id (:block/parent block))))
                                (outliner-save-block! {:db/id (:db/id block-right)
-                                                           :block/left (:db/id (:block/left block))}))
+                                                      :block/left (:db/id (:block/left block))}))
 
                              ;; update prev-block's children to point to the refed block
                              (when (or (:block/collapsed? prev-block)
@@ -838,18 +838,18 @@
                                  (doseq [child children]
                                    (when-not (= (:db/id child) (:db/id block))
                                      (outliner-save-block! {:db/id (:db/id child)
-                                                                 :block/parent (:db/id block)
-                                                                 :block/left (:db/id block)})))))
+                                                            :block/parent (:db/id block)
+                                                            :block/left (:db/id block)})))))
 
                              ;; parent will be removed
                              (when (= (:db/id prev-block) (:db/id (:block/parent block)))
                                (when-let [parent-right (outliner-core/get-right-sibling (db/get-db) (:db/id prev-block))]
                                  (outliner-save-block! {:db/id (:db/id parent-right)
-                                                             :block/left (:db/id block)})))
+                                                        :block/left (:db/id block)})))
 
                              (when db-based?
                                (outliner-save-block! {:db/id (:db/id block)
-                                                           :block/properties new-properties}))
+                                                      :block/properties new-properties}))
                              (when pos
                                (util/schedule
                                 (fn []
@@ -861,7 +861,7 @@
                              (save-block! repo prev-block new-content {:editor/op :delete})
                              (when db-based?
                                (outliner-save-block! {:db/id (:db/id prev-block)
-                                                           :block/properties new-properties})))))
+                                                      :block/properties new-properties})))))
 
                        :else
                        (delete-block-fn block))))))))))))

src/main/frontend/handler/editor/lifecycle.cljs

@@ -4,7 +4,8 @@
             [frontend.state :as state]
             [frontend.util :as util]
             [frontend.util.cursor :as cursor]
-            [goog.dom :as gdom]))
+            [goog.dom :as gdom]
+            [frontend.db :as db]))
 
 (defn did-mount!
   [state]
@@ -43,7 +44,9 @@
            (not (contains? #{:insert :indent-outdent :auto-save :undo :redo :delete} (state/get-editor-op)))
            ;; Don't trigger auto-save if the latest op is undo or redo
            (not (contains? #{:undo :redo :paste-blocks} (state/get-editor-latest-op))))
-      (editor-handler/save-block! (get-state) value)))
+      (let [state (get-state)]
+        (when (db/entity [:block/uuid (:block/uuid (:block state))]) ; block still exists
+          (editor-handler/save-block! state value)))))
   state)
 
 (def lifecycle

src/main/frontend/modules/outliner/pipeline.cljs

@@ -7,32 +7,35 @@
             [frontend.state :as state]
             [frontend.util.cursor :as cursor]
             [frontend.util.drawer :as drawer]
-            [frontend.modules.editor.undo-redo :as undo-redo]))
+            [frontend.modules.editor.undo-redo :as undo-redo]
+            [datascript.core :as d]))
 
 (defn- reset-editing-block-content!
   [tx-data tx-meta]
-  (let [repo (state/get-current-repo)
-        db? (config/db-based-graph? repo)]
-    (when-not (or (:undo? tx-meta) (:redo? tx-meta))
-      (when-let [edit-block (state/get-edit-block)]
-        (when-let [last-datom (-> (filter (fn [datom]
-                                            (and (= :block/content (:a datom))
-                                                 (= (:e datom) (:db/id edit-block)))) tx-data)
-                                  last)]
-          (when-let [input (state/get-input)]
-            (when (:added last-datom)
-              (let [entity (db/entity (:e last-datom))
-                    db-content (:block/content entity)
-                    content (if db? db-content
-                                (->> db-content
-                                     (property-util/remove-built-in-properties (or (:block/format entity) :markdown))
-                                     drawer/remove-logbook))
-                    pos (cursor/pos input)
-                    pos (when pos (if (zero? pos) (count content) 0))]
-                (when (not= (string/trim content)
-                            (string/trim (.-value input)))
-                  (state/set-edit-content! input content))
-                (when pos (cursor/move-cursor-to input pos))))))))))
+  ;; FIXME:
+  ;; (let [repo (state/get-current-repo)
+  ;;       db? (config/db-based-graph? repo)]
+  ;;   (when-not (or (:undo? tx-meta) (:redo? tx-meta))
+  ;;     (when-let [edit-block (state/get-edit-block)]
+  ;;       (when-let [last-datom (-> (filter (fn [datom]
+  ;;                                           (and (= :block/content (:a datom))
+  ;;                                                (= (:e datom) (:db/id edit-block)))) tx-data)
+  ;;                                 last)]
+  ;;         (when-let [input (state/get-input)]
+  ;;           (when (:added last-datom)
+  ;;             (let [entity (db/entity (:e last-datom))
+  ;;                   db-content (:block/content entity)
+  ;;                   content (if db? db-content
+  ;;                               (->> db-content
+  ;;                                    (property-util/remove-built-in-properties (or (:block/format entity) :markdown))
+  ;;                                    drawer/remove-logbook))
+  ;;                   pos (cursor/pos input)
+  ;;                   pos (when pos (if (zero? pos) (count content) 0))]
+  ;;               (when (not= (string/trim content)
+  ;;                           (string/trim (.-value input)))
+  ;;                 (state/set-edit-content! input content))
+  ;;               (when pos (cursor/move-cursor-to input pos)))))))))
+  )
 
 (defn store-undo-data!
   [{:keys [tx-meta] :as opts}]
@@ -47,12 +50,16 @@
 
 (defn invoke-hooks
   [{:keys [tx-meta tx-data deleted-block-uuids affected-keys blocks] :as opts}]
-  (store-undo-data! opts)
-  (let [{:keys [from-disk? new-graph?]} tx-meta
+  (let [{:keys [from-disk? new-graph? local-tx?]} tx-meta
         repo (state/get-current-repo)
         tx-report {:tx-meta tx-meta
                    :tx-data tx-data}]
 
+    (when local-tx? (store-undo-data! opts))
+
+    (let [conn (db/get-db repo false)]
+      (d/transact! conn tx-data tx-meta))
+
     (when (= (:outliner-op tx-meta) :delete-page)
       (state/pub-event! [:page/deleted repo (:deleted-page tx-meta) (:file-path tx-meta)]))
 
@@ -65,6 +72,7 @@
         (catch :default e
           (prn :reset-editing-block-content)
           (js/console.error e)))
+
       (let [importing? (:graph/importing @state/state)]
         (when-not importing?
           (react/refresh! repo tx-report affected-keys))

src/main/frontend/persist_db.cljs

@@ -18,9 +18,6 @@
  (defn <unsafe-delete [repo]
    (protocol/<unsafe-delete (get-impl) repo))
 
-(defn <transact-data [repo tx-data tx-meta]
-  (protocol/<transact-data (get-impl) repo tx-data tx-meta))
-
 (defn <export-db
   [repo opts]
   (protocol/<export-db (get-impl) repo opts))
@@ -44,12 +41,3 @@
   (p/let [_ (protocol/<new (get-impl) repo)
           _ (<export-db repo {})]
     (ipc/ipc :db-open repo)))
-
-(defn <release-access-handles
-  [repo]
-  (protocol/<release-access-handles (get-impl) repo))
-
-(defn transact-db->worker!
-  [repo tx-report]
-  (when-not (:pipeline-replace? (:tx-meta tx-report))
-    (<transact-data repo (:tx-data tx-report) (:tx-meta tx-report))))

src/main/frontend/persist_db/browser.cljs

@@ -11,7 +11,8 @@
             [cljs-bean.core :as bean]
             [frontend.state :as state]
             [electron.ipc :as ipc]
-            [frontend.handler.worker :as worker-handler]))
+            [frontend.handler.worker :as worker-handler]
+            [logseq.db :as ldb]))
 
 (defonce *worker (atom nil))
 
@@ -36,6 +37,25 @@
                  (when (seq new-state)
                    (.sync-app-state worker (pr-str new-state)))))))
 
+(defn- transact!
+  [^js worker repo tx-data tx-meta]
+  (let [tx-meta' (pr-str tx-meta)
+        tx-data' (pr-str tx-data)
+        context {:dev? config/dev?
+                 :node-test? util/node-test?
+                 :validate-db-options (:dev/validate-db-options (state/get-config))
+                 :importing? (:graph/importing @state/state)
+                 :date-formatter (state/get-date-formatter)
+                 :export-bullet-indentation (state/get-export-bullet-indentation)
+                 :preferred-format (state/get-preferred-format)
+                 :journals-directory (config/get-journals-directory)
+                 :whiteboards-directory (config/get-whiteboards-directory)
+                 :pages-directory (config/get-pages-directory)}]
+    (if worker
+      (.transact worker repo tx-data' tx-meta'
+                 (pr-str context))
+      (notification/show! "Latest change was not saved! Please restart the application." :error))))
+
 (defn start-db-worker!
   []
   (when-not (or config/publishing? util/node-test?)
@@ -50,9 +70,16 @@
                   _ (.sync-app-state wrapped-worker
                                      (pr-str
                                       {:git/current-repo (state/get-current-repo)
-                                       :config (:config @state/state)}))]
-            (sync-app-state! wrapped-worker)
-            (ask-persist-permission!))
+                                       :config (:config @state/state)}))
+                  _ (sync-app-state! wrapped-worker)
+                  _ (ask-persist-permission!)]
+            (ldb/register-transact-fn!
+             (fn worker-transact!
+               [_conn tx-data tx-meta]
+               (prn :debug :transact :tx-meta tx-meta :tx-data tx-data)
+               (transact! wrapped-worker (state/get-current-repo) tx-data
+                 ;; not from remote(rtc)
+                 (assoc tx-meta :local-tx? true)))))
           (p/catch (fn [error]
                      (prn :debug "Can't init SQLite wasm")
                      (js/console.error error)
@@ -97,26 +124,6 @@
     (when-let [^js sqlite @*worker]
       (.releaseAccessHandles sqlite repo)))
 
-  (<transact-data [_this repo tx-data tx-meta]
-    (let [^js sqlite @*worker]
-      (when-not (:pipeline-replace? tx-meta) ; from db worker
-        (let [tx-meta' (pr-str tx-meta)
-              tx-data' (pr-str tx-data)
-              context {:dev? config/dev?
-                       :node-test? util/node-test?
-                       :validate-db-options (:dev/validate-db-options (state/get-config))
-                       :importing? (:graph/importing @state/state)
-                       :date-formatter (state/get-date-formatter)
-                       :export-bullet-indentation (state/get-export-bullet-indentation)
-                       :preferred-format (state/get-preferred-format)
-                       :journals-directory (config/get-journals-directory)
-                       :whiteboards-directory (config/get-whiteboards-directory)
-                       :pages-directory (config/get-pages-directory)}]
-          (if sqlite
-            (.transact sqlite repo tx-data' tx-meta'
-                       (pr-str context))
-            (notification/show! "Latest change was not saved! Please restart the application." :error))))))
-
   (<fetch-initial-data [_this repo _opts]
     (when-let [^js sqlite @*worker]
       (-> (p/let [db-exists? (.dbExists sqlite repo)

src/main/frontend/persist_db/protocol.cljs

@@ -6,7 +6,6 @@
   (<new [this repo] "Create or open a graph")
   (<unsafe-delete [this repo] "Delete graph and its vfs")
   (<release-access-handles [this repo] "Release access file handles")
-  (<transact-data [this repo tx-data tx-meta] "Transact data to db")
   (<fetch-initial-data [this repo opts] "Fetch Initial data")
   (<export-db [this repo opts] "Save or get SQLite db")
   (<import-db [this repo data] "Import SQLite db"))

src/main/frontend/worker/pipeline.cljs

@@ -69,15 +69,14 @@
 
 (defn invoke-hooks
   [repo conn tx-report context]
-  (let [tx-meta (:tx-meta tx-report)
-        {:keys [from-disk? new-graph?]} tx-meta
-        fix-tx-data (validate-and-fix-db! repo conn tx-report context)]
-    (if (or from-disk? new-graph?)
-      {:tx-report tx-report}
-      (let [{:keys [pages blocks]} (ds-report/get-blocks-and-pages tx-report)
-            deleted-block-uuids (set (outliner-pipeline/filter-deleted-blocks (:tx-data tx-report)))
-            replace-tx (when-not (:pipeline-replace? tx-meta)
-                         (concat
+  (when-not (:pipeline-replace? (:tx-meta tx-report))
+    (let [tx-meta (:tx-meta tx-report)
+          {:keys [from-disk? new-graph?]} tx-meta]
+      (if (or from-disk? new-graph?)
+        {:tx-report tx-report}
+        (let [{:keys [pages blocks]} (ds-report/get-blocks-and-pages tx-report)
+              deleted-block-uuids (set (outliner-pipeline/filter-deleted-blocks (:tx-data tx-report)))
+              replace-tx (concat
                           ;; block path refs
                           (set (compute-block-path-refs-tx tx-report blocks))
 
@@ -93,22 +92,23 @@
                                     (when-let [db-id (:db/id b)]
                                       {:db/id db-id
                                        :block/tx-id tx-id})) updated-blocks)
-                             (remove nil?)))))
-            tx-report' (or
-                        (when (seq replace-tx)
-                          (ldb/transact! conn replace-tx {:replace? true
-                                                          :pipeline-replace? true}))
-                        tx-report)
-            full-tx-data (concat (:tx-data tx-report) fix-tx-data (:tx-data tx-report'))
-            final-tx-report (assoc tx-report' :tx-data full-tx-data)
-            affected-query-keys (when-not (:importing? context)
-                                  (worker-react/get-affected-queries-keys final-tx-report))]
-        (doseq [page pages]
-          (file/sync-to-file repo (:db/id page) tx-meta))
-        {:tx-report final-tx-report
-         :replace-tx-data (:tx-data tx-report')
-         :replace-tx-meta (:tx-meta tx-report')
-         :affected-keys affected-query-keys
-         :deleted-block-uuids deleted-block-uuids
-         :pages pages
-         :blocks blocks}))))
+                             (remove nil?))))
+              tx-report' (or
+                          (when (seq replace-tx)
+                            (ldb/transact! conn replace-tx {:replace? true
+                                                            :pipeline-replace? true}))
+                          tx-report)
+              fix-tx-data (validate-and-fix-db! repo conn tx-report context)
+              full-tx-data (concat (:tx-data tx-report) fix-tx-data (:tx-data tx-report'))
+              final-tx-report (assoc tx-report' :tx-data full-tx-data)
+              affected-query-keys (when-not (:importing? context)
+                                    (worker-react/get-affected-queries-keys final-tx-report))]
+          (doseq [page pages]
+            (file/sync-to-file repo (:db/id page) tx-meta))
+          {:tx-report final-tx-report
+           :replace-tx-data (:tx-data tx-report')
+           :replace-tx-meta (:tx-meta tx-report')
+           :affected-keys affected-query-keys
+           :deleted-block-uuids deleted-block-uuids
+           :pages pages
+           :blocks blocks})))))

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

@@ -489,7 +489,7 @@
       (swap! *depend-on-block-uuid-set conj target-uuid))))
 
 (defmethod local-block-ops->remote-ops-aux :update-op
-  [_ & {:keys [repo conn block update-op left-uuid parent-uuid *remote-ops]}]
+  [_ & {:keys [conn block update-op left-uuid parent-uuid *remote-ops]}]
   (let [block-uuid (:block/uuid block)
         attr-map (:updated-attrs (second update-op))
         attr-alias-map (when (contains? attr-map :alias)