refactor: fix data in server instead of client

deps/db-sync/src/logseq/db_sync/parent_missing.cljs

@@ -14,8 +14,7 @@
         (let [page (-> (sqlite-util/build-new-page common-config/recycle-page-name)
                        sqlite-create-graph/mark-block-as-built-in)
               {:keys [db-after]} (ldb/transact! conn [page] {:db-sync/recycle-page? true
-                                                             :outliner-op :create-page
-                                                             :rtc/fix? true})]
+                                                             :op :create-recycle-page})]
           (d/entity db-after [:block/uuid (:block/uuid page)])))))
 
 (defn get-missing-parent-datoms
@@ -35,8 +34,7 @@
     (outliner-tx/transact!
      {:persist-op? true
       :gen-undo-ops? false
-      :outliner-op :fix-missing-parent
-      :rtc/fix? true
+      :op :fix-missing-parent
       :transact-opts {:conn conn}}
      (outliner-core/move-blocks! conn blocks recycle-page {:sibling? false}))))
 

deps/db-sync/src/logseq/db_sync/worker.cljs

@@ -7,13 +7,11 @@
             [logseq.common.authorization :as authorization]
             [logseq.db :as ldb]
             [logseq.db-sync.common :as common :refer [cors-headers]]
-            [logseq.db-sync.cycle :as cycle]
             [logseq.db-sync.malli-schema :as db-sync-schema]
             [logseq.db-sync.parent-missing :as db-sync-parent-missing]
             [logseq.db-sync.protocol :as protocol]
             [logseq.db-sync.storage :as storage]
             [logseq.db-sync.worker-core :as worker-core]
-            [logseq.db.common.normalize :as db-normalize]
             [promesa.core :as p]
             [shadow.cljs.modern :refer (defclass)]))
 
@@ -328,52 +326,24 @@
                          addresses)))
     (set! (.-conn self) (storage/open-conn sql))))
 
-(defn- cycle-reject-response [db tx-data {:keys [attr entity]}]
-  (let [server-values (cycle/server-values-for db tx-data attr)]
-    (log/info :db-sync/cycle-reject
-              {:attr attr
-               :entity entity
-               :entity-title (entity-title db entity)
-               :server-values (count server-values)
-               :tx-count (count tx-data)})
-    {:type "tx/reject"
-     :reason "cycle"
-     :data (common/write-transit
-            {:attr attr
-             :server_values server-values})}))
-
-(defn- missing-parent-reject-response [db datoms]
-  (log/info :db-sync/missing-parent-reject
-            {:datoms datoms})
-  {:type "tx/reject"
-   :reason "missing-parent"
-   :data (common/write-transit
-          {:eids (map (fn [d] (let [block (d/entity db (:e d))]
-                                [:block/uuid (:block/uuid block)])) datoms)})})
-
 (defn- apply-tx! [^js self sender txs]
   (let [sql (.-sql self)
         conn (.-conn self)]
     (when-not conn
       (fail-fast :db-sync/missing-db {:op :apply-tx}))
-    (let [tx-data (protocol/transit->tx txs)
-          db @conn
-          tx-report (d/with db tx-data)
-          db' (:db-after tx-report)
-          missing-parent-datoms (db-sync-parent-missing/get-missing-parent-datoms tx-report)]
-      (if (seq missing-parent-datoms)
-        (missing-parent-reject-response db' missing-parent-datoms)
-        ;; TODO: move fix order to client to keep worker thin
-        (let [order-fixed (worker-core/fix-duplicate-orders db' tx-data)
-              cycle-info (cycle/detect-cycle db' order-fixed)]
-          (if cycle-info
-            (cycle-reject-response db order-fixed cycle-info)
-            (do
-              (ldb/transact! conn order-fixed)
-              (let [new-t (storage/get-t sql)]
-              ;; FIXME: no need to broadcast if client tx is less than remote tx
-                (broadcast! self sender {:type "changed" :t new-t})
-                new-t))))))))
+    (let [tx-data (protocol/transit->tx txs)]
+      (ldb/transact-with-temp-conn!
+       conn
+       {:apply-tx? true}
+       (fn [temp-conn *batch-tx-data]
+         (let [tx-report (d/transact! temp-conn tx-data {:op :apply-client-tx})]
+           ;; TODO: fix cycle
+           (db-sync-parent-missing/fix-parent-missing! temp-conn tx-report)
+           (worker-core/fix-duplicate-orders! temp-conn @*batch-tx-data))))
+      (let [new-t (storage/get-t sql)]
+        ;; FIXME: no need to broadcast if client tx is less than remote tx
+        (broadcast! self sender {:type "changed" :t new-t})
+        new-t))))
 
 (defn- handle-tx-batch! [^js self sender txs t-before]
   (let [current-t (t-now self)]

deps/db-sync/src/logseq/db_sync/worker_core.cljs

@@ -33,8 +33,9 @@
             :else nil))
         tx-data))
 
