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
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
6641c1d183
commit
e210376e05
@@ -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).
|
||||
|
||||
Reference in New Issue
Block a user