Add deposits rebuild endpoint and fix date serialization

Add POST /api/deposits/rebuild to re-project all bank_to_exchange events.
Fix date-off-by-one by converting java.sql.Date to LocalDate via
ReadableColumn and serializing with Jackson JavaTimeModule.
Sort all event/deposit queries by occurred_at, id.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

deps.edn

@@ -26,6 +26,8 @@
 
         ;; Serialization
         metosin/muuntaja                     {:mvn/version "0.6.11"}
+        metosin/jsonista                     {:mvn/version "0.3.12"}
+        com.fasterxml.jackson.datatype/jackson-datatype-jsr310 {:mvn/version "2.18.3"}
         luminus-transit/luminus-transit       {:mvn/version "0.1.6"}
 
         ;; Database

resources/migrations/20260308100000-add-wallet-type.down.sql

@@ -0,0 +1 @@
+ALTER TABLE wallets DROP COLUMN wallet_type;

resources/migrations/20260308100000-add-wallet-type.up.sql

@@ -0,0 +1,4 @@
+ALTER TABLE wallets ADD COLUMN wallet_type TEXT NOT NULL DEFAULT 'exchange';
+--;;
+UPDATE wallets SET wallet_type = 'wallet'
+WHERE name IN ('Cold', 'Coldcard', 'Nunchuk Multi Sig');

resources/migrations/20260308200000-events-occurred-at-date.down.sql

@@ -0,0 +1,5 @@
+ALTER TABLE deposits ALTER COLUMN occurred_at TYPE TIMESTAMPTZ USING occurred_at::timestamptz;
+--;;
+ALTER TABLE events ALTER COLUMN occurred_at SET DEFAULT NOW();
+--;;
+ALTER TABLE events ALTER COLUMN occurred_at TYPE TIMESTAMPTZ USING occurred_at::timestamptz;

resources/migrations/20260308200000-events-occurred-at-date.up.sql

@@ -0,0 +1,5 @@
+ALTER TABLE events ALTER COLUMN occurred_at TYPE DATE USING occurred_at::date;
+--;;
+ALTER TABLE events ALTER COLUMN occurred_at DROP DEFAULT;
+--;;
+ALTER TABLE deposits ALTER COLUMN occurred_at TYPE DATE USING occurred_at::date;

resources/queries.sql

@@ -70,15 +70,19 @@ SELECT rate_date, eur, dkk, eur_dkk, updated_at FROM currencies WHERE rate_date
 
 -- :name insert-wallet! :! :n
 -- :doc Insert a new wallet
-INSERT INTO wallets (name) VALUES (:name)
+INSERT INTO wallets (name, wallet_type) VALUES (:name, :wallet-type)
 
 -- :name get-all-wallets :? :*
 -- :doc Get all wallets
-SELECT id, name, created_at FROM wallets ORDER BY name
+SELECT id, name, wallet_type, created_at FROM wallets ORDER BY name
+
+-- :name get-wallets-by-type :? :*
+-- :doc Get wallets filtered by type
+SELECT id, name, wallet_type, created_at FROM wallets WHERE wallet_type = :wallet-type ORDER BY name
 
 -- :name get-wallet-by-id :? :1
 -- :doc Get a wallet by ID
-SELECT id, name, created_at FROM wallets WHERE id = :id
+SELECT id, name, wallet_type, created_at FROM wallets WHERE id = :id
 
 -- Events --------------------------------------------------------------------
 
@@ -98,7 +102,7 @@ SELECT e.id, e.event_type, e.occurred_at, e.recorded_at,
 FROM events e
 LEFT JOIN wallets fw ON fw.id = e.from_wallet_id
 LEFT JOIN wallets tw ON tw.id = e.to_wallet_id
-ORDER BY e.occurred_at DESC
+ORDER BY e.occurred_at DESC, e.id DESC
 
 -- :name get-events-by-type :? :*
 -- :doc Get events filtered by type
@@ -111,7 +115,7 @@ FROM events e
 LEFT JOIN wallets fw ON fw.id = e.from_wallet_id
 LEFT JOIN wallets tw ON tw.id = e.to_wallet_id
 WHERE e.event_type = :event-type