-(defn fix-duplicate-orders [db tx-data]
-  (let [updates (->> (attr-updates-from-tx db tx-data :block/order)
+(defn fix-duplicate-orders! [conn tx-data]
+  (let [db @conn
+        updates (->> (attr-updates-from-tx db tx-data :block/order)
                      (filter (fn [{:keys [e v]}] (and e v))))
         groups (group-by (fn [{:keys [e v]}]
                            (let [parent (:block/parent (d/entity db e))]
@@ -73,6 +74,5 @@
                    acc))
                []
                groups)]
-    (if (seq fixes)
-      (into tx-data fixes)
-      tx-data)))
+    (when (seq fixes)
+      (d/transact! conn fixes {:op :fix-duplicate-order}))))

deps/db-sync/test/logseq/db_sync/worker_test.cljs

@@ -23,7 +23,7 @@
                         :block/parent [:block/uuid parent]
                         :block/order order-b}])
     (let [tx [[:db/add [:block/uuid block-b] :block/order order-a]]
-          fixed (worker-core/fix-duplicate-orders @conn tx)
+          fixed (worker-core/fix-duplicate-orders! @conn tx)
           db' (d/db-with @conn fixed)
           order-a' (:block/order (d/entity db' [:block/uuid block-a]))
           order-b' (:block/order (d/entity db' [:block/uuid block-b]))]

deps/db/src/logseq/db.cljs

@@ -197,7 +197,7 @@
     (d/listen! temp-conn ::temp-conn-batch-tx
                (fn [{:keys [tx-data]}]
                  (vswap! *batch-tx-data into tx-data)))
-    (batch-tx-fn temp-conn)
+    (batch-tx-fn temp-conn *batch-tx-data)
     (let [tx-data @*batch-tx-data]
       (d/unlisten! temp-conn ::temp-conn-batch-tx)
       (reset! temp-conn nil)

deps/db/test/logseq/db_test.cljs

@@ -105,7 +105,7 @@
       (ldb/transact-with-temp-conn!
        conn
        {}
-       (fn [temp-conn]
+       (fn [temp-conn _*batch-tx-data]
          (ldb/transact! temp-conn [{:db/ident :logseq.class/Task
                                     :block/tags :logseq.class/Property}])
-         (ldb/transact! temp-conn [[:db/retract :logseq.class/Task :block/tags :logseq.class/Property]]))))))
+         (ldb/transact! temp-conn [[:db/retract :logseq.class/Task :block/tags :logseq.class/Property]]))))))

src/main/frontend/worker/db_sync.cljs

@@ -8,13 +8,9 @@
             [logseq.common.path :as path]
             [logseq.common.util :as common-util]
             [logseq.db :as ldb]
-            [logseq.db-sync.cycle :as db-sync-cycle]
             [logseq.db-sync.malli-schema :as db-sync-schema]
-            [logseq.db-sync.parent-missing :as db-sync-parent-missing]
             [logseq.db.common.normalize :as db-normalize]
             [logseq.db.sqlite.util :as sqlite-util]
-            [logseq.outliner.core :as outliner-core]
-            [logseq.outliner.transaction :as outliner-tx]
             [promesa.core :as p]))
 
 (defonce *repo->latest-remote-tx (atom {}))
@@ -232,123 +228,24 @@
                    (some-> (:block/uuid ent) str)))))
        (distinct)))
 
