220 lines
8.5 KiB
Clojure
220 lines
8.5 KiB
Clojure
(ns pmagnus.btcdata.web.routes.api
|
|
(:require
|
|
[clojure.data.json :as json]
|
|
[integrant.core :as ig]
|
|
[pmagnus.btcdata.frankfurter.rates :as rates]
|
|
[pmagnus.btcdata.web.controllers.health :as health]
|
|
[pmagnus.btcdata.web.controllers.transactions :as tx]
|
|
[pmagnus.btcdata.web.middleware.exception :as exception]
|
|
[pmagnus.btcdata.web.middleware.formats :as formats]
|
|
[reitit.coercion.malli :as malli]
|
|
[reitit.ring.coercion :as coercion]
|
|
[reitit.ring.middleware.muuntaja :as muuntaja]
|
|
[reitit.ring.middleware.parameters :as parameters]
|
|
[reitit.swagger :as swagger])
|
|
(:import
|
|
[io.undertow.websockets.core WebSockets WebSocketChannel]
|
|
[java.time LocalDate]
|
|
[java.time.format DateTimeParseException]
|
|
[java.util.concurrent LinkedBlockingQueue TimeUnit]))
|
|
|
|
(defn- price->json [{:keys [price prev-price recorded-at]} rates]
|
|
(json/write-str
|
|
(cond-> {:price (str price)
|
|
:prev_price (when prev-price (str prev-price))
|
|
:recorded_at (str recorded-at)}
|
|
rates (assoc :price_eur (str (.multiply (bigdec price) (:eur rates)))
|
|
:price_dkk (str (.multiply (bigdec price) (:dkk rates)))))))
|
|
|
|
(defn- start-ws-send-loop! [^WebSocketChannel channel price-atom rates-atom]
|
|
(let [queue (LinkedBlockingQueue.)
|
|
wkey (keyword (gensym "ws-"))
|
|
rkey (keyword (gensym "ws-r-"))]
|
|
(add-watch price-atom wkey
|
|
(fn [_ _ _ _] (.offer queue :update)))
|
|
(add-watch rates-atom rkey
|
|
(fn [_ _ _ _] (.offer queue :update)))
|
|
(when @price-atom
|
|
(.offer queue :update))
|
|
(future
|
|
(try
|
|
(loop []
|
|
(when (.isOpen channel)
|
|
(if (.poll queue 15 TimeUnit/SECONDS)
|
|
(when-let [v @price-atom]
|
|
(WebSockets/sendTextBlocking (price->json v @rates-atom) channel))
|
|
(WebSockets/sendTextBlocking "" channel))
|
|
(recur)))
|
|
(catch Exception _)
|
|
(finally
|
|
(remove-watch price-atom wkey)
|
|
(remove-watch rates-atom rkey)
|
|
(when (.isOpen channel)
|
|
(try (.close channel) (catch Exception _))))))))
|
|
|
|
(defn- ws-price-handler [{:keys [binance frankfurter]} _req]
|
|
(let [price-atom (:latest-price binance)
|
|
rates-atom (:rates frankfurter)]
|
|
{:undertow/websocket
|
|
{:on-open (fn [{:keys [^WebSocketChannel channel]}]
|
|
(start-ws-send-loop! channel price-atom rates-atom))}}))
|
|
|
|
(defn- start-strike-ws-loop! [^WebSocketChannel channel strike-atom]
|
|
(let [queue (LinkedBlockingQueue.)
|
|
wkey (keyword (gensym "ws-s-"))]
|
|
(add-watch strike-atom wkey
|
|
(fn [_ _ _ _] (.offer queue :update)))
|
|
(when @strike-atom
|
|
(.offer queue :update))
|
|
(future
|
|
(try
|
|
(loop []
|
|
(when (.isOpen channel)
|
|
(if (.poll queue 15 TimeUnit/SECONDS)
|
|
(when-let [rates @strike-atom]
|
|
(WebSockets/sendTextBlocking (json/write-str rates) channel))
|
|
(WebSockets/sendTextBlocking "" channel))
|
|
(recur)))
|
|
(catch Exception _)
|
|
(finally
|
|
(remove-watch strike-atom wkey)
|
|
(when (.isOpen channel)
|
|
(try (.close channel) (catch Exception _))))))))
|
|
|
|
(defn- ws-strike-handler [{:keys [strike]} _req]
|
|
(let [strike-atom (:latest-rates strike)]
|
|
{:undertow/websocket
|
|
{:on-open (fn [{:keys [^WebSocketChannel channel]}]
|
|
(start-strike-ws-loop! channel strike-atom))}}))
|
|
|
|
(defn- latest-price-handler [{:keys [binance frankfurter]} _req]
|
|
(let [price-atom (:latest-price binance)
|
|
rates-atom (:rates frankfurter)
|
|
v @price-atom]
|
|
(if v
|
|
{:status 200
|
|
:headers {"Content-Type" "application/json"}
|
|
:body (price->json v @rates-atom)}
|
|
{:status 503
|
|
:headers {"Content-Type" "application/json"}
|
|
:body (json/write-str {:error "No price data yet"})})))
|
|
|
|
(defn- rate-row->json [row]
|
|
(json/write-str {:rate_date (str (:rate_date row))
|
|
:eur (str (:eur row))
|
|
:dkk (str (:dkk row))
|
|
:eur_dkk (str (:eur_dkk row))}))
|
|
|
|
(defn- currencies-handler [{:keys [query-fn frankfurter]} req]
|
|
(let [date-str (get-in req [:path-params :date])]
|
|
(try
|
|
(let [date (LocalDate/parse date-str)
|
|
today (LocalDate/now)]
|
|
(cond
|
|
(.isAfter date today)
|
|
{:status 400
|
|
:headers {"Content-Type" "application/json"}
|
|
:body (json/write-str {:error "Future date not allowed"})}
|
|
|
|
(.isEqual date today)
|
|
(if-let [row (query-fn :get-latest-currency-rates {})]
|
|
{:status 200
|
|
:headers {"Content-Type" "application/json"}
|
|
:body (rate-row->json row)}
|
|
{:status 404
|
|
:headers {"Content-Type" "application/json"}
|
|
:body (json/write-str {:error "No rates available yet"})})
|
|
|
|
:else
|
|
(if-let [row (query-fn :get-currency-rates-by-date {:rate-date date})]
|
|
{:status 200
|
|
:headers {"Content-Type" "application/json"}
|
|
:body (rate-row->json row)}
|
|
(if-let [result (rates/fetch-and-persist-for-date!
|
|
(:client frankfurter) (:url frankfurter)
|
|
query-fn date)]
|
|
{:status 200
|
|
:headers {"Content-Type" "application/json"}
|
|
:body (rate-row->json result)}
|
|
{:status 502
|
|
:headers {"Content-Type" "application/json"}
|
|
:body (json/write-str {:error "Failed to fetch rates from Frankfurter"})}))))
|
|
(catch DateTimeParseException _
|
|
{:status 400
|
|
:headers {"Content-Type" "application/json"}
|
|
:body (json/write-str {:error "Invalid date format, use YYYY-MM-DD"})}))))
|
|
|
|
(defn- kraken-hour-handler [{:keys [query-fn]} _req]
|
|
(let [rows (query-fn :get-kraken-hour-latest-60 {})]
|
|
{:status 200
|
|
:headers {"Content-Type" "application/json"}
|
|
:body (json/write-str
|
|
(mapv (fn [r]
|
|
{:ts (:ts r)
|
|
:open (str (:open r))
|
|
:high (str (:high r))
|
|
:low (str (:low r))
|
|
:close (str (:close r))
|
|
:vwap (str (:vwap r))
|
|
:volume (str (:volume r))
|
|
:trade_count (:trade_count r)})
|
|
rows))}))
|
|
|
|
(defn- api-routes [opts]
|
|
[["/swagger.json"
|
|
{:get {:no-doc true
|
|
:swagger {:info {:title "btcdata API"}}
|
|
:handler (swagger/create-swagger-handler)}}]
|
|
["/health"
|
|
{:get health/healthcheck!}]
|
|
["/price/ws"
|
|
{:get (fn [req] (ws-price-handler opts req))
|
|
:no-doc true
|
|
:middleware []}]
|
|
["/price/latest"
|
|
{:get (fn [req] (latest-price-handler opts req))}]
|
|
["/currencies/:date"
|
|
{:get (fn [req] (currencies-handler opts req))}]
|
|
["/strike/ws"
|
|
{:get (fn [req] (ws-strike-handler opts req))
|
|
:no-doc true
|
|
:middleware []}]
|
|
["/wallets"
|
|
{:get (fn [req] (tx/list-wallets opts req))
|
|
:post (fn [req] (tx/create-wallet! opts req))}]
|
|
["/wallets/balances"
|
|
{:get (fn [req] (tx/get-wallet-balances opts req))}]
|
|
["/wallets/:id/events"
|
|
{:get (fn [req] (tx/get-wallet-events opts req))}]
|
|
["/events"
|
|
{:get (fn [req] (tx/list-events opts req))
|
|
:post (fn [req] (tx/create-event! opts req))}]
|
|
["/deposits"
|
|
{:get (fn [req] (tx/list-deposits opts req))}]
|
|
["/deposits/rebuild"
|
|
{:post (fn [req] (tx/rebuild-deposits! opts req))}]
|
|
["/kraken-hour"
|
|
{:get (fn [req] (kraken-hour-handler opts req))}]])
|
|
|
|
(defn route-data [opts]
|
|
(merge
|
|
opts
|
|
{:coercion malli/coercion
|
|
:muuntaja formats/instance
|
|
:swagger {:id ::api}
|
|
:middleware [parameters/parameters-middleware
|
|
muuntaja/format-negotiate-middleware
|
|
muuntaja/format-response-middleware
|
|
exception/exception-middleware
|
|
muuntaja/format-request-middleware
|
|
coercion/coerce-request-middleware
|
|
coercion/coerce-response-middleware]}))
|
|
|
|
(derive :reitit.routes/api :reitit/routes)
|
|
|
|
(defmethod ig/init-key :reitit.routes/api
|
|
[_ {:keys [base-path]
|
|
:or {base-path ""}
|
|
:as opts}]
|
|
[base-path (route-data opts) (api-routes opts)])
|