Clojure + Postgres Test Container + Flyway example

test-db.clj

(ns test-db
  "Example of using testcontainers and launching a postgres container with flyway migrations.
  NB. databases are created but not destroyed which can be handy for debugging tests.
  * Connecting to template1 from psql for example will cause create database to fail
  * Based on ideas from https://github.com/opentable/otj-pg-embedded"
  (:require [clj-test-containers.core :as tc]
            [clojure.java.jdbc :as jdbc]
            [clojure.string :as str])
  (:import (org.flywaydb.core
            Flyway)))

;; deps.edn needs:
;; clj-test-containers/clj-test-containers {:mvn/version "0.7.2"}
;; org.testcontainers/postgresql           {:mvn/version "1.16.3"}

(defonce ^:dynamic *pg-docker-container* nil)


;; adjust to your needs
(def +migration-locations+ ["filesystem:./resources/flyway-migrations"])

(def ^:const +pg-user+ "pg-user")
(def ^:const +pg-password+ "secret")
(def ^:const +pg-port+ 5432)
(def ^:private ^:dynamic *future* nil)
(def ^:private ^:dynamic *queue* (java.util.concurrent.SynchronousQueue.))

(defonce ^:dynamic *pg-db-spec*
  ;; :host and :port are merged in after the container is started
  ;; :dbname need to be determined by the user
  {:dbtype "postgresql"
   :user +pg-user+
   :password +pg-password+})

(defn pg-jdbc-url
  [database-name]
  (let [{:keys [host port]} *pg-db-spec*]
    (when-not (or host port)
      (throw (ex-info "pg container did not initialize? host/port not found" *pg-db-spec*)))
    (format "jdbc:postgresql://%s:%s/%s" host port database-name)))


(defn halt!
  []
  (some-> *future* future-cancel)
  (some-> *pg-docker-container* tc/stop!)
  (alter-var-root #'*pg-docker-container* (constantly nil))
  (alter-var-root #'*pg-db-spec* dissoc :host :port)
  :halted)


(defn- launch-container
  []
  (-> {:image-name "postgres:13.6"
       :exposed-ports [5432]
       :env-vars {"POSTGRES_USER" +pg-user+
                  "POSTGRES_PASSWORD" +pg-password+
                  "POSTGRES_DB" "postgres"}}
      tc/create
      tc/start!))

(defn- produce-next-db!
  [{:keys [user] :as connect-db}]
  (while (not (Thread/interrupted))
    (let [new-db-name (str/lower-case (str "pg_db_" (str/replace (random-uuid) #"[-]" "")))
          new-db-spec (assoc connect-db :dbname new-db-name)]
      (try
        (jdbc/execute! connect-db (format "create database \"%s\" owner \"%s\" encoding = 'utf8'" new-db-name user) {:transaction? false})
        (.put *queue* new-db-spec)
        (catch java.sql.SQLException ex
          (.put *queue* (assoc new-db-spec :ex ex)))))))

(defn init!
  "Returns a promise returns `:ok` when inited."
  [& {:keys [pre-migrate-fn post-migrate-fn]
      :or {pre-migrate-fn (constantly nil)
           post-migrate-fn (constantly nil)} :as _opts}]
  (let [p (promise)
        f (future
            (try
              (let [c (launch-container)]
                (alter-var-root #'*pg-docker-container* (constantly c))
                ;; fill out the spec
                (alter-var-root #'*pg-db-spec* assoc :host (:host *pg-docker-container*)
                                  :port (get (:mapped-ports *pg-docker-container*) +pg-port+))
                (let [db-name "template1"
                      db (assoc *pg-db-spec* :dbname db-name)]
                  (pre-migrate-fn db)
                  (-> (Flyway/configure)
                      (.locations (into-array String +migration-locations+))
                      (.dataSource (pg-jdbc-url db-name) +pg-user+ +pg-password+)
                      (.load)
                      (.migrate))
                  (post-migrate-fn db)
                  (deliver p :ok)
                  (produce-next-db! (assoc *pg-db-spec* :dbname db-name))))
              (catch Throwable t
                (deliver p t))))]
    (alter-var-root #'*future* (constantly f))
    p))


(defn next-db!
  []
  (let [{:keys [ex] :as db} (.take *queue*)]
    (when ex
      (throw (ex-info "db production failed" {} ex)))
    db))


(comment

  ;; TODO:
  ;; - Better error handling
  ;; - Keep only the 100 most recent databases. Drop older ones

  ;; Example usage:

  (def -promise (init! :pre-migrate-fn (fn [_db] (println "pre-migrate-fn called"))
                       :post-migrate-fn (fn [_db] (println "post-migrate-fn called"))))

  @-promise ;; should say :ok
  (def -next-db (next-db!))
  (jdbc/query -next-db ["select current_database()"])
  (halt!)

  )