Restart btcdata when its fenced database session closes

Reviewed-on: #2

.gitea/workflows/build.yml

@@ -0,0 +1,73 @@
+name: build
+
+on:
+  push:
+    branches: [master]
+    tags: ['v*']
+  workflow_dispatch:
+
+env:
+  REGISTRY: harbor.pmagnus.cloud
+  IMAGE: harbor.pmagnus.cloud/btcdata/btcdata
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Unit and PostgreSQL fencing tests
+        run: bash scripts/test-fencing.sh
+
+      - name: Generate immutable tags
+        id: tags
+        shell: bash
+        run: |
+          set -euo pipefail
+          SHA=$(git rev-parse --short HEAD)
+          case "$GITHUB_REF" in
+            refs/tags/v*) VERSION="${GITHUB_REF_NAME#v}" ;;
+            *) VERSION="main-$(date -u +%Y%m%d%H%M%S)" ;;
+          esac
+          echo "primary=$IMAGE:$VERSION" >> "$GITHUB_OUTPUT"
+          echo "qualified=$IMAGE:$VERSION-$SHA" >> "$GITHUB_OUTPUT"
+
+      # anvil is not a NetBird peer. Keep BuildKit on the runner host network
+      # until anvil can resolve and reach Harbor directly from job networks.
+      - uses: docker/setup-buildx-action@v3
+        with:
+          driver-opts: network=host
+
+      - uses: docker/login-action@v3
+        with:
+          registry: ${{ env.REGISTRY }}
+          username: ${{ secrets.HARBOR_ROBOT_USER }}
+          password: ${{ secrets.HARBOR_ROBOT_TOKEN }}
+
+      - uses: docker/build-push-action@v6
+        with:
+          context: .
+          platforms: linux/amd64
+          push: true
+          tags: ${{ steps.tags.outputs.primary }}
+          cache-from: type=registry,ref=${{ env.IMAGE }}:buildcache
+          cache-to: type=registry,ref=${{ env.IMAGE }}:buildcache,mode=max
+
+      - name: Wait for Harbor policy, then add SHA tag
+        env:
+          PRIMARY: ${{ steps.tags.outputs.primary }}
+          QUALIFIED: ${{ steps.tags.outputs.qualified }}
+        run: |
+          for attempt in $(seq 1 60); do
+            if docker buildx imagetools inspect "$PRIMARY" >/dev/null 2>&1; then
+              docker buildx imagetools create --tag "$QUALIFIED" "$PRIMARY"
+              exit 0
+            fi
+            echo "waiting for Harbor scan/pull policy ($attempt/60)"
+            sleep 10
+          done
+          echo "Harbor did not approve $PRIMARY within 10 minutes" >&2
+          exit 1
+
+      - name: Summary
+        run: echo "pushed ${{ steps.tags.outputs.primary }} and ${{ steps.tags.outputs.qualified }}"

.gitea/workflows/pr.yml

@@ -0,0 +1,14 @@
+name: pr
+on:
+  pull_request:
+    branches: [master]
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+      - name: Unit and PostgreSQL fencing tests
+        run: bash scripts/test-fencing.sh
+# No registry, deployment, or production credentials in PR jobs.

.gitignore

@@ -15,3 +15,4 @@ pom.xml.asc
 *.log
 /.env
 /token
+affald.txt

CLAUDE.md

@@ -37,15 +37,15 @@ Docker runs alongside local dev on different ports and databases:
 - **Framework:** Kit (Integrant-based)
 - **Server:** Undertow
 - **Routing:** Reitit
-- **Database:** PostgreSQL + conman + Migratus
+- **Database:** PostgreSQL + conman + Migratus — the Postgres lives on **citadel** (`192.168.50.195:5432`); `btcprice` for local dev, `btcprod` for the deployed container
 - **Port:** 4100 (env: BTCDATA_PORT)
 - **CORS:** Configured via CORS_ORIGIN env var
 
 ## 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

Dockerfile

@@ -11,8 +11,9 @@ FROM eclipse-temurin:21-jre-alpine
 COPY --from=build /build/target/btcdata-standalone.jar /btcdata/btcdata-standalone.jar
 
 EXPOSE 4101
+ENV PATH="/opt/java/openjdk/bin:${PATH}"
 ENV BTCDATA_PORT=4101
 ENV CORS_ORIGIN=http://localhost:4000
 ENV JDBC_URL=jdbc:postgresql://postgres:5432/btcprod
 
-CMD ["java", "-jar", "/btcdata/btcdata-standalone.jar"]
+CMD ["/opt/java/openjdk/bin/java", "-jar", "/btcdata/btcdata-standalone.jar"]

Makefile

@@ -1,4 +1,4 @@
-include .env
+-include .env
 export
 
 .PHONY: clean run repl test uberjar

README.md

@@ -9,6 +9,33 @@ Bitcoin data backend service. Fetches BTC prices from Binance (WebSocket) and Kr
 | `GET /api/price/ws` | WebSocket — sends HTML fragments with live price updates (HTMX-compatible) |
 | `GET /api/price/latest` | Latest price as JSON |
 | `GET /api/health` | Health check |
+| `GET /api/price/day/:date` | Completed UTC BTC/USD daily OHLC, e.g. `/api/price/day/2020-01-01` |
+| `GET /api/price/day?date=YYYY-MM-DD` | Same daily OHLC endpoint using a query parameter |
+| `GET /api/swagger.json` | Swagger specification |
+| `GET /swagger` | Interactive Swagger UI |
+
+### Daily OHLC
+
+Dates must be valid `YYYY-MM-DD` values strictly before today in UTC. Both
+`kraken_day` and `kraken_day_history` are checked before contacting Kraken.
+Missing recent candles are fetched from the OHLC API and saved in `kraken_day`.
+Older days outside Kraken's 720-candle window are calculated from paginated
+historical trades and saved in `kraken_day_history` for subsequent requests.
+Only trades in `[00:00 UTC, next midnight UTC)` contribute; VWAP is volume-weighted.
+
+Example response (prices and volume are JSON numbers):
+
+```json
+{"date":"2020-01-01","pair":"BTC/USD","open":7200,"high":7300,"low":7100,"close":7250,"vwap":7230.5,"volume":123.45,"trade_count":1000}
+```
+
+Errors are JSON: 400 for invalid, current or future dates, 404 for days without
+trades, and 502 for Kraken failures. Failed calculations are not cached.
+Historical requests wait for completion and can take several minutes; clients
+and reverse proxies need sufficiently long request timeouts. Kraken requests
+have a 60-second per-page timeout and historical pages are spaced one second apart.
+
+The `kraken_day_history` migration runs automatically on application startup.
 
 ## Quick Start
 

btcdata-deploy.md

