(ns kafka-clj.redis.single
  (:import
    (org.apache.commons.pool2 ObjectPool PooledObjectFactory)
    (org.apache.commons.pool2.impl GenericObjectPool DefaultPooledObject GenericKeyedObjectPool)
    (java.util.concurrent TimeUnit)
    (java.util UUID))
  (:require [taoensso.carmine
             (protocol :as protocol)
             (connections :as conns)
             (commands :as commands)]
            [kafka-clj.redis.protocol :refer [IRedis]]
            [taoensso.carmine :as car]
            [clojure.tools.logging :refer [info]]
            [taoensso.encore :as enc]))


(defn nil-safe [v default]
  (if (nil? v) default v))

(defn- make-connection-factory [{:keys [host port timeout password]}]
  (reify PooledObjectFactory
    (makeObject      [_ ]
      (DefaultPooledObject. (conns/make-new-connection (merge
                                                         {:host (nil-safe host "localhost")
                                                          :port (nil-safe port 6379)
                                                          :timeout (nil-safe timeout 300)}
                                                         (when password
                                                           {:password password})))))
    (activateObject  [_ pooled-obj])
    (validateObject  [_ pooled-obj] (let [conn (.getObject pooled-obj)]
                                      (conns/conn-alive? conn)))
    (passivateObject [_ pooled-obj])
    (destroyObject   [_ pooled-obj] (let [conn (.getObject pooled-obj)]
                                      (conns/close-conn conn)))))

(defn- set-pool-option [^ObjectPool pool [opt v]]
  (case opt

  org.apache.commons.pool2.impl . BaseGenericObjectPool
    :block-when-exhausted? (.setBlockWhenExhausted pool v) ; true
    :lifo?       (.setLifo          pool v) ; true
    :max-active   (.setMaxTotal      pool v) ; -1
    :max-total   (.setMaxTotal      pool v) ; -1

    :﻿max-total-per-key (.setMaxTotal pool v)
    :max-wait-ms (.setMaxWaitMillis pool v) ; -1
  1800000
  3
    :soft-min-evictable-idle-time-ms (.setSoftMinEvictableIdleTimeMillis pool v) ; -1
    :swallowed-exception-listener    (.setSwallowedExceptionListener     pool v)
    :test-on-borrow?  (.setTestOnBorrow  pool v) ; false
    :test-on-return?  (.setTestOnReturn  pool v) ; false
    :test-while-idle? (.setTestWhileIdle pool v) ; false
    :time-between-eviction-runs-ms (.setTimeBetweenEvictionRunsMillis pool v) ; -1

    (prn "Unknown pool option: " opt))
  pool)


(defn conn-pool ^java.io.Closeable [spec pool-opts]
  (let [jedis-defaults ; Ref. 
        {:test-while-idle?              true  ; from false
  from 3
  from 1800000
         :time-between-eviction-runs-ms 30000 ; from -1
         }
        carmine-defaults
        {
         :max-total 40
         }
        opts (merge jedis-defaults carmine-defaults pool-opts)
        object-pool (reduce set-pool-option
                            (GenericObjectPool. (make-connection-factory spec))
                            opts)]
    (conns/->ConnectionPool object-pool)))

(defn close-pool [{:keys [^ObjectPool pool]}] (.close pool))

(defn pooled-conn [{:keys [^ObjectPool pool]}]
  (.borrowObject pool))

(defn release-conn [{:keys [^ObjectPool pool]} conn]
  (.returnObject pool conn))