-(defn- reparent-cycle-block!
-  [conn block]
-  (when-let [page (:block/page block)]
-    (outliner-tx/transact!
-     {:transact-opts {:conn conn}
-      :persist-op? true
-      :gen-undo-ops? false
-      :outliner-op :fix-cycle-parent}
-     (outliner-core/move-blocks! conn [block] page {:sibling? false}))))
-
-(defn- fix-cycle-after-remote-tx!
-  [conn db tx-data]
-  ;; FIXME: replace `entity` with `eid`
-  (when-let [{:keys [attr entity]} (and (seq tx-data)
-                                        (db-sync-cycle/detect-cycle db tx-data))]
-    (let [eid entity]
-      (log/info :db-sync/remote-cycle-detected
-                {:attr attr
-                 :eid eid})
-      (when (= attr :block/parent)
-        (when-let [block (d/entity db eid)]
-          (reparent-cycle-block! conn block))))))
-
-(defn- reconcile-cycle! [repo attr server_values]
-  (if-let [conn (worker-state/get-datascript-conn repo)]
-    (let [db @conn
-          tx-data (reduce
-                   (fn [acc [eid value]]
-                     (let [entity (d/entity db eid)
-                           ;; FIXME: extends cardinality/many
-                           current (:db/id (get entity attr))]
-                       (cond
-                         (nil? entity) acc
-                         (nil? value)
-                         (if current
-                           (conj acc [:db/retract eid attr current])
-                           acc)
-                         :else
-                         (conj acc [:db/add eid attr value]))))
-                   []
-                   server_values)]
-      (log/info :db-sync/reconcile-cycle
-                {:repo repo
-                 :attr attr
-                 :server-values (count server_values)
-                 :tx-count (count tx-data)
-                 :entity-titles (->> (keys server_values)
-                                     (keep (fn [ref]
-                                             (when-let [ent (d/entity db ref)]
-                                               {:uuid (some-> (:block/uuid ent) str)
-                                                :title (or (:block/title ent)
-                                                           (:block/name ent))})))
-                                     (take 10))})
-      (when (seq tx-data)
-        (ldb/transact! conn tx-data {:rtc-tx? true})))
-    (fail-fast :db-sync/missing-db {:repo repo :op :reconcile-cycle})))
-
-(defn- normalize-entity-ref
-  [db entity]
-  (cond
-    (vector? entity) entity
-    (number? entity) (when-let [ent (d/entity db entity)]
-                       (cond
-                         (:block/uuid ent) [:block/uuid (:block/uuid ent)]
-                         (:db/ident ent) [:db/ident (:db/ident ent)]
-                         :else nil))
-    (uuid? entity) [:block/uuid entity]
-    (keyword? entity) [:db/ident entity]
-    :else nil))
-
-(defn- strip-cycle-attrs
-  [db tx-data {:keys [attr entity-refs]}]
-  (let [entity-refs (set entity-refs)]
-    (->> tx-data
-         (mapcat
-          (fn [tx]
-            (cond
-              (and (vector? tx) (= attr (nth tx 2 nil)))
-              (let [entity (nth tx 1 nil)
-                    entity-ref (normalize-entity-ref db entity)]
-                (if (and entity-ref (contains? entity-refs entity-ref))
-                  []
-                  [tx]))
-
-              (and (map? tx) (contains? tx attr))
-              (let [entity (or (:db/id tx) (:block/uuid tx) (:db/ident tx))
-                    entity-ref (normalize-entity-ref db entity)]
-                (if (and entity-ref (contains? entity-refs entity-ref))
-                  (let [tx' (dissoc tx attr)
-                        meaningful (seq (dissoc tx' :db/id :block/uuid :db/ident))]
-                    (if meaningful [tx'] []))
-                  [tx]))
-
-              :else
-              [tx]))))))
-
 (defn- client-ops-conn [repo]
   (worker-state/get-client-ops-conn repo))
 
-(defn- persist-local-tx! [repo tx-str tx-meta]
+(defn- persist-local-tx! [repo tx-str _tx-meta]
   (when-let [conn (client-ops-conn repo)]
     (let [tx-id (random-uuid)
           now (.now js/Date)]
       (ldb/transact! conn [{:db-sync/tx-id tx-id
                             :db-sync/tx tx-str
-                            :db-sync/created-at now
-                            :db-sync/fix? (:rtc/fix? tx-meta)}])
+                            :db-sync/created-at now}])
       tx-id)))
 
 (defn- pending-txs
   [repo limit]
   (when-let [conn (client-ops-conn repo)]
     (let [db @conn
-          datoms (take limit (d/datoms db :avet :db-sync/created-at))
-          fixs (d/datoms db :avet :db-sync/fix? true)
-          full-datoms (distinct (concat datoms fixs))]
-      (->> full-datoms
+          datoms (take limit (d/datoms db :avet :db-sync/created-at))]
+      (->> datoms
            (map (fn [datom]
                   (d/entity db (:e datom))))
            (keep (fn [ent]
@@ -398,60 +295,6 @@
                                :t_before (or (client-op/get-local-tx repo) 0)
                                :txs (sqlite-util/write-transit-str tx-data)})))))))))))
 
