bsv_wallet_cli/
relay.rs

1//! bsv-wallet-relay — store-and-forward ARC/Arcade callback receiver.
2//!
3//! ONE public callback receiver on a box; multiple wallet daemons (which may
4//! be localhost-only) drain their own queue OUTBOUND by callback token:
5//!
6//! ```text
7//! Arcade ──POST /arc-callback (Bearer <token>)──▶ relay (sqlite queue, per token)
8//! wallet ──GET  /pull?token=<token>&after=<id>──▶ relay (returns queued payloads)
9//! wallet ──GET  /pull?token=..&after=N&ack=N   ─▶ relay (deletes acked rows)
10//! ```
11//!
12//! # Token registration (simpler-but-safe design, documented trade-off)
13//!
14//! A token is auto-registered the first time a wallet pulls with it — wallets
15//! start polling at daemon boot, before any submit, so the token is always
16//! registered before Arcade can call back. Callbacks with unknown tokens are
17//! rejected (401) so the relay never stores unsolicited data. Tokens are
18//! high-entropy (32 hex), so pulling with a guessed token yields nothing
19//! (an empty, freshly-registered queue) and cannot read another wallet's
20//! payloads. Optionally, `POST /register {"token": "..."}` with
21//! `Authorization: Bearer <RELAY_ADMIN_TOKEN>` pre-registers tokens and
22//! `RELAY_REQUIRE_REGISTER=1` disables auto-registration entirely.
23//!
24//! # Endpoints
25//!
26//! | Route | Auth | Behavior |
27//! |-------|------|----------|
28//! | `GET /health` | none | `{"healthy":true}` |
29//! | `POST /arc-callback` | callback token (Bearer or `X-CallbackToken`) | queue payload for that token |
30//! | `GET /pull?token=&after=&ack=` | token in query | return payloads with `id > after` (max 100); delete `id <= ack` first |
31//! | `POST /register` | `RELAY_ADMIN_TOKEN` bearer | register `{"token": "..."}` |
32
33use axum::extract::{Query, State};
34use axum::http::{HeaderMap, StatusCode};
35use axum::response::{IntoResponse, Response};
36use axum::routing::{get, post};
37use axum::{Json, Router};
38use serde::Deserialize;
39use serde_json::json;
40use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
41use sqlx::SqlitePool;
42use std::sync::Arc;
43
44/// Shared relay state.
45#[derive(Clone)]
46pub struct RelayState {
47    /// Payload queue database.
48    pub pool: SqlitePool,
49    /// Admin bearer for `POST /register` (None disables the route).
50    pub admin_token: Option<Arc<String>>,
51    /// When false, `/pull` with an unknown token auto-registers it.
52    pub require_register: bool,
53}
54
55/// Open (and migrate) the relay sqlite database.
56pub async fn open_relay_db(path: &str) -> anyhow::Result<SqlitePool> {
57    let opts = SqliteConnectOptions::new()
58        .filename(path)
59        .create_if_missing(true);
60    let pool = SqlitePoolOptions::new()
61        .max_connections(5)
62        .connect_with(opts)
63        .await?;
64
65    sqlx::query(
66        r#"
67        CREATE TABLE IF NOT EXISTS tokens (
68            token       TEXT PRIMARY KEY,
69            created_at  TEXT NOT NULL
70        )
71        "#,
72    )
73    .execute(&pool)
74    .await?;
75
76    sqlx::query(
77        r#"
78        CREATE TABLE IF NOT EXISTS payloads (
79            id          INTEGER PRIMARY KEY AUTOINCREMENT,
80            token       TEXT NOT NULL,
81            received_at TEXT NOT NULL,
82            body        TEXT NOT NULL
83        )
84        "#,
85    )
86    .execute(&pool)
87    .await?;
88
89    sqlx::query("CREATE INDEX IF NOT EXISTS idx_payloads_token_id ON payloads (token, id)")
90        .execute(&pool)
91        .await?;
92
93    Ok(pool)
94}
95
96/// Build the relay router.
97pub fn make_relay_router(state: RelayState) -> Router {
98    Router::new()
99        .route("/health", get(health))
100        .route("/arc-callback", post(arc_callback))
101        .route("/pull", get(pull))
102        .route("/register", post(register))
103        .with_state(state)
104}
105
106async fn health() -> Json<serde_json::Value> {
107    Json(json!({"healthy": true, "service": "bsv-wallet-relay"}))
108}
109
110/// Extract the callback token from `Authorization: Bearer <tok>` (the ARC
111/// webhook convention) or an `X-CallbackToken` header (accepted for
112/// compatibility — implementations vary).
113fn extract_callback_token(headers: &HeaderMap) -> Option<String> {
114    if let Some(auth) = headers.get("authorization").and_then(|v| v.to_str().ok()) {
115        if let Some(tok) = auth.strip_prefix("Bearer ") {
116            if !tok.is_empty() {
117                return Some(tok.to_string());
118            }
119        }
120    }
121    headers
122        .get("x-callbacktoken")
123        .and_then(|v| v.to_str().ok())
124        .filter(|s| !s.is_empty())
125        .map(|s| s.to_string())
126}
127
128async fn token_registered(pool: &SqlitePool, token: &str) -> Result<bool, sqlx::Error> {
129    let row: Option<(String,)> = sqlx::query_as("SELECT token FROM tokens WHERE token = ?")
130        .bind(token)
131        .fetch_optional(pool)
132        .await?;
133    Ok(row.is_some())
134}
135
136async fn register_token(pool: &SqlitePool, token: &str) -> Result<(), sqlx::Error> {
137    sqlx::query("INSERT OR IGNORE INTO tokens (token, created_at) VALUES (?, ?)")
138        .bind(token)
139        .bind(chrono::Utc::now().to_rfc3339())
140        .execute(pool)
141        .await?;
142    Ok(())
143}
144
145fn err(status: StatusCode, code: &str, msg: &str) -> Response {
146    (status, Json(json!({"code": code, "message": msg}))).into_response()
147}
148
149/// `POST /arc-callback` — queue a callback payload for its token.
150async fn arc_callback(
151    State(state): State<RelayState>,
152    headers: HeaderMap,
153    Json(payload): Json<serde_json::Value>,
154) -> Response {
155    let Some(token) = extract_callback_token(&headers) else {
156        return err(
157            StatusCode::UNAUTHORIZED,
158            "NO_TOKEN",
159            "missing callback token",
160        );
161    };
162
163    match token_registered(&state.pool, &token).await {
164        Ok(true) => {}
165        Ok(false) => {
166            // Unknown token: never store unsolicited data.
167            return err(
168                StatusCode::UNAUTHORIZED,
169                "UNKNOWN_TOKEN",
170                "token not registered",
171            );
172        }
173        Err(e) => {
174            tracing::error!(error = %e, "relay: token lookup failed");
175            return err(
176                StatusCode::INTERNAL_SERVER_ERROR,
177                "DB_ERROR",
178                "storage error",
179            );
180        }
181    }
182
183    let body = payload.to_string();
184    if body.len() > 1_000_000 {
185        return err(
186            StatusCode::PAYLOAD_TOO_LARGE,
187            "TOO_LARGE",
188            "payload too large",
189        );
190    }
191
192    match sqlx::query("INSERT INTO payloads (token, received_at, body) VALUES (?, ?, ?)")
193        .bind(&token)
194        .bind(chrono::Utc::now().to_rfc3339())
195        .bind(&body)
196        .execute(&state.pool)
197        .await
198    {
199        Ok(_) => {
200            let txid = payload.get("txid").and_then(|v| v.as_str()).unwrap_or("?");
201            let status = payload
202                .get("txStatus")
203                .and_then(|v| v.as_str())
204                .unwrap_or("?");
205            tracing::info!(txid = %txid, tx_status = %status, "relay: payload queued");
206            (StatusCode::OK, Json(json!({"ok": true}))).into_response()
207        }
208        Err(e) => {
209            tracing::error!(error = %e, "relay: payload insert failed");
210            err(
211                StatusCode::INTERNAL_SERVER_ERROR,
212                "DB_ERROR",
213                "storage error",
214            )
215        }
216    }
217}
218
219#[derive(Debug, Deserialize)]
220struct PullQuery {
221    token: String,
222    /// Return payloads with `id > after` (default 0).
223    #[serde(default)]
224    after: i64,
225    /// Delete payloads with `id <= ack` before returning (optional).
226    ack: Option<i64>,
227}
228
229/// `GET /pull?token=<tok>&after=<id>[&ack=<id>]` — drain the queue.
230async fn pull(State(state): State<RelayState>, Query(q): Query<PullQuery>) -> Response {
231    if q.token.is_empty() {
232        return err(StatusCode::BAD_REQUEST, "NO_TOKEN", "token required");
233    }
234
235    // Auto-register on first pull (unless RELAY_REQUIRE_REGISTER).
236    match token_registered(&state.pool, &q.token).await {
237        Ok(true) => {}
238        Ok(false) if !state.require_register => {
239            if let Err(e) = register_token(&state.pool, &q.token).await {
240                tracing::error!(error = %e, "relay: auto-register failed");
241                return err(
242                    StatusCode::INTERNAL_SERVER_ERROR,
243                    "DB_ERROR",
244                    "storage error",
245                );
246            }
247            tracing::info!("relay: auto-registered new token on first pull");
248        }
249        Ok(false) => {
250            return err(
251                StatusCode::UNAUTHORIZED,
252                "UNKNOWN_TOKEN",
253                "token not registered",
254            );
255        }
256        Err(e) => {
257            tracing::error!(error = %e, "relay: token lookup failed");
258            return err(
259                StatusCode::INTERNAL_SERVER_ERROR,
260                "DB_ERROR",
261                "storage error",
262            );
263        }
264    }
265
266    if let Some(ack) = q.ack {
267        if let Err(e) = sqlx::query("DELETE FROM payloads WHERE token = ? AND id <= ?")
268            .bind(&q.token)
269            .bind(ack)
270            .execute(&state.pool)
271            .await
272        {
273            tracing::error!(error = %e, "relay: ack delete failed");
274            return err(
275                StatusCode::INTERNAL_SERVER_ERROR,
276                "DB_ERROR",
277                "storage error",
278            );
279        }
280    }
281
282    let rows: Result<Vec<(i64, String, String)>, sqlx::Error> = sqlx::query_as(
283        "SELECT id, received_at, body FROM payloads WHERE token = ? AND id > ? ORDER BY id LIMIT 100",
284    )
285    .bind(&q.token)
286    .bind(q.after)
287    .fetch_all(&state.pool)
288    .await;
289
290    match rows {
291        Ok(rows) => {
292            let items: Vec<serde_json::Value> = rows
293                .into_iter()
294                .map(|(id, received_at, body)| {
295                    let payload: serde_json::Value =
296                        serde_json::from_str(&body).unwrap_or(serde_json::Value::Null);
297                    json!({"id": id, "receivedAt": received_at, "payload": payload})
298                })
299                .collect();
300            (StatusCode::OK, Json(json!(items))).into_response()
301        }
302        Err(e) => {
303            tracing::error!(error = %e, "relay: pull query failed");
304            err(
305                StatusCode::INTERNAL_SERVER_ERROR,
306                "DB_ERROR",
307                "storage error",
308            )
309        }
310    }
311}
312
313#[derive(Debug, Deserialize)]
314struct RegisterBody {
315    token: String,
316}
317
318/// `POST /register {"token": "..."}` with the admin bearer.
319async fn register(
320    State(state): State<RelayState>,
321    headers: HeaderMap,
322    Json(body): Json<RegisterBody>,
323) -> Response {
324    let Some(admin) = state.admin_token.as_ref() else {
325        return err(
326            StatusCode::NOT_FOUND,
327            "REGISTER_DISABLED",
328            "set RELAY_ADMIN_TOKEN to enable /register",
329        );
330    };
331
332    let provided = headers
333        .get("authorization")
334        .and_then(|v| v.to_str().ok())
335        .and_then(|v| v.strip_prefix("Bearer "));
336    if provided != Some(admin.as_str()) {
337        return err(StatusCode::UNAUTHORIZED, "UNAUTHORIZED", "bad admin token");
338    }
339
340    if body.token.len() < 16 {
341        return err(
342            StatusCode::BAD_REQUEST,
343            "WEAK_TOKEN",
344            "token too short (min 16 chars)",
345        );
346    }
347
348    match register_token(&state.pool, &body.token).await {
349        Ok(()) => (StatusCode::OK, Json(json!({"ok": true}))).into_response(),
350        Err(e) => {
351            tracing::error!(error = %e, "relay: register failed");
352            err(
353                StatusCode::INTERNAL_SERVER_ERROR,
354                "DB_ERROR",
355                "storage error",
356            )
357        }
358    }
359}
360
361#[cfg(test)]
362mod tests {
363    use super::*;
364    use axum::body::Body;
365    use axum::http::Request;
366
367    async fn test_state(
368        require_register: bool,
369        admin: Option<&str>,
370    ) -> (RelayState, tempfile::TempDir) {
371        let dir = tempfile::tempdir().unwrap();
372        let db = dir.path().join("relay.db");
373        let pool = open_relay_db(db.to_str().unwrap()).await.unwrap();
374        (
375            RelayState {
376                pool,
377                admin_token: admin.map(|s| Arc::new(s.to_string())),
378                require_register,
379            },
380            dir,
381        )
382    }
383
384    async fn send(router: &Router, req: Request<Body>) -> (StatusCode, serde_json::Value) {
385        use tower::ServiceExt;
386        let resp = router.clone().oneshot(req).await.unwrap();
387        let status = resp.status();
388        let bytes = axum::body::to_bytes(resp.into_body(), 1_000_000)
389            .await
390            .unwrap();
391        let value: serde_json::Value =
392            serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null);
393        (status, value)
394    }
395
396    fn callback_req(token: &str, payload: &serde_json::Value) -> Request<Body> {
397        Request::builder()
398            .method("POST")
399            .uri("/arc-callback")
400            .header("authorization", format!("Bearer {}", token))
401            .header("content-type", "application/json")
402            .body(Body::from(payload.to_string()))
403            .unwrap()
404    }
405
406    fn pull_req(token: &str, after: i64, ack: Option<i64>) -> Request<Body> {
407        let mut uri = format!("/pull?token={}&after={}", token, after);
408        if let Some(a) = ack {
409            uri.push_str(&format!("&ack={}", a));
410        }
411        Request::builder()
412            .method("GET")
413            .uri(uri)
414            .body(Body::empty())
415            .unwrap()
416    }
417
418    #[tokio::test]
419    async fn health_ok() {
420        let (state, _dir) = test_state(false, None).await;
421        let router = make_relay_router(state);
422        let (status, body) = send(
423            &router,
424            Request::builder()
425                .method("GET")
426                .uri("/health")
427                .body(Body::empty())
428                .unwrap(),
429        )
430        .await;
431        assert_eq!(status, StatusCode::OK);
432        assert_eq!(body["healthy"], true);
433    }
434
435    #[tokio::test]
436    async fn queue_round_trip_with_ack() {
437        let (state, _dir) = test_state(false, None).await;
438        let router = make_relay_router(state);
439        let tok = "a1b2c3d4e5f60718293a4b5c6d7e8f90";
440
441        // 1. Wallet registers by pulling first (auto-register) — empty queue.
442        let (status, body) = send(&router, pull_req(tok, 0, None)).await;
443        assert_eq!(status, StatusCode::OK);
444        assert_eq!(body.as_array().unwrap().len(), 0);
445
446        // 2. Arcade posts two callbacks.
447        let p1 = serde_json::json!({"txid": "aa".repeat(32), "txStatus": "SEEN_ON_NETWORK"});
448        let p2 = serde_json::json!({"txid": "aa".repeat(32), "txStatus": "MINED",
449            "blockHeight": 850000, "blockHash": "bb".repeat(32), "merklePath": "deadbeef"});
450        let (s1, _) = send(&router, callback_req(tok, &p1)).await;
451        let (s2, _) = send(&router, callback_req(tok, &p2)).await;
452        assert_eq!(s1, StatusCode::OK);
453        assert_eq!(s2, StatusCode::OK);
454
455        // 3. Wallet drains the queue.
456        let (status, body) = send(&router, pull_req(tok, 0, None)).await;
457        assert_eq!(status, StatusCode::OK);
458        let items = body.as_array().unwrap().clone();
459        assert_eq!(items.len(), 2);
460        assert_eq!(items[0]["payload"]["txStatus"], "SEEN_ON_NETWORK");
461        assert_eq!(items[1]["payload"]["txStatus"], "MINED");
462        let last_id = items[1]["id"].as_i64().unwrap();
463
464        // 4. Ack deletes; nothing left after the acked id.
465        let (status, body) = send(&router, pull_req(tok, last_id, Some(last_id))).await;
466        assert_eq!(status, StatusCode::OK);
467        assert_eq!(body.as_array().unwrap().len(), 0);
468
469        // 5. Even pulling from 0 again returns nothing — rows deleted.
470        let (_, body) = send(&router, pull_req(tok, 0, None)).await;
471        assert_eq!(body.as_array().unwrap().len(), 0);
472    }
473
474    #[tokio::test]
475    async fn callback_with_unknown_token_rejected() {
476        let (state, _dir) = test_state(false, None).await;
477        let router = make_relay_router(state);
478
479        let p = serde_json::json!({"txid": "aa".repeat(32), "txStatus": "MINED"});
480        let (status, body) = send(&router, callback_req("never-registered-token", &p)).await;
481        assert_eq!(status, StatusCode::UNAUTHORIZED);
482        assert_eq!(body["code"], "UNKNOWN_TOKEN");
483    }
484
485    #[tokio::test]
486    async fn callback_without_token_rejected() {
487        let (state, _dir) = test_state(false, None).await;
488        let router = make_relay_router(state);
489
490        let req = Request::builder()
491            .method("POST")
492            .uri("/arc-callback")
493            .header("content-type", "application/json")
494            .body(Body::from("{}"))
495            .unwrap();
496        let (status, _) = send(&router, req).await;
497        assert_eq!(status, StatusCode::UNAUTHORIZED);
498    }
499
500    #[tokio::test]
501    async fn x_callback_token_header_accepted() {
502        let (state, _dir) = test_state(false, None).await;
503        let router = make_relay_router(state);
504        let tok = "ffeeddccbbaa99887766554433221100";
505
506        // register via pull
507        let _ = send(&router, pull_req(tok, 0, None)).await;
508
509        let p = serde_json::json!({"txid": "cc".repeat(32), "txStatus": "SEEN_ON_NETWORK"});
510        let req = Request::builder()
511            .method("POST")
512            .uri("/arc-callback")
513            .header("x-callbacktoken", tok)
514            .header("content-type", "application/json")
515            .body(Body::from(p.to_string()))
516            .unwrap();
517        let (status, _) = send(&router, req).await;
518        assert_eq!(status, StatusCode::OK);
519
520        let (_, body) = send(&router, pull_req(tok, 0, None)).await;
521        assert_eq!(body.as_array().unwrap().len(), 1);
522    }
523
524    #[tokio::test]
525    async fn tokens_are_isolated() {
526        let (state, _dir) = test_state(false, None).await;
527        let router = make_relay_router(state);
528        let tok_a = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
529        let tok_b = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
530        let _ = send(&router, pull_req(tok_a, 0, None)).await;
531        let _ = send(&router, pull_req(tok_b, 0, None)).await;
532
533        let p = serde_json::json!({"txid": "dd".repeat(32), "txStatus": "MINED"});
534        let _ = send(&router, callback_req(tok_a, &p)).await;
535
536        let (_, body_b) = send(&router, pull_req(tok_b, 0, None)).await;
537        assert_eq!(
538            body_b.as_array().unwrap().len(),
539            0,
540            "token B must not see token A's queue"
541        );
542        let (_, body_a) = send(&router, pull_req(tok_a, 0, None)).await;
543        assert_eq!(body_a.as_array().unwrap().len(), 1);
544    }
545
546    #[tokio::test]
547    async fn require_register_blocks_unknown_pull() {
548        let (state, _dir) = test_state(true, Some("admin-secret")).await;
549        let router = make_relay_router(state);
550        let tok = "1234567890abcdef1234567890abcdef";
551
552        let (status, _) = send(&router, pull_req(tok, 0, None)).await;
553        assert_eq!(status, StatusCode::UNAUTHORIZED);
554
555        // Admin registers the token.
556        let req = Request::builder()
557            .method("POST")
558            .uri("/register")
559            .header("authorization", "Bearer admin-secret")
560            .header("content-type", "application/json")
561            .body(Body::from(serde_json::json!({"token": tok}).to_string()))
562            .unwrap();
563        let (status, _) = send(&router, req).await;
564        assert_eq!(status, StatusCode::OK);
565
566        let (status, _) = send(&router, pull_req(tok, 0, None)).await;
567        assert_eq!(status, StatusCode::OK);
568    }
569
570    #[tokio::test]
571    async fn register_requires_admin_bearer() {
572        let (state, _dir) = test_state(true, Some("admin-secret")).await;
573        let router = make_relay_router(state);
574
575        let req = Request::builder()
576            .method("POST")
577            .uri("/register")
578            .header("authorization", "Bearer wrong")
579            .header("content-type", "application/json")
580            .body(Body::from(
581                serde_json::json!({"token": "1234567890abcdef"}).to_string(),
582            ))
583            .unwrap();
584        let (status, _) = send(&router, req).await;
585        assert_eq!(status, StatusCode::UNAUTHORIZED);
586    }
587}
bsv_wallet_cli/relay.rs

bsv_wallet_cli/
relay.rs
