lfg2025/archy
1
0
Fork 0
Code Issues 26 Pull Requests 4 Actions Packages Projects Releases 74 Wiki Activity

feat(mesh): MessageKey + Reply/Reaction variants and sender seq (Phase 2a)

Browse Source 
Per-target outbound seq counter on MeshState allocates a monotonic seq
before each typed envelope is encoded; send_typed_wire +
send_channel_typed_wire record it (alongside our own pubkey_hex) on the
Sent MeshMessage so the local store carries the same MessageKey the
receiver will see. TypedEnvelope.with_seq lets the RPC layer stamp the
seq AFTER signing (signature covers t/v/ts only).

New MessageKey struct pairs sender_pubkey+sender_seq as the stable
cross-transport identity. Adds variants 13 Reply and 14 Reaction with
ReplyPayload {target, text} and ReactionPayload {target, emoji}, plus
mesh.send-reply / mesh.send-reaction RPCs and receive-side dispatch
arms that store the payload json for the UI to index.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dorian
2026-04-13 13:19:30 -04:00
co-authored by Claude Opus 4.6
parent ab927afbaa
commit a360f90647
6 changed files with 235 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/archipelago/src/api/rpc/dispatcher.rs
+2
View File
@@ -297,6 +297,8 @@ impl RpcHandler {
"mesh.send-alert" => self.handle_mesh_send_alert(params).await,
"mesh.send-content" => self.handle_mesh_send_content(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,
"mesh.outbox" => self.handle_mesh_outbox(params).await,
"mesh.session-status" => self.handle_mesh_session_status(params).await,
"mesh.rotate-prekeys" => self.handle_mesh_rotate_prekeys().await,
core/archipelago/src/api/rpc/mesh/typed_messages.rs
+127 -28
View File
@@ -1,8 +1,8 @@
use super::super::RpcHandler;
use crate::blobs::DEFAULT_CAP_TTL_SECS;
use crate::mesh::message_types::{
self, AlertPayload, AlertType, ContentRefPayload, Coordinate, InvoicePayload, MeshMessageType,
TypedEnvelope,
self, AlertPayload, AlertType, ContentRefPayload, Coordinate, InvoicePayload, MessageKey,
MeshMessageType, ReactionPayload, ReplyPayload, TypedEnvelope,
};
use anyhow::Result;
use tracing::info;
@@ -30,16 +30,17 @@ impl RpcHandler {
payment_hash: None,
};
let payload = message_types::encode_payload(&invoice)?;
let envelope = TypedEnvelope::new(MeshMessageType::Invoice, payload);
let wire = envelope.to_wire()?;
// Send via mesh
let service = self.mesh_service.read().await;
let svc = service
.as_ref()
.ok_or_else(|| anyhow::anyhow!("Mesh service not running"))?;
let seq = svc.next_send_seq(contact_id).await;
let payload = message_types::encode_payload(&invoice)?;
let envelope = TypedEnvelope::new(MeshMessageType::Invoice, payload).with_seq(seq);
let wire = envelope.to_wire()?;
let display = format!(
"Invoice: {} sats{}",
amount_sats,
@@ -47,7 +48,7 @@ impl RpcHandler {
);
let typed_json = serde_json::to_value(&invoice).ok();
let msg = svc
.send_typed_wire(contact_id, wire, "invoice", &display, typed_json)
.send_typed_wire(contact_id, wire, "invoice", &display, typed_json, seq)
.await?;
info!(contact_id, amount_sats, "Sent invoice over mesh");
@@ -77,15 +78,17 @@ impl RpcHandler {
let label = params["label"].as_str().map(|s| s.to_string());
let coord = Coordinate::from_degrees(lat, lng, label);
let payload = message_types::encode_payload(&coord)?;
let envelope = TypedEnvelope::new(MeshMessageType::Coordinate, payload);
let wire = envelope.to_wire()?;
let service = self.mesh_service.read().await;
let svc = service
.as_ref()
.ok_or_else(|| anyhow::anyhow!("Mesh service not running"))?;
let seq = svc.next_send_seq(contact_id).await;
let payload = message_types::encode_payload(&coord)?;
let envelope = TypedEnvelope::new(MeshMessageType::Coordinate, payload).with_seq(seq);
let wire = envelope.to_wire()?;
let display = format!(
"Location: {:.6}, {:.6}{}",
coord.lat_degrees(),
@@ -94,7 +97,7 @@ impl RpcHandler {
);
let typed_json = serde_json::to_value(&coord).ok();
let msg = svc
.send_typed_wire(contact_id, wire, "coordinate", &display, typed_json)
.send_typed_wire(contact_id, wire, "coordinate", &display, typed_json, seq)
.await?;
info!(contact_id, "Sent coordinate over mesh");
@@ -149,7 +152,7 @@ impl RpcHandler {
let identity_dir = self.config.data_dir.join("identity");
let node_key_path = identity_dir.join("node_key");
let envelope = if node_key_path.exists() {
let unsigned_envelope = if node_key_path.exists() {
let key_bytes = tokio::fs::read(&node_key_path).await?;
if key_bytes.len() == 32 {
let mut seed = [0u8; 32];
@@ -163,8 +166,6 @@ impl RpcHandler {
TypedEnvelope::new(MeshMessageType::Alert, payload)
};
let wire = envelope.to_wire()?;
let service = self.mesh_service.read().await;
let svc = service
.as_ref()
@@ -172,21 +173,23 @@ impl RpcHandler {
let display = alert.message.clone();
let typed_json = serde_json::to_value(&alert).ok();
let signed = unsigned_envelope.sig.is_some();
if broadcast {
// Send on public channel (all peers) as raw bytes so the binary
// envelope is not corrupted by utf8 conversion.
svc.send_channel_typed_wire(0, wire, "alert", &display, typed_json.clone())
// Channel 0 uses target=0 for seq allocation. Signature covers
// (t, v, ts) — setting seq afterwards does NOT invalidate it.
let seq = svc.next_send_seq(0).await;
let envelope = unsigned_envelope.with_seq(seq);
let wire = envelope.to_wire()?;
svc.send_channel_typed_wire(0, wire, "alert", &display, typed_json.clone(), seq)
.await?;
info!(alert_type = alert_type_str, "Broadcast alert over mesh");
} else if let Some(contact_id) = params["contact_id"].as_u64() {
svc.send_typed_wire(
contact_id as u32,
wire,
"alert",
&display,
typed_json,
)
.await?;
let contact_id = contact_id as u32;
let seq = svc.next_send_seq(contact_id).await;
let envelope = unsigned_envelope.with_seq(seq);
let wire = envelope.to_wire()?;
svc.send_typed_wire(contact_id, wire, "alert", &display, typed_json, seq)
.await?;
info!(contact_id, alert_type = alert_type_str, "Sent alert to peer");
} else {
anyhow::bail!("Must specify contact_id or broadcast: true");
@@ -195,7 +198,7 @@ impl RpcHandler {
Ok(serde_json::json!({
"sent": true,
"alert_type": alert_type_str,
"signed": envelope.sig.is_some(),
"signed": signed,
}))
}
@@ -271,8 +274,9 @@ impl RpcHandler {
cap_exp: exp,
};
let seq = svc.next_send_seq(contact_id).await;
let payload = message_types::encode_payload(&content)?;
let envelope = TypedEnvelope::new(MeshMessageType::ContentRef, payload);
let envelope = TypedEnvelope::new(MeshMessageType::ContentRef, payload).with_seq(seq);
let wire = envelope.to_wire()?;
let display = match (&content.filename, &content.caption) {
@@ -283,7 +287,7 @@ impl RpcHandler {
};
let typed_json = serde_json::to_value(&content).ok();
let msg = svc
.send_typed_wire(contact_id, wire, "content_ref", &display, typed_json)
.send_typed_wire(contact_id, wire, "content_ref", &display, typed_json, seq)
.await?;
info!(contact_id, cid = %cid, size = meta.size, "Sent content_ref over mesh");
@@ -295,6 +299,101 @@ impl RpcHandler {
}))
}
/// 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
/// to be our peer — cross-transport replies to anyone work.
pub(in crate::api::rpc) async fn handle_mesh_send_reply(
&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 target_pubkey = params["target_pubkey"]
.as_str()
.ok_or_else(|| anyhow::anyhow!("Missing target_pubkey"))?
.to_string();
let target_seq = params["target_seq"]
.as_u64()
.ok_or_else(|| anyhow::anyhow!("Missing target_seq"))?;
let text = params["text"]
.as_str()
.ok_or_else(|| anyhow::anyhow!("Missing text"))?
.to_string();
let reply = ReplyPayload {
target: MessageKey { sender_pubkey: target_pubkey, sender_seq: target_seq },
text: text.clone(),
};
let service = self.mesh_service.read().await;
let svc = service
.as_ref()
.ok_or_else(|| anyhow::anyhow!("Mesh service not running"))?;
let seq = svc.next_send_seq(contact_id).await;
let payload = message_types::encode_payload(&reply)?;
let envelope = TypedEnvelope::new(MeshMessageType::Reply, payload).with_seq(seq);
let wire = envelope.to_wire()?;
let typed_json = serde_json::to_value(&reply).ok();
let msg = svc
.send_typed_wire(contact_id, wire, "reply", &text, typed_json, seq)
.await?;
info!(contact_id, seq, "Sent reply over mesh");
Ok(serde_json::json!({
"sent": true,
"message_id": msg.id,
"sender_seq": seq,
}))
}
/// mesh.send-reaction — Emoji reaction on an earlier message.
/// Params: { contact_id, target_pubkey, target_seq, emoji }. An empty
/// emoji string clears any existing reaction from us on that target.
pub(in crate::api::rpc) async fn handle_mesh_send_reaction(
&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 target_pubkey = params["target_pubkey"]
.as_str()
.ok_or_else(|| anyhow::anyhow!("Missing target_pubkey"))?
.to_string();
let target_seq = params["target_seq"]
.as_u64()
.ok_or_else(|| anyhow::anyhow!("Missing target_seq"))?;
let emoji = params["emoji"].as_str().unwrap_or("").to_string();
let reaction = ReactionPayload {
target: MessageKey { sender_pubkey: target_pubkey, sender_seq: target_seq },
emoji: emoji.clone(),
};
let service = self.mesh_service.read().await;
let svc = service
.as_ref()
.ok_or_else(|| anyhow::anyhow!("Mesh service not running"))?;
let seq = svc.next_send_seq(contact_id).await;
let payload = message_types::encode_payload(&reaction)?;
let envelope = TypedEnvelope::new(MeshMessageType::Reaction, payload).with_seq(seq);
let wire = envelope.to_wire()?;
let display = if emoji.is_empty() { "(cleared)".to_string() } else { emoji.clone() };
let typed_json = serde_json::to_value(&reaction).ok();
let msg = svc
.send_typed_wire(contact_id, wire, "reaction", &display, typed_json, seq)
.await?;
info!(contact_id, seq, emoji = %emoji, "Sent reaction over mesh");
Ok(serde_json::json!({
"sent": true,
"message_id": msg.id,
"sender_seq": seq,
}))
}
/// mesh.fetch-content — Fetch a ContentRef blob from the sender's onion and
/// persist it to our local blob store. Params must include everything the
/// receiver needs to construct and authorise the URL: `{ cid, sender_onion,