refactor: split package.rs, mod.rs, listener.rs, and lnd.rs into focused submodules
- R35: Split package.rs (1794 lines) into package/{mod,config,validation,lifecycle}.rs
- R36: Split mesh/listener.rs (1799 lines) into listener/{mod,session,frames,decode,dispatch,bitcoin}.rs
- R37: Split rpc/mod.rs into mod.rs + dispatcher.rs, middleware.rs, response.rs (54% reduction)
- R38: Split lnd.rs (1064 lines) into lnd/{mod,info,channels,wallet,payments}.rs
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dorian
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
8e4d352393
commit
77f550fb5e
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,374 @@
|
||||
//! Bitcoin relay operations: TX broadcast, confirmation tracking, peer messaging.
|
||||
|
||||
use super::MeshCommand;
|
||||
use super::MeshState;
|
||||
use super::super::crypto;
|
||||
use super::super::message_types;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use tracing::{debug, info, warn};
|
||||
|
||||
/// Called on an internet-connected node when it receives a TxRelay request.
|
||||
/// Broadcasts the raw TX to Bitcoin via RPC, sends the txid back, then
|
||||
/// monitors for 3 confirmations and sends updates back via mesh.
|
||||
pub(super) async fn handle_tx_relay_broadcast(
|
||||
relay: message_types::TxRelayPayload,
|
||||
sender_contact_id: u32,
|
||||
state: &Arc<MeshState>,
|
||||
) {
|
||||
let client = match reqwest::Client::builder()
|
||||
.timeout(std::time::Duration::from_secs(30))
|
||||
.build()
|
||||
{
|
||||
Ok(c) => c,
|
||||
Err(e) => {
|
||||
warn!("Failed to create HTTP client for TX relay: {}", e);
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let (rpc_user, rpc_pass) = crate::bitcoin_rpc::bitcoin_rpc_credentials().await;
|
||||
|
||||
// Pre-flight: check if Bitcoin Core is reachable and synced
|
||||
if !preflight_check(&client, &rpc_user, &rpc_pass, &relay, sender_contact_id, state).await {
|
||||
return;
|
||||
}
|
||||
|
||||
// Step 1: Broadcast via Bitcoin Core RPC sendrawtransaction
|
||||
let txid = match broadcast_transaction(&client, &rpc_user, &rpc_pass, &relay, sender_contact_id, state).await {
|
||||
Some(id) => id,
|
||||
None => return,
|
||||
};
|
||||
|
||||
info!(request_id = relay.request_id, txid = %txid, "TX broadcast successful — tracking confirmations");
|
||||
|
||||
// Step 2: Send TxRelayResponse with txid back to originator
|
||||
send_tx_relay_response(state, sender_contact_id, relay.request_id, Some(&txid), None, None).await;
|
||||
|
||||
// Step 3: Monitor confirmations (poll every 30s, up to 3 hours)
|
||||
track_confirmations(&client, &txid, relay.request_id, sender_contact_id, state).await;
|
||||
}
|
||||
|
||||
/// Pre-flight check: verify Bitcoin Core is reachable and synced.
|
||||
/// Returns `true` if the node is ready, `false` if an error response was sent.
|
||||
async fn preflight_check(
|
||||
client: &reqwest::Client,
|
||||
rpc_user: &str,
|
||||
rpc_pass: &str,
|
||||
relay: &message_types::TxRelayPayload,
|
||||
sender_contact_id: u32,
|
||||
state: &Arc<MeshState>,
|
||||
) -> bool {
|
||||
let preflight_body = serde_json::json!({
|
||||
"jsonrpc": "1.0",
|
||||
"id": "preflight",
|
||||
"method": "getblockchaininfo",
|
||||
"params": []
|
||||
});
|
||||
|
||||
match client
|
||||
.post(crate::constants::BITCOIN_RPC_URL)
|
||||
.basic_auth(rpc_user, Some(rpc_pass))
|
||||
.json(&preflight_body)
|
||||
.send()
|
||||
.await
|
||||
{
|
||||
Ok(resp) => {
|
||||
if let Ok(rpc_resp) = resp.json::<serde_json::Value>().await {
|
||||
if let Some(result) = rpc_resp.get("result") {
|
||||
let ibd = result.get("initialblockdownload")
|
||||
.and_then(|v| v.as_bool())
|
||||
.unwrap_or(false);
|
||||
let progress = result.get("verificationprogress")
|
||||
.and_then(|v| v.as_f64())
|
||||
.unwrap_or(0.0);
|
||||
if ibd || progress < 0.999 {
|
||||
let pct = (progress * 100.0) as u32;
|
||||
let msg = format!("Bitcoin node is syncing ({}%) — cannot broadcast yet", pct);
|
||||
warn!(request_id = relay.request_id, "{}", msg);
|
||||
send_tx_relay_response(state, sender_contact_id, relay.request_id, None, Some(&msg), Some("bitcoin_syncing")).await;
|
||||
return false;
|
||||
}
|
||||
} else if let Some(err) = rpc_resp.get("error").and_then(|e| e.as_object()) {
|
||||
let msg = err.get("message").and_then(|m| m.as_str()).unwrap_or("RPC error");
|
||||
warn!(request_id = relay.request_id, "Bitcoin pre-flight failed: {}", msg);
|
||||
send_tx_relay_response(state, sender_contact_id, relay.request_id, None, Some(&format!("Bitcoin node error: {}", msg)), Some("bitcoin_unreachable")).await;
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
let msg = format!("Bitcoin node unreachable — {}", if e.is_connect() {
|
||||
"connection refused (node may be stopped)"
|
||||
} else if e.is_timeout() {
|
||||
"connection timed out"
|
||||
} else {
|
||||
"network error"
|
||||
});
|
||||
warn!(request_id = relay.request_id, "Pre-flight: {}: {}", msg, e);
|
||||
send_tx_relay_response(state, sender_contact_id, relay.request_id, None, Some(&msg), Some("bitcoin_unreachable")).await;
|
||||
return false;
|
||||
}
|
||||
}
|
||||
true
|
||||
}
|
||||
|
||||
/// Broadcast a raw transaction via Bitcoin Core RPC.
|
||||
/// Returns the txid on success, or None if an error response was sent.
|
||||
async fn broadcast_transaction(
|
||||
client: &reqwest::Client,
|
||||
rpc_user: &str,
|
||||
rpc_pass: &str,
|
||||
relay: &message_types::TxRelayPayload,
|
||||
sender_contact_id: u32,
|
||||
state: &Arc<MeshState>,
|
||||
) -> Option<String> {
|
||||
let body = serde_json::json!({
|
||||
"jsonrpc": "1.0",
|
||||
"id": "mesh-relay",
|
||||
"method": "sendrawtransaction",
|
||||
"params": [relay.tx_hex]
|
||||
});
|
||||
|
||||
let txid = match client
|
||||
.post(crate::constants::BITCOIN_RPC_URL)
|
||||
.basic_auth(rpc_user, Some(rpc_pass))
|
||||
.json(&body)
|
||||
.send()
|
||||
.await
|
||||
{
|
||||
Ok(resp) => {
|
||||
match resp.json::<serde_json::Value>().await {
|
||||
Ok(rpc_resp) => {
|
||||
if let Some(err) = rpc_resp.get("error").and_then(|e| e.as_object()) {
|
||||
let code = err.get("code").and_then(|c| c.as_i64()).unwrap_or(0);
|
||||
let msg = err.get("message").and_then(|m| m.as_str()).unwrap_or("unknown");
|
||||
let user_msg = match code {
|
||||
-25 => format!("TX already in mempool or confirmed: {}", msg),
|
||||
-26 => format!("TX rejected by mempool policy: {}", msg),
|
||||
-27 => format!("TX already confirmed in a block"),
|
||||
_ => format!("Bitcoin rejected TX (code {}): {}", code, msg),
|
||||
};
|
||||
warn!(request_id = relay.request_id, rpc_code = code, "sendrawtransaction: {}", msg);
|
||||
send_tx_relay_response(state, sender_contact_id, relay.request_id, None, Some(&user_msg), Some(&format!("tx_rejected:{}", code))).await;
|
||||
return None;
|
||||
}
|
||||
rpc_resp.get("result").and_then(|r| r.as_str()).map(|s| s.to_string())
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("Failed to parse Bitcoin RPC response: {}", e);
|
||||
send_tx_relay_response(state, sender_contact_id, relay.request_id, None, Some("Failed to parse Bitcoin node response"), Some("rpc_parse_error")).await;
|
||||
return None;
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
let msg = format!("Bitcoin node unreachable during broadcast — {}", if e.is_connect() {
|
||||
"connection refused"
|
||||
} else if e.is_timeout() {
|
||||
"timed out"
|
||||
} else {
|
||||
"network error"
|
||||
});
|
||||
warn!("Bitcoin Core RPC unreachable: {}", e);
|
||||
send_tx_relay_response(state, sender_contact_id, relay.request_id, None, Some(&msg), Some("bitcoin_unreachable")).await;
|
||||
return None;
|
||||
}
|
||||
};
|
||||
|
||||
if txid.is_none() {
|
||||
send_tx_relay_response(state, sender_contact_id, relay.request_id, None, Some("Bitcoin node returned no transaction ID"), Some("rpc_parse_error")).await;
|
||||
}
|
||||
txid
|
||||
}
|
||||
|
||||
/// Monitor a transaction for confirmations (poll every 30s, up to 3 hours).
|
||||
async fn track_confirmations(
|
||||
client: &reqwest::Client,
|
||||
txid: &str,
|
||||
request_id: u64,
|
||||
sender_contact_id: u32,
|
||||
state: &Arc<MeshState>,
|
||||
) {
|
||||
let mut last_reported_confs: u32 = 0;
|
||||
for _ in 0..360 {
|
||||
tokio::time::sleep(Duration::from_secs(30)).await;
|
||||
|
||||
match check_tx_confirmations(client, txid).await {
|
||||
Ok((confs, block_height)) => {
|
||||
if confs > last_reported_confs && confs <= 3 {
|
||||
info!(txid = %txid, confirmations = confs, "Sending confirmation update via mesh");
|
||||
send_confirmation_update(state, sender_contact_id, request_id, txid, confs, block_height).await;
|
||||
last_reported_confs = confs;
|
||||
if confs >= 3 {
|
||||
info!(txid = %txid, "TX fully confirmed (3/3) — done tracking");
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
debug!(txid = %txid, "Confirmation check: {}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Send a TxRelayResponse back to the originating peer.
|
||||
async fn send_tx_relay_response(
|
||||
state: &Arc<MeshState>,
|
||||
dest_contact_id: u32,
|
||||
request_id: u64,
|
||||
txid: Option<&str>,
|
||||
error: Option<&str>,
|
||||
error_code: Option<&str>,
|
||||
) {
|
||||
let wire = match super::super::bitcoin_relay::build_tx_relay_response(request_id, txid, error, error_code) {
|
||||
Ok(w) => w,
|
||||
Err(e) => {
|
||||
warn!("Failed to build TX relay response: {}", e);
|
||||
return;
|
||||
}
|
||||
};
|
||||
send_to_peer(state, dest_contact_id, wire).await;
|
||||
}
|
||||
|
||||
/// Send a TxConfirmation update to the originator.
|
||||
async fn send_confirmation_update(
|
||||
state: &Arc<MeshState>,
|
||||
dest_contact_id: u32,
|
||||
request_id: u64,
|
||||
txid: &str,
|
||||
confirmations: u32,
|
||||
block_height: u64,
|
||||
) {
|
||||
let conf = message_types::TxConfirmationPayload {
|
||||
request_id,
|
||||
txid: txid.to_string(),
|
||||
confirmations,
|
||||
block_height,
|
||||
};
|
||||
if let Ok(payload_bytes) = message_types::encode_payload(&conf) {
|
||||
let envelope = message_types::TypedEnvelope::new(
|
||||
message_types::MeshMessageType::TxConfirmation,
|
||||
payload_bytes,
|
||||
);
|
||||
if let Ok(wire) = envelope.to_wire() {
|
||||
send_to_peer(state, dest_contact_id, wire).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Encrypt a typed wire payload for a specific peer.
|
||||
/// Attempts ratchet encryption first (forward secrecy), falls back to static
|
||||
/// shared secret, falls back to plaintext if neither is available.
|
||||
/// Respects the encrypt_relay config toggle for rollback.
|
||||
async fn encrypt_for_peer(
|
||||
state: &Arc<MeshState>,
|
||||
contact_id: u32,
|
||||
typed_wire: &[u8],
|
||||
) -> Vec<u8> {
|
||||
if !state.encrypt_relay {
|
||||
return typed_wire.to_vec();
|
||||
}
|
||||
|
||||
// Look up peer DID for ratchet session
|
||||
let peer_did = state.peers.read().await
|
||||
.get(&contact_id)
|
||||
.and_then(|p| p.did.clone());
|
||||
|
||||
// Try ratchet encryption first (forward secrecy)
|
||||
if let Some(ref did) = peer_did {
|
||||
if state.session_manager.has_session(did).await {
|
||||
match state.session_manager.encrypt_for_peer(did, typed_wire).await {
|
||||
Ok(ratchet_msg) => {
|
||||
let ratchet_bytes = ratchet_msg.to_bytes();
|
||||
let mut buf = Vec::with_capacity(1 + ratchet_bytes.len());
|
||||
buf.push(message_types::RATCHET_TYPED_MARKER);
|
||||
buf.extend_from_slice(&ratchet_bytes);
|
||||
debug!(contact_id, did = %did, "Encrypted with Double Ratchet (0xDD)");
|
||||
return buf;
|
||||
}
|
||||
Err(e) => {
|
||||
warn!(contact_id, did = %did, "Ratchet encrypt failed, trying static: {}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Fall back to static shared secret (0xEE)
|
||||
let secrets = state.shared_secrets.read().await;
|
||||
if let Some(secret) = secrets.get(&contact_id) {
|
||||
match crypto::encrypt(secret, typed_wire) {
|
||||
Ok(ciphertext) => {
|
||||
let mut buf = Vec::with_capacity(1 + ciphertext.len());
|
||||
buf.push(message_types::ENCRYPTED_TYPED_MARKER);
|
||||
buf.extend_from_slice(&ciphertext);
|
||||
debug!(contact_id, "Encrypted with static shared secret (0xEE)");
|
||||
return buf;
|
||||
}
|
||||
Err(e) => {
|
||||
warn!(contact_id, "Static encrypt failed, sending plaintext: {}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// No encryption available — send plaintext
|
||||
debug!(contact_id, "No encryption available, sending plaintext (0x02)");
|
||||
typed_wire.to_vec()
|
||||
}
|
||||
|
||||
/// Send raw wire bytes to a specific peer by contact_id.
|
||||
/// Encrypts directed messages via ratchet or shared secret when available.
|
||||
/// Falls back to channel 0 broadcast (plaintext) if peer's pubkey is unknown.
|
||||
async fn send_to_peer(state: &Arc<MeshState>, contact_id: u32, typed_wire: Vec<u8>) {
|
||||
let peers = state.peers.read().await;
|
||||
if let Some(peer) = peers.get(&contact_id) {
|
||||
if let Some(ref pk) = peer.pubkey_hex {
|
||||
if let Ok(pk_bytes) = hex::decode(pk) {
|
||||
if pk_bytes.len() >= 6 {
|
||||
let mut prefix = [0u8; 6];
|
||||
prefix.copy_from_slice(&pk_bytes[..6]);
|
||||
drop(peers);
|
||||
// Encrypt for this specific peer before sending
|
||||
let payload = encrypt_for_peer(state, contact_id, &typed_wire).await;
|
||||
let _ = state.cmd_tx.send(MeshCommand::SendRaw {
|
||||
dest_pubkey_prefix: prefix,
|
||||
payload,
|
||||
}).await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
drop(peers);
|
||||
// Broadcast fallback — plaintext (no specific peer to encrypt for)
|
||||
let _ = state.cmd_tx.send(MeshCommand::BroadcastChannel {
|
||||
channel: 0,
|
||||
payload: typed_wire,
|
||||
}).await;
|
||||
}
|
||||
|
||||
/// Check transaction confirmation count via Bitcoin Core RPC.
|
||||
async fn check_tx_confirmations(client: &reqwest::Client, txid: &str) -> anyhow::Result<(u32, u64)> {
|
||||
let body = serde_json::json!({
|
||||
"jsonrpc": "1.0",
|
||||
"id": "mesh-conf",
|
||||
"method": "gettransaction",
|
||||
"params": [txid]
|
||||
});
|
||||
let (rpc_user, rpc_pass) = crate::bitcoin_rpc::bitcoin_rpc_credentials().await;
|
||||
let resp = client
|
||||
.post(crate::constants::BITCOIN_RPC_URL)
|
||||
.basic_auth(&rpc_user, Some(&rpc_pass))
|
||||
.json(&body)
|
||||
.send()
|
||||
.await?;
|
||||
let rpc_resp: serde_json::Value = resp.json().await?;
|
||||
if let Some(result) = rpc_resp.get("result") {
|
||||
let confs = result.get("confirmations").and_then(|c| c.as_u64()).unwrap_or(0) as u32;
|
||||
let block_height = result.get("blockheight").and_then(|h| h.as_u64()).unwrap_or(0);
|
||||
Ok((confs, block_height))
|
||||
} else {
|
||||
anyhow::bail!("gettransaction returned no result")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,457 @@
|
||||
//! Message decoding: base64, encryption, chunk reassembly, peer resolution.
|
||||
|
||||
use super::MeshState;
|
||||
use super::super::crypto;
|
||||
use super::super::message_types::{self, TypedEnvelope};
|
||||
use super::super::types::*;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
use tracing::{debug, info, warn};
|
||||
|
||||
/// Try to base64-decode payload and check if the result is a typed envelope.
|
||||
/// Handles: plain typed (0x02), steganographic (0xAA), and encrypted (0xEE).
|
||||
/// Returns the decoded bytes if it's a valid base64-encoded TypedEnvelope.
|
||||
pub(super) fn try_base64_typed(payload: &[u8]) -> Option<Vec<u8>> {
|
||||
use base64::Engine;
|
||||
if payload.is_empty() || payload[0] == message_types::TYPED_MESSAGE_MARKER {
|
||||
return None;
|
||||
}
|
||||
let text = std::str::from_utf8(payload).ok()?;
|
||||
let decoded = base64::engine::general_purpose::STANDARD.decode(text.trim()).ok()?;
|
||||
unwrap_wire_layers(&decoded)
|
||||
}
|
||||
|
||||
/// Try to base64-decode and decrypt an encrypted typed message.
|
||||
/// Handles the common case where encrypted messages arrive as base64 text.
|
||||
pub(super) async fn try_decrypt_base64(
|
||||
payload: &[u8],
|
||||
sender_contact_id: u32,
|
||||
state: &Arc<MeshState>,
|
||||
) -> Option<Vec<u8>> {
|
||||
use base64::Engine;
|
||||
let text = std::str::from_utf8(payload).ok()?;
|
||||
let decoded = base64::engine::general_purpose::STANDARD.decode(text.trim()).ok()?;
|
||||
if decoded.first() != Some(&message_types::ENCRYPTED_TYPED_MARKER) {
|
||||
return None;
|
||||
}
|
||||
let secrets = state.shared_secrets.read().await;
|
||||
try_decrypt_typed(&decoded, sender_contact_id, &secrets)
|
||||
}
|
||||
|
||||
/// Try to decrypt a Double Ratchet encrypted message (0xDD prefix).
|
||||
/// Format: [0xDD] [RatchetHeader(40) + nonce(12) + ciphertext + tag(16)]
|
||||
/// Returns the decrypted typed wire bytes ([0x02][CBOR]) if successful.
|
||||
async fn try_decrypt_ratchet(
|
||||
decoded: &[u8],
|
||||
sender_contact_id: u32,
|
||||
state: &Arc<MeshState>,
|
||||
) -> Option<Vec<u8>> {
|
||||
if decoded.first() != Some(&message_types::RATCHET_TYPED_MARKER) {
|
||||
return None;
|
||||
}
|
||||
let ratchet_bytes = &decoded[1..]; // skip 0xDD marker
|
||||
|
||||
let ratchet_msg = match super::super::ratchet::RatchetMessage::from_bytes(ratchet_bytes) {
|
||||
Ok(msg) => msg,
|
||||
Err(e) => {
|
||||
warn!(contact_id = sender_contact_id, "Failed to parse ratchet message: {}", e);
|
||||
return None;
|
||||
}
|
||||
};
|
||||
|
||||
// Look up peer DID for session manager
|
||||
let peer_did = state.peers.read().await
|
||||
.get(&sender_contact_id)
|
||||
.and_then(|p| p.did.clone())?;
|
||||
|
||||
match state.session_manager.decrypt_from_peer(&peer_did, &ratchet_msg).await {
|
||||
Ok(plaintext) => {
|
||||
debug!(contact_id = sender_contact_id, did = %peer_did, "Decrypted ratchet message (0xDD)");
|
||||
// The plaintext should be the original [0x02][CBOR] typed wire
|
||||
if TypedEnvelope::is_typed(&plaintext) {
|
||||
Some(plaintext)
|
||||
} else {
|
||||
// Could be nested stego -> typed
|
||||
unwrap_wire_layers(&plaintext)
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
warn!(contact_id = sender_contact_id, "Ratchet decrypt failed: {}", e);
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Try to base64-decode and decrypt a ratchet-encrypted message.
|
||||
/// Handles the case where ratchet messages arrive as base64 text.
|
||||
pub(super) async fn try_decrypt_ratchet_base64(
|
||||
payload: &[u8],
|
||||
sender_contact_id: u32,
|
||||
state: &Arc<MeshState>,
|
||||
) -> Option<Vec<u8>> {
|
||||
use base64::Engine;
|
||||
let text = std::str::from_utf8(payload).ok()?;
|
||||
let decoded = base64::engine::general_purpose::STANDARD.decode(text.trim()).ok()?;
|
||||
if decoded.first() != Some(&message_types::RATCHET_TYPED_MARKER) {
|
||||
return None;
|
||||
}
|
||||
try_decrypt_ratchet(&decoded, sender_contact_id, state).await
|
||||
}
|
||||
|
||||
/// Unwrap wire layers: encrypted (0xEE) -> stego (0xAA) -> typed (0x02).
|
||||
/// Returns None if decoding fails at any layer (caller should use shared_secrets variant).
|
||||
fn unwrap_wire_layers(decoded: &[u8]) -> Option<Vec<u8>> {
|
||||
// Check for steganographic frame (0xAA prefix) — unwrap to typed envelope
|
||||
if decoded.first() == Some(&super::super::steganography::STEGO_MARKER) {
|
||||
match super::super::steganography::decode_typed_wire(decoded) {
|
||||
Ok(typed_wire) => return Some(typed_wire),
|
||||
Err(_) => return None,
|
||||
}
|
||||
}
|
||||
if TypedEnvelope::is_typed(decoded) {
|
||||
Some(decoded.to_vec())
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
/// Try to decrypt an encrypted typed message (0xEE prefix) using known shared secrets.
|
||||
/// Format: [0xEE] [nonce: 12] [ciphertext + tag: 16]
|
||||
fn try_decrypt_typed(
|
||||
decoded: &[u8],
|
||||
sender_contact_id: u32,
|
||||
shared_secrets: &HashMap<u32, [u8; 32]>,
|
||||
) -> Option<Vec<u8>> {
|
||||
if decoded.first() != Some(&message_types::ENCRYPTED_TYPED_MARKER) {
|
||||
return None;
|
||||
}
|
||||
let ciphertext = &decoded[1..]; // skip 0xEE marker
|
||||
|
||||
// Try sender's shared secret first (most likely)
|
||||
if let Some(secret) = shared_secrets.get(&sender_contact_id) {
|
||||
if let Ok(plaintext) = crypto::decrypt(secret, ciphertext) {
|
||||
return unwrap_wire_layers(&plaintext);
|
||||
}
|
||||
}
|
||||
|
||||
// Fallback: try all known shared secrets (in case contact_id mapping is stale)
|
||||
for (cid, secret) in shared_secrets {
|
||||
if *cid == sender_contact_id { continue; } // already tried
|
||||
if let Ok(plaintext) = crypto::decrypt(secret, ciphertext) {
|
||||
return unwrap_wire_layers(&plaintext);
|
||||
}
|
||||
}
|
||||
|
||||
None
|
||||
}
|
||||
|
||||
/// Check if payload is a mesh chunk ("MC" prefix) and try to reassemble.
|
||||
/// Format: MC{msg_id:2hex}{chunk_idx:2hex}{total:2hex}{base64_data}
|
||||
/// Returns Some(decoded_bytes) when all chunks have arrived.
|
||||
pub(super) async fn try_chunk_reassemble(
|
||||
payload: &[u8],
|
||||
sender_contact_id: u32,
|
||||
state: &Arc<MeshState>,
|
||||
) -> Option<Vec<u8>> {
|
||||
use base64::Engine;
|
||||
let text = std::str::from_utf8(payload).ok()?;
|
||||
if !text.starts_with("MC") || text.len() < 8 {
|
||||
return None;
|
||||
}
|
||||
|
||||
let msg_id = u8::from_str_radix(&text[2..4], 16).ok()?;
|
||||
let chunk_idx = u8::from_str_radix(&text[4..6], 16).ok()?;
|
||||
let total = u8::from_str_radix(&text[6..8], 16).ok()?;
|
||||
let chunk_data = &text[8..];
|
||||
|
||||
if total == 0 || total > 20 {
|
||||
return None; // sanity check
|
||||
}
|
||||
|
||||
let key = (sender_contact_id, msg_id);
|
||||
let mut buffer = state.chunk_buffer.write().await;
|
||||
|
||||
// Clean up stale entries (>120s old)
|
||||
buffer.retain(|_, v| v.created.elapsed().as_secs() < 120);
|
||||
|
||||
let assembly = buffer.entry(key).or_insert_with(|| super::ChunkAssembly {
|
||||
chunks: HashMap::new(),
|
||||
total,
|
||||
created: std::time::Instant::now(),
|
||||
});
|
||||
|
||||
assembly.chunks.insert(chunk_idx, chunk_data.to_string());
|
||||
assembly.total = total; // update in case first chunk had it wrong
|
||||
|
||||
debug!(msg_id, chunk_idx, total, received = assembly.chunks.len(), "Chunk received");
|
||||
|
||||
// Check if we have all chunks
|
||||
if assembly.chunks.len() < total as usize {
|
||||
return None;
|
||||
}
|
||||
|
||||
// All chunks received — reassemble in order
|
||||
let mut combined = String::new();
|
||||
for i in 0..total {
|
||||
match assembly.chunks.get(&i) {
|
||||
Some(data) => combined.push_str(data),
|
||||
None => {
|
||||
warn!(msg_id, missing = i, "Chunk missing during reassembly");
|
||||
return None;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if let Ok(decoded) = base64::engine::general_purpose::STANDARD.decode(&combined) {
|
||||
// Check for ratchet-encrypted frame (0xDD) — decrypt then unwrap
|
||||
if decoded.first() == Some(&message_types::RATCHET_TYPED_MARKER) {
|
||||
// Must drop buffer lock before calling async try_decrypt_ratchet
|
||||
let decoded_clone = decoded.clone();
|
||||
drop(buffer);
|
||||
if let Some(typed_wire) = try_decrypt_ratchet(&decoded_clone, sender_contact_id, state).await {
|
||||
info!(msg_id, chunks = total, total_len = typed_wire.len(), "Reassembled ratchet-encrypted chunked message");
|
||||
state.chunk_buffer.write().await.remove(&key);
|
||||
return Some(typed_wire);
|
||||
}
|
||||
buffer = state.chunk_buffer.write().await;
|
||||
}
|
||||
// Check for static-encrypted frame (0xEE) — decrypt then unwrap
|
||||
if decoded.first() == Some(&message_types::ENCRYPTED_TYPED_MARKER) {
|
||||
let secrets = state.shared_secrets.read().await;
|
||||
if let Some(typed_wire) = try_decrypt_typed(&decoded, sender_contact_id, &secrets) {
|
||||
info!(msg_id, chunks = total, total_len = typed_wire.len(), "Reassembled encrypted chunked message");
|
||||
buffer.remove(&key);
|
||||
return Some(typed_wire);
|
||||
}
|
||||
}
|
||||
// Check for stego frame — unwrap to typed envelope
|
||||
if decoded.first() == Some(&super::super::steganography::STEGO_MARKER) {
|
||||
if let Ok(typed_wire) = super::super::steganography::decode_typed_wire(&decoded) {
|
||||
info!(msg_id, chunks = total, total_len = typed_wire.len(), "Reassembled stego chunked message");
|
||||
buffer.remove(&key);
|
||||
return Some(typed_wire);
|
||||
}
|
||||
}
|
||||
if TypedEnvelope::is_typed(&decoded) {
|
||||
info!(msg_id, chunks = total, total_len = decoded.len(), "Reassembled chunked message");
|
||||
buffer.remove(&key);
|
||||
return Some(decoded);
|
||||
}
|
||||
}
|
||||
|
||||
warn!(msg_id, "All chunks received but decode failed");
|
||||
buffer.remove(&key);
|
||||
None
|
||||
}
|
||||
|
||||
/// Look up a peer by pubkey hex prefix. Returns (contact_id, display_name).
|
||||
pub(super) async fn resolve_peer(state: &Arc<MeshState>, sender_prefix: &str) -> (u32, String) {
|
||||
let peers = state.peers.read().await;
|
||||
peers
|
||||
.values()
|
||||
.find(|p| {
|
||||
p.pubkey_hex
|
||||
.as_ref()
|
||||
.map(|k| k.starts_with(sender_prefix))
|
||||
.unwrap_or(false)
|
||||
})
|
||||
.map(|p| (p.contact_id, p.advert_name.clone()))
|
||||
.unwrap_or((0, sender_prefix.to_string()))
|
||||
}
|
||||
|
||||
/// Store a plain-text (non-typed) message and emit an event.
|
||||
pub(super) async fn store_plain_message(
|
||||
state: &Arc<MeshState>,
|
||||
contact_id: u32,
|
||||
peer_name: &str,
|
||||
text: &str,
|
||||
) {
|
||||
let msg_id = state.next_id().await;
|
||||
let msg = MeshMessage {
|
||||
id: msg_id,
|
||||
direction: MessageDirection::Received,
|
||||
peer_contact_id: contact_id,
|
||||
peer_name: Some(peer_name.to_string()),
|
||||
plaintext: text.to_string(),
|
||||
timestamp: chrono::Utc::now().to_rfc3339(),
|
||||
delivered: true,
|
||||
encrypted: false,
|
||||
};
|
||||
state.store_message(msg.clone()).await;
|
||||
state.status.write().await.messages_received += 1;
|
||||
let _ = state.event_tx.send(MeshEvent::MessageReceived(msg));
|
||||
}
|
||||
|
||||
/// Handle a received identity broadcast from a peer.
|
||||
#[allow(dead_code)]
|
||||
pub(super) async fn handle_identity_received(
|
||||
contact_id: u32,
|
||||
rssi: i16,
|
||||
did: &str,
|
||||
ed_pubkey_hex: &str,
|
||||
x25519_pubkey_hex: &str,
|
||||
state: &Arc<MeshState>,
|
||||
our_x25519_secret: &[u8; 32],
|
||||
) {
|
||||
info!(
|
||||
contact_id,
|
||||
did = %did,
|
||||
rssi,
|
||||
"Archipelago peer discovered over mesh"
|
||||
);
|
||||
|
||||
// Verify Ed25519 public key is valid
|
||||
let ed_pubkey_bytes = match hex::decode(ed_pubkey_hex) {
|
||||
Ok(b) if b.len() == 32 => {
|
||||
let mut arr = [0u8; 32];
|
||||
arr.copy_from_slice(&b);
|
||||
arr
|
||||
}
|
||||
_ => {
|
||||
warn!(contact_id, "Rejecting identity: invalid Ed25519 public key");
|
||||
return;
|
||||
}
|
||||
};
|
||||
if ed25519_dalek::VerifyingKey::from_bytes(&ed_pubkey_bytes).is_err() {
|
||||
warn!(contact_id, "Rejecting identity: Ed25519 key is not a valid curve point");
|
||||
return;
|
||||
}
|
||||
|
||||
// Verify X25519 public key is consistent with Ed25519 key
|
||||
let expected_x25519 = match crypto::ed25519_pubkey_to_x25519(&ed_pubkey_bytes) {
|
||||
Ok(k) => k,
|
||||
Err(e) => {
|
||||
warn!(contact_id, "Rejecting identity: cannot derive X25519 from Ed25519: {}", e);
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
// Decode X25519 public key
|
||||
let x25519_bytes = match hex::decode(x25519_pubkey_hex) {
|
||||
Ok(b) if b.len() == 32 => {
|
||||
let mut arr = [0u8; 32];
|
||||
arr.copy_from_slice(&b);
|
||||
arr
|
||||
}
|
||||
_ => {
|
||||
warn!(contact_id, "Rejecting identity: invalid X25519 public key");
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
if x25519_bytes != expected_x25519 {
|
||||
warn!(contact_id, did = %did, "Rejecting identity: X25519 key does not match Ed25519 key");
|
||||
return;
|
||||
}
|
||||
|
||||
// Derive shared secret for encrypted messaging
|
||||
let shared_secret = crypto::x25519_shared_secret(our_x25519_secret, &x25519_bytes);
|
||||
state
|
||||
.shared_secrets
|
||||
.write()
|
||||
.await
|
||||
.insert(contact_id, shared_secret);
|
||||
|
||||
// Update peer record
|
||||
let peer = MeshPeer {
|
||||
contact_id,
|
||||
advert_name: format!("Archy-{}", &did[8..16.min(did.len())]),
|
||||
did: Some(did.to_string()),
|
||||
pubkey_hex: Some(ed_pubkey_hex.to_string()),
|
||||
x25519_pubkey: Some(x25519_bytes),
|
||||
rssi: Some(rssi),
|
||||
snr: None,
|
||||
last_heard: chrono::Utc::now().to_rfc3339(),
|
||||
hops: 0,
|
||||
};
|
||||
|
||||
let is_new = {
|
||||
let mut peers = state.peers.write().await;
|
||||
let is_new = !peers.contains_key(&contact_id);
|
||||
peers.insert(contact_id, peer.clone());
|
||||
is_new
|
||||
};
|
||||
state.update_peer_count().await;
|
||||
|
||||
let event = if is_new {
|
||||
MeshEvent::PeerDiscovered(peer)
|
||||
} else {
|
||||
MeshEvent::PeerUpdated(peer)
|
||||
};
|
||||
let _ = state.event_tx.send(event);
|
||||
let _ = state.event_tx.send(MeshEvent::IdentityReceived {
|
||||
contact_id,
|
||||
did: did.to_string(),
|
||||
pubkey_hex: ed_pubkey_hex.to_string(),
|
||||
x25519_pubkey: x25519_bytes,
|
||||
});
|
||||
}
|
||||
|
||||
/// Handle a received message (direct or channel).
|
||||
#[allow(dead_code)]
|
||||
pub(super) async fn handle_received_message(
|
||||
contact_id: u32,
|
||||
payload: &[u8],
|
||||
rssi: i16,
|
||||
is_channel: bool,
|
||||
state: &Arc<MeshState>,
|
||||
_our_x25519_secret: &[u8; 32],
|
||||
) {
|
||||
// Try to decrypt if we have a shared secret for this contact
|
||||
let shared_secrets = state.shared_secrets.read().await;
|
||||
let (plaintext, encrypted) = if let Some(secret) = shared_secrets.get(&contact_id) {
|
||||
match crypto::decrypt(secret, payload) {
|
||||
Ok(pt) => (String::from_utf8_lossy(&pt).to_string(), true),
|
||||
Err(_) => {
|
||||
// Not encrypted or wrong key — treat as plaintext
|
||||
(String::from_utf8_lossy(payload).to_string(), false)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
(String::from_utf8_lossy(payload).to_string(), false)
|
||||
};
|
||||
drop(shared_secrets);
|
||||
|
||||
// Update peer last_heard
|
||||
{
|
||||
let mut peers = state.peers.write().await;
|
||||
if let Some(peer) = peers.get_mut(&contact_id) {
|
||||
peer.last_heard = chrono::Utc::now().to_rfc3339();
|
||||
peer.rssi = Some(rssi);
|
||||
}
|
||||
}
|
||||
|
||||
let peer_name = state
|
||||
.peers
|
||||
.read()
|
||||
.await
|
||||
.get(&contact_id)
|
||||
.map(|p| p.advert_name.clone());
|
||||
|
||||
let msg_id = state.next_id().await;
|
||||
let msg = MeshMessage {
|
||||
id: msg_id,
|
||||
direction: MessageDirection::Received,
|
||||
peer_contact_id: contact_id,
|
||||
peer_name,
|
||||
plaintext: plaintext.clone(),
|
||||
timestamp: chrono::Utc::now().to_rfc3339(),
|
||||
delivered: true,
|
||||
encrypted,
|
||||
};
|
||||
|
||||
state.store_message(msg.clone()).await;
|
||||
{
|
||||
let mut status = state.status.write().await;
|
||||
status.messages_received += 1;
|
||||
}
|
||||
|
||||
info!(
|
||||
contact_id,
|
||||
encrypted,
|
||||
channel = is_channel,
|
||||
"Received mesh message"
|
||||
);
|
||||
|
||||
let _ = state.event_tx.send(MeshEvent::MessageReceived(msg));
|
||||
}
|
||||
@@ -0,0 +1,402 @@
|
||||
//! Typed message dispatch — routes TypedEnvelope messages to type-specific handlers.
|
||||
|
||||
use super::bitcoin::handle_tx_relay_broadcast;
|
||||
use super::decode::store_plain_message;
|
||||
use super::MeshState;
|
||||
use super::super::message_types::{self, MeshMessageType, TypedEnvelope};
|
||||
use super::super::types::*;
|
||||
use std::sync::Arc;
|
||||
use tracing::{debug, info, warn};
|
||||
|
||||
/// Store a typed message with a type label for UI rendering.
|
||||
async fn store_typed_message(
|
||||
state: &Arc<MeshState>,
|
||||
contact_id: u32,
|
||||
peer_name: &str,
|
||||
text: &str,
|
||||
type_label: &str,
|
||||
) {
|
||||
let msg_id = state.next_id().await;
|
||||
let msg = MeshMessage {
|
||||
id: msg_id,
|
||||
direction: MessageDirection::Received,
|
||||
peer_contact_id: contact_id,
|
||||
peer_name: Some(peer_name.to_string()),
|
||||
plaintext: format!("[{}] {}", type_label, text),
|
||||
timestamp: chrono::Utc::now().to_rfc3339(),
|
||||
delivered: true,
|
||||
encrypted: false,
|
||||
};
|
||||
state.store_message(msg.clone()).await;
|
||||
state.status.write().await.messages_received += 1;
|
||||
let _ = state.event_tx.send(MeshEvent::MessageReceived(msg));
|
||||
}
|
||||
|
||||
/// Handle a typed message envelope (0x02 prefix).
|
||||
/// Dispatches to type-specific handlers: BlockHeader, Alert, TxRelay, etc.
|
||||
pub(super) async fn handle_typed_message(
|
||||
payload: &[u8],
|
||||
sender_contact_id: u32,
|
||||
sender_name: &str,
|
||||
state: &Arc<MeshState>,
|
||||
) {
|
||||
let envelope = match TypedEnvelope::from_wire(payload) {
|
||||
Ok(e) => e,
|
||||
Err(e) => {
|
||||
warn!(
|
||||
payload_len = payload.len(),
|
||||
first_bytes = %hex::encode(&payload[..payload.len().min(16)]),
|
||||
"Failed to decode typed envelope: {}", e
|
||||
);
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
// Verify envelope signature if present, using the sender's known Ed25519 key
|
||||
if envelope.sig.is_some() {
|
||||
let peer_pubkey = state.peers.read().await
|
||||
.get(&sender_contact_id)
|
||||
.and_then(|p| p.pubkey_hex.as_ref())
|
||||
.and_then(|hex_str| hex::decode(hex_str).ok())
|
||||
.and_then(|bytes| {
|
||||
if bytes.len() == 32 {
|
||||
let mut arr = [0u8; 32];
|
||||
arr.copy_from_slice(&bytes);
|
||||
ed25519_dalek::VerifyingKey::from_bytes(&arr).ok()
|
||||
} else {
|
||||
None
|
||||
}
|
||||
});
|
||||
if let Some(vk) = peer_pubkey {
|
||||
match envelope.verify_signature(&vk) {
|
||||
Ok(true) => {}
|
||||
Ok(false) => {
|
||||
warn!(peer = sender_contact_id, "Dropping message with invalid signature");
|
||||
return;
|
||||
}
|
||||
Err(e) => {
|
||||
warn!(peer = sender_contact_id, "Signature verification error: {}", e);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let msg_type = envelope.message_type();
|
||||
let type_label = msg_type.map(|t| t.label()).unwrap_or("unknown");
|
||||
info!(
|
||||
msg_type = type_label,
|
||||
from = sender_contact_id,
|
||||
"Received typed mesh message"
|
||||
);
|
||||
|
||||
match msg_type {
|
||||
Some(MeshMessageType::BlockHeader) => {
|
||||
dispatch_block_header(&envelope, sender_contact_id, sender_name, state).await;
|
||||
}
|
||||
|
||||
Some(MeshMessageType::Alert) => {
|
||||
match message_types::decode_payload::<message_types::AlertPayload>(&envelope.v) {
|
||||
Ok(alert) => {
|
||||
let alert_type_str = format!("{:?}", alert.alert_type).to_lowercase();
|
||||
info!(
|
||||
alert_type = %alert_type_str,
|
||||
from = sender_contact_id,
|
||||
"Alert received via mesh: {}",
|
||||
alert.message
|
||||
);
|
||||
store_typed_message(
|
||||
state,
|
||||
sender_contact_id,
|
||||
sender_name,
|
||||
&alert.message,
|
||||
"alert",
|
||||
)
|
||||
.await;
|
||||
let _ = state.event_tx.send(MeshEvent::AlertReceived {
|
||||
alert_type: alert_type_str,
|
||||
message: alert.message,
|
||||
from_contact_id: sender_contact_id,
|
||||
});
|
||||
}
|
||||
Err(e) => warn!("Failed to decode alert payload: {}", e),
|
||||
}
|
||||
}
|
||||
|
||||
Some(MeshMessageType::TxRelay) => {
|
||||
match message_types::decode_payload::<message_types::TxRelayPayload>(&envelope.v) {
|
||||
Ok(relay) => {
|
||||
// Validate transaction before relaying
|
||||
if !super::super::bitcoin_relay::validate_raw_transaction(&relay.tx_hex) {
|
||||
warn!(peer = sender_contact_id, "Rejected invalid TX relay");
|
||||
return;
|
||||
}
|
||||
info!(
|
||||
request_id = relay.request_id,
|
||||
tx_len = relay.tx_hex.len(),
|
||||
"TX relay request received — broadcasting to Bitcoin network"
|
||||
);
|
||||
store_typed_message(
|
||||
state,
|
||||
sender_contact_id,
|
||||
sender_name,
|
||||
&format!("TX relay request #{} ({} hex chars)", relay.request_id, relay.tx_hex.len()),
|
||||
"tx_relay",
|
||||
)
|
||||
.await;
|
||||
|
||||
// Spawn async task to broadcast via Bitcoin RPC and track confirmations
|
||||
let relay_state = Arc::clone(state);
|
||||
let relay_contact = sender_contact_id;
|
||||
tokio::spawn(async move {
|
||||
handle_tx_relay_broadcast(relay, relay_contact, &relay_state).await;
|
||||
});
|
||||
}
|
||||
Err(e) => warn!("Failed to decode TX relay payload: {}", e),
|
||||
}
|
||||
}
|
||||
|
||||
Some(MeshMessageType::TxRelayResponse) => {
|
||||
dispatch_tx_relay_response(&envelope, sender_contact_id, sender_name, state).await;
|
||||
}
|
||||
|
||||
Some(MeshMessageType::LightningRelay) => {
|
||||
match message_types::decode_payload::<message_types::LightningRelayPayload>(
|
||||
&envelope.v,
|
||||
) {
|
||||
Ok(relay) => {
|
||||
info!(
|
||||
request_id = relay.request_id,
|
||||
amount_sats = relay.amount_sats,
|
||||
"Lightning relay request received"
|
||||
);
|
||||
store_typed_message(
|
||||
state,
|
||||
sender_contact_id,
|
||||
sender_name,
|
||||
&format!("Lightning relay: {} sats", relay.amount_sats),
|
||||
"lightning_relay",
|
||||
)
|
||||
.await;
|
||||
// Will be wired to LND in Week 9
|
||||
let _ = state.event_tx.send(MeshEvent::LightningRelayCompleted {
|
||||
request_id: relay.request_id,
|
||||
payment_hash: None,
|
||||
error: Some("Lightning relay processing not yet wired".to_string()),
|
||||
});
|
||||
}
|
||||
Err(e) => warn!("Failed to decode Lightning relay payload: {}", e),
|
||||
}
|
||||
}
|
||||
|
||||
Some(MeshMessageType::LightningRelayResponse) => {
|
||||
match message_types::decode_payload::<message_types::LightningRelayResponsePayload>(
|
||||
&envelope.v,
|
||||
) {
|
||||
Ok(resp) => {
|
||||
let status = if resp.payment_hash.is_some() { "paid" } else { "failed" };
|
||||
info!(request_id = resp.request_id, status, "Lightning relay response");
|
||||
let text = if let Some(ref hash) = resp.payment_hash {
|
||||
format!("Lightning paid! hash: {}...", &hash[..16.min(hash.len())])
|
||||
} else {
|
||||
format!("Lightning failed: {}", resp.error.as_deref().unwrap_or("unknown"))
|
||||
};
|
||||
store_typed_message(state, sender_contact_id, sender_name, &text, "lightning_relay_response").await;
|
||||
let _ = state.event_tx.send(MeshEvent::LightningRelayCompleted {
|
||||
request_id: resp.request_id,
|
||||
payment_hash: resp.payment_hash,
|
||||
error: resp.error,
|
||||
});
|
||||
}
|
||||
Err(e) => warn!("Failed to decode Lightning relay response: {}", e),
|
||||
}
|
||||
}
|
||||
|
||||
Some(MeshMessageType::Invoice) => {
|
||||
match message_types::decode_payload::<message_types::InvoicePayload>(&envelope.v) {
|
||||
Ok(invoice) => {
|
||||
let text = format!(
|
||||
"Invoice: {} sats{}",
|
||||
invoice.amount_sats,
|
||||
invoice.memo.as_ref().map(|m| format!(" — {}", m)).unwrap_or_default()
|
||||
);
|
||||
store_typed_message(state, sender_contact_id, sender_name, &text, "invoice").await;
|
||||
}
|
||||
Err(e) => warn!("Failed to decode invoice payload: {}", e),
|
||||
}
|
||||
}
|
||||
|
||||
Some(MeshMessageType::Coordinate) => {
|
||||
match message_types::decode_payload::<message_types::Coordinate>(&envelope.v) {
|
||||
Ok(coord) => {
|
||||
let text = format!(
|
||||
"Location: {:.6}, {:.6}{}",
|
||||
coord.lat_degrees(),
|
||||
coord.lng_degrees(),
|
||||
coord.label.as_ref().map(|l| format!(" ({})", l)).unwrap_or_default()
|
||||
);
|
||||
store_typed_message(state, sender_contact_id, sender_name, &text, "coordinate").await;
|
||||
}
|
||||
Err(e) => warn!("Failed to decode coordinate payload: {}", e),
|
||||
}
|
||||
}
|
||||
|
||||
Some(MeshMessageType::TxConfirmation) => {
|
||||
dispatch_tx_confirmation(&envelope, sender_contact_id, sender_name, state).await;
|
||||
}
|
||||
|
||||
Some(MeshMessageType::Text) => {
|
||||
// Typed text message — extract and store as plain text
|
||||
let text = String::from_utf8_lossy(&envelope.v).to_string();
|
||||
store_plain_message(state, sender_contact_id, sender_name, &text).await;
|
||||
}
|
||||
|
||||
_ => {
|
||||
debug!(
|
||||
msg_type = ?msg_type,
|
||||
"Unhandled typed message type"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Dispatch a BlockHeader typed message.
|
||||
async fn dispatch_block_header(
|
||||
envelope: &TypedEnvelope,
|
||||
sender_contact_id: u32,
|
||||
sender_name: &str,
|
||||
state: &Arc<MeshState>,
|
||||
) {
|
||||
// Compact binary format: height(8) + hash(32) + timestamp(4)
|
||||
match super::super::bitcoin_relay::decode_compact_block_header(&envelope.v) {
|
||||
Ok((height, hash_hex, timestamp)) => {
|
||||
// Validate header before accepting
|
||||
let last_known = state.block_header_cache.latest_height().await;
|
||||
if !super::super::bitcoin_relay::validate_block_header(height, &hash_hex, timestamp, last_known) {
|
||||
warn!(peer = sender_contact_id, height, "Rejected invalid block header");
|
||||
return;
|
||||
}
|
||||
|
||||
info!(
|
||||
height,
|
||||
hash = %hash_hex,
|
||||
"Block header received via mesh"
|
||||
);
|
||||
|
||||
// Store in block header cache for the Off-Grid Bitcoin panel
|
||||
let header_payload = message_types::BlockHeaderPayload {
|
||||
height,
|
||||
hash: hash_hex.clone(),
|
||||
prev_hash: String::new(),
|
||||
timestamp,
|
||||
announced_by: sender_name.to_string(),
|
||||
};
|
||||
let _ = state.block_header_cache.store_header(header_payload).await;
|
||||
|
||||
let text = format!(
|
||||
"Block #{} — {}...{}",
|
||||
height,
|
||||
&hash_hex[..8.min(hash_hex.len())],
|
||||
&hash_hex[hash_hex.len().saturating_sub(8)..]
|
||||
);
|
||||
store_typed_message(
|
||||
state,
|
||||
sender_contact_id,
|
||||
sender_name,
|
||||
&text,
|
||||
"block_header",
|
||||
)
|
||||
.await;
|
||||
let _ = state.event_tx.send(MeshEvent::BlockHeaderReceived {
|
||||
height,
|
||||
hash: hash_hex,
|
||||
});
|
||||
}
|
||||
Err(e) => warn!("Failed to decode block header: {}", e),
|
||||
}
|
||||
}
|
||||
|
||||
/// Dispatch a TxRelayResponse typed message.
|
||||
async fn dispatch_tx_relay_response(
|
||||
envelope: &TypedEnvelope,
|
||||
sender_contact_id: u32,
|
||||
sender_name: &str,
|
||||
state: &Arc<MeshState>,
|
||||
) {
|
||||
match message_types::decode_payload::<message_types::TxRelayResponsePayload>(&envelope.v) {
|
||||
Ok(resp) => {
|
||||
let status = if resp.txid.is_some() { "confirmed" } else { "failed" };
|
||||
info!(
|
||||
request_id = resp.request_id,
|
||||
status,
|
||||
error_code = resp.error_code.as_deref().unwrap_or("none"),
|
||||
"TX relay response received"
|
||||
);
|
||||
let text = if let Some(ref txid) = resp.txid {
|
||||
format!("TX relayed! txid: {}...{}", &txid[..8.min(txid.len())], &txid[txid.len().saturating_sub(8)..])
|
||||
} else if let Some(ref code) = resp.error_code {
|
||||
format!("TX relay failed [{}]: {}", code, resp.error.as_deref().unwrap_or("unknown"))
|
||||
} else {
|
||||
format!("TX relay failed: {}", resp.error.as_deref().unwrap_or("unknown"))
|
||||
};
|
||||
store_typed_message(state, sender_contact_id, sender_name, &text, "tx_relay_response").await;
|
||||
// Store result for frontend polling
|
||||
if let Some(ref tracker) = state.relay_tracker {
|
||||
tracker.store_result(super::super::bitcoin_relay::RelayResult {
|
||||
request_id: resp.request_id,
|
||||
txid: resp.txid.clone(),
|
||||
error: resp.error.clone(),
|
||||
error_code: resp.error_code.clone(),
|
||||
completed_at: chrono::Utc::now().to_rfc3339(),
|
||||
}).await;
|
||||
}
|
||||
let _ = state.event_tx.send(MeshEvent::TxRelayCompleted {
|
||||
request_id: resp.request_id,
|
||||
txid: resp.txid,
|
||||
error: resp.error,
|
||||
});
|
||||
}
|
||||
Err(e) => warn!("Failed to decode TX relay response: {}", e),
|
||||
}
|
||||
}
|
||||
|
||||
/// Dispatch a TxConfirmation typed message.
|
||||
async fn dispatch_tx_confirmation(
|
||||
envelope: &TypedEnvelope,
|
||||
sender_contact_id: u32,
|
||||
sender_name: &str,
|
||||
state: &Arc<MeshState>,
|
||||
) {
|
||||
match message_types::decode_payload::<message_types::TxConfirmationPayload>(&envelope.v) {
|
||||
Ok(conf) => {
|
||||
let status_text = if conf.confirmations >= 3 {
|
||||
format!("TX {} confirmed ({}/3) at block #{}", &conf.txid[..12.min(conf.txid.len())], conf.confirmations, conf.block_height)
|
||||
} else {
|
||||
format!("TX {} — {}/3 confirmations (block #{})", &conf.txid[..12.min(conf.txid.len())], conf.confirmations, conf.block_height)
|
||||
};
|
||||
info!(
|
||||
txid = %conf.txid,
|
||||
confirmations = conf.confirmations,
|
||||
block_height = conf.block_height,
|
||||
"TX confirmation update received"
|
||||
);
|
||||
store_typed_message(state, sender_contact_id, sender_name, &status_text, "tx_confirmation").await;
|
||||
// Store confirmation for frontend polling
|
||||
if let Some(ref tracker) = state.relay_tracker {
|
||||
tracker.store_result(super::super::bitcoin_relay::RelayResult {
|
||||
request_id: conf.request_id,
|
||||
txid: Some(conf.txid.clone()),
|
||||
error: None,
|
||||
error_code: None,
|
||||
completed_at: chrono::Utc::now().to_rfc3339(),
|
||||
}).await;
|
||||
}
|
||||
let _ = state.event_tx.send(MeshEvent::TxRelayCompleted {
|
||||
request_id: conf.request_id,
|
||||
txid: Some(conf.txid),
|
||||
error: None,
|
||||
});
|
||||
}
|
||||
Err(e) => warn!("Failed to decode TX confirmation: {}", e),
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,143 @@
|
||||
//! Inbound frame dispatcher — routes device frames to the appropriate handler.
|
||||
|
||||
use super::decode::{
|
||||
resolve_peer, store_plain_message, try_base64_typed, try_chunk_reassemble,
|
||||
try_decrypt_base64, try_decrypt_ratchet_base64,
|
||||
};
|
||||
use super::dispatch::handle_typed_message;
|
||||
use super::MeshState;
|
||||
use super::super::message_types::TypedEnvelope;
|
||||
use super::super::protocol;
|
||||
use std::sync::Arc;
|
||||
use tracing::{debug, info, warn};
|
||||
|
||||
/// Handle a single inbound frame from the device.
|
||||
/// Returns `true` if contacts should be refreshed from the device.
|
||||
pub(super) async fn handle_frame(
|
||||
frame: &protocol::InboundFrame,
|
||||
state: &Arc<MeshState>,
|
||||
our_x25519_secret: &[u8; 32],
|
||||
) -> bool {
|
||||
let _ = our_x25519_secret; // reserved for future per-frame decryption
|
||||
match frame.code {
|
||||
protocol::PUSH_NEW_CONTACT | protocol::PUSH_CONTACT_ADVERT => {
|
||||
info!(code = frame.code, "Contact discovery event — refreshing contacts");
|
||||
return true; // Signal caller to fetch contacts
|
||||
}
|
||||
|
||||
protocol::PUSH_ACK => {
|
||||
debug!("Message delivery confirmed");
|
||||
// Could track which message was ACKed from frame.data
|
||||
}
|
||||
|
||||
protocol::PUSH_MESSAGES_WAITING => {
|
||||
info!("Device has messages waiting — will sync");
|
||||
return true; // Signal caller to sync immediately
|
||||
}
|
||||
|
||||
protocol::RESP_CONTACT_MSG_V3 => {
|
||||
// Direct message received (v3 format) — check for typed envelope first
|
||||
match protocol::parse_contact_msg_v3_raw(&frame.data) {
|
||||
Ok((sender_prefix, payload, _snr)) => {
|
||||
if !payload.is_empty() {
|
||||
let (contact_id, name) = resolve_peer(state, &sender_prefix).await;
|
||||
if TypedEnvelope::is_typed(&payload) {
|
||||
handle_typed_message(&payload, contact_id, &name, state).await;
|
||||
} else if let Some(decoded) = try_base64_typed(&payload) {
|
||||
handle_typed_message(&decoded, contact_id, &name, state).await;
|
||||
} else if let Some(decoded) = try_decrypt_ratchet_base64(&payload, contact_id, state).await {
|
||||
handle_typed_message(&decoded, contact_id, &name, state).await;
|
||||
} else if let Some(decoded) = try_decrypt_base64(&payload, contact_id, state).await {
|
||||
handle_typed_message(&decoded, contact_id, &name, state).await;
|
||||
} else if let Some(decoded) = try_chunk_reassemble(&payload, contact_id, state).await {
|
||||
handle_typed_message(&decoded, contact_id, &name, state).await;
|
||||
} else if !payload.starts_with(b"MC") {
|
||||
let text = String::from_utf8_lossy(&payload).to_string();
|
||||
store_plain_message(state, contact_id, &name, &text).await;
|
||||
info!(from = %sender_prefix, "Received mesh DM (v3)");
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => warn!("Failed to parse v3 message: {}", e),
|
||||
}
|
||||
}
|
||||
|
||||
protocol::RESP_CONTACT_MSG => {
|
||||
// Direct message received (v1 format)
|
||||
match protocol::parse_contact_msg_v1_raw(&frame.data) {
|
||||
Ok((sender_prefix, payload)) => {
|
||||
if !payload.is_empty() {
|
||||
let (contact_id, name) = resolve_peer(state, &sender_prefix).await;
|
||||
if TypedEnvelope::is_typed(&payload) {
|
||||
handle_typed_message(&payload, contact_id, &name, state).await;
|
||||
} else if let Some(decoded) = try_base64_typed(&payload) {
|
||||
handle_typed_message(&decoded, contact_id, &name, state).await;
|
||||
} else if let Some(decoded) = try_decrypt_ratchet_base64(&payload, contact_id, state).await {
|
||||
handle_typed_message(&decoded, contact_id, &name, state).await;
|
||||
} else if let Some(decoded) = try_decrypt_base64(&payload, contact_id, state).await {
|
||||
handle_typed_message(&decoded, contact_id, &name, state).await;
|
||||
} else if let Some(decoded) = try_chunk_reassemble(&payload, contact_id, state).await {
|
||||
handle_typed_message(&decoded, contact_id, &name, state).await;
|
||||
} else if !payload.starts_with(b"MC") {
|
||||
let text = String::from_utf8_lossy(&payload).to_string();
|
||||
store_plain_message(state, contact_id, &name, &text).await;
|
||||
info!(from = %sender_prefix, "Received mesh DM (v1)");
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => warn!("Failed to parse v1 message: {}", e),
|
||||
}
|
||||
}
|
||||
|
||||
protocol::RESP_CHANNEL_MSG_V3 => {
|
||||
// Channel broadcast received (v3) — check for typed envelope
|
||||
match protocol::parse_channel_msg_v3_raw(&frame.data) {
|
||||
Ok((channel_idx, payload)) => {
|
||||
if !payload.is_empty() {
|
||||
let chan_contact_id = u32::MAX - (channel_idx as u32);
|
||||
let chan_name = format!("Channel {}", channel_idx);
|
||||
if TypedEnvelope::is_typed(&payload) {
|
||||
handle_typed_message(&payload, chan_contact_id, &chan_name, state).await;
|
||||
} else {
|
||||
let text = String::from_utf8_lossy(&payload).to_string();
|
||||
store_plain_message(state, chan_contact_id, &chan_name, &text).await;
|
||||
info!(channel = channel_idx, "Received mesh channel message (v3)");
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => warn!("Failed to parse v3 channel message: {}", e),
|
||||
}
|
||||
}
|
||||
|
||||
protocol::RESP_CHANNEL_MSG => {
|
||||
// Channel broadcast received (v1)
|
||||
match protocol::parse_channel_msg_v1_raw(&frame.data) {
|
||||
Ok((channel_idx, payload)) => {
|
||||
if !payload.is_empty() {
|
||||
let chan_contact_id = u32::MAX - (channel_idx as u32);
|
||||
let chan_name = format!("Channel {}", channel_idx);
|
||||
if TypedEnvelope::is_typed(&payload) {
|
||||
handle_typed_message(&payload, chan_contact_id, &chan_name, state).await;
|
||||
} else {
|
||||
let text = String::from_utf8_lossy(&payload).to_string();
|
||||
store_plain_message(state, chan_contact_id, &chan_name, &text).await;
|
||||
info!(channel = channel_idx, "Received mesh channel message");
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => warn!("Failed to parse channel message: {}", e),
|
||||
}
|
||||
}
|
||||
|
||||
protocol::PUSH_LOG_DATA | protocol::PUSH_PATH_UPDATE | protocol::PUSH_RAW_DATA => {
|
||||
// Internal device logging/path data — safe to ignore
|
||||
}
|
||||
|
||||
_ => {
|
||||
if protocol::is_push_notification(frame.code) {
|
||||
debug!(code = frame.code, "Unhandled push notification");
|
||||
}
|
||||
}
|
||||
}
|
||||
false
|
||||
}
|
||||
@@ -0,0 +1,218 @@
|
||||
//! Background mesh listener task.
|
||||
//!
|
||||
//! Runs as a long-lived tokio task that:
|
||||
//! - Maintains the serial connection to the Meshcore device
|
||||
//! - Reads incoming frames and dispatches events
|
||||
//! - Periodically broadcasts our identity advertisement
|
||||
//! - Reconnects on device disconnect
|
||||
//! - Manages peer cache and message store
|
||||
|
||||
mod bitcoin;
|
||||
mod decode;
|
||||
mod dispatch;
|
||||
mod frames;
|
||||
mod session;
|
||||
|
||||
use super::types::*;
|
||||
use std::collections::{HashMap, VecDeque};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use tokio::sync::{broadcast, mpsc, RwLock};
|
||||
use tracing::{error, info};
|
||||
|
||||
/// How often to broadcast our identity advertisement (seconds).
|
||||
const ADVERT_INTERVAL: Duration = Duration::from_secs(60);
|
||||
|
||||
/// How often to poll for queued messages when no push notifications.
|
||||
const SYNC_INTERVAL: Duration = Duration::from_secs(10);
|
||||
|
||||
/// Maximum stored messages (circular buffer).
|
||||
const MAX_MESSAGES: usize = 100;
|
||||
|
||||
/// Initial delay before reconnection attempt after device disconnect.
|
||||
const RECONNECT_DELAY_INIT: Duration = Duration::from_secs(5);
|
||||
|
||||
/// Maximum reconnect delay (cap for exponential backoff).
|
||||
const RECONNECT_DELAY_MAX: Duration = Duration::from_secs(60);
|
||||
|
||||
/// Number of consecutive write failures before we consider the device dead
|
||||
/// and trigger a reconnection cycle.
|
||||
const MAX_CONSECUTIVE_WRITE_FAILURES: u32 = 3;
|
||||
|
||||
/// Command sent from MeshService to the listener task (which owns the serial port).
|
||||
pub enum MeshCommand {
|
||||
SendText { dest_pubkey_prefix: [u8; 6], payload: Vec<u8> },
|
||||
/// Send pre-encoded binary (TypedEnvelope wire bytes) to a peer.
|
||||
SendRaw { dest_pubkey_prefix: [u8; 6], payload: Vec<u8> },
|
||||
/// Broadcast pre-encoded binary on a mesh channel.
|
||||
BroadcastChannel { channel: u8, payload: Vec<u8> },
|
||||
SendAdvert,
|
||||
}
|
||||
|
||||
/// Shared state for the mesh listener, accessible from RPC handlers.
|
||||
pub struct MeshState {
|
||||
pub peers: RwLock<HashMap<u32, MeshPeer>>,
|
||||
pub messages: RwLock<VecDeque<MeshMessage>>,
|
||||
pub shared_secrets: RwLock<HashMap<u32, [u8; 32]>>,
|
||||
pub status: RwLock<MeshStatus>,
|
||||
pub event_tx: broadcast::Sender<MeshEvent>,
|
||||
pub cmd_tx: mpsc::Sender<MeshCommand>,
|
||||
next_message_id: RwLock<u64>,
|
||||
/// Block header cache — populated when receiving headers from internet-connected peers.
|
||||
pub block_header_cache: Arc<super::bitcoin_relay::BlockHeaderCache>,
|
||||
/// Relay tracker — stores completed relay results for frontend polling.
|
||||
pub relay_tracker: Option<Arc<super::bitcoin_relay::RelayTracker>>,
|
||||
/// Steganography mode for outgoing/incoming messages.
|
||||
pub stego_mode: super::steganography::SteganographyMode,
|
||||
/// Chunk reassembly buffer for multi-frame messages.
|
||||
chunk_buffer: RwLock<HashMap<(u32, u8), ChunkAssembly>>,
|
||||
/// Double Ratchet session manager for forward-secret encryption.
|
||||
pub session_manager: Arc<super::session::SessionManager>,
|
||||
/// Whether to encrypt directed relay messages (config toggle for rollback).
|
||||
pub encrypt_relay: bool,
|
||||
}
|
||||
|
||||
/// In-progress chunk reassembly for a multi-frame message.
|
||||
struct ChunkAssembly {
|
||||
chunks: HashMap<u8, String>,
|
||||
total: u8,
|
||||
created: std::time::Instant,
|
||||
}
|
||||
|
||||
impl MeshState {
|
||||
pub fn new(
|
||||
channel_name: &str,
|
||||
block_header_cache: Arc<super::bitcoin_relay::BlockHeaderCache>,
|
||||
relay_tracker: Option<Arc<super::bitcoin_relay::RelayTracker>>,
|
||||
stego_mode: super::steganography::SteganographyMode,
|
||||
encrypt_relay: bool,
|
||||
session_manager: Arc<super::session::SessionManager>,
|
||||
) -> (Arc<Self>, broadcast::Receiver<MeshEvent>, mpsc::Receiver<MeshCommand>) {
|
||||
let (tx, rx) = broadcast::channel(64);
|
||||
let (cmd_tx, cmd_rx) = mpsc::channel(32);
|
||||
let state = Arc::new(Self {
|
||||
peers: RwLock::new(HashMap::new()),
|
||||
messages: RwLock::new(VecDeque::new()),
|
||||
shared_secrets: RwLock::new(HashMap::new()),
|
||||
cmd_tx,
|
||||
status: RwLock::new(MeshStatus {
|
||||
enabled: true,
|
||||
device_type: DeviceType::Unknown,
|
||||
device_path: None,
|
||||
device_connected: false,
|
||||
firmware_version: None,
|
||||
self_node_id: None,
|
||||
self_advert_name: None,
|
||||
peer_count: 0,
|
||||
channel_name: channel_name.to_string(),
|
||||
messages_sent: 0,
|
||||
messages_received: 0,
|
||||
}),
|
||||
event_tx: tx,
|
||||
next_message_id: RwLock::new(1),
|
||||
block_header_cache,
|
||||
relay_tracker,
|
||||
stego_mode,
|
||||
chunk_buffer: RwLock::new(HashMap::new()),
|
||||
session_manager,
|
||||
encrypt_relay,
|
||||
});
|
||||
(state, rx, cmd_rx)
|
||||
}
|
||||
|
||||
pub async fn next_id(&self) -> u64 {
|
||||
let mut id = self.next_message_id.write().await;
|
||||
let current = *id;
|
||||
*id += 1;
|
||||
current
|
||||
}
|
||||
|
||||
pub async fn store_message(&self, msg: MeshMessage) {
|
||||
let mut messages = self.messages.write().await;
|
||||
messages.push_back(msg);
|
||||
if messages.len() > MAX_MESSAGES {
|
||||
messages.pop_front();
|
||||
}
|
||||
}
|
||||
|
||||
async fn update_peer_count(&self) {
|
||||
let count = self.peers.read().await.len();
|
||||
self.status.write().await.peer_count = count;
|
||||
}
|
||||
}
|
||||
|
||||
/// Spawn the background mesh listener task.
|
||||
///
|
||||
/// This task manages the full lifecycle:
|
||||
/// 1. Detect and connect to Meshcore device
|
||||
/// 2. Initialize and set advert name
|
||||
/// 3. Main loop: read frames, dispatch events, periodic adverts
|
||||
/// 4. Reconnect on disconnect
|
||||
pub fn spawn_mesh_listener(
|
||||
state: Arc<MeshState>,
|
||||
device_path: Option<String>,
|
||||
our_did: String,
|
||||
our_ed_pubkey_hex: String,
|
||||
our_x25519_secret: [u8; 32],
|
||||
our_x25519_pubkey_hex: String,
|
||||
shutdown: tokio::sync::watch::Receiver<bool>,
|
||||
cmd_rx: mpsc::Receiver<MeshCommand>,
|
||||
) -> tokio::task::JoinHandle<()> {
|
||||
tokio::spawn(async move {
|
||||
let mut shutdown = shutdown;
|
||||
let mut cmd_rx = cmd_rx;
|
||||
let mut reconnect_delay = RECONNECT_DELAY_INIT;
|
||||
loop {
|
||||
if *shutdown.borrow() {
|
||||
info!("Mesh listener shutting down");
|
||||
return;
|
||||
}
|
||||
|
||||
match session::run_mesh_session(
|
||||
&state,
|
||||
device_path.as_deref(),
|
||||
&our_did,
|
||||
&our_ed_pubkey_hex,
|
||||
&our_x25519_secret,
|
||||
&our_x25519_pubkey_hex,
|
||||
&mut shutdown,
|
||||
&mut cmd_rx,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(()) => {
|
||||
info!("Mesh session ended cleanly");
|
||||
// Session was established before ending — reset backoff
|
||||
reconnect_delay = RECONNECT_DELAY_INIT;
|
||||
}
|
||||
Err(e) => {
|
||||
// Check if session was ever connected (vs failed to open)
|
||||
let was_connected = state.status.read().await.device_connected;
|
||||
if was_connected {
|
||||
reconnect_delay = RECONNECT_DELAY_INIT;
|
||||
}
|
||||
error!("Mesh session error: {} (retry in {:?})", e, reconnect_delay);
|
||||
}
|
||||
}
|
||||
|
||||
// Update status to disconnected
|
||||
{
|
||||
let mut status = state.status.write().await;
|
||||
status.device_connected = false;
|
||||
status.device_path = None;
|
||||
}
|
||||
let _ = state.event_tx.send(MeshEvent::DeviceDisconnected);
|
||||
|
||||
// Wait before reconnecting (exponential backoff)
|
||||
tokio::select! {
|
||||
_ = tokio::time::sleep(reconnect_delay) => {},
|
||||
_ = shutdown.changed() => {
|
||||
if *shutdown.borrow() { return; }
|
||||
},
|
||||
}
|
||||
|
||||
// Increase backoff for next failure, cap at max
|
||||
reconnect_delay = (reconnect_delay * 2).min(RECONNECT_DELAY_MAX);
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,337 @@
|
||||
//! Mesh session lifecycle: connect, initialize, main loop.
|
||||
|
||||
use super::{
|
||||
frames, MeshCommand, MeshState,
|
||||
ADVERT_INTERVAL, MAX_CONSECUTIVE_WRITE_FAILURES, SYNC_INTERVAL,
|
||||
};
|
||||
use super::super::serial::MeshcoreDevice;
|
||||
use super::super::types::*;
|
||||
use anyhow::Result;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use tokio::sync::mpsc;
|
||||
use tracing::{debug, error, info, warn};
|
||||
|
||||
/// Scan all candidate serial ports and open the first Meshcore device found.
|
||||
async fn auto_detect_and_open() -> Result<(String, MeshcoreDevice, DeviceInfo)> {
|
||||
let paths = super::super::serial::detect_serial_devices().await;
|
||||
if paths.is_empty() {
|
||||
anyhow::bail!("No serial devices found in /dev");
|
||||
}
|
||||
for path in &paths {
|
||||
debug!(path = %path, "Probing for Meshcore device");
|
||||
match MeshcoreDevice::open(path).await {
|
||||
Ok(mut dev) => match dev.initialize().await {
|
||||
Ok(info) => {
|
||||
info!(path = %path, firmware = %info.firmware_version, "Found Meshcore device via auto-detect");
|
||||
return Ok((path.clone(), dev, info));
|
||||
}
|
||||
Err(e) => debug!(path = %path, error = %e, "Not a Meshcore device"),
|
||||
},
|
||||
Err(e) => debug!(path = %path, error = %e, "Could not open serial port"),
|
||||
}
|
||||
}
|
||||
anyhow::bail!("No Meshcore device found on {} candidate ports: {:?}", paths.len(), paths)
|
||||
}
|
||||
|
||||
/// Fetch the contacts list from the device and update the peer cache.
|
||||
async fn refresh_contacts(
|
||||
device: &mut MeshcoreDevice,
|
||||
state: &Arc<MeshState>,
|
||||
) {
|
||||
match device.get_contacts().await {
|
||||
Ok(contacts) => {
|
||||
let mut peers = state.peers.write().await;
|
||||
for (idx, contact) in contacts.iter().enumerate() {
|
||||
let contact_id = idx as u32;
|
||||
let existing = peers.get(&contact_id);
|
||||
let peer = super::super::types::MeshPeer {
|
||||
contact_id,
|
||||
advert_name: contact.advert_name.clone(),
|
||||
did: existing.and_then(|p| p.did.clone()),
|
||||
pubkey_hex: Some(contact.public_key_hex.clone()),
|
||||
x25519_pubkey: existing.and_then(|p| p.x25519_pubkey),
|
||||
rssi: None,
|
||||
snr: None,
|
||||
last_heard: chrono::Utc::now().to_rfc3339(),
|
||||
hops: 0,
|
||||
};
|
||||
peers.insert(contact_id, peer);
|
||||
}
|
||||
drop(peers);
|
||||
state.update_peer_count().await;
|
||||
if !contacts.is_empty() {
|
||||
info!(count = contacts.len(), "Refreshed mesh contacts");
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("Failed to fetch contacts: {}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Drain any queued messages from the device.
|
||||
/// Returns `true` if a write/communication error occurred (for failure tracking).
|
||||
async fn sync_queued_messages(
|
||||
device: &mut MeshcoreDevice,
|
||||
state: &Arc<MeshState>,
|
||||
our_x25519_secret: &[u8; 32],
|
||||
) -> bool {
|
||||
match device.sync_messages().await {
|
||||
Ok(frames) => {
|
||||
for frame in &frames {
|
||||
frames::handle_frame(frame, state, our_x25519_secret).await;
|
||||
}
|
||||
if !frames.is_empty() {
|
||||
info!(count = frames.len(), "Synced queued mesh messages");
|
||||
}
|
||||
false
|
||||
}
|
||||
Err(e) => {
|
||||
debug!("Message sync: {}", e);
|
||||
true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Run a single mesh session (connect, initialize, main loop).
|
||||
pub(super) async fn run_mesh_session(
|
||||
state: &Arc<MeshState>,
|
||||
preferred_path: Option<&str>,
|
||||
our_did: &str,
|
||||
_our_ed_pubkey_hex: &str,
|
||||
our_x25519_secret: &[u8; 32],
|
||||
_our_x25519_pubkey_hex: &str,
|
||||
shutdown: &mut tokio::sync::watch::Receiver<bool>,
|
||||
cmd_rx: &mut mpsc::Receiver<MeshCommand>,
|
||||
) -> Result<()> {
|
||||
// Detect device — try preferred path first, fall back to auto-detect
|
||||
let (device_path, mut device, device_info) = if let Some(path) = preferred_path {
|
||||
match MeshcoreDevice::open(path).await {
|
||||
Ok(mut dev) => match dev.initialize().await {
|
||||
Ok(info) => (path.to_string(), dev, info),
|
||||
Err(e) => {
|
||||
warn!("Preferred path {} handshake failed: {} — trying auto-detect", path, e);
|
||||
auto_detect_and_open().await?
|
||||
}
|
||||
},
|
||||
Err(e) => {
|
||||
warn!("Preferred path {} open failed: {} — trying auto-detect", path, e);
|
||||
auto_detect_and_open().await?
|
||||
}
|
||||
}
|
||||
} else {
|
||||
auto_detect_and_open().await?
|
||||
};
|
||||
|
||||
// Update status
|
||||
{
|
||||
let mut status = state.status.write().await;
|
||||
status.device_connected = true;
|
||||
status.device_type = DeviceType::Meshcore;
|
||||
status.device_path = Some(device_path.clone());
|
||||
status.firmware_version = Some(device_info.firmware_version.clone());
|
||||
status.self_node_id = Some(device_info.node_id);
|
||||
status.self_advert_name = device.advert_name.clone();
|
||||
}
|
||||
|
||||
let _ = state.event_tx.send(MeshEvent::DeviceConnected(device_info));
|
||||
|
||||
// Set advert name to something identifiable
|
||||
let short_did = our_did.chars().skip(8).take(8).collect::<String>();
|
||||
let advert_name = format!("Archy-{}", short_did);
|
||||
if let Err(e) = device.set_advert_name(&advert_name).await {
|
||||
warn!("Failed to set advert name: {}", e);
|
||||
}
|
||||
|
||||
// Broadcast our advertisement so other nodes can discover us
|
||||
if let Err(e) = device.send_self_advert().await {
|
||||
warn!("Failed to send initial advert: {}", e);
|
||||
}
|
||||
|
||||
// Fetch existing contacts from the device
|
||||
refresh_contacts(&mut device, state).await;
|
||||
|
||||
// Sync any queued messages from before we connected
|
||||
let _ = sync_queued_messages(&mut device, state, our_x25519_secret).await;
|
||||
|
||||
// Main loop
|
||||
let mut advert_timer = tokio::time::interval(ADVERT_INTERVAL);
|
||||
let mut sync_timer = tokio::time::interval(SYNC_INTERVAL);
|
||||
advert_timer.tick().await; // skip first immediate tick
|
||||
sync_timer.tick().await;
|
||||
let mut consecutive_write_failures: u32 = 0;
|
||||
|
||||
loop {
|
||||
// If too many consecutive writes have failed, the serial port is dead —
|
||||
// bail out so the outer loop can reconnect to a (possibly re-enumerated) device.
|
||||
if consecutive_write_failures >= MAX_CONSECUTIVE_WRITE_FAILURES {
|
||||
error!(
|
||||
failures = consecutive_write_failures,
|
||||
"Serial port unresponsive — triggering reconnection"
|
||||
);
|
||||
anyhow::bail!("Serial port unresponsive after {} consecutive write failures", consecutive_write_failures);
|
||||
}
|
||||
|
||||
tokio::select! {
|
||||
// Check for incoming frames
|
||||
frame_result = device.try_recv_frame() => {
|
||||
match frame_result {
|
||||
Ok(Some(frame)) => {
|
||||
// Successful read resets the failure counter
|
||||
consecutive_write_failures = 0;
|
||||
let should_action = frames::handle_frame(
|
||||
&frame,
|
||||
state,
|
||||
our_x25519_secret,
|
||||
).await;
|
||||
if should_action {
|
||||
// Contact discovery or messages waiting — sync both
|
||||
refresh_contacts(&mut device, state).await;
|
||||
if sync_queued_messages(&mut device, state, our_x25519_secret).await {
|
||||
consecutive_write_failures += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(None) => {
|
||||
// No complete frame yet, that's fine
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
}
|
||||
Err(e) => {
|
||||
error!("Serial read error: {}", e);
|
||||
return Err(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Periodic advertisement broadcast + contact refresh
|
||||
_ = advert_timer.tick() => {
|
||||
debug!("Periodic self-advert broadcast");
|
||||
if let Err(e) = device.send_self_advert().await {
|
||||
consecutive_write_failures += 1;
|
||||
warn!(failures = consecutive_write_failures, "Failed to send advert: {}", e);
|
||||
} else {
|
||||
consecutive_write_failures = 0;
|
||||
}
|
||||
refresh_contacts(&mut device, state).await;
|
||||
}
|
||||
|
||||
// Process send commands from MeshService
|
||||
Some(cmd) = cmd_rx.recv() => {
|
||||
handle_send_command(cmd, &mut device, state, &mut consecutive_write_failures).await;
|
||||
}
|
||||
|
||||
// Periodic message sync
|
||||
_ = sync_timer.tick() => {
|
||||
if sync_queued_messages(&mut device, state, our_x25519_secret).await {
|
||||
consecutive_write_failures += 1;
|
||||
debug!(failures = consecutive_write_failures, "Message sync failed");
|
||||
} else {
|
||||
consecutive_write_failures = 0;
|
||||
}
|
||||
}
|
||||
|
||||
// Shutdown signal
|
||||
_ = shutdown.changed() => {
|
||||
if *shutdown.borrow() {
|
||||
info!("Mesh listener received shutdown signal");
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Process a single outbound command from MeshService.
|
||||
async fn handle_send_command(
|
||||
cmd: MeshCommand,
|
||||
device: &mut MeshcoreDevice,
|
||||
state: &Arc<MeshState>,
|
||||
consecutive_write_failures: &mut u32,
|
||||
) {
|
||||
match cmd {
|
||||
MeshCommand::SendText { dest_pubkey_prefix, payload } => {
|
||||
if let Err(e) = device.send_text(&dest_pubkey_prefix, &payload).await {
|
||||
*consecutive_write_failures += 1;
|
||||
warn!(failures = *consecutive_write_failures, "Failed to send text via mesh: {}", e);
|
||||
} else {
|
||||
*consecutive_write_failures = 0;
|
||||
info!(dest = %hex::encode(dest_pubkey_prefix), len = payload.len(), "Sent mesh message");
|
||||
}
|
||||
}
|
||||
MeshCommand::SendRaw { dest_pubkey_prefix, payload } => {
|
||||
// Apply steganographic encoding if configured
|
||||
let wire_payload = if state.stego_mode != super::super::steganography::SteganographyMode::Normal
|
||||
&& payload.first() == Some(&super::super::message_types::TYPED_MESSAGE_MARKER)
|
||||
{
|
||||
match super::super::steganography::encode_typed_wire(state.stego_mode, &payload) {
|
||||
Ok(stego) => stego,
|
||||
Err(e) => {
|
||||
warn!("Stego encode failed, sending plain: {}", e);
|
||||
payload
|
||||
}
|
||||
}
|
||||
} else {
|
||||
payload
|
||||
};
|
||||
// Base64 encode, then chunk if >140 chars (LoRa 160 byte limit)
|
||||
use base64::Engine;
|
||||
let encoded = base64::engine::general_purpose::STANDARD.encode(&wire_payload);
|
||||
|
||||
if encoded.len() <= 140 {
|
||||
// Single frame — fits in one LoRa packet
|
||||
if let Err(e) = device.send_text(&dest_pubkey_prefix, encoded.as_bytes()).await {
|
||||
*consecutive_write_failures += 1;
|
||||
warn!(failures = *consecutive_write_failures, "Failed to send raw via mesh: {}", e);
|
||||
} else {
|
||||
*consecutive_write_failures = 0;
|
||||
info!(dest = %hex::encode(dest_pubkey_prefix), len = encoded.len(), "Sent raw mesh message");
|
||||
}
|
||||
} else {
|
||||
// Multi-frame chunking: "MCxxyyzz..." where xx=msg_id, yy=chunk_idx, zz=total_chunks
|
||||
static CHUNK_MSG_ID: std::sync::atomic::AtomicU8 = std::sync::atomic::AtomicU8::new(0);
|
||||
let msg_id = CHUNK_MSG_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
||||
let chunk_data_size = 132; // 160 - 8 header bytes ("MCxxyyzz") = 152, leave margin
|
||||
let chunks: Vec<&str> = encoded.as_bytes().chunks(chunk_data_size)
|
||||
.map(|c| std::str::from_utf8(c).unwrap_or(""))
|
||||
.collect();
|
||||
let total = chunks.len() as u8;
|
||||
info!(
|
||||
dest = %hex::encode(dest_pubkey_prefix),
|
||||
raw_len = wire_payload.len(),
|
||||
b64_len = encoded.len(),
|
||||
chunks = total,
|
||||
"Sending chunked mesh message"
|
||||
);
|
||||
for (idx, chunk) in chunks.iter().enumerate() {
|
||||
let frame = format!("MC{:02x}{:02x}{:02x}{}", msg_id, idx as u8, total, chunk);
|
||||
if let Err(e) = device.send_text(&dest_pubkey_prefix, frame.as_bytes()).await {
|
||||
*consecutive_write_failures += 1;
|
||||
warn!(failures = *consecutive_write_failures, chunk = idx, "Chunk send failed: {}", e);
|
||||
break;
|
||||
}
|
||||
// Small delay between chunks to avoid overwhelming the radio
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
}
|
||||
*consecutive_write_failures = 0;
|
||||
}
|
||||
}
|
||||
MeshCommand::BroadcastChannel { channel, payload } => {
|
||||
if let Err(e) = device.send_channel_text(channel, &payload).await {
|
||||
*consecutive_write_failures += 1;
|
||||
warn!(failures = *consecutive_write_failures, "Failed to broadcast on channel {}: {}", channel, e);
|
||||
} else {
|
||||
*consecutive_write_failures = 0;
|
||||
info!(channel, len = payload.len(), "Broadcast on mesh channel");
|
||||
}
|
||||
}
|
||||
MeshCommand::SendAdvert => {
|
||||
if let Err(e) = device.send_self_advert().await {
|
||||
*consecutive_write_failures += 1;
|
||||
warn!(failures = *consecutive_write_failures, "Failed to send advert: {}", e);
|
||||
} else {
|
||||
*consecutive_write_failures = 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user