core: delivery receipts — "seen by <device>" instead of "did you see it?"
On every inbound clipboard the engine auto-sends a receipt back to the sender
(type:"receipt", to:sender, role:my-name, text:the-clipboard). Senders collect
them via Engine.receipts() -> [Receipt{from,name,source,text,ts}] so the UI can
show "delivered to iPhone" per sent item — visible read state, not manual asking.
Plumbing this required: SSE post now sends the full envelope (type/to/role), not
just text; the SSE stream forwards receipts as well as clipboards; RTC's raw-text
data channel skips non-clipboard messages (receipts ride SSE/LAN). Receipts
de-dupe across transports.
Proven by examples/receipt_demo.rs: A sends, B auto-acks, A.receipts() shows
"seen by Patrick's iPhone".
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,39 @@
|
|||||||
|
//! Proves delivery receipts: A sends a clipboard, B receives it and auto-acks,
|
||||||
|
//! and A.receipts() shows "seen by <B's name>" — no asking.
|
||||||
|
//! cargo run --example receipt_demo
|
||||||
|
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
|
use tethercore::{Engine, Message, MessageHandler};
|
||||||
|
|
||||||
|
struct Noop;
|
||||||
|
impl MessageHandler for Noop {
|
||||||
|
fn on_message(&self, _m: Message) {}
|
||||||
|
fn on_status(&self, _c: bool) {}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn main() {
|
||||||
|
let dead = "http://127.0.0.1:1".to_string(); // LAN-only
|
||||||
|
let room = "receipt-room".to_string();
|
||||||
|
let a = Engine::new(dead.clone(), room.clone(), "dev-a".into(), "macos".into(), "Patrick's Mac".into(), "".into());
|
||||||
|
let b = Engine::new(dead, room, "dev-b".into(), "ios".into(), "Patrick's iPhone".into(), "".into());
|
||||||
|
a.start(Box::new(Noop));
|
||||||
|
b.start(Box::new(Noop));
|
||||||
|
|
||||||
|
eprintln!("waiting for LAN link…");
|
||||||
|
std::thread::sleep(Duration::from_secs(5));
|
||||||
|
|
||||||
|
eprintln!("A sends a clipboard…");
|
||||||
|
a.send("important-link".to_string());
|
||||||
|
std::thread::sleep(Duration::from_secs(2)); // let B receive + ack, A record
|
||||||
|
|
||||||
|
let r = a.receipts();
|
||||||
|
if let Some(rec) = r.iter().find(|r| r.text == "important-link") {
|
||||||
|
println!("✅ A sees its clipboard was delivered → seen by {:?} [{}]", rec.name, rec.source);
|
||||||
|
} else {
|
||||||
|
println!("❌ no receipt recorded ({} total)", r.len());
|
||||||
|
std::process::exit(1);
|
||||||
|
}
|
||||||
|
a.stop();
|
||||||
|
b.stop();
|
||||||
|
}
|
||||||
+84
-12
@@ -74,6 +74,20 @@ pub struct Peer {
|
|||||||
/// engine's `nearby()` (reader). Instant is last-seen for staleness pruning.
|
/// engine's `nearby()` (reader). Instant is last-seen for staleness pruning.
|
||||||
type Discovered = Arc<Mutex<HashMap<String, (Peer, Instant)>>>;
|
type Discovered = Arc<Mutex<HashMap<String, (Peer, Instant)>>>;
|
||||||
|
|
||||||
|
/// A delivery acknowledgement: device `from` (named `name`) received a clipboard
|
||||||
|
/// whose text matches `text`. Lets the sender show "delivered to <name>" instead
|
||||||
|
/// of asking. Auto-sent by the engine on every inbound clipboard.
|
||||||
|
#[derive(Clone, Debug, uniffi::Record)]
|
||||||
|
pub struct Receipt {
|
||||||
|
pub from: String, // device that received it
|
||||||
|
pub name: String, // its friendly name
|
||||||
|
pub source: String, // its platform
|
||||||
|
pub text: String, // the clipboard text it received (for matching)
|
||||||
|
pub ts: i64,
|
||||||
|
}
|
||||||
|
|
||||||
|
type Receipts = Arc<Mutex<Vec<(Receipt, Instant)>>>;
|
||||||
|
|
||||||
/// Canonical room id for a shared identity: sha256(account)[..4] as 8 hex chars.
|
/// Canonical room id for a shared identity: sha256(account)[..4] as 8 hex chars.
|
||||||
/// Defined once here and exported so every client (Swift/Kotlin/agent) derives
|
/// Defined once here and exported so every client (Swift/Kotlin/agent) derives
|
||||||
/// the SAME room — that's what puts all of one account's devices together across
|
/// the SAME room — that's what puts all of one account's devices together across
|
||||||
@@ -121,6 +135,7 @@ pub struct Engine {
|
|||||||
transports: Vec<Arc<dyn Transport>>,
|
transports: Vec<Arc<dyn Transport>>,
|
||||||
dedup: Arc<Mutex<Dedup>>,
|
dedup: Arc<Mutex<Dedup>>,
|
||||||
discovered: Discovered,
|
discovered: Discovered,
|
||||||
|
receipts: Receipts,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[uniffi::export]
|
#[uniffi::export]
|
||||||
@@ -162,9 +177,19 @@ impl Engine {
|
|||||||
transports,
|
transports,
|
||||||
dedup: Arc::new(Mutex::new(Dedup::default())),
|
dedup: Arc::new(Mutex::new(Dedup::default())),
|
||||||
discovered,
|
discovered,
|
||||||
|
receipts: Arc::new(Mutex::new(Vec::new())),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Delivery acks received for clipboards we sent — "seen by <name>". Pruned
|
||||||
|
/// after 2 minutes. Poll alongside your sent items to show read state.
|
||||||
|
pub fn receipts(&self) -> Vec<Receipt> {
|
||||||
|
let now = Instant::now();
|
||||||
|
let mut v = self.receipts.lock().unwrap();
|
||||||
|
v.retain(|(_, t)| now.duration_since(*t) < Duration::from_secs(120));
|
||||||
|
v.iter().map(|(r, _)| r.clone()).collect()
|
||||||
|
}
|
||||||
|
|
||||||
/// Nearby tether devices seen on the local network (mDNS), for an
|
/// Nearby tether devices seen on the local network (mDNS), for an
|
||||||
/// AirDrop-style picker. Stale entries (>30s unseen) are pruned. Empty on
|
/// AirDrop-style picker. Stale entries (>30s unseen) are pruned. Empty on
|
||||||
/// iOS until the multicast entitlement lands.
|
/// iOS until the multicast entitlement lands.
|
||||||
@@ -183,9 +208,49 @@ impl Engine {
|
|||||||
let handler: Arc<dyn MessageHandler> = Arc::from(handler);
|
let handler: Arc<dyn MessageHandler> = Arc::from(handler);
|
||||||
let dedup = self.dedup.clone();
|
let dedup = self.dedup.clone();
|
||||||
let h_msg = handler.clone();
|
let h_msg = handler.clone();
|
||||||
let sink: Sink = Arc::new(move |m| {
|
let receipts = self.receipts.clone();
|
||||||
|
let cfg = self.cfg.clone();
|
||||||
|
let transports = self.transports.clone();
|
||||||
|
let rt = self.rt.clone();
|
||||||
|
let sink: Sink = Arc::new(move |m: Message| {
|
||||||
|
// Delivery acks addressed to us → record "seen by <name>".
|
||||||
|
if m.kind == "receipt" {
|
||||||
|
if m.to == cfg.from {
|
||||||
|
let mut v = receipts.lock().unwrap();
|
||||||
|
if !v.iter().any(|(r, _)| r.from == m.from && r.text == m.text) {
|
||||||
|
v.push((
|
||||||
|
Receipt {
|
||||||
|
from: m.from,
|
||||||
|
name: m.role, // sender packs its name in `role`
|
||||||
|
source: m.source,
|
||||||
|
text: m.text,
|
||||||
|
ts: m.ts,
|
||||||
|
},
|
||||||
|
Instant::now(),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
// Clipboard: de-dup, ack the sender, then deliver.
|
||||||
if dedup.lock().unwrap().seen(&m) {
|
if dedup.lock().unwrap().seen(&m) {
|
||||||
return; // same clipboard already delivered via another transport
|
return; // already delivered via another transport
|
||||||
|
}
|
||||||
|
if !m.from.is_empty() && m.from != cfg.from {
|
||||||
|
let ack = Message {
|
||||||
|
kind: "receipt".into(),
|
||||||
|
text: m.text.clone(),
|
||||||
|
from: cfg.from.clone(),
|
||||||
|
to: m.from.clone(),
|
||||||
|
role: cfg.name.clone(), // our friendly name for the receipt
|
||||||
|
source: cfg.source.clone(),
|
||||||
|
room: cfg.room.clone(),
|
||||||
|
ts: 0,
|
||||||
|
};
|
||||||
|
let h = rt.handle();
|
||||||
|
for t in &transports {
|
||||||
|
t.publish(h, ack.clone());
|
||||||
|
}
|
||||||
}
|
}
|
||||||
h_msg.on_message(m);
|
h_msg.on_message(m);
|
||||||
});
|
});
|
||||||
@@ -304,7 +369,7 @@ impl Transport for SseTransport {
|
|||||||
rt.spawn(async move {
|
rt.spawn(async move {
|
||||||
// A failed send is already implied by the connection-state log;
|
// A failed send is already implied by the connection-state log;
|
||||||
// don't log per-send (every clipboard change would spam offline).
|
// don't log per-send (every clipboard change would spam offline).
|
||||||
let _ = post(&http, &cfg, &msg.text).await;
|
let _ = post(&http, &cfg, &msg).await;
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -351,10 +416,13 @@ impl SseTransport {
|
|||||||
if line.is_empty() {
|
if line.is_empty() {
|
||||||
if !data.is_empty() {
|
if !data.is_empty() {
|
||||||
if let Ok(m) = serde_json::from_slice::<Message>(&data) {
|
if let Ok(m) = serde_json::from_slice::<Message>(&data) {
|
||||||
// Clipboard only — presence/signal chirps on the bus
|
// Clipboard + receipts go to the engine; presence/signal
|
||||||
// belong to the RTC layer, not the feed.
|
// chirps belong to the RTC layer, not here.
|
||||||
if m.from != self.cfg.from && (m.kind.is_empty() || m.kind == "clipboard") {
|
let kind = &m.kind;
|
||||||
sink(m); // engine de-dups + forwards to handler
|
if m.from != self.cfg.from
|
||||||
|
&& (kind.is_empty() || kind == "clipboard" || kind == "receipt")
|
||||||
|
{
|
||||||
|
sink(m); // engine routes + de-dups
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
data.clear();
|
data.clear();
|
||||||
@@ -370,13 +438,17 @@ impl SseTransport {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn post(http: &reqwest::Client, cfg: &Config, text: &str) -> Result<(), String> {
|
async fn post(http: &reqwest::Client, cfg: &Config, msg: &Message) -> Result<(), String> {
|
||||||
let url = format!("{}/api/send", cfg.server.trim_end_matches('/'));
|
let url = format!("{}/api/send", cfg.server.trim_end_matches('/'));
|
||||||
|
let kind = if msg.kind.is_empty() { "clipboard" } else { msg.kind.as_str() };
|
||||||
|
// Full envelope so receipts (type/to/role) survive the relay, not just text.
|
||||||
let body = serde_json::json!({
|
let body = serde_json::json!({
|
||||||
"type": "clipboard",
|
"type": kind,
|
||||||
"text": text,
|
"text": msg.text,
|
||||||
"from": cfg.from,
|
"from": msg.from,
|
||||||
"source": cfg.source,
|
"to": msg.to,
|
||||||
|
"role": msg.role,
|
||||||
|
"source": msg.source,
|
||||||
"room": cfg.room,
|
"room": cfg.room,
|
||||||
});
|
});
|
||||||
http.post(&url)
|
http.post(&url)
|
||||||
|
|||||||
@@ -490,6 +490,11 @@ impl Transport for RtcTransport {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn publish(&self, rt: &Handle, msg: Message) {
|
fn publish(&self, rt: &Handle, msg: Message) {
|
||||||
|
// The "tether" data channel carries raw clipboard text only; receipts
|
||||||
|
// and other typed messages go over SSE/LAN.
|
||||||
|
if !(msg.kind.is_empty() || msg.kind == "clipboard") {
|
||||||
|
return;
|
||||||
|
}
|
||||||
let peers = self.peers.clone();
|
let peers = self.peers.clone();
|
||||||
rt.spawn(async move {
|
rt.spawn(async move {
|
||||||
let map = peers.lock().await;
|
let map = peers.lock().await;
|
||||||
|
|||||||
Reference in New Issue
Block a user