📎 Add helper for check not referenced media

This commit is contained in:
Andrey Antukh 2024-01-30 19:14:40 +01:00
parent e1befadc18
commit 4e9b92b857

View file

@ -9,12 +9,14 @@
(:refer-clojure :exclude [parse-uuid]) (:refer-clojure :exclude [parse-uuid])
#_:clj-kondo/ignore #_:clj-kondo/ignore
(:require (:require
[app.binfile.common :as bfc]
[app.common.data :as d] [app.common.data :as d]
[app.common.exceptions :as ex] [app.common.exceptions :as ex]
[app.common.features :as cfeat] [app.common.features :as cfeat]
[app.common.files.changes :as cpc] [app.common.files.changes :as cpc]
[app.common.files.migrations :as pmg] [app.common.files.migrations :as fmg]
[app.common.files.repair :as repair] [app.common.files.repair :as repair]
[app.common.files.validate :as cfv]
[app.common.files.validate :as validate] [app.common.files.validate :as validate]
[app.common.logging :as l] [app.common.logging :as l]
[app.common.pprint :refer [pprint]] [app.common.pprint :refer [pprint]]
@ -38,7 +40,8 @@
[expound.alpha :as expound] [expound.alpha :as expound]
[promesa.core :as p] [promesa.core :as p]
[promesa.exec :as px] [promesa.exec :as px]
[promesa.exec.csp :as sp])) [promesa.exec.semaphore :as ps]
[promesa.util :as pu]))
(def ^:dynamic *system* nil) (def ^:dynamic *system* nil)
@ -62,15 +65,21 @@
{:data data} {:data data}
{:id id})))) {:id id}))))
(defn get-file (defn- get-file*
"Get the migrated data of one file." "Get the migrated data of one file."
[id & {:keys [migrate?] :or {migrate? true}}] [system id]
(db/run! main/system (db/run! system
(fn [system] (fn [system]
(binding [pmap/*load-fn* (partial feat.fdata/load-pointer system id)] (binding [pmap/*load-fn* (partial feat.fdata/load-pointer system id)]
(-> (files/get-file system id :migrate? migrate?) (-> (files/get-file system id :migrate? false)
(update :data feat.fdata/process-pointers deref) (update :data feat.fdata/process-pointers deref)
(update :data feat.fdata/process-objects (partial into {}))))))) (update :data feat.fdata/process-objects (partial into {}))
(fmg/migrate-file))))))
(defn get-file
"Get the migrated data of one file."
[id]
(get-file* main/system id))
(defn validate (defn validate
"Validate structure, referencial integrity and semantic coherence of "Validate structure, referencial integrity and semantic coherence of
@ -78,92 +87,89 @@
[id] [id]
(db/tx-run! main/system (db/tx-run! main/system
(fn [{:keys [::db/conn] :as system}] (fn [{:keys [::db/conn] :as system}]
(binding [pmap/*load-fn* (partial feat.fdata/load-pointer system id)]
(let [id (if (string? id) (parse-uuid id) id) (let [id (if (string? id) (parse-uuid id) id)
file (files/get-file system id) file (get-file* system id)
libs (->> (files/get-file-libraries conn id) libs (->> (files/get-file-libraries conn id)
(into [file] (map (fn [{:keys [id]}] (into [file] (map (fn [{:keys [id]}]
(binding [pmap/*load-fn* (partial feat.fdata/load-pointer system id)] (get-file* system id))))
(-> (files/get-file system id :migrate? false)
(update :data feat.fdata/process-pointers deref)
(pmg/migrate-file))))))
(d/index-by :id))] (d/index-by :id))]
(validate/validate-file file libs)))))) (validate/validate-file file libs)))))
(defn repair! (defn repair!
"Repair the list of errors detected by validation." "Repair the list of errors detected by validation."
[id] [id]
(db/tx-run! main/system (db/tx-run! main/system
(fn [{:keys [::db/conn] :as system}] (fn [{:keys [::db/conn] :as system}]
(binding [pmap/*tracked* (pmap/create-tracked)
pmap/*load-fn* (partial feat.fdata/load-pointer system id)]
(let [id (if (string? id) (parse-uuid id) id) (let [id (if (string? id) (parse-uuid id) id)
file (files/get-file system id) file (get-file* system id)
libs (->> (files/get-file-libraries conn id) libs (->> (files/get-file-libraries conn id)
(into [file] (map (fn [{:keys [id]}] (into [file] (map (fn [{:keys [id]}]
(binding [pmap/*load-fn* (partial feat.fdata/load-pointer system id)] (get-file* system id))))
(-> (files/get-file system id :migrate? false)
(update :data feat.fdata/process-pointers deref)
(pmg/migrate-file))))))
(d/index-by :id)) (d/index-by :id))
errors (validate/validate-file file libs) errors (validate/validate-file file libs)
changes (repair/repair-file file libs errors) changes (repair/repair-file file libs errors)
file (-> file file (-> file
(update :revn inc) (update :revn inc)
(update :data cpc/process-changes changes) (update :data cpc/process-changes changes))
(update :data blob/encode))]
(when (contains? (:features file) "fdata/pointer-map") file (if (contains? (:features file) "fdata/objects-map")
(feat.fdata/persist-pointers! system id)) (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 id)
file))
file)]
(db/update! conn :file (db/update! conn :file
{:revn (:revn file) {:revn (:revn file)
:data (:data file) :data (blob/encode (:data file))
:data-backend nil :data-backend nil
:modified-at (dt/now) :modified-at (dt/now)
:has-media-trimmed false} :has-media-trimmed false}
{:id (:id file)}) {:id (:id file)})
:repaired)))))
:repaired))))
(defn update-file! (defn update-file!
"Apply a function to the data of one file. Optionally save the changes or not. "Apply a function to the data of one file. Optionally save the changes or not.
The function receives the decoded and migrated file data." The function receives the decoded and migrated file data."
[& {:keys [update-fn id rollback? migrate? inc-revn?] [& {:keys [update-fn id rollback? inc-revn?]
:or {rollback? true migrate? true inc-revn? true}}] :or {rollback? true inc-revn? true}}]
(letfn [(process-file [{:keys [::db/conn] :as system} {:keys [features] :as file}] (letfn [(process-file [{:keys [::db/conn] :as system} file-id]
(binding [pmap/*tracked* (pmap/create-tracked) (let [file (get-file* system file-id)
pmap/*load-fn* (partial feat.fdata/load-pointer system id) file (cond-> (update-fn file)
cfeat/*wrap-with-pointer-map-fn*
(if (contains? features "fdata/pointer-map") pmap/wrap identity)
cfeat/*wrap-with-objects-map-fn*
(if (contains? features "fdata/objectd-map") omap/wrap identity)]
(let [file (cond-> (update-fn file)
inc-revn? (update :revn inc)) inc-revn? (update :revn inc))
features (db/create-array conn "text" (:features file))
data (blob/encode (:data file))] _ (cfv/validate-file-schema! file)
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 id)
file))
file)]
(db/update! conn :file (db/update! conn :file
{:data data {:data (blob/encode (:data file))
:revn (:revn file) :features (db/create-array conn "text" (:features file))
:features features} :revn (:revn file)}
{:id id})) {:id (:id file)})
(when (contains? (:features file) "fdata/pointer-map")
(feat.fdata/persist-pointers! system id))
(dissoc file :data)))] (dissoc file :data)))]
(db/tx-run! (or *system* main/system) (db/tx-run! (or *system* (assoc main/system ::db/rollback rollback?))
(fn [system] (fn [system]
(binding [*system* system] (binding [*system* system]
(try (process-file system id))))))
(->> (files/get-file system id :migrate? migrate?)
(process-file system))
(finally
(when rollback?
(db/rollback! system)))))))))
(def ^:private sql:get-file-ids (def ^:private sql:get-file-ids
@ -190,16 +196,11 @@
(strace/print-stack-trace cause)) (strace/print-stack-trace cause))
(process-file [{:keys [::db/conn] :as system} file-id] (process-file [{:keys [::db/conn] :as system} file-id]
(let [file (binding [pmap/*load-fn* (partial feat.fdata/load-pointer system file-id)] (let [file (get-file* system file-id)
(-> (files/get-file system file-id)
(update :data feat.fdata/process-pointers deref)))
libs (when with-libraries? libs (when with-libraries?
(->> (files/get-file-libraries conn file-id) (->> (files/get-file-libraries conn file-id)
(into [file] (map (fn [{:keys [id]}] (into [file] (map (fn [{:keys [id]}]
(binding [pmap/*load-fn* (partial feat.fdata/load-pointer system id)] (get-file* system id))))
(-> (files/get-file system id)
(update :data feat.fdata/process-pointers deref))))))
(d/index-by :id)))] (d/index-by :id)))]
(try (try
(if with-libraries? (if with-libraries?
@ -208,7 +209,7 @@
(catch Throwable cause (catch Throwable cause
((or on-error on-error*) cause file)))))] ((or on-error on-error*) cause file)))))]
(db/tx-run! main/system (db/tx-run! (assoc main/system ::db/rollback true)
(fn [{:keys [::db/conn] :as system}] (fn [{:keys [::db/conn] :as system}]
(try (try
(binding [*system* system] (binding [*system* system]
@ -217,83 +218,125 @@
(get-candidates conn))) (get-candidates conn)))
(finally (finally
(when (fn? on-end) (when (fn? on-end)
(ex/ignoring (on-end))) (ex/ignoring (on-end)))))))))
(db/rollback! system)))))))
(defn repair-file-media
[{: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))))))))
(defn process-files! (defn process-files!
"Apply a function to all files in the database, reading them in "Apply a function to all files in the database"
batches."
[& {:keys [max-items [& {:keys [max-items
workers max-jobs
start-at start-at
on-file on-file
on-error
on-end
on-init
rollback?] rollback?]
:or {workers 1 :or {max-jobs 1
rollback? true}}] rollback? true}}]
(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] (l/dbg :hint "process:start"
(println! "unexpected exception happened on processing file: " (:id file)) :rollback rollback?
(strace/print-stack-trace cause)) :max-jobs max-jobs
:max-items max-items)
(process-file [system file-id] (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 tpoint]
(try (try
(let [{:keys [features] :as file} (files/get-file system file-id)] (l/trc :hint "process:file:start" :file-id (str file-id))
(binding [pmap/*tracked* (pmap/create-tracked) (db/tx-run! (assoc main/system ::db/rollback rollback?)
pmap/*load-fn* (partial feat.fdata/load-pointer system file-id) (fn [{:keys [::db/conn] :as system}]
cfeat/*wrap-with-pointer-map-fn* (let [file' (get-file* system file-id)
(if (contains? features "fdata/pointer-map") pmap/wrap identity) file (binding [*system* system]
cfeat/*wrap-with-objects-map-fn* (on-file file'))]
(if (contains? features "fdata/objectd-map") omap/wrap identity)]
(on-file file) (when (and (some? file)
(not (identical? file file')))
(when (contains? features "fdata/pointer-map") (cfv/validate-file-schema! file)
(feat.fdata/persist-pointers! system file-id))))
(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))
: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)
:cause cause))
(finally
(ps/release! sjobs)
(let [elapsed (dt/format-duration (tpoint))]
(l/trc :hint "process:file:end"
:file-id (str file-id)
: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"])
(run! (fn [file-id]
(ps/acquire! sjobs)
(px/run! executor (partial process-file file-id (dt/tpoint))))
(->> (db/cursor conn [sql:get-file-ids (or start-at (dt/now))])
(take max-items)
(map :id)))
;; Close and await tasks
(pu/close! executor)))
(catch Throwable cause (catch Throwable cause
((or on-error on-error*) cause file-id)))) (l/dbg :hint "process:error" :cause cause))
(run-worker [in index] (finally
(db/tx-run! main/system (let [elapsed (dt/format-duration (tpoint))]
(fn [system] (l/dbg :hint "process:end"
(binding [*system* system] :rollback rollback?
(loop [i 0] :elapsed elapsed))))))
(when-let [file-id (sp/take! in)]
(println! "=> worker: index:" index "| loop:" i "| file:" (str file-id) "|" (px/get-name))
(process-file system file-id)
(recur (inc i)))))
(when rollback?
(db/rollback! system)))))
(run-producer [input]
(db/tx-run! main/system
(fn [{:keys [::db/conn]}]
(doseq [file-id (get-candidates conn)]
(println! "=> producer:" file-id "|" (px/get-name))
(sp/put! input file-id))
(sp/close! input))))]
(when (fn? on-init) (on-init))
(let [input (sp/chan :buf 25)
producer (px/thread
{:name "penpot/srepl/producer"}
(run-producer input))
threads (->> (range workers)
(map (fn [index]
(px/thread
{:name (str "penpot/srepl/worker/" index)}
(run-worker input index))))
(cons producer)
(doall))]
(run! p/await! threads)
(when (fn? on-end) (on-end)))))