@@ -0,0 +1,349 @@
+# btcdata Kubernetes deployment plan
+
+## 2026-10-09 incident: closed writer session never restarted
+
+The citadel pod was Running but not Ready, and both btcprice pods were unready.
+All writer queries failed with `Connection is closed`. Readiness correctly
+returned 503, but `/api/health` always returned 200, so Kubernetes did not restart
+the process. The initial connection-loss cause was not established from the
+available logs. Fenced connections deliberately never reconnect in place.
+
+Recovery deleted only the failed pod (normal graceful deletion); its Deployment
+recreated it, and btcdata and both btcprice pods became Ready. No lock was
+bypassed or database backend terminated for this incident.
+
+The source fix makes `/api/health` return 503 when the writer session is closed
+(or absent). This checks local JDBC state only: no SQL, external APIs, or Binance
+freshness are used in liveness. Existing probes will then restart an irrecoverable
+writer after three failed checks. Readiness retains database/freshness checks.
+Unit tests passed: 21 tests, 63 assertions. PostgreSQL integration tests were
+not run because the local Docker daemon was unavailable.
+
+The recovered pod still runs `main-20260930155338`, not the new source changes.
+Flux apps reconciliation is blocked on infrastructure health checks for image
+repositories; building/publishing and a successful Flux rollout remain necessary
+to activate the liveness and keepalive source fixes.
+
+## 2026-10-07 incident: orphaned writer session
+
+The `sejs/btcdata` pod on citadel entered CrashLoopBackOff with
+`Another btcdata writer holds the database lock`. The prior pod on srv1893094
+had been removed after readiness/network failures, but its PostgreSQL backend
+(PID 3531439, client 100.75.252.111) still held advisory lock 728194106. No old
+btcdata Java process or database TCP connection remained on that node.
+
+Recovery terminated only that verified orphaned backend. Kubernetes restarted
+btcdata successfully; btcdata and both btcprice pods became Ready. Do not bypass
+the advisory lock or terminate its owner without first verifying the old writer
+is stopped.
+
+Production database `btcprod` now has connection defaults:
+`tcp_keepalives_idle=60`, `tcp_keepalives_interval=10`,
+`tcp_keepalives_count=3`, and `tcp_user_timeout=90000` (milliseconds).
+These apply to **new connections**, not the already-running recovered session.
+Server-side dead-peer detection prevents a disappeared node from retaining the
+lock for the default multi-hour TCP timeout; it does not evict a live competing
+writer. The application now also sets these values on its lock-owning session
+before acquiring the fence (source change; not yet built/deployed).
+
+Validation: 20 unit tests / 58 assertions passed; disposable PostgreSQL integration
+tests were not run because the local Docker daemon was unavailable. New
+production connection settings were verified with `SHOW`.
+
+Rollback database defaults, if necessary, with `ALTER DATABASE btcprod RESET
+<setting>` for each of the four settings above; new connections inherit the
+server defaults again. Reverting the application change requires a normal image
+build and Flux rollout. The orphan-session termination is not reversible.
+
+## 2026-09-19 status and proposed recovery
+
+Kubernetes deployment already exists: one `Recreate` replica on citadel, eligible
+on citadel and srv1893094. Both nodes were Ready. At 14:03 UTC the live price
+WebSocket upgraded successfully but returned data last recorded at 08:36:42 UTC.
+Harbor was refusing connections and Flux apps reconciliation was dependency-blocked.
+No production restart or relocation was performed during this investigation.
+
+The `fix/websocket-failover` application change proposes bounded Binance reconnects,
+60-second in-memory freshness readiness, and PostgreSQL single-writer fencing.
+Every application query uses the session holding advisory lock `728194106`; loss
+of that session cannot cause writes through another pooled connection. Competing
+instances fail startup rather than becoming simultaneous writers. This requires
+all writers to run the fenced version; legacy deployments must stay stopped.
+
+Companion deployment changes in `magnus/deploy` add a **suspended** watchdog:
+three failed minute checks after startup grace request relocation to the other
+healthy node, with a ten-minute cooldown. See `apps/btcdata/FAILOVER.md` in that
+repository for review gates, acceptance tests, limitations, and rollback. This is
+proposed code, not a claim that automatic failover is enabled in production.
+
+Run `make test` for unit tests. Run `bash scripts/test-fencing.sh` for unit tests
+plus PostgreSQL lock-contention, session-loss, and pooled-lock-release tests in
+isolated disposable containers. No production credentials are required.
+
+The remaining sections below are the original migration plan, not current status.
+
+## Goal and interpretation
+
+Deploy `btcdata` through the existing Flux-managed two-node k3s cluster (`srv1893094` and `citadel`), with **exactly one btcdata pod running at a time**. The pod may run on either node and Kubernetes may recreate it on the other node after failure.
+
+The current Compose instance is assumed to be `/home/pmp/apps/btcdata` on **citadel**. It must be stopped and verified stopped before the Kubernetes replica is increased from zero to one, because btcdata runs Binance, Kraken, Strike, and currency-rate writers.
+
+Production traffic remains internal: `btcprice` proxies all browser HTTP/WebSocket traffic to btcdata. btcdata does not need an Ingress.
+
+> A one-replica Deployment is active/passive scheduling, not a warm standby. It normally prevents duplicate writers, but it cannot strictly fence a partitioned old process. If strict single-writer guarantees are required, add PostgreSQL advisory-lock leader election before deployment.
+
+## Target architecture
+
+```text
+btcprice pods (namespace btcprice)
+  -> http://btcdata.btcdata.svc.cluster.local:4101
+  -> ClusterIP Service btcdata
+  -> one btcdata pod on srv1893094 OR citadel
+       -> PostgreSQL btcprod on citadel
+       -> Frankfurter Service in namespace btcdata
+       -> Binance, Kraken, and Strike APIs
+```
+
+Kubernetes objects and image automation belong in `magnus/deploy`; application image building belongs in this repository. Do not deploy the workload with Compose or direct `kubectl apply`.
+
+## Important limitations
+
+- PostgreSQL remains on citadel and is still a single point of failure. Moving the btcdata pod to `srv1893094` does not help if citadel or its database is unavailable.
+- The database must be reachable from both k3s nodes. Use a stable reachable address, preferably citadel's NetBird address, rather than the LAN-only `192.168.50.195`, after verifying PostgreSQL listens on and permits that path.
+- One replica causes a short interruption during upgrades and node failover. `btcprice` should become unready and reconnect when btcdata returns.
+- Kubernetes `Deployment` with one replica is not absolute split-brain protection during a network partition. PostgreSQL advisory locking is the recommended follow-up.
+
+## Phase 1 — preflight and decisions
+
+1. Synchronize the GitOps checkout before editing:
+
+   ```bash
+   git -C ../deploy switch main
+   git -C ../deploy pull --ff-only
+   ```
+
+2. Confirm both nodes are Ready and amd64:
+   - `srv1893094`
+   - `citadel`
+3. From a disposable pod pinned to each node, verify:
+   - PostgreSQL TCP connectivity;
+   - authentication to `btcprod` with a temporary, non-logged command;
+   - outbound HTTPS/WebSocket access to Binance, Kraken, Strike, and Harbor.
+4. Decide and record the production JDBC hostname. Do not put the password in Git.
+5. Confirm the legacy workload is the Compose project at `/home/pmp/apps/btcdata` on citadel and record:
+   - current image ID;
+   - effective non-secret environment;
+   - container names and Compose project name;
+   - a tested restart command for rollback.
+6. Back up `btcprod` and test that the backup can be read/restored.
+7. Confirm whether strict fencing is required. If yes, implement a PostgreSQL advisory lock so only the lock holder starts all writer components before proceeding.
+
+**Gate:** both nodes can reach PostgreSQL and required external APIs, a database backup exists, and the exact legacy stop/rollback commands are known.
+
+## Phase 2 — application readiness
+
+1. Run the existing test suite with `make test`.
+2. Add or confirm tests for startup, shutdown, migrations, health, and disabled pollers.
+3. Add a dependency-aware `/api/ready` endpoint if practical. It should check the database and required in-process state without performing external API calls. Keep `/api/health` process-only for liveness.
+4. Ensure SIGTERM shuts down Undertow and all polling futures within the Kubernetes termination grace period.
+5. Review database migrations for backward compatibility so rollback to the Compose image remains possible.
+6. Keep runtime configuration in environment variables:
+   - `BTCDATA_PORT=4101`
+   - `JDBC_URL` from a Secret
+   - `STRIKE_API_KEY` from a Secret
+   - `CORS_ORIGIN` restricted to the actual frontend origins if direct browser access is ever enabled
+   - `FRANKFURTER_URL=http://frankfurter:8080`
+   - `BINANCE_ENABLED=true`
+   - `KRAKEN_ENABLED=true`
+   - `STRIKE_ENABLED=true`
+
+**Gate:** tests pass, termination is clean, and the production image can start with no embedded credentials.
+
+## Phase 3 — CI image publication
+
+Create `.gitea/workflows/build.yml`, following the established btcprice pipeline:
+
+1. Trigger on `main` and `v*` tags.
+2. Run `make test`.
+3. Build `linux/amd64` from the existing multi-stage `Dockerfile`.
+4. Publish to `harbor.pmagnus.cloud/btcdata/btcdata` using the Harbor CI robot.
+5. Publish immutable tags:
+   - `main-YYYYMMDDhhmmss`
+   - `main-YYYYMMDDhhmmss-<short-sha>`
+6. Wait for Harbor scanning/policy approval before creating the companion tag.
+7. Configure Harbor scan-on-push, vulnerability blocking, retention, and garbage collection consistently with btcprice.
+8. Verify both k3s nodes can pull the approved digest with the existing pull credentials.
+
+Do not give CI a kubeconfig or a GitOps repository write key. Flux remains responsible for deployment.
+
+**Gate:** an approved immutable image is available and pullable from both nodes.
+
+## Phase 4 — secrets
+
+The deploy repository currently has no encrypted-secret mechanism. Until SOPS or an external secret operator is introduced:
+
+1. Create a `btcdata` namespace.
+2. Provision a `btcdata-runtime` Secret out of band, using stdin or a temporary mode-`0600` file, with:
+   - `JDBC_URL`
+   - `STRIKE_API_KEY`
+3. Never put decoded values, command output, or generated Secret YAML in either repository.
+4. Document how to recreate the Secret after cluster recovery in a protected operations document.
+5. Reference the Secret with `env[].valueFrom.secretKeyRef` in the Deployment.
+
+Prefer adding SOPS/age to Flux as a follow-up so secret recovery is declarative and encrypted.
+
+## Phase 5 — GitOps manifests, initially scaled to zero
+
+Create `../deploy/apps/btcdata/`:
+
+```text
+apps/btcdata/
+  namespace.yaml
+  deployment.yaml
+  service.yaml
+  frankfurter-deployment.yaml
+  frankfurter-service.yaml
+  pdb.yaml
+  kustomization.yaml
+```
+
+Update:
+
+- `../deploy/apps/kustomization.yaml` to include `btcdata`;
+- `../deploy/infrastructure/image-automation/kustomization.yaml`;
+- add `../deploy/infrastructure/image-automation/btcdata-image.yaml`.
+
+### btcdata Deployment
+
+Use:
+
+- namespace `btcdata`;
+- image `harbor.pmagnus.cloud/btcdata/btcdata:<approved-main-tag>`;
+- Flux image-policy marker for `flux-system:btcdata`;
+- **`replicas: 0` for the initial reconciliation**;
+- `strategy.type: Recreate` to avoid overlapping writer pods during normal upgrades;
+- node affinity allowing only `srv1893094` and `citadel`;
+- container port `4101`, named `http`;
+- `terminationGracePeriodSeconds: 30` or longer if shutdown testing requires it;
+- startup and liveness probes on `/api/health`;
+- readiness on `/api/ready` if added, otherwise `/api/health` temporarily;
+- conservative initial requests/limits, then tune from observed usage;
+- no host ports, host networking, persistent volume, or Ingress.
+
+A one-replica PDB with `minAvailable: 1` protects against accidental voluntary eviction, but it also blocks a normal drain. Document the controlled PDB override/removal required for planned node maintenance.
+
+### Service
+
+Create a selector-based ClusterIP Service named `btcdata` in namespace `btcdata`, port `4101` to the named `http` container port.
+
+### Frankfurter
+
+Deploy `lineofflight/frankfurter` behind a ClusterIP Service named `frankfurter` on port `8080`. Pin an immutable image version or digest rather than a floating tag. One replica is sufficient because btcdata retains rates in PostgreSQL and can retry temporary failures.
+
+### Image automation
+
+Add an `ImageRepository` and `ImagePolicy` matching only plain main tags:
+
+```yaml
+filterTags:
+  pattern: '^main-(?P<ts>[0-9]{14})$'
+  extract: '$ts'
+policy:
+  numerical:
+    order: asc
+```
+
+Validate all kustomizations locally. Commit and push to `magnus/deploy`; do not manually apply the workload. Confirm Flux creates the namespace, Service, Frankfurter, and a zero-replica btcdata Deployment.
+
+**Gate:** Flux is Ready, Frankfurter is healthy, the btcdata Service exists, and no btcdata Kubernetes pod is running.
+
+## Phase 6 — controlled cutover
+
+This sequence deliberately includes a short outage to guarantee that the old and new writers do not intentionally overlap.
+
+1. Pause Flux btcdata image automation for the cutover window, or verify that no new image can be selected during the procedure.
+2. On citadel, stop **only** the legacy btcdata Compose project in `/home/pmp/apps/btcdata`.
+   - Do not stop PostgreSQL.
+   - Do not remove `btc-network`, images, volumes, or deployment files.
+3. Verify before Kubernetes startup:
+   - the btcdata Compose container is stopped;
+   - port `4102` is no longer listening for btcdata;
+   - no legacy Java btcdata process remains;
+   - database write timestamps stop advancing except for expected external writers.
+4. In `magnus/deploy`, change the btcdata Deployment from `replicas: 0` to **`replicas: 1`**, commit, and push.
+5. Let Flux reconcile. Do not use `kubectl scale`, because Git would revert it.
+6. Confirm exactly one pod is Running and Ready on either allowed node.
+7. Check startup logs for successful migrations, database connection, Binance WebSocket, Kraken pollers, Frankfurter rates, and Strike ticker.
+8. From disposable pods pinned to each node, test the new Service FQDN:
+   - `/api/health`;
+   - `/api/price/latest`;
+   - `/api/kraken-minute` and `/api/kraken-hour`;
+   - `/api/price/ws` and `/api/strike/ws` WebSockets.
+9. Confirm database rows advance once at the expected cadence and no duplicate-writer symptoms appear.
+
+**Gate:** one Kubernetes pod is healthy, data writes are correct, and both nodes can reach its ClusterIP Service.
+
+## Phase 7 — switch btcprice to Kubernetes btcdata
+
+Modify `../deploy/apps/btcprice/` in one GitOps change:
+
+1. Change the Deployment environment to:
+
+   ```text
+   BTCDATA_URL=http://btcdata.btcdata.svc.cluster.local:4101
+   ```
+
+2. Remove the legacy selectorless `btcdata` Service from the `btcprice` namespace.
+3. Remove `btcdata-endpointslice.yaml` and its kustomization entry.
+4. Keep btcprice's existing two-replica scheduling, readiness behavior, and WebSocket ingress timeouts.
+5. Validate both application kustomizations and push the change to `magnus/deploy`.
+6. Let Flux roll out btcprice and verify:
+   - both btcprice pods become Ready;
+   - `/api/ready` succeeds;
+   - chart/API requests return current data;
+   - `/ws/price` and `/ws/strike` reconnect and continue updating;
+   - `https://btcprice.k8s.pmagnus.cloud` and `https://btc.pmagnus.cloud` work.
+
+After successful validation, re-enable btcdata image automation.
+
+## Phase 8 — failover and acceptance tests
+
+1. Delete the btcdata pod and verify Kubernetes creates one replacement, never a steady state of two Ready replicas.
+2. Cordon the node currently hosting btcdata, perform a controlled pod deletion, and verify it starts on the other node. Uncordon afterward.
+3. Confirm btcprice becomes temporarily unready during failover and recovers without manual changes.
+4. Verify WebSocket clients reconnect and fresh price data resumes.
+5. Perform an automated image update and confirm `Recreate` stops the old pod before starting the new pod.
+6. Review logs and database data for duplicate Binance/Kraken/Strike writes.
+7. Observe CPU, memory, restart count, startup time, and failover duration; tune resources and probes.
+8. Keep the stopped Compose files and image for a defined rollback window, at least 24 hours.
+
+**Acceptance criteria:**
+
+- Flux reports all resources Ready.
+- Exactly one btcdata pod runs across the two eligible nodes.
+- btcprice uses only Kubernetes DNS for btcdata.
+- HTTP and WebSocket functions work through both production hostnames.
+- A source push can reach production through Gitea, Harbor, and Flux without SSH deployment or manual `kubectl apply`.
+- The legacy Compose btcdata remains stopped.
+
+## Rollback
+
+Rollback must also preserve the single-writer rule:
+
+1. Change the Kubernetes btcdata Deployment to `replicas: 0` in `magnus/deploy`, commit, and wait for Flux.
+2. Verify no btcdata pod or process remains in Kubernetes.
+3. Restore btcprice's selectorless Service and EndpointSlice pointing to citadel, and restore `BTCDATA_URL=http://btcdata:4102` through GitOps.
+4. Restart the recorded btcdata Compose service on citadel.
+5. Verify health, HTTP/WebSocket data, and database write cadence.
+6. Keep the failed Kubernetes image and logs for diagnosis; do not rewrite Git history or delete Harbor evidence.
+
+For an application-only rollback, pause image automation, commit the previous approved immutable image tag to `magnus/deploy`, and let Flux perform the `Recreate` rollout.
+
+## Follow-up improvements
+
+1. Add PostgreSQL advisory-lock leader election, then consider two btcdata pods with one writer leader and one standby/API replica.
+2. Move PostgreSQL to a highly available service reachable independently of citadel.
+3. Add SOPS/age or an external secret operator.
+4. Add dependency-aware readiness and metrics/alerts for stale prices, failed pollers, database errors, and leader status.
+5. Update this repository's `README.md` and `CLAUDE.md`, btcprice documentation, and system infrastructure documentation after the rollback window closes.
+6. Retire `deploy-citadel.sh` and legacy Compose deployment files only after rollback is no longer required.

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
@@ -44,10 +46,10 @@
                       com.lambdaisland/classpath        {:mvn/version "0.4.44"}
                       ring/ring-devel                   {:mvn/version "1.15.3"}}}
 
-  :test {:extra-paths ["env/test/resources"]
+  :test {:extra-paths ["test/clj" "env/test/resources"]
          :extra-deps  {io.github.cognitect-labs/test-runner {:git/tag "v0.5.1" :git/sha "dfb30dd"}
                        peridot/peridot                      {:mvn/version "0.5.4"}
-                       clj-commons/byte-streams              {:mvn/version "0.3.4"}}
+                       org.clj-commons/byte-streams          {:mvn/version "0.3.4"}}
          :main-opts   ["-m" "cognitect.test-runner"]
          :exec-fn     cognitect.test-runner.api/test}
 

docker-compose.yml

@@ -1,12 +1,30 @@
+networks:
+  btc-network:
+    name: btc-network
+    external: true
+
 services:
   btcdata:
     build: .
     restart: unless-stopped
     ports:
-      - "4101:4101"
+      - "${DOCKER_BTCDATA_PORT}:${DOCKER_BTCDATA_PORT}"
+    depends_on:
+      - frankfurter
+    networks:
+      - btc-network
     environment:
-      BTCDATA_PORT: "4101"
-      JDBC_URL: "jdbc:postgresql://postgres:5432/btcprod?user=postgres&password=ratata,123"
-      CORS_ORIGIN: "http://localhost:4041"
+      BTCDATA_PORT: "${DOCKER_BTCDATA_PORT}"
+      JDBC_URL: "${DOCKER_JDBC_URL}"
+      CORS_ORIGIN: "${DOCKER_CORS_ORIGIN}"
+      FRANKFURTER_URL: "${DOCKER_FRANKFURTER_URL}"
+      STRIKE_API_KEY: "${STRIKE_API_KEY}"
+      KRAKEN_ENABLED: "true"
     extra_hosts:
       - "postgres:host-gateway"
+
+  frankfurter:
+    image: lineofflight/frankfurter
+    restart: unless-stopped
+    networks:
+      - btc-network

env.example

@@ -0,0 +1,31 @@
+# Copy to .env and fill in. .env is gitignored — never commit real credentials.
+
+# --- Postgres (only used if you run a Postgres container from this compose file) ---
+POSTGRES_USER=postgres
+POSTGRES_PASSWORD=changeme
+POSTGRES_DB=btcprod
+PGPORT=5432
+
+# --- Local dev (make run) ---
+# The only required variable. No default in system.edn — the app won't start without it.
+JDBC_URL=jdbc:postgresql://localhost:5432/btcprice?user=postgres&password=changeme
+BTCDATA_PORT=4100
+CORS_ORIGIN=http://localhost:4000
+FRANKFURTER_URL=https://api.frankfurter.dev
+
+# Pollers. Leave false when developing against a shared database so you don't
+# write duplicate rows alongside the deployed instance.
+BINANCE_ENABLED=false
+STRIKE_ENABLED=false
+KRAKEN_ENABLED=false
+
+# --- Docker (docker compose up -d) ---
+# Runs alongside local dev on its own port. Points at the Postgres on citadel
+# (192.168.50.195) directly, so it works from any host that can reach that LAN.
+DOCKER_BTCDATA_PORT=4102
+DOCKER_JDBC_URL=jdbc:postgresql://192.168.50.195:5432/btcprod?user=postgres&password=changeme
+DOCKER_CORS_ORIGIN=http://localhost:4041
+DOCKER_FRANKFURTER_URL=http://frankfurter:8080
+
+# --- Strike ---
+STRIKE_API_KEY=

resources/migrations/20260306000000-create-strike-price.down.sql

@@ -0,0 +1 @@
+DROP TABLE strike_price;

resources/migrations/20260306000000-create-strike-price.up.sql

@@ -0,0 +1,5 @@
+CREATE TABLE strike_price (
+  id          BIGSERIAL PRIMARY KEY,
+  rates       JSONB       NOT NULL,
+  recorded_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);

resources/migrations/20260306100000-add-strike-quote-sats.down.sql

@@ -0,0 +1 @@
+ALTER TABLE strike_price DROP COLUMN quote_sats;

resources/migrations/20260306100000-add-strike-quote-sats.up.sql

@@ -0,0 +1 @@
+ALTER TABLE strike_price ADD COLUMN quote_sats BIGINT;

resources/migrations/20260306200000-create-currencies.down.sql

@@ -0,0 +1 @@
+DROP TABLE currencies;

resources/migrations/20260306200000-create-currencies.up.sql

@@ -0,0 +1,7 @@
+CREATE TABLE currencies (
+  rate_date DATE PRIMARY KEY,
+  eur NUMERIC(12,6) NOT NULL,
+  dkk NUMERIC(12,6) NOT NULL,
+  eur_dkk NUMERIC(12,6) NOT NULL,
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);

resources/migrations/20260307000000-create-wallets.down.sql

@@ -0,0 +1 @@
+DROP TABLE wallets;

resources/migrations/20260307000000-create-wallets.up.sql

