metabase.db.connection

(ns metabase.db.connection
  (:require
   [clojure.core.async.impl.dispatch :as a.impl.dispatch]
   [metabase.db.connection-pool-setup :as connection-pool-setup]
   [metabase.db.env :as mdb.env]
   [methodical.core :as methodical]
   [potemkin :as p]
   [toucan2.connection :as t2.conn]
   [toucan2.jdbc.connection :as t2.jdbc.conn]
   [toucan2.pipeline :as t2.pipeline])
  (:import
   (java.util.concurrent.locks ReentrantReadWriteLock)))

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

(defonce 
  application-db-counter
  (atom 0))

(p/defrecord+ ApplicationDB [^clojure.lang.Keyword db-type
                             ^javax.sql.DataSource data-source
                             ;; used by [[metabase.db/setup-db!]] and [[metabase.db/db-is-set-up?]] to record whether
                             ;; the usual setup steps have been performed (i.e., running Liquibase and Clojure-land data
                             ;; migrations).
                             ^clojure.lang.Atom    status
                             ;; A unique identifier generated for this specific application DB. Use this as a
                             ;; memoization/cache key. See [[unique-identifier]] for more information.
                             id
                             ;; Reentrant read-write lock for GETTING new connections. Lock doesn't track whether any
                             ;; existing connections are open! Holding the write lock will however prevent any NEW
                             ;; connections from being acquired.
                             ;;
                             ;; This is a reentrant read-write lock, which means any number of read locks are allowed at
                             ;; the same time, but the write lock is exclusive. So if you want to prevent anyone from
                             ;; getting new connections, lock the write lock.
                             ;;
                             ;; The main purpose of this is to power [[metabase.api.testing]] which allows you to reset
                             ;; the application DB with data from a SQL dump -- during the restore process it is
                             ;; important that we do not allow anyone to access the DB.
                             ^ReentrantReadWriteLock lock]
  javax.sql.DataSource
  (getConnection [_]
    (try
      (.. lock readLock lock)
      (.getConnection data-source)
      (finally
        (.. lock readLock unlock))))
  (getConnection [_ user password]
    (try
      (.. lock readLock lock)
      (.getConnection data-source user password)
      (finally
        (.. lock readLock unlock)))))

(alter-meta! #'->ApplicationDB assoc :private true)
(alter-meta! #'map->ApplicationDB assoc :private true)

(def ^:private initial-db-status nil)

(defn application-db
  ^ApplicationDB [db-type data-source & {:keys [create-pool?], :or {create-pool? false}}]
  ;; this doesn't use [[schema.core/defn]] because [[schema.core/defn]] doesn't like optional keyword args
  {:pre [(#{:h2 :mysql :postgres} db-type)
         (instance? javax.sql.DataSource data-source)]}
  (map->ApplicationDB
   {:db-type     db-type
    :data-source (if create-pool?
                   (connection-pool-setup/connection-pool-data-source db-type data-source)
                   data-source)
    :status      (atom initial-db-status)
    ;; for memoization purposes. See [[unique-identifier]] for more information.
    :id          (swap! application-db-counter inc)
    :lock        (ReentrantReadWriteLock.)}))

(def ^:dynamic ^ApplicationDB *application-db*
  (application-db mdb.env/db-type mdb.env/data-source :create-pool? true))

(defn db-type
  []
  (.db-type *application-db*))

(defn quoting-style
  [db-type]
  (case db-type
    :postgres :ansi
    :h2       :h2
    :mysql    :mysql))

(defn data-source
  ^javax.sql.DataSource []
  (.data-source *application-db*))

(defn unique-identifier
  []
  (.id *application-db*))

(methodical/defmethod t2.conn/do-with-connection :default
  [_connectable f]
  (t2.conn/do-with-connection *application-db* f))

(def ^:private ^:dynamic *transaction-depth* 0)

(defn- do-transaction [^java.sql.Connection connection f]
  (letfn [(thunk []
            (let [savepoint (.setSavepoint connection)]
              (try
                (let [result (f connection)]
                  (when (= *transaction-depth* 1)
                    ;; top-level transaction, commit
                    (.commit connection))
                  result)
                (catch Throwable e
                  (.rollback connection savepoint)
                  (throw e)))))]
    ;; optimization: don't set and unset autocommit if it's already false
    (if (.getAutoCommit connection)
      (try
        (.setAutoCommit connection false)
        (thunk)
        (finally
          (.setAutoCommit connection true)))
      (thunk))))

(comment
  ;; in toucan2.jdbc.connection, there is a 'defmethod' for t2.conn/do-with-transaction java.sql.Connection
  ;; since we don't want our implementation to be overwritten, we need to require it here first before defininng ours
  t2.jdbc.conn/keepme)

(methodical/defmethod t2.conn/do-with-transaction java.sql.Connection
  "Support nested transactions without introducing a lock like `next.jdbc` does, as that can cause deadlocks -- see
  https://github.com/seancorfield/next-jdbc/issues/244. Use `Savepoint`s because MySQL only supports nested
  transactions when done this way.
  See also https://metaboat.slack.com/archives/CKZEMT1MJ/p1694103570500929
  Note that these \"nested transactions\" are not the real thing (e.g., as in Oracle):
    - there is only one commit, meaning that every transaction in a tree of transactions can see the changes
      other transactions have made,
    - in the presence of unsynchronized concurrent threads running nested transactions, the effects of rollback
      are not well defined - a rollback will undo all work done by other transactions in the same tree that
      started later."
  [^java.sql.Connection connection {:keys [nested-transaction-rule] :or {nested-transaction-rule :allow} :as options} f]
  (assert (#{:allow :ignore :prohibit} nested-transaction-rule))
  (cond
    (and (pos? *transaction-depth*)
         (= nested-transaction-rule :ignore))
    (f connection)
    (and (pos? *transaction-depth*)
         (= nested-transaction-rule :prohibit))
    (throw (ex-info "Attempted to create nested transaction with :nested-transaction-rule set to :prohibit"
                    {:options options}))
    :else
    (binding [*transaction-depth* (inc *transaction-depth*)]
      (do-transaction connection f))))

(methodical/defmethod t2.pipeline/transduce-query :before :default
  "Make sure application database calls are not done inside core.async dispatch pool threads. This is done relatively
  early in the pipeline so the stacktrace when this fails isn't super enormous."
  [_rf _query-type₁ _model₂ _parsed-args resolved-query]
  (when (a.impl.dispatch/in-dispatch-thread?)
    (throw (ex-info "Application database calls are not allowed inside core.async dispatch pool threads."
                    {})))
  resolved-query)