feat: synchronize transports via sync messages

Author: link2xt

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!"

src/config.rs

@@ -819,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())?,
+                                format!(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| {

src/configure.rs

@@ -40,7 +40,7 @@ use crate::sync::Sync::*;
 use crate::tools::time;
 use crate::transport::{
     ConfiguredCertificateChecks, ConfiguredLoginParam, ConfiguredServerLoginParam,
-    ConnectionCandidate,
+    ConnectionCandidate, send_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?;
+        send_sync_transports(self).await?;
+
         Ok(())
     }
 
@@ -552,7 +562,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?;
     }
@@ -563,8 +574,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?;
+    send_sync_transports(ctx).await?;
 
     ctx.set_config_internal(Config::ConfiguredTimestamp, Some(&time().to_string()))
         .await?;

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;

src/sql/migrations.rs

@@ -1439,6 +1439,21 @@ CREATE INDEX imap_sync_index ON imap_sync(transport_id, folder);
         .await?;
     }
 
+    inc_and_check(&mut migration_version, 142)?;
+    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?

src/sync.rs

@@ -9,14 +9,15 @@ 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::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, sync_transports};
 use crate::{message, stock_str, token};
 use std::collections::HashSet;
 
@@ -52,6 +53,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 +95,28 @@ pub(crate) enum SyncData {
     DeleteMessages {
         msgs: Vec<String>, // RFC724 id (i.e. "Message-Id" header)
     },
+
+    /// Update transport configuration.
+    ///
+    /// This message contains a list of all added transports
+    /// together with their addition timestamp,
+    /// and all removed transports together with
+    /// the removal timestamp.
+    ///
+    /// In case of a tie, addition and removal timestamps
+    /// being the same, removal wins.
+    /// It is more likely that transport is added
+    /// and then removed within a second,
+    /// but unlikely the other way round
+    /// as adding new transport takes time
+    /// to run configuration.
+    Transports {
+        /// Active transports.
+        transports: Vec<TransportData>,
+
+        /// Removed transports with the timestamp of removal.
+        removed_transports: Vec<RemovedTransportData>,
+    },
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -274,6 +320,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,
+                    } => sync_transports(self, transports, removed_transports).await,
                 },
                 SyncDataOrUnknown::Unknown(data) => {
                     warn!(self, "Ignored unknown sync item: {data}.");

src/test_utils.rs

@@ -600,7 +600,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);