diff --git a/CLAUDE.md b/CLAUDE.md index 0c3c948..bdcc326 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -44,8 +44,8 @@ Docker runs alongside local dev on different ports and databases: ## API Endpoints - `GET /api/health` — Health check -- `GET /api/price/ws` — WebSocket endpoint for live HTML price fragments (HTMX-compatible) -- `GET /api/price/latest` — Latest price as JSON +- `GET /api/price/ws` — WebSocket endpoint for live HTML price fragments with EUR/DKK rates (HTMX-compatible) +- `GET /api/price/latest` — Latest price + EUR/DKK rates as JSON ## Source Layout @@ -55,6 +55,8 @@ src/clj/pmagnus/btcdata/ ├── config.clj # System config loader ├── ws/ │ └── binance.clj # Binance WebSocket client +├── frankfurter/ +│ └── rates.clj # EUR/DKK exchange rate poller ├── kraken/ │ ├── ohlc.clj # Hourly OHLC poller │ └── ohlc_daily.clj # Daily OHLC poller diff --git a/docker-compose.yml b/docker-compose.yml index 940043f..0b828e5 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -4,10 +4,13 @@ services: restart: unless-stopped ports: - "4101:4101" + depends_on: + - frankfurter environment: BTCDATA_PORT: "4101" JDBC_URL: "jdbc:postgresql://postgres:5432/btcprod?user=postgres&password=ratata,123" CORS_ORIGIN: "http://localhost:4041" + FRANKFURTER_URL: "http://frankfurter:8080" extra_hosts: - "postgres:host-gateway" diff --git a/resources/system.edn b/resources/system.edn index bdb8e96..71c0d8c 100644 --- a/resources/system.edn +++ b/resources/system.edn @@ -44,9 +44,10 @@ :migrate-on-init? true} :reitit.routes/api - {:base-path "/api" - :query-fn #ig/ref :db.sql/query-fn - :binance #ig/ref :ws/binance} + {:base-path "/api" + :query-fn #ig/ref :db.sql/query-fn + :binance #ig/ref :ws/binance + :frankfurter #ig/ref :frankfurter/rates} :ws/binance {:query-fn #ig/ref :db.sql/query-fn @@ -56,4 +57,7 @@ {:query-fn #ig/ref :db.sql/query-fn} :kraken/ohlc-day - {:query-fn #ig/ref :db.sql/query-fn}} + {:query-fn #ig/ref :db.sql/query-fn} + + :frankfurter/rates + {:url #or [#env FRANKFURTER_URL "http://localhost:8080"]}} diff --git a/src/clj/pmagnus/btcdata/frankfurter/rates.clj b/src/clj/pmagnus/btcdata/frankfurter/rates.clj new file mode 100644 index 0000000..8ff3ff5 --- /dev/null +++ b/src/clj/pmagnus/btcdata/frankfurter/rates.clj @@ -0,0 +1,66 @@ +(ns pmagnus.btcdata.frankfurter.rates + (:require + [clojure.data.json :as json] + [clojure.tools.logging :as log] + [integrant.core :as ig]) + (:import + [java.net URI] + [java.net.http HttpClient HttpRequest HttpResponse$BodyHandlers] + [java.time Instant])) + +(defn- fetch-rates + "GET /latest?from=USD&to=EUR,DKK from Frankfurter. Returns {:EUR x :DKK y}." + [^HttpClient client base-url] + (let [url (str base-url "/latest?from=USD&to=EUR,DKK") + request (-> (HttpRequest/newBuilder) + (.uri (URI. url)) + (.header "Accept" "application/json") + (.GET) + (.build)) + resp (.send client request (HttpResponse$BodyHandlers/ofString)) + body (json/read-str (.body resp) :key-fn keyword)] + (:rates body))) + +(defn- poll! + "Fetch rates and update the atom." + [client base-url rates-atom] + (let [rates (fetch-rates client base-url)] + (when (and (:EUR rates) (:DKK rates)) + (reset! rates-atom + {:eur (bigdec (str (:EUR rates))) + :dkk (bigdec (str (:DKK rates))) + :updated-at (Instant/now)}) + (log/info "Frankfurter rates — EUR:" (:EUR rates) "DKK:" (:DKK rates))))) + +(defn- start-poll-loop! + "Fetch immediately, then poll every 60 minutes." + [client base-url rates-atom running?] + (future + (try + (poll! client base-url rates-atom) + (catch Exception e + (log/error e "Frankfurter initial fetch failed"))) + (while @running? + (Thread/sleep 3600000) + (when @running? + (try + (poll! client base-url rates-atom) + (catch Exception e + (log/error e "Frankfurter poll failed"))))))) + +(defmethod ig/init-key :frankfurter/rates + [_ {:keys [url]}] + (log/info "Starting Frankfurter rates poller:" url) + (let [client (HttpClient/newHttpClient) + rates (atom nil) + running? (atom true) + fut (start-poll-loop! client url rates running?)] + {:rates rates + :running? running? + :future fut})) + +(defmethod ig/halt-key! :frankfurter/rates + [_ {:keys [running? future]}] + (log/info "Stopping Frankfurter rates poller") + (reset! running? false) + (future-cancel future)) diff --git a/src/clj/pmagnus/btcdata/web/routes/api.clj b/src/clj/pmagnus/btcdata/web/routes/api.clj index b816b4f..f6bc400 100644 --- a/src/clj/pmagnus/btcdata/web/routes/api.clj +++ b/src/clj/pmagnus/btcdata/web/routes/api.clj @@ -18,15 +18,17 @@ [java.time.format DateTimeFormatter] [java.util.concurrent LinkedBlockingQueue TimeUnit])) -(defn- price->json [{:keys [price prev-price recorded-at]}] +(defn- price->json [{:keys [price prev-price recorded-at]} rates] (json/write-str - {:price (str price) - :prev_price (when prev-price (str prev-price)) - :recorded_at (str recorded-at)})) + (cond-> {:price (str price) + :prev_price (when prev-price (str prev-price)) + :recorded_at (str recorded-at)} + rates (assoc :eur (str (:eur rates)) + :dkk (str (:dkk rates)))))) (def ^:private ts-fmt (DateTimeFormatter/ofPattern "dd-MM HH:mm:ss")) -(defn- price->html [{:keys [price prev-price recorded-at]}] +(defn- price->html [{:keys [price prev-price recorded-at]} rates] (let [p (double (bigdec price)) pp (when prev-price (double (bigdec prev-price))) color (cond @@ -39,50 +41,64 @@ "
" "
" "

