enhance: validate :file/writes chan

src/main/frontend/fs/sync.cljs

@@ -2701,7 +2701,7 @@
                     v)))))
        :flush-fn #(swap! *sync-state sync-state-reset-queued-local->remote-files)
        :stop-ch stop-chan
-       :distinct-coll? true
+       :distinct-key-fn identity
        :flush-now-ch private-immediately-local->remote-chan
        :refresh-timeout-ch private-recent-edited-chan)))
 

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

@@ -101,17 +101,17 @@
 (defn <ratelimit-file-writes!
   []
   (util/<ratelimit (state/get-file-write-chan) batch-write-interval
-                 :filter-fn
-                 (fn [[repo _ _ time]]
-                   (swap! *writes-finished? assoc repo {:time time
-                                                        :value false})
-                   true)
-                 :flush-fn
-                 (fn [col]
-                   (let [start-time (tc/to-long (t/now))
-                         repos (distinct (map first col))]
-                     (write-files! col)
-                     (doseq [repo repos]
-                       (let [last-write-time (get-in @*writes-finished? [repo :time])]
-                         (when (> start-time last-write-time)
-                           (swap! *writes-finished? assoc repo {:value true}))))))))
+                   :filter-fn
+                   (fn [[repo _ _ time]]
+                     (swap! *writes-finished? assoc repo {:time time
+                                                          :value false})
+                     true)
+                   :flush-fn
+                   (fn [col]
+                     (let [start-time (tc/to-long (t/now))
+                           repos (distinct (map first col))]
+                       (write-files! col)
+                       (doseq [repo repos]
+                         (let [last-write-time (get-in @*writes-finished? [repo :time])]
+                           (when (> start-time last-write-time)
+                             (swap! *writes-finished? assoc repo {:value true}))))))))

src/main/frontend/state.cljs

@@ -9,13 +9,14 @@
             [electron.ipc :as ipc]
             [frontend.colors :as colors]
             [frontend.mobile.util :as mobile-util]
-            [frontend.storage :as storage]
             [frontend.spec.storage :as storage-spec]
+            [frontend.storage :as storage]
             [frontend.util :as util]
             [frontend.util.cursor :as cursor]
             [goog.dom :as gdom]
             [goog.object :as gobj]
             [logseq.graph-parser.config :as gp-config]
+            [malli.core :as m]
             [medley.core :as medley]
             [promesa.core :as p]
             [rum.core :as rum]))
@@ -31,7 +32,12 @@
      :today                                 nil
      :system/events                         (async/chan 1000)
      :db/batch-txs                          (async/chan 1000)
-     :file/writes                           (async/chan 10000)
+     :file/writes                           (let [coercer (m/coercer [:catn
+                                                                      [:repo :string]
+                                                                      [:page-id :any]
+                                                                      [:outliner-op :any]
+                                                                      [:epoch :int]])]
+                                              (async/chan 10000 (map coercer)))
      :file/unlinked-dirs                    #{}
      :reactive/custom-queries               (async/chan 1000)
      :notification/show?                    false

src/main/frontend/util.cljc

@@ -1125,11 +1125,11 @@
                will poll it when its return value is channel,
   - :flush-fn exec flush-fn when time to flush, (flush-fn item-coll)
   - :stop-ch stop go-loop when stop-ch closed
-  - :distinct-coll? distinct coll when put into CH
+  - :distinct-key-fn distinct coll when put into CH
   - :chan-buffer buffer of return CH, default use (async/chan 1000)
   - :flush-now-ch flush the content in the queue immediately
   - :refresh-timeout-ch refresh (timeout max-duration)"
-     [in-ch max-duration & {:keys [filter-fn flush-fn stop-ch distinct-coll? chan-buffer flush-now-ch refresh-timeout-ch]}]
+     [in-ch max-duration & {:keys [filter-fn flush-fn stop-ch distinct-key-fn chan-buffer flush-now-ch refresh-timeout-ch]}]
      (let [ch (if chan-buffer (async/chan chan-buffer) (async/chan 1000))
            stop-ch* (or stop-ch (async/chan))
            flush-now-ch* (or flush-now-ch (async/chan))
@@ -1157,8 +1157,8 @@
                                (async/<! filter-v)
                                filter-v)]
                (if filter-v*
-                 (recur timeout-ch (cond-> (conj coll e)
-                                     distinct-coll? distinct
+                 (recur timeout-ch (cond->> (conj coll e)
+                                     distinct-key-fn (distinct-by distinct-key-fn)
                                      true vec))
                  (recur timeout-ch coll)))