use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use nostr_sdk::prelude::*;
use crate::state::{nostr_client, my_public_key, STATE};
use crate::types::{Message, Attachment};
use crate::crypto;
pub trait SendCallback: Send + Sync {
fn on_pending(&self, _chat_id: &str, _msg: &Message) {}
fn on_upload_progress(
&self,
_pending_id: &str,
_percentage: u8,
_bytes_sent: u64,
) -> Result<(), String> {
Ok(())
}
fn on_upload_complete(&self, _chat_id: &str, _pending_id: &str, _attachment_id: &str, _url: &str) {}
fn on_sent(&self, _chat_id: &str, _old_id: &str, _msg: &Message) {}
fn on_failed(&self, _chat_id: &str, _old_id: &str, _msg: &Message) {}
fn on_persist(&self, _chat_id: &str, _msg: &Message) {}
}
pub struct NoOpSendCallback;
impl SendCallback for NoOpSendCallback {}
pub struct SendConfig {
pub max_send_attempts: u32,
pub retry_delay: std::time::Duration,
pub self_send: bool,
pub cancel_token: Option<Arc<AtomicBool>>,
pub upload_retries: u32,
pub upload_retry_delay: std::time::Duration,
pub expiration: Option<u64>,
}
impl Default for SendConfig {
fn default() -> Self {
Self {
max_send_attempts: 1,
retry_delay: std::time::Duration::from_secs(5),
self_send: false,
cancel_token: None,
upload_retries: 3,
upload_retry_delay: std::time::Duration::from_secs(2),
expiration: None,
}
}
}
impl SendConfig {
pub fn gui() -> Self {
Self {
max_send_attempts: 12,
self_send: true,
..Default::default()
}
}
pub fn headless() -> Self {
Self {
max_send_attempts: 3,
self_send: true,
..Default::default()
}
}
}
#[derive(serde::Serialize, Clone, Debug)]
pub struct SendResult {
pub pending_id: String,
pub event_id: Option<String>,
pub chat_id: String,
}
struct WrapConfirm {
wrap_id: EventId,
chat_id: String,
pending_id: String,
rumor_event_id: String,
rumor: UnsignedEvent,
callback: Arc<dyn SendCallback>,
self_send: bool,
confirmed: AtomicBool,
rescued: AtomicBool,
loop_exited: AtomicBool,
notify: tokio::sync::Notify,
session: crate::state::SessionGuard,
registered_at: std::time::Instant,
}
static WRAP_CONFIRMS: std::sync::LazyLock<
std::sync::Mutex<std::collections::HashMap<EventId, Arc<WrapConfirm>>>,
> = std::sync::LazyLock::new(|| std::sync::Mutex::new(std::collections::HashMap::new()));
const WRAP_CONFIRM_TTL: std::time::Duration = std::time::Duration::from_secs(15 * 60);
fn register_wrap_confirm(entry: Arc<WrapConfirm>) {
let mut map = WRAP_CONFIRMS.lock().unwrap();
map.retain(|_, e| e.registered_at.elapsed() < WRAP_CONFIRM_TTL);
map.insert(entry.wrap_id, entry);
}
fn remove_wrap_confirm(wrap_id: &EventId) {
WRAP_CONFIRMS.lock().unwrap().remove(wrap_id);
}
pub fn clear_wrap_confirms() {
WRAP_CONFIRMS.lock().unwrap().clear();
}
pub fn note_relay_ok(event_id: &EventId, accepted: bool) {
if !accepted {
return;
}
let entry = {
let map = WRAP_CONFIRMS.lock().unwrap();
map.get(event_id).cloned()
};
let Some(entry) = entry else { return };
entry.confirmed.store(true, Ordering::SeqCst);
entry.notify.notify_one();
if !entry.loop_exited.load(Ordering::SeqCst)
|| entry.rescued.swap(true, Ordering::SeqCst)
{
return;
}
if !entry.session.is_valid() {
remove_wrap_confirm(&entry.wrap_id);
return;
}
tokio::spawn(async move {
rescue_failed_as_sent(&entry).await;
remove_wrap_confirm(&entry.wrap_id);
});
}
async fn rescue_failed_as_sent(entry: &WrapConfirm) {
if !entry.session.is_valid() {
return;
}
let finalized = {
let mut state = STATE.lock().await;
state.update_message(&entry.pending_id, |msg| {
msg.set_failed(false);
});
state.finalize_pending_message(&entry.chat_id, &entry.pending_id, &entry.rumor_event_id)
};
let Some((_old_id, ref msg)) = finalized else { return };
crate::log_info!(
"[Send] late relay OK confirmed wrap {} — message {} rescued to sent",
entry.wrap_id,
entry.rumor_event_id,
);
entry.callback.on_sent(&entry.chat_id, &entry.pending_id, msg);
entry.callback.on_persist(&entry.chat_id, msg);
if entry.self_send {
if let (Some(client), Some(my_pk)) = (nostr_client(), my_public_key()) {
spawn_self_send(client, my_pk, entry.rumor.clone());
}
}
}
fn spawn_self_send(client: Client, my_pk: PublicKey, rumor: UnsignedEvent) {
let rid_for_self = rumor.id;
let session = crate::state::SessionGuard::capture();
tokio::spawn(async move {
if !session.is_valid() { return; }
match crate::inbox_relays::send_gift_wrap_retained(
&client, &my_pk, rumor, [],
).await {
Ok(self_outcome) if !self_outcome.output.success.is_empty() => {
if !session.is_valid() { return; }
if let Some(rid) = rid_for_self {
if let Err(e) = crate::db::nip17_keys::store_wrap_key(
&self_outcome.wrap_event_id,
&rid,
&my_pk,
crate::db::nip17_keys::WrapRole::SelfSend,
&self_outcome.wrap_secret,
&self_outcome.targeted_relays,
) {
eprintln!("[NIP-17] failed to persist self-wrap key: {}", e);
}
}
}
_ => {}
}
});
}
async fn retry_send_gift_wrap(
client: &Client,
receiver: &PublicKey,
receiver_npub: &str,
pending_id: &str,
rumor: UnsignedEvent,
event_id: &str,
config: &SendConfig,
callback: Arc<dyn SendCallback>,
) -> Result<SendResult, String> {
let my_pk = my_public_key().ok_or("Public key not set")?;
let inner_rumor_id = rumor.id;
let mut built: Option<crate::inbox_relays::BuiltGiftWrap> = None;
let mut targets: Option<crate::inbox_relays::GiftWrapTargets> = None;
let mut confirm: Option<Arc<WrapConfirm>> = None;
let mut last_error: Option<String> = None;
let max_attempts = config.max_send_attempts.max(1);
for attempt in 0..max_attempts {
if built.is_none() {
let wrap_extra: Vec<Tag> = rumor.tags.iter()
.find(|t| t.as_slice().first().map(|k| k.as_str() == "expiration").unwrap_or(false))
.cloned()
.into_iter()
.collect();
match crate::inbox_relays::build_gift_wrap_retained(
client, receiver, rumor.clone(), wrap_extra,
).await {
Ok(b) => built = Some(b),
Err(e) => {
crate::log_warn!(
"[Send] attempt {}/{} — building gift-wrap failed: {}",
attempt + 1, max_attempts, e,
);
last_error = Some(e);
if attempt + 1 < max_attempts {
tokio::time::sleep(config.retry_delay).await;
}
continue;
}
}
}
let wrap = built.as_ref().unwrap();
if confirm.is_none() {
let entry = Arc::new(WrapConfirm {
wrap_id: wrap.event.id,
chat_id: receiver_npub.to_string(),
pending_id: pending_id.to_string(),
rumor_event_id: event_id.to_string(),
rumor: rumor.clone(),
callback: callback.clone(),
self_send: config.self_send,
confirmed: AtomicBool::new(false),
rescued: AtomicBool::new(false),
loop_exited: AtomicBool::new(false),
notify: tokio::sync::Notify::new(),
session: crate::state::SessionGuard::capture(),
registered_at: std::time::Instant::now(),
});
register_wrap_confirm(entry.clone());
confirm = Some(entry);
}
let confirm_ref = confirm.as_ref().unwrap();
if targets.is_none() {
let t = crate::inbox_relays::resolve_gift_wrap_targets(client, receiver).await;
if let Some(rid) = inner_rumor_id {
if let Err(e) = crate::db::nip17_keys::store_wrap_key(
&wrap.event.id,
&rid,
receiver,
crate::db::nip17_keys::WrapRole::Recipient,
&wrap.secret,
&t.targeted_relays,
) {
eprintln!("[NIP-17] failed to persist wrap key: {}", e);
}
}
targets = Some(t);
} else {
crate::inbox_relays::reconnect_gift_wrap_targets(targets.as_ref().unwrap()).await;
}
let targets_ref = targets.as_ref().unwrap();
match crate::inbox_relays::publish_gift_wrap_to_targets(
client, targets_ref, &wrap.event,
).await {
Ok(output) if !output.success.is_empty() => {
return Ok(finalize_gift_wrap_sent(
client, my_pk, receiver_npub, pending_id, event_id,
&rumor, config, &callback, confirm_ref, targets_ref,
).await);
}
Ok(output) => {
let failures: Vec<String> = output.failed.iter()
.map(|(url, err)| format!("{}: {}", url, err))
.collect();
crate::log_warn!(
"[Send] attempt {}/{} — 0 of {} relays accepted (targeted: {}). Per-relay errors: {}",
attempt + 1,
max_attempts,
targets_ref.targeted_relays.len(),
targets_ref.targeted_relays.join(", "),
if failures.is_empty() {
"(none reported — likely all timed out before responding)".to_string()
} else {
failures.join(" | ")
},
);
last_error = None;
}
Err(e) => {
crate::log_warn!(
"[Send] attempt {}/{} — publish errored: {}",
attempt + 1, max_attempts, e,
);
last_error = Some(e);
}
}
if confirm_ref.confirmed.load(Ordering::SeqCst) {
return Ok(finalize_gift_wrap_sent(
client, my_pk, receiver_npub, pending_id, event_id,
&rumor, config, &callback, confirm_ref, targets_ref,
).await);
}
if attempt + 1 < max_attempts {
let _ = tokio::time::timeout(
config.retry_delay,
confirm_ref.notify.notified(),
).await;
if confirm_ref.confirmed.load(Ordering::SeqCst) {
return Ok(finalize_gift_wrap_sent(
client, my_pk, receiver_npub, pending_id, event_id,
&rumor, config, &callback, confirm_ref, targets_ref,
).await);
}
}
}
if let Some(t) = targets.as_ref() {
crate::inbox_relays::teardown_gift_wrap_targets(client, t).await;
}
let failed_msg = {
let mut state = STATE.lock().await;
state.update_message(pending_id, |msg| {
msg.set_failed(true);
msg.set_pending(false);
})
};
if let Some((_chat_id, ref msg)) = failed_msg {
callback.on_failed(receiver_npub, pending_id, msg);
callback.on_persist(receiver_npub, msg);
}
if let Some(entry) = confirm.as_ref() {
entry.loop_exited.store(true, Ordering::SeqCst);
if entry.confirmed.load(Ordering::SeqCst)
&& !entry.rescued.swap(true, Ordering::SeqCst)
{
rescue_failed_as_sent(entry).await;
remove_wrap_confirm(&entry.wrap_id);
return Ok(SendResult {
pending_id: pending_id.to_string(),
event_id: Some(event_id.to_string()),
chat_id: receiver_npub.to_string(),
});
}
}
match last_error {
Some(e) => Err(format!("Failed to send DM after {} attempts: {}", max_attempts, e)),
None => Err(format!(
"Failed to send DM after {} attempts (no relays accepted the gift-wrap)",
max_attempts
)),
}
}
async fn finalize_gift_wrap_sent(
client: &Client,
my_pk: PublicKey,
receiver_npub: &str,
pending_id: &str,
event_id: &str,
rumor: &UnsignedEvent,
config: &SendConfig,
callback: &Arc<dyn SendCallback>,
confirm: &Arc<WrapConfirm>,
targets: &crate::inbox_relays::GiftWrapTargets,
) -> SendResult {
remove_wrap_confirm(&confirm.wrap_id);
crate::inbox_relays::teardown_gift_wrap_targets(client, targets).await;
let finalized = {
let mut state = STATE.lock().await;
state.finalize_pending_message(receiver_npub, pending_id, event_id)
};
if let Some((_old_id, ref finalized_msg)) = finalized {
callback.on_sent(receiver_npub, pending_id, finalized_msg);
callback.on_persist(receiver_npub, finalized_msg);
}
if config.self_send {
spawn_self_send(client.clone(), my_pk, rumor.clone());
}
SendResult {
pending_id: pending_id.to_string(),
event_id: Some(event_id.to_string()),
chat_id: receiver_npub.to_string(),
}
}
pub async fn send_dm(
receiver_npub: &str,
content: &str,
reply_to: Option<&str>,
config: &SendConfig,
callback: Arc<dyn SendCallback>,
) -> Result<SendResult, String> {
let client = nostr_client().ok_or("Not logged in")?;
let my_pk = my_public_key().ok_or("Public key not set")?;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH).unwrap();
let pending_id = format!("pending-{}", now.as_nanos());
let receiver = PublicKey::from_bech32(receiver_npub)
.map_err(|e| format!("Invalid npub: {}", e))?;
let emoji_tags = crate::emoji_packs::resolve_outbound_emoji_tags(content);
let msg = Message {
id: pending_id.clone(),
content: content.to_string(),
replied_to: reply_to.unwrap_or("").to_string(),
at: now.as_millis() as u64,
pending: true,
mine: true,
npub: my_pk.to_bech32().ok(),
emoji_tags: emoji_tags.clone(),
expiration: config.expiration,
..Default::default()
};
{
let mut state = STATE.lock().await;
state.add_message_to_participant(receiver_npub, msg.clone());
}
callback.on_pending(receiver_npub, &msg);
let milliseconds = now.as_millis() % 1000;
let mut rumor = EventBuilder::private_msg_rumor(receiver, content);
if let Some(reply_id) = reply_to {
if !reply_id.is_empty() {
rumor = rumor.tag(Tag::custom(
TagKind::e(),
[reply_id.to_string(), String::new(), "reply".to_string()],
));
}
}
let mut rumor = rumor.tag(Tag::custom(TagKind::custom("ms"), [milliseconds.to_string()]));
for et in &emoji_tags {
rumor = rumor.tag(Tag::custom(
TagKind::custom("emoji"),
[et.shortcode.clone(), et.url.clone()],
));
}
if let Some(exp) = config.expiration {
rumor = rumor.tag(Tag::expiration(Timestamp::from_secs(exp)));
}
let built_rumor = rumor.build(my_pk);
let event_id = built_rumor.id.ok_or("Rumor has no id")?.to_hex();
retry_send_gift_wrap(
&client, &receiver, receiver_npub, &pending_id,
built_rumor, &event_id, config, callback,
).await
}
pub async fn send_rumor_dm(
receiver_npub: &str,
pending_id: &str,
rumor: UnsignedEvent,
config: &SendConfig,
callback: Arc<dyn SendCallback>,
) -> Result<SendResult, String> {
let client = nostr_client().ok_or("Not logged in")?;
let receiver = PublicKey::from_bech32(receiver_npub)
.map_err(|e| format!("Invalid npub: {}", e))?;
let event_id = rumor.id.ok_or("Rumor has no id")?.to_hex();
retry_send_gift_wrap(
&client, &receiver, receiver_npub, pending_id,
rumor, &event_id, config, callback,
).await
}
pub async fn send_file_dm(
receiver_npub: &str,
file_bytes: Arc<Vec<u8>>,
filename: &str,
extension: &str,
content: Option<&str>,
config: &SendConfig,
callback: Arc<dyn SendCallback>,
) -> Result<SendResult, String> {
let client = nostr_client().ok_or("Not logged in")?;
let my_pk = my_public_key().ok_or("Public key not set")?;
let signer = client.signer().await
.map_err(|e| format!("Signer unavailable: {}", e))?;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH).unwrap();
let pending_id = format!("pending-{}", now.as_nanos());
let milliseconds = now.as_millis() % 1000;
let receiver = PublicKey::from_bech32(receiver_npub)
.map_err(|e| format!("Invalid npub: {}", e))?;
let file_hash = crypto::sha256_hex(&file_bytes);
let mime_type = crypto::mime_from_extension(extension);
let webxdc_topic = (extension.eq_ignore_ascii_case("xdc"))
.then(|| crate::webxdc::mint_topic_id(&file_hash, &my_pk.to_hex()));
let download_dir = crate::db::get_download_dir();
let _ = std::fs::create_dir_all(&download_dir);
let local_name = if filename.is_empty() {
format!("{}.{}", &file_hash, extension)
} else {
let stem = filename.rsplit_once('.').map(|(s, _)| s).unwrap_or(filename);
format!("{}.{}", stem, extension)
};
let local_path = crypto::resolve_unique_filename(&download_dir, &local_name);
let tmp = download_dir.join(format!(".{}.tmp", &file_hash));
let _ = std::fs::write(&tmp, &*file_bytes);
let _ = std::fs::rename(&tmp, &local_path);
let local_path_str = local_path.to_string_lossy().to_string();
let img_meta = crypto::generate_image_metadata(&file_bytes);
let params = crypto::generate_encryption_params();
let encrypted = crypto::encrypt_data(&file_bytes, ¶ms)?;
let encrypted_size = encrypted.len() as u64;
let attachment = Attachment {
id: file_hash.clone(), key: params.key.clone(), nonce: params.nonce.clone(),
extension: extension.to_string(), name: filename.to_string(),
url: String::new(), path: local_path_str.clone(), size: encrypted_size,
img_meta: img_meta.clone(), downloading: false, downloaded: true,
webxdc_topic: webxdc_topic.clone(),
..Default::default()
};
let msg = Message {
id: pending_id.clone(), content: content.unwrap_or("").to_string(),
at: now.as_millis() as u64, pending: true, mine: true,
npub: my_pk.to_bech32().ok(), attachments: vec![attachment],
expiration: config.expiration,
..Default::default()
};
{
let mut state = STATE.lock().await;
state.add_message_to_participant(receiver_npub, msg.clone());
}
callback.on_pending(receiver_npub, &msg);
let servers = crate::state::get_blossom_servers();
let cb_for_progress = callback.clone();
let pid_for_progress = pending_id.clone();
let progress_cb: crate::blossom::ProgressCallback = Arc::new(move |percentage, bytes| {
cb_for_progress.on_upload_progress(
&pid_for_progress,
percentage.unwrap_or(0),
bytes.unwrap_or(0),
)
});
let upload_url = match crate::blossom::upload_blob_with_progress_and_failover(
signer.clone(), servers, Arc::new(encrypted), Some(mime_type),
true,
progress_cb, Some(config.upload_retries), Some(config.upload_retry_delay),
config.cancel_token.clone(),
).await {
Ok(url) => url,
Err(e) => {
let failed_msg = {
let mut state = STATE.lock().await;
state.update_message(&pending_id, |msg| {
msg.set_failed(true);
msg.set_pending(false);
})
};
if let Some((_chat_id, ref msg)) = failed_msg {
callback.on_failed(receiver_npub, &pending_id, msg);
callback.on_persist(receiver_npub, msg);
}
return Err(format!("Upload failed: {}", e));
}
};
{
let mut state = STATE.lock().await;
state.update_message(&pending_id, |msg| {
if let Some(att) = msg.attachments.last_mut() {
att.url = upload_url.clone().into_boxed_str();
}
});
}
callback.on_upload_complete(receiver_npub, &pending_id, &file_hash, &upload_url);
let mut file_rumor = EventBuilder::new(Kind::from_u16(15), &upload_url)
.tag(Tag::public_key(receiver))
.tag(Tag::custom(TagKind::custom("file-type"), [mime_type]))
.tag(Tag::custom(TagKind::custom("size"), [encrypted_size.to_string()]))
.tag(Tag::custom(TagKind::custom("encryption-algorithm"), ["aes-gcm"]))
.tag(Tag::custom(TagKind::custom("decryption-key"), [params.key.as_str()]))
.tag(Tag::custom(TagKind::custom("decryption-nonce"), [params.nonce.as_str()]))
.tag(Tag::custom(TagKind::custom("ox"), [file_hash.clone()]));
if !filename.is_empty() {
file_rumor = file_rumor.tag(Tag::custom(TagKind::custom("name"), [filename]));
}
if let Some(ref topic) = webxdc_topic {
file_rumor = file_rumor.tag(Tag::custom(TagKind::custom("webxdc-topic"), [topic.as_str()]));
}
if let Some(ref meta) = img_meta {
if !meta.thumbhash.is_empty() {
file_rumor = file_rumor.tag(Tag::custom(TagKind::custom("thumb"), [meta.thumbhash.as_str()]));
}
file_rumor = file_rumor.tag(Tag::custom(TagKind::custom("dim"), [format!("{}x{}", meta.width, meta.height)]));
}
file_rumor = file_rumor.tag(Tag::custom(TagKind::custom("ms"), [milliseconds.to_string()]));
if let Some(exp) = config.expiration {
file_rumor = file_rumor.tag(Tag::expiration(Timestamp::from_secs(exp)));
}
let built_rumor = file_rumor.build(my_pk);
let event_id = built_rumor.id.ok_or("Rumor has no id")?.to_hex();
retry_send_gift_wrap(
&client, &receiver, receiver_npub, &pending_id,
built_rumor, &event_id, config, callback,
).await
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Mutex;
#[derive(Debug, Clone, PartialEq)]
enum CbEvent {
Pending(String),
UploadProgress(String, u8, u64),
UploadComplete(String, String),
Sent(String, String),
Failed(String, String),
Persist(String),
}
struct MockCallback {
events: Mutex<Vec<CbEvent>>,
cancel_at: Option<u8>,
}
impl MockCallback {
fn new() -> Self { Self { events: Mutex::new(vec![]), cancel_at: None } }
fn with_cancel(pct: u8) -> Self { Self { events: Mutex::new(vec![]), cancel_at: Some(pct) } }
fn events(&self) -> Vec<CbEvent> { self.events.lock().unwrap().clone() }
}
impl SendCallback for MockCallback {
fn on_pending(&self, cid: &str, _: &Message) {
self.events.lock().unwrap().push(CbEvent::Pending(cid.into()));
}
fn on_upload_progress(&self, pid: &str, pct: u8, bytes: u64) -> Result<(), String> {
self.events.lock().unwrap().push(CbEvent::UploadProgress(pid.into(), pct, bytes));
if self.cancel_at.map_or(false, |c| pct >= c) { return Err("Cancelled".into()); }
Ok(())
}
fn on_upload_complete(&self, cid: &str, _: &str, _: &str, url: &str) {
self.events.lock().unwrap().push(CbEvent::UploadComplete(cid.into(), url.into()));
}
fn on_sent(&self, cid: &str, old: &str, _: &Message) {
self.events.lock().unwrap().push(CbEvent::Sent(cid.into(), old.into()));
}
fn on_failed(&self, cid: &str, old: &str, _: &Message) {
self.events.lock().unwrap().push(CbEvent::Failed(cid.into(), old.into()));
}
fn on_persist(&self, cid: &str, _: &Message) {
self.events.lock().unwrap().push(CbEvent::Persist(cid.into()));
}
}
#[test]
fn config_default() {
let c = SendConfig::default();
assert_eq!(c.max_send_attempts, 1);
assert!(!c.self_send);
assert!(c.cancel_token.is_none());
assert_eq!(c.upload_retries, 3);
}
#[test]
fn config_gui() {
let c = SendConfig::gui();
assert_eq!(c.max_send_attempts, 12);
assert!(c.self_send);
}
#[test]
fn config_headless() {
let c = SendConfig::headless();
assert_eq!(c.max_send_attempts, 3);
assert!(c.self_send);
}
#[test]
fn config_custom_override() {
let c = SendConfig { max_send_attempts: 5, ..SendConfig::gui() };
assert_eq!(c.max_send_attempts, 5);
assert!(c.self_send);
}
#[test]
fn noop_callback_all_methods() {
let cb = NoOpSendCallback;
let msg = Message::default();
cb.on_pending("c", &msg);
assert!(cb.on_upload_progress("p", 50, 1024).is_ok());
cb.on_upload_complete("c", "p", "a", "url");
cb.on_sent("c", "o", &msg);
cb.on_failed("c", "o", &msg);
cb.on_persist("c", &msg);
}
#[test]
fn text_dm_sequence() {
let cb = MockCallback::new();
let msg = Message::default();
cb.on_pending("r", &msg);
cb.on_sent("r", "p-1", &msg);
cb.on_persist("r", &msg);
assert_eq!(cb.events(), vec![
CbEvent::Pending("r".into()),
CbEvent::Sent("r".into(), "p-1".into()),
CbEvent::Persist("r".into()),
]);
}
#[test]
fn file_dm_sequence() {
let cb = MockCallback::new();
let msg = Message::default();
cb.on_pending("r", &msg);
cb.on_upload_progress("p", 0, 0).ok();
cb.on_upload_progress("p", 50, 5000).ok();
cb.on_upload_progress("p", 100, 10000).ok();
cb.on_upload_complete("r", "p", "h", "https://blossom/h");
cb.on_sent("r", "p", &msg);
cb.on_persist("r", &msg);
let e = cb.events();
assert_eq!(e.len(), 7);
assert!(matches!(&e[4], CbEvent::UploadComplete(_, url) if url.contains("blossom")));
}
#[test]
fn failed_sequence() {
let cb = MockCallback::new();
let msg = Message::default();
cb.on_pending("r", &msg);
cb.on_failed("r", "p", &msg);
cb.on_persist("r", &msg);
assert_eq!(cb.events(), vec![
CbEvent::Pending("r".into()),
CbEvent::Failed("r".into(), "p".into()),
CbEvent::Persist("r".into()),
]);
}
#[test]
fn cancel_upload_at_threshold() {
let cb = MockCallback::with_cancel(50);
assert!(cb.on_upload_progress("p", 25, 512).is_ok());
assert!(cb.on_upload_progress("p", 50, 1024).is_err());
assert_eq!(cb.events().len(), 2);
}
#[test]
fn cancel_triggers_failed() {
let cb = MockCallback::with_cancel(30);
let msg = Message::default();
cb.on_pending("r", &msg);
cb.on_upload_progress("p", 10, 1000).ok();
assert!(cb.on_upload_progress("p", 30, 3000).is_err());
cb.on_failed("r", "p", &msg);
assert!(matches!(cb.events().last(), Some(CbEvent::Failed(..))));
}
#[test]
fn send_result_serialize() {
let r = SendResult { pending_id: "p".into(), event_id: Some("e".into()), chat_id: "c".into() };
let j = serde_json::to_string(&r).unwrap();
assert!(j.contains("\"pending_id\":\"p\""));
}
#[test]
fn send_result_none_event() {
let r = SendResult { pending_id: "p".into(), event_id: None, chat_id: "c".into() };
let j = serde_json::to_string(&r).unwrap();
assert!(j.contains("null"));
}
#[test]
fn file_dm_fresh_upload_full_sequence() {
let cb = MockCallback::new();
let msg = Message::default();
cb.on_pending("npub1recv", &msg);
cb.on_upload_progress("pending-42", 0, 0).unwrap();
cb.on_upload_progress("pending-42", 25, 2500).unwrap();
cb.on_upload_progress("pending-42", 50, 5000).unwrap();
cb.on_upload_progress("pending-42", 75, 7500).unwrap();
cb.on_upload_progress("pending-42", 100, 10000).unwrap();
cb.on_upload_complete("npub1recv", "pending-42", "deadbeef", "https://blossom.example/deadbeef");
cb.on_sent("npub1recv", "pending-42", &msg);
cb.on_persist("npub1recv", &msg);
let e = cb.events();
assert_eq!(e.len(), 9);
assert!(matches!(&e[0], CbEvent::Pending(c) if c == "npub1recv"));
assert!(matches!(&e[1], CbEvent::UploadProgress(_, 0, 0)));
assert!(matches!(&e[5], CbEvent::UploadProgress(_, 100, 10000)));
assert!(matches!(&e[6], CbEvent::UploadComplete(_, url) if url.contains("deadbeef")));
assert!(matches!(&e[7], CbEvent::Sent(..)));
assert!(matches!(&e[8], CbEvent::Persist(..)));
}
#[test]
fn file_dm_skip_upload_sequence() {
let cb = MockCallback::new();
let msg = Message::default();
cb.on_pending("npub1recv", &msg);
cb.on_upload_complete("npub1recv", "pending-99", "existinghash", "https://blossom.example/existing");
cb.on_sent("npub1recv", "pending-99", &msg);
cb.on_persist("npub1recv", &msg);
let e = cb.events();
assert_eq!(e.len(), 4);
assert!(!e.iter().any(|ev| matches!(ev, CbEvent::UploadProgress(..))));
assert!(matches!(&e[1], CbEvent::UploadComplete(..)));
}
#[test]
fn file_dm_upload_cancelled_at_30pct() {
let cb = MockCallback::with_cancel(30);
let msg = Message::default();
cb.on_pending("npub1recv", &msg);
assert!(cb.on_upload_progress("p", 10, 1000).is_ok());
assert!(cb.on_upload_progress("p", 20, 2000).is_ok());
let err = cb.on_upload_progress("p", 30, 3000);
assert!(err.is_err());
assert!(err.unwrap_err().contains("Cancelled"));
cb.on_failed("npub1recv", "p", &msg);
let e = cb.events();
assert_eq!(e.len(), 5);
assert!(!e.iter().any(|ev| matches!(ev, CbEvent::Sent(..))));
assert!(matches!(e.last(), Some(CbEvent::Failed(..))));
}
#[test]
fn file_dm_upload_fails_marks_failed() {
let cb = MockCallback::new();
let msg = Message::default();
cb.on_pending("npub1recv", &msg);
cb.on_upload_progress("p", 0, 0).ok();
cb.on_upload_progress("p", 10, 500).ok();
cb.on_failed("npub1recv", "p", &msg);
cb.on_persist("npub1recv", &msg);
let e = cb.events();
assert_eq!(e.len(), 5);
assert!(matches!(&e[3], CbEvent::Failed(..)));
assert!(matches!(&e[4], CbEvent::Persist(..)));
assert!(!e.iter().any(|ev| matches!(ev, CbEvent::UploadComplete(..))));
assert!(!e.iter().any(|ev| matches!(ev, CbEvent::Sent(..))));
}
#[test]
fn file_dm_gift_wrap_fails_after_upload() {
let cb = MockCallback::new();
let msg = Message::default();
cb.on_pending("npub1recv", &msg);
cb.on_upload_progress("p", 100, 10000).ok();
cb.on_upload_complete("npub1recv", "p", "hash", "https://blossom/hash");
cb.on_failed("npub1recv", "p", &msg);
cb.on_persist("npub1recv", &msg);
let e = cb.events();
assert_eq!(e.len(), 5);
assert!(matches!(&e[2], CbEvent::UploadComplete(..)));
assert!(matches!(&e[3], CbEvent::Failed(..)));
}
#[test]
fn file_dm_with_image_metadata_sequence() {
let cb = MockCallback::new();
let msg = Message::default();
cb.on_pending("npub1recv", &msg);
cb.on_upload_progress("p", 0, 0).ok();
cb.on_upload_progress("p", 50, 50000).ok();
cb.on_upload_progress("p", 100, 100000).ok();
cb.on_upload_complete("npub1recv", "p", "imghash", "https://blossom/imghash.jpg");
cb.on_sent("npub1recv", "p", &msg);
cb.on_persist("npub1recv", &msg);
let e = cb.events();
assert_eq!(e.len(), 7);
assert!(matches!(&e[0], CbEvent::Pending(..)));
assert!(matches!(&e[4], CbEvent::UploadComplete(_, url) if url.ends_with(".jpg")));
assert!(matches!(&e[5], CbEvent::Sent(..)));
}
#[test]
fn cancel_token_config_with_upload() {
let token = Arc::new(std::sync::atomic::AtomicBool::new(false));
let c = SendConfig {
cancel_token: Some(token.clone()),
..SendConfig::gui()
};
assert!(c.cancel_token.is_some());
assert!(!token.load(std::sync::atomic::Ordering::Relaxed));
token.store(true, std::sync::atomic::Ordering::Relaxed);
assert!(c.cancel_token.as_ref().unwrap().load(std::sync::atomic::Ordering::Relaxed));
}
}