metabase.sync.util

(ns metabase.sync.util
  (:require
   [clojure.math.numeric-tower :as math]
   [clojure.string :as str]
   [java-time.api :as t]
   [medley.core :as m]
   [metabase.driver :as driver]
   [metabase.driver.util :as driver.u]
   [metabase.events :as events]
   [metabase.models.interface :as mi]
   [metabase.models.task-history :as task-history]
   [metabase.query-processor.interface :as qp.i]
   [metabase.sync.interface :as i]
   [metabase.util :as u]
   [metabase.util.date-2 :as u.date]
   [metabase.util.log :as log]
   [metabase.util.malli :as mu]
   [metabase.util.malli.registry :as mr]
   [metabase.util.malli.schema :as ms]
   [toucan2.core :as t2]
   [toucan2.realize :as t2.realize])
  (:import
   (java.time.temporal Temporal)))

(set! *warn-on-reflection* true)

(derive ::event :metabase/event)

(def ^:private sync-event-topics
  #{:event/sync-begin
    :event/sync-end
    :event/analyze-begin
    :event/analyze-end
    :event/refingerprint-begin
    :event/refingerprint-end
    :event/cache-field-values-begin
    :event/cache-field-values-end
    :event/sync-metadata-begin
    :event/sync-metadata-end})

(doseq [topic sync-event-topics]
  (derive topic ::event))

