diff --git a/backend/src/app/srepl/main.clj b/backend/src/app/srepl/main.clj index ab1987239a..11b2732c19 100644 --- a/backend/src/app/srepl/main.clj +++ b/backend/src/app/srepl/main.clj @@ -828,45 +828,69 @@ ;; CASCADE FIXING ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; +(defn- submit-deleted-cascade + "Replay the deletion cascade of every soft-deleted row of `table`, by + submitting one `:delete-object` task per row. + + Reads the rows with a server-side cursor (`db/plan`) instead of materialising + the whole table, and enqueues the task instead of running it inline, so it + can be used against production in bounded batches. The task is idempotent (it + only marks the children with the row's own date), `label` is the row id and + `dedupe` drops not-yet-due duplicates. + + Options: `:limit` (max rows read), `:ids` (restrict the run to these row ids) + and `:dry-run?` (count without enqueueing). `table` and `object` are internal + constants, never operator input. + + Returns `{:found n :submitted m}`. + + Note: the submissions are given the outer `system` (with the pool) on + purpose; handing them the read-only transaction cfg would write on it." + [system {:keys [table object limit ids dry-run?]}] + (let [clauses (cond-> ["deleted_at is not null"] + (seq ids) (conj "id = any(?)")) + sql (str "select id, deleted_at from " table + " where " (str/join " and " clauses) + (when limit (str " limit " (long limit))))] + + (db/tx-run! (assoc system ::db/read-only true) + (fn [{:keys [::db/conn]}] + (let [query (cond-> [sql] + (seq ids) (conj (db/create-array conn "uuid" ids))) + found (reduce (fn [total {:keys [id deleted-at]}] + (when-not dry-run? + (wrk/submit! (-> system + (assoc ::wrk/task :delete-object) + (assoc ::wrk/params {:object object + :deleted-at deleted-at + :id id}) + (assoc ::wrk/label (str id)) + (assoc ::wrk/dedupe true)))) + (inc total)) + 0 + (db/plan conn query))] + {:found found + :submitted (if dry-run? 0 found)}))))) + (defn process-deleted-profiles-cascade - [] - (->> (db/exec! sys/system ["select id, deleted_at from profile where deleted_at is not null"]) - (run! (fn [{:keys [id deleted-at]}] - (wrk/invoke! (-> sys/system - (assoc ::wrk/task :delete-object) - (assoc ::wrk/params {:object :profile - :deleted-at deleted-at - :id id}))))))) + "Replay the deletion cascade of every soft-deleted profile." + [& {:as opts}] + (submit-deleted-cascade sys/system (merge {:table "profile" :object :profile} opts))) (defn process-deleted-teams-cascade - [] - (->> (db/exec! sys/system ["select id, deleted_at from team where deleted_at is not null"]) - (run! (fn [{:keys [id deleted-at]}] - (wrk/invoke! (-> sys/system - (assoc ::wrk/task :delete-object) - (assoc ::wrk/params {:object :team - :deleted-at deleted-at - :id id}))))))) + "Replay the deletion cascade of every soft-deleted team." + [& {:as opts}] + (submit-deleted-cascade sys/system (merge {:table "team" :object :team} opts))) (defn process-deleted-projects-cascade - [] - (->> (db/exec! sys/system ["select id, deleted_at from project where deleted_at is not null"]) - (run! (fn [{:keys [id deleted-at]}] - (wrk/invoke! (-> sys/system - (assoc ::wrk/task :delete-object) - (assoc ::wrk/params {:object :project - :deleted-at deleted-at - :id id}))))))) + "Replay the deletion cascade of every soft-deleted project." + [& {:as opts}] + (submit-deleted-cascade sys/system (merge {:table "project" :object :project} opts))) (defn process-deleted-files-cascade - [] - (->> (db/exec! sys/system ["select id, deleted_at from file where deleted_at is not null"]) - (run! (fn [{:keys [id deleted-at]}] - (wrk/invoke! (-> sys/system - (assoc ::wrk/task :delete-object) - (assoc ::wrk/params {:object :file - :deleted-at deleted-at - :id id}))))))) + "Replay the deletion cascade of every soft-deleted file." + [& {:as opts}] + (submit-deleted-cascade sys/system (merge {:table "file" :object :file} opts))) ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;; SSO diff --git a/backend/test/backend_tests/srepl_main_test.clj b/backend/test/backend_tests/srepl_main_test.clj index 9897415d19..1e2add5361 100644 --- a/backend/test/backend_tests/srepl_main_test.clj +++ b/backend/test/backend_tests/srepl_main_test.clj @@ -2,13 +2,45 @@ ;; License, v. 2.0. If a copy of the MPL was not distributed with this ;; file, You can obtain one at http://mozilla.org/MPL/2.0/. ;; -;; Copyright (c) KALEIDOS SUBSIDIARY SL +;; Copyright (c) KALEIDOS INC Sucursal en España SL (ns backend-tests.srepl-main-test (:require + [app.common.time :as ct] [app.srepl.main :as srepl] + [app.system :as sys] + [app.worker :as wrk] + [backend-tests.helpers :as th] [clojure.test :as t])) +(t/use-fixtures :once th/state-init) +(t/use-fixtures :each th/database-reset) + +;; --- HELPERS + +(defn- capture-submissions! + "Runs `f` with `wrk/submit!` replaced by a collector, and returns the option + maps it was called with, in order." + [f] + (let [submitted (atom [])] + (with-redefs [wrk/submit! (fn [opts] + (swap! submitted conj opts) + (get-in opts [::wrk/params :id]))] + (f)) + @submitted)) + +(defn- params-of + [submitted] + (mapv ::wrk/params submitted)) + +(defn- create-deleted-profile! + [i deleted-at] + (let [profile (th/create-profile* i)] + (th/db-update! :profile {:deleted-at deleted-at} {:id (:id profile)}) + profile)) + +;; --- TESTS + (t/deftest parse-emails (t/is (= ["some@example.com"] (srepl/parse-emails "some@example.com"))) @@ -26,3 +58,95 @@ (t/is (thrown? clojure.lang.ExceptionInfo (srepl/parse-emails 42)))) + +(t/deftest cascade-enqueues-one-delete-object-task-per-deleted-row + (let [deleted (create-deleted-profile! 991 (ct/in-past {:days 30})) + _ (th/create-profile* 992) + submitted (capture-submissions! + #(#'srepl/submit-deleted-cascade + th/*system* {:table "profile" :object :profile}))] + + (t/testing "only the soft-deleted row is processed" + (t/is (= 1 (count submitted)))) + + (t/testing "the enqueued task is a delete-object for that row" + (let [opts (first submitted)] + (t/is (= :delete-object (::wrk/task opts))) + (t/is (= {:object :profile :id (:id deleted)} + (select-keys (::wrk/params opts) [:object :id]))))) + + (t/testing "the date is the row's own, not now" + (t/is (ct/is-before? (:deleted-at (::wrk/params (first submitted))) + (ct/in-past {:days 29})))) + + (t/testing "label is the row id and dedupe is on" + (t/is (= (str (:id deleted)) (::wrk/label (first submitted)))) + (t/is (true? (::wrk/dedupe (first submitted))))))) + +(t/deftest cascade-honours-limit-and-ids + (let [p1 (create-deleted-profile! 993 (ct/in-past {:days 10})) + _ (create-deleted-profile! 994 (ct/in-past {:days 9})) + _ (create-deleted-profile! 995 (ct/in-past {:days 8}))] + + (t/testing "limit bounds how many rows are read" + (let [submitted (capture-submissions! + #(#'srepl/submit-deleted-cascade + th/*system* {:table "profile" :object :profile :limit 2}))] + (t/is (= 2 (count submitted))))) + + (t/testing "ids restricts the run to the given rows" + (let [submitted (capture-submissions! + #(#'srepl/submit-deleted-cascade + th/*system* {:table "profile" :object :profile :ids [(:id p1)]}))] + (t/is (= [{:object :profile :id (:id p1)}] + (mapv #(select-keys % [:object :id]) (params-of submitted)))))))) + +(t/deftest cascade-dry-run-counts-without-enqueueing + (let [_ (create-deleted-profile! 996 (ct/in-past {:days 10})) + submitted (atom []) + result (with-redefs [wrk/submit! (fn [opts] (swap! submitted conj opts))] + (#'srepl/submit-deleted-cascade + th/*system* {:table "profile" :object :profile :dry-run? true}))] + + (t/is (= 1 (:found result))) + (t/is (= 0 (:submitted result))) + (t/is (empty? @submitted)))) + +(t/deftest public-helpers-read-their-own-table + (let [profile (th/create-profile* 997) + team (th/create-team* 997 {:profile-id (:id profile)}) + project (th/create-project* 997 {:profile-id (:id profile) + :team-id (:id team)}) + file (th/create-file* 997 {:profile-id (:id profile) + :project-id (:id project)}) + old (ct/in-past {:days 10})] + + (th/db-update! :profile {:deleted-at old} {:id (:id profile)}) + (th/db-update! :team {:deleted-at old} {:id (:id team)}) + (th/db-update! :project {:deleted-at old} {:id (:id project)}) + (th/db-update! :file {:deleted-at old} {:id (:id file)}) + + (with-redefs [sys/system th/*system*] + (let [ids (fn [submitted] (into #{} (map :id (params-of submitted))))] + + (t/testing "profiles helper" + (let [submitted (capture-submissions! #(srepl/process-deleted-profiles-cascade))] + (t/is (contains? (ids submitted) (:id profile))))) + + (t/testing "teams helper" + (let [submitted (capture-submissions! #(srepl/process-deleted-teams-cascade))] + (t/is (contains? (ids submitted) (:id team))))) + + (t/testing "projects helper" + (let [submitted (capture-submissions! #(srepl/process-deleted-projects-cascade))] + (t/is (contains? (ids submitted) (:id project))))) + + (t/testing "files helper" + (let [submitted (capture-submissions! #(srepl/process-deleted-files-cascade))] + (t/is (contains? (ids submitted) (:id file))))) + + (t/testing "the helpers accept options" + (let [submitted (capture-submissions! + #(srepl/process-deleted-files-cascade :ids [(:id file)]))] + (t/is (= [{:object :file :id (:id file)}] + (mapv #(select-keys % [:object :id]) (params-of submitted))))))))))