feat(mesh): Telegram primitives pass + attachment transport router
Bundles the Phase 2b/3/4/5 work that accumulated across prior sessions and the new attachment chunking router from this session. Everything ships in one shot so the full mesh surface stays coherent on-wire. Telegram primitives (variants 13–18, 20–22): - Reply / Reaction / ReadReceipt / Forward / Edit / Delete - Presence heartbeat + last-seen tracking - ChannelInvite + ContactCard payload types - MessageKey (sender_pubkey, sender_seq) as cross-transport identity - Action menu, reply banner, edit banner, tombstones, (edited) marker - Debounced auto-read-receipts on scroll + message arrival Activated prototypes (Phase 4): - PsbtHash send RPC - Contacts CRUD (in-memory alias/notes/pinned/blocked) - Outbox 📤 badge, rotate-prekeys button - Chunked send fallback (MCIIXXTT framing) as auto-failover inside send_typed_wire when a typed wire exceeds the LoRa per-frame budget Unified inbox (Phase 1): - conversations.list + conversations.messages RPCs (UI collapse deferred) Attachment transport router (new this session): - ContentInline variant 23 + ContentInlinePayload carrying file bytes directly in the envelope for small files with no Tor path - mesh.send-content-inline RPC — mirrors to local BlobStore, rides send_typed_wire which auto-chunks over MCIIXXTT framing (~2.3 KB cap) - mesh.transport-advice RPC as single source of truth for tier decisions: auto-mesh / choose / tor-only / impossible - Receive arm writes inline bytes to local BlobStore so the existing content_ref card renderer handles both transports uniformly - MeshState.blob_store field + order-independent propagation from RpcHandler::set_blob_store / set_mesh_service - Frontend handleAttachFile calls advice first, branches into silent auto-send, transport-chooser modal, Tor-only path, or red error - Transport modal with 📡 mesh / 🧅 Tor options + ETA + disabled state when peer has no Tor reachability Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dorian
parent
5616bb74e6
commit
6760d11a57
@ -261,6 +261,9 @@ impl RpcHandler {
|
||||
"federation.peer-address-changed" => self.handle_federation_peer_address_changed(params).await,
|
||||
"federation.notify-did-change" => self.handle_federation_notify_did_change(params).await,
|
||||
"federation.peer-did-changed" => self.handle_federation_peer_did_changed(params).await,
|
||||
"federation.list-pending-requests" => self.handle_federation_list_pending_requests().await,
|
||||
"federation.approve-request" => self.handle_federation_approve_request(params).await,
|
||||
"federation.reject-request" => self.handle_federation_reject_request(params).await,
|
||||
|
||||
// VPN & Remote Access
|
||||
"vpn.status" => self.handle_vpn_status().await,
|
||||
@ -296,6 +299,8 @@ impl RpcHandler {
|
||||
"mesh.send-coordinate" => self.handle_mesh_send_coordinate(params).await,
|
||||
"mesh.send-alert" => self.handle_mesh_send_alert(params).await,
|
||||
"mesh.send-content" => self.handle_mesh_send_content(params).await,
|
||||
"mesh.send-content-inline" => self.handle_mesh_send_content_inline(params).await,
|
||||
"mesh.transport-advice" => self.handle_mesh_transport_advice(params).await,
|
||||
"mesh.fetch-content" => self.handle_mesh_fetch_content(params).await,
|
||||
"mesh.send-reply" => self.handle_mesh_send_reply(params).await,
|
||||
"mesh.send-reaction" => self.handle_mesh_send_reaction(params).await,
|
||||
|
||||
@ -59,9 +59,7 @@ impl RpcHandler {
|
||||
{
|
||||
use base64::Engine;
|
||||
let b64 = base64::engine::general_purpose::STANDARD.encode(&payload);
|
||||
let _ = shared_state
|
||||
.cmd_tx
|
||||
.send(crate::mesh::listener::MeshCommand::BroadcastChannel {
|
||||
let _ = shared_state.send_cmd(crate::mesh::listener::MeshCommand::BroadcastChannel {
|
||||
channel: 0,
|
||||
payload: b64.into_bytes(),
|
||||
})
|
||||
@ -95,9 +93,7 @@ impl RpcHandler {
|
||||
wire.clone()
|
||||
};
|
||||
|
||||
let _ = svc.shared_state()
|
||||
.cmd_tx
|
||||
.send(crate::mesh::listener::MeshCommand::SendRaw {
|
||||
let _ = svc.shared_state().send_cmd(crate::mesh::listener::MeshCommand::SendRaw {
|
||||
dest_pubkey_prefix: prefix,
|
||||
payload,
|
||||
})
|
||||
@ -243,9 +239,7 @@ impl RpcHandler {
|
||||
wire.clone()
|
||||
};
|
||||
|
||||
let _ = svc.shared_state()
|
||||
.cmd_tx
|
||||
.send(crate::mesh::listener::MeshCommand::SendRaw {
|
||||
let _ = svc.shared_state().send_cmd(crate::mesh::listener::MeshCommand::SendRaw {
|
||||
dest_pubkey_prefix: prefix,
|
||||
payload,
|
||||
})
|
||||
|
||||
@ -185,7 +185,7 @@ impl RpcHandler {
|
||||
if pk_bytes.len() >= 6 {
|
||||
let mut prefix = [0u8; 6];
|
||||
prefix.copy_from_slice(&pk_bytes[..6]);
|
||||
let _ = svc.shared_state().cmd_tx.send(
|
||||
let _ = svc.shared_state().send_cmd(
|
||||
crate::mesh::listener::MeshCommand::SendRaw {
|
||||
dest_pubkey_prefix: prefix,
|
||||
payload: wire,
|
||||
|
||||
@ -1,9 +1,9 @@
|
||||
use super::super::RpcHandler;
|
||||
use crate::blobs::DEFAULT_CAP_TTL_SECS;
|
||||
use crate::mesh::message_types::{
|
||||
self, AlertPayload, AlertType, ChannelInvitePayload, ContentRefPayload, Coordinate,
|
||||
DeletePayload, EditPayload, ForwardPayload, InvoicePayload, MessageKey, MeshMessageType,
|
||||
PsbtHashPayload, ReactionPayload, ReadReceiptPayload, ReplyPayload,
|
||||
self, AlertPayload, AlertType, ChannelInvitePayload, ContentInlinePayload, ContentRefPayload,
|
||||
Coordinate, DeletePayload, EditPayload, ForwardPayload, InvoicePayload, MessageKey,
|
||||
MeshMessageType, PsbtHashPayload, ReactionPayload, ReadReceiptPayload, ReplyPayload,
|
||||
TypedEnvelope,
|
||||
};
|
||||
use anyhow::Result;
|
||||
@ -352,6 +352,201 @@ impl RpcHandler {
|
||||
}))
|
||||
}
|
||||
|
||||
/// mesh.send-content-inline — Carry file bytes directly in a typed envelope.
|
||||
/// Params: { contact_id, mime, filename?, caption?, bytes_b64 }. The
|
||||
/// underlying `send_typed_wire` auto-chunks via MCIIXXTT framing when the
|
||||
/// envelope exceeds the LoRa per-frame budget. Sender also writes the
|
||||
/// blob to its own BlobStore so the chat history renders identically to
|
||||
/// ContentRef on both sides.
|
||||
pub(in crate::api::rpc) async fn handle_mesh_send_content_inline(
|
||||
&self,
|
||||
params: Option<serde_json::Value>,
|
||||
) -> Result<serde_json::Value> {
|
||||
use base64::{engine::general_purpose::STANDARD as B64, Engine as _};
|
||||
|
||||
let params = params.ok_or_else(|| anyhow::anyhow!("Missing params"))?;
|
||||
let contact_id = params["contact_id"]
|
||||
.as_u64()
|
||||
.ok_or_else(|| anyhow::anyhow!("Missing contact_id"))? as u32;
|
||||
let mime = params["mime"]
|
||||
.as_str()
|
||||
.unwrap_or("application/octet-stream")
|
||||
.to_string();
|
||||
let filename = params["filename"].as_str().map(|s| s.to_string());
|
||||
let caption = params["caption"].as_str().map(|s| s.to_string());
|
||||
let bytes_b64 = params["bytes_b64"]
|
||||
.as_str()
|
||||
.ok_or_else(|| anyhow::anyhow!("Missing bytes_b64"))?;
|
||||
let bytes = B64
|
||||
.decode(bytes_b64)
|
||||
.map_err(|e| anyhow::anyhow!("Invalid base64: {}", e))?;
|
||||
|
||||
// Hard ceiling matching the chunked-send capacity (~20 chunks * 152
|
||||
// b64 chars after MCIIXXTT framing). Anything larger must go via
|
||||
// ContentRef over Tor.
|
||||
const INLINE_HARD_MAX: usize = 2300;
|
||||
if bytes.len() > INLINE_HARD_MAX {
|
||||
anyhow::bail!(
|
||||
"Payload {} bytes exceeds inline max {} — use mesh.send-content (ContentRef) instead",
|
||||
bytes.len(),
|
||||
INLINE_HARD_MAX
|
||||
);
|
||||
}
|
||||
|
||||
// Mirror to local BlobStore so the Sent record renders the same
|
||||
// attachment card as the receiver's.
|
||||
let blob_store = {
|
||||
let guard = self.blob_store.read().await;
|
||||
guard
|
||||
.as_ref()
|
||||
.ok_or_else(|| anyhow::anyhow!("Blob store not initialised"))?
|
||||
.clone()
|
||||
};
|
||||
let meta = blob_store
|
||||
.put(&bytes, &mime, filename.clone(), None)
|
||||
.await?;
|
||||
|
||||
let service = self.mesh_service.read().await;
|
||||
let svc = service
|
||||
.as_ref()
|
||||
.ok_or_else(|| anyhow::anyhow!("Mesh service not running"))?;
|
||||
|
||||
let content = ContentInlinePayload {
|
||||
mime: mime.clone(),
|
||||
filename: filename.clone(),
|
||||
caption: caption.clone(),
|
||||
bytes,
|
||||
};
|
||||
let seq = svc.next_send_seq(contact_id).await;
|
||||
let payload = message_types::encode_payload(&content)?;
|
||||
let envelope = TypedEnvelope::new(MeshMessageType::ContentInline, payload).with_seq(seq);
|
||||
let wire = envelope.to_wire()?;
|
||||
|
||||
let display = match (&filename, &caption) {
|
||||
(Some(f), Some(c)) => format!("📎 {} — {}", f, c),
|
||||
(Some(f), None) => format!("📎 {}", f),
|
||||
(None, Some(c)) => format!("📎 {}", c),
|
||||
(None, None) => format!("📎 {} ({} bytes)", mime, meta.size),
|
||||
};
|
||||
// Render as a content_ref card on the sender side (UI already knows
|
||||
// how to draw it from cid + mime + filename + size).
|
||||
let typed_json = serde_json::json!({
|
||||
"cid": meta.cid,
|
||||
"size": meta.size,
|
||||
"mime": mime,
|
||||
"filename": filename,
|
||||
"caption": caption,
|
||||
"inline": true,
|
||||
});
|
||||
|
||||
let msg = svc
|
||||
.send_typed_wire(
|
||||
contact_id,
|
||||
wire,
|
||||
"content_ref",
|
||||
&display,
|
||||
Some(typed_json),
|
||||
seq,
|
||||
)
|
||||
.await?;
|
||||
|
||||
info!(
|
||||
contact_id,
|
||||
size = meta.size,
|
||||
cid = %meta.cid,
|
||||
"Sent content_inline over mesh"
|
||||
);
|
||||
Ok(serde_json::json!({
|
||||
"sent": true,
|
||||
"message_id": msg.id,
|
||||
"cid": meta.cid,
|
||||
"size": meta.size,
|
||||
}))
|
||||
}
|
||||
|
||||
/// mesh.transport-advice — Recommend how to send an attachment of a given
|
||||
/// size to a given peer. Single source of truth for the frontend tier
|
||||
/// router. Params: { contact_id, size }. Returns:
|
||||
/// { tier, est_seconds, has_tor, reason }
|
||||
/// where tier ∈ "auto-mesh" | "choose" | "tor-only" | "impossible".
|
||||
pub(in crate::api::rpc) async fn handle_mesh_transport_advice(
|
||||
&self,
|
||||
params: Option<serde_json::Value>,
|
||||
) -> Result<serde_json::Value> {
|
||||
let params = params.ok_or_else(|| anyhow::anyhow!("Missing params"))?;
|
||||
let contact_id = params["contact_id"]
|
||||
.as_u64()
|
||||
.ok_or_else(|| anyhow::anyhow!("Missing contact_id"))? as u32;
|
||||
let size = params["size"]
|
||||
.as_u64()
|
||||
.ok_or_else(|| anyhow::anyhow!("Missing size"))?;
|
||||
|
||||
// Knobs — keep in sync with the frontend modal copy.
|
||||
const MESH_AUTO_MAX: u64 = 1024;
|
||||
const MESH_HARD_MAX: u64 = 2300;
|
||||
const TOR_LARGE_WARN: u64 = 5 * 1024 * 1024;
|
||||
const LORA_BYTES_PER_SEC: u64 = 50;
|
||||
|
||||
// Resolve peer Tor reachability via federation node list.
|
||||
let service = self.mesh_service.read().await;
|
||||
let svc = service
|
||||
.as_ref()
|
||||
.ok_or_else(|| anyhow::anyhow!("Mesh service not running"))?;
|
||||
let state = svc.shared_state();
|
||||
let (peer_pubkey_hex, peer_did) = {
|
||||
let peers = state.peers.read().await;
|
||||
match peers.get(&contact_id) {
|
||||
Some(p) => (p.pubkey_hex.clone(), p.did.clone()),
|
||||
None => (None, None),
|
||||
}
|
||||
};
|
||||
let nodes = crate::federation::load_nodes(&self.config.data_dir)
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
let has_tor = peer_pubkey_hex
|
||||
.as_ref()
|
||||
.map(|pk| nodes.iter().any(|n| &n.pubkey == pk))
|
||||
.unwrap_or(false)
|
||||
|| peer_did
|
||||
.as_ref()
|
||||
.map(|d| nodes.iter().any(|n| &n.did == d))
|
||||
.unwrap_or(false);
|
||||
|
||||
let est_seconds = (size.saturating_add(LORA_BYTES_PER_SEC - 1) / LORA_BYTES_PER_SEC).max(1);
|
||||
|
||||
let (tier, reason) = if size <= MESH_AUTO_MAX {
|
||||
("auto-mesh", "Small enough to send inline over mesh")
|
||||
} else if size <= MESH_HARD_MAX {
|
||||
if has_tor {
|
||||
("choose", "Fits over mesh (slow) or Tor (instant)")
|
||||
} else {
|
||||
("auto-mesh", "No Tor path — sending inline over mesh")
|
||||
}
|
||||
} else if size <= TOR_LARGE_WARN {
|
||||
if has_tor {
|
||||
("tor-only", "Too large for mesh — Tor only")
|
||||
} else {
|
||||
("impossible", "Too large for mesh, and peer has no Tor path")
|
||||
}
|
||||
} else {
|
||||
if has_tor {
|
||||
("tor-only", "Large file — receiver fetch may be slow")
|
||||
} else {
|
||||
("impossible", "Too large, and peer has no Tor path")
|
||||
}
|
||||
};
|
||||
|
||||
Ok(serde_json::json!({
|
||||
"tier": tier,
|
||||
"est_seconds": est_seconds,
|
||||
"has_tor": has_tor,
|
||||
"reason": reason,
|
||||
"size": size,
|
||||
"mesh_auto_max": MESH_AUTO_MAX,
|
||||
"mesh_hard_max": MESH_HARD_MAX,
|
||||
}))
|
||||
}
|
||||
|
||||
/// mesh.send-reply — Send a text reply targeted at an earlier message.
|
||||
/// Params: { contact_id, target_pubkey, target_seq, text }. The target
|
||||
/// MessageKey identifies the message being replied to; it does NOT need
|
||||
|
||||
@ -142,6 +142,12 @@ impl RpcHandler {
|
||||
|
||||
/// Set the mesh service (called after identity is loaded).
|
||||
pub async fn set_mesh_service(&self, service: crate::mesh::MeshService) {
|
||||
// If the blob store is already initialised, propagate it into the
|
||||
// freshly-started mesh state so the listener can persist inline
|
||||
// attachments. Mirrors `set_blob_store`'s forward-propagation.
|
||||
if let Some(store) = self.blob_store.read().await.as_ref().cloned() {
|
||||
*service.shared_state().blob_store.write().await = Some(store);
|
||||
}
|
||||
*self.mesh_service.write().await = Some(service);
|
||||
}
|
||||
|
||||
@ -153,8 +159,13 @@ impl RpcHandler {
|
||||
/// Share the blob store + our pubkey so mesh.send-content / fetch-content
|
||||
/// can reach them. Called once from ApiHandler::new.
|
||||
pub async fn set_blob_store(&self, store: Arc<crate::blobs::BlobStore>, self_pubkey_hex: String) {
|
||||
*self.blob_store.write().await = Some(store);
|
||||
*self.blob_store.write().await = Some(store.clone());
|
||||
*self.self_pubkey_hex.write().await = Some(self_pubkey_hex);
|
||||
// Propagate into a running mesh service if one is already up — keeps
|
||||
// `set_blob_store` and `set_mesh_service` order-independent.
|
||||
if let Some(svc) = self.mesh_service.read().await.as_ref() {
|
||||
*svc.shared_state().blob_store.write().await = Some(store);
|
||||
}
|
||||
}
|
||||
|
||||
/// Get reference to the mesh service Arc (for MeshTransport wrapper).
|
||||
|
||||
@ -331,7 +331,7 @@ async fn send_to_peer(state: &Arc<MeshState>, contact_id: u32, typed_wire: Vec<u
|
||||
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 {
|
||||
let _ = state.send_cmd(MeshCommand::SendRaw {
|
||||
dest_pubkey_prefix: prefix,
|
||||
payload,
|
||||
}).await;
|
||||
@ -342,7 +342,7 @@ async fn send_to_peer(state: &Arc<MeshState>, contact_id: u32, typed_wire: Vec<u
|
||||
}
|
||||
drop(peers);
|
||||
// Broadcast fallback — plaintext (no specific peer to encrypt for)
|
||||
let _ = state.cmd_tx.send(MeshCommand::BroadcastChannel {
|
||||
let _ = state.send_cmd(MeshCommand::BroadcastChannel {
|
||||
channel: 0,
|
||||
payload: typed_wire,
|
||||
}).await;
|
||||
|
||||
@ -145,6 +145,23 @@ fn try_decrypt_typed(
|
||||
None
|
||||
}
|
||||
|
||||
/// Cheap structural check: does this payload look like an `MCxxyyzz…`
|
||||
/// chunk frame? Used by the receive dispatcher to decide whether a `None`
|
||||
/// from `try_chunk_reassemble` means "not a chunk" (fall through to the
|
||||
/// plaintext store) or "chunk buffered, waiting for more frames" (do not
|
||||
/// store anything yet — partial chunks must never be persisted as their
|
||||
/// own messages, which was the cause of the raw `MC0301…` bubbles users
|
||||
/// were seeing in chat).
|
||||
pub(super) fn is_mc_chunk_frame(payload: &[u8]) -> bool {
|
||||
let Ok(text) = std::str::from_utf8(payload) else { return false };
|
||||
if !text.starts_with("MC") || text.len() < 8 {
|
||||
return false;
|
||||
}
|
||||
u8::from_str_radix(&text[2..4], 16).is_ok()
|
||||
&& u8::from_str_radix(&text[4..6], 16).is_ok()
|
||||
&& u8::from_str_radix(&text[6..8], 16).is_ok()
|
||||
}
|
||||
|
||||
/// 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.
|
||||
|
||||
@ -478,6 +478,47 @@ pub(crate) async fn handle_typed_envelope_direct(
|
||||
}
|
||||
}
|
||||
|
||||
Some(MeshMessageType::ContentInline) => {
|
||||
match message_types::decode_payload::<message_types::ContentInlinePayload>(&envelope.v) {
|
||||
Ok(content) => {
|
||||
let text = match (&content.filename, &content.caption) {
|
||||
(Some(fname), Some(c)) => format!("📎 {} — {}", fname, c),
|
||||
(Some(fname), None) => format!("📎 {}", fname),
|
||||
(None, Some(c)) => format!("📎 {}", c),
|
||||
(None, None) => format!("📎 {} ({} bytes)", content.mime, content.bytes.len()),
|
||||
};
|
||||
// Write bytes to local BlobStore so the standard
|
||||
// content_ref renderer can display it via /blob/<cid>.
|
||||
let mut cid_opt: Option<String> = None;
|
||||
let mut size: u64 = content.bytes.len() as u64;
|
||||
if let Some(store) = state.blob_store.read().await.as_ref().cloned() {
|
||||
match store
|
||||
.put(&content.bytes, &content.mime, content.filename.clone(), None)
|
||||
.await
|
||||
{
|
||||
Ok(meta) => {
|
||||
cid_opt = Some(meta.cid);
|
||||
size = meta.size;
|
||||
}
|
||||
Err(e) => warn!("Failed to persist inline attachment: {}", e),
|
||||
}
|
||||
} else {
|
||||
warn!("Blob store not set on mesh state; inline attachment not persisted");
|
||||
}
|
||||
let json = serde_json::json!({
|
||||
"cid": cid_opt,
|
||||
"size": size,
|
||||
"mime": content.mime,
|
||||
"filename": content.filename,
|
||||
"caption": content.caption,
|
||||
"inline": true,
|
||||
});
|
||||
store_typed_message(state, sender_contact_id, sender_name, &text, "content_ref", Some(json), Some(envelope.seq)).await;
|
||||
}
|
||||
Err(e) => warn!("Failed to decode content_inline payload: {}", e),
|
||||
}
|
||||
}
|
||||
|
||||
Some(MeshMessageType::ContentRef) => {
|
||||
match message_types::decode_payload::<message_types::ContentRefPayload>(&envelope.v) {
|
||||
Ok(content) => {
|
||||
|
||||
@ -1,8 +1,8 @@
|
||||
//! 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,
|
||||
is_mc_chunk_frame, 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;
|
||||
@ -136,7 +136,129 @@ async fn handle_channel_payload(
|
||||
channel_idx: u8,
|
||||
payload: &[u8],
|
||||
) {
|
||||
// DM-via-channel wrapper: [marker(1)][dest_prefix(6)][inner…]
|
||||
// DM-via-channel wrapper (text form): the channel text carries an
|
||||
// ASCII "@DM:<base64>" token somewhere in the body. We locate the
|
||||
// marker anywhere in the payload (the firmware auto-prepends the
|
||||
// sender's `"<advert_name>: "` before our bytes, so the marker is
|
||||
// not at offset 0), then base64-decode to get `[dest(6)][inner]`.
|
||||
// Using a text marker + base64 avoids the C-string NUL truncation
|
||||
// that broke the previous raw-byte wrapper on the firmware's side.
|
||||
let text_view = std::str::from_utf8(payload).ok();
|
||||
|
||||
// v2 format: `@DM2:` + base64(`[dest(6)][sender_arch(6)][inner…]`).
|
||||
// Carries a sender prefix so we can attribute the message to the real
|
||||
// sender's contact_id instead of guessing — fixes the long-standing
|
||||
// bug where every inbound DM-via-channel was misattributed to whichever
|
||||
// `Archy-*` peer happened to have the lowest contact_id in the firmware
|
||||
// contact table (the third-device thread on .39/.76).
|
||||
if let Some(idx) = text_view.and_then(|t| t.find("@DM2:")) {
|
||||
use base64::Engine;
|
||||
let b64 = &text_view.unwrap()[idx + 5..];
|
||||
let b64 = b64.trim_end_matches(|c: char| c == '\0' || c.is_whitespace());
|
||||
match base64::engine::general_purpose::STANDARD.decode(b64) {
|
||||
Ok(body) if body.len() >= 12 => {
|
||||
let dest_prefix: [u8; 6] = body[..6].try_into().expect("sliced 6 bytes");
|
||||
let sender_prefix: [u8; 6] = body[6..12].try_into().expect("sliced 6 bytes");
|
||||
let inner_vec = body[12..].to_vec();
|
||||
let inner: &[u8] = &inner_vec;
|
||||
let addressed_to_us = dest_prefix_is_us(state, &dest_prefix).await;
|
||||
if !addressed_to_us {
|
||||
debug!(
|
||||
dest = %hex::encode(dest_prefix),
|
||||
inner_len = inner.len(),
|
||||
"Dropping DM2-via-channel (not for us)"
|
||||
);
|
||||
return;
|
||||
}
|
||||
info!(
|
||||
dest = %hex::encode(dest_prefix),
|
||||
sender = %hex::encode(sender_prefix),
|
||||
inner_len = inner.len(),
|
||||
channel = channel_idx,
|
||||
"Received DM2 via channel (addressed to us)"
|
||||
);
|
||||
let (contact_id, name) =
|
||||
resolve_sender_by_arch_prefix(state, &sender_prefix, &dest_prefix).await;
|
||||
if TypedEnvelope::is_typed(inner) {
|
||||
handle_typed_message(inner, contact_id, &name, state).await;
|
||||
} else if let Some(decoded) = try_base64_typed(inner) {
|
||||
handle_typed_message(&decoded, contact_id, &name, state).await;
|
||||
} else if let Some(decoded) = try_chunk_reassemble(inner, contact_id, state).await {
|
||||
if TypedEnvelope::is_typed(&decoded) {
|
||||
handle_typed_message(&decoded, contact_id, &name, state).await;
|
||||
} else {
|
||||
let text = String::from_utf8_lossy(&decoded).to_string();
|
||||
store_plain_message(state, contact_id, &name, &text).await;
|
||||
}
|
||||
} else if is_mc_chunk_frame(inner) {
|
||||
// Chunk buffered for reassembly — do not store the raw
|
||||
// `MCxxyyzz…` frame as its own plaintext message. Wait
|
||||
// for the rest of the chunks to arrive.
|
||||
debug!(inner_len = inner.len(), "DM2 chunk buffered");
|
||||
} else {
|
||||
let text = String::from_utf8_lossy(inner).to_string();
|
||||
store_plain_message(state, contact_id, &name, &text).await;
|
||||
}
|
||||
return;
|
||||
}
|
||||
Ok(_) => debug!("DM2-via-channel b64 decoded too short"),
|
||||
Err(e) => debug!("DM2-via-channel b64 decode failed: {}", e),
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(idx) = text_view.and_then(|t| t.find("@DM:")) {
|
||||
use base64::Engine;
|
||||
let b64 = &text_view.unwrap()[idx + 4..];
|
||||
// Trim any trailing whitespace / NULs that firmware may append.
|
||||
let b64 = b64.trim_end_matches(|c: char| c == '\0' || c.is_whitespace());
|
||||
match base64::engine::general_purpose::STANDARD.decode(b64) {
|
||||
Ok(body) if body.len() >= 6 => {
|
||||
let dest_prefix: [u8; 6] = body[..6]
|
||||
.try_into()
|
||||
.expect("sliced 6 bytes");
|
||||
let inner_vec = body[6..].to_vec();
|
||||
let inner: &[u8] = &inner_vec;
|
||||
let addressed_to_us = dest_prefix_is_us(state, &dest_prefix).await;
|
||||
if !addressed_to_us {
|
||||
debug!(
|
||||
dest = %hex::encode(dest_prefix),
|
||||
inner_len = inner.len(),
|
||||
"Dropping DM-via-channel (not for us)"
|
||||
);
|
||||
return;
|
||||
}
|
||||
info!(
|
||||
dest = %hex::encode(dest_prefix),
|
||||
inner_len = inner.len(),
|
||||
channel = channel_idx,
|
||||
"Received DM via channel (addressed to us)"
|
||||
);
|
||||
let (contact_id, name) = resolve_counterparty(state, &dest_prefix).await;
|
||||
if TypedEnvelope::is_typed(inner) {
|
||||
handle_typed_message(inner, contact_id, &name, state).await;
|
||||
} else if let Some(decoded) = try_base64_typed(inner) {
|
||||
handle_typed_message(&decoded, contact_id, &name, state).await;
|
||||
} else if let Some(decoded) = try_chunk_reassemble(inner, contact_id, state).await {
|
||||
if TypedEnvelope::is_typed(&decoded) {
|
||||
handle_typed_message(&decoded, contact_id, &name, state).await;
|
||||
} else {
|
||||
let text = String::from_utf8_lossy(&decoded).to_string();
|
||||
store_plain_message(state, contact_id, &name, &text).await;
|
||||
}
|
||||
} else if is_mc_chunk_frame(inner) {
|
||||
debug!(inner_len = inner.len(), "DM chunk buffered");
|
||||
} else {
|
||||
let text = String::from_utf8_lossy(inner).to_string();
|
||||
store_plain_message(state, contact_id, &name, &text).await;
|
||||
}
|
||||
return;
|
||||
}
|
||||
Ok(_) => debug!("DM-via-channel b64 decoded too short"),
|
||||
Err(e) => debug!("DM-via-channel b64 decode failed: {}", e),
|
||||
}
|
||||
}
|
||||
|
||||
// Legacy raw-byte wrapper kept as a defensive no-op.
|
||||
if payload.len() >= 7 && payload[0] == protocol::DM_VIA_CHANNEL_MARKER {
|
||||
let dest_prefix: [u8; 6] = payload[1..7]
|
||||
.try_into()
|
||||
@ -234,6 +356,34 @@ async fn dest_prefix_is_us(state: &Arc<MeshState>, dest_prefix: &[u8; 6]) -> boo
|
||||
true
|
||||
}
|
||||
|
||||
/// Look up the contact_id for a `@DM2:` sender by matching the 6-byte
|
||||
/// archipelago ed25519 prefix against `state.peers`. Federation-seeded
|
||||
/// peers (added by `mesh::upsert_federation_peer` at startup and after
|
||||
/// every federation mutation) carry the archipelago key in `pubkey_hex`,
|
||||
/// so the prefix lookup resolves to the unified federation chat thread —
|
||||
/// which is exactly where we want both LoRa-arrived and Tor-arrived
|
||||
/// messages from the same peer to land. If no peer matches (sender isn't
|
||||
/// in our federation list yet), we fall through to `resolve_counterparty`
|
||||
/// so the message still lands somewhere visible rather than being dropped.
|
||||
async fn resolve_sender_by_arch_prefix(
|
||||
state: &Arc<MeshState>,
|
||||
sender_arch_prefix: &[u8; 6],
|
||||
dest_prefix: &[u8; 6],
|
||||
) -> (u32, String) {
|
||||
let prefix_hex = hex::encode(sender_arch_prefix);
|
||||
let peers = state.peers.read().await;
|
||||
if let Some(p) = peers.values().find(|p| {
|
||||
p.pubkey_hex
|
||||
.as_deref()
|
||||
.map(|k| k.starts_with(&prefix_hex))
|
||||
.unwrap_or(false)
|
||||
}) {
|
||||
return (p.contact_id, p.advert_name.clone());
|
||||
}
|
||||
drop(peers);
|
||||
resolve_counterparty(state, dest_prefix).await
|
||||
}
|
||||
|
||||
/// Pick a "counterparty" contact_id when dispatching a DM-via-channel
|
||||
/// whose sender we don't otherwise know. We look for any archipelago
|
||||
/// (type-1, "Archy-*") peer in the contact table whose prefix ISN'T the
|
||||
@ -243,20 +393,31 @@ async fn resolve_counterparty(
|
||||
state: &Arc<MeshState>,
|
||||
dest_prefix: &[u8; 6],
|
||||
) -> (u32, String) {
|
||||
// Collect every `Archy-*` peer whose meshcore pubkey-prefix differs
|
||||
// from dest_prefix (dest is ours, so "not dest" = "not us"), then
|
||||
// pick the lowest contact_id. HashMap iteration order is randomized,
|
||||
// so sorting is required to avoid flapping between peers across
|
||||
// receives (which was producing doubled chat threads in the UI).
|
||||
let peers = state.peers.read().await;
|
||||
for p in peers.values() {
|
||||
if !p.advert_name.starts_with("Archy-") {
|
||||
continue;
|
||||
let mut candidates: Vec<(u32, String)> = peers
|
||||
.values()
|
||||
.filter(|p| p.advert_name.starts_with("Archy-"))
|
||||
.filter_map(|p| {
|
||||
let hex_pk = p.pubkey_hex.as_deref()?;
|
||||
if hex_pk.len() < 12 {
|
||||
return None;
|
||||
}
|
||||
if let Some(hex_pk) = p.pubkey_hex.as_deref() {
|
||||
if hex_pk.len() >= 12 {
|
||||
if let Ok(bytes) = hex::decode(&hex_pk[..12]) {
|
||||
let bytes = hex::decode(&hex_pk[..12]).ok()?;
|
||||
if bytes.len() == 6 && bytes[..] != dest_prefix[..] {
|
||||
return (p.contact_id, p.advert_name.clone());
|
||||
Some((p.contact_id, p.advert_name.clone()))
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
(0, "dm-via-channel".to_string())
|
||||
})
|
||||
.collect();
|
||||
candidates.sort_by_key(|(id, _)| *id);
|
||||
candidates
|
||||
.into_iter()
|
||||
.next()
|
||||
.unwrap_or((0, "dm-via-channel".to_string()))
|
||||
}
|
||||
|
||||
@ -95,6 +95,15 @@ pub struct MeshState {
|
||||
pub presence: RwLock<HashMap<String, (String, u32, u64)>>,
|
||||
/// Contacts store — alias/notes/pinned/blocked per peer pubkey hex.
|
||||
pub contacts: RwLock<HashMap<String, ContactEntry>>,
|
||||
/// Our archipelago ed25519 public key (hex). Used by outbound DM-via-channel
|
||||
/// to embed a sender prefix on the wire so receivers can attribute inbound
|
||||
/// messages to the correct contact_id even when multiple `Archy-*` peers
|
||||
/// share the LoRa channel.
|
||||
pub our_ed_pubkey_hex: String,
|
||||
/// Shared blob store for writing received inline attachments. Populated
|
||||
/// by `RpcHandler` after startup so the mesh listener can persist inline
|
||||
/// file bytes into the same store the HTTP layer serves.
|
||||
pub blob_store: RwLock<Option<Arc<crate::blobs::BlobStore>>>,
|
||||
}
|
||||
|
||||
/// Contact metadata kept alongside MeshState.peers. Pinned contacts sort to
|
||||
@ -126,6 +135,7 @@ impl MeshState {
|
||||
stego_mode: super::steganography::SteganographyMode,
|
||||
encrypt_relay: bool,
|
||||
session_manager: Arc<super::session::SessionManager>,
|
||||
our_ed_pubkey_hex: String,
|
||||
) -> (Arc<Self>, broadcast::Receiver<MeshEvent>, mpsc::Receiver<MeshCommand>) {
|
||||
let (tx, rx) = broadcast::channel(64);
|
||||
let (cmd_tx, cmd_rx) = mpsc::channel(32);
|
||||
@ -158,6 +168,8 @@ impl MeshState {
|
||||
encrypt_relay,
|
||||
presence: RwLock::new(HashMap::new()),
|
||||
contacts: RwLock::new(HashMap::new()),
|
||||
our_ed_pubkey_hex,
|
||||
blob_store: RwLock::new(None),
|
||||
});
|
||||
(state, rx, cmd_rx)
|
||||
}
|
||||
|
||||
@ -34,19 +34,123 @@ async fn auto_detect_and_open() -> Result<(String, MeshcoreDevice, DeviceInfo)>
|
||||
anyhow::bail!("No Meshcore device found on {} candidate ports: {:?}", paths.len(), paths)
|
||||
}
|
||||
|
||||
/// Wrap a direct-message payload as a channel-1 broadcast body. Format:
|
||||
/// `[DM_VIA_CHANNEL_MARKER(1)][dest_pubkey_prefix(6)][inner_payload…]`
|
||||
/// Receivers that see this marker on a channel frame extract the header,
|
||||
/// filter by destination, and dispatch the inner payload as if it were a
|
||||
/// direct unicast message.
|
||||
fn wrap_dm_via_channel(dest_pubkey_prefix: &[u8; 6], inner: &[u8]) -> Vec<u8> {
|
||||
let mut out = Vec::with_capacity(1 + 6 + inner.len());
|
||||
out.push(super::super::protocol::DM_VIA_CHANNEL_MARKER);
|
||||
out.extend_from_slice(dest_pubkey_prefix);
|
||||
out.extend_from_slice(inner);
|
||||
/// ASCII marker for the original DM-via-channel format:
|
||||
/// `@DM:` + base64(`[dest_prefix(6)][inner…]`). No sender info on the wire,
|
||||
/// so the receiver has to guess the sender from its contact table — which
|
||||
/// misattributes traffic when more than one `Archy-*` peer is in range.
|
||||
/// Kept for backwards compatibility with peers that haven't been upgraded.
|
||||
pub(super) const DM_V1_MARKER: &str = "@DM:";
|
||||
|
||||
/// ASCII marker for the v2 DM-via-channel format:
|
||||
/// `@DM2:` + base64(`[dest_prefix(6)][sender_arch_prefix(6)][inner…]`).
|
||||
/// `sender_arch_prefix` is the first 6 bytes of the sender's archipelago
|
||||
/// ed25519 public key, which the receiver looks up against its mesh peer
|
||||
/// table (where federation-seeded peers carry the archipelago key) to
|
||||
/// route inbound DMs to the correct contact_id thread.
|
||||
pub(super) const DM_V2_MARKER: &str = "@DM2:";
|
||||
|
||||
fn wrap_dm_for_channel(
|
||||
dest_pubkey_prefix: &[u8; 6],
|
||||
sender_arch_prefix: &[u8; 6],
|
||||
inner: &[u8],
|
||||
) -> String {
|
||||
use base64::Engine;
|
||||
let mut body = Vec::with_capacity(12 + inner.len());
|
||||
body.extend_from_slice(dest_pubkey_prefix);
|
||||
body.extend_from_slice(sender_arch_prefix);
|
||||
body.extend_from_slice(inner);
|
||||
let b64 = base64::engine::general_purpose::STANDARD.encode(&body);
|
||||
format!("{}{}", DM_V2_MARKER, b64)
|
||||
}
|
||||
|
||||
/// Compute our sender_arch_prefix from `state.our_ed_pubkey_hex`. Returns
|
||||
/// `[0u8; 6]` if the stored hex is malformed (which would only happen if a
|
||||
/// caller constructed `MeshState` with a bad value — empty string yields
|
||||
/// all-zero, which won't match any real peer on the receiver side).
|
||||
fn our_sender_prefix(state: &Arc<MeshState>) -> [u8; 6] {
|
||||
let mut out = [0u8; 6];
|
||||
if state.our_ed_pubkey_hex.len() >= 12 {
|
||||
if let Ok(bytes) = hex::decode(&state.our_ed_pubkey_hex[..12]) {
|
||||
if bytes.len() == 6 {
|
||||
out.copy_from_slice(&bytes);
|
||||
}
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// Send an arbitrary-size payload as one or more DM-via-channel frames.
|
||||
/// Single-frame if the wrapped wire fits under the LoRa budget; otherwise
|
||||
/// chunked as `MCxxyyzz…` base64 frames (receiver reassembles via
|
||||
/// `try_chunk_reassemble`). Each chunk is independently `@DM2:`-wrapped
|
||||
/// with both the recipient and sender prefixes so attribution works on
|
||||
/// the receiver side.
|
||||
async fn send_dm_via_channel(
|
||||
device: &mut MeshcoreDevice,
|
||||
state: &Arc<MeshState>,
|
||||
dest_pubkey_prefix: &[u8; 6],
|
||||
payload: &[u8],
|
||||
consecutive_write_failures: &mut u32,
|
||||
) {
|
||||
use base64::Engine;
|
||||
let sender_prefix = our_sender_prefix(state);
|
||||
// First try a single frame with the raw payload directly wrapped.
|
||||
// This keeps small plain-text messages at minimal overhead.
|
||||
let single = wrap_dm_for_channel(dest_pubkey_prefix, &sender_prefix, payload);
|
||||
if single.len() <= 140 {
|
||||
match device.send_channel_text(0, single.as_bytes()).await {
|
||||
Ok(()) => {
|
||||
*consecutive_write_failures = 0;
|
||||
info!(
|
||||
dest = %hex::encode(dest_pubkey_prefix),
|
||||
len = payload.len(),
|
||||
wire_len = single.len(),
|
||||
"Sent mesh message (DM via channel)"
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
*consecutive_write_failures += 1;
|
||||
warn!(failures = *consecutive_write_failures, "Failed to send DM via channel: {}", e);
|
||||
}
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// Payload too large for one wrap — base64 then MC-chunk. Receiver
|
||||
// reassembles base64 chunks and routes the decoded bytes back through
|
||||
// the typed-envelope ladder in handle_channel_payload.
|
||||
let encoded = base64::engine::general_purpose::STANDARD.encode(payload);
|
||||
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 = 80;
|
||||
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 = payload.len(),
|
||||
b64_len = encoded.len(),
|
||||
chunks = total,
|
||||
"Sending chunked mesh message (DM via channel)"
|
||||
);
|
||||
let mut any_err = false;
|
||||
for (idx, chunk) in chunks.iter().enumerate() {
|
||||
let frame = format!("MC{:02x}{:02x}{:02x}{}", msg_id, idx as u8, total, chunk);
|
||||
let wrapped = wrap_dm_for_channel(dest_pubkey_prefix, &sender_prefix, frame.as_bytes());
|
||||
if let Err(e) = device.send_channel_text(0, wrapped.as_bytes()).await {
|
||||
*consecutive_write_failures += 1;
|
||||
warn!(failures = *consecutive_write_failures, chunk = idx, "Chunk DM-via-channel send failed: {}", e);
|
||||
any_err = true;
|
||||
break;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
}
|
||||
if !any_err {
|
||||
*consecutive_write_failures = 0;
|
||||
}
|
||||
}
|
||||
|
||||
/// Fetch the contacts list from the device and update the peer cache.
|
||||
async fn refresh_contacts(
|
||||
device: &mut MeshcoreDevice,
|
||||
@ -301,28 +405,9 @@ async fn handle_send_command(
|
||||
) {
|
||||
match cmd {
|
||||
MeshCommand::SendText { dest_pubkey_prefix, payload } => {
|
||||
// Route the DM as a DM-via-channel broadcast: meshcore's
|
||||
// direct-unicast path silently drops between our two nodes
|
||||
// (proven via `mode=flood resp_code=6` diag — the firmware
|
||||
// transmits but nothing arrives), while channel-1 broadcasts
|
||||
// reliably flood via the FreeMadeira repeater. Wrap the
|
||||
// payload with a recipient pubkey-prefix header so the
|
||||
// receiver side can tell it apart from normal channel text.
|
||||
let wrapped = wrap_dm_via_channel(&dest_pubkey_prefix, &payload);
|
||||
if let Err(e) = device.send_channel_text(0, &wrapped).await {
|
||||
*consecutive_write_failures += 1;
|
||||
warn!(failures = *consecutive_write_failures, "Failed to send DM via channel: {}", e);
|
||||
} else {
|
||||
*consecutive_write_failures = 0;
|
||||
info!(
|
||||
dest = %hex::encode(dest_pubkey_prefix),
|
||||
len = payload.len(),
|
||||
"Sent mesh message (DM via channel)"
|
||||
);
|
||||
}
|
||||
send_dm_via_channel(device, state, &dest_pubkey_prefix, &payload, consecutive_write_failures).await;
|
||||
}
|
||||
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)
|
||||
{
|
||||
@ -336,58 +421,7 @@ async fn handle_send_command(
|
||||
} 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);
|
||||
|
||||
// Route via DM-via-channel wrapper. Channel-1 broadcasts are
|
||||
// the only proven-working path between our two nodes, so we
|
||||
// send the base64 chunk on channel 1 with a recipient header
|
||||
// the receiver can use to filter. Chunk size is reduced by 7
|
||||
// bytes (1 marker + 6 dest-prefix) so each wrapped frame
|
||||
// still fits inside the LoRa 160-byte budget.
|
||||
if encoded.len() <= 133 {
|
||||
// Single frame — wraps under 160B. 160 − 7 wrapper − some
|
||||
// safety margin leaves ≈133 bytes for the base64 payload.
|
||||
let wrapped = wrap_dm_via_channel(&dest_pubkey_prefix, encoded.as_bytes());
|
||||
if let Err(e) = device.send_channel_text(0, &wrapped).await {
|
||||
*consecutive_write_failures += 1;
|
||||
warn!(failures = *consecutive_write_failures, "Failed to send raw DM-via-channel: {}", e);
|
||||
} else {
|
||||
*consecutive_write_failures = 0;
|
||||
info!(dest = %hex::encode(dest_pubkey_prefix), len = encoded.len(), "Sent raw mesh message (DM via channel)");
|
||||
}
|
||||
} else {
|
||||
// Multi-frame chunking: "MCxxyyzz..." where xx=msg_id, yy=chunk_idx, zz=total_chunks.
|
||||
// Chunk size shrunk from 132 → 125 to leave room for the
|
||||
// DM wrapper header (7 bytes) on top of the "MCxxyyzz" (8).
|
||||
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 = 125;
|
||||
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 (DM via channel)"
|
||||
);
|
||||
for (idx, chunk) in chunks.iter().enumerate() {
|
||||
let frame = format!("MC{:02x}{:02x}{:02x}{}", msg_id, idx as u8, total, chunk);
|
||||
let wrapped = wrap_dm_via_channel(&dest_pubkey_prefix, frame.as_bytes());
|
||||
if let Err(e) = device.send_channel_text(0, &wrapped).await {
|
||||
*consecutive_write_failures += 1;
|
||||
warn!(failures = *consecutive_write_failures, chunk = idx, "Chunk DM-via-channel 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;
|
||||
}
|
||||
send_dm_via_channel(device, state, &dest_pubkey_prefix, &wire_payload, consecutive_write_failures).await;
|
||||
}
|
||||
MeshCommand::BroadcastChannel { channel, payload } => {
|
||||
if let Err(e) = device.send_channel_text(channel, &payload).await {
|
||||
|
||||
@ -65,6 +65,11 @@ pub enum MeshMessageType {
|
||||
/// Shareable contact card — advertises a federation node (did, onion, pubkey).
|
||||
/// Lets the receiver one-click-federate with that node.
|
||||
ContactCard = 22,
|
||||
/// Inline attachment: file bytes carried directly in the envelope. Used
|
||||
/// when the file is small enough to chunk over LoRa (<~2.3 KB after
|
||||
/// MCIIXXTT framing) and the peer has no Tor path. Recipient writes the
|
||||
/// bytes to its local BlobStore on reassembly.
|
||||
ContentInline = 23,
|
||||
}
|
||||
|
||||
impl MeshMessageType {
|
||||
@ -93,6 +98,7 @@ impl MeshMessageType {
|
||||
20 => Some(Self::Presence),
|
||||
21 => Some(Self::ChannelInvite),
|
||||
22 => Some(Self::ContactCard),
|
||||
23 => Some(Self::ContentInline),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
@ -125,6 +131,7 @@ impl MeshMessageType {
|
||||
"presence" => Some(Self::Presence),
|
||||
"channel_invite" => Some(Self::ChannelInvite),
|
||||
"contact_card" => Some(Self::ContactCard),
|
||||
"content_inline" => Some(Self::ContentInline),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
@ -154,6 +161,7 @@ impl MeshMessageType {
|
||||
Self::Presence => "presence",
|
||||
Self::ChannelInvite => "channel_invite",
|
||||
Self::ContactCard => "contact_card",
|
||||
Self::ContentInline => "content_inline",
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -450,6 +458,20 @@ pub struct ContentRefPayload {
|
||||
pub cap_exp: u64,
|
||||
}
|
||||
|
||||
/// Inline attachment payload — file bytes carried directly in the envelope.
|
||||
/// Used when the file is small enough to chunk over LoRa and the peer has no
|
||||
/// Tor path. Receiver writes `bytes` to its local BlobStore on reassembly
|
||||
/// and renders it as a normal attachment card.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct ContentInlinePayload {
|
||||
pub mime: String,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub filename: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub caption: Option<String>,
|
||||
pub bytes: Vec<u8>,
|
||||
}
|
||||
|
||||
/// Read receipt — "I have seen every message from this sender up to and
|
||||
/// including `up_to`." Receivers apply this to fold the ✓✓ "seen" marker
|
||||
/// onto all local messages whose MessageKey ≤ `up_to` for that pubkey.
|
||||
|
||||
@ -234,6 +234,7 @@ impl MeshService {
|
||||
config.steganography_mode,
|
||||
config.encrypt_relay_messages,
|
||||
Arc::clone(&session_manager),
|
||||
ed_pubkey_hex.to_string(),
|
||||
);
|
||||
|
||||
// Derive X25519 keys from Ed25519 identity
|
||||
|
||||
@ -407,6 +407,49 @@ export const useMeshStore = defineStore('mesh', () => {
|
||||
}
|
||||
}
|
||||
|
||||
async function transportAdvice(contactId: number, size: number) {
|
||||
return rpcClient.call<{
|
||||
tier: 'auto-mesh' | 'choose' | 'tor-only' | 'impossible'
|
||||
est_seconds: number
|
||||
has_tor: boolean
|
||||
reason: string
|
||||
size: number
|
||||
mesh_auto_max: number
|
||||
mesh_hard_max: number
|
||||
}>({
|
||||
method: 'mesh.transport-advice',
|
||||
params: { contact_id: contactId, size },
|
||||
})
|
||||
}
|
||||
|
||||
async function sendContentInline(
|
||||
contactId: number,
|
||||
mime: string,
|
||||
bytes: Uint8Array,
|
||||
filename?: string,
|
||||
caption?: string,
|
||||
) {
|
||||
try {
|
||||
sending.value = true
|
||||
error.value = null
|
||||
// Base64-encode bytes for JSON transport.
|
||||
let binary = ''
|
||||
for (let i = 0; i < bytes.byteLength; i++) binary += String.fromCharCode(bytes[i]!)
|
||||
const bytes_b64 = btoa(binary)
|
||||
const res = await rpcClient.call<{ sent: boolean; message_id: number; cid: string; size: number }>({
|
||||
method: 'mesh.send-content-inline',
|
||||
params: { contact_id: contactId, mime, filename, caption, bytes_b64 },
|
||||
})
|
||||
if (res.sent) await fetchMessages()
|
||||
return res
|
||||
} catch (err: unknown) {
|
||||
error.value = err instanceof Error ? err.message : 'Failed to send inline content'
|
||||
throw err
|
||||
} finally {
|
||||
sending.value = false
|
||||
}
|
||||
}
|
||||
|
||||
async function sendReply(contactId: number, targetPubkey: string, targetSeq: number, text: string) {
|
||||
sending.value = true
|
||||
try {
|
||||
@ -633,6 +676,8 @@ export const useMeshStore = defineStore('mesh', () => {
|
||||
sendCoordinate,
|
||||
sendAlert,
|
||||
sendContent,
|
||||
sendContentInline,
|
||||
transportAdvice,
|
||||
fetchContent,
|
||||
sendReply,
|
||||
sendReaction,
|
||||
|
||||
@ -1076,18 +1076,54 @@ const attachError = ref<string | null>(null)
|
||||
const fetchingCids = ref<Set<string>>(new Set())
|
||||
const fetchedUrls = ref<Map<string, string>>(new Map())
|
||||
|
||||
async function handleAttachFile(ev: Event) {
|
||||
const input = ev.target as HTMLInputElement
|
||||
const file = input.files?.[0]
|
||||
if (!file) return
|
||||
if (!activeChatPeer.value) {
|
||||
attachError.value = 'Pick a peer first'
|
||||
if (input) input.value = ''
|
||||
return
|
||||
// Transport chooser modal state — populated when advice comes back as
|
||||
// "choose" (size fits both inline-over-mesh AND Tor). User picks a path;
|
||||
// `transportChoiceResolve` finishes the promise started by handleAttachFile.
|
||||
interface PendingTransportChoice {
|
||||
file: File
|
||||
size: number
|
||||
est_seconds: number
|
||||
has_tor: boolean
|
||||
}
|
||||
const transportChoice = ref<PendingTransportChoice | null>(null)
|
||||
let transportChoiceResolve: ((choice: 'mesh' | 'tor' | 'cancel') => void) | null = null
|
||||
|
||||
function pickTransport(choice: 'mesh' | 'tor' | 'cancel') {
|
||||
if (transportChoiceResolve) {
|
||||
transportChoiceResolve(choice)
|
||||
transportChoiceResolve = null
|
||||
}
|
||||
attaching.value = true
|
||||
attachError.value = null
|
||||
transportChoice.value = null
|
||||
}
|
||||
|
||||
async function resolveFederationOnion(peerName: string): Promise<string | undefined> {
|
||||
try {
|
||||
const fed = await rpcClient.federationListNodes()
|
||||
const hit = fed.nodes.find(
|
||||
(n: { name?: string; onion?: string }) =>
|
||||
(n.name ?? '').toLowerCase() === peerName.toLowerCase() ||
|
||||
(n.name ?? '').toLowerCase().includes(peerName.toLowerCase()) ||
|
||||
peerName.toLowerCase().includes((n.name ?? '').toLowerCase()),
|
||||
)
|
||||
return hit?.onion ?? undefined
|
||||
} catch {
|
||||
return undefined
|
||||
}
|
||||
}
|
||||
|
||||
async function sendViaMeshInline(file: File, peerContactId: number) {
|
||||
const buf = await file.arrayBuffer()
|
||||
const bytes = new Uint8Array(buf)
|
||||
await mesh.sendContentInline(
|
||||
peerContactId,
|
||||
file.type || 'application/octet-stream',
|
||||
bytes,
|
||||
file.name,
|
||||
messageText.value.trim() || undefined,
|
||||
)
|
||||
}
|
||||
|
||||
async function sendViaTorContentRef(file: File, peerContactId: number, peerName: string) {
|
||||
const buf = await file.arrayBuffer()
|
||||
const up = await fetch('/api/blob', {
|
||||
method: 'POST',
|
||||
@ -1099,36 +1135,52 @@ async function handleAttachFile(ev: Event) {
|
||||
credentials: 'include',
|
||||
body: buf,
|
||||
})
|
||||
if (!up.ok) {
|
||||
attachError.value = `upload failed: ${up.status}`
|
||||
if (!up.ok) throw new Error(`upload failed: ${up.status}`)
|
||||
const { cid } = (await up.json()) as { cid: string }
|
||||
const peerOnion = await resolveFederationOnion(peerName)
|
||||
await mesh.sendContent(peerContactId, cid, messageText.value.trim() || undefined, peerOnion)
|
||||
}
|
||||
|
||||
async function handleAttachFile(ev: Event) {
|
||||
const input = ev.target as HTMLInputElement
|
||||
const file = input.files?.[0]
|
||||
if (!file) return
|
||||
if (!activeChatPeer.value) {
|
||||
attachError.value = 'Pick a peer first'
|
||||
if (input) input.value = ''
|
||||
return
|
||||
}
|
||||
const { cid } = (await up.json()) as { cid: string }
|
||||
// Resolve the federation onion for this mesh peer. Meshcore adverts
|
||||
// don't carry an archipelago DID so the backend can't link them on its
|
||||
// own — we match on name (both sides use the node's display name).
|
||||
// Falls back to undefined; the backend will try its own DID lookup or
|
||||
// error out if no federation path exists.
|
||||
let peerOnion: string | undefined
|
||||
const peer = activeChatPeer.value
|
||||
attaching.value = true
|
||||
attachError.value = null
|
||||
try {
|
||||
const fed = await rpcClient.federationListNodes()
|
||||
const peerName = activeChatPeer.value.advert_name
|
||||
const hit = fed.nodes.find(
|
||||
(n: { name?: string; onion?: string }) =>
|
||||
(n.name ?? '').toLowerCase() === peerName.toLowerCase() ||
|
||||
(n.name ?? '').toLowerCase().includes(peerName.toLowerCase()) ||
|
||||
peerName.toLowerCase().includes((n.name ?? '').toLowerCase()),
|
||||
)
|
||||
peerOnion = hit?.onion ?? undefined
|
||||
} catch {
|
||||
/* non-fatal — backend will try its own lookup */
|
||||
const advice = await mesh.transportAdvice(peer.contact_id, file.size)
|
||||
let transport: 'mesh' | 'tor' | 'cancel'
|
||||
if (advice.tier === 'auto-mesh') {
|
||||
transport = 'mesh'
|
||||
} else if (advice.tier === 'tor-only') {
|
||||
transport = 'tor'
|
||||
} else if (advice.tier === 'impossible') {
|
||||
attachError.value = `Cannot send: ${advice.reason} (${(file.size / 1024).toFixed(1)} KB)`
|
||||
return
|
||||
} else {
|
||||
// "choose" — open modal and wait for user to pick
|
||||
transport = await new Promise<'mesh' | 'tor' | 'cancel'>((resolve) => {
|
||||
transportChoiceResolve = resolve
|
||||
transportChoice.value = {
|
||||
file,
|
||||
size: file.size,
|
||||
est_seconds: advice.est_seconds,
|
||||
has_tor: advice.has_tor,
|
||||
}
|
||||
})
|
||||
if (transport === 'cancel') return
|
||||
}
|
||||
if (transport === 'mesh') {
|
||||
await sendViaMeshInline(file, peer.contact_id)
|
||||
} else {
|
||||
await sendViaTorContentRef(file, peer.contact_id, peer.advert_name)
|
||||
}
|
||||
await mesh.sendContent(
|
||||
activeChatPeer.value.contact_id,
|
||||
cid,
|
||||
messageText.value.trim() || undefined,
|
||||
peerOnion,
|
||||
)
|
||||
messageText.value = ''
|
||||
nextTick(() => scrollChatToBottom())
|
||||
} catch (e) {
|
||||
@ -1698,6 +1750,37 @@ function isImageMime(mime?: string): boolean {
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<!-- Transport chooser modal: shown when attachment size fits both mesh
|
||||
(inline-chunked) and Tor. User picks which path to send it over. -->
|
||||
<div v-if="transportChoice" class="mesh-transport-modal-backdrop" @click.self="pickTransport('cancel')">
|
||||
<div class="glass-card mesh-transport-modal">
|
||||
<h3 class="mesh-transport-title">📎 How should I send this?</h3>
|
||||
<p class="mesh-transport-sub">
|
||||
<strong>{{ transportChoice.file.name }}</strong>
|
||||
· {{ (transportChoice.size / 1024).toFixed(1) }} KB
|
||||
</p>
|
||||
<div class="mesh-transport-options">
|
||||
<button class="mesh-transport-option" @click="pickTransport('mesh')">
|
||||
<span class="mesh-transport-icon">📡</span>
|
||||
<span class="mesh-transport-label">Over mesh</span>
|
||||
<span class="mesh-transport-meta">~{{ transportChoice.est_seconds }}s · works offline</span>
|
||||
</button>
|
||||
<button
|
||||
class="mesh-transport-option"
|
||||
:disabled="!transportChoice.has_tor"
|
||||
@click="pickTransport('tor')"
|
||||
>
|
||||
<span class="mesh-transport-icon">🧅</span>
|
||||
<span class="mesh-transport-label">Over Tor</span>
|
||||
<span class="mesh-transport-meta">
|
||||
{{ transportChoice.has_tor ? 'instant · needs onion' : 'no Tor path to peer' }}
|
||||
</span>
|
||||
</button>
|
||||
</div>
|
||||
<button class="mesh-transport-cancel" @click="pickTransport('cancel')">Cancel</button>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
</div>
|
||||
</template>
|
||||
|
||||
|
||||
@ -305,3 +305,18 @@ select.mesh-bitcoin-input option { background: #1a1a2e; color: rgba(255,255,255,
|
||||
.mesh-chat-pending-clear { flex: 0 0 auto; align-self: center; margin-left: auto; background: rgba(255,255,255,0.08); border: 1px solid rgba(255,255,255,0.15); color: rgba(255,255,255,0.85); width: 28px; height: 28px; border-radius: 50%; display: inline-flex; align-items: center; justify-content: center; cursor: pointer; font-size: 0.95rem; line-height: 1; transition: background 0.15s ease, color 0.15s ease, border-color 0.15s ease, transform 0.1s ease; }
|
||||
.mesh-chat-pending-clear:hover { background: rgba(239,68,68,0.3); color: #fff; border-color: rgba(239,68,68,0.6); transform: scale(1.08); }
|
||||
.mesh-chat-pending-clear:active { transform: scale(0.92); }
|
||||
|
||||
/* Transport chooser modal (attachment size router) */
|
||||
.mesh-transport-modal-backdrop { position: fixed; inset: 0; background: rgba(0,0,0,0.6); backdrop-filter: blur(4px); display: flex; align-items: center; justify-content: center; z-index: 1000; }
|
||||
.mesh-transport-modal { max-width: 420px; width: 92%; padding: 24px; display: flex; flex-direction: column; gap: 14px; }
|
||||
.mesh-transport-title { margin: 0; font-size: 1.1rem; color: #fff; }
|
||||
.mesh-transport-sub { margin: 0; color: rgba(255,255,255,0.6); font-size: 0.85rem; overflow-wrap: anywhere; }
|
||||
.mesh-transport-options { display: flex; flex-direction: column; gap: 10px; margin-top: 6px; }
|
||||
.mesh-transport-option { display: flex; align-items: center; gap: 12px; padding: 14px 16px; border-radius: 12px; background: rgba(255,255,255,0.05); border: 1px solid rgba(255,255,255,0.12); color: #fff; cursor: pointer; text-align: left; transition: background 0.15s ease, border-color 0.15s ease, transform 0.1s ease; }
|
||||
.mesh-transport-option:hover:not(:disabled) { background: rgba(255,255,255,0.1); border-color: rgba(255,255,255,0.25); transform: translateY(-1px); }
|
||||
.mesh-transport-option:disabled { opacity: 0.4; cursor: not-allowed; }
|
||||
.mesh-transport-icon { font-size: 1.5rem; flex: 0 0 auto; }
|
||||
.mesh-transport-label { flex: 1 1 auto; font-weight: 600; }
|
||||
.mesh-transport-meta { flex: 0 0 auto; font-size: 0.75rem; color: rgba(255,255,255,0.5); }
|
||||
.mesh-transport-cancel { margin-top: 4px; padding: 8px; background: transparent; border: none; color: rgba(255,255,255,0.5); cursor: pointer; font-size: 0.85rem; }
|
||||
.mesh-transport-cancel:hover { color: #fff; }
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user