(def ^:private Topic
  [:and
   events/Topic
   [:fn
    {:error/message "Sync event deriving from :metabase.sync.util/event"}
    #(isa? % ::event)]])

(defonce ^:private operation->db-ids (atom {}))

(defn with-duplicate-ops-prevented
  {:style/indent [:form]}
  [operation database-or-id f]
  (fn []
    (when-not (contains? (@operation->db-ids operation) (u/the-id database-or-id))
      (try
        ;; mark this database as currently syncing so we can prevent duplicate sync attempts (#2337)
        (swap! operation->db-ids update operation #(conj (or % #{}) (u/the-id database-or-id)))
        (log/debug "Sync operations in flight:" (m/filter-vals seq @operation->db-ids))
        ;; do our work
        (f)
        ;; always take the ID out of the set when we are through
        (finally
          (swap! operation->db-ids update operation #(disj % (u/the-id database-or-id))))))))

(mu/defn- with-sync-events
  {:style/indent [:form]}
  ;; we can do everyone a favor and infer the name of the individual begin and sync events
  ([event-name-prefix database-or-id f]
   (letfn [(event-keyword [prefix suffix]
             (keyword (or (namespace event-name-prefix) "event")
                      (str (name prefix) suffix)))]
     (with-sync-events
      (event-keyword event-name-prefix "-begin")
      (event-keyword event-name-prefix "-end")
      database-or-id
      f)))
  ([begin-event-name :- Topic
    end-event-name   :- Topic
    database-or-id
    f]
   (fn []
     (let [start-time    (System/nanoTime)
           tracking-hash (str (random-uuid))]
       (events/publish-event! begin-event-name {:database_id (u/the-id database-or-id), :custom_id tracking-hash})
       (let [return        (f)
             total-time-ms (int (/ (- (System/nanoTime) start-time)
                                   1000000.0))]
         (events/publish-event! end-event-name {:database_id  (u/the-id database-or-id)
                                                :custom_id    tracking-hash
                                                :running_time total-time-ms})
         return)))))

(defn- with-start-and-finish-logging*
  {:style/indent [:form]}
  [log-fn message f]
  (let [start-time (System/nanoTime)
        _          (log-fn (u/format-color 'magenta "STARTING: %s" message))
        result     (f)]
    (log-fn (u/format-color 'magenta "FINISHED: %s (%s)"
                            message
                            (u/format-nanoseconds (- (System/nanoTime) start-time))))
    result))

(defn- with-start-and-finish-logging
  {:style/indent [:form]}
  [message f]
  (fn []
    (with-start-and-finish-logging* #(log/info %) message f)))

(defn- do-with-start-and-finish-debug-logging
  {:style/indent [:form]}
  [message f]
  (with-start-and-finish-logging* #(log/info %) message f))

(defn- with-db-logging-disabled
  {:style/indent [:form]}
  [f]
  (fn []
    (binding [qp.i/*disable-qp-logging* true]
      (f))))

(defn- sync-in-context
  [database f]
  (fn []
    (driver/sync-in-context (driver.u/database->driver database) database
                            f)))

(doseq [klass [java.net.ConnectException
               java.net.NoRouteToHostException
               java.net.UnknownHostException
               com.mchange.v2.resourcepool.CannotAcquireResourceException
               javax.net.ssl.SSLHandshakeException]]
  (derive klass ::exception-class-not-to-retry))

(def ^:dynamic *log-exceptions-and-continue?*
  true)

(defn do-with-error-handling
  ([f]
   (do-with-error-handling "Error running sync step" f))
  ([message f]
   (try
     (f)
     (catch Throwable e
       (if *log-exceptions-and-continue?*
         (do
           (log/warn e message)
           e)
         (throw e))))))

(defmacro with-error-handling
  {:style/indent 1}
  [message & body]
  `(do-with-error-handling ~message (fn [] ~@body)))

(mu/defn do-sync-operation
  [operation :- :keyword                ; something like `:sync-metadata` or `:refingerprint`
   database  :- (ms/InstanceOf :model/Database)
   message   :- ms/NonBlankString
   f         :- fn?]
  ((with-duplicate-ops-prevented
    operation database
    (with-sync-events
     operation database
     (with-start-and-finish-logging
      message
      (with-db-logging-disabled
       (sync-in-context database
                        (partial do-with-error-handling (format "Error in sync step %s" message) f))))))))

(defmacro sync-operation
  {:style/indent 3}
  [operation database message & body]
  `(do-sync-operation ~operation ~database ~message (fn [] ~@body)))

(def ^:private ^:const ^Integer emoji-meter-width 50)

(def ^:private progress-emoji
  ["😱"   ; face screaming in fear
   "😢"   ; crying face
   "😞"   ; disappointed face
   "😒"   ; unamused face
   "😕"   ; confused face
   "😐"   ; neutral face
   "😬"   ; grimacing face
   "😌"   ; relieved face
   "😏"   ; smirking face
   "😋"   ; face savouring delicious food
   "😊"   ; smiling face with smiling eyes
   "😍"   ; smiling face with heart shaped eyes
   "😎"]) ; smiling face with sunglasses

(defn- percent-done->emoji [percent-done]
  (progress-emoji (int (math/round (* percent-done (dec (count progress-emoji)))))))

(defn emoji-progress-bar
  [completed total log-every-n]
  (let [percent-done (float (/ completed total))
        filleds      (int (* percent-done emoji-meter-width))
        blanks       (- emoji-meter-width filleds)]
    (when (or (zero? (mod completed log-every-n))
              (= completed total))
      (str "["
           (str/join (repeat filleds "*"))
           (str/join (repeat blanks "·"))
           (format "] %s  %3.0f%%" (u/emoji (percent-done->emoji percent-done)) (* percent-done 100.0))))))

(defmacro with-emoji-progress-bar
  {:style/indent 1}
  [[emoji-progress-fn-binding total-count] & body]
  `(let [finished-count#            (atom 0)
         total-count#               ~total-count
         log-every-n#               (Math/ceil (/ total-count# 10))
         ~emoji-progress-fn-binding (fn [] (emoji-progress-bar (swap! finished-count# inc) total-count# log-every-n#))]
     ~@body))

(defn set-initial-table-sync-complete!
  [table]
  (when (not= (:initial_sync_status table) "complete")
    (t2/update! :model/Table (u/the-id table) {:initial_sync_status "complete"})))

(def ^:private sync-tables-kv-args
  {:active          true
   :visibility_type nil})

(defn set-initial-table-sync-complete-for-db!
  [database-or-id]
  (t2/update! :model/Table (merge sync-tables-kv-args {:db_id (u/the-id database-or-id)})
              {:initial_sync_status "complete"}))

(defn set-initial-database-sync-complete!
  [database]
  (when (not= (:initial_sync_status database) "complete")
    (t2/update! :model/Database (u/the-id database) {:initial_sync_status "complete"})))

(defn set-initial-database-sync-aborted!
  [database]
  (when (not= (:initial_sync_status database) "complete")
    (t2/update! :model/Database (u/the-id database) {:initial_sync_status "aborted"})))

(def sync-tables-clause
  (into [:and] (for [[k v] sync-tables-kv-args]
                 [:= k v])))

(defn reducible-sync-tables
  [database-or-id & {:keys [schema-names table-names]}]
  (eduction (map t2.realize/realize)
            (t2/reducible-select :model/Table
                                 :db_id (u/the-id database-or-id)
                                 {:where [:and sync-tables-clause
                                          (when (seq schema-names) [:in :schema schema-names])
                                          (when (seq table-names) [:in :name table-names])]})))

(defn sync-tables-count
  [database-or-id]
  (t2/count :model/Table :db_id (u/the-id database-or-id) {:where sync-tables-clause}))

(defn refingerprint-reducible-sync-tables
  [database-or-id]
  (eduction (map t2.realize/realize)
            (t2/reducible-select :model/Table
                                 {:select    [:t.*]
                                  :from      [[(t2/table-name :model/Table) :t]]
                                  :left-join [[{:select   [:table_id
                                                           [[:min :last_analyzed] :earliest_last_analyzed]]
                                                :from     [(t2/table-name :model/Field)]
                                                :group-by [:table_id]} :sub]
                                              [:= :t.id :sub.table_id]]
                                  :where     [:and sync-tables-clause [:= :t.db_id (u/the-id database-or-id)]]
                                  :order-by  [[:sub.earliest_last_analyzed :asc]]})))

(defn sync-schemas
  [database-or-id]
  (vec (map :schema (t2/query {:select-distinct [:schema]
                               :from            [:metabase_table]
                               :where           [:and sync-tables-clause [:= :db_id (u/the-id database-or-id)]]}))))

(defmulti name-for-logging
  {:arglists '([instance])}
  mi/model)

(defmethod name-for-logging :model/Database
  [{database-name :name, id :id, engine :engine}]
  (format "%s Database %s ''%s''" (name engine) (str (or id "")) database-name))

(defn table-name-for-logging
  [& {:keys [id schema name]}]
  (format "Table %s ''%s''" (or (str id) "") (str (when (seq schema) (str schema ".")) name)))

(defmethod name-for-logging :model/Table [table]
  (table-name-for-logging table))

(defn field-name-for-logging
  [& {:keys [id name]}]
  (format "Field %s ''%s''" (or (str id) "") name))

(defmethod name-for-logging :model/Field [field]
  (field-name-for-logging field))

(defmethod name-for-logging :default [{field-name :name}]
  (format "Field ''%s''" field-name))

(mu/defn calculate-duration-str :- :string
  "Given two datetimes, caculate the time between them, return the result as a string"
  [begin-time :- (ms/InstanceOfClass Temporal)
   end-time   :- (ms/InstanceOfClass Temporal)]
  (u/format-nanoseconds (.toNanos (t/duration begin-time end-time))))

(def ^:private TimedSyncMetadata
  [:map
   [:start-time                  (ms/InstanceOfClass Temporal)]
   [:end-time   {:optional true} (ms/InstanceOfClass Temporal)]])

(mr/def ::StepRunMetadata
  [:merge
   TimedSyncMetadata
   [:map
    [:log-summary-fn [:maybe [:=> [:cat [:ref ::StepRunMetadata]] :string]]]]])

(def ^:private StepRunMetadata
  [:ref ::StepRunMetadata])

(mr/def ::StepNameWithMetadata
  [:tuple
   ;; step name
   :string
   ;; step metadata
   StepRunMetadata])

(def StepNameWithMetadata
  [:ref ::StepNameWithMetadata])

(def ^:private SyncOperationMetadata
  [:merge
   TimedSyncMetadata
   [:map
    [:steps [:maybe [:sequential StepNameWithMetadata]]]]])

(def ^:private LogSummaryFunction
  [:=> [:cat StepRunMetadata] :string])

(def ^:private StepDefinition
  [:map
   [:sync-fn        [:=> [:cat StepRunMetadata] i/DatabaseInstance]]
   [:step-name      :string]
   [:log-summary-fn [:maybe LogSummaryFunction]]])

(defn create-sync-step
  ([step-name sync-fn]
   (create-sync-step step-name sync-fn nil))
  ([step-name sync-fn log-summary-fn]
   {:sync-fn        sync-fn
    :step-name      step-name
    :log-summary-fn (when log-summary-fn
                      (comp str log-summary-fn))}))

(mu/defn- run-step-with-metadata :- StepNameWithMetadata
  "Runs `step` on `database` returning metadata from the run"
  [database :- i/DatabaseInstance
   {:keys [step-name sync-fn log-summary-fn] :as _step} :- StepDefinition]
  (let [start-time (t/zoned-date-time)
        results    (do-with-start-and-finish-debug-logging
                    (format "step ''%s'' for %s"
                            step-name
                            (name-for-logging database))
                    (fn [& args]
                      (try
                        (task-history/with-task-history
                         {:task            step-name
                          :db_id           (u/the-id database)
                          :on-success-info (fn [update-map result]
                                             (if (instance? Throwable result)
                                               (throw result)
                                               (assoc update-map :task_details (dissoc result :start-time :end-time :log-summary-fn))))}
                          (apply sync-fn database args))
                        (catch Throwable e
                          (if *log-exceptions-and-continue?*
                            (do
                              (log/warnf e "Error running step ''%s'' for %s" step-name (name-for-logging database))
                              {:throwable e})
                            (throw e))))))
        end-time   (t/zoned-date-time)]
    [step-name (assoc results
                      :start-time start-time
                      :end-time end-time
                      :log-summary-fn log-summary-fn)]))

(mu/defn- make-log-sync-summary-str
  [operation :- :string
   database :- i/DatabaseInstance
   {:keys [start-time end-time steps]} :- SyncOperationMetadata]
  (str
   (apply format
          (str "\n#################################################################\n"
               "# %s\n"
               "# %s\n"
               "# %s\n"
               "# %s\n")
          [(format "Completed %s on %s" operation (:name database))
           (format "Start: %s" (u.date/format start-time))
           (format "End: %s" (u.date/format end-time))
           (format "Duration: %s" (calculate-duration-str start-time end-time))])
   (apply str (for [[step-name {:keys [start-time end-time log-summary-fn] :as step-info}] steps]
                (apply format (str "# ---------------------------------------------------------------\n"
                                   "# %s\n"
                                   "# %s\n"
                                   "# %s\n"
                                   "# %s\n"
                                   (when log-summary-fn
                                     (format "# %s\n" (log-summary-fn step-info))))
                       [(format "Completed step ''%s''" step-name)
                        (format "Start: %s" (u.date/format start-time))
                        (format "End: %s" (u.date/format end-time))
                        (format "Duration: %s" (calculate-duration-str start-time end-time))])))
   "#################################################################\n"))

(mu/defn- log-sync-summary
  [operation :- :string
   database :- i/DatabaseInstance
   sync-metadata :- SyncOperationMetadata]
  ;; Note this needs to either stay nested in the `debug` macro call or be guarded by an log/enabled?
  ;; call. Constructing the log below requires some work, no need to incur that cost debug logging isn't enabled
  (log/debug (make-log-sync-summary-str operation database sync-metadata)))

(defn- do-not-retry-exception? [e]
  (or (isa? (class e) ::exception-class-not-to-retry)
      (some-> (ex-cause e) recur)))

(defn abandon-sync?
  [step-results]
  (when-let [caught-exception (:throwable step-results)]
    (do-not-retry-exception? caught-exception)))

(mu/defn run-sync-operation
  [operation :- :string
   database :- i/DatabaseInstance
   sync-steps :- [:maybe [:sequential StepDefinition]]]
  (task-history/with-task-history {:task  operation
                                   :db_id (u/the-id database)}
    (let [start-time    (t/zoned-date-time)
          step-metadata (loop [[step-defn & rest-defns] sync-steps
                               result                   []]
                          (let [[step-name r] (run-step-with-metadata database step-defn)
                                new-result    (conj result [step-name r])]
                            (cond (abandon-sync? r) new-result
                                  (not (seq rest-defns)) new-result
                                  :else (recur rest-defns new-result))))
          end-time      (t/zoned-date-time)
          sync-metadata {:start-time start-time
                         :end-time   end-time
                         :steps      step-metadata}]
      (log-sync-summary operation database sync-metadata)
      sync-metadata)))

(defn sum-numbers
  [f coll]
  (reduce + (for [item coll
                  :let [result (f item)]
                  :when (number? result)]
              result)))

(defn sum-for*
  [results]
  (reduce + (filter number? results)))

(defmacro sum-for
  {:style/indent 1}
  [[item-binding coll & more-for-bindings] & body]
  `(sum-for* (for [~item-binding ~coll
                   ~@more-for-bindings]
               (do ~@body))))