@@ -0,0 +1,9 @@
+CREATE TABLE wallets (
+  id         UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+  name       TEXT NOT NULL UNIQUE,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+--;;
+INSERT INTO wallets (name) VALUES
+  ('Crypto.Com'), ('Nexo'), ('CoinCorner'), ('Strike'),
+  ('Cold'), ('Coldcard'), ('Nunchuk Multi Sig');

resources/migrations/20260307100000-create-events.down.sql

@@ -0,0 +1 @@
+DROP TABLE events;

resources/migrations/20260307100000-create-events.up.sql

@@ -0,0 +1,17 @@
+CREATE TABLE events (
+  id             BIGSERIAL PRIMARY KEY,
+  event_type     TEXT NOT NULL,
+  occurred_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  recorded_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  sats           BIGINT,
+  fee_sats       BIGINT,
+  fiat_amount    NUMERIC(18,2),
+  fiat_currency  TEXT,
+  from_wallet_id UUID REFERENCES wallets(id),
+  to_wallet_id   UUID REFERENCES wallets(id),
+  note           TEXT
+);
+--;;
+CREATE INDEX idx_events_type ON events (event_type);
+--;;
+CREATE INDEX idx_events_occurred ON events (occurred_at);

resources/migrations/20260308000000-create-deposits.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS deposits;

resources/migrations/20260308000000-create-deposits.up.sql

@@ -0,0 +1,12 @@
+CREATE TABLE deposits (
+  id          BIGSERIAL PRIMARY KEY,
+  event_id    BIGINT NOT NULL REFERENCES events(id) UNIQUE,
+  occurred_at TIMESTAMPTZ NOT NULL,
+  exchange    TEXT NOT NULL,
+  fiat_amount NUMERIC(18,2) NOT NULL,
+  fiat_currency TEXT NOT NULL,
+  amount_eur  NUMERIC(18,2),
+  amount_dkk  NUMERIC(18,2),
+  amount_usd  NUMERIC(18,2),
+  note        TEXT
+);

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/migrations/20260313000000-create-kraken-minute.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS kraken_minute;

resources/migrations/20260313000000-create-kraken-minute.up.sql

@@ -0,0 +1,11 @@
+CREATE TABLE kraken_minute (
+  ts          BIGINT         NOT NULL PRIMARY KEY,
+  open        NUMERIC(18,8)  NOT NULL,
+  high        NUMERIC(18,8)  NOT NULL,
+  low         NUMERIC(18,8)  NOT NULL,
+  close       NUMERIC(18,8)  NOT NULL,
+  vwap        NUMERIC(18,8)  NOT NULL,
+  volume      NUMERIC(24,8)  NOT NULL,
+  trade_count INTEGER        NOT NULL,
+  created_at  TIMESTAMPTZ    NOT NULL DEFAULT NOW()
+);

resources/migrations/20260316000000-create-buys.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS buys;

resources/migrations/20260316000000-create-buys.up.sql

@@ -0,0 +1,7 @@
+CREATE TABLE buys (
+  id          BIGSERIAL PRIMARY KEY,
+  event_id    BIGINT NOT NULL REFERENCES events(id) UNIQUE,
+  occurred_at DATE NOT NULL,
+  wallet      TEXT NOT NULL,
+  amount_eur  NUMERIC(18,2) NOT NULL
+);

resources/migrations/20260316100000-buys-add-sats.down.sql

@@ -0,0 +1,3 @@
+ALTER TABLE buys DROP COLUMN fee_sats;
+--;;
+ALTER TABLE buys DROP COLUMN sats;

resources/migrations/20260316100000-buys-add-sats.up.sql

@@ -0,0 +1,3 @@
+ALTER TABLE buys ADD COLUMN sats BIGINT NOT NULL DEFAULT 0;
+--;;
+ALTER TABLE buys ADD COLUMN fee_sats BIGINT;

resources/migrations/20260317000000-create-kraken-day-history.down.sql

@@ -0,0 +1 @@
+DROP TABLE kraken_day_history;

resources/migrations/20260317000000-create-kraken-day-history.up.sql

@@ -0,0 +1,11 @@
+CREATE TABLE kraken_day_history (
+  ts          BIGINT         NOT NULL PRIMARY KEY,
+  open        NUMERIC(18,8)  NOT NULL,
+  high        NUMERIC(18,8)  NOT NULL,
+  low         NUMERIC(18,8)  NOT NULL,
+  close       NUMERIC(18,8)  NOT NULL,
+  vwap        NUMERIC(18,8)  NOT NULL,
+  volume      NUMERIC(24,8)  NOT NULL,
+  trade_count INTEGER        NOT NULL,
+  created_at  TIMESTAMPTZ    NOT NULL DEFAULT NOW()
+);

resources/queries.sql

@@ -1,5 +1,9 @@
 -- queries for btcdata
 
+-- :name check-database :? :1
+-- :doc Verify that PostgreSQL accepts queries
+SELECT 1 AS healthy
+
 -- :name insert-binance-price! :! :n
 -- :doc Insert a Binance BTC price
 INSERT INTO binance_price (price) VALUES (:price)
@@ -42,3 +46,198 @@ SET open = EXCLUDED.open,
 -- :name get-latest-kraken-day :? :1
 -- :doc Get the most recent Kraken daily candle by timestamp
 SELECT ts FROM kraken_day ORDER BY ts DESC LIMIT 1
+
+-- :name get-kraken-day-by-ts :? :1
+SELECT ts, open, high, low, close, vwap, volume, trade_count
+FROM kraken_day WHERE ts = :ts
+
+-- :name get-kraken-day-history-by-ts :? :1
+SELECT ts, open, high, low, close, vwap, volume, trade_count
+FROM kraken_day_history WHERE ts = :ts
+
+-- :name upsert-kraken-day-history! :! :n
+INSERT INTO kraken_day_history (ts, open, high, low, close, vwap, volume, trade_count)
+VALUES (:ts, :open, :high, :low, :close, :vwap, :volume, :trade-count)
+ON CONFLICT (ts) DO UPDATE
+SET open = EXCLUDED.open,
+    high = EXCLUDED.high,
+    low = EXCLUDED.low,
+    close = EXCLUDED.close,
+    vwap = EXCLUDED.vwap,
+    volume = EXCLUDED.volume,
+    trade_count = EXCLUDED.trade_count
+
+-- :name insert-strike-price! :! :n
+-- :doc Insert a Strike ticker snapshot
+INSERT INTO strike_price (rates, quote_sats) VALUES (:rates, :quote-sats)
+
+-- :name get-latest-strike-price :? :1
+-- :doc Get the most recent Strike ticker snapshot
+SELECT rates, recorded_at FROM strike_price ORDER BY id DESC LIMIT 1
+
+-- :name upsert-currency-rates! :! :n
+-- :doc Upsert daily currency rates from Frankfurter
+INSERT INTO currencies (rate_date, eur, dkk, eur_dkk, updated_at)
+VALUES (:rate-date, :eur, :dkk, :eur-dkk, NOW())
+ON CONFLICT (rate_date) DO UPDATE
+SET eur = EXCLUDED.eur, dkk = EXCLUDED.dkk, eur_dkk = EXCLUDED.eur_dkk, updated_at = NOW()
+
+-- :name get-latest-currency-rates :? :1
+-- :doc Get the most recent currency rates
+SELECT rate_date, eur, dkk, eur_dkk, updated_at FROM currencies ORDER BY rate_date DESC LIMIT 1
+
+-- :name get-currency-rates-by-date :? :1
+-- :doc Get currency rates for a specific date
+SELECT rate_date, eur, dkk, eur_dkk, updated_at FROM currencies WHERE rate_date = :rate-date
+
+-- Wallets -------------------------------------------------------------------
+
+-- :name insert-wallet! :! :n
+-- :doc Insert a new wallet
+INSERT INTO wallets (name, wallet_type) VALUES (:name, :wallet-type)
+
+-- :name get-all-wallets :? :*
+-- :doc Get all wallets
+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, wallet_type, created_at FROM wallets WHERE id = :id
+
+-- Events --------------------------------------------------------------------
+
+-- :name insert-event! :? :1
+-- :doc Insert a new event, returning its id
+INSERT INTO events (event_type, occurred_at, sats, fee_sats, fiat_amount, fiat_currency, from_wallet_id, to_wallet_id, note)
+VALUES (:event-type, :occurred-at, :sats, :fee-sats, :fiat-amount, :fiat-currency, :from-wallet-id, :to-wallet-id, :note)
+RETURNING id
+
+-- :name get-all-events :? :*
+-- :doc Get all events with wallet names
+SELECT e.id, e.event_type, e.occurred_at, e.recorded_at,
+       e.sats, e.fee_sats, e.fiat_amount, e.fiat_currency,
+       e.from_wallet_id, fw.name AS from_wallet_name,
+       e.to_wallet_id, tw.name AS to_wallet_name,
+       e.note
+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, e.id DESC
+
+-- :name get-events-by-type :? :*
+-- :doc Get events filtered by type
+SELECT e.id, e.event_type, e.occurred_at, e.recorded_at,
+       e.sats, e.fee_sats, e.fiat_amount, e.fiat_currency,
+       e.from_wallet_id, fw.name AS from_wallet_name,
+       e.to_wallet_id, tw.name AS to_wallet_name,
+       e.note
+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, e.id DESC
+
+-- :name get-events-by-wallet :? :*
+-- :doc Get events involving a specific wallet
+SELECT e.id, e.event_type, e.occurred_at, e.recorded_at,
+       e.sats, e.fee_sats, e.fiat_amount, e.fiat_currency,
+       e.from_wallet_id, fw.name AS from_wallet_name,
+       e.to_wallet_id, tw.name AS to_wallet_name,
+       e.note
+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, 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
+INSERT INTO deposits (event_id, occurred_at, exchange, fiat_amount, fiat_currency, amount_eur, amount_dkk, amount_usd, note)
+VALUES (:event-id, :occurred-at, :exchange, :fiat-amount, :fiat-currency, :amount-eur, :amount-dkk, :amount-usd, :note)
+
+-- :name get-all-deposits :? :*
+-- :doc Get all projected deposits ordered by date descending
+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, id DESC
+
+-- :name truncate-buys! :! :n
+-- :doc Delete all rows from the buys read table
+TRUNCATE buys
+
+-- :name get-buy-events :? :*
+-- :doc Get exchange_to_wallet events with wallet name, oldest first
+SELECT e.id, e.occurred_at, e.sats, e.fee_sats, e.fiat_amount, e.fiat_currency,
+       e.to_wallet_id, tw.name AS wallet
+FROM events e
+LEFT JOIN wallets tw ON tw.id = e.to_wallet_id
+WHERE e.event_type = 'exchange_to_wallet'
+ORDER BY e.occurred_at ASC, e.id ASC
+
+-- :name insert-buy! :! :n
+-- :doc Insert a projected buy row
+INSERT INTO buys (event_id, occurred_at, wallet, sats, fee_sats, amount_eur)
+VALUES (:event-id, :occurred-at, :wallet, :sats, :fee-sats, :amount-eur)
+
+-- :name get-all-buys :? :*
+-- :doc Get all projected buys ordered by date descending
+SELECT id, event_id, occurred_at, wallet, sats, fee_sats, amount_eur
+FROM buys
+ORDER BY occurred_at DESC, id DESC
+
+-- :name get-kraken-hour-latest-24 :? :*
+-- :doc Get the 24 most recent Kraken hourly candles
+SELECT ts, open, high, low, close, vwap, volume, trade_count, created_at
+FROM kraken_hour ORDER BY ts DESC LIMIT 24
+
+-- :name get-kraken-minute-latest-60 :? :*
+-- :doc Get the 60 most recent Kraken minute candles
+SELECT ts, open, high, low, close, vwap, volume, trade_count, created_at
+FROM kraken_minute ORDER BY ts DESC LIMIT 60
+
+-- :name upsert-kraken-minute! :! :n
+-- :doc Upsert a Kraken minute OHLC candle
+INSERT INTO kraken_minute (ts, open, high, low, close, vwap, volume, trade_count)
+VALUES (:ts, :open, :high, :low, :close, :vwap, :volume, :trade-count)
+ON CONFLICT (ts) DO UPDATE
+SET open = EXCLUDED.open,
+    high = EXCLUDED.high,
+    low  = EXCLUDED.low,
+    close = EXCLUDED.close,
+    vwap  = EXCLUDED.vwap,
+    volume = EXCLUDED.volume,
+    trade_count = EXCLUDED.trade_count
+
+-- :name get-latest-kraken-minute :? :1
+-- :doc Get the most recent Kraken minute candle timestamp
+SELECT ts FROM kraken_minute ORDER BY ts DESC LIMIT 1
+
+-- :name get-wallet-balances :? :*
+-- :doc Compute sats balance per wallet from events
+SELECT w.id, w.name,
+  COALESCE(SUM(CASE WHEN e.to_wallet_id = w.id THEN e.sats ELSE 0 END), 0)
+  - COALESCE(SUM(CASE WHEN e.from_wallet_id = w.id THEN e.sats ELSE 0 END), 0)
+  - COALESCE(SUM(CASE WHEN e.from_wallet_id = w.id THEN e.fee_sats ELSE 0 END), 0)
+  AS balance_sats
+FROM wallets w
+LEFT JOIN events e ON e.from_wallet_id = w.id OR e.to_wallet_id = w.id
+GROUP BY w.id, w.name
+ORDER BY w.name

resources/system.edn

@@ -43,17 +43,41 @@
   :db               {:datasource #ig/ref :db.sql/connection}
   :migrate-on-init? true}
 
+ :writer/query-fn
+ {:conn #ig/ref :db.sql/connection
+  :query-fn #ig/ref :db.sql/query-fn
+  :migrations #ig/ref :db.sql/migrations}
+
  :reitit.routes/api
  {:base-path    "/api"
-  :query-fn  #ig/ref :db.sql/query-fn
-  :binance   #ig/ref :ws/binance}
+  :query-fn     #ig/ref :writer/query-fn
+  :binance      #ig/ref :ws/binance
+  :frankfurter  #ig/ref :frankfurter/rates
+  :strike       #ig/ref :strike/ticker}
 
  :ws/binance
- {:query-fn #ig/ref :db.sql/query-fn
-  :uri      "wss://stream.binance.com:9443/ws/btcusdt@trade"}
+ {:query-fn #ig/ref :writer/query-fn
+  :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}
+ {:query-fn #ig/ref :writer/query-fn
+  :enabled? #or [#env KRAKEN_ENABLED "true"]}
 
  :kraken/ohlc-day
- {:query-fn #ig/ref :db.sql/query-fn}}
+ {:query-fn #ig/ref :writer/query-fn
+  :enabled? #or [#env KRAKEN_ENABLED "true"]}
+
+ :kraken/ohlc-minute
+ {:query-fn #ig/ref :writer/query-fn
+  :enabled? #or [#env KRAKEN_ENABLED "true"]}
+
+ :frankfurter/rates
+ {:url      #or [#env FRANKFURTER_URL "http://localhost:8080"]
+  :query-fn #ig/ref :writer/query-fn}
+
+ :strike/ticker
+ {:query-fn        #ig/ref :writer/query-fn
+  :api-key         #env STRIKE_API_KEY
+  :sats-eur-amount #or [#env SATS_EUR_AMOUNT "55"]
+  :enabled?        #or [#env STRIKE_ENABLED "true"]}}

scripts/fetch-year.sh

@@ -0,0 +1,154 @@
+#!/usr/bin/env bash
+# Usage: ./scripts/fetch-year.sh YEAR [--reverse]
+# Fetches sequentially; prints JSON responses to stdout and errors to stderr.
+set -euo pipefail
+
+reverse=false
+year=
+for arg in "$@"; do
+  case "$arg" in
+    --reverse) reverse=true ;;
+    *)
+      if [[ -n $year ]]; then
+        echo "Usage: $0 YEAR [--reverse]" >&2
+        exit 1
+      fi
+      year=$arg
+      ;;
+  esac
+done
+if [[ -z $year ]]; then
+  echo "Usage: $0 YEAR [--reverse]" >&2
+  exit 1
+fi
+if [[ ! $year =~ ^[0-9]{4}$ || $year == 0000 ]]; then
+  echo "YEAR must be a four-digit year from 0001 to 9999." >&2
+  exit 1
+fi
+
+year_number=$((10#$year))
+days_in_month=(31 28 31 30 31 30 31 31 30 31 30 31)
+if (( year_number % 400 == 0 || (year_number % 4 == 0 && year_number % 100 != 0) )); then
+  days_in_month[1]=29
+fi
+
+# Reuse a reachable service, or keep a port-forward alive while fetching.
+service_ready() {
+  curl --fail --silent --output /dev/null --connect-timeout 1 --max-time 2 \
+    'http://localhost:4101/api/health'
+}
+
+forward_pid=
+cleanup() {
+  if [[ -n $forward_pid ]]; then
+    kill "$forward_pid" 2>/dev/null || true
+    wait "$forward_pid" 2>/dev/null || true
+    forward_pid=
+  fi
+}
+trap cleanup EXIT
+trap 'exit 130' INT
+trap 'exit 143' TERM
+
+wait_for_pod() {
+  local selector
+  # Use the service's selector rather than assuming a particular pod label.
+  if ! selector=$(kubectl -n sejs get svc btcdata --request-timeout=10s \
+    -o go-template='{{range $key, $value := .spec.selector}}{{$key}}={{$value}},{{end}}'); then
+    return 1
+  fi
+  selector=${selector%,}
+  if [[ -z $selector ]]; then
+    echo "Service sejs/btcdata has no pod selector." >&2
+    return 1
+  fi
+  echo "Waiting up to 5 minutes for btcdata pods to become Ready..." >&2
+  kubectl -n sejs wait --for=condition=Ready pod --selector="$selector" \
+    --timeout=300s --request-timeout=310s >&2
+}
+
+start_forward() {
+  local attempt
+  cleanup
+  if ! command -v kubectl >/dev/null 2>&1; then
+    echo "Service unavailable on localhost:4101 and kubectl is not installed." >&2
+    return 1
+  fi
+  if ! wait_for_pod; then
+    echo "btcdata pod readiness check failed; not opening a tunnel." >&2
+    return 1
+  fi
+  echo "Starting kubectl port-forward for sejs/btcdata..." >&2
+  kubectl -n sejs port-forward svc/btcdata 4101:4101 >&2 &
+  forward_pid=$!
+
+  for (( attempt = 0; attempt < 30; attempt++ )); do
+    if ! kill -0 "$forward_pid" 2>/dev/null; then
+      echo "Port-forward exited before the service became ready." >&2
+      cleanup
+      return 1
+    fi
+    if service_ready; then
+      return 0
+    fi
+    sleep 1
+  done
+  echo "Timed out waiting for the service on localhost:4101." >&2
+  cleanup
+  return 1
+}
+
+ensure_service() {
+  local retry
+  if service_ready; then
+    return 0
+  fi
+  for (( retry = 1; retry <= 3; retry++ )); do
+    echo "Recovering port-forward (attempt $retry/3)..." >&2
+    if start_forward; then
+      return 0
+    fi
+    if (( retry < 3 )); then
+      sleep 2
+    fi
+  done
+  echo "Unable to restore the service on localhost:4101." >&2
+  return 1
+}
+
+fetch_date() {
+  curl --fail --silent --show-error --connect-timeout 5 -X GET \
+    "http://localhost:4101/api/price/day/$1" \
+    -H "accept: application/json" --write-out '\n'
+}
+
+ensure_service || exit 1
+
+# Generate dates without relying on platform-specific date command options.
+status=0
+for (( month_index = 0; month_index < 12; month_index++ )); do
+  month=$((month_index + 1))
+  if [[ $reverse == true ]]; then
+    month=$((12 - month_index))
+  fi
+  days=${days_in_month[month - 1]}
+  for (( day_index = 0; day_index < days; day_index++ )); do
+    day=$((day_index + 1))
+    if [[ $reverse == true ]]; then
+      day=$((days - day_index))
+    fi
+    printf -v date '%s-%02d-%02d' "$year" "$month" "$day"
+    if ! fetch_date "$date"; then
+      if ! service_ready; then
+        ensure_service || exit 1
+        echo "Retrying $date after tunnel recovery..." >&2
+        if fetch_date "$date"; then
+          continue
+        fi
+      fi
+      echo "Failed to fetch $date" >&2
+      status=1
+    fi
+  done
+done
+exit "$status"

scripts/test-fencing.sh

@@ -0,0 +1,32 @@
+#!/usr/bin/env bash
+# Disposable isolated PostgreSQL; no production credentials or published ports.
+set -euo pipefail
+name="btcdata-test-${GITHUB_RUN_ID:-$$}-${GITHUB_RUN_ATTEMPT:-1}"
+cleanup() {
+  docker rm -f "${name}-runner" "$name" >/dev/null 2>&1 || true
+  docker network rm "$name" >/dev/null 2>&1 || true
+}
+trap cleanup EXIT
+docker network create "$name" >/dev/null
+docker run -d --name "$name" --network "$name" \
+  -e POSTGRES_HOST_AUTH_METHOD=trust postgres:16-alpine >/dev/null
+ready=false
+for _ in $(seq 1 60); do
+  if docker exec "$name" pg_isready -U postgres >/dev/null 2>&1; then
+    ready=true
+    break
+  fi
+  sleep 1
+done
+if [ "$ready" != true ]; then
+  echo 'Disposable PostgreSQL did not become ready' >&2
+  exit 1
+fi
+# The CI workspace is inside the runner container, not on the Docker host.
+# Copy only test inputs; never mount a nonexistent host workspace or copy .env/.git.
+docker create --name "${name}-runner" --network "$name" -w /app \
+  -e "BTCDATA_TEST_JDBC_URL=jdbc:postgresql://$name:5432/postgres?user=postgres" \
+  clojure:temurin-21-tools-deps-alpine clj -M:test >/dev/null
+COPYFILE_DISABLE=1 tar --no-xattrs -cf - deps.edn src resources test env/test | docker cp - "${name}-runner:/app"
+docker start --attach "${name}-runner"
+exit "$(docker inspect --format '{{.State.ExitCode}}' "${name}-runner")"

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]]
 
@@ -15,10 +16,29 @@
    [pmagnus.btcdata.web.routes.api]
    ;; WebSocket clients
    [pmagnus.btcdata.ws.binance]
+   [pmagnus.btcdata.writer]
    ;; Pollers
    [pmagnus.btcdata.kraken.ohlc]
-   [pmagnus.btcdata.kraken.ohlc-daily])
-  (:gen-class))
+   [pmagnus.btcdata.kraken.ohlc-daily]
+   [pmagnus.btcdata.kraken.ohlc-minute]
+   [pmagnus.btcdata.frankfurter.rates]
+   [pmagnus.btcdata.strike.ticker])
+  (: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/frankfurter/rates.clj

@@ -0,0 +1,114 @@
+(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 LocalDate]))
+
+(defn- fetch-rates*
+  "GET a Frankfurter endpoint. Returns parsed body."
+  [^HttpClient client url]
+  (let [request (-> (HttpRequest/newBuilder)
+                    (.uri (URI. url))
+                    (.header "Accept" "application/json")
+                    (.GET)
+                    (.build))
+        resp    (.send client request (HttpResponse$BodyHandlers/ofString))]
+    (json/read-str (.body resp) :key-fn keyword)))
+
+(defn- fetch-rates
+  "GET /latest?from=USD&to=EUR,DKK from Frankfurter. Returns {:EUR x :DKK y}."
+  [^HttpClient client base-url]
+  (:rates (fetch-rates* client (str base-url "/v1/latest?from=USD&to=EUR,DKK"))))
+
+(defn fetch-and-persist-for-date!
+  "Fetch rates for a specific date from Frankfurter, persist, and return the rate map."
+  [^HttpClient client base-url query-fn ^LocalDate date]
+  (let [url   (str base-url "/v1/" date "?from=USD&to=EUR,DKK")
+        rates (:rates (fetch-rates* client url))]
+    (when (and (:EUR rates) (:DKK rates))
+      (let [eur     (bigdec (str (:EUR rates)))
+            dkk     (bigdec (str (:DKK rates)))
+            eur-dkk (.divide dkk eur 6 java.math.RoundingMode/HALF_UP)]
+        (query-fn :upsert-currency-rates!
+                  {:rate-date date :eur eur :dkk dkk :eur-dkk eur-dkk})
+        (log/info "Frankfurter historical rates for" (str date) "— EUR:" eur "DKK:" dkk)
+        {:rate_date  date :eur eur :dkk dkk :eur_dkk eur-dkk}))))
+
+(defn- poll!
+  "Fetch rates, update the atom, and persist to DB."
+  [client base-url rates-atom query-fn]
+  (let [rates (fetch-rates client base-url)]
+    (when (and (:EUR rates) (:DKK rates))
+      (let [eur (bigdec (str (:EUR rates)))
+            dkk (bigdec (str (:DKK rates)))
+            eur-dkk (.divide dkk eur 6 java.math.RoundingMode/HALF_UP)]
+        (reset! rates-atom
+                {:eur        eur
+                 :dkk        dkk
+                 :updated-at (Instant/now)})
+        (query-fn :upsert-currency-rates!
+                  {:rate-date (LocalDate/now)
+                   :eur       eur
+                   :dkk       dkk
+                   :eur-dkk   eur-dkk})
+        (log/info "Frankfurter rates — EUR:" eur "DKK:" dkk "EUR/DKK:" eur-dkk)))))
+
+(defn- start-poll-loop!
+  "Fetch immediately, then poll every 60 minutes."
+  [client base-url rates-atom running? query-fn]
+  (future
+    (loop [retries 12]
+      (let [ok? (try
+                  (poll! client base-url rates-atom query-fn)
+                  true
+                  (catch Exception e
+                    (log/error e (str "Frankfurter fetch failed, retrying in 10s (" retries " left)"))
+                    false))]
+        (when (and (not ok?) @running? (pos? retries))
+          (Thread/sleep 10000)
+          (recur (dec retries)))))
+    (while @running?
+      (Thread/sleep 3600000)
+      (when @running?
+        (try
+          (poll! client base-url rates-atom query-fn)
+          (catch Exception e
+            (log/error e "Frankfurter poll failed")))))))
+
+(defn- seed-from-db!
+  "Load latest rates from DB into the atom so we have data immediately."
+  [rates-atom query-fn]
+  (try
+    (when-let [row (query-fn :get-latest-currency-rates {})]
+      (reset! rates-atom
+              {:eur        (:eur row)
+               :dkk        (:dkk row)
+               :updated-at (:updated_at row)})
+      (log/info "Seeded Frankfurter rates from DB — EUR:" (:eur row) "DKK:" (:dkk row)))
+    (catch Exception e
+      (log/warn e "Could not seed rates from DB"))))
+
+(defmethod ig/init-key :frankfurter/rates
+  [_ {:keys [url query-fn]}]
+  (log/info "Starting Frankfurter rates poller:" url)
+  (let [client   (HttpClient/newHttpClient)
+        rates    (atom nil)
+        running? (atom true)]
+    (seed-from-db! rates query-fn)
+    (let [fut (start-poll-loop! client url rates running? query-fn)]
+      {:rates    rates
+       :running? running?
+       :future   fut
+       :client   client
+       :url      url
+       :query-fn query-fn})))
+
+(defmethod ig/halt-key! :frankfurter/rates
+  [_ {:keys [running? future]}]
+  (log/info "Stopping Frankfurter rates poller")
+  (reset! running? false)
+  (future-cancel future))

src/clj/pmagnus/btcdata/kraken/day.clj

@@ -0,0 +1,111 @@
+(ns pmagnus.btcdata.kraken.day
+  "On-demand completed UTC candles. Database failures are not upstream failures."
+  (:require [clojure.data.json :as json])
+  (:import [java.net URI]
+           [java.net.http HttpClient HttpRequest HttpResponse$BodyHandlers]
+           [java.time Duration LocalDate ZoneOffset]
+           [java.math RoundingMode]))
+
+(defonce ^:private client
+  (delay (-> (HttpClient/newBuilder)
+             (.connectTimeout (Duration/ofSeconds 15))
+             (.build))))
+
+(defn- upstream-error [message]
+  (ex-info message {:type ::upstream}))
+
+(defn request!
+  "Fetch a Kraken public result, with a per-page timeout (not an overall deadline)."
+  [endpoint]
+  (try
+    (let [request (-> (HttpRequest/newBuilder)
+                      (.uri (URI/create (str "https://api.kraken.com/0/public/" endpoint)))
+                      (.timeout (Duration/ofSeconds 60))
+                      (.header "Accept" "application/json")
+                      (.GET) (.build))
+          response (.send ^HttpClient @client request (HttpResponse$BodyHandlers/ofString))]
+      (when-not (= 200 (.statusCode response))
+        (throw (upstream-error "Kraken returned an HTTP error")))
+      (let [body (json/read-str (.body response) :key-fn keyword :bigdec true)]
+        (when (or (seq (:error body)) (not (map? (:result body))))
+          (throw (upstream-error "Invalid or unsuccessful Kraken response")))
+        (:result body)))
+    (catch InterruptedException e
+      (.interrupt (Thread/currentThread))
+      (throw (ex-info "Kraken request interrupted" {:type ::upstream} e)))
+    (catch Exception e
+      (throw (ex-info "Failed to fetch Kraken data" {:type ::upstream} e)))))
+
+(defn- pair-rows [result]
+  (let [rows (:XXBTZUSD result)]
+    (when-not (vector? rows)
+      (throw (upstream-error "Missing BTC/USD data in Kraken response")))
+    rows))
+
+(defn- candle [[ts open high low close vwap volume trade-count]]
+  {:ts (long ts) :open (bigdec open) :high (bigdec high)
+   :low (bigdec low) :close (bigdec close) :vwap (bigdec vwap)
+   :volume (bigdec volume) :trade-count (long trade-count)})
+
+(defn- add-trade [acc [price volume]]
+  (let [p (bigdec price) v (bigdec volume)]
+    (-> (or acc {:open p :high p :low p :volume 0M :value 0M :trade-count 0})
+        (assoc :close p)
+        (update :high max p)
+        (update :low min p)
+        (update :volume + v)
+        (update :value + (* p v))
+        (update :trade-count inc))))
+
+(defn pause! [] (Thread/sleep 1000))
+
+(defn- historical-candle! [start]
+  (let [end (+ start 86400)]
+    (loop [cursor (str (*' (max 0 start) 1000000000)) acc nil]
+      (let [result (request! (str "Trades?pair=XBTUSD&since=" cursor))
+            rows (pair-rows result)
+            acc (reduce add-trade acc
+                        (filter (fn [row]
+                                  (let [ts (bigdec (nth row 2))]
+                                    (and (<= start ts) (< ts end))))
+                                rows))
+            done? (or (empty? rows) (some #(>= (bigdec (nth % 2)) end) rows))]
+        (if done?
+          (when acc
+            (when-not (pos? (:volume acc))
+              (throw (upstream-error "Kraken returned zero daily volume")))
+            (-> acc
+                (assoc :ts start :vwap (.divide ^java.math.BigDecimal (:value acc)
+                                               ^java.math.BigDecimal (:volume acc)
+                                               8 RoundingMode/HALF_UP))
+                (dissoc :value)))
+          (let [next-cursor (str (:last result))]
+            (when-not (> (bigint next-cursor) (bigint cursor))
+              (throw (upstream-error "Kraken trade pagination did not advance")))
+            (pause!)
+            (recur next-cursor acc)))))))
+
+(defn- fetch-candle! [ts]
+  (try
+    (let [rows (pair-rows (request! (str "OHLC?pair=XBTUSD&interval=1440&since=" (max 0 ts))))
+          match (some #(when (= ts (long (first %))) %) rows)]
+      (cond
+        match [:upsert-kraken-day! (candle match)]
+        (or (empty? rows) (< ts (long (ffirst rows))))
+        (when-let [c (historical-candle! ts)] [:upsert-kraken-day-history! c])
+        :else nil))
+    (catch Exception e
+      (throw (ex-info "Failed to obtain Kraken daily candle" {:type ::upstream} e)))))
+
+(defn find-day!
+  "Look up both caches before fetching. Returns a candle or nil. Date must be completed."
+  [query-fn ^LocalDate date]
+  (let [ts (.toEpochSecond (.atStartOfDay date) ZoneOffset/UTC)
+        params {:ts ts}]
+    (or (query-fn :get-kraken-day-by-ts params)
+        (query-fn :get-kraken-day-history-by-ts params)
+        (when-let [[save-query c] (fetch-candle! ts)]
+          (query-fn save-query c)
+          ;; Read stored numeric precision so first and cached responses match.
+          (query-fn (if (= save-query :upsert-kraken-day!)
+                      :get-kraken-day-by-ts :get-kraken-day-history-by-ts) params)))))

src/clj/pmagnus/btcdata/kraken/ohlc.clj

@@ -54,68 +54,53 @@
           :ts))
 
 (defn- poll!
-  "Fetch OHLC data, drop the last (in-progress) candle, save completed ones.
+  "Fetch OHLC data, upsert candles. On initial fetch saves all completed candles
+   (drops last in-progress). On subsequent fetches saves last 5 including in-progress.
    Returns the count of saved candles."
-  [client query-fn since-atom]
+  [client query-fn since-atom initial?]
   (let [result   (fetch-ohlc client @since-atom)
         ;; Kraken returns a map with the pair key and a "last" key
         last-ts  (:last result)
         pair-key (first (remove #{:last} (keys result)))
         raw      (get result pair-key)
-        candles  (map parse-candle (butlast raw))]
+        candles  (map parse-candle (if initial? (butlast raw) (take-last 5 raw)))]
     (when (seq candles)
       (save-candles! query-fn candles)
       (when last-ts
         (reset! since-atom last-ts))
-      (log/info "Fetched" (count candles) "completed Kraken hourly candles"))
+      (log/info "Fetched" (count candles) "Kraken hourly candles"))
     (count candles)))
 
-(defn- ms-until-next-poll
-  "Milliseconds from now until next HH:01:00 UTC."
-  []
-  (let [now    (java.time.ZonedDateTime/now java.time.ZoneOffset/UTC)
-        next   (-> now
-                   (.truncatedTo java.time.temporal.ChronoUnit/HOURS)
-                   (.plusMinutes 1))
-        target (if (.isAfter now next)
-                 (.plusHours next 1)
-                 next)]
-    (.toMillis (java.time.Duration/between now target))))
-
 (defn- start-poll-loop!
-  "Start a background future that polls Kraken aligned to HH:01:00 UTC.
-   When `has-data?` is false, fetches immediately to backfill."
-  [client query-fn since-atom running? has-data?]
+  "Start a background future that polls Kraken every 60 seconds."
+  [client query-fn since-atom running?]
   (future
-    (when-not has-data?
     (try
-        (poll! client query-fn since-atom)
+      (poll! client query-fn since-atom true)
       (catch Exception e
-          (log/error e "Kraken OHLC initial poll failed"))))
+        (log/error e "Kraken OHLC initial poll failed")))
     (while @running?
-      (let [wait (ms-until-next-poll)]
-        (log/info "Next Kraken OHLC poll in" (quot wait 60000) "minutes")
-        (Thread/sleep wait)
+      (Thread/sleep 60000)
       (when @running?
         (try
-            (poll! client query-fn since-atom)
+          (poll! client query-fn since-atom false)
           (catch Exception e
-              (log/error e "Kraken OHLC poll failed"))))))))
+            (log/error e "Kraken OHLC poll failed")))))))
 
 (defmethod ig/init-key :kraken/ohlc
-  [_ {:keys [query-fn]}]
+  [_ {:keys [query-fn enabled?]}]
+  (if (= enabled? "false")
+    (do (log/info "Kraken OHLC poller disabled")
+        {:running? (atom false)})
     (let [client    (HttpClient/newHttpClient)
           db-since  (seed-since-from-db query-fn)
           since     (atom db-since)
           running?  (atom true)
-        has-data? (some? db-since)
-        fut       (start-poll-loop! client query-fn since running? has-data?)]
-    (if has-data?
-      (log/info "Starting Kraken OHLC poller, next fetch in" (quot (ms-until-next-poll) 60000) "minutes")
-      (log/info "Starting Kraken OHLC poller, fetching immediately (no data)"))
+          fut       (start-poll-loop! client query-fn since running?)]
+      (log/info "Starting Kraken OHLC poller (60s interval)")
       {:running? running?
        :future   fut
-     :since    since}))
+       :since    since})))
 
 (defmethod ig/halt-key! :kraken/ohlc
   [_ {:keys [running? future]}]

src/clj/pmagnus/btcdata/kraken/ohlc_daily.clj

@@ -102,7 +102,10 @@
               (log/error e "Kraken daily OHLC poll failed"))))))))
 
 (defmethod ig/init-key :kraken/ohlc-day
-  [_ {:keys [query-fn]}]
+  [_ {:keys [query-fn enabled?]}]
+  (if (= enabled? "false")
+    (do (log/info "Kraken daily OHLC poller disabled")
+        {:running? (atom false)})
     (let [client    (HttpClient/newHttpClient)
           db-since  (seed-since-from-db query-fn)
           since     (atom db-since)
@@ -114,7 +117,7 @@
         (log/info "Starting Kraken daily OHLC poller, fetching immediately (no data)"))
       {:running? running?
        :future   fut
-     :since    since}))
+       :since    since})))
 
 (defmethod ig/halt-key! :kraken/ohlc-day
   [_ {:keys [running? future]}]

src/clj/pmagnus/btcdata/kraken/ohlc_minute.clj

@@ -0,0 +1,114 @@
+(ns pmagnus.btcdata.kraken.ohlc-minute
+  (: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 ZoneOffset]
+   [java.time.format DateTimeFormatter]))
+
+(def ^:private kraken-ohlc-url
+  "https://api.kraken.com/0/public/OHLC?pair=XBTUSD&interval=1")
+
+(defn- fetch-ohlc
+  "HTTP GET to Kraken OHLC endpoint for 1-minute candles.
+   When `since` is provided, appends &since= to fetch only newer candles."
+  [^HttpClient client since]
+  (let [url     (if since
+                  (str kraken-ohlc-url "&since=" since)
+                  kraken-ohlc-url)
+        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)]
+    (when-let [errors (seq (:error body))]
+      (throw (ex-info "Kraken API error" {:errors errors})))
+    (:result body)))
+
+(defn- parse-candle
+  "Convert a Kraken OHLC array [ts, open, high, low, close, vwap, volume, count]
+   to a map with bigdec values."
+  [[ts open high low close vwap volume count]]
+  {:ts          (long ts)
+   :open        (bigdec open)
+   :high        (bigdec high)
+   :low         (bigdec low)
+   :close       (bigdec close)
+   :vwap        (bigdec vwap)
+   :volume      (bigdec volume)
+   :trade-count (int count)})
+
+(defn- save-candles!
+  "Upsert each candle into the kraken_minute table."
+  [query-fn candles]
+  (doseq [candle candles]
+    (query-fn :upsert-kraken-minute! candle)))
+
+(defn- seed-since-from-db
+  "Query DB for the latest candle timestamp. Returns it or nil."
+  [query-fn]
+  (some-> (query-fn :get-latest-kraken-minute {})
+          :ts))
+
+(defn- poll!
+  "Fetch OHLC data including the last in-progress candle, upsert all.
+   On initial fetch saves everything; subsequent fetches only take last 5.
+   Returns the count of saved candles."
+  [client query-fn since-atom initial?]
+  (let [result   (fetch-ohlc client @since-atom)
+        last-ts  (:last result)
+        pair-key (first (remove #{:last} (keys result)))
+        raw      (get result pair-key)
+        candles  (map parse-candle (if initial? raw (take-last 5 raw)))]
+    (when (seq candles)
+      (save-candles! query-fn candles)
+      (when last-ts
+        (reset! since-atom last-ts))
+      (let [fmt    (DateTimeFormatter/ofPattern "HH:mm")
+            c      (last candles)
+            ts-str (.format (.atOffset (Instant/ofEpochSecond (:ts c)) ZoneOffset/UTC) fmt)]
+        (log/info "Kraken minute:" (count candles) "candles, latest" ts-str "UTC"
+                  "O" (str (:open c)) "H" (str (:high c)) "L" (str (:low c)) "C" (str (:close c)))))
+    (count candles)))
+
+(defn- start-poll-loop!
+  "Start a background future that polls Kraken every 15 seconds."
+  [client query-fn since-atom running?]
+  (future
+    (try
+      (poll! client query-fn since-atom true)
+      (catch Exception e
+        (log/error e "Kraken minute initial poll failed")))
+    (while @running?
+      (Thread/sleep 15000)
+      (when @running?
+        (try
+          (poll! client query-fn since-atom false)
+          (catch Exception e
+            (log/error e "Kraken minute poll failed")))))))
+
+(defmethod ig/init-key :kraken/ohlc-minute
+  [_ {:keys [query-fn enabled?]}]
+  (if (= enabled? "false")
+    (do (log/info "Kraken minute poller disabled")
+        {:running? (atom false)})
+    (let [client    (HttpClient/newHttpClient)
+          db-since  (seed-since-from-db query-fn)
+          since     (atom db-since)
+          running?  (atom true)
+          fut       (start-poll-loop! client query-fn since running?)]
+      (log/info "Starting Kraken minute poller (15s interval)")
+      {:running? running?
+       :future   fut
+       :since    since})))
+
+(defmethod ig/halt-key! :kraken/ohlc-minute
+  [_ {:keys [running? future]}]
+  (log/info "Stopping Kraken minute poller")
+  (reset! running? false)
+  (future-cancel future))

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

@@ -0,0 +1,159 @@
+(ns pmagnus.btcdata.strike.ticker
+  (:require
+   [clojure.data.json :as json]
+   [clojure.string :as str]
+   [clojure.tools.logging :as log]
+   [integrant.core :as ig])
+  (:import
+   [java.net URI]
+   [java.net.http HttpClient HttpRequest HttpResponse$BodyHandlers]
+   [java.time Duration Instant]
+   [org.postgresql.util PGobject]))
+
+(defn- ->jsonb
+  "Wrap a JSON string as a PGobject with type jsonb."
+  [^String s]
+  (doto (PGobject.)
+    (.setType "jsonb")
+    (.setValue s)))
+
+(defn- fetch-rates
+  "GET /v1/rates/ticker from Strike. Returns the raw JSON response body string."
+  [^HttpClient client ^String api-key]
+  (let [request (-> (HttpRequest/newBuilder)
+                    (.uri (URI. "https://api.strike.me/v1/rates/ticker"))
+                    (.header "Accept" "application/json")
+                    (.header "Authorization" (str "Bearer " api-key))
+                    (.GET)
+                    (.build))
+        resp    (.send client request (HttpResponse$BodyHandlers/ofString))
+        body    (.body resp)]
+    body))
+
+(defn- post-quote
+  "POST /v1/currency-exchange-quotes. Returns parsed response map or nil."
+  [^HttpClient client ^String api-key ^String eur-amount]
+  (let [body-str (json/write-str {:sell "EUR" :buy "BTC"
+                                  :amount {:amount eur-amount :currency "EUR"}})
+        request  (-> (HttpRequest/newBuilder)
+                     (.uri (URI. "https://api.strike.me/v1/currency-exchange-quotes"))
+                     (.header "Accept" "application/json")
+                     (.header "Content-Type" "application/json")
+                     (.header "Authorization" (str "Bearer " api-key))
+                     (.POST (java.net.http.HttpRequest$BodyPublishers/ofString body-str))
+                     (.build))
+        resp     (.send client request (HttpResponse$BodyHandlers/ofString))
+        status   (.statusCode resp)
+        body     (.body resp)]
+    (if (= 200 status)
+      (json/read-str body :key-fn keyword)
+      (do (log/warn "Strike quote HTTP" status body)
+          nil))))
+
+(def ^:private fee-max-age (Duration/ofHours 24))
+
+(defn- refresh-fee!
+  "Fetch fee for sats-eur-amount and cache it. Returns the fee BigDecimal or nil."
+  [client api-key sats-eur-amount fee-cache]
+  (when-let [parsed (post-quote client api-key sats-eur-amount)]
+    (let [fee (bigdec (get-in parsed [:fee :amount]))]
+      (log/info "Strike fee for" sats-eur-amount "EUR:" fee "EUR")
+      (reset! fee-cache {:fee fee :fetched-at (Instant/now)})
+      fee)))
+
+(defn- cached-fee
+  "Return cached fee if fresh, otherwise refresh. Returns BigDecimal or nil."
+  [client api-key sats-eur-amount fee-cache]
+  (let [{:keys [fee fetched-at]} @fee-cache]
+    (if (and fee fetched-at
+             (.isBefore (Instant/now) (.plus ^Instant fetched-at fee-max-age)))
+      fee
+      (refresh-fee! client api-key sats-eur-amount fee-cache))))
+
+(defn- fetch-quote-sats
+  "Get sats for a total EUR amount (including fee). Uses cached fee to calculate
+   pre-fee amount, then quotes that. Returns sats as long or nil."
+  [client api-key sats-eur-amount fee-cache]
+  (let [total   (bigdec sats-eur-amount)
+        fee     (cached-fee client api-key sats-eur-amount fee-cache)
+        pre-fee (if fee (.toPlainString (.subtract total fee)) sats-eur-amount)]
+    (when-let [parsed (post-quote client api-key pre-fee)]
+      (let [btc-amt (get-in parsed [:target :amount])]
+        (when btc-amt
+          (long (* (bigdec btc-amt) 100000000)))))))
+
+(defn- format-rates
+  "Build a compact log string from the parsed rates array."
+  [rates]
+  (->> rates
+       (map (fn [{:keys [sourceCurrency targetCurrency amount]}]
+              (str sourceCurrency "/" targetCurrency " " amount)))
+       (str/join "  ")))
+
+(defn- rates->map
+  "Index the rates array by sourceCurrency/targetCurrency pair."
+  [rates]
+  (into {}
+    (map (fn [{:keys [sourceCurrency targetCurrency amount]}]
+           [(str sourceCurrency "/" targetCurrency) amount])
+         rates)))
+
+(defn- poll!
+  "Fetch rates and insert into the database. Updates latest-atom."
+  [client api-key query-fn latest-atom sats-eur-amount fee-cache]
+  (let [body       (fetch-rates client api-key)
+        rates      (json/read-str body :key-fn keyword)
+        rmap       (rates->map rates)
+        quote-sats (when sats-eur-amount
+                     (try
+                       (let [sats (fetch-quote-sats client api-key sats-eur-amount fee-cache)]
+                         (when sats
+                           (log/info "Strike quote:" sats-eur-amount "EUR ->" sats "sats"))
+                         sats)
+                       (catch Exception e
+                         (log/error e "Strike quote fetch failed")
+                         nil)))
+        rmap       (if quote-sats (assoc rmap "quote-sats" quote-sats) rmap)]
+    (query-fn :insert-strike-price! {:rates     (->jsonb body)
+                                     :quote-sats quote-sats})
+    (reset! latest-atom rmap)
+    (log/info "Strike:" (format-rates rates))))
+
+(defn- start-poll-loop!
+  "Fetch immediately, then poll every 10 seconds."
+  [client api-key query-fn running? latest-atom sats-eur-amount fee-cache]
+  (future
+    (try
+      (poll! client api-key query-fn latest-atom sats-eur-amount fee-cache)
+      (catch Exception e
+        (log/error e "Strike initial fetch failed")))
+    (while @running?
+      (Thread/sleep 10000)
+      (when @running?
+        (try
+          (poll! client api-key query-fn latest-atom sats-eur-amount fee-cache)
+          (catch Exception e
+            (log/error e "Strike poll failed")))))))
+
+(defmethod ig/init-key :strike/ticker
+  [_ {: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)
+              fee-cache    (atom nil)
+              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}))))
+
+(defmethod ig/halt-key! :strike/ticker
+  [_ {:keys [running? future]}]
+  (log/info "Stopping Strike ticker poller")
+  (reset! running? false)
+  (when future (future-cancel future)))

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

