Merge pull request #5053 from penpot/niwinz-update-file-refactor

♻️ Refactor file-update for make it more reusable

backend/scripts/repl

@@ -24,7 +24,7 @@ export PENPOT_FLAGS="\
        enable-rpc-climit \
        enable-rpc-rlimit \
        enable-soft-rpc-rlimit \
-       enable-file-snapshot \
+       enable-auto-file-snapshot \
        enable-webhooks \
        enable-access-tokens \
        enable-tiered-file-data-storage \

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

@@ -38,6 +38,20 @@
    [clojure.set :as set]
    [promesa.exec :as px]))
 
+(declare ^:private get-lagged-changes)
+(declare ^:private send-notifications!)
+(declare ^:private update-file)
+(declare ^:private update-file*)
+(declare ^:private process-changes-and-validate)
+(declare ^:private take-snapshot?)
+(declare ^:private delete-old-snapshots!)
+
+;; PUBLIC API; intended to be used outside of this module
+(declare update-file!)
+(declare update-file-data!)
+(declare persist-file!)
+(declare get-file)
+
 ;; --- SCHEMA
 
 (def ^:private
@@ -97,41 +111,6 @@
   (or (contains? library-change-types type)
       (contains? file-change-types type)))
 
-(def ^:private sql:get-file
-  "SELECT f.*, p.team_id
-     FROM file AS f
-     JOIN project AS p ON (p.id = f.project_id)
-    WHERE f.id = ?
-      AND (f.deleted_at IS NULL OR
-           f.deleted_at > now())
-      FOR KEY SHARE")
-
-(defn get-file
-  [conn id]
-  (let [file (db/exec-one! conn [sql:get-file id])]
-    (when-not file
-      (ex/raise :type :not-found
-                :code :object-not-found
-                :hint (format "file with id '%s' does not exists" id)))
-    (update file :features db/decode-pgarray #{})))
-
-(defn- wrap-with-pointer-map-context
-  [f]
-  (fn [cfg {:keys [id] :as file}]
-    (binding [pmap/*tracked* (pmap/create-tracked)
-              pmap/*load-fn* (partial feat.fdata/load-pointer cfg id)]
-      (let [result (f cfg file)]
-        (feat.fdata/persist-pointers! cfg id)
-        result))))
-
-(declare ^:private delete-old-snapshots!)
-(declare ^:private get-lagged-changes)
-(declare ^:private send-notifications!)
-(declare ^:private take-snapshot?)
-(declare ^:private update-file)
-(declare ^:private update-file*)
-(declare ^:private update-file-data)
-
 ;; If features are specified from params and the final feature
 ;; set is different than the persisted one, update it on the
 ;; database.
@@ -147,7 +126,8 @@
    ::sm/result schema:update-file-result
    ::doc/module :files
    ::doc/added "1.17"}
-  [cfg {:keys [::rpc/profile-id id] :as params}]
+  [{:keys [::mtx/metrics] :as cfg}
+   {:keys [::rpc/profile-id id changes changes-with-metadata] :as params}]
   (db/tx-run! cfg (fn [{:keys [::db/conn] :as cfg}]
                     (files/check-edition-permissions! conn profile-id id)
                     (db/xact-lock! conn id)
@@ -161,43 +141,21 @@
                                        (cfeat/check-client-features! (:features params))
                                        (cfeat/check-file-features! (:features file) (:features params)))
 
-                          params   (assoc params
+                          changes  (if changes-with-metadata
-                                          :profile-id profile-id
+                                     (->> changes-with-metadata (mapcat :changes) vec)
-                                          :features features
+                                     (vec changes))
-                                          :team team
+
-                                          :file file)
+                          params   (-> params
+                                       (assoc :profile-id profile-id)
+                                       (assoc :features features)
+                                       (assoc :team team)
+                                       (assoc :file file)
+                                       (assoc :changes changes))
+
+                          cfg      (assoc cfg ::timestamp (dt/now))
 
                           tpoint   (dt/tpoint)]
 
-                      ;; When newly computed features does not match exactly with
-                      ;; the features defined on team row, we update it.
-                      (when (not= features (:features team))
-                        (let [features (db/create-array conn "text" features)]
-                          (db/update! conn :team
-                                      {:features features}
-                                      {:id (:id team)})))
-
-                      (binding [l/*context* (some-> (meta params)
-                                                    (get :app.http/request)
-                                                    (errors/request->context))]
-                        (-> (update-file cfg params)
-                            (rph/with-defer #(let [elapsed (tpoint)]
-                                               (l/trace :hint "update-file" :time (dt/format-duration elapsed))))))))))
-
-(defn update-file
-  [{:keys [::mtx/metrics] :as cfg}
-   {:keys [file features changes changes-with-metadata] :as params}]
-  (let [features  (-> features
-                      (set/difference cfeat/frontend-only-features)
-                      (set/union (:features file)))
-
-        update-fn (cond-> update-file*
-                    (contains? features "fdata/pointer-map")
-                    (wrap-with-pointer-map-context))
-
-        changes   (if changes-with-metadata
-                    (->> changes-with-metadata (mapcat :changes) vec)
-                    (vec changes))]
 
                       (when (> (:revn params)
                                (:revn file))
@@ -207,59 +165,133 @@
                                   :context {:incoming-revn (:revn params)
                                             :stored-revn (:revn file)}))
 
+                      ;; When newly computed features does not match exactly with
+                      ;; the features defined on team row, we update it.
+                      (when (not= features (:features team))
+                        (let [features (db/create-array conn "text" features)]
+                          (db/update! conn :team
+                                      {:features features}
+                                      {:id (:id team)})))
+
                       (mtx/run! metrics {:id :update-file-changes :inc (count changes)})
 
-    (binding [cfeat/*current*  features
+                      (binding [l/*context* (some-> (meta params)
-              cfeat/*previous* (:features file)]
+                                                    (get :app.http/request)
-      (let [file   (assoc file :features features)
+                                                    (errors/request->context))]
-            params (-> params
+                        (-> (update-file* cfg params)
-                       (assoc :file file)
+                            (rph/with-defer #(let [elapsed (tpoint)]
-                       (assoc :changes changes)
+                                               (l/trace :hint "update-file" :time (dt/format-duration elapsed))))))))))
-                       (assoc ::created-at (dt/now)))]
-
-        (-> (update-fn cfg params)
-            (vary-meta assoc ::audit/replace-props
-                       {:id         (:id file)
-                        :name       (:name file)
-                        :features   (:features file)
-                        :project-id (:project-id file)
-                        :team-id    (:team-id file)}))))))
 
 (defn- update-file*
-  [{:keys [::db/conn ::wrk/executor] :as cfg}
+  "Internal function, part of the update-file process, that encapsulates
-   {:keys [profile-id file changes session-id ::created-at skip-validate] :as params}]
+  the changes application offload to a separated thread and emit all
+  corresponding notifications.
+
+  Follow the inner implementation to `update-file-data!` function.
+
+  Only intended for internal use on this module."
+  [{:keys [::db/conn ::wrk/executor ::timestamp] :as cfg}
+   {:keys [profile-id file features changes session-id skip-validate] :as params}]
+
   (let [;; Retrieve the file data
         file (feat.fdata/resolve-file-data cfg file)
 
+        file (assoc file :features
+                    (-> features
+                        (set/difference cfeat/frontend-only-features)
+                        (set/union (:features file))))
+
         ;; Process the file data on separated thread for avoid to do
         ;; the CPU intensive operation on vthread.
+        file  (px/invoke! executor
+                          (fn []
+                            (binding [cfeat/*current*  features
+                                      cfeat/*previous* (:features file)]
+                              (update-file-data! cfg file
+                                                 process-changes-and-validate
+                                                 changes skip-validate))))]
 
-        file     (px/invoke! executor (partial update-file-data cfg file changes skip-validate))
+    (when (feat.fdata/offloaded? file)
+      (let [storage (sto/resolve cfg ::db/reuse-conn true)]
+        (some->> (:data-ref-id file) (sto/touch-object! storage))))
+
+    ;; TODO: move this to asynchronous task
+    (when (::snapshot-data file)
+      (delete-old-snapshots! cfg file))
+
+    (persist-file! cfg file)
+
+    (let [params   (assoc params :file file)
+          response {:revn (:revn file)
+                    :lagged (get-lagged-changes conn params)}
           features (db/create-array conn "text" (:features file))]
 
-    ;; NOTE: if file was offloaded, we need to touch the referenced
+      ;; Insert change (xlog)
-    ;; storage object because on this update operation the data will
-    ;; be overwritted.
-    (when (= "objects-storage" (:data-backend file))
-      (let [storage (sto/resolve cfg ::db/reuse-conn true)]
-        (sto/touch-object! storage (:data-ref-id file))))
-
       (db/insert! conn :file-change
                   {:id (uuid/next)
                    :session-id session-id
                    :profile-id profile-id
-                 :created-at created-at
+                   :created-at timestamp
                    :file-id (:id file)
                    :revn (:revn file)
                    :version (:version file)
+                   :features features
                    :label (::snapshot-label file)
                    :data (::snapshot-data file)
-                 :features (db/create-array conn "text" (:features file))
                    :changes (blob/encode changes)}
                   {::db/return-keys false})
 
-    (when (::snapshot-data file)
+      ;; Send asynchronous notifications
-      (delete-old-snapshots! cfg file))
+      (send-notifications! cfg params)
+
+      (vary-meta response assoc ::audit/replace-props
+                 {:id         (:id file)
+                  :name       (:name file)
+                  :features   (:features file)
+                  :project-id (:project-id file)
+                  :team-id    (:team-id file)}))))
+
+(defn update-file!
+  "A public api that allows apply a transformation to a file with all context setup."
+  [cfg file-id update-fn & args]
+  (let [file (get-file cfg file-id)
+        file (apply update-file-data! cfg file update-fn args)]
+    (persist-file! cfg file)))
+
+(def ^:private sql:get-file
+  "SELECT f.*, p.team_id
+     FROM file AS f
+     JOIN project AS p ON (p.id = f.project_id)
+    WHERE f.id = ?
+      AND (f.deleted_at IS NULL OR
+           f.deleted_at > now())
+      FOR KEY SHARE")
+
+(defn get-file
+  "Get not-decoded file, only decodes the features set."
+  [conn id]
+  (let [file (db/exec-one! conn [sql:get-file id])]
+    (when-not file
+      (ex/raise :type :not-found
+                :code :object-not-found
+                :hint (format "file with id '%s' does not exists" id)))
+    (update file :features db/decode-pgarray #{})))
+
+(defn persist-file!
+  "Function responsible of persisting already encoded file. Should be
+  used together with `get-file` and `update-file-data!`.
+
+  It also updates the project modified-at attr."
+  [{:keys [::db/conn ::timestamp]} file]
+  (let [features (db/create-array conn "text" (:features file))
+        ;; The timestamp can be nil because this function is also
+        ;; intended to be used outside of this module
+        modified-at (or timestamp (dt/now))]
+
+    (db/update! conn :project
+                {:modified-at modified-at}
+                {:id (:project-id file)}
+                {::db/return-keys false})
 
     (db/update! conn :file
                 {:revn (:revn file)
@@ -268,42 +300,28 @@
                  :features features
                  :data-backend nil
                  :data-ref-id nil
-                 :modified-at created-at
+                 :modified-at modified-at
                  :has-media-trimmed false}
-                {:id (:id file)})
+                {:id (:id file)}
+                {::db/return-keys false})))
 
-    (db/update! conn :project
+(defn- update-file-data!
-                {:modified-at created-at}
+  "Perform a file data transformation in with all update context setup.
-                {:id (:project-id file)})
 
-    (let [params (assoc params :file file)]
+  This function expected not-decoded file and transformation function. Returns
-      ;; Send asynchronous notifications
+  an encoded file.
-      (send-notifications! cfg params)
 
-      {:revn (:revn file)
+  This function is not responsible of saving the file. It only saves
-       :lagged (get-lagged-changes conn params)})))
+  fdata/pointer-map modified fragments."
 
-(defn- soft-validate-file-schema!
+  [cfg {:keys [id] :as file} update-fn & args]
-  [file]
+  (binding [pmap/*tracked* (pmap/create-tracked)
-  (try
+            pmap/*load-fn* (partial feat.fdata/load-pointer cfg id)]
-    (val/validate-file-schema! file)
-    (catch Throwable cause
-      (l/error :hint "file schema validation error" :cause cause))))
-
-(defn- soft-validate-file!
-  [file libs]
-  (try
-    (val/validate-file! file libs)
-    (catch Throwable cause
-      (l/error :hint "file validation error"
-               :cause cause))))
-
-(defn- update-file-data
-  [{:keys [::db/conn] :as cfg} file changes skip-validate]
     (let [file (update file :data (fn [data]
                                     (-> data
                                         (blob/decode)
                                         (assoc :id (:id file)))))
+
           ;; For avoid unnecesary overhead of creating multiple pointers
           ;; and handly internally with objects map in their worst
           ;; case (when probably all shapes and all pointers will be
@@ -316,31 +334,10 @@
                      (fmg/migrate-file))
                  file)
 
-        ;; WARNING: this ruins performance; maybe we need to find
+          file (apply update-fn cfg file args)
-        ;; some other way to do general validation
-        libs (when (and (or (contains? cf/flags :file-validation)
-                            (contains? cf/flags :soft-file-validation))
-                        (not skip-validate))
-               (->> (files/get-file-libraries conn (:id file))
-                    (into [file] (map (fn [{:keys [id]}]
-                                        (binding [pmap/*load-fn* (partial feat.fdata/load-pointer cfg id)
-                                                  pmap/*tracked* nil]
-                                          ;; We do not resolve the objects maps here
-                                          ;; because there is a lower probability that all
-                                          ;; shapes needed to be loded into memory, so we
-                                          ;; leeave it on lazy status
-                                          (-> (files/get-file cfg id :migrate? false)
-                                              (update :data feat.fdata/process-pointers deref) ; ensure all pointers resolved
-                                              (update :data feat.fdata/process-objects (partial into {}))
-                                              (fmg/migrate-file))))))
-                    (d/index-by :id)))
-
-
-        file (-> (files/check-version! file)
-                 (update :revn inc)
-                 (update :data cpc/process-changes changes)
-                 (update :data d/without-nils))
 
+          ;; TODO: reuse operations if file is migrated
+          ;; TODO: move encoding to a separated thread
           file (if (take-snapshot? file)
                  (let [tpoint   (dt/tpoint)
                        snapshot (-> (:data file)
@@ -359,7 +356,68 @@
                    (-> file
                        (assoc ::snapshot-data snapshot)
                        (assoc ::snapshot-label label)))
-               file)]
+                 file)
+
+          file (cond-> file
+                 (contains? cfeat/*current* "fdata/objects-map")
+                 (feat.fdata/enable-objects-map)
+
+                 (contains? cfeat/*current* "fdata/pointer-map")
+                 (feat.fdata/enable-pointer-map)
+
+                 :always
+                 (update :data blob/encode))]
+
+      (feat.fdata/persist-pointers! cfg id)
+
+      file)))
+
+(defn- get-file-libraries
+  "A helper for preload file libraries, mainly used for perform file
+  semantical and structural validation"
+  [{:keys [::db/conn] :as cfg} file]
+  (->> (files/get-file-libraries conn (:id file))
+       (into [file] (map (fn [{:keys [id]}]
+                           (binding [pmap/*load-fn* (partial feat.fdata/load-pointer cfg id)
+                                     pmap/*tracked* nil]
+                             ;; We do not resolve the objects maps here
+                             ;; because there is a lower probability that all
+                             ;; shapes needed to be loded into memory, so we
+                             ;; leeave it on lazy status
+                             (-> (files/get-file cfg id :migrate? false)
+                                 (update :data feat.fdata/process-pointers deref) ; ensure all pointers resolved
+                                 (update :data feat.fdata/process-objects (partial into {}))
+                                 (fmg/migrate-file))))))
+       (d/index-by :id)))
+
+(defn- soft-validate-file-schema!
+  [file]
+  (try
+    (val/validate-file-schema! file)
+    (catch Throwable cause
+      (l/error :hint "file schema validation error" :cause cause))))
+
+(defn- soft-validate-file!
+  [file libs]
+  (try
+    (val/validate-file! file libs)
+    (catch Throwable cause
+      (l/error :hint "file validation error"
+               :cause cause))))
+
+(defn- process-changes-and-validate
+  [cfg file changes skip-validate]
+  (let [;; WARNING: this ruins performance; maybe we need to find
+        ;; some other way to do general validation
+        libs (when (and (or (contains? cf/flags :file-validation)
+                            (contains? cf/flags :soft-file-validation))
+                        (not skip-validate))
+               (get-file-libraries cfg file))
+
+        file (-> (files/check-version! file)
+                 (update :revn inc)
+                 (update :data cpc/process-changes changes)
+                 (update :data d/without-nils))]
 
     (binding [pmap/*tracked* nil]
       (when (contains? cf/flags :soft-file-validation)
@@ -376,15 +434,7 @@
                  (not skip-validate))
         (val/validate-file-schema! file)))
 
-    (cond-> file
+    file))
-      (contains? cfeat/*current* "fdata/objects-map")
-      (feat.fdata/enable-objects-map)
-
-      (contains? cfeat/*current* "fdata/pointer-map")
-      (feat.fdata/enable-pointer-map)
-
-      :always
-      (update :data blob/encode))))
 
 (defn- take-snapshot?
   "Defines the rule when file `data` snapshot should be saved."
@@ -426,8 +476,7 @@
           result    (db/exec-one! conn [sql:delete-snapshots id last-date])]
       (l/trc :hint "delete old snapshots" :file-id (str id) :total (db/get-update-count result)))))
 
-(def ^:private
+(def ^:private sql:lagged-changes
-  sql:lagged-changes
   "select s.id, s.revn, s.file_id,
           s.session_id, s.changes
      from file_change as s

backend/test/backend_tests/helpers.clj

@@ -76,7 +76,7 @@
    :enable-feature-fdata-pointer-map
    :enable-feature-fdata-objets-map
    :enable-feature-components-v2
-   :enable-file-snapshot
+   :enable-auto-file-snapshot
    :disable-file-validation])
 
 (defn state-init
@@ -304,16 +304,18 @@
   ([params] (update-file* *system* params))
   ([system {:keys [file-id changes session-id profile-id revn]
             :or {session-id (uuid/next) revn 0}}]
-   (db/tx-run! system (fn [{:keys [::db/conn] :as system}]
+   (-> system
+       (assoc ::files.update/timestamp (dt/now))
+       (db/tx-run! (fn [{:keys [::db/conn] :as system}]
                      (let [file (files.update/get-file conn file-id)]
-                          (files.update/update-file system
+                       (#'files.update/update-file* system
                                                     {:id file-id
                                                      :revn revn
                                                      :file file
                                                      :features (:features file)
                                                      :changes changes
                                                      :session-id session-id
-                                                     :profile-id profile-id}))))))
+                                                     :profile-id profile-id})))))))
 
 (declare command!)