-(defn- pending-txs-by-ids
-  [repo tx-ids]
-  (if-let [conn (client-ops-conn repo)]
-    (let [db @conn]
-      (keep (fn [tx-id]
-              (when-let [ent (d/entity db [:db-sync/tx-id tx-id])]
-                (when-let [tx (:db-sync/tx ent)]
-                  {:tx-id tx-id
-                   :tx tx})))
-            tx-ids))
-    (fail-fast :db-sync/missing-db {:repo repo :op :pending-txs-by-ids})))
-
-(defn- requeue-non-parent-txs!
-  [repo attr server_values entries]
-  (let [db (some-> (worker-state/get-datascript-conn repo) deref)
-        entity-refs (when (seq server_values) (set (keys server_values)))
-        scoped? (and db attr (seq entity-refs))
-        requeued (volatile! 0)
-        stripped (volatile! 0)]
-    (if-not scoped?
-      (fail-fast :db-sync/missing-field
-                 {:repo repo
-                  :has-db? (boolean db)
-                  :attr attr
-                  :server-values (count server_values)
-                  :entries (count entries)})
-      (do
-        (doseq [{:keys [tx]} entries]
-          (when (string? tx)
-            (vswap! stripped inc)
-            (let [tx-data (parse-transit tx {:repo repo :op :requeue-non-parent-txs})
-                  filtered (strip-cycle-attrs db tx-data {:attr attr :entity-refs entity-refs})]
-              (when (seq filtered)
-                (vswap! requeued inc)
-                (persist-local-tx! repo (sqlite-util/write-transit-str filtered) {:rtc/fix? true})))))
-        (log/info :db-sync/requeue-non-parent-txs
-                  {:repo repo
-                   :entries (count entries)
-                   :stripped @stripped
-                   :requeued @requeued})))))
-
-(defn- cycle-entity-titles
-  [repo server_values]
-  (if-let [conn (worker-state/get-datascript-conn repo)]
-    (let [db @conn]
-      (->> (keys server_values)
-           (keep (fn [ref]
-                   (when-let [ent (d/entity db ref)]
-                     {:uuid (some-> (:block/uuid ent) str)
-                      :title (or (:block/title ent)
-                                 (:block/name ent))})))
-           (take 10)))
-    (fail-fast :db-sync/missing-db {:repo repo :op :cycle-entity-titles})))
-
 (defn- ensure-client-state! [repo]
   (let [client {:repo repo
                 :send-queue (atom (p/resolved nil))
@@ -679,60 +522,15 @@
 (defn- apply-remote-tx! [repo client tx-data]
   (if-let [conn (worker-state/get-datascript-conn repo)]
     (try
-      (let [report' (d/with @conn tx-data)
-            fix-tx-data (db-sync-parent-missing/fix-parent-missing! conn report')
-            tx-report (ldb/transact! conn fix-tx-data {:rtc-tx? true})
+      (let [tx-report (ldb/transact! conn tx-data {:rtc-tx? true})
             db-after (:db-after tx-report)
             asset-uuids (asset-uuids-from-tx db-after (:tx-data tx-report))]
-        ;; FIXME: cycle should be fixed before `ldb/transact!`
-        (fix-cycle-after-remote-tx! conn db-after tx-data)
         (when (seq asset-uuids)
           (enqueue-asset-downloads! repo client asset-uuids)))
       (catch :default e
         (log/error :db-sync/apply-remote-tx-failed {:error e})))
     (fail-fast :db-sync/missing-db {:repo repo :op :apply-remote-tx})))
 
-(defn- fix-cycle!
-  [repo client message local-tx remote-tx]
-  (when (nil? (:data message))
-    (fail-fast :db-sync/missing-field
-               {:repo repo :type "tx/reject" :field :data}))
-  (let [{:keys [attr server_values]}
-        (parse-transit (:data message) {:repo repo :type "tx/reject"})]
-    (when (nil? attr)
-      (fail-fast :db-sync/missing-field
-                 {:repo repo :type "tx/reject" :field :attr}))
-    (when (nil? server_values)
-      (fail-fast :db-sync/missing-field
-                 {:repo repo :type "tx/reject" :field :server_values}))
-    ;; FIXME: fix cycle shouldn't re-trigger uploading
-    (let [inflight-ids @(:inflight client)
-          inflight-entries (pending-txs-by-ids repo inflight-ids)]
-      (log/info :db-sync/tx-reject-cycle
-                {:repo repo
-                 :attr attr
-                 :server-values (count server_values)
-                 :entity-titles (cycle-entity-titles repo server_values)
-                 :inflight-ids (count inflight-ids)
-                 :local-tx local-tx
-                 :remote-tx remote-tx})
-      (reconcile-cycle! repo attr server_values)
-      (remove-pending-txs! repo inflight-ids)
-      (reset! (:inflight client) [])
-      (requeue-non-parent-txs! repo attr server_values inflight-entries))
-    (flush-pending! repo client)))
-
-(defn- fix-parent-missing!
-  [repo client message]
-  (if-let [conn (worker-state/get-datascript-conn repo)]
-    (let [db @conn
-          {:keys [eids]} (parse-transit (:data message) {:repo repo :type "tx/reject"})
-          blocks (keep #(d/entity db %) eids)]
-      (when (seq blocks)
-        (db-sync-parent-missing/move-blocks-to-recycle! conn blocks))
-      (flush-pending! repo client))
-    (fail-fast :db-sync/missing-db {:repo repo :op :reconcile-cycle})))
-
 (defn- handle-message! [repo client raw]
   (let [message (-> raw parse-message coerce-ws-server-message)]
     (when-not (map? message)
@@ -781,10 +579,6 @@
                       (case reason
                         "stale"
                         (send! (:ws client) {:type "pull" :since local-tx})
-                        "cycle"
-                        (fix-cycle! repo client message local-tx remote-tx)
-                        "parent-missing"
-                        (fix-parent-missing! repo client message)
 
                         (fail-fast :db-sync/invalid-field
                                    {:repo repo :type "tx/reject" :reason reason})))
@@ -845,7 +639,7 @@
         nil))))
 
 (defn- connect! [repo client url]
-  (when-let [current (:ws client)]
+  (when (:ws client)
     (stop-client! client))
   (let [ws (js/WebSocket. (append-token url (auth-token)))
         updated (assoc client :ws ws)]

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

@@ -99,8 +99,7 @@
    :local-tx {:db/index true}
    :graph-uuid {:db/index true}
    :db-sync/tx-id {:db/unique :db.unique/identity}
-   :db-sync/created-at {:db/index true}
-   :db-sync/fix? {:db/index true}})
+   :db-sync/created-at {:db/index true}})
 
 (defn update-graph-uuid
   [repo graph-uuid]

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

@@ -663,7 +663,7 @@
         (js/console.groupCollapsed "rtc/apply-remote-ops-log")
         (ldb/transact-with-temp-conn!
          conn tx-meta
-         (fn [temp-conn]
+         (fn [temp-conn _*batch-tx-data]
            (worker-util/profile :ensure-refed-blocks-exist (ensure-refed-blocks-exist repo temp-conn refed-blocks))
            (worker-util/profile :apply-remote-update-page-ops (apply-remote-update-page-ops repo temp-conn update-page-ops))
            (worker-util/profile :apply-remote-move-ops (apply-remote-move-ops temp-conn sorted-move-ops))

src/test/frontend/worker/rtc/rtc_fns_test.cljs

@@ -159,7 +159,7 @@
   (ldb/transact-with-temp-conn!
    conn
    {}
-   (fn [temp-conn]
+   (fn [temp-conn _*batch-tx-data]
      (#'r.remote/apply-remote-move-ops temp-conn move-ops))))
 
 (deftest apply-remote-move-ops-test