@@ -0,0 +1,45 @@
+(ns pmagnus.btcdata.web.controllers.day
+  (:require [pmagnus.btcdata.kraken.day :as day])
+  (:import [java.time LocalDate ZoneOffset]
+           [java.time.format DateTimeParseException]))
+
+(defn- response [status body]
+  {:status status :body body})
+
+(defn- parse-date [s]
+  (when (and (string? s) (re-matches #"[0-9]{4}-[0-9]{2}-[0-9]{2}" s))
+    (try (LocalDate/parse s) (catch DateTimeParseException _ nil))))
+
+(defn handle [{:keys [query-fn]} req]
+  (let [s (or (get-in req [:path-params :date])
+              (get-in req [:query-params "date"]))
+        date (parse-date s)]
+    (cond
+      (nil? date) (response 400 {:error "Invalid date, use YYYY-MM-DD"})
+      (not (.isBefore date (LocalDate/now ZoneOffset/UTC)))
+      (response 400 {:error "Only completed UTC days are allowed"})
+      :else
+      (try
+        (if-let [row (day/find-day! query-fn date)]
+          (response 200 (assoc (select-keys row [:open :high :low :close :vwap :volume :trade_count])
+                               :date (str date) :pair "BTC/USD"))
+          (response 404 {:error "No BTC/USD trading data for this day"}))
+        (catch clojure.lang.ExceptionInfo e
+          (if (= ::day/upstream (:type (ex-data e)))
+            (response 502 {:error "Failed to fetch or calculate Kraken daily data"})
+            (throw e)))))))
+
+(def response-schema
+  [:map [:date :string] [:pair :string]
+   [:open number?] [:high number?] [:low number?] [:close number?]
+   [:vwap number?] [:volume number?] [:trade_count int?]])
+
+(defn route [opts parameter-location]
+  {:get {:summary "BTC/USD daily OHLC for a completed UTC date"
+         :description "Checks stored candles first. Missing recent candles are fetched from Kraken; older days are calculated from historical trades and cached. Historical requests wait for completion and may take several minutes."
+         :parameters {parameter-location [:map [:date :string]]}
+         :responses {200 {:body response-schema}
+                     400 {:body [:map [:error :string]]}
+                     404 {:body [:map [:error :string]]}
+                     502 {:body [:map [:error :string]]}}
+         :handler (partial handle opts)}})

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

@@ -1,11 +1,31 @@
 (ns pmagnus.btcdata.web.controllers.health
   (:require
-   [ring.util.http-response :as response])
+   [clojure.tools.logging :as log]
+   [ring.util.http-response :as response]
+   [pmagnus.btcdata.ws.binance :as binance]
+   [pmagnus.btcdata.writer :as writer])
   (:import
    [java.lang.management ManagementFactory]))
 
-(defn healthcheck! [_req]
+(defn healthcheck!
+  "No remote dependency checks. A permanently closed writer session needs restart."
+  [query-fn _req]
+  (if (writer/session-live? query-fn)
     (response/ok
      {:time    (str (java.time.Instant/now))
       :up-time (.. ManagementFactory getRuntimeMXBean getUptime)
-    :app     {:status "up"}}))
+      :app     {:status "up"}})
+    (response/service-unavailable {:status "restart required"
+                                   :reason "writer session closed"})))
+
+(defn readiness-check!
+  "Require the lock-owning database session and a fresh upstream stream."
+  [query-fn component _req]
+  (try
+    (query-fn :check-database {})
+    (if (binance/fresh? component)
+      (response/ok {:status "ready"})
+      (response/service-unavailable {:status "not ready" :reason "binance stale"}))
+    (catch Exception e
+      (log/warn e "Database readiness check failed")
+      (response/service-unavailable {:status "not ready"}))))

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