-ORDER BY e.occurred_at DESC
+ORDER BY e.occurred_at DESC, e.id DESC
 
 -- :name get-events-by-wallet :? :*
 -- :doc Get events involving a specific wallet
@@ -124,7 +128,20 @@ FROM events e
 LEFT JOIN wallets fw ON fw.id = e.from_wallet_id
 LEFT JOIN wallets tw ON tw.id = e.to_wallet_id
 WHERE e.from_wallet_id = :wallet-id OR e.to_wallet_id = :wallet-id
-ORDER BY e.occurred_at DESC
+ORDER BY e.occurred_at DESC, e.id DESC
+
+-- :name truncate-deposits! :! :n
+-- :doc Delete all rows from the deposits read table
+TRUNCATE deposits
+
+-- :name get-deposit-events :? :*
+-- :doc Get bank_to_exchange events with exchange name, oldest first
+SELECT e.id, e.occurred_at, e.fiat_amount, e.fiat_currency,
+       e.to_wallet_id, tw.name AS exchange, e.note
+FROM events e
+LEFT JOIN wallets tw ON tw.id = e.to_wallet_id
+WHERE e.event_type = 'bank_to_exchange'
+ORDER BY e.occurred_at ASC, e.id ASC
 
 -- :name insert-deposit! :! :n
 -- :doc Insert a projected deposit row
@@ -136,7 +153,7 @@ VALUES (:event-id, :occurred-at, :exchange, :fiat-amount, :fiat-currency, :amoun
 SELECT id, event_id, occurred_at, exchange, fiat_amount, fiat_currency,
        amount_eur, amount_dkk, amount_usd, note
 FROM deposits
-ORDER BY occurred_at DESC
+ORDER BY occurred_at DESC, id DESC
 
 -- :name get-wallet-balances :? :*
 -- :doc Compute sats balance per wallet from events

resources/system.edn

@@ -52,7 +52,8 @@
 
  :ws/binance
  {:query-fn #ig/ref :db.sql/query-fn
-  :uri      "wss://stream.binance.com:9443/ws/btcusdt@trade"}
+  :uri      "wss://stream.binance.com:9443/ws/btcusdt@trade"
+  :enabled? #or [#env BINANCE_ENABLED "true"]}
 
  :kraken/ohlc
  {:query-fn #ig/ref :db.sql/query-fn}
@@ -67,4 +68,5 @@
  :strike/ticker
  {:query-fn        #ig/ref :db.sql/query-fn
   :api-key         #env STRIKE_API_KEY
-  :sats-eur-amount #or [#env SATS_EUR_AMOUNT "55"]}}
+  :sats-eur-amount #or [#env SATS_EUR_AMOUNT "55"]
+  :enabled?        #or [#env STRIKE_ENABLED "true"]}}

src/clj/pmagnus/btcdata/core.clj

@@ -2,6 +2,7 @@
   (:require
    [clojure.tools.logging :as log]
    [integrant.core :as ig]
+   [next.jdbc.result-set]
    [pmagnus.btcdata.config :as config]
    [pmagnus.btcdata.env :refer [defaults]]
 
@@ -20,7 +21,22 @@
    [pmagnus.btcdata.kraken.ohlc-daily]
    [pmagnus.btcdata.frankfurter.rates]
    [pmagnus.btcdata.strike.ticker])
-  (:gen-class))
+  (:gen-class)
+  (:import
+   [java.sql Date Timestamp]))
+
+;; Read SQL DATE as LocalDate and TIMESTAMP as Instant (timezone-safe)
+(extend-protocol next.jdbc.result-set/ReadableColumn
+  Date
+  (read-column-by-label [v _]
+    (.toLocalDate v))
+  (read-column-by-index [v _ _]
+    (.toLocalDate v))
+  Timestamp
+  (read-column-by-label [v _]
+    (.toInstant v))
+  (read-column-by-index [v _ _]
+    (.toInstant v)))
 
 (defonce system (atom nil))
 

src/clj/pmagnus/btcdata/strike/ticker.clj

@@ -136,8 +136,13 @@
             (log/error e "Strike poll failed")))))))
 
 (defmethod ig/init-key :strike/ticker
-  [_ {:keys [query-fn api-key sats-eur-amount]}]
-  (log/info "Starting Strike ticker poller")
+  [_ {:keys [query-fn api-key sats-eur-amount enabled?] :or {enabled? "true"}}]
+  (if-not (= enabled? "true")
+    (do (log/info "Strike ticker disabled")
+        {:running?     (atom false)
+         :future       nil
+         :latest-rates (atom nil)})
+    (do (log/info "Starting Strike ticker poller")
         (let [client       (HttpClient/newHttpClient)
               running?     (atom true)
               latest-rates (atom nil)
@@ -145,10 +150,10 @@
               fut          (start-poll-loop! client api-key query-fn running? latest-rates sats-eur-amount fee-cache)]
           {:running?     running?
            :future       fut
-     :latest-rates latest-rates}))
+           :latest-rates latest-rates}))))
 
 (defmethod ig/halt-key! :strike/ticker
   [_ {:keys [running? future]}]
   (log/info "Stopping Strike ticker poller")
   (reset! running? false)
-  (future-cancel future))
+  (when future (future-cancel future)))

