♻️ Refactor srepl helpers

backend/src/app/http/debug.clj

@@ -413,13 +413,8 @@
                 :code :invalid-version
                 :hint "provided invalid version"))
 
-    (binding [srepl/*system* cfg]
+    (db/tx-run! cfg srepl/process-file! file-id #(update % :data assoc :version version))
-      (srepl/process-file! :id file-id
+
-                           :update-fn (fn [file]
-                                        (update file :data assoc :version version))
-                           :migrate? false
-                           :inc-revn? false
-                           :save? true))
     {::rres/status  200
      ::rres/headers {"content-type" "text/plain"}
      ::rres/body    "OK"}))

backend/src/app/rpc/commands/files_snapshot.clj

@@ -147,8 +147,9 @@
                       (restore-file-snapshot! cfg params)))))
 
 (defn take-file-snapshot!
-  [{:keys [::db/conn]} {:keys [file-id label]}]
+  [cfg {:keys [file-id label]}]
-  (let [file  (db/get conn :file {:id file-id})
+  (let [conn  (db/get-connection cfg)
+        file  (db/get conn :file {:id file-id})
         id    (uuid/next)]
 
     (l/debug :hint "creating file snapshot"

backend/src/app/srepl/fixes.clj

@@ -0,0 +1,94 @@
+;; This Source Code Form is subject to the terms of the Mozilla Public
+;; License, v. 2.0. If a copy of the MPL was not distributed with this
+;; file, You can obtain one at http://mozilla.org/MPL/2.0/.
+;;
+;; Copyright (c) KALEIDOS INC
+
+(ns app.srepl.fixes
+  "A misc of fix functions"
+  (:refer-clojure :exclude [parse-uuid])
+  (:require
+   [app.binfile.common :as bfc]
+   [app.common.data :as d]
+   [app.common.files.changes :as cpc]
+   [app.common.files.repair :as cfr]
+   [app.common.files.validate :as cfv]
+   [app.common.logging :as l]
+   [app.common.uuid :as uuid]
+   [app.db :as db]
+   [app.srepl.helpers :as h]))
+
+(defn repair-file-media
+  "A helper intended to be used with `srepl.main/process-files!` that
+  fixes all not propertly referenced file-media-object for a file"
+  [{:keys [id data] :as file} & _]
+  (let [conn  (db/get-connection h/*system*)
+        used  (bfc/collect-used-media data)
+        ids   (db/create-array conn "uuid" used)
+        sql   (str "SELECT * FROM file_media_object WHERE id = ANY(?)")
+        rows  (db/exec! conn [sql ids])
+        index (reduce (fn [index media]
+                        (if (not= (:file-id media) id)
+                          (let [media-id (uuid/next)]
+                            (l/wrn :hint "found not referenced media"
+                                   :file-id (str id)
+                                   :media-id (str (:id media)))
+
+                            (db/insert! conn :file-media-object
+                                        (-> media
+                                            (assoc :file-id id)
+                                            (assoc :id media-id)))
+                            (assoc index (:id media) media-id))
+                          index))
+                      {}
+                      rows)]
+
+    (when (seq index)
+      (binding [bfc/*state* (atom {:index index})]
+        (update file :data (fn [fdata]
+                             (-> fdata
+                                 (update :pages-index #'bfc/relink-shapes)
+                                 (update :components #'bfc/relink-shapes)
+                                 (update :media #'bfc/relink-media)
+                                 (d/without-nils))))))))
+
+
+(defn repair-file
+  "Internal helper for validate and repair the file. The operation is
+  applied multiple times untile file is fixed or max iteration counter
+  is reached (default 10)"
+  [file libs & {:keys [max-iterations] :or {max-iterations 10}}]
+
+  (let [validate-and-repair
+        (fn [file libs iteration]
+          (when-let [errors (not-empty (cfv/validate-file file libs))]
+            (l/trc :hint "repairing file"
+                   :file-id (str (:id file))
+                   :iteration iteration
+                   :errors (count errors))
+            (let [changes (cfr/repair-file file libs errors)]
+              (-> file
+                  (update :revn inc)
+                  (update :data cpc/process-changes changes)))))
+
+        process-file
+        (fn [file libs]
+          (loop [file      file
+                 iteration 0]
+            (if (< iteration max-iterations)
+              (if-let [file (validate-and-repair file libs iteration)]
+                (recur file (inc iteration))
+                file)
+              (do
+                (l/wrn :hint "max retry num reached on repairing file"
+                       :file-id (str (:id file))
+                       :iteration iteration)
+                file))))
+
+        file'
+        (process-file file libs)]
+
+    (when (not= (:revn file) (:revn file'))
+      (l/trc :hint "file repaired" :file-id (str (:id file))))
+
+    file'))

backend/src/app/srepl/helpers.clj

@@ -7,39 +7,18 @@
 (ns app.srepl.helpers
   "A  main namespace for server repl."
   (:refer-clojure :exclude [parse-uuid])
-  #_:clj-kondo/ignore
   (:require
-   [app.binfile.common :as bfc]
    [app.common.data :as d]
-   [app.common.exceptions :as ex]
-   [app.common.features :as cfeat]
-   [app.common.files.changes :as cpc]
    [app.common.files.migrations :as fmg]
    [app.common.files.validate :as cfv]
-   [app.common.logging :as l]
-   [app.common.pprint :refer [pprint]]
-   [app.common.spec :as us]
-   [app.common.uuid :as uuid]
-   [app.config :as cfg]
    [app.db :as db]
-   [app.db.sql :as sql]
+   [app.features.components-v2 :as feat.comp-v2]
    [app.features.fdata :as feat.fdata]
    [app.main :as main]
    [app.rpc.commands.files :as files]
-   [app.rpc.commands.files-update :as files-update]
+   [app.rpc.commands.files-snapshot :as fsnap]
    [app.util.blob :as blob]
-   [app.util.objects-map :as omap]
+   [app.util.pointer-map :as pmap]))
-   [app.util.pointer-map :as pmap]
-   [app.util.time :as dt]
-   [clojure.spec.alpha :as s]
-   [clojure.stacktrace :as strace]
-   [clojure.walk :as walk]
-   [cuerdas.core :as str]
-   [expound.alpha :as expound]
-   [promesa.core :as p]
-   [promesa.exec :as px]
-   [promesa.exec.semaphore :as ps]
-   [promesa.util :as pu]))
 
 (def ^:dynamic *system* nil)
 
@@ -54,25 +33,6 @@
     (d/parse-uuid v)
     v))
 
-;; (def ^:private sql:get-and-lock-team-files
-;;   "SELECT f.id
-;;      FROM file AS f
-;;      JOIN project AS p ON (p.id = f.project_id)
-;;     WHERE p.team_id = ?
-;;       FOR UPDATE")
-
-;; (defn get-and-lock-team-files
-;;   [conn team-id]
-;;   (->> (db/exec! conn [sql:get-and-lock-team-files team-id])
-;;        (into #{} (map :id))))
-
-;; (defn get-team
-;;   [system team-id]
-;;   (-> (db/get system :team {:id team-id}
-;;               {::db/remove-deleted false
-;;                ::db/check-deleted false})
-;;       (update :features db/decode-pgarray #{})))
-
 (defn get-file
   "Get the migrated data of one file."
   ([id] (get-file (or *system* main/system) id))
@@ -107,8 +67,10 @@
                 {:revn (:revn file)
                  :data (:data file)
                  :features (:features file)
+                 :deleted-at (:deleted-at file)
+                 :created-at (:created-at file)
+                 :modified-at (:modified-at file)
                  :data-backend nil
-                 :modified-at (dt/now)
                  :has-media-trimmed false}
                 {:id (:id file)})))
 
@@ -125,213 +87,90 @@
 
 (defn reset-file-data!
   "Hardcode replace of the data of one file."
-  [id data]
+  [system id data]
-  (db/tx-run! main/system
+  (db/tx-run! system
               (fn [system]
                 (db/update! system :file
                             {:data data}
                             {:id id}))))
 
-(defn process-file*
-  [system file-id update-fn]
-  (let [file (get-file system file-id)
-        file (-> (update-fn file)
-                 (update :revn inc))]
 
-    (cfv/validate-file-schema! file)
+(def ^:private sql:snapshots-with-file
-    (update-file! system file)
+  "WITH files AS (
-    (dissoc file :data)))
+     SELECT f.id AS file_id,
+            (SELECT fc.id
+               FROM file_change AS fc
+              WHERE fc.label = ?
+                AND fc.file_id = f.id
+              ORDER BY fc.created_at DESC
+              LIMIT 1) AS id
+       FROM file AS f
+   ) SELECT * FROM files
+      WHERE file_id = ANY(?)
+        AND id IS NOT NULL")
+
+(defn get-file-snapshots
+  "Get a seq parirs of file-id and snapshot-id for a set of files
+   and specified label"
+  [conn label ids]
+  (db/exec! conn [sql:snapshots-with-file label
+                  (db/create-array conn "uuid" ids)]))
+
+(defn take-team-snapshot!
+  [system team-id label]
+  (let [conn (db/get-connection system)]
+    (->> (feat.comp-v2/get-and-lock-team-files conn team-id)
+         (map (fn [file-id]
+                {:file-id file-id
+                 :label label}))
+         (reduce (fn [result params]
+                   (fsnap/take-file-snapshot! conn params)
+                   (inc result))
+                 0))))
+
+(defn restore-team-snapshot!
+  [system team-id label]
+  (let [conn (db/get-connection system)
+        ids  (->> (feat.comp-v2/get-and-lock-team-files conn team-id)
+                  (into #{}))
+
+        snap (get-file-snapshots conn label ids)
+
+        ids' (into #{} (map :file-id) snap)
+        team (-> (feat.comp-v2/get-team conn team-id)
+                 (update :features disj "components/v2"))]
+
+    (when (not= ids ids')
+      (throw (RuntimeException. "no uniform snapshot available")))
+
+    (feat.comp-v2/update-team! conn team)
+    (reduce (fn [result params]
+              (fsnap/restore-file-snapshot! conn params)
+              (inc result))
+            0
+            snap)))
 
 (defn process-file!
-  "Apply a function to the data of one file. Optionally save the changes or not.
+  [system file-id update-fn & {:keys [label validate? with-libraries?] :or {validate? true} :as opts}]
-  The function receives the decoded and migrated file data."
-  [& {:keys [update-fn id rollback?]
-      :or {rollback? true}}]
 
-  (let [system (or *system* (assoc main/system ::db/rollback rollback?))]
+  (when (string? label)
-    (db/tx-run! system
+    (fsnap/take-file-snapshot! system {:file-id file-id :label label}))
-                (fn [system]
-                  (binding [*system* system]
-                    (process-file* system id update-fn))))))
 
+  (let [conn  (db/get-connection system)
+        file  (get-file system file-id)
+        libs  (when with-libraries?
+                (->> (files/get-file-libraries conn file-id)
+                     (into [file] (map (fn [{:keys [id]}]
+                                         (get-file system id))))
+                     (d/index-by :id)))
 
-(def ^:private sql:get-file-ids
+        file' (if with-libraries?
-  "SELECT id FROM file
+                (update-fn file libs opts)
-    WHERE created_at < ? AND deleted_at is NULL
+                (update-fn file opts))]
-    ORDER BY created_at DESC")
 
-(defn analyze-files
+    (when (and (some? file)
-  "Apply a function to all files in the database, reading them in
+               (not (identical? file file')))
-  batches. Do not change data.
+      (when validate? (cfv/validate-file-schema! file'))
-
+      (let [file' (update file' :revn inc)]
-  The `on-file` parameter should be a function that receives the file
+        (update-file! system file')
-  and the previous state and returns the new state.
+        true))))
-
-  Emits rollback at the end of operation."
-  [& {:keys [max-items start-at on-file on-error on-end on-init with-libraries?]}]
-  (letfn [(get-candidates [conn]
-            (cond->> (db/cursor conn [sql:get-file-ids (or start-at (dt/now))])
-              (some? max-items)
-              (take max-items)))
-
-          (on-error* [cause file]
-            (println "unexpected exception happened on processing file: " (:id file))
-            (strace/print-stack-trace cause))
-
-          (process-file [{:keys [::db/conn] :as system} file-id]
-            (let [file (get-file system file-id)
-                  libs (when with-libraries?
-                         (->> (files/get-file-libraries conn file-id)
-                              (into [file] (map (fn [{:keys [id]}]
-                                                  (get-file system id))))
-                              (d/index-by :id)))]
-              (try
-                (if with-libraries?
-                  (on-file file libs)
-                  (on-file file))
-                (catch Throwable cause
-                  ((or on-error on-error*) cause file)))))]
-
-    (db/tx-run! (assoc main/system ::db/rollback true)
-                (fn [{:keys [::db/conn] :as system}]
-                  (try
-                    (binding [*system* system]
-                      (when (fn? on-init) (on-init))
-                      (run! (partial process-file system)
-                            (get-candidates conn)))
-                    (finally
-                      (when (fn? on-end)
-                        (ex/ignoring (on-end)))))))))
-
-(defn process-files!
-  "Apply a function to all files in the database"
-  [& {:keys [max-items
-             max-jobs
-             start-at
-             on-file
-             validate?
-             rollback?]
-      :or {max-jobs 1
-           max-items Long/MAX_VALUE
-           validate? true
-           rollback? true}}]
-
-  (l/dbg :hint "process:start"
-         :rollback rollback?
-         :max-jobs max-jobs
-         :max-items max-items)
-
-  (let [tpoint    (dt/tpoint)
-        factory   (px/thread-factory :virtual false :prefix "penpot/file-process/")
-        executor  (px/cached-executor :factory factory)
-        sjobs     (ps/create :permits max-jobs)
-
-        process-file
-        (fn [file-id idx tpoint]
-          (try
-            (l/trc :hint "process:file:start" :file-id (str file-id) :index idx)
-            (db/tx-run! (assoc main/system ::db/rollback rollback?)
-                        (fn [{:keys [::db/conn] :as system}]
-                          (let [file' (get-file system file-id)
-                                file  (binding [*system* system]
-                                        (on-file file'))]
-
-                            (when (and (some? file) (not (identical? file file')))
-
-                              (when validate?
-                                (cfv/validate-file-schema! file))
-
-                              (let [file (if (contains? (:features file) "fdata/objects-map")
-                                           (feat.fdata/enable-objects-map file)
-                                           file)
-
-                                    file (if (contains? (:features file) "fdata/pointer-map")
-                                           (binding [pmap/*tracked* (pmap/create-tracked)]
-                                             (let [file (feat.fdata/enable-pointer-map file)]
-                                               (feat.fdata/persist-pointers! system file-id)
-                                               file))
-                                           file)]
-
-                                (db/update! conn :file
-                                            {:data (blob/encode (:data file))
-                                             :deleted-at (:deleted-at file)
-                                             :created-at (:created-at file)
-                                             :modified-at (:modified-at file)
-                                             :features (db/create-array conn "text" (:features file))
-                                             :revn (:revn file)}
-                                            {:id file-id}))))))
-            (catch Throwable cause
-              (l/wrn :hint "unexpected error on processing file (skiping)"
-                     :file-id (str file-id)
-                     :index idx
-                     :cause cause))
-            (finally
-              (ps/release! sjobs)
-              (let [elapsed (dt/format-duration (tpoint))]
-                (l/trc :hint "process:file:end"
-                       :file-id (str file-id)
-                       :index idx
-                       :elapsed elapsed)))))]
-
-    (try
-      (db/tx-run! main/system
-                  (fn [{:keys [::db/conn] :as system}]
-                    (db/exec! conn ["SET statement_timeout = 0"])
-                    (db/exec! conn ["SET idle_in_transaction_session_timeout = 0"])
-
-                    (try
-                      (reduce (fn [idx file-id]
-                                (ps/acquire! sjobs)
-                                (px/run! executor (partial process-file file-id idx (dt/tpoint)))
-                                (inc idx))
-                              0
-                              (->> (db/cursor conn [sql:get-file-ids (or start-at (dt/now))])
-                                   (take max-items)
-                                   (map :id)))
-                      (finally
-                        ;; Close and await tasks
-                        (pu/close! executor)))))
-
-      (catch Throwable cause
-        (l/dbg :hint "process:error" :cause cause))
-
-      (finally
-        (let [elapsed (dt/format-duration (tpoint))]
-          (l/dbg :hint "process:end"
-                 :rollback rollback?
-                 :elapsed elapsed))))))
-
-
-(defn repair-file-media
-  "A helper intended to be used with `process-files!` that fixes all
-  not propertly referenced file-media-object for a file"
-  [{:keys [id data] :as file}]
-  (let [conn  (db/get-connection *system*)
-        used  (bfc/collect-used-media data)
-        ids   (db/create-array conn "uuid" used)
-        sql   (str "SELECT * FROM file_media_object WHERE id = ANY(?)")
-        rows  (db/exec! conn [sql ids])
-        index (reduce (fn [index media]
-                        (if (not= (:file-id media) id)
-                          (let [media-id (uuid/next)]
-                            (l/wrn :hint "found not referenced media"
-                                   :file-id (str id)
-                                   :media-id (str (:id media)))
-
-                            (db/insert! *system* :file-media-object
-                                        (-> media
-                                            (assoc :file-id id)
-                                            (assoc :id media-id)))
-                            (assoc index (:id media) media-id))
-                          index))
-                      {}
-                      rows)]
-
-    (when (seq index)
-      (binding [bfc/*state* (atom {:index index})]
-        (update file :data (fn [fdata]
-                             (-> fdata
-                                 (update :pages-index #'bfc/relink-shapes)
-                                 (update :components #'bfc/relink-shapes)
-                                 (update :media #'bfc/relink-media)
-                                 (d/without-nils))))))))

backend/src/app/srepl/main.clj

@@ -12,9 +12,8 @@
    [app.binfile.common :as bfc]
    [app.common.data :as d]
    [app.common.data.macros :as dm]
+   [app.common.exceptions :as ex]
    [app.common.features :as cfeat]
-   [app.common.files.changes :as cpc]
-   [app.common.files.repair :as cfr]
    [app.common.files.validate :as cfv]
    [app.common.logging :as l]
    [app.common.pprint :as p]
@@ -31,17 +30,19 @@
    [app.rpc.commands.files-snapshot :as fsnap]
    [app.rpc.commands.management :as mgmt]
    [app.rpc.commands.profile :as profile]
-   [app.srepl.cli :as cli]
+   [app.srepl.fixes :as fixes]
    [app.srepl.helpers :as h]
-   [app.storage :as sto]
    [app.util.blob :as blob]
-   [app.util.objects-map :as omap]
    [app.util.pointer-map :as pmap]
    [app.util.time :as dt]
    [app.worker :as wrk]
-   [clojure.pprint :refer [pprint print-table]]
+   [clojure.pprint :refer [print-table]]
+   [clojure.stacktrace :as strace]
    [clojure.tools.namespace.repl :as repl]
-   [cuerdas.core :as str]))
+   [cuerdas.core :as str]
+   [promesa.exec :as px]
+   [promesa.exec.semaphore :as ps]
+   [promesa.util :as pu]))
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; TASKS
@@ -138,24 +139,20 @@
 ;; FEATURES
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
+(declare process-file!)
+
 (defn enable-objects-map-feature-on-file!
-  [& {:keys [save? id]}]
+  [file-id & {:as opts}]
-  (h/process-file! main/system
+  (process-file! file-id feat.fdata/enable-objects-map opts))
-                   :id id
-                   :update-fn feat.fdata/enable-objects-map
-                   :save? save?))
 
 (defn enable-pointer-map-feature-on-file!
-  [& {:keys [save? id]}]
+  [file-id & {:as opts}]
-  (h/process-file! main/system
+  (process-file! file-id feat.fdata/enable-pointer-map opts))
-                   :id id
-                   :update-fn feat.fdata/enable-pointer-map
-                   :save? save?))
 
 (defn enable-storage-features-on-file!
-  [& {:as params}]
+  [file-id & {:as opts}]
-  (enable-objects-map-feature-on-file! main/system params)
+  (enable-objects-map-feature-on-file! file-id opts)
-  (enable-pointer-map-feature-on-file! main/system params))
+  (enable-pointer-map-feature-on-file! file-id opts))
 
 (defn enable-team-feature!
   [team-id feature]
@@ -308,84 +305,40 @@
     (db/tx-run! main/system fsnap/take-file-snapshot! {:file-id file-id :label label})))
 
 (defn restore-file-snapshot!
-  [& {:keys [file-id id]}]
+  [file-id label]
-  (db/tx-run! main/system
+  (let [file-id (h/parse-uuid file-id)]
-              (fn [cfg]
+    (db/tx-run! main/system
-                (let [file-id (h/parse-uuid file-id)
+                (fn [{:keys [::db/conn] :as system}]
-                      id      (h/parse-uuid id)]
+                  (when-let [snapshot (->> (h/get-file-snapshots conn label #{file-id})
-                  (if (and (uuid? id) (uuid? file-id))
+                                           (map :id)
-                    (fsnap/restore-file-snapshot! cfg {:id id :file-id file-id})
+                                           (first))]
-                    (println "=> invalid parameters"))))))
+                    (fsnap/restore-file-snapshot! system
-
+                                                  {:id (:id snapshot)
+                                                   :file-id file-id}))))))
 
 (defn list-file-snapshots!
-  [& {:keys [file-id limit]}]
+  [file-id & {:keys [limit]}]
-  (db/tx-run! main/system
+  (let [file-id (h/parse-uuid file-id)]
-              (fn [system]
+    (db/tx-run! main/system
-                (let [params {:file-id (h/parse-uuid file-id)
+                (fn [system]
-                              :limit limit}]
+                  (let [params {:file-id file-id :limit limit}]
-                  (->> (fsnap/get-file-snapshots system (d/without-nils params))
+                    (->> (fsnap/get-file-snapshots system (d/without-nils params))
-                       (print-table [:id :revn :created-at :label]))))))
+                         (print-table [:label :id :revn :created-at])))))))
 
 (defn take-team-snapshot!
-  [& {:keys [team-id label rollback?]
+  [team-id & {:keys [label rollback?] :or {rollback? true}}]
-      :or {rollback? true}}]
   (let [team-id (h/parse-uuid team-id)
-        label   (or label (fsnap/generate-snapshot-label))
+        label   (or label (fsnap/generate-snapshot-label))]
-
-        take-snapshot
-        (fn [{:keys [::db/conn] :as system}]
-          (->> (feat.comp-v2/get-and-lock-team-files conn team-id)
-               (map (fn [file-id]
-                      {:file-id file-id
-                       :label label}))
-               (run! (partial fsnap/take-file-snapshot! system))))]
-
     (-> (assoc main/system ::db/rollback rollback?)
-        (db/tx-run! take-snapshot))))
+        (db/tx-run! h/take-team-snapshot! team-id label))))
-
-(def ^:private sql:snapshots-with-file
-  "WITH files AS (
-     SELECT f.id AS file_id,
-            (SELECT fc.id
-               FROM file_change AS fc
-              WHERE fc.label = ?
-                AND fc.file_id = f.id
-              ORDER BY fc.created_at DESC
-              LIMIT 1) AS id
-       FROM file AS f
-   ) SELECT * FROM files
-      WHERE file_id = ANY(?)
-        AND id IS NOT NULL")
 
 (defn restore-team-snapshot!
   "Restore a snapshot on all files of the team. The snapshot should
   exists for all files; if is not the case, an exception is raised."
-  [& {:keys [team-id label rollback?] :or {rollback? true}}]
+  [team-id label & {:keys [rollback?] :or {rollback? true}}]
-  (let [team-id (h/parse-uuid team-id)
+  (let [team-id (h/parse-uuid team-id)]
-
-        get-file-snapshots
-        (fn [conn ids]
-          (db/exec! conn [sql:snapshots-with-file label
-                          (db/create-array conn "uuid" ids)]))
-
-        restore-snapshot
-        (fn [{:keys [::db/conn] :as system}]
-          (let [ids  (->> (feat.comp-v2/get-and-lock-team-files conn team-id)
-                          (into #{}))
-                snap (get-file-snapshots conn ids)
-                ids' (into #{} (map :file-id) snap)
-                team (-> (feat.comp-v2/get-team conn team-id)
-                         (update :features disj "components/v2"))]
-
-            (when (not= ids ids')
-              (throw (RuntimeException. "no uniform snapshot available")))
-
-            (feat.comp-v2/update-team! conn team)
-            (run! (partial fsnap/restore-file-snapshot! system) snap)))]
-
     (-> (assoc main/system ::db/rollback rollback?)
-        (db/tx-run! restore-snapshot))))
+        (db/tx-run! h/restore-team-snapshot! team-id label))))
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; FILE VALIDATION & REPAIR
@@ -394,68 +347,163 @@
 (defn validate-file
   "Validate structure, referencial integrity and semantic coherence of
   all contents of a file. Returns a list of errors."
-  [id]
+  [file-id]
-  (db/tx-run! main/system
+  (let [file-id (h/parse-uuid file-id)]
-              (fn [{:keys [::db/conn] :as system}]
+    (db/tx-run! (assoc main/system ::db/rollback true)
-                (let [id   (if (string? id) (parse-uuid id) id)
-                      file (h/get-file system id)
-                      libs (->> (files/get-file-libraries conn id)
-                                (into [file] (map (fn [{:keys [id]}]
-                                                    (h/get-file system id))))
-                                (d/index-by :id))]
-                  (cfv/validate-file file libs)))))
-
-(defn- repair-file*
-  "Internal helper for validate and repair the file. The operation is
-  applied multiple times untile file is fixed or max iteration counter
-  is reached (default 10)"
-  [system id & {:keys [max-iterations label] :or {max-iterations 10}}]
-  (let [id (parse-uuid id)
-
-        validate-and-repair
-        (fn [file libs iteration]
-          (when-let [errors (not-empty (cfv/validate-file file libs))]
-            (l/trc :hint "repairing file"
-                   :file-id (str id)
-                   :iteration iteration
-                   :errors (count errors))
-            (let [changes (cfr/repair-file file libs errors)]
-              (-> file
-                  (update :revn inc)
-                  (update :data cpc/process-changes changes)))))
-
-        process-file
-        (fn [file libs]
-          (loop [file      file
-                 iteration 0]
-            (if (< iteration max-iterations)
-              (if-let [file (validate-and-repair file libs iteration)]
-                (recur file (inc iteration))
-                file)
-              (do
-                (l/wrn :hint "max retry num reached on repairing file"
-                       :file-id (str id)
-                       :iteration iteration)
-                file))))]
-
-    (db/tx-run! system
                 (fn [{:keys [::db/conn] :as system}]
-                  (when (string? label)
+                  (let [file (h/get-file system file-id)
-                    (fsnap/take-file-snapshot! system {:file-id id :label label}))
+                        libs (->> (files/get-file-libraries conn file-id)
-
-                  (let [file (h/get-file system id)
-                        libs (->> (files/get-file-libraries conn id)
                                   (into [file] (map (fn [{:keys [id]}]
                                                       (h/get-file system id))))
-                                  (d/index-by :id))
+                                  (d/index-by :id))]
-                        file (process-file file libs)]
+                    (cfv/validate-file file libs))))))
-                    (h/update-file! system file))))))
 
 (defn repair-file!
   "Repair the list of errors detected by validation."
   [file-id & {:keys [rollback?] :or {rollback? true} :as opts}]
-  (let [system (assoc main/system ::db/rollback rollback?)]
+  (let [system  (assoc main/system ::db/rollback rollback?)
-    (repair-file* system file-id (dissoc opts :rollback?))))
+        file-id (h/parse-uuid file-id)
+        opts    (assoc opts :with-libraries? true)]
+    (db/tx-run! system h/process-file! file-id fixes/repair-file opts)))
+
+;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
+;; PROCESSING
+;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
+
+(def ^:private
+  sql:get-file-ids
+  "SELECT id FROM file
+    WHERE created_at < ? AND deleted_at is NULL
+    ORDER BY created_at DESC")
+
+(defn analyze-files
+  "Apply a function to all files in the database, reading them in
+  batches. Do not change data.
+
+  The `on-file` parameter should be a function that receives the file
+  and the previous state and returns the new state.
+
+  Emits rollback at the end of operation."
+  [on-file & {:keys [max-items start-at with-libraries?]}]
+  (letfn [(get-candidates [conn]
+            (cond->> (db/cursor conn [sql:get-file-ids (or start-at (dt/now))])
+              (some? max-items)
+              (take max-items)))
+
+          (process-file [{:keys [::db/conn] :as system} file-id]
+            (let [file (h/get-file system file-id)
+                  libs (when with-libraries?
+                         (->> (files/get-file-libraries conn file-id)
+                              (into [file] (map (fn [{:keys [id]}]
+                                                  (h/get-file system id))))
+                              (d/index-by :id)))]
+              (if with-libraries?
+                (on-file file libs)
+                (on-file file))))]
+
+    (db/tx-run! (assoc main/system ::db/rollback true)
+                (fn [{:keys [::db/conn] :as system}]
+                  (binding [h/*system* system]
+                    (run! (partial process-file system)
+                          (get-candidates conn)))))))
+
+(defn process-file!
+  "Apply a function to the file. Optionally save the changes or not.
+  The function receives the decoded and migrated file data."
+  [file-id update-fn & {:keys [rollback?] :or {rollback? true}}]
+  (db/tx-run! (assoc main/system ::db/rollback rollback?)
+              (fn [system]
+                (binding [h/*system* system]
+                  (h/process-file! system file-id update-fn)))))
+
+(defn process-team-files!
+  "Apply a function to each file of the specified team."
+  [team-id update-fn & {:keys [rollback? label] :or {rollback? true} :as opts}]
+  (let [team-id (h/parse-uuid team-id)
+        opts    (dissoc opts :label)]
+    (db/tx-run! (assoc main/system ::db/rollback rollback?)
+                (fn [{:keys [::db/conn] :as system}]
+                  (when (string? label)
+                    (h/take-team-snapshot! system team-id label))
+
+                  (binding [h/*system* system]
+                    (->> (feat.comp-v2/get-and-lock-team-files conn team-id)
+                         (reduce (fn [result file-id]
+                                   (if (h/process-file! system file-id update-fn opts)
+                                     (inc result)
+                                     result))
+                                 0)))))))
+
+(defn process-files!
+  "Apply a function to all files in the database"
+  [update-fn & {:keys [max-items
+                       max-jobs
+                       start-at
+                       rollback?]
+                :or {max-jobs 1
+                     max-items Long/MAX_VALUE
+                     rollback? true}
+                :as opts}]
+
+  (l/dbg :hint "process:start"
+         :rollback rollback?
+         :max-jobs max-jobs
+         :max-items max-items)
+
+  (let [tpoint    (dt/tpoint)
+        factory   (px/thread-factory :virtual false :prefix "penpot/file-process/")
+        executor  (px/cached-executor :factory factory)
+        sjobs     (ps/create :permits max-jobs)
+
+        process-file
+        (fn [file-id idx tpoint]
+          (try
+            (l/trc :hint "process:file:start" :file-id (str file-id) :index idx)
+            (let [system (assoc main/system ::db/rollback rollback?)]
+              (db/tx-run! system h/process-file! file-id update-fn opts))
+
+            (catch Throwable cause
+              (l/wrn :hint "unexpected error on processing file (skiping)"
+                     :file-id (str file-id)
+                     :index idx
+                     :cause cause))
+            (finally
+              (ps/release! sjobs)
+              (let [elapsed (dt/format-duration (tpoint))]
+                (l/trc :hint "process:file:end"
+                       :file-id (str file-id)
+                       :index idx
+                       :elapsed elapsed)))))
+
+        process-files
+        (fn [{:keys [::db/conn] :as system}]
+          (db/exec! conn ["SET statement_timeout = 0"])
+          (db/exec! conn ["SET idle_in_transaction_session_timeout = 0"])
+
+          (try
+            (reduce (fn [idx file-id]
+                      (ps/acquire! sjobs)
+                      (px/run! executor (partial process-file file-id idx (dt/tpoint)))
+                      (inc idx))
+                    0
+                    (->> (db/cursor conn [sql:get-file-ids (or start-at (dt/now))])
+                         (take max-items)
+                         (map :id)))
+            (finally
+              ;; Close and await tasks
+              (pu/close! executor))))]
+
+    (try
+      (db/tx-run! main/system process-files)
+
+      (catch Throwable cause
+        (l/dbg :hint "process:error" :cause cause))
+
+      (finally
+        (let [elapsed (dt/format-duration (tpoint))]
+          (l/dbg :hint "process:end"
+                 :rollback rollback?
+                 :elapsed elapsed))))))
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; MISC