bsv_wallet_cli/
arc_ingest.rs

1//! Shared ingestion of ARC/Arcade callback payloads.
2//!
3//! Both proof-delivery push paths converge here:
4//! - the daemon's own `POST /arc-callback` route (direct webhook), and
5//! - the `bsv-wallet-relay` poller (store-and-forward webhook).
6//!
7//! Payload shape (ARC webhook convention, Arcade V2 verified):
8//! `{ "txid", "txStatus", "blockHash"?, "blockHeight"?, "merklePath"?, ... }`
9//! — `merklePath` (BUMP hex) is present on `MINED`.
10//!
11//! With a merkle path present the proof is validated and stored via the
12//! toolbox's `StorageSqlx::ingest_merkle_proof` (ChainTracker validation →
13//! `proven_txs` insert → complete `proven_tx_reqs`/`transactions`). Status-only
14//! payloads map through the same status transitions as the Monitor's SSE task.
15
16use anyhow::{anyhow, Result};
17use bsv_wallet_toolbox::monitor::ArcadeEventsTask;
18use bsv_wallet_toolbox::{MonitorStorage, ProofIngestOutcome, StorageSqlx};
19use std::sync::atomic::AtomicBool;
20
21/// What ingesting a callback payload did.
22#[derive(Debug, Clone, PartialEq, Eq)]
23pub enum IngestAction {
24    /// A merkle proof was validated and stored; records completed.
25    ProofIngested,
26    /// The proof was rejected (invalid root / unparseable) — NOT stored.
27    ProofRejected(String),
28    /// A status-only update was applied to storage.
29    StatusApplied,
30    /// A status-only update matched no records (unknown txid or already final).
31    StatusIgnored,
32}
33
34/// Parse and apply one ARC/Arcade callback payload against wallet storage.
35pub async fn ingest_arc_payload(
36    storage: &StorageSqlx,
37    payload: &serde_json::Value,
38) -> Result<IngestAction> {
39    let txid = payload
40        .get("txid")
41        .and_then(|v| v.as_str())
42        .ok_or_else(|| anyhow!("payload missing txid"))?;
43    if txid.len() != 64 || !txid.chars().all(|c| c.is_ascii_hexdigit()) {
44        return Err(anyhow!("invalid txid"));
45    }
46    let tx_status = payload
47        .get("txStatus")
48        .and_then(|v| v.as_str())
49        .unwrap_or("");
50
51    let merkle_path_hex = payload
52        .get("merklePath")
53        .and_then(|v| v.as_str())
54        .filter(|s| !s.is_empty());
55
56    if let Some(mp_hex) = merkle_path_hex {
57        let merkle_path = hex::decode(mp_hex).map_err(|e| anyhow!("merklePath not hex: {}", e))?;
58        let block_height = payload
59            .get("blockHeight")
60            .and_then(|v| v.as_u64())
61            .ok_or_else(|| anyhow!("merklePath payload missing blockHeight"))?
62            as u32;
63        let block_hash = payload
64            .get("blockHash")
65            .and_then(|v| v.as_str())
66            .unwrap_or_default();
67
68        // Ensure spendability transition happened even if we never saw
69        // SEEN_ON_NETWORK (webhook may be the only delivery path).
70        let _ = storage.mark_transaction_seen_on_network(txid).await;
71
72        match storage
73            .ingest_merkle_proof(txid, &merkle_path, block_height, block_hash, None)
74            .await
75            .map_err(|e| anyhow!("ingest_merkle_proof: {}", e))?
76        {
77            ProofIngestOutcome::Ingested(status) => {
78                tracing::info!(
79                    txid = %txid,
80                    block_height = ?status.block_height,
81                    "arc-callback: merkle proof ingested"
82                );
83                Ok(IngestAction::ProofIngested)
84            }
85            ProofIngestOutcome::InvalidMerkleRoot { computed_root } => {
86                tracing::warn!(txid = %txid, computed_root = %computed_root, "arc-callback: proof rejected (invalid merkle root)");
87                Ok(IngestAction::ProofRejected("invalid merkle root".into()))
88            }
89            ProofIngestOutcome::InvalidProof(e) => {
90                tracing::warn!(txid = %txid, error = %e, "arc-callback: proof rejected (unparseable)");
91                Ok(IngestAction::ProofRejected(e))
92            }
93            ProofIngestOutcome::TrackerError(e) => {
94                tracing::warn!(txid = %txid, error = %e, "arc-callback: proof deferred (ChainTracker error) — polling sync will retry");
95                Ok(IngestAction::ProofRejected(format!("tracker error: {}", e)))
96            }
97        }
98    } else {
99        // Status-only payload — identical mapping to the Monitor's SSE task.
100        // Since arcade v0.10.1 (#259) MINED webhooks DO carry merklePath and
101        // take the proof branch above; a status-only MINED still lands here
102        // when upstream enrichment best-effort-misses (BUMP-cache eviction /
103        // reorg race), and the proof is then fetched by the SSE task's
104        // trigger / the polling backstop.
105        let trigger = AtomicBool::new(false);
106        let updated =
107            ArcadeEventsTask::<StorageSqlx>::apply_status_event(storage, txid, tx_status, &trigger)
108                .await
109                .map_err(|e| anyhow!("apply_status_event: {}", e))?;
110        tracing::info!(
111            txid = %txid,
112            status = %tx_status,
113            updated,
114            "arc-callback: status webhook received"
115        );
116        if updated {
117            Ok(IngestAction::StatusApplied)
118        } else {
119            Ok(IngestAction::StatusIgnored)
120        }
121    }
122}
bsv_wallet_cli/arc_ingest.rs

bsv_wallet_cli/
arc_ingest.rs
