WIP: synchronization of transports

link2xt · link2xt · commit 3b2f96a477e1 · 2025-11-27T00:10:29.000Z
diff --git a/deltachat-ffi/src/lib.rs b/deltachat-ffi/src/lib.rs
@@ -559,6 +559,7 @@ pub unsafe extern "C" fn dc_event_get_id(event: *mut dc_event_t) -> libc::c_int
         EventType::IncomingCallAccepted { .. } => 2560,
         EventType::OutgoingCallAccepted { .. } => 2570,
         EventType::CallEnded { .. } => 2580,
+        EventType::TransportsModified => 2600,
         #[allow(unreachable_patterns)]
         #[cfg(test)]
         _ => unreachable!("This is just to silence a rust_analyzer false-positive"),
@@ -593,7 +594,8 @@ pub unsafe extern "C" fn dc_event_get_data1_int(event: *mut dc_event_t) -> libc:
         | EventType::AccountsBackgroundFetchDone
         | EventType::ChatlistChanged
         | EventType::AccountsChanged
-        | EventType::AccountsItemChanged => 0,
+        | EventType::AccountsItemChanged
+        | EventType::TransportsModified => 0,
         EventType::IncomingReaction { contact_id, .. }
         | EventType::IncomingWebxdcNotify { contact_id, .. } => contact_id.to_u32() as libc::c_int,
         EventType::MsgsChanged { chat_id, .. }
@@ -681,7 +683,8 @@ pub unsafe extern "C" fn dc_event_get_data2_int(event: *mut dc_event_t) -> libc:
         | EventType::IncomingCallAccepted { .. }
         | EventType::OutgoingCallAccepted { .. }
         | EventType::CallEnded { .. }
-        | EventType::EventChannelOverflow { .. } => 0,
+        | EventType::EventChannelOverflow { .. }
+        | EventType::TransportsModified => 0,
         EventType::MsgsChanged { msg_id, .. }
         | EventType::ReactionsChanged { msg_id, .. }
         | EventType::IncomingReaction { msg_id, .. }
@@ -780,7 +783,8 @@ pub unsafe extern "C" fn dc_event_get_data2_str(event: *mut dc_event_t) -> *mut
         | EventType::AccountsChanged
         | EventType::AccountsItemChanged
         | EventType::IncomingCallAccepted { .. }