(defmacro _wcar
  "
  The ConnectionPool returned from conn-pool must be the first argument to this macro
  Evaluates body in the context of a fresh thread-bound pooled connection to
  Redis server. Sends Redis commands to server as pipeline and returns the
  server's response. Releases connection back to pool when done.
  `conn-opts` arg is a map with connection pool and spec options:
    {:pool {} :spec {:host \"127.0.0.1\" :port 6379}} ; Default
    {:pool {} :spec {:uri \"redis::9475/\"}}
    {:pool {} :spec {:host \"127.0.0.1\" :port 6379
                     :password \"secret\"
                     :timeout-ms 6000
                     :db 3}}
  A `nil` or `{}` `conn-opts` will use defaults. A `:none` pool can be used
  to skip connection pooling (not recommended).
  For other pool options, Ref. ,
                                (defaults).
  Note that because of thread-binding, you'll probably want to avoid lazy Redis
  command calls in `wcar`'s body unless you know what you're doing. Compare:
  `(wcar {} (for   [k [:k1 :k2]] (car/set k :val))` ; Lazy, NO commands run
  `(wcar {} (doseq [k [:k1 :k2]] (car/set k :val))` ; Not lazy, commands run
  See also `with-replies`."
  {:arglists '([conn-pool :as-pipeline & body] [conn-pool & body])}
  [^ObjectPool pool & sigs] ; [conn-opts & [s1 & sn :as sigs]]
  `(let [conn# (kafka-clj.redis.single/pooled-conn ~pool)

  To support ` wcar ` nesting with req planning , we mimic
         ;; `with-replies` stashing logic here to simulate immediate writes:
         ?stashed-replies#
         (when protocol/*context*
           (protocol/execute-requests :get-replies :as-pipeline))]

     (try
       (let [response# (protocol/with-context conn#
                                              (protocol/with-replies ~@sigs))]
         (release-conn ~pool conn#)
         response#)

       (catch Throwable t# ; nb Throwable to catch assertions, etc.
         (release-conn ~pool conn#) (throw t#))

       ;; Restore any stashed replies to preexisting context:
       (finally
         (when ?stashed-replies#
           (car/parse nil ; Already parsed on stashing
                      (enc/backport-run! car/return ?stashed-replies#)))))))

(def ^:private lkey (partial car/key :carmine :lock))

(defn _acquire-lock
  "Attempts to acquire a distributed lock, returning an owner UUID iff successful."
  TODO Waiting on  for simpler ( non - Lua ) solution
  [pool lock-name timeout-ms wait-ms]
  (let [max-udt (+ wait-ms (System/currentTimeMillis))
        uuid    (str (UUID/randomUUID))]
    (_wcar pool ; Hold one connection for all attempts
           (loop []
             (when (> max-udt (System/currentTimeMillis))
               (if (-> (car/lua
                         "if redis.call('setnx', _:lkey, _:uuid) == 1 then
                           redis.call('pexpire', _:lkey, _:timeout-ms)
                           return 1
                         else
                           return 0
                         end"
                         {:lkey       (lkey lock-name)}
                         {:uuid       uuid
                          :timeout-ms timeout-ms})
                       car/with-replies car/as-bool)
                 (car/return uuid)
                 (do (Thread/sleep 1) (recur))))))))

(defn _release-lock
  "Attempts to release a distributed lock, returning true iff successful."
  [pool lock-name owner-uuid]
  (_wcar pool
         (car/parse-bool
           (car/lua
             "if redis.call('get', _:lkey) == _:uuid then
                redis.call('del', _:lkey)
                return 1
              else
                return 0
              end"
             {:lkey (lkey lock-name)}
             {:uuid owner-uuid}))))

(defn _have-lock? [pool lock-name owner-uuid]
  (= (_wcar pool (car/get (lkey lock-name))) owner-uuid))

(defn- release-all-locks! [pool]
  (when-let [lkeys (seq (_wcar pool (car/keys (lkey :*))))]
    (_wcar pool (apply car/del lkeys))))

(defn lua [_ script]
  (car/lua script {:a 1} {:b 2}))

;;pool is of type #taoensso.carmine.connections.ConnectionPool {:pool ObjectPool}
(defrecord SingleRedis [pool])

(defn ^ObjectPool get-object-pool
  "Internal helper function that returns the internal ObjectPool"
  [carmine-conn]
  (:pool (:pool carmine-conn)))

(extend-type SingleRedis
  IRedis

  (-conn-pool-idle [this] (.getNumIdle (get-object-pool this)))
  (-conn-pool-active [this] (.getNumActive (get-object-pool this)))

  (-lpush* [_ queue obj-coll]
    (apply car/lpush queue obj-coll))
  (-lpush [_ queue obj]
    (car/lpush queue obj))
  (-llen  [_ queue]
    (car/llen queue))
  (-lrem  [_ queue n obj]
    (car/lrem queue 1 obj))
  (-get [_ k]
    (car/get k))
  (-set   [_ k v]
    (car/set k v))
  (-lrange [_ q n limit]
    (car/lrange q n limit))
  (-brpoplpush [_ queue queue2 n]
    ;n is milliseconds but redis only handles seconds
    (car/brpoplpush queue queue2 (.toSeconds (TimeUnit/MILLISECONDS) (long n))))

  (-acquire-lock [{:keys [pool]} lock-name timeout-ms wait-ms]
    (_acquire-lock pool lock-name timeout-ms wait-ms))
  (-release-lock [{:keys [pool]} lock-name owner-uuid]
    (_release-lock pool lock-name owner-uuid))
  (-have-lock?   [{:keys [pool]} lock-name owner-uuid]
    (_have-lock? pool lock-name owner-uuid))

  (-close! [{:keys [pool]}] (close-pool pool))
  (-wcar [{:keys [pool]} f]
    (_wcar pool (f)))

  (-lua [{:keys [pool]} script-str]
    (lua pool script-str))

  (-flushall [_]
    (car/flushall)))

(defn create
  "Main public function that should be called to create a Redis pool instance"
  [spec opts]
  (->SingleRedis (conn-pool spec opts)))
 true
 true
 -1
 -1
 -1
 -1
 false
 false
 false
 -1
 Ref. 
 from false
 from -1
 Default
 Lazy, NO commands run
 Not lazy, commands run
 [conn-opts & [s1 & sn :as sigs]]
 `with-replies` stashing logic here to simulate immediate writes:
 nb Throwable to catch assertions, etc.
 Restore any stashed replies to preexisting context:
 Already parsed on stashing
 Hold one connection for all attempts
pool is of type #taoensso.carmine.connections.ConnectionPool {:pool ObjectPool}
n is milliseconds but redis only handles seconds
(ns kafka-clj.redis.single
  (:import
    (org.apache.commons.pool2 ObjectPool PooledObjectFactory)
    (org.apache.commons.pool2.impl GenericObjectPool DefaultPooledObject GenericKeyedObjectPool)
    (java.util.concurrent TimeUnit)
    (java.util UUID))
  (:require [taoensso.carmine
             (protocol :as protocol)
             (connections :as conns)
             (commands :as commands)]
            [kafka-clj.redis.protocol :refer [IRedis]]
            [taoensso.carmine :as car]
            [clojure.tools.logging :refer [info]]
            [taoensso.encore :as enc]))


(defn nil-safe [v default]
  (if (nil? v) default v))

(defn- make-connection-factory [{:keys [host port timeout password]}]
  (reify PooledObjectFactory
    (makeObject      [_ ]
      (DefaultPooledObject. (conns/make-new-connection (merge
                                                         {:host (nil-safe host "localhost")
                                                          :port (nil-safe port 6379)
                                                          :timeout (nil-safe timeout 300)}
                                                         (when password
                                                           {:password password})))))
    (activateObject  [_ pooled-obj])
    (validateObject  [_ pooled-obj] (let [conn (.getObject pooled-obj)]
                                      (conns/conn-alive? conn)))
    (passivateObject [_ pooled-obj])
    (destroyObject   [_ pooled-obj] (let [conn (.getObject pooled-obj)]
                                      (conns/close-conn conn)))))

(defn- set-pool-option [^ObjectPool pool [opt v]]
  (case opt

  org.apache.commons.pool2.impl . BaseGenericObjectPool

    :﻿max-total-per-key (.setMaxTotal pool v)
  1800000
  3
    :swallowed-exception-listener    (.setSwallowedExceptionListener     pool v)

    (prn "Unknown pool option: " opt))
  pool)


(defn conn-pool ^java.io.Closeable [spec pool-opts]
  from 3
  from 1800000
         }
        carmine-defaults
        {
         :max-total 40
         }
        opts (merge jedis-defaults carmine-defaults pool-opts)
        object-pool (reduce set-pool-option
                            (GenericObjectPool. (make-connection-factory spec))
                            opts)]
    (conns/->ConnectionPool object-pool)))

(defn close-pool [{:keys [^ObjectPool pool]}] (.close pool))

(defn pooled-conn [{:keys [^ObjectPool pool]}]
  (.borrowObject pool))

(defn release-conn [{:keys [^ObjectPool pool]} conn]
  (.returnObject pool conn))


(defmacro _wcar
  "
  The ConnectionPool returned from conn-pool must be the first argument to this macro
  Evaluates body in the context of a fresh thread-bound pooled connection to
  Redis server. Sends Redis commands to server as pipeline and returns the
  server's response. Releases connection back to pool when done.
  `conn-opts` arg is a map with connection pool and spec options:
    {:pool {} :spec {:uri \"redis::9475/\"}}
    {:pool {} :spec {:host \"127.0.0.1\" :port 6379
                     :password \"secret\"
                     :timeout-ms 6000
                     :db 3}}
  A `nil` or `{}` `conn-opts` will use defaults. A `:none` pool can be used
  to skip connection pooling (not recommended).
  For other pool options, Ref. ,
                                (defaults).
  Note that because of thread-binding, you'll probably want to avoid lazy Redis
  command calls in `wcar`'s body unless you know what you're doing. Compare:
  See also `with-replies`."
  {:arglists '([conn-pool :as-pipeline & body] [conn-pool & body])}
  `(let [conn# (kafka-clj.redis.single/pooled-conn ~pool)

  To support ` wcar ` nesting with req planning , we mimic
         ?stashed-replies#
         (when protocol/*context*
           (protocol/execute-requests :get-replies :as-pipeline))]

     (try
       (let [response# (protocol/with-context conn#
                                              (protocol/with-replies ~@sigs))]
         (release-conn ~pool conn#)
         response#)

         (release-conn ~pool conn#) (throw t#))

       (finally
         (when ?stashed-replies#
                      (enc/backport-run! car/return ?stashed-replies#)))))))

(def ^:private lkey (partial car/key :carmine :lock))

(defn _acquire-lock
  "Attempts to acquire a distributed lock, returning an owner UUID iff successful."
  TODO Waiting on  for simpler ( non - Lua ) solution
  [pool lock-name timeout-ms wait-ms]
  (let [max-udt (+ wait-ms (System/currentTimeMillis))
        uuid    (str (UUID/randomUUID))]
           (loop []
             (when (> max-udt (System/currentTimeMillis))
               (if (-> (car/lua
                         "if redis.call('setnx', _:lkey, _:uuid) == 1 then
                           redis.call('pexpire', _:lkey, _:timeout-ms)
                           return 1
                         else
                           return 0
                         end"
                         {:lkey       (lkey lock-name)}
                         {:uuid       uuid
                          :timeout-ms timeout-ms})
                       car/with-replies car/as-bool)
                 (car/return uuid)
                 (do (Thread/sleep 1) (recur))))))))

(defn _release-lock
  "Attempts to release a distributed lock, returning true iff successful."
  [pool lock-name owner-uuid]
  (_wcar pool
         (car/parse-bool
           (car/lua
             "if redis.call('get', _:lkey) == _:uuid then
                redis.call('del', _:lkey)
                return 1
              else
                return 0
              end"
             {:lkey (lkey lock-name)}
             {:uuid owner-uuid}))))

(defn _have-lock? [pool lock-name owner-uuid]
  (= (_wcar pool (car/get (lkey lock-name))) owner-uuid))

(defn- release-all-locks! [pool]
  (when-let [lkeys (seq (_wcar pool (car/keys (lkey :*))))]
    (_wcar pool (apply car/del lkeys))))

(defn lua [_ script]
  (car/lua script {:a 1} {:b 2}))

(defrecord SingleRedis [pool])

(defn ^ObjectPool get-object-pool
  "Internal helper function that returns the internal ObjectPool"
  [carmine-conn]
  (:pool (:pool carmine-conn)))

(extend-type SingleRedis
  IRedis

  (-conn-pool-idle [this] (.getNumIdle (get-object-pool this)))
  (-conn-pool-active [this] (.getNumActive (get-object-pool this)))

  (-lpush* [_ queue obj-coll]
    (apply car/lpush queue obj-coll))
  (-lpush [_ queue obj]
    (car/lpush queue obj))
  (-llen  [_ queue]
    (car/llen queue))
  (-lrem  [_ queue n obj]
    (car/lrem queue 1 obj))
  (-get [_ k]
    (car/get k))
  (-set   [_ k v]
    (car/set k v))
  (-lrange [_ q n limit]
    (car/lrange q n limit))
  (-brpoplpush [_ queue queue2 n]
    (car/brpoplpush queue queue2 (.toSeconds (TimeUnit/MILLISECONDS) (long n))))

  (-acquire-lock [{:keys [pool]} lock-name timeout-ms wait-ms]
    (_acquire-lock pool lock-name timeout-ms wait-ms))
  (-release-lock [{:keys [pool]} lock-name owner-uuid]
    (_release-lock pool lock-name owner-uuid))
  (-have-lock?   [{:keys [pool]} lock-name owner-uuid]
    (_have-lock? pool lock-name owner-uuid))

  (-close! [{:keys [pool]}] (close-pool pool))
  (-wcar [{:keys [pool]} f]
    (_wcar pool (f)))

  (-lua [{:keys [pool]} script-str]
    (lua pool script-str))

  (-flushall [_]
    (car/flushall)))

(defn create
  "Main public function that should be called to create a Redis pool instance"
  [spec opts]
  (->SingleRedis (conn-pool spec opts)))