Merge branch 'feat/db' into feat/property-default-value

deps/graph-parser/src/logseq/graph_parser/block.cljs

@@ -320,8 +320,8 @@
                :block/title original-page-name'}
               (when (and original-page-name
                          (not= (string/lower-case original-page-name)
-                               (string/lower-case original-page-name')))
-
+                               (string/lower-case original-page-name'))
+                         (not @*export-to-db-graph?))
                 {:block.temp/original-page-name original-page-name})
               (if (and class? page-entity (:db/ident page-entity))
                 {:block/uuid (:block/uuid page-entity)

deps/graph-parser/src/logseq/graph_parser/exporter.cljs

@@ -85,8 +85,15 @@
        (swap! all-idents assoc (keyword class-name) (:db/ident m))
        (with-meta m {:new-class? true})))))
 
-(defn- find-or-gen-class-uuid [page-names-to-uuids page-name db-ident]
-  (or (get @page-names-to-uuids page-name)
+(defn- find-or-gen-class-uuid [page-names-to-uuids page-name db-ident & {:keys [temp-new-class?]}]
+  (or (if temp-new-class?
+        ;; First lookup by possible parent b/c page-names-to-uuids erroneously has the child name
+        ;; and full name. To not guess at the parent name we would need to save all properties-from-classes
+        (or (some #(when (string/ends-with? (key %) (str ns-util/parent-char page-name))
+                     (val %))
+                  @page-names-to-uuids)
+            (get @page-names-to-uuids page-name))
+        (get @page-names-to-uuids page-name))
       (let [new-uuid (common-uuid/gen-uuid :db-ident-block-uuid db-ident)]
         (swap! page-names-to-uuids assoc page-name new-uuid)
         new-uuid)))
@@ -128,7 +135,7 @@
     (let [class-m (find-or-create-class db new-class all-idents)
           class-m' (merge class-m
                           {:block/uuid
-                           (find-or-gen-class-uuid page-names-to-uuids (common-util/page-name-sanity-lc new-class) (:db/ident class-m))})]
+                           (find-or-gen-class-uuid page-names-to-uuids (common-util/page-name-sanity-lc new-class) (:db/ident class-m) {:temp-new-class? true})})]
       (when (:new-class? (meta class-m)) (swap! classes-tx conj class-m'))
       (assert (:block/uuid class-m') "Class must have a :block/uuid")
       [:block/uuid (:block/uuid class-m')])
@@ -978,10 +985,12 @@
         ;; Fetch all named ents once per import file to speed up named lookups
         all-existing-page-uuids (get-all-existing-page-uuids @conn)
         all-pages (map #(modify-page-tx % all-existing-page-uuids) all-pages*)
+        all-new-page-uuids (->> all-pages
+                                (remove #(all-existing-page-uuids (or (::original-name %) (:block/name %))))
+                                (map (juxt (some-fn ::original-name :block/name) :block/uuid))
+                                (into {}))
         ;; Stateful because new page uuids can occur via tags
-        page-names-to-uuids (atom (merge all-existing-page-uuids
-                                         (into {} (map (juxt (some-fn ::original-name :block/name) :block/uuid)
-                                                       (remove all-existing-page-uuids all-pages)))))
+        page-names-to-uuids (atom (merge all-existing-page-uuids all-new-page-uuids))
         per-file-state {:page-names-to-uuids page-names-to-uuids
                         :classes-tx (:classes-tx options)}
         all-pages-m (mapv #(handle-page-properties % @conn per-file-state all-pages options)

deps/graph-parser/test/logseq/graph_parser/exporter_test.cljs

@@ -193,14 +193,14 @@
 
       ;; Counts
       ;; Includes journals as property values e.g. :logseq.task/deadline
-      (is (= 22 (count (d/q '[:find ?b :where [?b :block/type "journal"]] @conn))))
-      (is (= 22 (count (d/q '[:find ?b :where [?b :block/tags :logseq.class/Journal]] @conn))))
+      (is (= 23 (count (d/q '[:find ?b :where [?b :block/type "journal"]] @conn))))
+      (is (= 23 (count (d/q '[:find ?b :where [?b :block/tags :logseq.class/Journal]] @conn))))
 
       (is (= 4 (count (d/q '[:find ?b :where [?b :block/tags :logseq.class/Task]] @conn))))
       (is (= 3 (count (d/q '[:find ?b :where [?b :block/tags :logseq.class/Query]] @conn))))
 
       ;; Don't count pages like url.md that have properties but no content
-      (is (= 9
+      (is (= 10
              (count (->> (d/q '[:find [(pull ?b [:block/title :block/type]) ...]
                                 :where [?b :block/title] [_ :block/page ?b] (not [?b :logseq.property/built-in?])] @conn)
                          (filter ldb/internal-page?))))

deps/graph-parser/test/resources/exporter-test-graph/journals/2024_11_18.md

@@ -1 +1,2 @@
+- Block with journal ref [[Nov 3th, 2020]]
 - Another block with #Quotes/life

deps/graph-parser/test/resources/exporter-test-graph/pages/Priority.md

@@ -0,0 +1,2 @@
+parent:: [[HumanConcept]]
+- this page triggers bug with a class created via :parent that is then used by :type

deps/graph-parser/test/resources/exporter-test-graph/pages/later.md

@@ -0,0 +1 @@
+type:: [[Priority]]

src/dev-cljs/shadow/user.clj

@@ -16,13 +16,20 @@
   ([]
    (when-let [runtime-id (->> (api/repl-runtimes :app)
                               (filter (fn [runtime] (= :browser-worker (:host runtime))))
-                              first
-                              :client-id)]
-     (prn :worker-runtime-id runtime-id)
+                              (map :client-id)
+                              (apply max))]
      (worker-repl runtime-id)))
-  ([runtime-id]
-   (assert runtime-id "runtime-id shouldn't be empty")
-   (api/repl :app {:runtime-id runtime-id})))
+  ([runtime-id-or-which]
+   (assert runtime-id-or-which "runtime-id shouldn't be empty")
+   (if
+    (number? runtime-id-or-which)
+     (do (prn :worker-runtime-id runtime-id-or-which)
+         (api/repl :app {:runtime-id runtime-id-or-which}))
+     (let [runtime-ids (->> (api/repl-runtimes :app)
+                            (filter (fn [runtime] (= :browser-worker (:host runtime))))
+                            (map :client-id))
+           runtime-id (apply (if (= :old runtime-id-or-which) min max) runtime-ids)]
+       (worker-repl runtime-id)))))
 
 (defn runtime-id-list
   []

src/main/frontend/components/block.cljs

@@ -2194,7 +2194,7 @@
         heading (if (true? heading) (min (inc level) 6) heading)
         elem (if heading
                (keyword (str "h" heading ".block-title-wrap.as-heading"
-                             (when block-ref? ".inline")))
+                             (when block-ref? ".as-inline")))
                :span.block-title-wrap)]
     (->elem
      elem

src/main/frontend/components/block.css

@@ -276,6 +276,10 @@
         @apply w-5 h-5;
       }
     }
+
+    &.as-inline {
+      @apply inline;
+    }
   }
 
   &:has(.dsl-query), &:has(.embed-page) {

src/main/frontend/components/header.cljs

@@ -150,7 +150,8 @@
                      :icon (ui/icon "bulb")})
 
                   ;; Disable login on Web until RTC is ready
-                  (when (and (not login?) (not util/web-platform?))
+                  (when (and (not login?) (or (not util/web-platform?)
+                                              config/dev?))
                     {:title (t :login)
                      :options {:on-click #(state/pub-event! [:user/login])}
                      :icon (ui/icon "user")})

src/main/frontend/components/page.cljs

@@ -240,16 +240,17 @@
                                            config)
                             config (common-handler/config-with-document-mode hiccup-config)
                             blocks (if block? [block] (db/sort-by-order children block))]
-                        [:div
-                         (page-blocks-inner page-e blocks config sidebar? whiteboard? block-id)
-                         (when-not (or config/publishing?
-                                       (let [last-child-id (model/get-block-deep-last-open-child-id (db/get-db) (:db/id (last blocks)))
-                                             block' (if last-child-id (db/entity last-child-id) (last blocks))]
-                                         (string/blank? (:block/title block'))))
+                        (let [add-button? (not (or config/publishing?
+                                                 (let [last-child-id (model/get-block-deep-last-open-child-id (db/get-db) (:db/id (last blocks)))
+                                                       block' (if last-child-id (db/entity last-child-id) (last blocks))]
+                                                   (string/blank? (:block/title block')))))]
+                          [:div
+                           {:class (when add-button? "show-add-button")}
+                           (page-blocks-inner page-e blocks config sidebar? whiteboard? block-id)
                            (let [args (if block-id
                                         {:block-uuid block-id}
                                         {:page page-name})]
-                             (add-button args (:container-id config))))]))]
+                             (add-button args (:container-id config)))])))]
          (if (and db-based? (or (ldb/class? block) (ldb/property? block)))
            [:div.mt-4.ml-2.-mb-1
             (ui/foldable

src/main/frontend/components/page.css

@@ -246,3 +246,13 @@ html.is-native-ios {
     @apply pt-2;
   }
 }
+
+.add-button-link-wrap {
+  @apply invisible;
+}
+
+.show-add-button {
+  .add-button-link-wrap {
+    @apply visible;
+  }
+}

src/main/frontend/components/reference.cljs

@@ -159,7 +159,7 @@
     (reset! *ref-pages ref-pages)
     (when (or (seq (:included filters)) (seq (:excluded filters)) (> filter-n 0))
       [:div.references.page-linked.flex-1.flex-row
-       [:div.content.pt-6
+       [:div.content.pt-2
         (references-cp page-entity *filters total filter-n filtered-ref-blocks' *ref-pages)]])))
 
 (rum/defcs references* < rum/reactive db-mixins/query

src/main/frontend/components/views.cljs

@@ -1074,28 +1074,38 @@
 
 (rum/defc table-view < rum/static
   [table option row-selection add-new-object! *scroller-ref]
-  (let [selected-rows (shui/table-get-selection-rows row-selection (:rows table))]
+  (let [selected-rows (shui/table-get-selection-rows row-selection (:rows table))
+        [ready? set-ready?] (rum/use-state false)
+        *rows-wrap (rum/use-ref nil)]
+
+    (rum/use-effect!
+      (fn [] (set-ready? true))
+      [])
+
     (shui/table
      (let [columns' (:columns table)
            rows (:rows table)]
        [:div.ls-table-rows.content.overflow-x-auto.force-visible-scrollbar
-        [:div.relative
-         (table-header table columns' option selected-rows)
-
-         (ui/virtualized-list
-          {:ref #(reset! *scroller-ref %)
-           :custom-scroll-parent (gdom/getElement "main-content-container")
-           :increase-viewport-by {:top 300 :bottom 300}
-           :compute-item-key (fn [idx]
-                               (let [block (nth rows idx)]
-                                 (str "table-row-" (:db/id block))))
-           :total-count (count rows)
-           :item-content (fn [idx]
-                           (let [row (nth rows idx)]
-                             (table-row table row columns' {} option)))})
-
-         (when add-new-object!
-           (shui/table-footer (add-new-row table)))]]))))
+        {:ref *rows-wrap}
+        (when ready?
+          [:div.relative
+           (table-header table columns' option selected-rows)
+
+           (ui/virtualized-list
+             {:ref #(reset! *scroller-ref %)
+              :custom-scroll-parent (or (some-> (rum/deref *rows-wrap) (.closest ".sidebar-item-list"))
+                                      (gdom/getElement "main-content-container"))
+              :increase-viewport-by {:top 300 :bottom 300}
+              :compute-item-key (fn [idx]
+                                  (let [block (nth rows idx)]
+                                    (str "table-row-" (:db/id block))))
+              :total-count (count rows)
+              :item-content (fn [idx]
+                              (let [row (nth rows idx)]
+                                (table-row table row columns' {} option)))})
+
+           (when add-new-object!
+             (shui/table-footer (add-new-row table)))])]))))
 
 (rum/defc list-view < rum/static
   [config view-entity result]

src/main/frontend/config.cljs

@@ -508,6 +508,11 @@
   (when-let [repo-dir (get-repo-dir (state/get-current-repo))]
     (path/path-join repo-dir "assets")))
 
+(defn get-repo-assets-root
+  [repo]
+  (when-let [repo-dir (get-repo-dir repo)]
+    (path/path-join repo-dir "assets")))
+
 (defn get-custom-js-path
   ([]
    (get-custom-js-path (state/get-current-repo)))

src/main/frontend/handler/assets.cljs

@@ -1,5 +1,7 @@
 (ns ^:no-doc frontend.handler.assets
-  (:require [clojure.string :as string]
+  (:require [cljs-http.client :as http]
+            [clojure.string :as string]
+            [frontend.common.missionary-util :as c.m]
             [frontend.config :as config]
             [frontend.fs :as fs]
             [frontend.fs.nfs :as nfs]
@@ -10,6 +12,7 @@
             [logseq.common.path :as path]
             [logseq.common.util :as common-util]
             [medley.core :as medley]
+            [missionary.core :as m]
             [promesa.core :as p]))
 
 (defn alias-enabled?
@@ -213,7 +216,8 @@
 (defn <get-all-assets
   []
   (when-let [path (config/get-current-repo-assets-root)]
-    (p/let [result (fs/readdir path {:path-only? true})]
+    (p/let [result (p/catch (fs/readdir path {:path-only? true})
+                       (constantly nil))]
       (p/all (map (fn [path]
                     (p/let [data (fs/read-file path "" {})]
                       (let [path' (util/node-path.join "assets" (util/node-path.basename path))]
@@ -231,3 +235,81 @@
   [filename]
   (p/let [[repo-dir assets-dir] (ensure-assets-dir! (state/get-current-repo))]
     (path/path-join repo-dir assets-dir filename)))
+
+(defn <get-all-asset-file-paths
+  [repo]
+  (when-let [path (config/get-repo-assets-root repo)]
+    (p/catch (fs/readdir path {:path-only? true})
+             (constantly nil))))
+
+(defn <read-asset
+  [repo asset-block-id asset-type]
+  (let [repo-dir (config/get-repo-dir repo)
+        file-path (path/path-join common-config/local-assets-dir
+                                  (str asset-block-id "." asset-type))]
+    (fs/read-file repo-dir file-path {})))
+
+(defn <get-asset-file-metadata
+  [repo asset-block-id asset-type]
+  (-> (p/let [file (<read-asset repo asset-block-id asset-type)
+              blob (js/Blob. (array file) (clj->js {:type "image"}))
+              checksum (get-file-checksum blob)]
+        {:checksum checksum})
+      (p/catch (constantly nil))))
+
+(defn <write-asset
+  [repo asset-block-id asset-type data]
+  (let [asset-block-id-str (str asset-block-id)
+        repo-dir (config/get-repo-dir repo)
+        file-path (path/path-join common-config/local-assets-dir
+                                  (str asset-block-id-str "." asset-type))]
+    (fs/write-file! repo repo-dir file-path data {})))
+
+(defn <unlink-asset
+  [repo asset-block-id asset-type]
+  (let [file-path (path/path-join (config/get-repo-dir repo)
+                                  common-config/local-assets-dir
+                                  (str asset-block-id "." asset-type))]
+    (p/catch (fs/unlink! repo file-path {}) (constantly nil))))
+
+(defn new-task--rtc-upload-asset
+  [repo asset-block-uuid-str asset-type checksum put-url]
+  (assert (and asset-type checksum))
+  (m/sp
+    (let [asset-file (c.m/<? (<read-asset repo asset-block-uuid-str asset-type))
+          {:keys [status] :as r}
+          (c.m/<? (http/put put-url {:headers {"x-amz-meta-checksum" checksum
+                                               "x-amz-meta-type" asset-type}
+                                     :body asset-file
+                                     :with-credentials? false}))]
+      (when-not (http/unexceptional-status? status)
+        {:ex-data {:type :rtc.exception/upload-asset-failed :data r}}))))
+
+(defn new-task--rtc-download-asset
+  [repo asset-block-uuid-str asset-type get-url]
+  (m/sp
+    (let [{:keys [status body] :as r} (c.m/<? (http/get get-url {:with-credentials? false
+                                                                 :response-type :array-buffer}))]
+      (if-not (http/unexceptional-status? status)
+        {:ex-data {:type :rtc.exception/download-asset-failed :data r}}
+        (do (c.m/<? (<write-asset repo asset-block-uuid-str asset-type body))
+            nil)))))
+
+(comment
+  ;; read asset
+  (p/let [repo "logseq_db_demo"
+          ;; Existing asset block's id
+          asset-block-id-str "672c5a1d-8171-4259-9f35-470c3c67e37f"
+          asset-type "png"
+          data (<read-asset repo asset-block-id-str asset-type)]
+    (js/console.dir data))
+
+  ;; write asset
+  (p/let [repo "logseq_db_demo"
+          ;; Existing asset block's id
+          asset-block-id-str "672c5a1d-8171-4259-9f35-470c3c67e37f"
+          asset-type "png"
+          data (<read-asset repo asset-block-id-str asset-type)
+          new-asset-id (random-uuid)
+          result (<write-asset repo new-asset-id asset-type data)]
+    (js/console.dir result)))

src/main/frontend/handler/worker.cljs

@@ -53,7 +53,7 @@
           (let [data (.-data event)]
             (if (= data "keepAliveResponse")
               (.postMessage worker "keepAliveRequest")
-              (when-not (= (.-type data) "RAW")
+              (when-not (contains? #{"RAW" "APPLY" "RELEASE"} (.-type data))
                 ;; Log thrown exceptions from comlink
                 ;; https://github.com/GoogleChromeLabs/comlink/blob/dffe9050f63b1b39f30213adeb1dd4b9ed7d2594/src/comlink.ts#L223-L236
                 (if (and (= "HANDLER" (.-type data)) (= "throw" (.-name data)))

src/main/frontend/persist_db/browser.cljs

@@ -14,7 +14,8 @@
             [frontend.handler.worker :as worker-handler]
             [logseq.db :as ldb]
             [frontend.db.transact :as db-transact]
-            [frontend.date :as date]))
+            [frontend.date :as date]
+            [frontend.handler.assets :as assets-handler]))
 
 (defonce *worker state/*db-worker)
 
@@ -84,6 +85,35 @@
                  (ldb/write-transit-str context))
       (notification/show! "Latest change was not saved! Please restart the application." :error))))
 
+(defn- with-write-transit-str
+  [p]
+  (p/chain p ldb/write-transit-str))
+
+(deftype Main []
+  Object
+  (readAsset [_this repo asset-block-id asset-type]
+    (assets-handler/<read-asset repo asset-block-id asset-type))
+  (writeAsset [_this repo asset-block-id asset-type data]
+    (assets-handler/<write-asset repo asset-block-id asset-type data))
+  (unlinkAsset [_this repo asset-block-id asset-type]
+    (assets-handler/<unlink-asset repo asset-block-id asset-type))
+  (get-all-asset-file-paths [_this repo]
+    (with-write-transit-str
+      (assets-handler/<get-all-asset-file-paths repo)))
+  (get-asset-file-metadata [_this repo asset-block-id asset-type]
+    (with-write-transit-str
+      (assets-handler/<get-asset-file-metadata repo asset-block-id asset-type)))
+  (rtc-upload-asset [_this repo asset-block-uuid-str asset-type checksum put-url]
+    (with-write-transit-str
+      (js/Promise.
+       (assets-handler/new-task--rtc-upload-asset repo asset-block-uuid-str asset-type checksum put-url))))
+  (rtc-download-asset [_this repo asset-block-uuid-str asset-type get-url]
+    (with-write-transit-str
+      (js/Promise.
+       (assets-handler/new-task--rtc-download-asset repo asset-block-uuid-str asset-type get-url))))
+  (testFn [_this]
+    (prn :debug :works)))
+
 (defn start-db-worker!
   []
   (when-not util/node-test?
@@ -93,6 +123,7 @@
           worker (js/Worker. (str worker-url "?electron=" (util/electron?) "&publishing=" config/publishing?))
           wrapped-worker (Comlink/wrap worker)
           t1 (util/time-ms)]
+      (Comlink/expose (Main.) worker)
       (worker-handler/handle-message! worker wrapped-worker)
       (reset! *worker wrapped-worker)
       (-> (p/let [_ (.init wrapped-worker config/RTC-WS-URL)

src/main/frontend/worker/db_worker.cljs

@@ -31,14 +31,14 @@
             [logseq.common.util :as common-util]
             [logseq.db :as ldb]
             [logseq.db.frontend.order :as db-order]
+            [logseq.db.frontend.schema :as db-schema]
             [logseq.db.sqlite.common-db :as sqlite-common-db]
             [logseq.db.sqlite.create-graph :as sqlite-create-graph]
             [logseq.db.sqlite.util :as sqlite-util]
             [logseq.outliner.op :as outliner-op]
+            [me.tonsky.persistent-sorted-set :as set :refer [BTSet]]
             [promesa.core :as p]
-            [shadow.cljs.modern :refer [defclass]]
-            [logseq.db.frontend.schema :as db-schema]
-            [me.tonsky.persistent-sorted-set :as set :refer [BTSet]]))
+            [shadow.cljs.modern :refer [defclass]]))
 
 (defonce *sqlite worker-state/*sqlite)
 (defonce *sqlite-conns worker-state/*sqlite-conns)
@@ -915,7 +915,8 @@
     (worker-state/set-worker-object! obj)
     (file/<ratelimit-file-writes!)
     (js/setInterval #(.postMessage js/self "keepAliveResponse") (* 1000 25))
-    (Comlink/expose obj)))
+    (Comlink/expose obj)
+    (reset! worker-state/*main-thread (Comlink/wrap js/self))))
 
 (comment
   (defn <remove-all-files!

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

@@ -1,177 +1,301 @@
 (ns frontend.worker.rtc.asset
   "Fns to sync assets.
   some notes:
-  - has :logseq.property.asset/type
-  - block/content, store the asset name
-  - an asset-block not having :file/path indicates need to download asset from server
+  - has :logseq.property.asset/type, :logseq.property.asset/size, :logseq.property.asset/checksum
+  - block/title, store the asset name
   - an asset-block not having :logseq.property.asset/remote-metadata
-    indicates need to upload the asset to server
-  - if an asset-block doesn't have both :file/path and :logseq.property.asset/remote-metadata,
-    it means the other client hasn't uploaded the asset to server
-"
-  (:require [cljs-http.client :as http]
+    indicates need to upload the asset to server"
+  (:require [clojure.set :as set]
             [datascript.core :as d]
             [frontend.common.missionary-util :as c.m]
+            [frontend.worker.rtc.client-op :as client-op]
             [frontend.worker.rtc.log-and-state :as rtc-log-and-state]
             [frontend.worker.rtc.ws-util :as ws-util]
+            [frontend.worker.state :as worker-state]
+            [logseq.common.path :as path]
+            [logseq.db :as ldb]
             [malli.core :as ma]
             [missionary.core :as m])
   (:import [missionary Cancelled]))
 
-(defn get-all-asset-blocks
-  [db]
-  (->> (d/q
-        '[:find (pull ?asset [*])
-          :in $
-          :where
-          [?asset :block/uuid]
-          [?asset :logseq.property.asset/type]]
-        db)
-       (apply concat)))
-
-(defn asset-block->upload+download-action
-  [asset-block]
-  (let [local-file-path (:file/path asset-block)
-        remote-metadata (:logseq.property.asset/remote-metadata asset-block)]
-    (cond
-      (and local-file-path remote-metadata) nil
-      (nil? local-file-path) :download
-      (nil? remote-metadata) :upload)))
-
-(defn get-action->asset-blocks
-  [db]
-  (reduce
-   (fn [action->asset-blocks asset-block]
-     (if-let [action (asset-block->upload+download-action asset-block)]
-       (update action->asset-blocks action (fnil conj #{}) asset-block)
-       action->asset-blocks))
-   {} (get-all-asset-blocks db)))
-
-(defn new-task--upload-assets
-  [get-ws-create-task conn graph-uuid asset-uuids]
-  {:pre [(every? uuid? asset-uuids)]}
+(defn- create-local-updates-check-flow
+  "Return a flow that emits value if need to push local-updates"
+  [repo *auto-push? interval-ms]
+  (let [auto-push-flow (m/watch *auto-push?)
+        clock-flow (c.m/clock interval-ms :clock)
+        merge-flow (m/latest vector auto-push-flow clock-flow)]
+    (m/eduction (filter first)
+                (map second)
+                (filter (fn [v] (when (pos? (client-op/get-unpushed-asset-ops-count repo)) v)))
+                merge-flow)))
+
+(def ^:private remote-asset-updates-schema
+  [:sequential
+   [:map {:closed true}
+    [:op [:enum :update-asset :remove-asset]]
+    [:block/uuid :uuid]
+    [:malli.core/default [:map-of :keyword :any]]]])
+
+(def ^:private *remote-asset-updates (atom nil :validator (ma/validator remote-asset-updates-schema)))
+(def ^:private remote-asset-updates-flow (m/buffer 10 (m/watch *remote-asset-updates)))
+
+(comment
+  (def cancel ((m/reduce (fn [_ v] (prn :v v)) remote-asset-updates-flow) prn prn)))
+
+(defn- new-task--get-asset-file-metadata
+  "Return nil if this asset not exist"
+  [repo block-uuid asset-type]
   (m/sp
-   (when (seq asset-uuids)
-     (let [asset-uuid->url (->> (m/? (ws-util/send&recv get-ws-create-task
-                                                        {:action "get-assets-upload-urls"
-                                                         :graph-uuid graph-uuid
-                                                         :asset-uuid->metadata
-                                                         (into {}
-                                                               (map (fn [asset-uuid] [asset-uuid {"checksum" "TEST-CHECKSUM"}]))
-                                                               asset-uuids)}))
-                                :asset-uuid->url)]
-       (doseq [[asset-uuid put-url] asset-uuid->url]
-         (assert (uuid? asset-uuid) asset-uuid)
-         (let [{:keys [status] :as r}
-               (c.m/<? (http/put put-url {:headers {"x-amz-meta-checksum" "TEST-CHECKSUM"}
-                                          :body (js/JSON.stringify
-                                                 (clj->js {:TEST-ASSET true
-                                                           :asset-uuid (str asset-uuid)
-                                                           :graph-uuid (str graph-uuid)}))
-                                          :with-credentials? false}))]
-           (if (not= 200 status)
-             (prn :debug-failed-upload-asset {:resp r :asset-uuid asset-uuid :graph-uuid graph-uuid})
-
-             (when (some? (d/entity @conn [:block/uuid asset-uuid]))
-               (d/transact! conn [{:block/uuid asset-uuid
-                                   :logseq.property.asset/remote-metadata {:checksum "TEST"}}])))))))))
-
-(defn new-task--download-assets
-  [get-ws-create-task conn graph-uuid asset-uuids]
-  {:pre [(every? uuid? asset-uuids)]}
+    (ldb/read-transit-str
+     (c.m/<?
+      (.get-asset-file-metadata ^js @worker-state/*main-thread repo (str block-uuid) asset-type)))))
+
+(defn- remote-block-ops=>remote-asset-ops
+  [db-before remove-ops]
+  (keep
+   (fn [remove-op]
+     (let [block-uuid (:block-uuid remove-op)]
+       (when-let [ent (d/entity db-before [:block/uuid block-uuid])]
+         (when-let [asset-type (:logseq.property.asset/type ent)]
+           {:op :remove-asset
+            :block/uuid block-uuid
+            :logseq.property.asset/type asset-type}))))
+   remove-ops))
+
+(defn emit-remote-asset-updates-from-block-ops
+  [db-before remove-ops]
+  (when-let [asset-update-ops
+             (not-empty (remote-block-ops=>remote-asset-ops db-before remove-ops))]
+    (reset! *remote-asset-updates asset-update-ops)))
+
+(defn new-task--emit-remote-asset-updates-from-push-asset-upload-updates
+  [repo db push-asset-upload-updates-message]
   (m/sp
-   (when (seq asset-uuids)
-     (let [asset-uuid->url
-           (->> (m/? (ws-util/send&recv get-ws-create-task {:action "get-assets-download-urls"
-                                                            :graph-uuid graph-uuid
-                                                            :asset-uuids asset-uuids}))
-                :asset-uuid->url)]
-       (doseq [[asset-uuid get-url] asset-uuid->url]
-         (assert (uuid? asset-uuid) asset-uuid)
-         (let [{:keys [status _body] :as r} (c.m/<? (http/get get-url {:with-credentials? false}))]
-           (if (not= 200 status)
-             (prn :debug-failed-download-asset {:resp r :asset-uuid asset-uuid :graph-uuid graph-uuid})
-             (when (d/entity @conn [:block/uuid asset-uuid])
-               (d/transact! conn [{:block/uuid asset-uuid
-                                   :file/path "TEST-FILE-PATH"}])
-               (prn :debug-succ-download-asset asset-uuid)))))))))
+    (let [{:keys [uploaded-assets]} push-asset-upload-updates-message]
+      (when-let [asset-update-ops
+                 (->> uploaded-assets
+                      (map
+                       (fn [[asset-uuid remote-metadata]]
+                         (m/sp
+                           (let [ent (d/entity db [:block/uuid asset-uuid])
+                                 asset-type (:logseq.property.asset/type ent)
+                                 local-checksum (:logseq.property.asset/checksum ent)
+                                 remote-checksum (get remote-metadata "checksum")]
+                             (when (or (and local-checksum remote-checksum
+                                            (not= local-checksum remote-checksum))
+                                       (and asset-type
+                                            (nil? (m/? (new-task--get-asset-file-metadata
+                                                        repo asset-uuid asset-type)))))
+                               {:op :update-asset
+                                :block/uuid asset-uuid})))))
+                      (apply m/join vector)
+                      m/?
+                      (remove nil?)
+                      not-empty)]
+        (reset! *remote-asset-updates asset-update-ops)))))
+
+(defn- create-mixed-flow
+  "Return a flow that emits different events:
+  - `:local-update-check`: event to notify check if there're some new local-updates on assets
+  - `:remote-updates`: remote asset updates "
+  [repo *auto-push?]
+  (let [remote-update-flow (m/eduction
+                            (map (fn [v] {:type :remote-updates :value v}))
+                            remote-asset-updates-flow)
+        local-update-check-flow (m/eduction
+                                 (map (fn [v] {:type :local-update-check :value v}))
+                                 (create-local-updates-check-flow repo *auto-push? 2500))]
+    (c.m/mix remote-update-flow local-update-check-flow)))
 
 (defonce ^:private *assets-sync-lock (atom nil))
 (defn- holding-assets-sync-lock
   "Use this to prevent multiple assets-sync loops at same time."
   [started-dfv task]
   (m/sp
-   (when-not (compare-and-set! *assets-sync-lock nil true)
-     (let [e (ex-info "Must not run multiple assets-sync loops"
-                      {:type :assets-sync.exception/lock-failed
-                       :missionary/retry true})]
-       (started-dfv e)
-       (throw e)))
-   (try
-     (m/? task)
-     (finally
-       (reset! *assets-sync-lock nil)))))
-
-(def ^:private asset-change-event-schema
-  [:map-of
-   [:enum :download :upload
-    ;; Why don't need :delete event?
-    ;; when remove-block-op sync to server, server will know this asset need to be deleted
-    ;; :delete
-    ]
-   [:set :uuid]])
-
-(def ^:private asset-change-event-validator (ma/validator asset-change-event-schema))
-
-(defonce *global-asset-change-event (atom nil :validator asset-change-event-validator))
-
-(defonce ^:private global-asset-change-event-flow
-  (m/buffer 20 (m/watch *global-asset-change-event)))
+    (when-not (compare-and-set! *assets-sync-lock nil true)
+      (let [e (ex-info "Must not run multiple assets-sync loops"
+                       {:type :assets-sync.exception/lock-failed
+                        :missionary/retry true})]
+        (started-dfv e)
+        (throw e)))
+    (try
+      (m/? task)
+      (finally
+        (reset! *assets-sync-lock nil)))))
+
+(defn- clean-asset-ops!
+  [repo all-asset-uuids handled-asset-uuids]
+  (doseq [asset-uuid (set/difference (set all-asset-uuids) (set handled-asset-uuids))]
+    (client-op/remove-asset-op repo asset-uuid)))
+
+(defn- new-task--push-local-asset-updates
+  [repo get-ws-create-task conn graph-uuid add-log-fn]
+  (m/sp
+    (when-let [asset-ops (not-empty (client-op/get-all-asset-ops repo))]
+      (let [upload-asset-uuids (keep
+                                (fn [asset-op]
+                                  (when (contains? asset-op :update-asset)
+                                    (:block/uuid asset-op)))
+                                asset-ops)
+            remove-asset-uuids (keep
+                                (fn [asset-op]
+                                  (when (contains? asset-op :remove-asset)
+                                    (:block/uuid asset-op)))
+                                asset-ops)
+            asset-uuid->asset-type+checksum
+            (into {}
+                  (keep
+                   (fn [asset-uuid]
+                     (let [ent (d/entity @conn [:block/uuid asset-uuid])]
+                       (when-let [tp (:logseq.property.asset/type ent)]
+                         (when-let [checksum (:logseq.property.asset/checksum ent)]
+                           [asset-uuid [tp checksum]])))))
+                  upload-asset-uuids)
+            asset-uuid->url
+            (when (seq asset-uuid->asset-type+checksum)
+              (->> (m/? (ws-util/send&recv get-ws-create-task
+                                           {:action "get-assets-upload-urls"
+                                            :graph-uuid graph-uuid
+                                            :asset-uuid->metadata
+                                            (into {}
+                                                  (map (fn [[asset-uuid [asset-type checksum]]]
+                                                         [asset-uuid {"checksum" checksum "type" asset-type}]))
+                                                  asset-uuid->asset-type+checksum)}))
+                   :asset-uuid->url))]
+        (when (seq asset-uuid->url)
+          (add-log-fn :rtc.asset.log/upload-assets {:asset-uuids (keys asset-uuid->url)}))
+        (doseq [[asset-uuid put-url] asset-uuid->url]
+          (let [[asset-type checksum] (get asset-uuid->asset-type+checksum asset-uuid)
+                r (ldb/read-transit-str
+                   (c.m/<?
+                    (.rtc-upload-asset
+                     ^js @worker-state/*main-thread
+                     repo (str asset-uuid) asset-type checksum put-url)))]
+            (when (:ex-data r)
+              (throw (ex-info "upload asset failed" r)))
+            (d/transact! conn
+                         [{:block/uuid asset-uuid
+                           :logseq.property.asset/remote-metadata {:checksum checksum :type asset-type}}]
+                         ;; Don't generate rtc ops again, (block-ops & asset-ops)
+                         {:persist-op? false})
+            (client-op/remove-asset-op repo asset-uuid)))
+        (when (seq remove-asset-uuids)
+          (add-log-fn :rtc.asset.log/remove-assets {:asset-uuids remove-asset-uuids})
+          (m/? (ws-util/send&recv get-ws-create-task
+                                  {:action "delete-assets"
+                                   :graph-uuid graph-uuid
+                                   :asset-uuids remove-asset-uuids}))
+          (doseq [asset-uuid remove-asset-uuids]
+            (client-op/remove-asset-op repo asset-uuid)))
+        (clean-asset-ops! repo
+                          (map :block/uuid asset-ops)
+                          (concat (keys asset-uuid->url) remove-asset-uuids))))))
+
+(defn- new-task--pull-remote-asset-updates
+  [repo get-ws-create-task conn graph-uuid add-log-fn asset-update-ops]
+  (m/sp
+    (when (seq asset-update-ops)
+      (let [update-asset-uuids (keep (fn [op]
+                                       (when (= :update-asset (:op op))
+                                         (:block/uuid op)))
+                                     asset-update-ops)
+            remove-asset-uuid->asset-type
+            (into {} (keep (fn [op]
+                             (when (= :remove-asset (:op op))
+                               [(:block/uuid op) (:logseq.property.asset/type op)])))
+                  asset-update-ops)
+            asset-uuid->asset-type (into {}
+                                         (keep (fn [asset-uuid]
+                                                 (when-let [tp (:logseq.property.asset/type
+                                                                (d/entity @conn [:block/uuid asset-uuid]))]
+                                                   [asset-uuid tp])))
+                                         update-asset-uuids)
+            asset-uuid->url
+            (when (seq asset-uuid->asset-type)
+              (->> (m/? (ws-util/send&recv get-ws-create-task
+                                           {:action "get-assets-download-urls"
+                                            :graph-uuid graph-uuid
+                                            :asset-uuids (keys asset-uuid->asset-type)}))
+                   :asset-uuid->url))]
+        (doseq [[asset-uuid asset-type] remove-asset-uuid->asset-type]
+          (c.m/<? (.unlinkAsset ^js @worker-state/*main-thread repo (str asset-uuid) asset-type)))
+        (when (seq asset-uuid->url)
+          (add-log-fn :rtc.asset.log/download-assets {:asset-uuids (keys asset-uuid->url)}))
+        (doseq [[asset-uuid get-url] asset-uuid->url]
+          (prn :start-download-asset asset-uuid)
+          (let [r (ldb/read-transit-str
+                   (c.m/<?
+                    (.rtc-download-asset
+                     ^js @worker-state/*main-thread
+                     repo (str asset-uuid) (get asset-uuid->asset-type asset-uuid) get-url)))]
+            (when-let [edata (:ex-data r)]
+              ;; if download-url return 404, ignore this asset
+              (when (not= 404 (:status (:data edata)))
+                (throw (ex-info "download asset failed" r))))))))))
+
+(defn- get-all-asset-blocks
+  [db]
+  (d/q '[:find [(pull ?b [:block/uuid
+                          :logseq.property.asset/type
+                          :logseq.property.asset/checksum])
+                ...]
+         :where
+         [?b :block/uuid]
+         [?b :logseq.property.asset/type]]
+       db))
+
+(defn- new-task--initial-download-missing-assets
+  [repo get-ws-create-task graph-uuid conn add-log-fn]
+  (m/sp
+    (let [local-all-asset-file-paths (ldb/read-transit-str
+                                      (c.m/<? (.get-all-asset-file-paths ^js @worker-state/*main-thread repo)))
+          local-all-asset-file-uuids (set (map (comp parse-uuid path/file-stem) local-all-asset-file-paths))
+          local-all-asset-uuids (set (map :block/uuid (get-all-asset-blocks @conn)))]
+      (when-let [asset-update-ops
+                 (not-empty
+                  (map (fn [asset-uuid] {:op :update-asset :block/uuid asset-uuid})
+                       (set/difference local-all-asset-uuids local-all-asset-file-uuids)))]
+        (add-log-fn :rtc.asset.log/initial-download-missing-assets-count {:count (count asset-update-ops)})
+        (m/? (new-task--pull-remote-asset-updates
+              repo get-ws-create-task conn graph-uuid add-log-fn asset-update-ops))))))
 
 (defn create-assets-sync-loop
-  [get-ws-create-task graph-uuid conn]
+  [repo get-ws-create-task graph-uuid conn *auto-push?]
   (let [started-dfv         (m/dfv)
-        asset-change-event-flow global-asset-change-event-flow
         add-log-fn (fn [type message]
                      (assert (map? message) message)
-                     (rtc-log-and-state/rtc-log type (assoc message :graph-uuid graph-uuid)))]
+                     (rtc-log-and-state/rtc-log type (assoc message :graph-uuid graph-uuid)))
+        mixed-flow (create-mixed-flow repo *auto-push?)]
     {:onstarted-task started-dfv
      :assets-sync-loop-task
      (holding-assets-sync-lock
       started-dfv
       (m/sp
-       (try
-         (started-dfv true)
-         (let [action->asset-blocks (get-action->asset-blocks @conn)]
-           (m/?
-            (m/join
-             (constantly nil)
-             (m/sp
-                ;; init phase:
-                ;; generate all asset-change-events from db
-              (when (or (seq (action->asset-blocks :download))
-                        (seq (action->asset-blocks :upload)))
-                (prn "init phase: generate all asset-change-events from db" action->asset-blocks))
-              (m/? (new-task--download-assets
-                    get-ws-create-task conn graph-uuid (map :block/uuid (action->asset-blocks :download))))
-              (m/? (new-task--upload-assets
-                    get-ws-create-task conn graph-uuid (map :block/uuid (action->asset-blocks :upload)))))
-             (->>
-              (let [{asset-uuids-to-download :download
-                     asset-uuids-to-upload :upload} (m/?> asset-change-event-flow)]
-                (m/? (new-task--download-assets get-ws-create-task conn graph-uuid asset-uuids-to-download))
-                (m/? (new-task--upload-assets get-ws-create-task conn graph-uuid asset-uuids-to-upload)))
-              m/ap (m/reduce {} nil)))))
-
-         (catch Cancelled e
-           (add-log-fn :rtc.asset.log/cancelled {})
-           (throw e)))))}))
+        (try
+          (started-dfv true)
+          (m/? (new-task--initial-download-missing-assets repo get-ws-create-task graph-uuid conn add-log-fn))
+          (->>
+           (let [event (m/?> mixed-flow)]
+             (case (:type event)
+               :remote-updates
+               (when-let [asset-update-ops (not-empty (:value event))]
+                 (m/? (new-task--pull-remote-asset-updates
+                       repo get-ws-create-task conn graph-uuid add-log-fn asset-update-ops)))
+               :local-update-check
+               (m/? (new-task--push-local-asset-updates
+                     repo get-ws-create-task conn graph-uuid add-log-fn))))
+           m/ap
+           (m/reduce {} nil)
+           m/?)
+          (catch Cancelled e
+            (add-log-fn :rtc.asset.log/cancelled {})
+            (throw e)))))}))
 
 (comment
   (def x (atom 1))
   (def f (m/ap
-          (let [r (m/?> (m/buffer 10 (m/watch x)))]
-            (m/? (m/sleep 2000))
-            r)))
+           (let [r (m/?> (m/buffer 10 (m/watch x)))]
+             (m/? (m/sleep 2000))
+             r)))
 
   (def cancel ((m/reduce (fn [r e] (prn :e e)) f) prn prn)))

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

@@ -1,38 +1,40 @@
 (ns frontend.worker.rtc.asset-db-listener
   "Listen asset-block changes in db, generate asset-sync operations"
   (:require [datascript.core :as d]
-            [frontend.common.schema-register :as sr]
             [frontend.worker.db-listener :as db-listener]
-            [frontend.worker.rtc.asset :as r.asset]
             [frontend.worker.rtc.client-op :as client-op]
             [logseq.db :as ldb]))
 
-(defn entity-datoms=>action+asset-uuid
-  [db-after entity-datoms]
+(defn- max-t
+  [entity-datoms]
+  (apply max (map (fn [[_e _a _v t]] t) entity-datoms)))
+
+(defn- asset-related-attrs-changed?
+  [entity-datoms]
+  (some (fn [[_e a]] (= :logseq.property.asset/checksum a)) entity-datoms))
+
+(defn- entity-datoms=>ops
+  [db-before db-after entity-datoms]
   (when-let [e (ffirst entity-datoms)]
-    (let [ent (d/entity db-after e)
-          block-uuid (:block/uuid ent)]
-      (when (and block-uuid (ldb/asset? ent))
-        (when-let [action (r.asset/asset-block->upload+download-action ent)]
-          [action block-uuid])))))
+    (let [ent-after (d/entity db-after e)
+          ent-before (d/entity db-before e)]
+      (cond
+        (and (some-> ent-after ldb/asset?)
+             (asset-related-attrs-changed? entity-datoms))
+        [[:update-asset (max-t entity-datoms) {:block-uuid (:block/uuid ent-after)}]]
 
-(defn generate-asset-change-events
-  [db-after same-entity-datoms-coll]
-  (let [action->asset-uuids
-        (->> same-entity-datoms-coll
-             (keep (partial entity-datoms=>action+asset-uuid db-after))
-             (reduce
-              (fn [action->asset-uuids [action asset-uuid]]
-                (update action->asset-uuids action (fnil conj #{}) asset-uuid))
-              {}))]
-    (reset! r.asset/*global-asset-change-event action->asset-uuids)))
+        (and (some-> ent-before ldb/asset?)
+             (nil? ent-after))
+        [[:remove-asset (max-t entity-datoms) {:block-uuid (:block/uuid ent-before)}]]))))
 
-(sr/defkeyword :generate-asset-change-events?
-  "tx-meta option, generate events to notify asset-sync (default true)")
+(defn generate-asset-ops
+  [repo db-before db-after same-entity-datoms-coll]
+  (when-let [ops (not-empty (mapcat (partial entity-datoms=>ops db-before db-after) same-entity-datoms-coll))]
+    (client-op/add-asset-ops repo ops)))
 
 (defmethod db-listener/listen-db-changes :gen-asset-change-events
-  [_ {:keys [_tx-data tx-meta _db-before db-after
+  [_ {:keys [_tx-data tx-meta db-before db-after
              repo _id->attr->datom _e->a->add?->v->t same-entity-datoms-coll]}]
   (when (and (client-op/rtc-db-graph? repo)
-             (:generate-asset-change-events? tx-meta true))
-    (generate-asset-change-events db-after same-entity-datoms-coll)))
+             (:persist-op? tx-meta true))
+    (generate-asset-ops repo db-before db-after same-entity-datoms-coll)))

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

@@ -95,10 +95,17 @@
               ;; [a v] as key for card-many attr, `a` as key for card-one attr
          ]
     (if-not av
-      (sort-by #(nth % 2) (vals r))
+      (vals r)
       (let [[a v _t _add?] av
             av-key (if (card-many-attr? db a) [a v] a)]
-        (recur others (assoc r av-key av))))))
+        (if-let [old-av (get r av-key)]
+          (recur others
+                 (cond
+                   (< (nth old-av 2) (nth av 2)) (assoc r av-key av)
+                   (> (nth old-av 2) (nth av 2)) r
+                   (true? (nth av 3)) (assoc r av-key av)
+                   :else r))
+          (recur others (assoc r av-key av)))))))
 
 (defn- remove-non-exist-ref-av
   "Remove av if its v is ref(block-uuid) and not exist"
@@ -320,7 +327,7 @@
   "Return a task: push local updates"
   [repo conn graph-uuid date-formatter get-ws-create-task add-log-fn]
   (m/sp
-    (let [block-ops-map-coll (client-op/get&remove-all-ops repo)]
+    (let [block-ops-map-coll (client-op/get&remove-all-block-ops repo)]
       (when-let [block-uuid->remote-ops (not-empty (gen-block-uuid->remote-ops @conn block-ops-map-coll))]
         (when-let [ops-for-remote (rtc-const/to-ws-ops-decoder
                                    (sort-remote-ops

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

@@ -40,13 +40,29 @@
      [:t :int]
      [:value [:map
               [:block-uuid :uuid]
-              [:av-coll [:sequential rtc-const/av-schema]]]]]]])
+              [:av-coll [:sequential rtc-const/av-schema]]]]]]
+
+   [:update-asset
+    [:catn
+     [:op :keyword]
+     [:t :int]
+     [:value [:map
+              [:block-uuid :uuid]]]]]
+   [:remove-asset
+    [:catn
+     [:op :keyword]
+     [:t :int]
+     [:value [:map
+              [:block-uuid :uuid]]]]]])
 
 (def ops-schema [:sequential op-schema])
 (def ops-coercer (ma/coercer ops-schema mt/json-transformer nil
                              #(do (prn ::bad-ops (:value %))
                                   (ma/-fail! ::ops-schema %))))
 
+(def ^:private block-op-types #{:move :remove :update-page :remove-page :update})
+(def ^:private asset-op-types #{:update-asset :remove-asset})
+
 (def schema-in-db
   "TODO: rename this db-name from client-op to client-metadata+op.
   and move it to its own namespace."
@@ -158,35 +174,33 @@
     (assert (some? conn) repo)
     (add-ops* conn ops)))
 
-(defn- get-all-op-datoms
-  [conn]
-  (->> (d/datoms @conn :eavt)
-       (group-by :e)))
-
-(defn- get-all-ops*
+(defn- get-all-block-ops*
   "Return e->op-map"
+  [db]
+  (->> (d/datoms db :eavt)
+       (group-by :e)
+       (keep (fn [[e datoms]]
+               (let [op-map (into {}
+                                  (keep (fn [datom]
+                                          (let [a (:a datom)]
+                                            (when (or (= :block/uuid a) (contains? block-op-types a))
+                                              [a (:v datom)]))))
+                                  datoms)]
+                 (when (and (:block/uuid op-map)
+                            ;; count>1 = contains some `block-op-types`
+                            (> (count op-map) 1))
+                   [e op-map]))))
+       (into {})))
+
+(defn get&remove-all-block-ops*
   [conn]
-  (let [e->datoms (get-all-op-datoms conn)]
-    (into {}
-          (keep (fn [[e same-ent-datoms]]
-                  (let [op-map (into {} (map (juxt :a :v)) same-ent-datoms)]
-                    (when (:block/uuid op-map)
-                      [e op-map])))
-                e->datoms))))
-
-(defn get&remove-all-ops*
-  [conn]
-  (let [e->op-map (get-all-ops* conn)
-        retract-all-tx-data (map (fn [e] [:db.fn/retractEntity e]) (keys e->op-map))]
+  (let [e->op-map (get-all-block-ops* @conn)
+        retract-all-tx-data (mapcat (fn [e] (map (fn [a] [:db.fn/retractAttribute e a]) block-op-types))
+                                    (keys e->op-map))]
     (d/transact! conn retract-all-tx-data)
     (vals e->op-map)))
 
-(defn get-all-ops
-  "Return coll of
-  {:block/uuid ...
-   :update ...
-   :move ...
-   ...}"
+(defn get-all-block-ops
   [repo]
   (when-let [conn (worker-state/get-client-ops-conn repo)]
     (mapcat
@@ -194,9 +208,9 @@
        (keep (fn [[k v]]
                (when (not= :block/uuid k) v))
              m))
-     (vals (get-all-ops* conn)))))
+     (vals (get-all-block-ops* @conn)))))
 
-(defn get&remove-all-ops
+(defn get&remove-all-block-ops
   "Return coll of
   {:block/uuid ...
    :update ...
@@ -204,12 +218,12 @@
    ...}"
   [repo]
   (when-let [conn (worker-state/get-client-ops-conn repo)]
-    (get&remove-all-ops* conn)))
+    (get&remove-all-block-ops* conn)))
 
-(defn get-unpushed-ops-count
+(defn get-unpushed-block-ops-count
   [repo]
   (when-let [conn (worker-state/get-client-ops-conn repo)]
-    (count (get-all-op-datoms conn))))
+    (count (get-all-block-ops* @conn))))
 
 (defn rtc-db-graph?
   "Is db-graph & RTC enabled"
@@ -218,11 +232,11 @@
        (or (exists? js/process)
            (some? (get-local-tx repo)))))
 
-(defn create-pending-ops-count-flow
+(defn create-pending-block-ops-count-flow
   [repo]
   (when-let [conn (worker-state/get-client-ops-conn repo)]
     (letfn [(datom-count [db]
-              (count (d/datoms db :avet :block/uuid)))]
+              (count (get-all-block-ops* db)))]
       (m/relieve
        (m/observe
         (fn ctor [emit!]
@@ -232,3 +246,69 @@
           (emit! (datom-count @conn))
           (fn dtor []
             (d/unlisten! conn :create-pending-ops-count-flow))))))))
+
+;;; asset ops
+(defn add-asset-ops
+  [repo asset-ops]
+  (let [conn (worker-state/get-client-ops-conn repo)
+        ops (ops-coercer asset-ops)]
+    (assert (some? conn) repo)
+    (letfn [(already-removed? [remove-op t]
+              (some-> remove-op second (> t)))
+            (update-after-remove? [update-op t]
+              (some-> update-op second (> t)))]
+      (doseq [op ops]
+        (let [[op-type t value] op
+              {:keys [block-uuid]} value
+              exist-block-ops-entity (d/entity @conn [:block/uuid block-uuid])
+              e (:db/id exist-block-ops-entity)]
+          (when-let [tx-data
+                     (not-empty
+                      (case op-type
+                        :update-asset
+                        (let [remove-asset-op (get exist-block-ops-entity :remove-asset)]
+                          (when-not (already-removed? remove-asset-op t)
+                            (cond-> [{:block/uuid block-uuid
+                                      :update-asset op}]
+                              remove-asset-op (conj [:db.fn/retractAttribute e :remove-asset]))))
+                        :remove-asset
+                        (let [update-asset-op (get exist-block-ops-entity :update-asset)]
+                          (when-not (update-after-remove? update-asset-op t)
+                            (cond-> [{:block/uuid block-uuid
+                                      :remove-asset op}]
+                              update-asset-op (conj [:db.fn/retractAttribute e :update-asset]))))))]
+            (d/transact! conn tx-data)))))))
+
+(defn- get-all-asset-ops*
+  [db]
+  (->> (d/datoms db :eavt)
+       (group-by :e)
+       (keep (fn [[e datoms]]
+               (let [op-map (into {}
+                                  (keep (fn [datom]
+                                          (let [a (:a datom)]
+                                            (when (or (= :block/uuid a) (contains? asset-op-types a))
+                                              [a (:v datom)]))))
+                                  datoms)]
+                 (when (and (:block/uuid op-map)
+                            ;; count>1 = contains some `asset-op-types`
+                            (> (count op-map) 1))
+                   [e op-map]))))
+       (into {})))
+
+(defn get-unpushed-asset-ops-count
+  [repo]
+  (when-let [conn (worker-state/get-client-ops-conn repo)]
+    (count (get-all-asset-ops* @conn))))
+
+(defn get-all-asset-ops
+  [repo]
+  (when-let [conn (worker-state/get-client-ops-conn repo)]
+    (vals (get-all-asset-ops* @conn))))
+
+(defn remove-asset-op
+  [repo asset-uuid]
+  (when-let [conn (worker-state/get-client-ops-conn repo)]
+    (let [ent (d/entity @conn [:block/uuid asset-uuid])]
+      (when-let [e (:db/id ent)]
+        (d/transact! conn (map (fn [a] [:db.fn/retractAttribute e a]) asset-op-types))))))

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

@@ -154,6 +154,7 @@
         [:op :keyword]
         [:block-uuid :uuid]]]]]]
    [:asset-uuid->url {:optional true} [:map-of :uuid :string]]
+   [:uploaded-assets {:optional true} [:map-of :uuid :map]]
    [:ex-data {:optional true} [:map [:type :keyword]]]
    [:ex-message {:optional true} :string]])
 
@@ -261,6 +262,12 @@
       [:action :string]
       [:graph-uuid :string]
       [:asset-uuids [:sequential :uuid]]]]
+    ["delete-assets"
+     [:map
+      [:req-id :string]
+      [:action :string]
+      [:graph-uuid :string]
+      [:asset-uuids [:sequential :uuid]]]]
     ["get-user-devices"
      [:map
       [:req-id :string]

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

@@ -27,14 +27,23 @@
 
 (def ^:private sentinel (js-obj))
 (defn- get-remote-updates
-  "Return a flow: receive messages from ws, and filter messages with :req-id=`push-updates` or `online-users-updated`."
+  "Return a flow: receive messages from ws,
+  and filter messages with :req-id=
+  - `push-updates`
+  - `online-users-updated`.
+  - `push-asset-upload-updates`"
   [get-ws-create-task]
   (m/ap
    (loop []
      (let [ws (m/? get-ws-create-task)
            x (try
                (m/?> (m/eduction
-                      (filter (fn [data] (contains? #{"online-users-updated" "push-updates"} (:req-id data))))
+                      (filter (fn [data]
+                                (contains?
+                                 #{"online-users-updated"
+                                   "push-updates"
+                                   "push-asset-upload-updates"}
+                                 (:req-id data))))
                       (ws/recv-flow ws)))
                (catch js/CloseEvent _
                  sentinel))]
@@ -50,7 +59,7 @@
         merge-flow (m/latest vector auto-push-flow clock-flow)]
     (m/eduction (filter first)
                 (map second)
-                (filter (fn [v] (when (pos? (client-op/get-unpushed-ops-count repo)) v)))
+                (filter (fn [v] (when (pos? (client-op/get-unpushed-block-ops-count repo)) v)))
                 merge-flow)))
 
 (defn- create-pull-remote-updates-flow
@@ -76,6 +85,7 @@
 (defn- create-mixed-flow
   "Return a flow that emits all kinds of events:
   `:remote-update`: remote-updates data from server
+  `:remote-asset-update`: remote asset-updates from server
   `:local-update-check`: event to notify to check if there're some new local-updates, then push to remote.
   `:online-users-updated`: online users info updated
   `:pull-remote-updates`: pull remote updates"
@@ -84,7 +94,8 @@
                              (map (fn [data]
                                     (case (:req-id data)
                                       "push-updates" {:type :remote-update :value data}
-                                      "online-users-updated" {:type :online-users-updated :value data})))
+                                      "online-users-updated" {:type :online-users-updated :value data}
+                                      "push-asset-upload-updates" {:type :remote-asset-update :value data})))
                              (get-remote-updates get-ws-create-task))
         local-updates-check-flow (m/eduction
                                   (map (fn [data] {:type :local-update-check :value data}))
@@ -150,7 +161,7 @@
         get-ws-create-task         (r.client/ensure-register-graph-updates
                                     get-ws-create-task graph-uuid repo conn *last-calibrate-t *online-users)
         {:keys [assets-sync-loop-task]}
-        (r.asset/create-assets-sync-loop get-ws-create-task graph-uuid conn)
+        (r.asset/create-assets-sync-loop repo get-ws-create-task graph-uuid conn *auto-push?)
         mixed-flow                 (create-mixed-flow repo get-ws-create-task *auto-push?)]
     (assert (some? *current-ws))
     {:rtc-state-flow     (create-rtc-state-flow (create-ws-state-flow *current-ws))
@@ -161,40 +172,44 @@
      (holding-rtc-lock
       started-dfv
       (m/sp
-       (try
+        (try
           ;; init run to open a ws
-         (m/? get-ws-create-task)
-         (started-dfv true)
-         (reset! *assets-sync-loop-canceler
-                 (c.m/run-task assets-sync-loop-task :assets-sync-loop-task))
-         (->>
-          (let [event (m/?> mixed-flow)]
-            (case (:type event)
-              :remote-update
-              (try (r.remote-update/apply-remote-update graph-uuid repo conn date-formatter event add-log-fn)
-                   (catch :default e
-                     (when (= ::r.remote-update/need-pull-remote-data (:type (ex-data e)))
-                       (m/? (r.client/new-task--pull-remote-data
-                             repo conn graph-uuid date-formatter get-ws-create-task add-log-fn)))))
-
-              :local-update-check
-              (m/? (r.client/new-task--push-local-ops
-                    repo conn graph-uuid date-formatter
-                    get-ws-create-task add-log-fn))
-
-              :online-users-updated
-              (reset! *online-users (:online-users (:value event)))
-
-              :pull-remote-updates
-              (m/? (r.client/new-task--pull-remote-data
-                    repo conn graph-uuid date-formatter get-ws-create-task add-log-fn))))
-          (m/ap)
-          (m/reduce {} nil)
-          (m/?))
-         (catch Cancelled e
-           (when @*assets-sync-loop-canceler (@*assets-sync-loop-canceler))
-           (add-log-fn :rtc.log/cancelled {})
-           (throw e)))))}))
+          (m/? get-ws-create-task)
+          (started-dfv true)
+          (reset! *assets-sync-loop-canceler
+                  (c.m/run-task assets-sync-loop-task :assets-sync-loop-task))
+          (->>
+           (let [event (m/?> mixed-flow)]
+             (case (:type event)
+               :remote-update
+               (try (r.remote-update/apply-remote-update graph-uuid repo conn date-formatter event add-log-fn)
+                    (catch :default e
+                      (when (= ::r.remote-update/need-pull-remote-data (:type (ex-data e)))
+                        (m/? (r.client/new-task--pull-remote-data
+                              repo conn graph-uuid date-formatter get-ws-create-task add-log-fn)))))
+               :remote-asset-update
+               (m/? (r.asset/new-task--emit-remote-asset-updates-from-push-asset-upload-updates
+                     repo @conn (:value event)))
+
+               :local-update-check
+               (m/? (r.client/new-task--push-local-ops
+                     repo conn graph-uuid date-formatter
+                     get-ws-create-task add-log-fn))
+
+               :online-users-updated
+               (reset! *online-users (:online-users (:value event)))
+
+               :pull-remote-updates
+               (m/? (r.client/new-task--pull-remote-data
+                     repo conn graph-uuid date-formatter get-ws-create-task add-log-fn))))
+           (m/ap)
+           (m/reduce {} nil)
+           (m/?))
+          (catch Cancelled e
+            (add-log-fn :rtc.log/cancelled {})
+            (throw e))
+          (finally
+            (when @*assets-sync-loop-canceler (@*assets-sync-loop-canceler))))))}))
 
 (def ^:private empty-rtc-loop-metadata
   {:graph-uuid nil
@@ -294,27 +309,27 @@
 (def ^:private create-get-state-flow
   (let [rtc-loop-metadata-flow (m/watch *rtc-loop-metadata)]
     (m/ap
-     (let [{rtc-lock :*rtc-lock :keys [repo graph-uuid user-uuid rtc-state-flow *rtc-auto-push? *online-users]}
-           (m/?< rtc-loop-metadata-flow)]
-       (try
-         (when (and repo rtc-state-flow *rtc-auto-push? rtc-lock)
-           (m/?<
-            (m/latest
-             (fn [rtc-state rtc-auto-push? rtc-lock online-users pending-local-ops-count local-tx remote-tx]
-               {:graph-uuid graph-uuid
-                :user-uuid user-uuid
-                :unpushed-block-update-count pending-local-ops-count
-                :local-tx local-tx
-                :remote-tx remote-tx
-                :rtc-state rtc-state
-                :rtc-lock rtc-lock
-                :auto-push? rtc-auto-push?
-                :online-users online-users})
-             rtc-state-flow (m/watch *rtc-auto-push?) (m/watch rtc-lock) (m/watch *online-users)
-             (client-op/create-pending-ops-count-flow repo)
-             (rtc-log-and-state/create-local-t-flow graph-uuid)
-             (rtc-log-and-state/create-remote-t-flow graph-uuid))))
-         (catch Cancelled _))))))
+      (let [{rtc-lock :*rtc-lock :keys [repo graph-uuid user-uuid rtc-state-flow *rtc-auto-push? *online-users]}
+            (m/?< rtc-loop-metadata-flow)]
+        (try
+          (when (and repo rtc-state-flow *rtc-auto-push? rtc-lock)
+            (m/?<
+             (m/latest
+              (fn [rtc-state rtc-auto-push? rtc-lock online-users pending-local-ops-count local-tx remote-tx]
+                {:graph-uuid graph-uuid
+                 :user-uuid user-uuid
+                 :unpushed-block-update-count pending-local-ops-count
+                 :local-tx local-tx
+                 :remote-tx remote-tx
+                 :rtc-state rtc-state
+                 :rtc-lock rtc-lock
+                 :auto-push? rtc-auto-push?
+                 :online-users online-users})
+              rtc-state-flow (m/watch *rtc-auto-push?) (m/watch rtc-lock) (m/watch *online-users)
+              (client-op/create-pending-block-ops-count-flow repo)
+              (rtc-log-and-state/create-local-t-flow graph-uuid)
+              (rtc-log-and-state/create-remote-t-flow graph-uuid))))
+          (catch Cancelled _))))))
 
 (defn new-task--get-debug-state
   []

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

@@ -5,7 +5,8 @@
             [frontend.common.schema-register :include-macros true :as sr]
             [frontend.worker.db-listener :as db-listener]
             [frontend.worker.rtc.client-op :as client-op]
-            [logseq.db :as ldb]))
+            [logseq.db :as ldb]
+            [logseq.db.frontend.property :as db-property]))
 
 (defn- latest-add?->v->t
   [add?->v->t]
@@ -26,9 +27,7 @@
     :db/index :db/valueType :db/cardinality})
 
 (def ^:private watched-attr-ns
-  #{"logseq.property" "logseq.property.tldraw" "logseq.property.pdf" "logseq.property.fsrs"
-    "logseq.property.linked-references" "logseq.task" "logseq.property.node" "logseq.property.code"
-    "logseq.class" "logseq.kv"})
+  (conj db-property/logseq-property-namespaces "logseq.class" "logseq.kv"))
 
 (defn- watched-attr?
   [attr]

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

@@ -30,6 +30,12 @@ the server will put it to s3 and return its presigned-url to clients.")
 (sr/defkeyword :rtc.exception/different-graph-skeleton
   "remote graph skeleton data is different from local's.")
 
+(sr/defkeyword :rtc.exception/bad-request-body
+  "bad request body, rejected by server-schema")
+
+(sr/defkeyword :rtc.exception/not-allowed
+  "this api-call is not allowed")
+
 (def ex-remote-graph-not-exist
   (ex-info "remote graph not exist" {:type :rtc.exception/remote-graph-not-exist}))
 
@@ -43,6 +49,12 @@ the server will put it to s3 and return its presigned-url to clients.")
 (def ex-local-not-rtc-graph
   (ex-info "RTC is not supported for this local-graph" {:type :rtc.exception/not-rtc-graph}))
 
+(def ex-bad-request-body
+  (ex-info "bad request body" {:type :rtc.exception/bad-request-body}))
+
+(def ex-not-allowed
+  (ex-info "not allowed" {:type :rtc.exception/not-allowed}))
+
 (def ex-unknown-server-error
   (ex-info "Unknown server error" {:type :rtc.exception/unknown-server-error}))
 

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

@@ -193,7 +193,10 @@
     (merge block
            (update-vals (select-keys block card-one-attrs-in-block)
                         (fn [v]
-                          (if (coll? v) (first v) v))))))
+                          (if (or (sequential? v)
+                                  (set? v))
+                            (first v)
+                            v))))))
 
 (defn- transact-block-refs!
   [repo]
@@ -258,13 +261,13 @@
   [all-blocks repo graph-uuid]
   (let [{:keys [t blocks]} all-blocks
         card-one-attrs (blocks->card-one-attrs blocks)
-        blocks (worker-util/profile :convert-card-one-value-from-value-coll
+        blocks1 (worker-util/profile :convert-card-one-value-from-value-coll
                                     (map (partial convert-card-one-value-from-value-coll card-one-attrs) blocks))
-        blocks (worker-util/profile :normalize-remote-blocks
-                                    (normalized-remote-blocks-coercer blocks))
+        blocks2 (worker-util/profile :normalize-remote-blocks
+                 (normalized-remote-blocks-coercer blocks1))
         ;;TODO: remove this, client/schema already converted to :db/cardinality, :db/valueType by remote,
         ;; and :client/schema should be removed by remote too
-        blocks (map #(dissoc % :client/schema) blocks)
+        blocks (map #(dissoc % :client/schema) blocks2)
         blocks (fill-block-fields blocks)
         [schema-blocks normal-blocks] (blocks->schema-blocks+normal-blocks blocks)
         tx-data (concat

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

@@ -25,7 +25,11 @@
    :rtc.log/apply-remote-update
    :rtc.log/push-local-update
 
-   :rtc.asset.log/cancelled])
+   :rtc.asset.log/cancelled
+   :rtc.asset.log/upload-assets
+   :rtc.asset.log/download-assets
+   :rtc.asset.log/remove-assets
+   :rtc.asset.log/initial-download-missing-assets-count])
 
 (def ^:private rtc-log-type-validator (ma/validator rtc-log-type-schema))
 

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

@@ -6,6 +6,7 @@
             [datascript.core :as d]
             [frontend.common.schema-register :as sr]
             [frontend.worker.handler.page :as worker-page]
+            [frontend.worker.rtc.asset :as r.asset]
             [frontend.worker.rtc.client-op :as client-op]
             [frontend.worker.rtc.const :as rtc-const]
             [frontend.worker.rtc.log-and-state :as rtc-log-and-state]
@@ -14,6 +15,7 @@
             [logseq.clj-fractional-indexing :as index]
             [logseq.common.util :as common-util]
             [logseq.db :as ldb]
+            [logseq.db.frontend.property :as db-property]
             [logseq.db.frontend.property.util :as db-property-util]
             [logseq.graph-parser.whiteboard :as gp-whiteboard]
             [logseq.outliner.batch-tx :as batch-tx]
@@ -276,7 +278,7 @@
 
 (defn- affected-blocks->diff-type-ops
   [repo affected-blocks]
-  (let [unpushed-ops (client-op/get-all-ops repo)
+  (let [unpushed-ops (client-op/get-all-block-ops repo)
         affected-blocks-map* (if unpushed-ops
                                (update-remote-data-by-local-unpushed-ops
                                 affected-blocks unpushed-ops)
@@ -335,12 +337,16 @@
     :property/schema.classes
     :property.value/content})
 
+(def ^:private watched-attr-ns
+  (conj db-property/logseq-property-namespaces "logseq.class" "logseq.kv"))
+
 (defn- update-op-watched-attr?
   [attr]
   (or (contains? update-op-watched-attrs attr)
       (when-let [ns (namespace attr)]
-        (or (= "logseq.task" ns)
-            (string/ends-with? ns ".property")))))
+        (or (contains? watched-attr-ns ns)
+            (string/ends-with? ns ".property")
+            (string/ends-with? ns ".class")))))
 
 (defn- diff-block-kv->tx-data
   [db db-schema e k local-v remote-v]
@@ -359,15 +365,18 @@
       [true false]
       (let [remote-block-uuid (if (coll? remote-v) (first remote-v) remote-v)]
         (when (not= local-v remote-block-uuid)
-          (when-let [db-id (:db/id (d/entity db [:block/uuid remote-block-uuid]))]
-            [[:db/add e k db-id]])))
+          (if (nil? remote-block-uuid)
+            [[:db/retract e k]]
+            (when-let [db-id (:db/id (d/entity db [:block/uuid remote-block-uuid]))]
+              [[:db/add e k db-id]]))))
+
       [false false]
       (let [remote-v* (if (coll? remote-v)
                         (first (map ldb/read-transit-str remote-v))
                         (ldb/read-transit-str remote-v))]
         (when (not= local-v remote-v*)
           (if (nil? remote-v*)
-            [[:db/retract e k local-v]]
+            [[:db/retract e k]]
             [[:db/add e k remote-v*]])))
 
       [false true]
@@ -547,7 +556,8 @@
               sorted-move-ops (move-ops-map->sorted-move-ops move-ops-map)
               update-ops (vals update-ops-map)
               update-page-ops (vals update-page-ops-map)
-              remove-page-ops (vals remove-page-ops-map)]
+              remove-page-ops (vals remove-page-ops-map)
+              db-before @conn]
           (js/console.groupCollapsed "rtc/apply-remote-ops-log")
           (batch-tx/with-batch-tx-mode conn {:rtc-tx? true
                                              :persist-op? false
@@ -561,6 +571,9 @@
           ;; NOTE: we cannot set :persist-op? = true when batch-tx/with-batch-tx-mode (already set to false)
           ;; and there're some transactions in `apply-remote-remove-ops` need to :persist-op?=true
           (worker-util/profile :apply-remote-remove-ops (apply-remote-remove-ops repo conn date-formatter remove-ops))
+          ;; wait all remote-ops transacted into db,
+          ;; then start to check any asset-updates in remote
+          (r.asset/emit-remote-asset-updates-from-block-ops db-before remove-ops)
           (js/console.groupEnd)
 
           (client-op/update-local-tx repo remote-t)

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

@@ -13,7 +13,9 @@
 (defn- handle-remote-ex
   [resp]
   (if-let [e ({:graph-not-exist r.ex/ex-remote-graph-not-exist
-               :graph-not-ready r.ex/ex-remote-graph-not-ready}
+               :graph-not-ready r.ex/ex-remote-graph-not-ready
+               :bad-request-body r.ex/ex-bad-request-body
+               :not-allowed r.ex/ex-not-allowed}
               (:type (:ex-data resp)))]
     (throw e)
     resp))
@@ -58,6 +60,7 @@
 
 (defn get-ws-url
   [token]
+  (assert (some? token))
   (gstring/format @worker-state/*rtc-ws-url token))
 
 (defn- gen-get-ws-create-map

src/main/frontend/worker/state.cljs

@@ -10,6 +10,9 @@
 (sr/defkeyword :undo/repo->page-block-uuid->redo-ops
   "{repo {<page-block-uuid> [op1 op2 ...]}}")
 
+
+(defonce *main-thread (atom nil))
+
 (defonce *state (atom {:worker/object nil
 
                        :db/latest-transact-time {}

src/rtc_e2e_test/client_steps.cljs

@@ -27,7 +27,7 @@
               [[:block/updated-at "[\"~#'\",1724836490810]" true]
                [:block/created-at "[\"~#'\",1724836490810]" true]
                [:block/title "[\"~#'\",\"block1\"]" true]]]}
-           (set (map helper/simplify-client-op (client-op/get-all-ops const/downloaded-test-repo)))))))
+           (set (map helper/simplify-client-op (client-op/get-all-block-ops const/downloaded-test-repo)))))))
    :client2 nil})
 
 (def ^:private step1

src/rtc_e2e_test/helper.cljs

@@ -103,7 +103,7 @@
     (let [r (m/? (m/timeout
                   (m/reduce (fn [_ v]
                               (when (and (= :rtc.log/push-local-update (:type v))
-                                         (empty? (client-op/get-all-ops const/downloaded-test-repo)))
+                                         (empty? (client-op/get-all-block-ops const/downloaded-test-repo)))
                                 (is (nil? (:ex-data v)))
                                 (reduced v)))
                             rtc-log-and-state/rtc-log-flow)

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

@@ -131,7 +131,7 @@
                               :create-first-block? false})
         (is (some? (d/pull @conn '[*] [:block/uuid page-uuid])))
         (is (= {page-uuid #{:update-page :update}}
-               (ops-coll=>block-uuid->op-types (client-op/get&remove-all-ops repo)))))
+               (ops-coll=>block-uuid->op-types (client-op/get&remove-all-block-ops repo)))))
       (testing "add blocks to this page"
         (let [target-entity (d/entity @conn [:block/uuid page-uuid])]
           (batch-tx/with-batch-tx-mode conn
@@ -147,7 +147,7 @@
           (is (=
                {block-uuid1 #{:move :update}
                 block-uuid2 #{:move :update}}
-               (ops-coll=>block-uuid->op-types (client-op/get&remove-all-ops repo))))))
+               (ops-coll=>block-uuid->op-types (client-op/get&remove-all-block-ops repo))))))
 
       (testing "delete a block"
         (batch-tx/with-batch-tx-mode conn
@@ -156,4 +156,4 @@
 
         (is (=
              {block-uuid1 #{:remove}}
-             (ops-coll=>block-uuid->op-types (client-op/get&remove-all-ops repo))))))))
+             (ops-coll=>block-uuid->op-types (client-op/get&remove-all-block-ops repo))))))))

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

@@ -3,11 +3,13 @@
             [datascript.core :as d]
             [frontend.worker.rtc.remote-update :as subject]
             [logseq.db :as ldb]
-            [logseq.db.frontend.schema :as db-schema]))
+            [logseq.db.frontend.schema :as db-schema]
+            [logseq.db.sqlite.create-graph :as sqlite-create-graph]))
 
 (deftest remote-op-value->tx-data-test
   (let [[block-uuid ref-uuid1 ref-uuid2] (repeatedly random-uuid)
-        db (d/empty-db db-schema/schema-for-db-based-graph)]
+        db (d/db-with (d/empty-db db-schema/schema-for-db-based-graph)
+                      (sqlite-create-graph/build-db-initial-data {}))]
     (testing ":block/title"
       (let [db (d/db-with db [{:block/uuid block-uuid
                                :block/title "local-content"}])
@@ -49,6 +51,16 @@
                (set (#'subject/remote-op-value->tx-data db (d/entity db [:block/uuid block-uuid]) op-value))))))
     (testing ":block/updated-at"
       (let [db (d/db-with db [{:block/uuid block-uuid
-                               :block/updated-at 1}])]
-        (is (= [[:db/retract 1 :block/updated-at 1]]
-               (#'subject/remote-op-value->tx-data db (d/entity db [:block/uuid block-uuid]) {})))))))
+                               :block/updated-at 1}])
+            ent (d/entity db [:block/uuid block-uuid])]
+        (is (= [[:db/retract (:db/id ent) :block/updated-at]]
+               (#'subject/remote-op-value->tx-data db ent {})))))
+    (testing ":logseq.task/status, op-value don't have this attr, means remove this attr"
+      (let [db (d/db-with db [{:db/id "ref1"
+                               :block/uuid ref-uuid1}
+                              {:block/uuid block-uuid
+                               :logseq.task/status "ref1"}])
+            op-value {}
+            ent (d/entity db [:block/uuid block-uuid])]
+        (is (= [[:db/retract (:db/id ent) :logseq.task/status]]
+               (#'subject/remote-op-value->tx-data db ent op-value)))))))