3pp-mirror/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2025-05-24 00:16:11 +02:00
Code Issues Projects Releases Packages Wiki Activity

✨ Reorganize fdata/pointer-map feature helpers

Browse source 
Mainly move all pointer-map related helpers from app.rpc.commands.files
to the the app.features.fdata namespace and normalizes codestile around
feature handling on all affected code.

This commit also comes with several features related bugifxes on the
components-v2 migration code:

- properly migrate legacy feature names on apply components-v2 migration
- start using new fdata feature related functions
- prevent generation of a ephimeral pointer on each graphic migration
  operation; on large files this caused a very noticiable overhead of
  creating a big number of completly unused pointer maps
- do persistence after validation and not before
This commit is contained in:
Andrey Antukh 2023-12-07 11:51:52 +01:00 • committed by Andrés Moya
parent 5669bfc260
commit 417366d998
13 changed files with 649 additions and 610 deletions
Download patch file Download diff file Expand all files Collapse all files

6
backend/src/app/db.clj
View file

@ -249,6 +249,12 @@
:code :unable-resolve-connection :code :unable-resolve-connection
:hint "expected conn or system map")))) :hint "expected conn or system map"))))
(defn connection-map?
"Check if the provided value is a map like data structure that
contains a database connection."
[o]
(and (map? o) (connection? (::conn o))))
(defn- get-connectable (defn- get-connectable
[o] [o]
(cond (cond

81
backend/src/app/features/components_v2.clj
View file

@ -14,7 +14,7 @@
[app.common.files.changes-builder :as fcb] [app.common.files.changes-builder :as fcb]
[app.common.files.helpers :as cfh] [app.common.files.helpers :as cfh]
[app.common.files.libraries-helpers :as cflh] [app.common.files.libraries-helpers :as cflh]
[app.common.files.migrations :as pmg] [app.common.files.migrations :as fmg]
[app.common.files.shapes-helpers :as cfsh] [app.common.files.shapes-helpers :as cfsh]
[app.common.files.validate :as cfv] [app.common.files.validate :as cfv]
[app.common.geom.point :as gpt] [app.common.geom.point :as gpt]
@ -32,6 +32,7 @@
[app.common.types.shape-tree :as ctst] [app.common.types.shape-tree :as ctst]
[app.common.uuid :as uuid] [app.common.uuid :as uuid]
[app.db :as db] [app.db :as db]
[app.features.fdata :as fdata]
[app.http.sse :as sse] [app.http.sse :as sse]
[app.media :as media] [app.media :as media]
[app.rpc.commands.files :as files] [app.rpc.commands.files :as files]
@ -40,7 +41,6 @@
[app.storage :as sto] [app.storage :as sto]
[app.storage.tmp :as tmp] [app.storage.tmp :as tmp]
[app.util.blob :as blob] [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] [app.util.time :as dt]
[buddy.core.codecs :as bc] [buddy.core.codecs :as bc]
@ -703,7 +703,6 @@
(try (try
(->> (d/zip media-group grid) (->> (d/zip media-group grid)
(map (fn [[mobj position]] (map (fn [[mobj position]]
(l/trc :hint "submit graphic processing" :file-id (str (:id fdata)) :id (str (:id mobj)))
(sse/tap {:type :migration-progress (sse/tap {:type :migration-progress
:section :graphics :section :graphics
:name (:name mobj)}) :name (:name mobj)})
@ -761,7 +760,7 @@
(create-media-grid fdata page-id (:id frame) grid assets) (create-media-grid fdata page-id (:id frame) grid assets)
(gpt/add position (gpt/point 0 (+ height (* 2 grid-gap) frame-gap)))))))))) (gpt/add position (gpt/point 0 (+ height (* 2 grid-gap) frame-gap))))))))))
(defn- migrate-file-data (defn- migrate-fdata
[fdata libs] [fdata libs]
(let [migrated? (dm/get-in fdata [:options :components-v2])] (let [migrated? (dm/get-in fdata [:options :components-v2])]
(if migrated? (if migrated?
@ -770,42 +769,38 @@
fdata (migrate-graphics fdata)] fdata (migrate-graphics fdata)]
(update fdata :options assoc :components-v2 true))))) (update fdata :options assoc :components-v2 true)))))
(defn- process-file (defn- process-fdata
[{:keys [id] :as file} & {:keys [validate? throw-on-validate?]}] [fdata id]
(let [conn (::db/conn *system*)] (-> fdata
(binding [pmap/*tracked* (atom {}) (assoc :id id)
pmap/*load-fn* (partial files/load-pointer conn id) (fdata/process-pointers deref)
cfeat/*wrap-with-pointer-map-fn* (fmg/migrate-data)))
(if (contains? (:features file) "fdata/pointer-map") pmap/wrap identity)
cfeat/*wrap-with-objects-map-fn*
(if (contains? (:features file) "fdata/objectd-map") omap/wrap identity)]
(let [file (-> file (defn- process-file
(update :data blob/decode) [{:keys [::db/conn] :as system} id & {:keys [validate? throw-on-validate?]}]
(update :data assoc :id id) (binding [pmap/*tracked* (pmap/create-tracked)
(pmg/migrate-file)) pmap/*load-fn* (partial fdata/load-pointer *system* id)]
(let [file (binding [cfeat/*new* (atom #{})]
(-> (files/get-file system id :migrate? false)
(update :data process-fdata id)
(update :features into (deref cfeat/*new*))
(update :features cfeat/migrate-legacy-features)))
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 files/load-pointer conn id)] (binding [pmap/*load-fn* (partial fdata/load-pointer system id)]
(-> (db/get conn :file {:id id}) (-> (files/get-file system id :migrate? false)
(files/decode-row) (update :data process-fdata id))))))
(files/process-pointers deref) ; ensure all pointers resolved
(pmg/migrate-file))))))
(d/index-by :id)) (d/index-by :id))
pmap? (contains? (:features file) "fdata/pointer-map")
file (-> file file (-> file
(update :data migrate-file-data libs) (update :data migrate-fdata libs)
(update :features conj "components/v2"))] (update :features conj "components/v2")
(cond-> pmap? (fdata/enable-pointer-map)))
(when (contains? (:features file) "fdata/pointer-map") ]
(files/persist-pointers! conn id))
(db/update! conn :file
{:data (blob/encode (:data file))
:features (db/create-array conn "text" (:features file))
:revn (:revn file)}
{:id (:id file)})
(when validate? (when validate?
(if throw-on-validate? (if throw-on-validate?
@ -816,7 +811,16 @@
:file-name (:name file) :file-name (:name file)
:error error)))) :error error))))
(dissoc file :data))))) (db/update! conn :file
{:data (blob/encode (:data file))
:features (db/create-array conn "text" (:features file))
:revn (:revn file)}
{:id (:id file)})
(when pmap?
(fdata/persist-pointers! system id))
(dissoc file :data))))
(defn migrate-file! (defn migrate-file!
[system file-id & {:keys [validate? throw-on-validate?]}] [system file-id & {:keys [validate? throw-on-validate?]}]
@ -830,13 +834,12 @@
(let [system (update system ::sto/storage media/configure-assets-storage)] (let [system (update system ::sto/storage media/configure-assets-storage)]
(db/tx-run! system (db/tx-run! system
(fn [{:keys [::db/conn] :as system}] (fn [system]
(binding [*system* system] (binding [*system* system]
(fsnap/take-file-snapshot! system {:file-id file-id :label "migration/components-v2"}) (fsnap/take-file-snapshot! system {:file-id file-id :label "migration/components-v2"})
(-> (db/get conn :file {:id file-id}) (process-file system file-id
(update :features db/decode-pgarray #{}) :validate? validate?
(process-file :validate? validate? :throw-on-validate? throw-on-validate?)))))
:throw-on-validate? throw-on-validate?))))))
(finally (finally
(let [elapsed (tpoint) (let [elapsed (tpoint)

98
backend/src/app/features/fdata.clj
View file

@ -7,42 +7,90 @@
(ns app.features.fdata (ns app.features.fdata
"A `fdata/*` related feature migration helpers" "A `fdata/*` related feature migration helpers"
(:require (:require
[app.common.data :as d]
[app.common.exceptions :as ex]
[app.common.logging :as l]
[app.db :as db]
[app.util.blob :as blob]
[app.util.objects-map :as omap] [app.util.objects-map :as omap]
[app.util.pointer-map :as pmap])) [app.util.pointer-map :as pmap]))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; OBJECTS-MAP
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(defn enable-objects-map (defn enable-objects-map
[file] [file]
(let [update-fn #(d/update-when % :objects omap/wrap)]
(-> file (-> file
(update :data (fn [data] (update :data (fn [fdata]
(-> data (-> fdata
(update :pages-index update-vals #(update % :objects omap/wrap)) (update :pages-index update-vals update-fn)
(update :components update-vals #(update % :objects omap/wrap))))) (update :components update-vals update-fn))))
(update :features conj "fdata/objects-map"))) (update :features conj "fdata/objects-map"))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; POINTER-MAP
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(defn load-pointer
"A database loader pointer helper"
[system file-id id]
(let [{:keys [content]} (db/get system :file-data-fragment
{:id id :file-id file-id}
{::db/columns [:content]
::db/check-deleted? false})]
(when-not content
(ex/raise :type :internal
:code :fragment-not-found
:hint "fragment not found"
:file-id file-id
:fragment-id id))
(blob/decode content)))
(defn persist-pointers!
"Given a database connection and the final file-id, persist all
pointers to the underlying storage (the database)."
[system file-id]
(doseq [[id item] @pmap/*tracked*]
(when (pmap/modified? item)
(l/trc :hint "persist pointer" :file-id (str file-id) :id (str id))
(let [content (-> item deref blob/encode)]
(db/insert! system :file-data-fragment
{:id id
:file-id file-id
:content content})))))
(defn process-pointers
"Apply a function to all pointers on the file. Usuly used for
dereference the pointer to a plain value before some processing."
[fdata update-fn]
(cond-> fdata
(contains? fdata :pages-index)
(update :pages-index process-pointers update-fn)
:always
(update-vals (fn [val]
(if (pmap/pointer-map? val)
(update-fn val)
val)))))
(defn get-used-pointer-ids
"Given a file, return all pointer ids used in the data."
[fdata]
(->> (concat (vals fdata)
(vals (:pages-index fdata)))
(into #{} (comp (filter pmap/pointer-map?)
(map pmap/get-id)))))
(defn enable-pointer-map (defn enable-pointer-map
"Enable the fdata/pointer-map feature on the file."
[file] [file]
(-> file (-> file
(update :data (fn [data] (update :data (fn [fdata]
(-> data (-> fdata
(update :pages-index update-vals pmap/wrap) (update :pages-index update-vals pmap/wrap)
(update :components pmap/wrap)))) (update :components pmap/wrap))))
(update :features conj "fdata/pointer-map"))) (update :features conj "fdata/pointer-map")))
;; (defn enable-shape-data-type
;; [file]
;; (letfn [(update-object [object]
;; (-> object
;; (d/update-when :selrect grc/make-rect)
;; (d/update-when :svg-viewbox grc/make-rect)
;; (cts/map->Shape)))
;; (update-container [container]
;; (d/update-when container :objects update-vals update-object))]
;; (-> file
;; (update :data (fn [data]
;; (-> data
;; (update :pages-index update-vals update-container)
;; (update :components update-vals update-container))))
;; (update :features conj "fdata/shape-data-type"))))

29
backend/src/app/rpc/commands/binfile.clj
View file

@ -21,8 +21,8 @@
[app.common.uuid :as uuid] [app.common.uuid :as uuid]
[app.config :as cf] [app.config :as cf]
[app.db :as db] [app.db :as db]
[app.features.components-v2 :as features.components-v2] [app.features.components-v2 :as feat.compv2]
[app.features.fdata :as features.fdata] [app.features.fdata :as feat.fdata]
[app.http.sse :as sse] [app.http.sse :as sse]
[app.loggers.audit :as-alias audit] [app.loggers.audit :as-alias audit]
[app.loggers.webhooks :as-alias webhooks] [app.loggers.webhooks :as-alias webhooks]
@ -305,25 +305,21 @@
(defn- get-files (defn- get-files
[cfg ids] [cfg ids]
(letfn [(get-files* [{:keys [::db/conn]}] (db/run! cfg (fn [{:keys [::db/conn]}]
(let [sql (str "SELECT id FROM file " (let [sql (str "SELECT id FROM file "
" WHERE id = ANY(?) ") " WHERE id = ANY(?) ")
ids (db/create-array conn "uuid" ids)] ids (db/create-array conn "uuid" ids)]
(->> (db/exec! conn [sql ids]) (->> (db/exec! conn [sql ids])
(into [] (map :id)) (into [] (map :id))
(not-empty))))] (not-empty))))))
(db/run! cfg get-files*)))
(defn- get-file (defn- get-file
[cfg file-id] [cfg file-id]
(letfn [(get-file* [{:keys [::db/conn]}] (db/run! cfg (fn [{:keys [::db/conn] :as cfg}]
(binding [pmap/*load-fn* (partial files/load-pointer conn file-id)] (binding [pmap/*load-fn* (partial feat.fdata/load-pointer cfg file-id)]
(some-> (db/get* conn :file {:id file-id} {::db/remove-deleted? false}) (some-> (db/get* conn :file {:id file-id} {::db/remove-deleted? false})
(files/decode-row) (files/decode-row)
(files/process-pointers deref))))] (update :data feat.fdata/process-pointers deref))))))
(db/run! cfg get-file*)))
(defn- get-file-media (defn- get-file-media
[{:keys [::db/pool]} {:keys [data id] :as file}] [{:keys [::db/pool]} {:keys [data id] :as file}]
@ -666,7 +662,7 @@
(doseq [[feature file-id] (-> *state* deref :pending-to-migrate)] (doseq [[feature file-id] (-> *state* deref :pending-to-migrate)]
(case feature (case feature
"components/v2" "components/v2"
(features.components-v2/migrate-file! options file-id (feat.compv2/migrate-file! options file-id
:validate? validate? :validate? validate?
:throw-on-validate? true) :throw-on-validate? true)
@ -702,11 +698,11 @@
(cond-> file (cond-> file
(and (contains? cfeat/*current* "fdata/objects-map") (and (contains? cfeat/*current* "fdata/objects-map")
(not (contains? cfeat/*previous* "fdata/objects-map"))) (not (contains? cfeat/*previous* "fdata/objects-map")))
(features.fdata/enable-objects-map) (feat.fdata/enable-objects-map)
(and (contains? cfeat/*current* "fdata/pointer-map") (and (contains? cfeat/*current* "fdata/pointer-map")
(not (contains? cfeat/*previous* "fdata/pointer-map"))) (not (contains? cfeat/*previous* "fdata/pointer-map")))
(features.fdata/enable-pointer-map))) (feat.fdata/enable-pointer-map)))
(defn- get-remaped-thumbnails (defn- get-remaped-thumbnails
[thumbnails file-id] [thumbnails file-id]
@ -717,7 +713,7 @@
thumbnails)) thumbnails))
(defmethod read-section :v1/files (defmethod read-section :v1/files
[{:keys [::db/conn ::input ::project-id ::enabled-features ::timestamp ::overwrite?]}] [{:keys [::db/conn ::input ::project-id ::enabled-features ::timestamp ::overwrite?] :as system}]
(doseq [expected-file-id (-> *state* deref :files)] (doseq [expected-file-id (-> *state* deref :files)]
(let [file (read-obj! input) (let [file (read-obj! input)
@ -773,7 +769,6 @@
cfeat/*previous* (:features file) cfeat/*previous* (:features file)
pmap/*tracked* (atom {})] pmap/*tracked* (atom {})]
(let [params (-> file (let [params (-> file
(assoc :id file-id') (assoc :id file-id')
(assoc :features features) (assoc :features features)
@ -821,7 +816,7 @@
(create-or-update-file! conn params) (create-or-update-file! conn params)
(db/insert! conn :file params)) (db/insert! conn :file params))
(files/persist-pointers! conn file-id') (feat.fdata/persist-pointers! system file-id')
(when overwrite? (when overwrite?
(db/delete! conn :file-thumbnail {:file-id file-id'})) (db/delete! conn :file-thumbnail {:file-id file-id'}))

57
backend/src/app/rpc/commands/comments.clj
View file

@ -12,6 +12,7 @@
[app.common.spec :as us] [app.common.spec :as us]
[app.common.uuid :as uuid] [app.common.uuid :as uuid]
[app.db :as db] [app.db :as db]
[app.features.fdata :as feat.fdata]
[app.loggers.audit :as-alias audit] [app.loggers.audit :as-alias audit]
[app.loggers.webhooks :as-alias webhooks] [app.loggers.webhooks :as-alias webhooks]
[app.rpc :as-alias rpc] [app.rpc :as-alias rpc]
@ -43,15 +44,17 @@
(defn- get-file (defn- get-file
"A specialized version of get-file for comments module." "A specialized version of get-file for comments module."
[conn file-id page-id] [{:keys [::db/conn] :as cfg} file-id page-id]
(binding [pmap/*load-fn* (partial files/load-pointer conn file-id)] (if-let [{:keys [data] :as file} (some-> (db/exec-one! conn [sql:get-file file-id])
(if-let [{:keys [data] :as file} (some-> (db/exec-one! conn [sql:get-file file-id]) (files/decode-row))] (files/decode-row))]
(binding [pmap/*load-fn* (partial feat.fdata/load-pointer cfg file-id)]
(-> file (-> file
(assoc :page-name (dm/get-in data [:pages-index page-id :name])) (assoc :page-name (dm/get-in data [:pages-index page-id :name]))
(assoc :page-id page-id)) (assoc :page-id page-id)))
(ex/raise :type :not-found (ex/raise :type :not-found
:code :object-not-found :code :object-not-found
:hint "file not found")))) :hint "file not found")))
(defn- get-comment-thread (defn- get-comment-thread
[conn thread-id & {:as opts}] [conn thread-id & {:as opts}]
@ -288,12 +291,11 @@
(sv/defmethod ::create-comment-thread (sv/defmethod ::create-comment-thread
{::doc/added "1.15" {::doc/added "1.15"
::webhooks/event? true} ::webhooks/event? true}
[{:keys [::db/pool] :as cfg} [cfg {:keys [::rpc/profile-id ::rpc/request-at file-id page-id share-id position content frame-id]}]
{:keys [::rpc/profile-id ::rpc/request-at file-id page-id share-id position content frame-id]}] (db/tx-run! cfg
(fn [{:keys [::db/conn] :as cfg}]
(db/with-atomic [conn pool]
(let [{:keys [team-id project-id page-name] :as file} (get-file conn file-id page-id)]
(files/check-comment-permissions! conn profile-id file-id share-id) (files/check-comment-permissions! conn profile-id file-id share-id)
(let [{:keys [team-id project-id page-name] :as file} (get-file cfg file-id page-id)]
(run! (partial quotes/check-quote! conn) (run! (partial quotes/check-quote! conn)
(list {::quotes/id ::quotes/comment-threads-per-file (list {::quotes/id ::quotes/comment-threads-per-file
@ -319,7 +321,7 @@
:page-name page-name :page-name page-name
:position position :position position
:content content :content content
:frame-id frame-id}))))) :frame-id frame-id}))))))
(defn- create-comment-thread (defn- create-comment-thread
@ -402,8 +404,7 @@
;; --- COMMAND: Add Comment ;; --- COMMAND: Add Comment
(declare get-comment-thread) (declare ^:private get-comment-thread)
(declare create-comment)
(s/def ::create-comment (s/def ::create-comment
(s/keys :req [::rpc/profile-id] (s/keys :req [::rpc/profile-id]
@ -413,10 +414,11 @@
(sv/defmethod ::create-comment (sv/defmethod ::create-comment
{::doc/added "1.15" {::doc/added "1.15"
::webhooks/event? true} ::webhooks/event? true}
[{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id ::rpc/request-at thread-id share-id content] :as params}] [cfg {:keys [::rpc/profile-id ::rpc/request-at thread-id share-id content]}]
(db/with-atomic [conn pool] (db/tx-run! cfg
(fn [{:keys [::db/conn] :as cfg}]
(let [{:keys [file-id page-id] :as thread} (get-comment-thread conn thread-id ::db/for-update? true) (let [{:keys [file-id page-id] :as thread} (get-comment-thread conn thread-id ::db/for-update? true)
{:keys [team-id project-id page-name] :as file} (get-file conn file-id page-id)] {:keys [team-id project-id page-name] :as file} (get-file cfg file-id page-id)]
(files/check-comment-permissions! conn profile-id (:id file) share-id) (files/check-comment-permissions! conn profile-id (:id file) share-id)
(quotes/check-quote! conn (quotes/check-quote! conn
@ -455,7 +457,8 @@
;; current thread. ;; current thread.
(upsert-comment-thread-status! conn profile-id thread-id request-at) (upsert-comment-thread-status! conn profile-id thread-id request-at)
(vary-meta comment assoc ::audit/props props))))) (vary-meta comment assoc ::audit/props props))))))
;; --- COMMAND: Update Comment ;; --- COMMAND: Update Comment
@ -466,8 +469,10 @@
(sv/defmethod ::update-comment (sv/defmethod ::update-comment
{::doc/added "1.15"} {::doc/added "1.15"}
[{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id ::rpc/request-at id share-id content] :as params}] [cfg {:keys [::rpc/profile-id ::rpc/request-at id share-id content]}]
(db/with-atomic [conn pool]
(db/tx-run! cfg
(fn [{:keys [::db/conn] :as cfg}]
(let [{:keys [thread-id owner-id] :as comment} (get-comment conn id ::db/for-update? true) (let [{:keys [thread-id owner-id] :as comment} (get-comment conn id ::db/for-update? true)
{:keys [file-id page-id] :as thread} (get-comment-thread conn thread-id ::db/for-update? true)] {:keys [file-id page-id] :as thread} (get-comment-thread conn thread-id ::db/for-update? true)]
@ -478,7 +483,7 @@
(ex/raise :type :validation (ex/raise :type :validation
:code :not-allowed)) :code :not-allowed))
(let [{:keys [page-name] :as file} (get-file conn file-id page-id)] (let [{:keys [page-name] :as file} (get-file cfg file-id page-id)]
(db/update! conn :comment (db/update! conn :comment
{:content content {:content content
:modified-at request-at} :modified-at request-at}
@ -488,7 +493,7 @@
{:modified-at request-at {:modified-at request-at
:page-name page-name} :page-name page-name}
{:id thread-id}) {:id thread-id})
nil)))) nil)))))
;; --- COMMAND: Delete Comment Thread ;; --- COMMAND: Delete Comment Thread
@ -499,7 +504,7 @@
(sv/defmethod ::delete-comment-thread (sv/defmethod ::delete-comment-thread
{::doc/added "1.15"} {::doc/added "1.15"}
[{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id id share-id] :as params}] [{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id id share-id]}]
(db/with-atomic [conn pool] (db/with-atomic [conn pool]
(let [{:keys [owner-id file-id] :as thread} (get-comment-thread conn id ::db/for-update? true)] (let [{:keys [owner-id file-id] :as thread} (get-comment-thread conn id ::db/for-update? true)]
(files/check-comment-permissions! conn profile-id file-id share-id) (files/check-comment-permissions! conn profile-id file-id share-id)
@ -539,12 +544,12 @@
(sv/defmethod ::update-comment-thread-position (sv/defmethod ::update-comment-thread-position
{::doc/added "1.15"} {::doc/added "1.15"}
[{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id id position frame-id share-id] :as params}] [{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id ::rpc/request-at id position frame-id share-id]}]
(db/with-atomic [conn pool] (db/with-atomic [conn pool]
(let [{:keys [file-id] :as thread} (get-comment-thread conn id ::db/for-update? true)] (let [{:keys [file-id] :as thread} (get-comment-thread conn id ::db/for-update? true)]
(files/check-comment-permissions! conn profile-id file-id share-id) (files/check-comment-permissions! conn profile-id file-id share-id)
(db/update! conn :comment-thread (db/update! conn :comment-thread
{:modified-at (::rpc/request-at params) {:modified-at request-at
:position (db/pgpoint position) :position (db/pgpoint position)
:frame-id frame-id} :frame-id frame-id}
{:id (:id thread)}) {:id (:id thread)})
@ -559,12 +564,12 @@
(sv/defmethod ::update-comment-thread-frame (sv/defmethod ::update-comment-thread-frame
{::doc/added "1.15"} {::doc/added "1.15"}
[{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id id frame-id share-id] :as params}] [{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id ::rpc/request-at id frame-id share-id]}]
(db/with-atomic [conn pool] (db/with-atomic [conn pool]
(let [{:keys [file-id] :as thread} (get-comment-thread conn id ::db/for-update? true)] (let [{:keys [file-id] :as thread} (get-comment-thread conn id ::db/for-update? true)]
(files/check-comment-permissions! conn profile-id file-id share-id) (files/check-comment-permissions! conn profile-id file-id share-id)
(db/update! conn :comment-thread (db/update! conn :comment-thread
{:modified-at (::rpc/request-at params) {:modified-at request-at
:frame-id frame-id} :frame-id frame-id}
{:id id}) {:id id})
nil))) nil)))

252
backend/src/app/rpc/commands/files.clj
View file

@ -20,6 +20,7 @@
[app.common.types.file :as ctf] [app.common.types.file :as ctf]
[app.config :as cf] [app.config :as cf]
[app.db :as db] [app.db :as db]
[app.features.fdata :as feat.fdata]
[app.loggers.audit :as-alias audit] [app.loggers.audit :as-alias audit]
[app.loggers.webhooks :as-alias webhooks] [app.loggers.webhooks :as-alias webhooks]
[app.rpc :as-alias rpc] [app.rpc :as-alias rpc]
@ -181,62 +182,6 @@
:code :object-not-found :code :object-not-found
:hint "not found")))) :hint "not found"))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; FEATURES: pointer-map
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(defn load-pointer
[conn file-id id]
(dm/assert!
"expected valid connection"
(db/connection? conn))
(let [{:keys [content]} (db/get conn :file-data-fragment
{:id id :file-id file-id}
{::db/columns [:content]
::db/check-deleted? false})]
(when-not content
(ex/raise :type :internal
:code :fragment-not-found
:hint "fragment not found"
:file-id file-id
:fragment-id id))
(blob/decode content)))
(defn persist-pointers!
[conn file-id]
(doseq [[id item] @pmap/*tracked*]
(when (pmap/modified? item)
(let [content (-> item deref blob/encode)]
(db/insert! conn :file-data-fragment
{:id id
:file-id file-id
:content content})))))
(defn process-pointers
[file update-fn]
(update file :data (fn resolve-fn [data]
(cond-> data
(contains? data :pages-index)
(update :pages-index resolve-fn)
:always
(update-vals (fn [val]
(if (pmap/pointer-map? val)
(update-fn val)
val)))))))
(defn get-all-pointer-ids
"Given a file, return all pointer ids used in the data."
[fdata]
(->> (concat (vals fdata)
(vals (:pages-index fdata)))
(into #{} (comp (filter pmap/pointer-map?)
(map pmap/get-id)))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; QUERY COMMANDS ;; QUERY COMMANDS
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
@ -277,32 +222,12 @@
[:id ::sm/uuid] [:id ::sm/uuid]
[:project-id {:optional true} ::sm/uuid]])) [:project-id {:optional true} ::sm/uuid]]))
(defn get-file (defn- migrate-file
[conn id & {:keys [project-id migrate? [{:keys [::db/conn] :as cfg} {:keys [id] :as file}]
include-deleted? (binding [pmap/*load-fn* (partial feat.fdata/load-pointer cfg id)
lock-for-update?]
:or {include-deleted? false
lock-for-update? false}}]
(dm/assert!
"expected raw connection"
(db/connection? conn))
(binding [pmap/*load-fn* (partial load-pointer conn id)
pmap/*tracked* (pmap/create-tracked) pmap/*tracked* (pmap/create-tracked)
cfeat/*new* (atom #{})] cfeat/*new* (atom #{})]
(let [file (fmg/migrate-file file)]
(let [params (merge {:id id}
(when (some? project-id)
{:project-id project-id}))
file (-> (db/get conn :file params
{::db/check-deleted? (not include-deleted?)
::db/remove-deleted? (not include-deleted?)
::db/for-update? lock-for-update?})
(decode-row)
(cond-> migrate?
(fmg/migrate-file)))]
;; NOTE: when file is migrated, we break the rule of no perform ;; NOTE: when file is migrated, we break the rule of no perform
;; mutations on get operations and update the file with all ;; mutations on get operations and update the file with all
;; migrations applied ;; migrations applied
@ -317,10 +242,32 @@
{:data (blob/encode (:data file)) {:data (blob/encode (:data file))
:features (db/create-array conn "text" features)} :features (db/create-array conn "text" features)}
{:id id}) {:id id})
(persist-pointers! conn id) (feat.fdata/persist-pointers! cfg id)
(assoc file :features features)) (assoc file :features features))
file)))) file))))
(defn get-file
[{:keys [::db/conn] :as cfg} id & {:keys [project-id migrate?
include-deleted?
lock-for-update?]
:or {include-deleted? false
lock-for-update? false}}]
(dm/assert!
"expected cfg with valid connection"
(db/connection-map? cfg))
(let [params (merge {:id id}
(when (some? project-id)
{:project-id project-id}))
file (-> (db/get conn :file params
{::db/check-deleted? (not include-deleted?)
::db/remove-deleted? (not include-deleted?)
::db/for-update? lock-for-update?})
(decode-row))]
(if migrate?
(migrate-file cfg file)
file)))
(defn get-minimal-file (defn get-minimal-file
[{:keys [::db/pool] :as cfg} id] [{:keys [::db/pool] :as cfg} id]
(db/get pool :file {:id id} {:columns [:id :modified-at :revn]})) (db/get pool :file {:id id} {:columns [:id :modified-at :revn]}))
@ -345,7 +292,7 @@
:project-id project-id :project-id project-id
:file-id id) :file-id id)
file (-> (get-file conn id :project-id project-id) file (-> (get-file cfg id :project-id project-id)
(assoc :permissions perms) (assoc :permissions perms)
(check-version!)) (check-version!))
@ -358,8 +305,8 @@
;; pointers on backend and return a complete file. ;; pointers on backend and return a complete file.
file (if (and (contains? (:features file) "fdata/pointer-map") file (if (and (contains? (:features file) "fdata/pointer-map")
(not (contains? (:features params) "fdata/pointer-map"))) (not (contains? (:features params) "fdata/pointer-map")))
(binding [pmap/*load-fn* (partial load-pointer conn id)] (binding [pmap/*load-fn* (partial feat.fdata/load-pointer cfg id)]
(process-pointers file deref)) (update file :data feat.fdata/process-pointers deref))
file)] file)]
(vary-meta file assoc ::cond/key (get-file-etag params file))))))) (vary-meta file assoc ::cond/key (get-file-etag params file)))))))
@ -498,6 +445,7 @@
(defn get-page (defn get-page
[{:keys [::db/conn] :as cfg} {:keys [profile-id file-id page-id object-id] :as params}] [{:keys [::db/conn] :as cfg} {:keys [profile-id file-id page-id object-id] :as params}]
(when (and (uuid? object-id) (when (and (uuid? object-id)
(not (uuid? page-id))) (not (uuid? page-id)))
(ex/raise :type :validation (ex/raise :type :validation
@ -508,13 +456,13 @@
:profile-id profile-id :profile-id profile-id
:file-id file-id) :file-id file-id)
file (get-file conn file-id) file (get-file cfg file-id)
_ (-> (cfeat/get-team-enabled-features cf/flags team) _ (-> (cfeat/get-team-enabled-features cf/flags team)
(cfeat/check-client-features! (:features params)) (cfeat/check-client-features! (:features params))
(cfeat/check-file-features! (:features file) (:features params))) (cfeat/check-file-features! (:features file) (:features params)))
page (binding [pmap/*load-fn* (partial load-pointer conn file-id)] page (binding [pmap/*load-fn* (partial feat.fdata/load-pointer cfg file-id)]
(let [page-id (or page-id (-> file :data :pages first)) (let [page-id (or page-id (-> file :data :pages first))
page (dm/get-in file [:data :pages-index page-id])] page (dm/get-in file [:data :pages-index page-id])]
(if (pmap/pointer-map? page) (if (pmap/pointer-map? page)
@ -573,17 +521,16 @@
and p.team_id = ? and p.team_id = ?
order by f.modified_at desc") order by f.modified_at desc")
;; FIXME: i'm not sure about feature handling here... ???
(defn get-team-shared-files (defn- get-library-summary
[conn team-id] [cfg {:keys [id data] :as file}]
(letfn [(assets-sample [assets limit] (letfn [(assets-sample [assets limit]
(let [sorted-assets (->> (vals assets) (let [sorted-assets (->> (vals assets)
(sort-by #(str/lower (:name %))))] (sort-by #(str/lower (:name %))))]
{:count (count sorted-assets) {:count (count sorted-assets)
:sample (into [] (take limit sorted-assets))})) :sample (into [] (take limit sorted-assets))}))]
(library-summary [{:keys [id data] :as file}] (binding [pmap/*load-fn* (partial feat.fdata/load-pointer cfg id)]
(binding [pmap/*load-fn* (partial load-pointer conn id)]
(let [load-objects (fn [component] (let [load-objects (fn [component]
(ctf/load-component-objects data component)) (ctf/load-component-objects data component))
components-sample (-> (assets-sample (ctkl/components data) 4) components-sample (-> (assets-sample (ctkl/components data) 4)
@ -591,8 +538,11 @@
{:components components-sample {:components components-sample
:media (assets-sample (:media data) 3) :media (assets-sample (:media data) 3)
:colors (assets-sample (:colors data) 3) :colors (assets-sample (:colors data) 3)
:typographies (assets-sample (:typographies data) 3)})))] :typographies (assets-sample (:typographies data) 3)}))))
(defn- get-team-shared-files
[{:keys [::db/conn] :as cfg} {:keys [team-id profile-id]}]
(teams/check-read-permissions! conn profile-id team-id)
(->> (db/exec! conn [sql:team-shared-files team-id]) (->> (db/exec! conn [sql:team-shared-files team-id])
(into #{} (comp (into #{} (comp
(map decode-row) (map decode-row)
@ -602,8 +552,8 @@
(dissoc :media-id) (dissoc :media-id)
(assoc :thumbnail-uri (resolve-public-uri media-id))) (assoc :thumbnail-uri (resolve-public-uri media-id)))
(dissoc row :media-id)))) (dissoc row :media-id))))
(map #(assoc % :library-summary (library-summary %))) (map #(assoc % :library-summary (get-library-summary cfg %)))
(map #(dissoc % :data))))))) (map #(dissoc % :data))))))
(def ^:private schema:get-team-shared-files (def ^:private schema:get-team-shared-files
[:map {:title "get-team-shared-files"} [:map {:title "get-team-shared-files"}
@ -613,10 +563,8 @@
"Get all file (libraries) for the specified team." "Get all file (libraries) for the specified team."
{::doc/added "1.17" {::doc/added "1.17"
::sm/params schema:get-team-shared-files} ::sm/params schema:get-team-shared-files}
[{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id team-id]}] [cfg {:keys [::rpc/profile-id] :as params}]
(dm/with-open [conn (db/open pool)] (db/tx-run! cfg get-team-shared-files (assoc params :profile-id profile-id)))
(teams/check-read-permissions! conn profile-id team-id)
(get-team-shared-files conn team-id)))
;; --- COMMAND QUERY: get-file-libraries ;; --- COMMAND QUERY: get-file-libraries
@ -744,20 +692,15 @@
;; --- COMMAND QUERY: get-file-summary ;; --- COMMAND QUERY: get-file-summary
(sv/defmethod ::get-file-summary (defn- get-file-summary
"Retrieve a file summary by its ID. Only authenticated users." [{:keys [::db/conn] :as cfg} {:keys [profile-id id project-id] :as params}]
{::doc/added "1.20"
::sm/params schema:get-file}
[cfg {:keys [::rpc/profile-id id project-id] :as params}]
(db/tx-run! cfg
(fn [{:keys [::db/conn] :as cfg}]
(check-read-permissions! conn profile-id id) (check-read-permissions! conn profile-id id)
(let [team (teams/get-team conn (let [team (teams/get-team conn
:profile-id profile-id :profile-id profile-id
:project-id project-id :project-id project-id
:file-id id) :file-id id)
file (get-file conn id :project-id project-id)] file (get-file cfg id :project-id project-id)]
(-> (cfeat/get-team-enabled-features cf/flags team) (-> (cfeat/get-team-enabled-features cf/flags team)
(cfeat/check-client-features! (:features params)) (cfeat/check-client-features! (:features params))
@ -767,7 +710,14 @@
:components-count (count (ctkl/components-seq (:data file))) :components-count (count (ctkl/components-seq (:data file)))
:graphics-count (count (get-in file [:data :media] [])) :graphics-count (count (get-in file [:data :media] []))
:colors-count (count (get-in file [:data :colors] [])) :colors-count (count (get-in file [:data :colors] []))
:typography-count (count (get-in file [:data :typographies] []))})))) :typography-count (count (get-in file [:data :typographies] []))}))
(sv/defmethod ::get-file-summary
"Retrieve a file summary by its ID. Only authenticated users."
{::doc/added "1.20"
::sm/params schema:get-file}
[cfg {:keys [::rpc/profile-id] :as params}]
(db/tx-run! cfg get-file-summary (assoc params :profile-id profile-id)))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; MUTATION COMMANDS ;; MUTATION COMMANDS
@ -827,10 +777,15 @@
ORDER BY f.created_at ASC;") ORDER BY f.created_at ASC;")
(defn- absorb-library-by-file! (defn- absorb-library-by-file!
[conn ldata file-id] [cfg ldata file-id]
(binding [pmap/*load-fn* (partial load-pointer conn file-id)
(dm/assert!
"expected cfg with valid connection"
(db/connection-map? cfg))
(binding [pmap/*load-fn* (partial feat.fdata/load-pointer cfg file-id)
pmap/*tracked* (pmap/create-tracked)] pmap/*tracked* (pmap/create-tracked)]
(let [file (-> (get-file conn file-id (let [file (-> (get-file cfg file-id
:include-deleted? true :include-deleted? true
:lock-for-update? true) :lock-for-update? true)
(update :data ctf/absorb-assets ldata))] (update :data ctf/absorb-assets ldata))]
@ -839,42 +794,36 @@
:library-id (str (:id ldata)) :library-id (str (:id ldata))
:file-id (str file-id)) :file-id (str file-id))
(db/update! conn :file (db/update! cfg :file
{:revn (inc (:revn file)) {:revn (inc (:revn file))
:data (blob/encode (:data file)) :data (blob/encode (:data file))
:modified-at (dt/now)} :modified-at (dt/now)}
{:id file-id}) {:id file-id})
(persist-pointers! conn file-id)))) (feat.fdata/persist-pointers! cfg file-id))))
(defn- absorb-library! (defn- absorb-library!
"Find all files using a shared library, and absorb all library assets "Find all files using a shared library, and absorb all library assets
into the file local libraries" into the file local libraries"
[conn {:keys [id] :as library}] [cfg {:keys [id] :as library}]
(let [ldata (binding [pmap/*load-fn* (partial load-pointer conn id)]
(-> library (process-pointers deref) :data)) (dm/assert!
ids (->> (db/exec! conn [sql:get-referenced-files id]) "expected cfg with valid connection"
(db/connection-map? cfg))
(let [ldata (binding [pmap/*load-fn* (partial feat.fdata/load-pointer cfg id)]
(-> library :data (feat.fdata/process-pointers deref)))
ids (->> (db/exec! cfg [sql:get-referenced-files id])
(map :id))] (map :id))]
(l/trc :hint "absorbing library" (l/trc :hint "absorbing library"
:library-id (str id) :library-id (str id)
:files (str/join "," (map str ids))) :files (str/join "," (map str ids)))
(run! (partial absorb-library-by-file! conn ldata) ids))) (run! (partial absorb-library-by-file! cfg ldata) ids)))
(def ^:private (defn- set-file-shared
schema:set-file-shared [{:keys [::db/conn] :as cfg} {:keys [profile-id id] :as params}]
(sm/define
[:map {:title "set-file-shared"}
[:id ::sm/uuid]
[:is-shared :boolean]]))
(sv/defmethod ::set-file-shared
{::doc/added "1.17"
::webhooks/event? true
::sm/params schema:set-file-shared}
[{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id id] :as params}]
(db/with-atomic [conn pool]
(check-edition-permissions! conn profile-id id) (check-edition-permissions! conn profile-id id)
(let [file (db/get-by-id conn id {:columns [:id :name :is-shared]}) (let [file (db/get-by-id conn id {:columns [:id :name :is-shared]})
file (cond file (cond
@ -884,7 +833,7 @@
;; file, we need to perform more complex operation, ;; file, we need to perform more complex operation,
;; so in this case we retrieve the complete file and ;; so in this case we retrieve the complete file and
;; perform all required validations. ;; perform all required validations.
(let [file (-> (get-file conn id :lock-for-update? true) (let [file (-> (get-file cfg id :lock-for-update? true)
(check-version!) (check-version!)
(assoc :is-shared false)) (assoc :is-shared false))
team (teams/get-team conn team (teams/get-team conn
@ -895,7 +844,7 @@
(cfeat/check-client-features! (:features params)) (cfeat/check-client-features! (:features params))
(cfeat/check-file-features! (:features file))) (cfeat/check-file-features! (:features file)))
(absorb-library! conn file) (absorb-library! cfg file)
(db/delete! conn :file-library-rel {:library-file-id id}) (db/delete! conn :file-library-rel {:library-file-id id})
(db/update! conn :file (db/update! conn :file
@ -922,7 +871,21 @@
(select-keys file [:id :name :is-shared]) (select-keys file [:id :name :is-shared])
{::audit/props {:name (:name file) {::audit/props {:name (:name file)
:project-id (:project-id file) :project-id (:project-id file)
:is-shared (:is-shared file)}})))) :is-shared (:is-shared file)}})))
(def ^:private
schema:set-file-shared
(sm/define
[:map {:title "set-file-shared"}
[:id ::sm/uuid]
[:is-shared :boolean]]))
(sv/defmethod ::set-file-shared
{::doc/added "1.17"
::webhooks/event? true
::sm/params schema:set-file-shared}
[cfg {:keys [::rpc/profile-id] :as params}]
(db/tx-run! cfg set-file-shared (assoc params :profile-id profile-id)))
;; --- MUTATION COMMAND: delete-file ;; --- MUTATION COMMAND: delete-file
@ -939,12 +902,8 @@
[:map {:title "delete-file"} [:map {:title "delete-file"}
[:id ::sm/uuid]])) [:id ::sm/uuid]]))
(sv/defmethod ::delete-file (defn- delete-file
{::doc/added "1.17" [{:keys [::db/conn] :as cfg} {:keys [profile-id id] :as params}]
::webhooks/event? true
::sm/params schema:delete-file}
[{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id id] :as params}]
(db/with-atomic [conn pool]
(check-edition-permissions! conn profile-id id) (check-edition-permissions! conn profile-id id)
(let [file (mark-file-deleted! conn id)] (let [file (mark-file-deleted! conn id)]
@ -952,7 +911,7 @@
;; the whole file, proceed with feature checking and properly execute ;; the whole file, proceed with feature checking and properly execute
;; the absorb-library procedure ;; the absorb-library procedure
(when (:is-shared file) (when (:is-shared file)
(let [file (-> (get-file conn id (let [file (-> (get-file cfg id
:lock-for-update? true :lock-for-update? true
:include-deleted? true) :include-deleted? true)
(check-version!)) (check-version!))
@ -967,13 +926,20 @@
(cfeat/check-client-features! (:features params)) (cfeat/check-client-features! (:features params))
(cfeat/check-file-features! (:features file))) (cfeat/check-file-features! (:features file)))
(absorb-library! conn file))) (absorb-library! cfg file)))
(rph/with-meta (rph/wrap) (rph/with-meta (rph/wrap)
{::audit/props {:project-id (:project-id file) {::audit/props {:project-id (:project-id file)
:name (:name file) :name (:name file)
:created-at (:created-at file) :created-at (:created-at file)
:modified-at (:modified-at file)}})))) :modified-at (:modified-at file)}})))
(sv/defmethod ::delete-file
{::doc/added "1.17"
::webhooks/event? true
::sm/params schema:delete-file}
[cfg {:keys [::rpc/profile-id] :as params}]
(db/tx-run! cfg delete-file (assoc params :profile-id profile-id)))
;; --- MUTATION COMMAND: link-file-to-library ;; --- MUTATION COMMAND: link-file-to-library

17
backend/src/app/rpc/commands/files_create.clj
View file

@ -7,12 +7,14 @@
(ns app.rpc.commands.files-create (ns app.rpc.commands.files-create
(:require (:require
[app.common.data :as d] [app.common.data :as d]
[app.common.data.macros :as dm]
[app.common.features :as cfeat] [app.common.features :as cfeat]
[app.common.schema :as sm] [app.common.schema :as sm]
[app.common.types.file :as ctf] [app.common.types.file :as ctf]
[app.common.uuid :as uuid] [app.common.uuid :as uuid]
[app.config :as cf] [app.config :as cf]
[app.db :as db] [app.db :as db]
[app.features.fdata :as feat.fdata]
[app.loggers.audit :as-alias audit] [app.loggers.audit :as-alias audit]
[app.loggers.webhooks :as-alias webhooks] [app.loggers.webhooks :as-alias webhooks]
[app.rpc :as-alias rpc] [app.rpc :as-alias rpc]
@ -44,13 +46,20 @@
:or {is-shared false revn 0 create-page true} :or {is-shared false revn 0 create-page true}
:as params}] :as params}]
(dm/assert!
"expected a valid connection"
(db/connection? conn))
(let [id (or id (uuid/next)) (let [id (or id (uuid/next))
pointers (atom {}) pointers (pmap/create-tracked)
pmap? (contains? features "fdata/pointer-map")
omap? (contains? features "fdata/objects-map")
data (binding [pmap/*tracked* pointers data (binding [pmap/*tracked* pointers
cfeat/*current* features cfeat/*current* features
cfeat/*wrap-with-objects-map-fn* (if (features "fdata/objects-map") omap/wrap identity) cfeat/*wrap-with-objects-map-fn* (if omap? omap/wrap identity)
cfeat/*wrap-with-pointer-map-fn* (if (features "fdata/pointer-map") pmap/wrap identity)] cfeat/*wrap-with-pointer-map-fn* (if pmap? pmap/wrap identity)]
(if create-page (if create-page
(ctf/make-file-data id) (ctf/make-file-data id)
(ctf/make-file-data id nil))) (ctf/make-file-data id nil)))
@ -72,7 +81,7 @@
:deleted-at deleted-at}))] :deleted-at deleted-at}))]
(binding [pmap/*tracked* pointers] (binding [pmap/*tracked* pointers]
(files/persist-pointers! conn id)) (feat.fdata/persist-pointers! cfg id))
(->> (assoc params :file-id id :role :owner) (->> (assoc params :file-id id :role :owner)
(create-file-role! conn)) (create-file-role! conn))

29
backend/src/app/rpc/commands/files_thumbnails.clj
View file

@ -16,6 +16,7 @@
[app.common.types.shape-tree :as ctt] [app.common.types.shape-tree :as ctt]
[app.config :as cf] [app.config :as cf]
[app.db :as db] [app.db :as db]
[app.features.fdata :as feat.fdata]
[app.loggers.audit :as-alias audit] [app.loggers.audit :as-alias audit]
[app.loggers.webhooks :as-alias webhooks] [app.loggers.webhooks :as-alias webhooks]
[app.media :as media] [app.media :as media]
@ -100,28 +101,28 @@
;; loading all pages into memory for find the frame set for thumbnail. ;; loading all pages into memory for find the frame set for thumbnail.
(defn get-file-data-for-thumbnail (defn get-file-data-for-thumbnail
[conn {:keys [data id] :as file}] [{:keys [::db/conn] :as cfg} {:keys [data id] :as file}]
(letfn [;; function responsible on finding the frame marked to be (letfn [;; function responsible on finding the frame marked to be
;; used as thumbnail; the returned frame always have ;; used as thumbnail; the returned frame always have
;; the :page-id set to the page that it belongs. ;; the :page-id set to the page that it belongs.
(get-thumbnail-frame [data] (get-thumbnail-frame [file]
;; NOTE: this is a hack for avoid perform blocking ;; NOTE: this is a hack for avoid perform blocking
;; operation inside the for loop, clojure lazy-seq uses ;; operation inside the for loop, clojure lazy-seq uses
;; synchronized blocks that does not plays well with ;; synchronized blocks that does not plays well with
;; virtual threads, so we need to perform the load ;; virtual threads where all rpc methods calls are
;; operation first. This operation forces all pointer maps ;; dispatched, so we need to perform the load operation
;; load into the memory. ;; first. This operation forces all pointer maps load into
(->> (-> data :pages-index vals) ;; the memory.
(filter pmap/pointer-map?) ;;
(run! pmap/load!)) ;; FIXME: this is no longer true with clojure>=1.12
(let [{:keys [data]} (update file :data feat.fdata/process-pointers pmap/load!)]
;; Then proceed to find the frame set for thumbnail ;; Then proceed to find the frame set for thumbnail
(d/seek #(or (:use-for-thumbnail %) (d/seek #(or (:use-for-thumbnail %)
(:use-for-thumbnail? %)) ; NOTE: backward comp (remove on v1.21) (:use-for-thumbnail? %)) ; NOTE: backward comp (remove on v1.21)
(for [page (-> data :pages-index vals) (for [page (-> data :pages-index vals)
frame (-> page :objects ctt/get-frames)] frame (-> page :objects ctt/get-frames)]
(assoc frame :page-id (:id page))))) (assoc frame :page-id (:id page))))))
;; function responsible to filter objects data structure of ;; function responsible to filter objects data structure of
;; all unneeded shapes if a concrete frame is provided. If no ;; all unneeded shapes if a concrete frame is provided. If no
@ -165,8 +166,8 @@
objects)))] objects)))]
(binding [pmap/*load-fn* (partial files/load-pointer conn id)] (binding [pmap/*load-fn* (partial feat.fdata/load-pointer cfg id)]
(let [frame (get-thumbnail-frame data) (let [frame (get-thumbnail-frame file)
frame-id (:id frame) frame-id (:id frame)
page-id (or (:page-id frame) page-id (or (:page-id frame)
(-> data :pages first)) (-> data :pages first))
@ -220,7 +221,7 @@
:profile-id profile-id :profile-id profile-id
:file-id file-id) :file-id file-id)
file (files/get-file conn file-id)] file (files/get-file cfg file-id)]
(-> (cfeat/get-team-enabled-features cf/flags team) (-> (cfeat/get-team-enabled-features cf/flags team)
(cfeat/check-client-features! (:features params)) (cfeat/check-client-features! (:features params))
@ -228,7 +229,7 @@
{:file-id file-id {:file-id file-id
:revn (:revn file) :revn (:revn file)
:page (get-file-data-for-thumbnail conn file)})))) :page (get-file-data-for-thumbnail cfg file)}))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; MUTATION COMMANDS ;; MUTATION COMMANDS

24
backend/src/app/rpc/commands/files_update.clj
View file

@ -17,7 +17,7 @@
[app.common.uuid :as uuid] [app.common.uuid :as uuid]
[app.config :as cf] [app.config :as cf]
[app.db :as db] [app.db :as db]
[app.features.fdata :refer [enable-pointer-map enable-objects-map]] [app.features.fdata :as feat.fdata]
[app.http.errors :as errors] [app.http.errors :as errors]
[app.loggers.audit :as audit] [app.loggers.audit :as audit]
[app.loggers.webhooks :as webhooks] [app.loggers.webhooks :as webhooks]
@ -106,12 +106,12 @@
(defn- wrap-with-pointer-map-context (defn- wrap-with-pointer-map-context
[f] [f]
(fn [{:keys [::db/conn] :as cfg} {:keys [id] :as file}] (fn [cfg {:keys [id] :as file}]
(binding [pmap/*tracked* (atom {}) (binding [pmap/*tracked* (pmap/create-tracked)
pmap/*load-fn* (partial files/load-pointer conn id) pmap/*load-fn* (partial feat.fdata/load-pointer cfg id)
cfeat/*wrap-with-pointer-map-fn* pmap/wrap] cfeat/*wrap-with-pointer-map-fn* pmap/wrap]
(let [result (f cfg file)] (let [result (f cfg file)]
(files/persist-pointers! conn id) (feat.fdata/persist-pointers! cfg id)
result)))) result))))
(defn- wrap-with-objects-map-context (defn- wrap-with-objects-map-context
@ -236,7 +236,7 @@
;; to be executed on a separated executor for avoid to do the ;; to be executed on a separated executor for avoid to do the
;; CPU intensive operation on vthread. ;; CPU intensive operation on vthread.
update-fdata-fn (partial update-file-data conn file changes skip-validate) update-fdata-fn (partial update-file-data cfg file changes skip-validate)
file (-> (climit/configure cfg :update-file/global) file (-> (climit/configure cfg :update-file/global)
(climit/run! update-fdata-fn executor))] (climit/run! update-fdata-fn executor))]
@ -290,7 +290,7 @@
file) file)
(defn- update-file-data (defn- update-file-data
[conn file changes skip-validate] [{:keys [::db/conn] :as cfg} file changes skip-validate]
(let [file (update file :data (fn [data] (let [file (update file :data (fn [data]
(-> data (-> data
(blob/decode) (blob/decode)
@ -304,10 +304,10 @@
(not skip-validate)) (not skip-validate))
(->> (files/get-file-libraries conn (:id file)) (->> (files/get-file-libraries conn (:id file))
(into [file] (map (fn [{:keys [id]}] (into [file] (map (fn [{:keys [id]}]
(binding [pmap/*load-fn* (partial files/load-pointer conn id) (binding [pmap/*load-fn* (partial feat.fdata/load-pointer cfg id)
pmap/*tracked* nil] pmap/*tracked* nil]
(-> (files/get-file conn id :migrate? false) (-> (files/get-file cfg id :migrate? false)
(files/process-pointers deref) ; ensure all pointers resolved (feat.fdata/process-pointers deref) ; ensure all pointers resolved
(fmg/migrate-file)))))) (fmg/migrate-file))))))
(d/index-by :id)))] (d/index-by :id)))]
@ -332,11 +332,11 @@
(cond-> (and (contains? cfeat/*current* "fdata/objects-map") (cond-> (and (contains? cfeat/*current* "fdata/objects-map")
(not (contains? cfeat/*previous* "fdata/objects-map"))) (not (contains? cfeat/*previous* "fdata/objects-map")))
(enable-objects-map)) (feat.fdata/enable-objects-map))
(cond-> (and (contains? cfeat/*current* "fdata/pointer-map") (cond-> (and (contains? cfeat/*current* "fdata/pointer-map")
(not (contains? cfeat/*previous* "fdata/pointer-map"))) (not (contains? cfeat/*previous* "fdata/pointer-map")))
(enable-pointer-map)) (feat.fdata/enable-pointer-map))
(update :data blob/encode)))) (update :data blob/encode))))

58
backend/src/app/rpc/commands/management.clj
View file

@ -14,11 +14,11 @@
[app.common.schema :as sm] [app.common.schema :as sm]
[app.common.uuid :as uuid] [app.common.uuid :as uuid]
[app.db :as db] [app.db :as db]
[app.features.fdata :as feat.fdata]
[app.http.sse :as sse] [app.http.sse :as sse]
[app.loggers.webhooks :as-alias webhooks] [app.loggers.webhooks :as-alias webhooks]
[app.rpc :as-alias rpc] [app.rpc :as-alias rpc]
[app.rpc.commands.binfile :as binfile] [app.rpc.commands.binfile :as binfile]
[app.rpc.commands.files :as files]
[app.rpc.commands.projects :as proj] [app.rpc.commands.projects :as proj]
[app.rpc.commands.teams :as teams :refer [create-project-role create-project]] [app.rpc.commands.teams :as teams :refer [create-project-role create-project]]
[app.rpc.doc :as-alias doc] [app.rpc.doc :as-alias doc]
@ -49,9 +49,8 @@
{::doc/added "1.16" {::doc/added "1.16"
::webhooks/event? true ::webhooks/event? true
::sm/params schema:duplicate-file} ::sm/params schema:duplicate-file}
[{:keys [::db/pool] :as cfg} {:keys [::rpc/profile-id] :as params}] [cfg {:keys [::rpc/profile-id] :as params}]
(db/with-atomic [conn pool] (db/tx-run! cfg duplicate-file (assoc params :profile-id profile-id)))
(duplicate-file conn (assoc params :profile-id profile-id))))
(defn- remap-id (defn- remap-id
[item index key] [item index key]
@ -60,7 +59,7 @@
(assoc key (get index (get item key) (get item key))))) (assoc key (get index (get item key) (get item key)))))
(defn- process-file (defn- process-file
[conn {:keys [id] :as file} index] [cfg index {:keys [id] :as file}]
(letfn [(process-form [form] (letfn [(process-form [form]
(cond-> form (cond-> form
;; Relink library items ;; Relink library items
@ -103,26 +102,28 @@
(dissoc k)) (dissoc k))
res))) res)))
media media
media))] media))
(-> file
(update :id #(get index %)) (update-fdata [fdata new-id]
(update :data (-> fdata
(fn [data]
(binding [pmap/*load-fn* (partial files/load-pointer conn id)
pmap/*tracked* (atom {})]
(let [file-id (get index id)
data (-> data
(blob/decode) (blob/decode)
(assoc :id file-id) (assoc :id new-id)
(pmg/migrate-data) (pmg/migrate-data)
(update :pages-index relink-shapes) (update :pages-index relink-shapes)
(update :components relink-shapes) (update :components relink-shapes)
(update :media relink-media) (update :media relink-media)
(d/without-nils) (d/without-nils)
(files/process-pointers pmap/clone) (feat.fdata/process-pointers pmap/clone)
(blob/encode))] (blob/encode)))]
(files/persist-pointers! conn file-id)
data))))))) (binding [pmap/*load-fn* (partial feat.fdata/load-pointer cfg id)
pmap/*tracked* (pmap/create-tracked)]
(let [new-id (get index id)
file (-> file
(assoc :id new-id)
(update :data update-fdata new-id))]
(feat.fdata/persist-pointers! cfg new-id)
file))))
(def sql:get-used-libraries (def sql:get-used-libraries
"select flr.* "select flr.*
@ -139,7 +140,7 @@
and so.deleted_at is null") and so.deleted_at is null")
(defn duplicate-file* (defn duplicate-file*
[conn {:keys [profile-id file index project-id name flibs fmeds]} {:keys [reset-shared-flag]}] [{:keys [::db/conn] :as cfg} {:keys [profile-id file index project-id name flibs fmeds]} {:keys [reset-shared-flag]}]
(let [flibs (or flibs (db/exec! conn [sql:get-used-libraries (:id file)])) (let [flibs (or flibs (db/exec! conn [sql:get-used-libraries (:id file)]))
fmeds (or fmeds (db/exec! conn [sql:get-used-media-objects (:id file)])) fmeds (or fmeds (db/exec! conn [sql:get-used-media-objects (:id file)]))
@ -182,7 +183,7 @@
(assoc :modified-at now) (assoc :modified-at now)
(assoc :ignore-sync-until ignore)) (assoc :ignore-sync-until ignore))
file (process-file conn file index)] file (process-file cfg index file)]
(db/insert! conn :file file) (db/insert! conn :file file)
(db/insert! conn :file-profile-rel (db/insert! conn :file-profile-rel
@ -201,13 +202,15 @@
file)) file))
(defn duplicate-file (defn duplicate-file
[conn {:keys [profile-id file-id] :as params}] [{:keys [::db/conn] :as cfg} {:keys [profile-id file-id] :as params}]
(let [file (db/get-by-id conn :file file-id) (let [file (db/get-by-id conn :file file-id)
index {file-id (uuid/next)} index {file-id (uuid/next)}
params (assoc params :index index :file file)] params (assoc params :index index :file file)]
(proj/check-edition-permissions! conn profile-id (:project-id file)) (proj/check-edition-permissions! conn profile-id (:project-id file))
(db/exec-one! conn ["SET CONSTRAINTS ALL DEFERRED"]) (db/exec-one! conn ["SET CONSTRAINTS ALL DEFERRED"])
(-> (duplicate-file* conn params {:reset-shared-flag true})
;; FIXME: why we decode the data here?
(-> (duplicate-file* cfg params {:reset-shared-flag true})
(update :data blob/decode) (update :data blob/decode)
(update :features db/decode-pgarray #{})))) (update :features db/decode-pgarray #{}))))
@ -227,12 +230,11 @@
{::doc/added "1.16" {::doc/added "1.16"
::webhooks/event? true ::webhooks/event? true
::sm/params schema:duplicate-project} ::sm/params schema:duplicate-project}
[{:keys [::db/pool] :as cfg} params] [cfg {:keys [::rpc/profile-id] :as params}]
(db/with-atomic [conn pool] (db/tx-run! cfg duplicate-project (assoc params :profile-id profile-id)))
(duplicate-project conn (assoc params :profile-id (::rpc/profile-id params)))))
(defn duplicate-project (defn duplicate-project
[conn {:keys [profile-id project-id name] :as params}] [{:keys [::db/conn] :as cfg} {:keys [profile-id project-id name] :as params}]
;; Defer all constraints ;; Defer all constraints
(db/exec-one! conn ["SET CONSTRAINTS ALL DEFERRED"]) (db/exec-one! conn ["SET CONSTRAINTS ALL DEFERRED"])
@ -270,7 +272,7 @@
(let [file (db/get-by-id conn :file id) (let [file (db/get-by-id conn :file id)
params (assoc params :file file) params (assoc params :file file)
opts {:reset-shared-flag false}] opts {:reset-shared-flag false}]
(duplicate-file* conn params opts)))) (duplicate-file* cfg params opts))))
;; return the created project ;; return the created project
project)) project))

2
backend/src/app/rpc/commands/viewer.clj
View file

@ -29,7 +29,7 @@
(defn- get-view-only-bundle (defn- get-view-only-bundle
[{:keys [::db/conn] :as cfg} {:keys [profile-id file-id ::perms] :as params}] [{:keys [::db/conn] :as cfg} {:keys [profile-id file-id ::perms] :as params}]
(let [file (files/get-file conn file-id) (let [file (files/get-file cfg file-id)
project (db/get conn :project project (db/get conn :project
{:id (:project-id file)} {:id (:project-id file)}

128
backend/src/app/srepl/helpers.clj
View file

@ -23,6 +23,7 @@
[app.config :as cfg] [app.config :as cfg]
[app.db :as db] [app.db :as db]
[app.db.sql :as sql] [app.db.sql :as sql]
[app.features.fdata :as feat.fdata]
[app.main :refer [system]] [app.main :refer [system]]
[app.rpc.commands.files :as files] [app.rpc.commands.files :as files]
[app.rpc.commands.files-update :as files-update] [app.rpc.commands.files-update :as files-update]
@ -39,7 +40,6 @@
[promesa.exec :as px] [promesa.exec :as px]
[promesa.exec.csp :as sp])) [promesa.exec.csp :as sp]))
(def ^:dynamic *conn* nil)
(def ^:dynamic *system* nil) (def ^:dynamic *system* nil)
(defn println! (defn println!
@ -63,71 +63,75 @@
(defn get-file (defn get-file
"Get the migrated data of one file." "Get the migrated data of one file."
[system id] [system id & {:keys [migrate?] :or {migrate? true}}]
(db/run! system (db/run! system
(fn [{:keys [::db/conn]}] (fn [system]
(binding [pmap/*load-fn* (partial files/load-pointer conn id)] (binding [pmap/*load-fn* (partial feat.fdata/load-pointer system id)]
(-> (files/get-file conn id) (-> (files/get-file system id :migrate? migrate?)
(files/process-pointers deref)))))) (update :data feat.fdata/process-pointers deref))))))
(defn validate (defn validate
"Validate structure, referencial integrity and semantic coherence of "Validate structure, referencial integrity and semantic coherence of
all contents of a file. Returns a list of errors." all contents of a file. Returns a list of errors."
[system id] [system id]
(db/with-atomic [conn (:app.db/pool system)] (db/tx-run! system
(binding [pmap/*load-fn* (partial files/load-pointer conn id)] (fn [{:keys [::db/conn] :as system}]
(let [file (files/get-file conn id) (binding [pmap/*load-fn* (partial feat.fdata/load-pointer system id)]
libraries (->> (files/get-file-libraries conn id) (let [id (if (string? id) (parse-uuid id) id)
file (files/get-file system 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 files/load-pointer conn id)] (binding [pmap/*load-fn* (partial feat.fdata/load-pointer system id)]
(-> (db/get conn :file {:id id}) (-> (files/get-file system id :migrate? false)
(files/decode-row) (feat.fdata/process-pointers deref)
(files/process-pointers deref) ; ensure all pointers resolved
(pmg/migrate-file)))))) (pmg/migrate-file))))))
(d/index-by :id))] (d/index-by :id))]
(validate/validate-file file libraries))))) (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."
[system id] [system id]
(db/with-atomic [conn (:app.db/pool system)] (db/tx-run! system
(let [file (files/get-file conn id)] (fn [{:keys [::db/conn] :as system}]
(binding [*conn* conn (binding [pmap/*tracked* (pmap/create-tracked)
pmap/*tracked* (atom {}) pmap/*load-fn* (partial feat.fdata/load-pointer system id)]
pmap/*load-fn* (partial files/load-pointer conn id)] (let [id (if (string? id) (parse-uuid id) id)
(let [libraries (->> (files/get-file-libraries conn id) file (files/get-file system 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 files/load-pointer conn id)] (binding [pmap/*load-fn* (partial feat.fdata/load-pointer system id)]
(-> (db/get conn :file {:id id}) (-> (files/get-file system id :migrate? false)
(files/decode-row) (feat.fdata/process-pointers deref)
(files/process-pointers deref) ; ensure all pointers resolved
(pmg/migrate-file)))))) (pmg/migrate-file))))))
(d/index-by :id)) (d/index-by :id))
errors (validate/validate-file file libraries) errors (validate/validate-file file libs)
changes (-> (repair/repair-file (:data file) libraries errors) changes (-> (repair/repair-file (:data file) libs errors) :redo-changes)
(get :redo-changes))
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))] (update :data blob/encode))]
(files/persist-pointers! conn id) (when (contains? (:features file) "fdata/pointer-map")
(feat.fdata/persist-pointers! system id))
(db/update! conn :file (db/update! conn :file
{:revn (:revn file) {:revn (:revn file)
:data (:data file) :data (: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)))))
(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."
[system & {:keys [update-fn id rollback? migrate? inc-revn?] [system & {:keys [update-fn id rollback? migrate? inc-revn?]
:or {rollback? true migrate? true inc-revn? true}}] :or {rollback? true migrate? true inc-revn? true}}]
(letfn [(process-file [conn {:keys [features] :as file}] (letfn [(process-file [{:keys [::db/conn] :as system} {:keys [features] :as file}]
(binding [pmap/*tracked* (atom {}) (binding [pmap/*tracked* (pmap/create-tracked)
pmap/*load-fn* (partial files/load-pointer conn id) pmap/*load-fn* (partial feat.fdata/load-pointer system id)
cfeat/*wrap-with-pointer-map-fn* cfeat/*wrap-with-pointer-map-fn*
(if (contains? features "fdata/pointer-map") pmap/wrap identity) (if (contains? features "fdata/pointer-map") pmap/wrap identity)
cfeat/*wrap-with-objects-map-fn* cfeat/*wrap-with-objects-map-fn*
@ -145,19 +149,19 @@
{:id id})) {:id id}))
(when (contains? (:features file) "fdata/pointer-map") (when (contains? (:features file) "fdata/pointer-map")
(files/persist-pointers! conn id)) (feat.fdata/persist-pointers! system id))
(dissoc file :data)))] (dissoc file :data)))]
(db/tx-run! system (db/tx-run! system
(fn [{:keys [::db/conn] :as system}] (fn [system]
(binding [*conn* conn *system* system] (binding [*system* system]
(try (try
(->> (files/get-file conn id :migrate? migrate?) (->> (files/get-file system id :migrate? migrate?)
(process-file conn)) (process-file system))
(finally (finally
(when rollback? (when rollback?
(db/rollback! conn))))))))) (db/rollback! system)))))))))
(defn analyze-files (defn analyze-files
"Apply a function to all files in the database, reading them in "Apply a function to all files in the database, reading them in
@ -187,17 +191,17 @@
(println "unexpected exception happened on processing file: " (:id file)) (println "unexpected exception happened on processing file: " (:id file))
(strace/print-stack-trace cause)) (strace/print-stack-trace cause))
(process-file [conn file-id] (process-file [{:keys [::db/conn] :as system} file-id]
(let [file (binding [pmap/*load-fn* (partial files/load-pointer conn file-id)] (let [file (binding [pmap/*load-fn* (partial feat.fdata/load-pointer system file-id)]
(-> (files/get-file conn file-id) (-> (files/get-file system file-id)
(files/process-pointers deref))) (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 files/load-pointer conn id)] (binding [pmap/*load-fn* (partial feat.fdata/load-pointer system id)]
(-> (files/get-file conn id) (-> (files/get-file system id)
(files/process-pointers deref)))))) (update :data feat.fdata/process-pointers deref))))))
(d/index-by :id)))] (d/index-by :id)))]
(try (try
(if with-libraries? (if with-libraries?
@ -209,19 +213,19 @@
(db/tx-run! system (db/tx-run! system
(fn [{:keys [::db/conn] :as system}] (fn [{:keys [::db/conn] :as system}]
(try (try
(binding [*conn* conn *system* system] (binding [*system* system]
(when (fn? on-init) (on-init)) (when (fn? on-init) (on-init))
(run! (partial process-file conn) (get-candidates conn))) (run! (partial process-file system) (get-candidates conn)))
(finally (finally
(when (fn? on-end) (when (fn? on-end)
(ex/ignoring (on-end))) (ex/ignoring (on-end)))
(db/rollback! conn))))))) (db/rollback! system)))))))
(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, reading them in
batches." batches."
[{:keys [::db/pool] :as system} & {:keys [chunk-size [system & {:keys [chunk-size
max-items max-items
workers workers
start-at start-at
@ -252,11 +256,11 @@
(println! "unexpected exception happened on processing file: " (:id file)) (println! "unexpected exception happened on processing file: " (:id file))
(strace/print-stack-trace cause)) (strace/print-stack-trace cause))
(process-file [conn file-id] (process-file [system file-id]
(try (try
(let [{:keys [features] :as file} (files/get-file conn file-id)] (let [{:keys [features] :as file} (files/get-file system file-id)]
(binding [pmap/*tracked* (atom {}) (binding [pmap/*tracked* (pmap/create-tracked)
pmap/*load-fn* (partial files/load-pointer conn file-id) pmap/*load-fn* (partial feat.fdata/load-pointer system file-id)
cfeat/*wrap-with-pointer-map-fn* cfeat/*wrap-with-pointer-map-fn*
(if (contains? features "fdata/pointer-map") pmap/wrap identity) (if (contains? features "fdata/pointer-map") pmap/wrap identity)
cfeat/*wrap-with-objects-map-fn* cfeat/*wrap-with-objects-map-fn*
@ -265,30 +269,30 @@
(on-file file) (on-file file)
(when (contains? features "fdata/pointer-map") (when (contains? features "fdata/pointer-map")
(files/persist-pointers! conn file-id)))) (feat.fdata/persist-pointers! system file-id))))
(catch Throwable cause (catch Throwable cause
((or on-error on-error*) cause file-id)))) ((or on-error on-error*) cause file-id))))
(run-worker [in index] (run-worker [in index]
(db/tx-run! system (db/tx-run! system
(fn [{:keys [::db/conn] :as system}] (fn [system]
(binding [*conn* conn *system* system] (binding [*system* system]
(loop [i 0] (loop [i 0]
(when-let [file-id (sp/take! in)] (when-let [file-id (sp/take! in)]
(println! "=> worker: index:" index "| loop:" i "| file:" (str file-id) "|" (px/get-name)) (println! "=> worker: index:" index "| loop:" i "| file:" (str file-id) "|" (px/get-name))
(process-file conn file-id) (process-file system file-id)
(recur (inc i))))) (recur (inc i)))))
(when rollback? (when rollback?
(db/rollback! conn))))) (db/rollback! system)))))
(run-producer [input] (run-producer [input]
(db/with-atomic [conn pool] (db/tx-run! system (fn [{:keys [::db/conn]}]
(doseq [file-id (get-candidates conn)] (doseq [file-id (get-candidates conn)]
(println! "=> producer:" file-id "|" (px/get-name)) (println! "=> producer:" file-id "|" (px/get-name))
(sp/put! input file-id)) (sp/put! input file-id))
(sp/close! input)))] (sp/close! input))))]
(when (fn? on-init) (on-init)) (when (fn? on-init) (on-init))

14
backend/src/app/tasks/file_gc.clj
View file

@ -19,8 +19,8 @@
[app.common.types.shape-tree :as ctt] [app.common.types.shape-tree :as ctt]
[app.config :as cf] [app.config :as cf]
[app.db :as db] [app.db :as db]
[app.features.fdata :as feat.fdata]
[app.media :as media] [app.media :as media]
[app.rpc.commands.files :as files]
[app.storage :as sto] [app.storage :as sto]
[app.util.blob :as blob] [app.util.blob :as blob]
[app.util.pointer-map :as pmap] [app.util.pointer-map :as pmap]
@ -271,9 +271,9 @@
" limit 1;") " limit 1;")
rows (db/exec! conn [sql file-id cursor])] rows (db/exec! conn [sql file-id cursor])]
[(some-> rows peek :created-at) [(some-> rows peek :created-at)
(mapcat (comp files/get-all-pointer-ids blob/decode :data) rows)]))] (mapcat (comp feat.fdata/get-used-pointer-ids blob/decode :data) rows)]))]
(let [used (into (files/get-all-pointer-ids data) (let [used (into (feat.fdata/get-used-pointer-ids data)
(d/iteration get-pointers-chunk (d/iteration get-pointers-chunk
:vf second :vf second
:kf first :kf first
@ -290,10 +290,10 @@
(defn- process-file (defn- process-file
[{:keys [::db/conn] :as cfg} {:keys [id data revn modified-at features] :as file}] [{:keys [::db/conn] :as cfg} {:keys [id data revn modified-at features] :as file}]
(l/dbg :hint "processing file" :id id :modified-at modified-at) (l/dbg :hint "processing file" :file-id (str id) :modified-at modified-at)
(binding [pmap/*load-fn* (partial files/load-pointer conn id) (binding [pmap/*load-fn* (partial feat.fdata/load-pointer cfg id)
pmap/*tracked* (atom {})] pmap/*tracked* (pmap/create-tracked)]
(let [data (-> (blob/decode data) (let [data (-> (blob/decode data)
(assoc :id id) (assoc :id id)
(pmg/migrate-data))] (pmg/migrate-data))]
@ -311,4 +311,4 @@
{:has-media-trimmed true} {:has-media-trimmed true}
{:id id}) {:id id})
(files/persist-pointers! conn id)))) (feat.fdata/persist-pointers! cfg id))))