Copy real test inputs into nested CI containers

The database moved to citadel (192.168.50.195); .env and the deployed
container already point there, but env.example still handed out the old
omarchy address.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EWRh3vxBcqwDbuWv2dfQYF

.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.

CLAUDE.md

@@ -37,7 +37,7 @@ 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
 

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

btcdata-deploy.md

@@ -0,0 +1,294 @@
+# btcdata Kubernetes deployment plan
+
+## 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.

deploy-citadel.sh

@@ -0,0 +1,47 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+HOST="${BTCDATA_DEPLOY_HOST:-citadel}"
+APP_DIR="${BTCDATA_DEPLOY_DIR:-/home/pmp/apps/btcdata}"
+IMAGE_NAME="${BTCDATA_IMAGE_NAME:-$(basename "$APP_DIR")-btcdata:latest}"
+PLATFORM="${BTCDATA_PLATFORM:-}"
+NETWORK="${BTCDATA_NETWORK:-btc-network}"
+FILES=(docker-compose.yml .env)
+
+if [[ -z "$PLATFORM" ]]; then
+  remote_arch="$(ssh "$HOST" uname -m)"
+  case "$remote_arch" in
+  x86_64 | amd64) PLATFORM="linux/amd64" ;;
+  aarch64 | arm64) PLATFORM="linux/arm64" ;;
+  *)
+    echo "Unsupported target architecture: $remote_arch" >&2
+    echo "Set BTCDATA_PLATFORM manually, e.g. BTCDATA_PLATFORM=linux/amd64" >&2
+    exit 1
+    ;;
+  esac
+fi
+
+for file in "${FILES[@]}"; do
+  if [[ ! -f "$file" ]]; then
+    echo "Missing required file: $file" >&2
+    exit 1
+  fi
+done
+
+echo "Building $IMAGE_NAME locally for $PLATFORM..."
+docker build --platform "$PLATFORM" -t "$IMAGE_NAME" .
+
+echo "Preparing $APP_DIR on $HOST..."
+ssh "$HOST" "mkdir -p '$APP_DIR'"
+
+echo "Ensuring external network $NETWORK exists on $HOST..."
+ssh "$HOST" "docker network inspect '$NETWORK' >/dev/null 2>&1 || docker network create '$NETWORK'"
+
+echo "Copying docker-compose.yml and .env to $HOST:$APP_DIR..."
+scp "${FILES[@]}" "$HOST:$APP_DIR/"
+
+echo "Copying Docker image $IMAGE_NAME to $HOST..."
+docker save "$IMAGE_NAME" | ssh "$HOST" docker load
+
+echo "Starting docker compose on $HOST in $APP_DIR..."
+ssh "$HOST" "cd '$APP_DIR' && docker compose up -d --no-build"

deps.edn

@@ -46,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}
 

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/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/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)
@@ -155,6 +159,30 @@ SELECT id, event_id, occurred_at, exchange, fiat_amount, fiat_currency,
 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

resources/system.edn

@@ -43,36 +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
+  :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
+ {: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 :db.sql/query-fn
+ {: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 :db.sql/query-fn}
+  :query-fn #ig/ref :writer/query-fn}
 
  :strike/ticker
- {:query-fn        #ig/ref :db.sql/query-fn
+ {: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/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

@@ -16,6 +16,7 @@
    [pmagnus.btcdata.web.routes.api]
    ;; WebSocket clients
    [pmagnus.btcdata.ws.binance]
+   [pmagnus.btcdata.writer]
    ;; Pollers
    [pmagnus.btcdata.kraken.ohlc]
    [pmagnus.btcdata.kraken.ohlc-daily]

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

@@ -1,6 +1,8 @@
 (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])
   (:import
    [java.lang.management ManagementFactory]))
 
@@ -9,3 +11,15 @@
    {:time    (str (java.time.Instant/now))
     :up-time (.. ManagementFactory getRuntimeMXBean getUptime)
     :app     {:status "up"}}))
+
+(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

@@ -69,6 +69,29 @@
     (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)]
@@ -93,6 +116,13 @@
                             (: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]
@@ -128,6 +158,40 @@
                  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

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