src/clj/pmagnus/btcdata/web/controllers/transactions.clj

@@ -5,17 +5,20 @@
    [pmagnus.btcdata.frankfurter.rates :as rates]
    [ring.util.http-response :as response])
   (:import
-   [java.time Instant LocalDate ZoneOffset]
+   [java.time LocalDate]
    [java.util UUID]))
 
-(defn list-wallets [{:keys [query-fn]} _req]
-  (response/ok (query-fn :get-all-wallets {})))
+(defn list-wallets [{:keys [query-fn]} req]
+  (let [wallet-type (get-in req [:query-params "type"])]
+    (if wallet-type
+      (response/ok (query-fn :get-wallets-by-type {:wallet-type wallet-type}))
+      (response/ok (query-fn :get-all-wallets {})))))
 
 (defn create-wallet! [{:keys [query-fn]} req]
-  (let [{:keys [name]} (:body-params req)]
+  (let [{:keys [name wallet_type]} (:body-params req)]
     (if (str/blank? name)
       (response/bad-request {:error "name is required"})
-      (do (query-fn :insert-wallet! {:name name})
+      (do (query-fn :insert-wallet! {:name name :wallet-type (or wallet_type "exchange")})
           (response/created "/api/wallets" {:name name})))))
 
 (defn get-wallet-balances [{:keys [query-fn]} _req]
@@ -48,11 +51,10 @@
   (try
     (let [wallet   (query-fn :get-wallet-by-id {:id to-wallet-id})
           exchange (or (:name wallet) "Unknown")
-          date     (.toLocalDate (.atOffset occurred-at ZoneOffset/UTC))
-          rates    (or (query-fn :get-currency-rates-by-date {:rate-date date})
+          rates    (or (query-fn :get-currency-rates-by-date {:rate-date occurred-at})
                        (rates/fetch-and-persist-for-date!
                          (:client frankfurter) (:url frankfurter)
-                         query-fn date))
+                         query-fn occurred-at))
           amounts  (if rates
                      (convert-amount fiat-amount fiat-currency rates)
                      {:amount-eur nil :amount-dkk nil :amount-usd nil})]
@@ -72,9 +74,9 @@
         event-type (:event_type params)]
     (if-not (#{"bank_to_exchange" "exchange_to_wallet" "wallet_to_wallet"} event-type)
       (response/bad-request {:error "Invalid event_type"})
-      (let [occurred-at (if-let [ts (:occurred_at params)]
-                          (Instant/parse ts)
-                          (Instant/now))
+      (let [occurred-at (if-let [d (:occurred_at params)]
+                          (LocalDate/parse d)
+                          (LocalDate/now))
             row {:event-type     event-type
                  :occurred-at    occurred-at
                  :sats           (:sats params)
@@ -96,6 +98,36 @@
 (defn list-deposits [{:keys [query-fn]} _req]
   (response/ok (query-fn :get-all-deposits {})))
 
+(defn rebuild-deposits! [{:keys [query-fn frankfurter]} _req]
+  (query-fn :truncate-deposits! {})
+  (let [events (query-fn :get-deposit-events {})
+        n      (reduce
+                 (fn [cnt {:keys [id occurred_at fiat_amount fiat_currency exchange note]}]
+                   (if-not fiat_amount
+                     cnt
+                     (try
+                       (let [rates   (or (query-fn :get-currency-rates-by-date {:rate-date occurred_at})
+                                         (rates/fetch-and-persist-for-date!
+                                           (:client frankfurter) (:url frankfurter)
+                                           query-fn occurred_at))
+                             amounts (if rates
+                                       (convert-amount fiat_amount fiat_currency rates)
+                                       {:amount-eur nil :amount-dkk nil :amount-usd nil})]
+                         (query-fn :insert-deposit!
+                                   (merge {:event-id      id
+                                           :occurred-at   occurred_at
+                                           :exchange      (or exchange "Unknown")
+                                           :fiat-amount   fiat_amount
+                                           :fiat-currency fiat_currency
+                                           :note          note}
+                                          amounts))
+                         (inc cnt))
+                       (catch Exception e
+                         (log/error e "Failed to rebuild deposit for event" id)
+                         cnt))))
+                 0 events)]
+    (response/ok {:rebuilt n})))
+
 (defn list-events [{:keys [query-fn]} req]
   (let [event-type (get-in req [:query-params "type"])]
     (if event-type

src/clj/pmagnus/btcdata/web/middleware/formats.clj

@@ -1,11 +1,23 @@
 (ns pmagnus.btcdata.web.middleware.formats
   (:require
+   [jsonista.core :as j]
    [luminus-transit.time :as time]
-   [muuntaja.core :as m]))
+   [muuntaja.core :as m])
+  (:import
+   [com.fasterxml.jackson.databind SerializationFeature]
+   [com.fasterxml.jackson.datatype.jsr310 JavaTimeModule]))
+
+(def ^:private mapper
+  (j/object-mapper
+   {:modules [(JavaTimeModule.)]
+    :decode-key-fn true
+    :configure {SerializationFeature/WRITE_DATES_AS_TIMESTAMPS false}}))
 
 (def instance
   (m/create
    (-> m/default-options
+       (assoc-in [:formats "application/json" :decoder-opts] {:mapper mapper})
+       (assoc-in [:formats "application/json" :encoder-opts] {:mapper mapper})
        (update-in [:formats "application/transit+json" :decoder-opts]
                   (partial merge time/time-deserialization-handlers))
        (update-in [:formats "application/transit+json" :encoder-opts]

src/clj/pmagnus/btcdata/web/routes/api.clj

@@ -174,7 +174,9 @@
     {: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))}]])
+    {:get (fn [req] (tx/list-deposits opts req))}]
+   ["/deposits/rebuild"
+    {:post (fn [req] (tx/rebuild-deposits! opts req))}]])
 
 (defn route-data [opts]
   (merge

src/clj/pmagnus/btcdata/ws/binance.clj

@@ -71,8 +71,12 @@
         (.join))))
 
 (defmethod ig/init-key :ws/binance
-  [_ {:keys [query-fn uri]}]
-  (log/info "Starting Binance WebSocket listener:" uri)
+  [_ {:keys [query-fn uri enabled?] :or {enabled? "true"}}]
+  (if-not (= enabled? "true")
+    (do (log/info "Binance WebSocket disabled")
+        {:state        (atom {:running? false})
+         :latest-price (atom nil)})
+    (do (log/info "Starting Binance WebSocket listener:" uri)
         (let [latest-price (atom nil)
               state        (atom {:running?   true
                                   :last-write 0
@@ -81,7 +85,7 @@
               ws           (connect! uri query-fn state latest-price)]
           (swap! state assoc :ws ws)
           {:state        state
-     :latest-price latest-price}))
+           :latest-price latest-price}))))
 
 (defmethod ig/halt-key! :ws/binance
   [_ {:keys [state]}]