-        | EventType::WebxdcRealtimeAdvertisementReceived { .. } => ptr::null_mut(),
+        | EventType::WebxdcRealtimeAdvertisementReceived { .. }
+        | EventType::TransportsModified => ptr::null_mut(),
         EventType::IncomingCall {
             place_call_info, ..
         } => {
diff --git a/deltachat-jsonrpc/src/api/types/events.rs b/deltachat-jsonrpc/src/api/types/events.rs
@@ -460,6 +460,11 @@ pub enum EventType {
         /// ID of the chat which the message belongs to.
         chat_id: u32,
     },
+
+    /// One or more transports has changed.
+    ///
+    /// Transport list should be refreshed.
+    TransportsModified,
 }
 
 impl From<CoreEventType> for EventType {
@@ -642,6 +647,8 @@ impl From<CoreEventType> for EventType {
                 msg_id: msg_id.to_u32(),
                 chat_id: chat_id.to_u32(),
             },
+            CoreEventType::TransportsModified => TransportsModified,
+
             #[allow(unreachable_patterns)]
             #[cfg(test)]
             _ => unreachable!("This is just to silence a rust_analyzer false-positive"),
diff --git a/deltachat-rpc-client/src/deltachat_rpc_client/const.py b/deltachat-rpc-client/src/deltachat_rpc_client/const.py
@@ -80,6 +80,7 @@ class EventType(str, Enum):
     CONFIG_SYNCED = "ConfigSynced"
     WEBXDC_REALTIME_DATA = "WebxdcRealtimeData"
     WEBXDC_REALTIME_ADVERTISEMENT_RECEIVED = "WebxdcRealtimeAdvertisementReceived"
+    TRANSPORTS_MODIFIED = "TransportsModified"
 
 
 class ChatId(IntEnum):
diff --git a/deltachat-rpc-client/tests/test_multitransport.py b/deltachat-rpc-client/tests/test_multitransport.py
@@ -1,5 +1,6 @@
 import pytest
 
+from deltachat_rpc_client import EventType
 from deltachat_rpc_client.rpc import JsonRpcError
 
 
@@ -156,3 +157,47 @@ def test_reconfigure_transport(acfactory) -> None:
     # Reconfiguring the transport should not reset
     # the settings as if when configuring the first transport.
     assert account.get_config("mvbox_move") == "1"
+
+
+def test_transport_synchronization(acfactory, log) -> None:
+    """Test synchronization of transports between devices."""
+    ac1, ac2 = acfactory.get_online_accounts(2)
+    ac1_clone = ac1.clone()
+    ac1_clone.bring_online()
+
+    qr = acfactory.get_account_qr()
+
+    ac1.add_transport_from_qr(qr)
+    ac1_clone.wait_for_event(EventType.TRANSPORTS_MODIFIED)
+    assert len(ac1.list_transports()) == 2
+    assert len(ac1_clone.list_transports()) == 2
+
+    ac1_clone.add_transport_from_qr(qr)
+    ac1.wait_for_event(EventType.TRANSPORTS_MODIFIED)
+    assert len(ac1.list_transports()) == 3
+    assert len(ac1_clone.list_transports()) == 3
+
+    log.section("ac1 clone removes second transport")
+    [transport1, transport2, transport3] = ac1_clone.list_transports()
+    addr3 = transport3["addr"]
+    ac1_clone.delete_transport(transport2["addr"])
+
+    ac1.wait_for_event(EventType.TRANSPORTS_MODIFIED)
+    [transport1, transport3] = ac1.list_transports()
+
+    log.section("ac1 changes the primary transport")
+    ac1.set_config("configured_addr", transport3["addr"])
+
+    log.section("ac1 removes the first transport")
+    ac1.delete_transport(transport1["addr"])
+
+    ac1_clone.wait_for_event(EventType.TRANSPORTS_MODIFIED)
+    [transport3] = ac1_clone.list_transports()
+    assert transport3["addr"] == addr3
+    assert ac1_clone.get_config("configured_addr") == addr3
+
+    ac2_chat = ac2.create_chat(ac1)
+    ac2_chat.send_text("Hello!")
+
+    assert ac1.wait_for_incoming_msg().get_snapshot().text == "Hello!"
+    assert ac1_clone.wait_for_incoming_msg().get_snapshot().text == "Hello!"
diff --git a/src/config.rs b/src/config.rs
@@ -21,7 +21,6 @@ use crate::mimefactory::RECOMMENDED_FILE_SIZE;
 use crate::provider::{Provider, get_provider_by_id};
 use crate::sync::{self, Sync::*, SyncData};
 use crate::tools::get_abs_path;
-use crate::transport::ConfiguredLoginParam;
 use crate::{constants, stats};
 
 /// The available configuration keys.
@@ -820,11 +819,19 @@ impl Context {
                         self,
                         "Creating a pseudo configured account which will not be able to send or receive messages. Only meant for tests!"
                     );
-                    ConfiguredLoginParam::from_json(&format!(
-                        r#"{{"addr":"{addr}","imap":[],"imap_user":"","imap_password":"","smtp":[],"smtp_user":"","smtp_password":"","certificate_checks":"Automatic","oauth2":false}}"#
-                    ))?
-                    .save_to_transports_table(self, &EnteredLoginParam::default())
-                    .await?;
+                    self.sql
+                        .execute(
+                            "INSERT INTO transports (addr, entered_param, configured_param) VALUES (?, ?, ?)",
+                            (
+                                addr,
+                                serde_json::to_string(&EnteredLoginParam::default())?,
+                                r#"{{"addr":"{addr}","imap":[],"imap_user":"","imap_password":"","smtp":[],"smtp_user":"","smtp_password":"","certificate_checks":"Automatic","oauth2":false}}"#
+                            ),
+                        )
+                        .await?;
+                    self.sql
+                        .set_raw_config(Config::ConfiguredAddr.as_ref(), Some(addr))
+                        .await?;
                 }
                 self.sql
                     .transaction(|transaction| {
diff --git a/src/configure.rs b/src/configure.rs
@@ -40,7 +40,7 @@ use crate::sync::Sync::*;
 use crate::tools::time;
 use crate::transport::{
     ConfiguredCertificateChecks, ConfiguredLoginParam, ConfiguredServerLoginParam,
-    ConnectionCandidate,
+    ConnectionCandidate, sync_transports,
 };
 use crate::{EventType, stock_str};
 use crate::{chat, provider};
@@ -205,6 +205,7 @@ impl Context {
     /// Removes the transport with the specified email address
     /// (i.e. [EnteredLoginParam::addr]).
     pub async fn delete_transport(&self, addr: &str) -> Result<()> {
+        let now = time();
         self.sql
             .transaction(|transaction| {
                 let primary_addr = transaction.query_row(
@@ -232,10 +233,19 @@ impl Context {
                     "DELETE FROM imap_sync WHERE transport_id=?",
                     (transport_id,),
                 )?;
+                transaction.execute(
+                    "INSERT INTO removed_transports (addr, remove_timestamp)
+                     VALUES (?, ?)
+                     ON CONFLICT (addr)
+                     DO UPDATE SET remove_timestamp = excluded.remove_timestamp",
+                    (addr, now),
+                )?;
 
                 Ok(())
             })
             .await?;
+        sync_transports(self).await?;
+
         Ok(())
     }
 
@@ -553,7 +563,8 @@ async fn configure(ctx: &Context, param: &EnteredLoginParam) -> Result<Option<&'
 
     progress!(ctx, 900);
 
-    if !ctx.is_configured().await? {
+    let is_configured = ctx.is_configured().await?;
+    if !is_configured {
         ctx.sql.set_raw_config("mvbox_move", Some("0")).await?;
         ctx.sql.set_raw_config("only_fetch_mvbox", None).await?;
     }
@@ -568,8 +579,10 @@ async fn configure(ctx: &Context, param: &EnteredLoginParam) -> Result<Option<&'
 
     let provider = configured_param.provider;
     configured_param
-        .save_to_transports_table(ctx, param)
+        .clone()
+        .save_to_transports_table(ctx, param, time())
         .await?;
+    sync_transports(ctx).await?;
 
     ctx.set_config_internal(Config::ConfiguredTimestamp, Some(&time().to_string()))
         .await?;
diff --git a/src/events/payload.rs b/src/events/payload.rs
@@ -417,6 +417,11 @@ pub enum EventType {
         chat_id: ChatId,
     },
 
+    /// One or more transports has changed.
+    ///
+    /// Transport list should be refreshed.
+    TransportsModified,
+
     /// Event for using in tests, e.g. as a fence between normally generated events.
     #[cfg(test)]
     Test,
diff --git a/src/receive_imf.rs b/src/receive_imf.rs
@@ -827,6 +827,11 @@ pub(crate) async fn receive_imf_inner(
     if let Some(ref sync_items) = mime_parser.sync_items {
         if from_id == ContactId::SELF {
             if mime_parser.was_encrypted() {
+                // Receiving encrypted message from self updates primary transport.
+                context
+                    .sql
+                    .set_raw_config("configured_addr", Some(&mime_parser.from.addr))
+                    .await?;
                 context
                     .execute_sync_items(sync_items, mime_parser.timestamp_sent)
                     .await;
diff --git a/src/sql/migrations.rs b/src/sql/migrations.rs
@@ -1445,6 +1445,21 @@ CREATE INDEX imap_sync_index ON imap_sync(transport_id, folder);
         .await?;
     }
 
+    inc_and_check(&mut migration_version, 141)?;
+    if dbversion < migration_version {
+        sql.execute_migration(
+            "ALTER TABLE transports
+             ADD COLUMN add_timestamp INTEGER NOT NULL DEFAULT 0;
+             CREATE TABLE removed_transports (
+                 addr TEXT NOT NULL,
+                 remove_timestamp INTEGER NOT NULL,
+                 UNIQUE(addr)
+             ) STRICT;",
+            migration_version,
+        )
+        .await?;
+    }
+
     let new_version = sql
         .get_raw_config_int(VERSION_CFG)
         .await?
diff --git a/src/sync.rs b/src/sync.rs
@@ -9,14 +9,16 @@ use crate::config::Config;
 use crate::constants::Blocked;
 use crate::contact::ContactId;
 use crate::context::Context;
-use crate::log::LogExt;
-use crate::log::warn;
+use crate::events::EventType;
+use crate::log::{LogExt as _, warn};
+use crate::login_param::EnteredLoginParam;
 use crate::message::{Message, MsgId, Viewtype};
 use crate::mimeparser::SystemMessage;
 use crate::param::Param;
 use crate::sync::SyncData::{AddQrToken, AlterChat, DeleteQrToken};
 use crate::token::Namespace;
 use crate::tools::time;
+use crate::transport::{ConfiguredLoginParamJson, save_transport};
 use crate::{message, stock_str, token};
 use std::collections::HashSet;
 
@@ -52,6 +54,29 @@ pub(crate) struct QrTokenData {
     pub(crate) grpid: Option<String>,
 }
 
+#[derive(Debug, Serialize, Deserialize)]
+pub(crate) struct TransportData {
+    /// Configured login parameters.
+    pub(crate) configured: ConfiguredLoginParamJson,
+
+    /// Login parameters entered by the user.
+    ///
+    /// They can be used to reconfigure the transport.
+    pub(crate) entered: EnteredLoginParam,
+
+    /// Timestamp of when the transport was last time (re)configured.
+    pub(crate) timestamp: i64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub(crate) struct RemovedTransportData {
+    /// Address of the removed transport.
+    pub(crate) addr: String,
+
+    /// Timestamp of when the transport was removed.
+    pub(crate) timestamp: i64,
+}
+
 #[derive(Debug, Serialize, Deserialize)]
 pub(crate) enum SyncData {
     AddQrToken(QrTokenData),
@@ -71,6 +96,13 @@ pub(crate) enum SyncData {
     DeleteMessages {
         msgs: Vec<String>, // RFC724 id (i.e. "Message-Id" header)
     },
+
+    /// Update transport configuration.
+    Transports {
+        transports: Vec<TransportData>,
+
+        removed_transports: Vec<RemovedTransportData>,
+    },
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -274,6 +306,10 @@ impl Context {
                     SyncData::Config { key, val } => self.sync_config(key, val).await,
                     SyncData::SaveMessage { src, dest } => self.save_message(src, dest).await,
                     SyncData::DeleteMessages { msgs } => self.sync_message_deletion(msgs).await,
+                    SyncData::Transports {
+                        transports,
+                        removed_transports,
+                    } => self.sync_transports(transports, removed_transports).await,
                 },
                 SyncDataOrUnknown::Unknown(data) => {
                     warn!(self, "Ignored unknown sync item: {data}.");
@@ -347,6 +383,46 @@ impl Context {
         message::delete_msgs_locally_done(self, &msg_ids, modified_chat_ids).await?;
         Ok(())
     }
+
+    /// Process received data for transport synchronization.
+    async fn sync_transports(
+        &self,
+        transports: &[TransportData],
+        removed_transports: &[RemovedTransportData],
+    ) -> Result<()> {
+        for TransportData {
+            configured,
+            entered,
+            timestamp,
+        } in transports
+        {
+            save_transport(self, entered, configured, *timestamp).await?;
+        }
+
+        self.sql
+            .transaction(|transaction| {
+                for RemovedTransportData { addr, timestamp } in removed_transports {
+                    transaction.execute(
+                        "DELETE FROM transports
+                         WHERE addr=? AND add_timestamp<=?",
+                        (addr, timestamp),
+                    )?;
+                    transaction.execute(
+                        "INSERT INTO removed_transports (addr, remove_timestamp)
+                         VALUES (?, ?)
+                         ON CONFLICT (addr) DO
+                         UPDATE SET remove_timestamp = excluded.remove_timestamp
+                         WHERE excluded.remove_timestamp > remove_timestamp",
+                        (addr, timestamp),
+                    )?;
+                }
+                Ok(())
+            })
+            .await?;
+
+        self.emit_event(EventType::TransportsModified);
+        Ok(())
+    }
 }
 
 #[cfg(test)]
diff --git a/src/test_utils.rs b/src/test_utils.rs
@@ -587,7 +587,7 @@ impl TestContext {
         self.ctx
             .set_config(Config::ConfiguredAddr, Some(addr))
             .await
-            .unwrap();
+            .expect("Failed to configure address");
 
         if let Some(name) = addr.split('@').next() {
             self.set_name(name);
diff --git a/src/transport.rs b/src/transport.rs