@@ -87,10 +87,14 @@
      {:on-open (fn [{:keys [^WebSocketChannel channel]}]
                  (start-strike-ws-loop! channel strike-atom))}}))
 
-(defn- latest-price-handler [{:keys [binance frankfurter]} _req]
+(defn- latest-price-handler [{:keys [binance frankfurter query-fn]} _req]
   (let [price-atom (:latest-price binance)
         rates-atom (:rates frankfurter)
-        v          @price-atom]
+        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"}
@@ -176,6 +180,8 @@
            :handler (swagger/create-swagger-handler)}}]
    ["/health"
     {:get health/healthcheck!}]
+   ["/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
@@ -202,6 +208,10 @@
     {: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"

src/clj/pmagnus/btcdata/writer.clj

@@ -0,0 +1,36 @@
+(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 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)
+      (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,97 +1,102 @@
-(ns pmagnus.btcdata.ws.binance
-  (:require
-   [clojure.data.json :as json]
-   [clojure.tools.logging :as log]
-   [integrant.core :as ig])
-  (:import
-   [java.net URI]
-   [java.net.http HttpClient WebSocket WebSocket$Listener]
-   [java.util.concurrent CompletableFuture CompletionStage]))
-
-(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"))))
-
-(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
-                   (onOpen [_ ws]
-                     (log/info "Binance WebSocket connected")
-                     (.request ws 1))
-
-                   (onText [_ ws data last?]
-                     (let [buf (:buffer @state)]
-                       (.append buf data)
-                       (when last?
-                         (let [text (str buf)]
-                           (.setLength buf 0)
-                           (try
-                             (let [msg   (json/read-str text :key-fn keyword)
-                                   price (some-> (:p msg) bigdec)]
-                               (when price
-                                 (let [now (System/currentTimeMillis)]
-                                   (when (> (- now (:last-write @state)) 5000)
-                                     (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)}))))))
-                             (catch Exception e
-                               (log/error e "Failed to parse Binance message")))))
-                       (let [^CompletionStage cf (CompletableFuture/completedFuture nil)]
-                         (.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)
-        (.buildAsync (URI. uri) listener)
-        (.join))))
-
-(defmethod ig/init-key :ws/binance
-  [_ {:keys [query-fn uri enabled?] :or {enabled? "true"}}]
-  (if-not (= enabled? "true")
-    (do (log/info "Binance WebSocket disabled")
-        {:state        (atom {:running? false})
-         :latest-price (atom nil)})
-    (do (log/info "Starting Binance WebSocket listener:" uri)
-        (let [latest-price (atom nil)
-              state        (atom {:running?   true
-                                  :last-write 0
-                                  :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)]
-    (try
-      (.sendClose ws WebSocket/NORMAL_CLOSURE "shutting down")
-      (catch Exception _))))
+(ns pmagnus.btcdata.ws.binance
+  (:require [clojure.data.json :as json]
+            [clojure.tools.logging :as log]
+            [integrant.core :as ig])
+  (:import [java.net URI]
+           [java.net.http HttpClient WebSocket WebSocket$Listener]
+           [java.time Duration Instant]
+           [java.util.concurrent CompletableFuture TimeUnit]))
+
+(def stale-after-ms 60000)
+
+(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?]
+            (when (current?)
+              (.append buffer data)
+              (when last?
+                (let [text (str buffer)]
+                  (.setLength buffer 0)
+                  (try
+                    (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)
+                          (reset! latest-price {:price price
+                                                :prev-price (:price @latest-price)
+                                                :recorded-at (Instant/now)}))))
+                    (catch Exception e
+                      (log/error e "Failed to process Binance price"))))))
+            (.request ws 1)
+            (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)
+        (.get 15 TimeUnit/SECONDS))))
+
+(defmethod ig/init-key :ws/binance
+  [_ {: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
+              (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/web/controllers/health_test.clj

@@ -0,0 +1,24 @@
+(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 healthcheck-is-process-only
+  (is (= 200 (:status (health/healthcheck! {})))))
+
+(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,52 @@
+(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 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 {}))))
+        (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.
+        (.close (:connection (meta 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))))))))