$" (format "%,.2f" p) "

" + (when rates + (let [eur (double (:eur rates)) + dkk (double (:dkk rates))] + (str "
" + "EUR " (format "%.4f" eur) "" + "DKK " (format "%.4f" dkk) "" + "
"))) "

Updated " ts "

" "
"))) -(defn- start-ws-send-loop! [^WebSocketChannel channel price-atom] +(defn- start-ws-send-loop! [^WebSocketChannel channel price-atom rates-atom] (let [queue (LinkedBlockingQueue.) - wkey (keyword (gensym "ws-"))] + wkey (keyword (gensym "ws-")) + rkey (keyword (gensym "ws-r-"))] (.set (.getReceiveSetter channel) (proxy [AbstractReceiveListener] [])) (.resumeReceives channel) (add-watch price-atom wkey - (fn [_ _ _ v] (.offer queue v))) - (when-let [v @price-atom] - (.offer queue v)) + (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-let [v (.poll queue 15 TimeUnit/SECONDS)] - (WebSockets/sendTextBlocking (price->html v) channel) + (if (.poll queue 15 TimeUnit/SECONDS) + (when-let [v @price-atom] + (WebSockets/sendTextBlocking (price->html 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]} req] +(defn- ws-price-handler [{:keys [binance frankfurter]} req] (let [^HttpServerExchange exchange (:server-exchange req) price-atom (:latest-price binance) + rates-atom (:rates frankfurter) callback (proxy [WebSocketConnectionCallback] [] (onConnect [_ws-exchange channel] - (start-ws-send-loop! channel price-atom))) + (start-ws-send-loop! channel price-atom rates-atom))) handler (WebSocketProtocolHandshakeHandler. callback)] (.handleRequest handler exchange) nil)) -(defn- latest-price-handler [{:keys [binance]} _req] +(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)} + :body (price->json v @rates-atom)} {:status 503 :headers {"Content-Type" "application/json"} :body (json/write-str {:error "No price data yet"})})))