fix: use initial topic on resend

Author: Septias

src/mimefactory.rs

@@ -7,7 +7,9 @@ use std::path::Path;
 use anyhow::{bail, Context as _, Result};
 use base64::Engine as _;
 use chrono::TimeZone;
+use data_encoding::BASE32_NOPAD;
 use deltachat_contact_tools::sanitize_bidi_characters;
+use iroh_gossip::proto::TopicId;
 use mail_builder::headers::address::{Address, EmailAddress};
 use mail_builder::headers::HeaderType;
 use mail_builder::mime::MimePart;
@@ -29,7 +31,7 @@ use crate::location;
 use crate::message::{self, Message, MsgId, Viewtype};
 use crate::mimeparser::{is_hidden, SystemMessage};
 use crate::param::Param;
-use crate::peer_channels::create_iroh_header;
+use crate::peer_channels::{create_iroh_header, get_iroh_topic_for_msg};
 use crate::peerstate::Peerstate;
 use crate::simplify::escape_message_footer_marks;
 use crate::stock_str;
@@ -126,6 +128,9 @@ pub struct MimeFactory {
 
     /// True if the avatar should be attached.
     pub attach_selfavatar: bool,
+
+    /// Sustain webxdc topic on resend.
+    webxdc_topic: Option<TopicId>,
 }
 
 /// Result of rendering a message, ready to be submitted to a send job.
@@ -319,6 +324,7 @@ impl MimeFactory {
             member_timestamps.is_empty()
                 || to.len() + past_members.len() == member_timestamps.len()
         );
+        let webxdc_topic = get_iroh_topic_for_msg(context, msg.id).await?;
         let factory = MimeFactory {
             from_addr,
             from_displayname,
@@ -336,6 +342,7 @@ impl MimeFactory {
             last_added_location_id: None,
             sync_ids_to_delete: None,
             attach_selfavatar,
+            webxdc_topic,
         };
         Ok(factory)
     }
@@ -370,6 +377,7 @@ impl MimeFactory {
             last_added_location_id: None,
             sync_ids_to_delete: None,
             attach_selfavatar: false,
+            webxdc_topic: None,
         };
 
         Ok(res)
@@ -1569,10 +1577,13 @@ impl MimeFactory {
             let json = msg.param.get(Param::Arg).unwrap_or_default();
             parts.push(context.build_status_update_part(json));
         } else if msg.viewtype == Viewtype::Webxdc {
+            let topic = self
+                .webxdc_topic
+                .map(|top| BASE32_NOPAD.encode(top.as_bytes()).to_ascii_lowercase())
+                .unwrap_or(create_iroh_header(context, msg.id).await?);
             headers.push((
                 HeaderDef::IrohGossipTopic.into(),
-                mail_builder::headers::raw::Raw::new(create_iroh_header(context, msg.id).await?)
-                    .into(),
+                mail_builder::headers::raw::Raw::new(topic).into(),
             ));
             if let (Some(json), _) = context
                 .render_webxdc_status_update_object(

src/peer_channels.rs

@@ -108,7 +108,7 @@ impl Iroh {
 
         info!(
             ctx,
-            "IROH_REALTIME: Joining gossip with peers: {:?}", node_ids,
+            "IROH_REALTIME: Joining gossip {topic} with peers: {:?}", node_ids,
         );
 
         // Inform iroh of potentially new node addresses
@@ -141,11 +141,11 @@ impl Iroh {
     pub async fn maybe_add_gossip_peer(&self, topic: TopicId, peer: NodeAddr) -> Result<()> {
         if self.iroh_channels.read().await.get(&topic).is_some() {
             self.router.endpoint().add_node_addr(peer.clone())?;
-
-            self.gossip
-                .subscribe_with_opts(topic, JoinOptions::with_bootstrap(vec![peer.node_id]));
+            self.gossip.subscribe(topic, vec![peer.node_id])?;
+            Ok(())
+        } else {
+            anyhow::bail!("can not read iroh channels");
         }
-        Ok(())
     }
 
     /// Send realtime data to the gossip swarm.
@@ -316,7 +316,7 @@ impl Context {
         if let Some(iroh) = &*self.iroh.read().await {
             info!(
                 self,
-                "Adding (maybe existing) peer with id {} to gossip", peer.node_id
+                "Adding (maybe existing) peer with id {} to {topic}", peer.node_id
             );
             iroh.maybe_add_gossip_peer(topic, peer).await?;
         }
@@ -559,6 +559,10 @@ async fn subscribe_loop(
 
 #[cfg(test)]
 mod tests {
+    use std::time::Duration;
+
+    use tokio::time::timeout;
+
     use super::*;
     use crate::{
         chat::{self, add_contact_to_chat, resend_msgs, send_msg, ChatId, ProtectionStatus},
@@ -986,7 +990,10 @@ mod tests {
         let fiona_advert = fiona.pop_sent_msg().await;
         alice.recv_msg_trash(&fiona_advert).await;
 
-        fiona_connect_future.await.unwrap();
+        timeout(Duration::from_secs(2), fiona_connect_future)
+            .await
+            .unwrap()
+            .unwrap();
         send_webxdc_realtime_data(alice, instance.id, b"alice -> bob & fiona".into())
             .await
             .unwrap();
@@ -1030,17 +1037,19 @@ mod tests {
             .unwrap();
         let alice_advertisement = alice.pop_sent_msg().await;
 
-        send_webxdc_realtime_advertisement(bob, bob_webxdc.id)
+        let bob_advertisement_future = send_webxdc_realtime_advertisement(bob, bob_webxdc.id)
             .await
+            .unwrap()
             .unwrap();
         let bob_advertisement = bob.pop_sent_msg().await;
 
         eprintln!("Receiving advertisements");
         bob.recv_msg_trash(&alice_advertisement).await;
         alice.recv_msg_trash(&bob_advertisement).await;
 
-        eprintln!("Alice waits for connection");
+        eprintln!("Alice and Bob wait for connection");
         alice_advertisement_future.await.unwrap();
+        bob_advertisement_future.await.unwrap();
 
         // Alice sends ephemeral message
         eprintln!("Sending ephemeral message");

src/receive_imf.rs

@@ -1659,7 +1659,7 @@ RETURNING id
         created_db_entries.push(row_id);
     }
 
-    // check all parts whether they contain a new logging webxdc
+    // Maybe set logging xdc and add gossip topics for webxdcs.
     for (part, msg_id) in mime_parser.parts.iter().zip(&created_db_entries) {
         // check if any part contains a webxdc topic id
         if part.typ == Viewtype::Webxdc {