Handle deletable state when enabling persistence (#23118)

* Handle deletable state when enabling persistence

When enabling persistence all models are supposed to be set to creating
unless they have been explicitly turned "off".

Before, we were only looking for new models without attached
PersistedInfos when enabling, but missed deletable PersistedInfos that
needed to be re-enabled. It's the correct thing to do during the refresh
as you wouldn't want refresh to re-enable but it's incorrect when
enabling.

* Remove unecessary thread-last

enterprise/backend/test/metabase_enterprise/sandbox/api/permissions_test.clj

@@ -135,7 +135,7 @@
               (is (db/exists? GroupTableAccessPolicy :group_id (u/the-id (perms-group/all-users)), :table_id (mt/id :venues))))))))))
 
 (defn- fake-persist-card! [card]
-  (let [persisted-info (persisted-info/turn-on! (mt/user->id :rasta) card)]
+  (let [persisted-info (persisted-info/turn-on-model! (mt/user->id :rasta) card)]
     (db/update-where! PersistedInfo {:card_id (u/the-id card)}
                       :definition (json/encode
                                     (persisted-info/metadata->definition

src/metabase/api/card.clj

@@ -785,7 +785,7 @@
                          :database    (:name database)})))
       (when-not dataset
         (throw (ex-info (tru "Card is not a model") {:status-code 400})))
-      (when-let [persisted-info (persisted-info/turn-on! api/*current-user-id* card)]
+      (when-let [persisted-info (persisted-info/turn-on-model! api/*current-user-id* card)]
         (task.persist-refresh/schedule-refresh-for-individual! persisted-info))
       api/generic-204-no-content)))
 

src/metabase/events/persisted_info.clj

@@ -33,7 +33,7 @@
                (public-settings/persisted-models-enabled)
                (get-in (Database (:database_id card)) [:options :persist-models-enabled])
                (nil? (db/select-one-field :id PersistedInfo :card_id (:id card))))
-      (persisted-info/turn-on! (:actor_id card) card))
+      (persisted-info/turn-on-model! (:actor_id card) card))
     (catch Throwable e
       (log/warn (format "Failed to process persisted-info event. %s" (:topic event)) e))))
 

src/metabase/models/persisted_info.clj

@@ -110,7 +110,7 @@
      :creator_id      user-id}))
 
 (defn ready-unpersisted-models!
-  "Looks for all models in database and creates a persisted-info ready to be synced."
+  "Looks for all new models in database and creates a persisted-info ready to be synced."
   [database-id]
   (let [cards (db/select Card {:where [:and
                                        [:= :database_id database-id]
@@ -123,7 +123,7 @@
 (comment
   (ready-unpersisted-models! 183))
 
-(defn turn-on!
+(defn turn-on-model!
   "Marks PersistedInfo as `creating`, these will at some point be persisted by the PersistRefresh task."
   [user-id card]
   (let [card-id (u/the-id card)
@@ -138,3 +138,16 @@
                                        :active false, :state "creating", :state_change_at :%now)
                            (db/select-one PersistedInfo :card_id card-id)))]
     persisted-info))
+
+(defn ready-database!
+  "Sets PersistedInfo state to `creating` for models without a PeristedInfo or those in a `deletable` state.
+   Will ignore explicitly set `off` models."
+  [database-id]
+  (db/update! PersistedInfo
+              {:where [:and
+                       [:= :database_id database-id]
+                       [:= :state "deletable"]]
+               :set {:active false,
+                     :state "creating",
+                     :state_change_at :%now}})
+  (ready-unpersisted-models! database-id))

src/metabase/task/persist_refresh.clj

@@ -297,7 +297,7 @@
      (u/format-color 'green
                      "Scheduling persistence refreshes for database %d: trigger: %s"
                      (u/the-id database) (.. ^Trigger tggr getKey getName)))
-    (persisted-info/ready-unpersisted-models! (u/the-id database))
+    (persisted-info/ready-database! (u/the-id database))
     (try (task/add-trigger! tggr)
          (catch ObjectAlreadyExistsException _e
            (log/info