@@ -0,0 +1,203 @@
+(ns pmagnus.btcdata.web.controllers.transactions
+  (:require
+   [clojure.string :as str]
+   [clojure.tools.logging :as log]
+   [pmagnus.btcdata.frankfurter.rates :as rates]
+   [ring.util.http-response :as response])
+  (:import
+   [java.time LocalDate]
+   [java.util UUID]))
+
+(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 wallet_type]} (:body-params req)]
+    (if (str/blank? name)
+      (response/bad-request {:error "name is required"})
+      (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]
+  (response/ok (query-fn :get-wallet-balances {})))
+
+(defn- convert-amount
+  "Convert fiat-amount in fiat-currency to EUR, DKK, USD using rates map.
+   rates has keys :eur (EUR/USD), :dkk (DKK/USD), :eur_dkk (DKK/EUR)."
+  [fiat-amount fiat-currency rates]
+  (let [amt (bigdec fiat-amount)
+        eur (:eur rates)
+        dkk (:dkk rates)
+        eur-dkk (:eur_dkk rates)]
+    (case fiat-currency
+      "EUR" {:amount-eur amt
+             :amount-dkk (.setScale (* amt eur-dkk) 2 java.math.RoundingMode/HALF_UP)
+             :amount-usd (.setScale (.divide amt eur 6 java.math.RoundingMode/HALF_UP) 2 java.math.RoundingMode/HALF_UP)}
+      "USD" {:amount-eur (.setScale (* amt eur) 2 java.math.RoundingMode/HALF_UP)
+             :amount-dkk (.setScale (* amt dkk) 2 java.math.RoundingMode/HALF_UP)
+             :amount-usd amt}
+      "DKK" {:amount-eur (.setScale (.divide amt eur-dkk 6 java.math.RoundingMode/HALF_UP) 2 java.math.RoundingMode/HALF_UP)
+             :amount-dkk amt
+             :amount-usd (.setScale (.divide amt dkk 6 java.math.RoundingMode/HALF_UP) 2 java.math.RoundingMode/HALF_UP)}
+      ;; Unknown currency — store nil conversions
+      {:amount-eur nil :amount-dkk nil :amount-usd nil})))
+
+(defn- project-deposit!
+  "Project a bank_to_exchange event into the deposits read table."
+  [{:keys [query-fn frankfurter]} event-id occurred-at to-wallet-id fiat-amount fiat-currency note]
+  (try
+    (let [wallet   (query-fn :get-wallet-by-id {:id to-wallet-id})
+          exchange (or (:name wallet) "Unknown")
+          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      event-id
+                        :occurred-at   occurred-at
+                        :exchange      exchange
+                        :fiat-amount   fiat-amount
+                        :fiat-currency fiat-currency
+                        :note          note}
+                       amounts)))
+    (catch Exception e
+      (log/error e "Failed to project deposit for event" event-id))))
+
+(defn- project-buy!
+  "Project an exchange_to_wallet event into the buys read table."
+  [{:keys [query-fn frankfurter]} event-id occurred-at to-wallet-id sats fee-sats fiat-amount fiat-currency]
+  (try
+    (let [wallet-name (or (:name (query-fn :get-wallet-by-id {:id to-wallet-id})) "Unknown")
+          amount-eur  (if (= fiat-currency "EUR")
+                        (bigdec fiat-amount)
+                        (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))]
+                          (:amount-eur (convert-amount fiat-amount fiat-currency rates))))]
+      (when amount-eur
+        (query-fn :insert-buy!
+                  {:event-id    event-id
+                   :occurred-at occurred-at
+                   :wallet      wallet-name
+                   :sats        (or sats 0)
+                   :fee-sats    fee-sats
+                   :amount-eur  amount-eur})))
+    (catch Exception e
+      (log/error e "Failed to project buy for event" event-id))))
+
+(defn create-event! [{:keys [query-fn] :as opts} req]
+  (let [params (:body-params req)
+        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 [d (:occurred_at params)]
+                          (LocalDate/parse d)
+                          (LocalDate/now))
+            row {:event-type     event-type
+                 :occurred-at    occurred-at
+                 :sats           (:sats params)
+                 :fee-sats       (:fee_sats params)
+                 :fiat-amount    (:fiat_amount params)
+                 :fiat-currency  (:fiat_currency params)
+                 :from-wallet-id (some-> (:from_wallet_id params) UUID/fromString)
+                 :to-wallet-id   (some-> (:to_wallet_id params) UUID/fromString)
+                 :note           (:note params)}
+            result (query-fn :insert-event! row)]
+        (when (= event-type "bank_to_exchange")
+          (project-deposit! opts (:id result) occurred-at
+                            (some-> (:to_wallet_id params) UUID/fromString)
+                            (:fiat_amount params)
+                            (:fiat_currency params)
+                            (:note params)))
+        (when (and (= event-type "exchange_to_wallet") (:fiat_amount params))
+          (project-buy! opts (:id result) occurred-at
+                         (some-> (:to_wallet_id params) UUID/fromString)
+                         (:sats params)
+                         (:fee_sats params)
+                         (:fiat_amount params)
+                         (:fiat_currency params)))
+        (response/created "/api/events" {:status "ok"})))))
+
+(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-buys [{:keys [query-fn]} _req]
+  (response/ok (query-fn :get-all-buys {})))
+
+(defn rebuild-buys! [{:keys [query-fn frankfurter]} _req]
+  (query-fn :truncate-buys! {})
+  (let [events (query-fn :get-buy-events {})
+        n      (reduce
+                 (fn [cnt {:keys [id occurred_at sats fee_sats fiat_amount fiat_currency wallet]}]
+                   (if-not fiat_amount
+                     cnt
+                     (try
+                       (let [amount-eur
+                             (if (= fiat_currency "EUR")
+                               (bigdec fiat_amount)
+                               (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))]
+                                 (:amount-eur (convert-amount fiat_amount fiat_currency rates))))]
+                         (when amount-eur
+                           (query-fn :insert-buy!
+                                     {:event-id    id
+                                      :occurred-at occurred_at
+                                      :wallet      (or wallet "Unknown")
+                                      :sats        (or sats 0)
+                                      :fee-sats    fee_sats
+                                      :amount-eur  amount-eur}))
+                         (inc cnt))
+                       (catch Exception e
+                         (log/error e "Failed to rebuild buy 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
+      (response/ok (query-fn :get-events-by-type {:event-type event-type}))
+      (response/ok (query-fn :get-all-events {})))))
+
+(defn get-wallet-events [{:keys [query-fn]} req]
+  (let [id (UUID/fromString (get-in req [:path-params :id]))]
+    (response/ok (query-fn :get-events-by-wallet {:wallet-id id}))))

src/clj/pmagnus/btcdata/web/handler.clj

@@ -2,6 +2,7 @@
   (:require
    [integrant.core :as ig]
    [reitit.ring :as ring]
+   [reitit.swagger-ui :as swagger-ui]
    [pmagnus.btcdata.web.middleware.core :as middleware]))
 
 (defmethod ig/init-key :handler/ring
@@ -10,7 +11,8 @@
    (router)
    (ring/routes
     (when (some? api-path)
-      (reitit.ring/create-default-handler))
+      (swagger-ui/create-swagger-ui-handler
+       {:path "/swagger" :url (str api-path "/swagger.json")}))
     (ring/create-default-handler
      {:not-found          (constantly {:status 404 :body "Page not found"})
       :method-not-allowed (constantly {:status 405 :body "Not allowed"})

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

@@ -9,7 +9,7 @@
     (if (= :options (:request-method request))
       {:status  204
        :headers {"Access-Control-Allow-Origin"  origin
-                 "Access-Control-Allow-Methods" "GET, OPTIONS"
+                 "Access-Control-Allow-Methods" "GET, POST, DELETE, OPTIONS"
                  "Access-Control-Allow-Headers" "Content-Type"
                  "Access-Control-Max-Age"       "86400"}}
       (when-let [resp (handler request)]
@@ -22,7 +22,7 @@
   [handler]
   (fn [request]
     (let [resp (handler request)]
-      (when (:status resp) resp))))
+      (when (or (:status resp) (:undertow/websocket resp)) resp))))
 
 (defn wrap-base [{:keys [site-defaults-config cors-origin]}]
   (let [origin (or cors-origin

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

@@ -2,7 +2,10 @@
   (: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.day :as day]
+   [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]
@@ -11,95 +14,211 @@
    [reitit.ring.middleware.parameters :as parameters]
    [reitit.swagger :as swagger])
   (:import
-   [io.undertow.server HttpServerExchange]
-   [io.undertow.websockets WebSocketConnectionCallback WebSocketProtocolHandshakeHandler]
-   [io.undertow.websockets.core AbstractReceiveListener WebSockets WebSocketChannel]
-   [java.time Instant ZoneId]
-   [java.time.format DateTimeFormatter]
+   [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]}]
+(defn- price->json [{:keys [price prev-price recorded-at]} rates]
   (json/write-str
-   {:price       (str price)
+   (cond-> {:price       (str price)
             :prev_price  (when prev-price (str prev-price))
-    :recorded_at (str recorded-at)}))
+            :recorded_at (str recorded-at)}
+     rates (assoc :price_eur (str (.multiply (bigdec price) (:eur rates)))
+                  :price_dkk (str (.multiply (bigdec price) (:dkk rates)))))))
 
-(def ^:private ts-fmt (DateTimeFormatter/ofPattern "dd-MM HH:mm:ss"))
-
-(defn- price->html [{:keys [price prev-price recorded-at]}]
-  (let [p  (double (bigdec price))
-        pp (when prev-price (double (bigdec prev-price)))
-        color (cond
-                (nil? pp) "text-gray-900"
-                (> p pp)  "text-green-600"
-                (< p pp)  "text-red-600"
-                :else     "text-gray-900")
-        ts (.format (.atZone (Instant/from recorded-at) (ZoneId/systemDefault)) ts-fmt)]
-    (str "<div id=\"price-display\">"
-         "<div class=\"bg-white rounded-2xl shadow-sm border border-gray-200 p-6\">"
-         "<div class=\"text-center\">"
-         "<p class=\"text-4xl font-bold " color "\">$" (format "%,.2f" p) "</p>"
-         "<p class=\"text-xs text-gray-400 mt-2\">Updated " ts "</p>"
-         "</div></div></div>")))
-
-(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-"))]
-    (.set (.getReceiveSetter channel)
-          (proxy [AbstractReceiveListener] []))
-    (.resumeReceives channel)
+        wkey  (keyword (gensym "ws-"))
+        rkey  (keyword (gensym "ws-r-"))]
     (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->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]} req]
-  (let [^HttpServerExchange exchange (:server-exchange req)
-        price-atom (:latest-price binance)
-        callback (proxy [WebSocketConnectionCallback] []
-                   (onConnect [_ws-exchange channel]
-                     (start-ws-send-loop! channel price-atom)))
-        handler (WebSocketProtocolHandshakeHandler. callback)]
-    (.handleRequest handler exchange)
-    nil))
-
-(defn- latest-price-handler [{:keys [binance]} _req]
+(defn- ws-price-handler [{:keys [binance frankfurter]} _req]
   (let [price-atom (:latest-price binance)
-        v          @price-atom]
+        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 query-fn]} _req]
+  (let [price-atom (:latest-price binance)
+        rates-atom (:rates frankfurter)
+        v          (or @price-atom
+                       (when-let [row (query-fn :get-latest-binance-price {})]
+                         {:price       (:price row)
+                          :prev-price  (:price row)
+                          :recorded-at (:recorded_at row)}))]
     (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"})})))
 
+(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"})}))))
+
+(def ^:private ohlc-fmt
+  (java.time.format.DateTimeFormatter/ofPattern "yyyy-MM-dd HH:mm:ss"))
+
+(defn- ohlc-row->map [r]
+  {:ts          (:ts r)
+   :datetime    (.format (.atOffset (java.time.Instant/ofEpochSecond (:ts r))
+                                    java.time.ZoneOffset/UTC) ohlc-fmt)
+   :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)})
+
+(defn- kraken-hour-handler [{:keys [query-fn]} _req]
+  {:status  200
+   :headers {"Content-Type" "application/json"}
+   :body    (json/write-str (mapv ohlc-row->map (query-fn :get-kraken-hour-latest-24 {})))})
+
+(defn- kraken-minute-handler [{:keys [query-fn]} _req]
+  {:status  200
+   :headers {"Content-Type" "application/json"}
+   :body    (json/write-str (mapv ohlc-row->map (query-fn :get-kraken-minute-latest-60 {})))})
+
 (defn- api-routes [opts]
   [["/swagger.json"
     {:get {:no-doc  true
            :swagger {:info {:title "btcdata API"}}
            :handler (swagger/create-swagger-handler)}}]
    ["/health"
-    {:get health/healthcheck!}]
+    {:get (fn [req] (health/healthcheck! (:query-fn opts) req))}]
+   ["/ready"
+    {:get (fn [req] (health/readiness-check! (:query-fn opts) (:binance opts) req))}]
    ["/price/ws"
     {:get        (fn [req] (ws-price-handler opts req))
      :no-doc     true
      :middleware []}]
    ["/price/latest"
-    {:get (fn [req] (latest-price-handler opts req))}]])
+    {:get (fn [req] (latest-price-handler opts req))}]
+   ["/price/day" (day/route opts :query)]
+   ["/price/day/:date" (day/route opts :path)]
+   ["/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))}]
+   ["/buys"
+    {:get (fn [req] (tx/list-buys opts req))}]
+   ["/buys/rebuild"
+    {:post (fn [req] (tx/rebuild-buys! opts req))}]
+   ["/kraken-hour"
+    {:get (fn [req] (kraken-hour-handler opts req))}]
+   ["/kraken-minute"
+    {:get (fn [req] (kraken-minute-handler opts req))}]])
 
 (defn route-data [opts]
   (merge

src/clj/pmagnus/btcdata/writer.clj

@@ -0,0 +1,55 @@
+(ns pmagnus.btcdata.writer
+  "Fence all application queries with one PostgreSQL session advisory lock.
+   Queries use the lock-owning connection itself: a lost session cannot keep
+   writing through a different pooled connection. Never reconnect this session."
+  (:require [integrant.core :as ig]
+            [next.jdbc :as jdbc]))
+
+(def lock-id 728194106)
+
+(defn configure-session! [connection]
+  ;; Client-side network timeouts cannot reap a server session left behind by a
+  ;; dead node. Server TCP keepalives release its fence after ~90s of silence.
+  (doseq [sql ["SET tcp_keepalives_idle = 60"
+               "SET tcp_keepalives_interval = 10"
+               "SET tcp_keepalives_count = 3"
+               "SET tcp_user_timeout = 90000"]]
+    (jdbc/execute-one! connection [sql])))
+
+(defn session-live?
+  "Local-only liveness check; a closed fenced session can never recover in place."
+  [query-fn]
+  (if-let [connection (:connection (meta query-fn))]
+    (try
+      (not (.isClosed ^java.sql.Connection connection))
+      (catch Exception _ false))
+    false))
+
+(defn guarded-query [connection query-fn]
+  (fn [query params]
+    (locking connection
+      (query-fn connection query params))))
+
+(defmethod ig/init-key :writer/query-fn
+  [_ {:keys [conn query-fn]}]
+  (let [connection (jdbc/get-connection conn)]
+    (try
+      ;; Bound lock/query stalls; a failed session is never replaced behind the fence.
+      (.setNetworkTimeout connection
+        (reify java.util.concurrent.Executor (execute [_ task] (.run ^Runnable task)))
+        10000)
+      (configure-session! connection)
+      (when-not (:locked (jdbc/execute-one! connection
+                           ["select pg_try_advisory_lock(?) as locked" lock-id]))
+        (throw (ex-info "Another btcdata writer holds the database lock" {})))
+      (with-meta (guarded-query connection query-fn) {:connection connection})
+      (catch Exception e
+        (.close connection)
+        (throw e)))))
+
+(defmethod ig/halt-key! :writer/query-fn [_ query-fn]
+  (let [connection (:connection (meta query-fn))]
+    (locking connection
+      (try
+        (jdbc/execute-one! connection ["select pg_advisory_unlock(?)" lock-id])
+        (finally (.close connection))))))

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

@@ -1,93 +1,102 @@
 (ns pmagnus.btcdata.ws.binance
-  (:require
-   [clojure.data.json :as json]
+  (:require [clojure.data.json :as json]
             [clojure.tools.logging :as log]
             [integrant.core :as ig])
-  (:import
-   [java.net URI]
+  (:import [java.net URI]
            [java.net.http HttpClient WebSocket WebSocket$Listener]
-   [java.util.concurrent CompletableFuture CompletionStage]))
+           [java.time Duration Instant]
+           [java.util.concurrent CompletableFuture TimeUnit]))
 
-(defn- save-price! [query-fn price]
-  (try
-    (query-fn :insert-binance-price! {:price price})
-    (catch Exception e
-      (log/error e "Failed to save BTC price"))))
+(def stale-after-ms 60000)
 
-(defn- connect!
-  "Opens a WebSocket to Binance trade stream and returns the WebSocket instance.
-   `state` is an atom with keys :running?, :last-write, :buffer.
-   `latest-price` is an atom reset with each throttled price for SSE consumers."
-  [uri query-fn state latest-price]
-  (let [client   (HttpClient/newHttpClient)
-        listener (reify WebSocket$Listener
+(defn fresh?
+  "Only in-process stream data counts; database fallback is not stream health."
+  ([component] (fresh? component (System/currentTimeMillis)))
+  ([{:keys [state latest-price]} now]
+   (let [{:keys [enabled? connected?]} @state
+         recorded-at (:recorded-at @latest-price)]
+     (or (false? enabled?)
+         (and connected? (some? recorded-at)
+              (<= 0 (- now (.toEpochMilli ^Instant recorded-at)) stale-after-ms))))))
+
+(defn- connect! [client uri query-fn state latest-price generation]
+  (let [buffer (StringBuilder.)
+        current? #(and (:running? @state) (= generation (:generation @state)))
+        listener
+        (reify WebSocket$Listener
           (onOpen [_ ws]
+            (if (current?)
+              (do (swap! state assoc :connected? true)
                   (log/info "Binance WebSocket connected")
                   (.request ws 1))
-
+              (.abort ws)))
           (onText [_ ws data last?]
-                     (let [buf (:buffer @state)]
-                       (.append buf data)
+            (when (current?)
+              (.append buffer data)
               (when last?
-                         (let [text (str buf)]
-                           (.setLength buf 0)
+                (let [text (str buffer)]
+                  (.setLength buffer 0)
                   (try
-                             (let [msg   (json/read-str text :key-fn keyword)
-                                   price (some-> (:p msg) bigdec)]
-                               (when price
+                    (when-let [price (some-> (json/read-str text :key-fn keyword) :p bigdec)]
                       (let [now (System/currentTimeMillis)]
+                        (swap! state assoc :last-message now)
                         (when (> (- now (:last-write @state)) 5000)
+                          ;; Publish only after persistence succeeds.
+                          (query-fn :insert-binance-price! {:price price})
                           (swap! state assoc :last-write now)
-                                     (log/info "BTC price:" (str price))
-                                     (save-price! query-fn price)
-                                     (let [prev-price (:price @latest-price)]
-                                       (reset! latest-price
-                                               {:price      price
-                                                :prev-price prev-price
-                                                :recorded-at (java.time.Instant/now)}))))))
+                          (reset! latest-price {:price price
+                                                :prev-price (:price @latest-price)
+                                                :recorded-at (Instant/now)}))))
                     (catch Exception e
-                               (log/error e "Failed to parse Binance message")))))
-                       (let [^CompletionStage cf (CompletableFuture/completedFuture nil)]
+                      (log/error e "Failed to process Binance price"))))))
             (.request ws 1)
-                         cf)))
-
-                   (onClose [_ _ws code reason]
-                     (log/warn "Binance WebSocket closed:" code reason)
-                     (when (:running? @state)
-                       (future
-                         (Thread/sleep 3000)
-                         (when (:running? @state)
-                           (log/info "Reconnecting to Binance...")
-                           (try
-                             (let [ws (connect! uri query-fn state latest-price)]
-                               (swap! state assoc :ws ws))
-                             (catch Exception e
-                               (log/error e "Binance reconnect failed")))))))
-
-                   (onError [_ _ws error]
-                     (log/error error "Binance WebSocket error")))]
-    (-> (.newWebSocketBuilder client)
+            (CompletableFuture/completedFuture nil))
+          (onClose [_ _ code reason]
+            (when (current?)
+              (swap! state assoc :connected? false)
+              (log/warn "Binance WebSocket closed:" code reason))
+            (CompletableFuture/completedFuture nil))
+          (onError [_ _ error]
+            (when (current?)
+              (swap! state assoc :connected? false)
+              (log/warn error "Binance WebSocket error"))))]
+    (-> (.newWebSocketBuilder ^HttpClient client)
+        (.connectTimeout (Duration/ofSeconds 10))
         (.buildAsync (URI. uri) listener)
-        (.join))))
+        (.get 15 TimeUnit/SECONDS))))
 
 (defmethod ig/init-key :ws/binance
-  [_ {:keys [query-fn uri]}]
-  (log/info "Starting Binance WebSocket listener:" uri)
-  (let [latest-price (atom nil)
-        state        (atom {:running?   true
-                            :last-write 0
-                            :buffer     (StringBuilder.)
-                            :ws         nil})
-        ws           (connect! uri query-fn state latest-price)]
-    (swap! state assoc :ws ws)
-    {:state        state
-     :latest-price latest-price}))
-
-(defmethod ig/halt-key! :ws/binance
-  [_ {:keys [state]}]
-  (log/info "Stopping Binance WebSocket listener")
-  (swap! state assoc :running? false)
-  (when-let [ws (:ws @state)]
+  [_ {:keys [query-fn uri enabled?] :or {enabled? "true"}}]
+  (let [enabled (= enabled? "true")
+        latest-price (atom nil)
+        state (atom {:enabled? enabled :running? enabled :connected? false
+                     :last-write 0 :last-message 0 :generation nil :ws nil})
+        client (when enabled (-> (HttpClient/newBuilder)
+                                 (.connectTimeout (Duration/ofSeconds 10)) .build))
+        worker
+        (when enabled
+          (future
             (try
-      (.sendClose ws WebSocket/NORMAL_CLOSURE "shutting down")
-      (catch Exception _))))
+              (while (:running? @state)
+                (let [generation (Object.)]
+                  (swap! state assoc :generation generation :connected? false
+                         :last-message (System/currentTimeMillis))
+                  (try
+                    (let [ws (connect! client uri query-fn state latest-price generation)]
+                      (swap! state assoc :ws ws)
+                      (try
+                        (while (and (:running? @state) (:connected? @state)
+                                    (< (- (System/currentTimeMillis) (:last-message @state)) stale-after-ms))
+                          (Thread/sleep 1000))
+                        (finally (.abort ^WebSocket ws))))
+                    (catch InterruptedException e (throw e))
+                    (catch Exception e (log/warn e "Binance connection failed; retrying"))
+                    (finally (swap! state assoc :connected? false :generation nil :ws nil))))
+                (when (:running? @state) (Thread/sleep 3000)))
+              (catch InterruptedException _))))]
+    {:state state :latest-price latest-price :worker worker}))
+
+(defmethod ig/halt-key! :ws/binance [_ {:keys [state worker]}]
+  (swap! state assoc :running? false :connected? false :generation nil)
+  (when-let [ws (:ws @state)] (.abort ^WebSocket ws))
+  (when worker (future-cancel worker)))

test/clj/pmagnus/btcdata/kraken/day_test.clj

@@ -0,0 +1,77 @@
+(ns pmagnus.btcdata.kraken.day-test
+  (:require [clojure.test :refer [deftest is]]
+            [pmagnus.btcdata.kraken.day :as day])
+  (:import [java.time LocalDate ZoneOffset]))
+
+(def date (LocalDate/parse "2020-01-01"))
+(def ts (.toEpochSecond (.atStartOfDay date) ZoneOffset/UTC))
+
+(defn database [store writes]
+  (fn [q p]
+    (case q
+      :get-kraken-day-by-ts (get @store [:recent (:ts p)])
+      :get-kraken-day-history-by-ts (get @store [:history (:ts p)])
+      (:upsert-kraken-day! :upsert-kraken-day-history!)
+      (let [table (if (= q :upsert-kraken-day!) :recent :history)]
+        (swap! writes conj q)
+        (swap! store assoc [table (:ts p)]
+               (-> p (assoc :trade_count (:trade-count p)) (dissoc :trade-count)))))))
+
+(deftest cached-data-never-fetches
+  (doseq [table [:recent :history]]
+    (let [row {:ts ts} store (atom {[table ts] row})]
+      (with-redefs [day/request! (fn [_] (throw (Exception. "Unexpected fetch")))]
+        (is (= row (day/find-day! (database store (atom [])) date)))))))
+
+(deftest recent-candle-is-persisted
+  (let [store (atom {}) writes (atom []) q (database store writes)]
+    (with-redefs [day/request! (fn [_] {:XXBTZUSD [[ts "10" "12" "9" "11" "10.5" "2" 3]]})]
+      (is (= 11M (:close (day/find-day! q date))))
+      (is (= [:upsert-kraken-day!] @writes)))))
+
+(deftest historical-pagination-and-cache
+  (let [store (atom {}) writes (atom []) q (database store writes)
+        calls (atom []) cursor (str (*' (+ ts 5) 1000000000))]
+    (with-redefs [day/pause! (fn [])
+                  day/request!
+                  (fn [url]
+                    (swap! calls conj url)
+                    (case (count @calls)
+                      1 {:XXBTZUSD [[(+ ts 86400) "1" "1" "1" "1" "1" "1" 1]]}
+                      2 {:XXBTZUSD [["10" "2" ts] ["20" "1" (+ ts 5)]] :last cursor}
+                      3 {:XXBTZUSD [["5" "1" (+ ts 86399.9M)]
+                                   ["999" "1" (+ ts 86400)]] :last "unused"}))]
+      (let [result (day/find-day! q date)]
+        (is (= {:open 10M :high 20M :low 5M :close 5M :volume 4M
+                :vwap 11.25000000M :trade_count 3 :ts ts} result))
+        (is (= result (day/find-day! q date)))
+        (is (= 3 (count @calls)))
+        (is (.endsWith (last @calls) cursor))
+        (is (= [:upsert-kraken-day-history!] @writes))))))
+
+(deftest failed-or-empty-history-is-not-persisted
+  (doseq [result [{:XXBTZUSD [] :last "0"}
+                  {:XXBTZUSD [["1" "1" ts]] :last "0"}
+                  {:wrong []}]]
+    (let [writes (atom []) calls (atom 0)]
+      (with-redefs [day/request! (fn [_]
+                                  (if (= 1 (swap! calls inc))
+                                    {:XXBTZUSD [[(+ ts 86400)]]} result))]
+        (if (= [] (:XXBTZUSD result))
+          (is (nil? (day/find-day! (database (atom {}) writes) date)))
+          (is (thrown? clojure.lang.ExceptionInfo
+                       (day/find-day! (database (atom {}) writes) date))))
+        (is (empty? @writes))))))
+
+(deftest upstream-failure-does-not-save-a-partial-day
+  (let [writes (atom []) calls (atom 0)]
+    (with-redefs [day/pause! (fn [])
+                  day/request! (fn [_]
+                                 (case (swap! calls inc)
+                                   1 {:XXBTZUSD [[(+ ts 86400)]]}
+                                   2 {:XXBTZUSD [["10" "1" ts]]
+                                      :last (str (*' (inc ts) 1000000000))}
+                                   (throw (Exception. "Connection failed"))))]
+      (is (thrown? clojure.lang.ExceptionInfo
+                   (day/find-day! (database (atom {}) writes) date)))
+      (is (empty? @writes)))))

test/clj/pmagnus/btcdata/web/controllers/day_test.clj

@@ -0,0 +1,54 @@
+(ns pmagnus.btcdata.web.controllers.day-test
+  (:require [clojure.test :refer [deftest is]]
+            [clojure.data.json :as json]
+            [integrant.core :as ig]
+            [pmagnus.btcdata.kraken.day :as day]
+            [pmagnus.btcdata.web.controllers.day :as controller]
+            [pmagnus.btcdata.web.routes.api :as api]
+            [pmagnus.btcdata.web.handler]
+            [reitit.ring :as ring]))
+
+(def app
+  (ring/ring-handler
+   (ring/router [(ig/init-key :reitit.routes/api {:base-path "/api" :query-fn nil})])))
+
+(defn request [uri query]
+  (app {:request-method :get :uri uri :query-string query
+        :headers {"accept" "application/json"}}))
+
+(deftest routes-and-numeric-json
+  (with-redefs [day/find-day! (fn [_ _]
+                              {:open 10M :high 12M :low 9M :close 11M
+                               :vwap 10.5M :volume 2M :trade_count 3})]
+    (doseq [[uri query] [["/api/price/day/2020-01-01" nil]
+                        ["/api/price/day" "date=2020-01-01"]]]
+      (let [r (request uri query) body (json/read-str (slurp (:body r)) :key-fn keyword)]
+        (is (= 200 (:status r)))
+        (is (= "BTC/USD" (:pair body)))
+        (is (= "2020-01-01" (:date body)))
+        (is (number? (:open body)))))))
+
+(deftest invalid-dates
+  (doseq [s [nil "bad" "2020-02-30" "2020-1-01" "9999-01-01"
+             (str (java.time.LocalDate/now java.time.ZoneOffset/UTC))]]
+    (with-redefs [day/find-day! (fn [& _] (throw (Exception. "Must not fetch")))]
+      (is (= 400 (:status (controller/handle {} {:query-params {"date" s}})))))))
+
+(deftest error-responses
+  (with-redefs [day/find-day! (constantly nil)]
+    (is (= 404 (:status (request "/api/price/day/2020-01-01" nil)))))
+  (with-redefs [day/find-day! (fn [& _] (throw (ex-info "upstream" {:type ::day/upstream})))]
+    (is (= 502 (:status (request "/api/price/day/2020-01-01" nil))))))
+
+(deftest swagger-documents-both-routes
+  (let [r (request "/api/swagger.json" nil)
+        body (json/read-str (slurp (:body r)))]
+    (is (= 200 (:status r)))
+    (is (get-in body ["paths" "/api/price/day" "get"]))
+    (is (get-in body ["paths" "/api/price/day/{date}" "get"]))))
+
+(deftest swagger-ui-is-served
+  (let [handler (ig/init-key :handler/ring
+                            {:router #(ring/router []) :api-path "/api"})]
+    (is (= 302 (:status (handler {:request-method :get :uri "/swagger" :headers {}}))))
+    (is (= 200 (:status (handler {:request-method :get :uri "/swagger/" :headers {}}))))))

test/clj/pmagnus/btcdata/web/controllers/health_test.clj

@@ -0,0 +1,34 @@
+(ns pmagnus.btcdata.web.controllers.health-test
+  (:require [clojure.test :refer [deftest is]]
+            [pmagnus.btcdata.web.controllers.health :as health]))
+
+(def disabled-stream {:state (atom {:enabled? false}) :latest-price (atom nil)})
+
+(deftest liveness-check-is-local-and-detects-lost-writer
+  (let [closed? (atom false)
+        connection (proxy [java.sql.Connection] []
+                     (isClosed [] @closed?))
+        query-fn (with-meta (fn [& _] (throw (Exception. "Must not query database")))
+                   {:connection connection})]
+    (is (= 200 (:status (health/healthcheck! query-fn {}))))
+    (reset! closed? true)
+    (is (= 503 (:status (health/healthcheck! query-fn {}))))
+    (is (= "writer session closed"
+           (get-in (health/healthcheck! query-fn {}) [:body :reason]))))
+  (is (= 503 (:status (health/healthcheck! (fn [& _]) {})))))
+
+(deftest readiness-check-tests-database
+  (let [called (atom nil)
+        query-fn (fn [query params] (reset! called [query params]))]
+    (is (= 200 (:status (health/readiness-check! query-fn disabled-stream {}))))
+    (is (= [:check-database {}] @called)))
+  (is (= 503 (:status (health/readiness-check!
+                       (fn [_ _] (throw (ex-info "database unavailable" {})))
+                       disabled-stream {})))))
+
+(deftest stale-stream-is-not-ready
+  (is (= 503 (:status (health/readiness-check!
+                       (fn [_ _] {})
+                       {:state (atom {:enabled? true :connected? true})
+                        :latest-price (atom {:recorded-at java.time.Instant/EPOCH})}
+                       {})))))

test/clj/pmagnus/btcdata/writer_test.clj

@@ -0,0 +1,80 @@
+(ns pmagnus.btcdata.writer-test
+  (:require [clojure.test :refer [deftest is]]
+            [integrant.core :as ig]
+            [conman.core :as conman]
+            [next.jdbc :as jdbc]
+            [pmagnus.btcdata.writer :as writer]))
+
+(deftest queries-use-lock-session
+  (let [connection (Object.)
+        seen (atom nil)
+        query (writer/guarded-query connection
+                (fn [& args] (reset! seen args) :result))]
+    (is (= :result (query :insert-price {:price 1})))
+    (is (= [connection :insert-price {:price 1}] @seen))))
+
+(deftest session-liveness-fails-closed
+  (is (false? (writer/session-live? (fn [& _]))))
+  (let [connection (proxy [java.sql.Connection] []
+                     (isClosed [] (throw (java.sql.SQLException. "broken"))))]
+    (is (false? (writer/session-live? (with-meta (fn [& _])
+                                       {:connection connection}))))))
+
+(deftest server-session-has-bounded-dead-peer-detection
+  (let [connection (Object.)
+        calls (atom [])]
+    (with-redefs [jdbc/execute-one! (fn [conn sql] (swap! calls conj [conn sql]))]
+      (writer/configure-session! connection))
+    (is (= (mapv #(vector connection [%])
+                 ["SET tcp_keepalives_idle = 60"
+                  "SET tcp_keepalives_interval = 10"
+                  "SET tcp_keepalives_count = 3"
+                  "SET tcp_user_timeout = 90000"])
+           @calls))))
+
+(deftest postgres-session-fencing
+  ;; Optional disposable PostgreSQL integration test; never use production here.
+  (if-let [url (System/getenv "BTCDATA_TEST_JDBC_URL")]
+    (let [ds (jdbc/get-datasource {:jdbcUrl url})
+          opts {:conn ds :query-fn (fn [conn _ _]
+                                    (jdbc/execute-one! conn ["select 1 as healthy"]))}
+          first-writer (ig/init-key :writer/query-fn opts)]
+      (try
+        (is (= 1 (:healthy (first-writer :check {}))))
+        (let [connection (:connection (meta first-writer))]
+          (doseq [[setting expected] [["tcp_keepalives_idle" "60"]
+                                     ["tcp_keepalives_interval" "10"]
+                                     ["tcp_keepalives_count" "3"]
+                                     ["tcp_user_timeout" "90000"]]]
+            (is (= expected (:value (jdbc/execute-one! connection
+                                     ["select current_setting(?) as value" setting]))))))
+        (is (thrown-with-msg? clojure.lang.ExceptionInfo #"Another btcdata writer"
+              (ig/init-key :writer/query-fn opts)))
+        ;; Closing the owning session cannot transparently reconnect for writes.
+        (is (writer/session-live? first-writer))
+        (.close (:connection (meta first-writer)))
+        (is (false? (writer/session-live? first-writer)))
+        (is (thrown? Exception (first-writer :check {})))
+        (let [replacement (ig/init-key :writer/query-fn opts)]
+          (try
+            (is (= 1 (:healthy (replacement :check {}))))
+            (finally (ig/halt-key! :writer/query-fn replacement))))
+        (finally
+          (when-not (.isClosed (:connection (meta first-writer)))
+            (ig/halt-key! :writer/query-fn first-writer)))))
+    (println "SKIP PostgreSQL fencing integration: set BTCDATA_TEST_JDBC_URL")))
+
+(deftest pooled-session-releases-lock-on-halt
+  (when-let [url (System/getenv "BTCDATA_TEST_JDBC_URL")]
+    (let [pool (conman/connect! {:jdbc-url url :maximum-pool-size 3})
+          opts {:conn pool :query-fn (fn [_ _ _] :ok)}]
+      (try
+        (let [leader (ig/init-key :writer/query-fn opts)]
+          (is (thrown? Exception (ig/init-key :writer/query-fn opts)))
+          (ig/halt-key! :writer/query-fn leader))
+        ;; An independent connection can acquire after pooled-session shutdown.
+        (let [other-opts (assoc opts :conn (jdbc/get-datasource {:jdbcUrl url}))
+              leader (ig/init-key :writer/query-fn other-opts)]
+          (is (= :ok (leader :check {})))
+          (ig/halt-key! :writer/query-fn leader))
+        (finally (conman/disconnect! pool))))))

test/clj/pmagnus/btcdata/ws/binance_test.clj

@@ -0,0 +1,41 @@
+(ns pmagnus.btcdata.ws.binance-test
+  (:require [clojure.test :refer [deftest is]]
+            [integrant.core :as ig]
+            [pmagnus.btcdata.ws.binance :as binance])
+  (:import [java.time Instant]))
+
+(deftest freshness-boundaries
+  (let [state (atom {:enabled? true :connected? true})
+        price (atom {:recorded-at (Instant/ofEpochMilli 100000)})
+        component {:state state :latest-price price}]
+    (is (binance/fresh? component 100000))
+    (is (binance/fresh? component 160000))
+    (is (not (binance/fresh? component 160001)))
+    (is (not (binance/fresh? component 99999)))
+    (swap! state assoc :connected? false)
+    (is (not (binance/fresh? component 100000)))
+    (reset! price nil)
+    (is (not (binance/fresh? component 100000)))))
+
+(deftest disabled-stream-does-not-connect
+  (let [component (ig/init-key :ws/binance {:enabled? "false" :uri "invalid"})]
+    (is (binance/fresh? component))
+    (is (nil? (:worker component)))
+    (ig/halt-key! :ws/binance component)
+    (is (false? (:running? @(:state component))))))
+
+(deftest initial-failure-is-retried-and-worker-stops
+  (let [attempts (atom 0)
+        connect-var (ns-resolve 'pmagnus.btcdata.ws.binance 'connect!)]
+    (with-redefs-fn
+      {connect-var (fn [& _] (swap! attempts inc) (throw (ex-info "test failure" {})))}
+      (fn []
+        (let [component (ig/init-key :ws/binance {:enabled? "true" :uri "test"})]
+          (try
+            (let [deadline (+ (System/currentTimeMillis) 6000)]
+              (while (and (< @attempts 2) (< (System/currentTimeMillis) deadline))
+                (Thread/sleep 20)))
+            (is (>= @attempts 2))
+            (is (not (binance/fresh? component)))
+            (finally (ig/halt-key! :ws/binance component)))
+          (is (future-cancelled? (:worker component))))))))