Switch to simpler fix

This reverts commit b501ab1532.

.github/workflows/ci.yml

@@ -40,6 +40,18 @@ jobs:
       - name: Check
         run: cargo check --workspace --all-targets --all-features
 
+  npm_constants:
+    name: Check if node constants are up to date
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          show-progress: false
+      - name: Rebuild constants
+        run: npm run build:core:constants
+      - name: Check that constants are not changed
+        run: git diff --exit-code
+
   cargo_deny:
     name: cargo deny
     runs-on: ubuntu-latest

.github/workflows/jsonrpc-client-npm-package.yml

@@ -1,82 +1,38 @@
-name: "jsonrpc js client build"
+name: "Publish @deltachat/jsonrpc-client"
 on:
-  pull_request:
-  push:
-    tags:
-      - "*"
-      - "!py-*"
+  workflow_dispatch:
+  release:
+    types: [published]
 
 jobs:
   pack-module:
-    name: "Package @deltachat/jsonrpc-client and upload to download.delta.chat"
+    name: "Publish @deltachat/jsonrpc-client"
     runs-on: ubuntu-20.04
+    permissions:
+      id-token: write
+      contents: read
     steps:
-      - name: Install tree
-        run: sudo apt install tree
       - uses: actions/checkout@v4
         with:
           show-progress: false
+
       - uses: actions/setup-node@v4
         with:
-          node-version: "18"
-      - name: Get tag
-        id: tag
-        uses: dawidd6/action-get-tag@v1
-        continue-on-error: true
-      - name: Get Pull Request ID
-        id: prepare
-        run: |
-          tag=${{ steps.tag.outputs.tag }}
-          if [ -z "$tag" ]; then
-            node -e "console.log('DELTACHAT_JSONRPC_TAR_GZ=deltachat-jsonrpc-client-' + '${{ github.ref }}'.split('/')[2] + '.tar.gz')" >> $GITHUB_ENV
-          else
-            echo "DELTACHAT_JSONRPC_TAR_GZ=deltachat-jsonrpc-client-${{ steps.tag.outputs.tag }}.tar.gz" >> $GITHUB_ENV
-            echo "No preview will be uploaded this time, but the $tag release"
-          fi
-      - name: System info
-        run: |
-          npm --version
-          node --version
-          echo $DELTACHAT_JSONRPC_TAR_GZ
+          node-version: 20
+          registry-url: "https://registry.npmjs.org"
+
       - name: Install dependencies without running scripts
         working-directory: deltachat-jsonrpc/typescript
         run: npm install --ignore-scripts
+
       - name: Package
-        shell: bash
         working-directory: deltachat-jsonrpc/typescript
         run: |
           npm run build
           npm pack .
-          ls -lah
-          mv $(find deltachat-jsonrpc-client-*) $DELTACHAT_JSONRPC_TAR_GZ
-      - name: Upload Prebuild
-        uses: actions/upload-artifact@v4
-        with:
-          name: deltachat-jsonrpc-client.tgz
-          path: deltachat-jsonrpc/typescript/${{ env.DELTACHAT_JSONRPC_TAR_GZ }}
-      # Upload to download.delta.chat/node/preview/
-      - name: Upload deltachat-jsonrpc-client preview to download.delta.chat/node/preview/
-        if: ${{ ! steps.tag.outputs.tag }}
-        id: upload-preview
-        shell: bash
-        run: |
-          echo -e "${{ secrets.SSH_KEY }}" >__TEMP_INPUT_KEY_FILE
-          chmod 600 __TEMP_INPUT_KEY_FILE
-          scp -o StrictHostKeyChecking=no -v -i __TEMP_INPUT_KEY_FILE -P "22" -r deltachat-jsonrpc/typescript/$DELTACHAT_JSONRPC_TAR_GZ "${{ secrets.USERNAME }}"@"download.delta.chat":"/var/www/html/download/node/preview/"
-        continue-on-error: true
-      - name: Post links to details
-        if: steps.upload-preview.outcome == 'success'
-        run: node ./node/scripts/postLinksToDetails.js
+
+      - name: Publish
+        working-directory: deltachat-jsonrpc/typescript
+        run: npm publish --provenance deltachat-jsonrpc-client-*
         env:
-          URL: preview/${{ env.DELTACHAT_JSONRPC_TAR_GZ }}
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-          MSG_CONTEXT: Download the deltachat-jsonrpc-client.tgz
-      # Upload to download.delta.chat/node/
-      - name: Upload deltachat-jsonrpc-client build to download.delta.chat/node/
-        if: ${{ steps.tag.outputs.tag }}
-        id: upload
-        shell: bash
-        run: |
-          echo -e "${{ secrets.SSH_KEY }}" >__TEMP_INPUT_KEY_FILE
-          chmod 600 __TEMP_INPUT_KEY_FILE
-          scp -o StrictHostKeyChecking=no -v -i __TEMP_INPUT_KEY_FILE -P "22" -r deltachat-jsonrpc/typescript/$DELTACHAT_JSONRPC_TAR_GZ "${{ secrets.USERNAME }}"@"download.delta.chat":"/var/www/html/download/node/"
+          NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}

.gitignore

@@ -34,7 +34,6 @@ deltachat-ffi/xml
 coverage/
 .DS_Store
 .vscode
-.vscode/launch.json
 python/accounts.txt
 python/all-testaccounts.txt
 tmp/

Cargo.toml

@@ -60,7 +60,10 @@ hex = "0.4.0"
 hickory-resolver = "0.24"
 humansize = "2"
 image = { version = "0.25.1", default-features=false, features = ["gif", "jpeg", "ico", "png", "pnm", "webp", "bmp"] }
-iroh = { version = "0.4.2", default-features = false }
+iroh_old = { version = "0.4.2", default-features = false, package = "iroh"}
+iroh-net = { git = "https://github.com/link2xt/iroh", branch="link2xt/keep-connection" }
+iroh-gossip = { git = "https://github.com/link2xt/iroh", branch="link2xt/keep-connection", features = ["net"] }
+quinn = "0.10.0"
 kamadak-exif = "0.5.3"
 lettre_email = { git = "https://github.com/deltachat/lettre", branch = "master" }
 libc = "0.2"
@@ -79,7 +82,7 @@ quick-xml = "0.31"
 quoted_printable = "0.5"
 rand = "0.8"
 regex = { workspace = true }
-reqwest = { version = "0.12.2", features = ["json"] }
+reqwest = { version = "0.11.27", features = ["json"] }
 rusqlite = { workspace = true, features = ["sqlcipher"] }
 rust-hsluv = "0.1"
 sanitize-filename = "0.5"
@@ -101,6 +104,7 @@ tokio-util = "0.7.9"
 toml = "0.8"
 url = "2"
 uuid = { version = "1", features = ["serde", "v4"] }
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 
 # Pin OpenSSL to 3.1 releases.
 # OpenSSL 3.2 has a regression tracked at <https://github.com/openssl/openssl/issues/23376>
@@ -109,6 +113,8 @@ uuid = { version = "1", features = ["serde", "v4"] }
 # According to <https://www.openssl.org/policies/releasestrat.html>
 # 3.1 branch will be supported until 2025-03-14.
 openssl-src = "~300.1"
+tracing = "0.1.40"
+
 
 [dev-dependencies]
 ansi_term = "0.12.0"

RELEASE.md

@@ -4,21 +4,18 @@ For example, to release version 1.116.0 of the core, do the following steps.
 
 1. Resolve all [blocker issues](https://github.com/deltachat/deltachat-core-rust/labels/blocker).
 
-2. Run `npm run build:core:constants` in the root of the repository
-   and commit generated `node/constants.js`, `node/events.js` and `node/lib/constants.js`.
+2. Update the changelog: `git cliff --unreleased --tag 1.116.0 --prepend CHANGELOG.md` or `git cliff -u -t 1.116.0 -p CHANGELOG.md`.
 
-3. Update the changelog: `git cliff --unreleased --tag 1.116.0 --prepend CHANGELOG.md` or `git cliff -u -t 1.116.0 -p CHANGELOG.md`.
-
-4. add a link to compare previous with current version to the end of CHANGELOG.md:
+3. add a link to compare previous with current version to the end of CHANGELOG.md:
   `[1.116.0]: https://github.com/deltachat/deltachat-core-rust/compare/v1.115.2...v1.116.0`
 
-5. Update the version by running `scripts/set_core_version.py 1.116.0`.
+4. Update the version by running `scripts/set_core_version.py 1.116.0`.
 
-6. Commit the changes as `chore(release): prepare for 1.116.0`.
+5. Commit the changes as `chore(release): prepare for 1.116.0`.
    Optionally, use a separate branch like `prep-1.116.0` for this commit and open a PR for review.
 
-7. Tag the release: `git tag -a v1.116.0`.
+6. Tag the release: `git tag -a v1.116.0`.
 
-8. Push the release tag: `git push origin v1.116.0`.
+7. Push the release tag: `git push origin v1.116.0`.
 
-9. Create a GitHub release: `gh release create v1.116.0 -n ''`.
+8. Create a GitHub release: `gh release create v1.116.0 -n ''`.

benches/contacts.rs

@@ -1,3 +1,4 @@
+#![recursion_limit = "256"]
 use criterion::{black_box, criterion_group, criterion_main, Criterion};
 use deltachat::contact::Contact;
 use deltachat::context::Context;

benches/create_account.rs

@@ -1,3 +1,4 @@
+#![recursion_limit = "256"]
 use std::path::PathBuf;
 
 use criterion::{black_box, criterion_group, criterion_main, Criterion};

benches/get_chat_msgs.rs

@@ -1,3 +1,4 @@
+#![recursion_limit = "256"]
 use std::path::Path;
 
 use criterion::{black_box, criterion_group, criterion_main, Criterion};

benches/get_chatlist.rs

@@ -1,3 +1,4 @@
+#![recursion_limit = "256"]
 use std::path::Path;
 
 use criterion::{black_box, criterion_group, criterion_main, Criterion};

benches/receive_emails.rs

@@ -1,3 +1,4 @@
+#![recursion_limit = "256"]
 use std::path::PathBuf;
 
 use criterion::{black_box, criterion_group, criterion_main, Criterion};

benches/search_msgs.rs

@@ -1,3 +1,4 @@
+#![recursion_limit = "256"]
 use std::path::Path;
 
 use criterion::{black_box, criterion_group, criterion_main, Criterion};

benches/send_events.rs

@@ -1,3 +1,4 @@
+#![recursion_limit = "256"]
 use criterion::{criterion_group, criterion_main, Criterion};
 
 use deltachat::context::Context;

deltachat-ffi/src/lib.rs

@@ -1,3 +1,4 @@
+#![recursion_limit = "256"]
 #![warn(unused, clippy::all)]
 #![allow(
     non_camel_case_types,
@@ -561,6 +562,7 @@ pub unsafe extern "C" fn dc_event_get_id(event: *mut dc_event_t) -> libc::c_int
         EventType::ConfigSynced { .. } => 2111,
         EventType::WebxdcStatusUpdate { .. } => 2120,
         EventType::WebxdcInstanceDeleted { .. } => 2121,
+        EventType::WebxdcRealtimeData { .. } => 2150,
         EventType::AccountsBackgroundFetchDone => 2200,
         EventType::ChatlistChanged => 2300,
         EventType::ChatlistItemChanged { .. } => 2301,
@@ -616,8 +618,9 @@ pub unsafe extern "C" fn dc_event_get_data1_int(event: *mut dc_event_t) -> libc:
         | EventType::SecurejoinJoinerProgress { contact_id, .. } => {
             contact_id.to_u32() as libc::c_int
         }
-        EventType::WebxdcStatusUpdate { msg_id, .. } => msg_id.to_u32() as libc::c_int,
-        EventType::WebxdcInstanceDeleted { msg_id, .. } => msg_id.to_u32() as libc::c_int,
+        EventType::WebxdcRealtimeData { msg_id, .. }
+        | EventType::WebxdcStatusUpdate { msg_id, .. }
+        | EventType::WebxdcInstanceDeleted { msg_id, .. } => msg_id.to_u32() as libc::c_int,
         EventType::ChatlistItemChanged { chat_id } => {
             chat_id.unwrap_or_default().to_u32() as libc::c_int
         }
@@ -655,6 +658,7 @@ pub unsafe extern "C" fn dc_event_get_data2_int(event: *mut dc_event_t) -> libc:
         | EventType::ConnectivityChanged
         | EventType::WebxdcInstanceDeleted { .. }
         | EventType::IncomingMsgBunch { .. }
+        | EventType::WebxdcRealtimeData { .. }
         | EventType::SelfavatarChanged
         | EventType::AccountsBackgroundFetchDone
         | EventType::ChatlistChanged
@@ -721,6 +725,7 @@ pub unsafe extern "C" fn dc_event_get_data2_str(event: *mut dc_event_t) -> *mut
         | EventType::SelfavatarChanged
         | EventType::WebxdcStatusUpdate { .. }
         | EventType::WebxdcInstanceDeleted { .. }
+        | EventType::WebxdcRealtimeData { .. }
         | EventType::AccountsBackgroundFetchDone
         | EventType::ChatEphemeralTimerModified { .. }
         | EventType::IncomingMsgBunch { .. }

deltachat-jsonrpc/src/api.rs

@@ -18,12 +18,14 @@ use deltachat::constants::DC_MSG_ID_DAYMARKER;
 use deltachat::contact::{may_be_valid_addr, Contact, ContactId, Origin};
 use deltachat::context::get_info;
 use deltachat::ephemeral::Timer;
-use deltachat::imex;
 use deltachat::location;
 use deltachat::message::get_msg_read_receipts;
 use deltachat::message::{
     self, delete_msgs, markseen_msgs, Message, MessageState, MsgId, Viewtype,
 };
+use deltachat::peer_channels::{
+    leave_webxdc_realtime, send_webxdc_realtime_advertisement, send_webxdc_realtime_data,
+};
 use deltachat::provider::get_provider_info;
 use deltachat::qr::{self, Qr};
 use deltachat::qr_code_generator::{generate_backup_qr, get_securejoin_qr_svg};
@@ -32,6 +34,7 @@ use deltachat::securejoin;
 use deltachat::stock_str::StockMessage;
 use deltachat::webxdc::StatusUpdateSerial;
 use deltachat::EventEmitter;
+use deltachat::{imex, info};
 use sanitize_filename::is_sanitized;
 use tokio::fs;
 use tokio::sync::{watch, Mutex, RwLock};
@@ -185,6 +188,16 @@ impl CommandApi {
         self.accounts.write().await.add_account().await
     }
 
+    /// Imports/migrated an existing account from a database path into this account manager.
+    /// Returns the ID of new account.
+    async fn migrate_account(&self, path_to_db: String) -> Result<u32> {
+        self.accounts
+            .write()
+            .await
+            .migrate_account(std::path::PathBuf::from(path_to_db))
+            .await
+    }
+
     async fn remove_account(&self, account_id: u32) -> Result<()> {
         self.accounts
             .write()
@@ -329,6 +342,11 @@ impl CommandApi {
         ctx.get_info().await
     }
 
+    async fn get_blob_dir(&self, account_id: u32) -> Result<Option<String>> {
+        let ctx = self.get_context(account_id).await?;
+        Ok(ctx.get_blobdir().to_str().map(|s| s.to_owned()))
+    }
+
     async fn draft_self_report(&self, account_id: u32) -> Result<u32> {
         let ctx = self.get_context(account_id).await?;
         Ok(ctx.draft_self_report().await?.to_u32())
@@ -1748,6 +1766,37 @@ impl CommandApi {
             .await
     }
 
+    async fn send_webxdc_realtime_data(
+        &self,
+        account_id: u32,
+        instance_msg_id: u32,
+        data: Vec<u8>,
+    ) -> Result<()> {
+        let ctx = self.get_context(account_id).await?;
+        send_webxdc_realtime_data(&ctx, MsgId::new(instance_msg_id), data).await
+    }
+
+    async fn send_webxdc_realtime_advertisement(
+        &self,
+        account_id: u32,
+        instance_msg_id: u32,
+    ) -> Result<()> {
+        let ctx = self.get_context(account_id).await?;
+        let fut = send_webxdc_realtime_advertisement(&ctx, MsgId::new(instance_msg_id)).await?;
+        if let Some(fut) = fut {
+            tokio::spawn(async move {
+                fut.await.ok();
+                info!(ctx, "send_webxdc_realtime_advertisement done")
+            });
+        }
+        Ok(())
+    }
+
+    async fn leave_webxdc_realtime(&self, account_id: u32, instance_message_id: u32) -> Result<()> {
+        let ctx = self.get_context(account_id).await?;
+        leave_webxdc_realtime(&ctx, MsgId::new(instance_message_id)).await
+    }
+
     async fn get_webxdc_status_updates(
         &self,
         account_id: u32,

deltachat-jsonrpc/src/api/types/contact.rs

@@ -89,7 +89,7 @@ impl ContactObject {
     }
 }
 
-#[derive(Serialize, TypeDef, schemars::JsonSchema)]
+#[derive(Clone, Serialize, TypeDef, schemars::JsonSchema)]
 #[serde(rename_all = "camelCase")]
 pub struct VcardContact {
     /// Email address.

deltachat-jsonrpc/src/api/types/events.rs

@@ -240,6 +240,10 @@ pub enum EventType {
         status_update_serial: u32,
     },
 
+    /// Data received over an ephemeral peer channel.
+    #[serde(rename_all = "camelCase")]
+    WebxdcRealtimeData { msg_id: u32, data: Vec<u8> },
+
     /// Inform that a message containing a webxdc instance has been deleted
     #[serde(rename_all = "camelCase")]
     WebxdcInstanceDeleted { msg_id: u32 },
@@ -362,6 +366,10 @@ impl From<CoreEventType> for EventType {
                 msg_id: msg_id.to_u32(),
                 status_update_serial: status_update_serial.to_u32(),
             },
+            CoreEventType::WebxdcRealtimeData { msg_id, data } => WebxdcRealtimeData {
+                msg_id: msg_id.to_u32(),
+                data,
+            },
             CoreEventType::WebxdcInstanceDeleted { msg_id } => WebxdcInstanceDeleted {
                 msg_id: msg_id.to_u32(),
             },

deltachat-jsonrpc/src/api/types/message.rs

@@ -1,3 +1,4 @@
+use crate::api::VcardContact;
 use anyhow::{Context as _, Result};
 use deltachat::chat::Chat;
 use deltachat::chat::ChatItem;
@@ -87,6 +88,8 @@ pub struct MessageObject {
     download_state: DownloadState,
 
     reactions: Option<JSONRPCReactions>,
+
+    vcard_contact: Option<VcardContact>,
 }
 
 #[derive(Serialize, TypeDef, schemars::JsonSchema)]
@@ -173,6 +176,13 @@ impl MessageObject {
             Some(reactions.into())
         };
 
+        let vcard_contacts: Vec<VcardContact> = message
+            .vcard_contacts(context)
+            .await?
+            .into_iter()
+            .map(Into::into)
+            .collect();
+
         Ok(MessageObject {
             id: msg_id.to_u32(),
             chat_id: message.get_chat_id().to_u32(),
@@ -232,6 +242,8 @@ impl MessageObject {
             download_state,
 
             reactions,
+
+            vcard_contact: vcard_contacts.first().cloned(),
         })
     }
 }
@@ -379,6 +391,9 @@ pub enum SystemMessageType {
 
     /// Webxdc info added with `info` set in `send_webxdc_status_update()`.
     WebxdcInfoMessage,
+
+    /// This message contains a users iroh node address.
+    IrohNodeAddr,
 }
 
 impl From<deltachat::mimeparser::SystemMessage> for SystemMessageType {
@@ -401,6 +416,7 @@ impl From<deltachat::mimeparser::SystemMessage> for SystemMessageType {
             SystemMessage::WebxdcStatusUpdate => SystemMessageType::WebxdcStatusUpdate,
             SystemMessage::WebxdcInfoMessage => SystemMessageType::WebxdcInfoMessage,
             SystemMessage::InvalidUnencryptedMail => SystemMessageType::InvalidUnencryptedMail,
+            SystemMessage::IrohNodeAddr => SystemMessageType::IrohNodeAddr,
             SystemMessage::SecurejoinWait => SystemMessageType::SecurejoinWait,
             SystemMessage::SecurejoinWaitTimeout => SystemMessageType::SecurejoinWaitTimeout,
         }

deltachat-jsonrpc/src/lib.rs

@@ -1,3 +1,4 @@
+#![recursion_limit = "256"]
 pub mod api;
 pub use yerpc;
 

deltachat-jsonrpc/src/webserver.rs

@@ -1,3 +1,4 @@
+#![recursion_limit = "256"]
 use std::net::SocketAddr;
 use std::path::PathBuf;
 

deltachat-repl/src/cmdline.rs

@@ -19,6 +19,7 @@ use deltachat::location;
 use deltachat::log::LogExt;
 use deltachat::message::{self, Message, MessageState, MsgId, Viewtype};
 use deltachat::mimeparser::SystemMessage;
+use deltachat::peer_channels::{send_webxdc_realtime_advertisement, send_webxdc_realtime_data};
 use deltachat::peerstate::*;
 use deltachat::qr::*;
 use deltachat::reaction::send_reaction;
@@ -642,6 +643,30 @@ pub async fn cmdline(context: Context, line: &str, chat_id: &mut ChatId) -> Resu
             println!("{cnt} chats");
             println!("{time_needed:?} to create this list");
         }
+        "start-realtime" => {
+            if arg1.is_empty() {
+                bail!("missing msgid");
+            }
+            let msg_id = MsgId::new(arg1.parse()?);
+            let res = send_webxdc_realtime_advertisement(&context, msg_id).await?;
+
+            if let Some(res) = res {
+                println!("waiting for peer channel join");
+                res.await?;
+            }
+            println!("joined peer channel");
+        }
+        "send-realtime" => {
+            if arg1.is_empty() {
+                bail!("missing msgid");
+            }
+            if arg2.is_empty() {
+                bail!("no message");
+            }
+            let msg_id = MsgId::new(arg1.parse()?);
+            send_webxdc_realtime_data(&context, msg_id, arg2.as_bytes().to_vec()).await?;
+            println!("sent realtime message");
+        }
         "chat" => {
             if sel_chat.is_none() && arg1.is_empty() {
                 bail!("Argument [chat-id] is missing.");

deltachat-repl/src/main.rs

@@ -1,3 +1,4 @@
+#![recursion_limit = "256"]
 //! This is a CLI program and a little testing frame.  This file must not be
 //! included when using Delta Chat Core as a library.
 //!

deltachat-rpc-client/src/deltachat_rpc_client/const.py

@@ -62,6 +62,7 @@ class EventType(str, Enum):
     CHATLIST_CHANGED = "ChatlistChanged"
     CHATLIST_ITEM_CHANGED = "ChatlistItemChanged"
     CONFIG_SYNCED = "ConfigSynced"
+    WEBXDC_REALTIME_DATA = "WebxdcRealtimeData"
 
 
 class ChatId(IntEnum):

deltachat-rpc-client/src/deltachat_rpc_client/rpc.py

@@ -177,7 +177,7 @@ class Rpc:
                 account_id = event["contextId"]
                 queue = self.get_queue(account_id)
                 event = event["event"]
-                logging.debug("account_id=%d got an event %s", account_id, event)
+                print("account_id=%d got an event %s" % (account_id, event), file=sys.stderr)
                 queue.put(event)
         except Exception:
             # Log an exception if the event loop dies.

deltachat-rpc-client/tests/test_webxdc_iroh.py

@@ -0,0 +1,123 @@
+#!/usr/bin/env python3
+"""
+Testing webxdc iroh connectivity
+
+If you want to debug iroh at rust-trace/log level set
+
+    RUST_LOG=iroh_net=trace,iroh_gossip=trace
+"""
+
+import pytest
+
+import time
+import os
+import sys
+import logging
+import random
+import itertools
+import sys
+
+from deltachat_rpc_client import DeltaChat, EventType, SpecialContactId
+
+
+@pytest.fixture()
+def path_to_webxdc():
+    return "../test-data/webxdc/chess.xdc"
+
+
+def test_realtime_sequentially(acfactory, path_to_webxdc):
+    """Test two peers trying to establish connection sequentially."""
+    ac1, ac2 = acfactory.get_online_accounts(2)
+    ac1.create_chat(ac2)
+    ac2.create_chat(ac1)
+    acfactory.send_message(from_account=ac1, to_account=ac2, text="ping0")
+    snapshot = ac2.get_message_by_id(ac2.wait_for_incoming_msg_event().msg_id).get_snapshot()
+    assert snapshot.text == "ping0"
+
+    def log(msg):
+        print()
+        print("*" * 80 + "\n" + msg + "\n", file=sys.stderr)
+        print()
+
+    # share a webxdc app between ac1 and ac2
+    ac1_webxdc_msg = acfactory.send_message(from_account=ac1, to_account=ac2, text="play", file=path_to_webxdc)
+    ac2_webxdc_msg = ac2.get_message_by_id(ac2.wait_for_incoming_msg_event().msg_id)
+    snapshot = ac2_webxdc_msg.get_snapshot()
+    assert snapshot.text == "play"
+
+    # send iroh announcements sequentially
+    log("sending ac1 -> ac2 realtime advertisement and additional message")
+    ac1._rpc.send_webxdc_realtime_advertisement(ac1.id, ac1_webxdc_msg.id)
+    acfactory.send_message(from_account=ac1, to_account=ac2, text="ping1")
+
+    log("waiting for incoming message on ac2")
+    snapshot = ac2.get_message_by_id(ac2.wait_for_incoming_msg_event().msg_id).get_snapshot()
+    assert snapshot.text == "ping1"
+
+    log("sending ac2 -> ac1 realtime advertisement and additional message")
+    ac2._rpc.send_webxdc_realtime_advertisement(ac2.id, ac2_webxdc_msg.id)
+    acfactory.send_message(from_account=ac2, to_account=ac1, text="ping2")
+
+    log("waiting for incoming message on ac1")
+    snapshot = ac1.get_message_by_id(ac1.wait_for_incoming_msg_event().msg_id).get_snapshot()
+    assert snapshot.text == "ping2"
+
+    log("sending realtime data ac1 -> ac2")
+    ac1._rpc.send_webxdc_realtime_data(ac1.id, ac1_webxdc_msg.id, [13, 15, 17])
+
+    log("ac2: waiting for realtime data")
+    while 1:
+        event = ac2.wait_for_event()
+        if event.kind == EventType.WEBXDC_REALTIME_DATA:
+            assert event.data == [13, 15, 17]
+            break
+
+
+def test_realtime_simultaneously(acfactory, path_to_webxdc):
+    """Test two peers trying to establish connection simultaneously."""
+    ac1, ac2 = acfactory.get_online_accounts(2)
+    ac1.create_chat(ac2)
+    ac2.create_chat(ac1)
+    acfactory.send_message(from_account=ac1, to_account=ac2, text="ping0")
+    snapshot = ac2.get_message_by_id(ac2.wait_for_incoming_msg_event().msg_id).get_snapshot()
+    assert snapshot.text == "ping0"
+
+    def log(msg):
+        print()
+        print("*" * 80 + "\n" + msg + "\n", file=sys.stderr)
+        print()
+
+    # share a webxdc app between ac1 and ac2
+    ac1_webxdc_msg = acfactory.send_message(from_account=ac1, to_account=ac2, text="play", file=path_to_webxdc)
+    ac2_webxdc_msg = ac2.get_message_by_id(ac2.wait_for_incoming_msg_event().msg_id)
+    snapshot = ac2_webxdc_msg.get_snapshot()
+    assert snapshot.text == "play"
+
+    # send iroh announcements simultaneously
+    log("sending ac1 -> ac2 realtime advertisement and additional message")
+    ac1._rpc.send_webxdc_realtime_advertisement(ac1.id, ac1_webxdc_msg.id)
+    acfactory.send_message(from_account=ac1, to_account=ac2, text="ping1")
+
+    log("sending ac2 -> ac1 realtime advertisement and additional message")
+    ac2._rpc.send_webxdc_realtime_advertisement(ac2.id, ac2_webxdc_msg.id)
+    acfactory.send_message(from_account=ac2, to_account=ac1, text="ping2")
+
+    # Ensure that advertisements have been received.
+
+    log("waiting for incoming message on ac2")
+    snapshot = ac2.get_message_by_id(ac2.wait_for_incoming_msg_event().msg_id).get_snapshot()
+    assert snapshot.text == "ping1"
+
+    log("waiting for incoming message on ac1")
+    snapshot = ac1.get_message_by_id(ac1.wait_for_incoming_msg_event().msg_id).get_snapshot()
+    assert snapshot.text == "ping2"
+
+    log("sending realtime data ac1 -> ac2")
+    ac1._rpc.send_webxdc_realtime_data(ac1.id, ac1_webxdc_msg.id, [13, 15, 17])
+
+    log("ac2: waiting for realtime data")
+    while 1:
+        event = ac2.wait_for_event()
+        if event.kind == EventType.WEBXDC_REALTIME_DATA:
+            assert event.data == [13, 15, 17]
+            break

deltachat-rpc-server/Cargo.toml

@@ -22,6 +22,7 @@ serde = { version = "1.0", features = ["derive"] }
 tokio = { version = "1.37.0", features = ["io-std"] }
 tokio-util = "0.7.9"
 yerpc = { version = "0.5.2", features = ["anyhow_expose", "openrpc"] }
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 
 [features]
 default = ["vendored"]

deltachat-rpc-server/npm-package/.gitignore

@@ -1,2 +1,3 @@
 platform_package
-*.tgz
+*.tgz
+package-lock.json

deltachat-rpc-server/npm-package/index.d.ts

@@ -20,17 +20,20 @@ export function getRPCServerPath(
 
 
 export type DeltaChatOverJsonRpcServer = StdioDeltaChat & {
-    shutdown: () => Promise<void>;
     readonly pathToServerBinary: string;
 };
 
+export interface StartOptions {
+  /** whether to disable outputting stderr to the parent process's stderr */
+  muteStdErr: boolean;
+}
 
 /**
  * 
  * @param directory directory for accounts folder
  * @param options 
  */
-export function startDeltaChat(directory: string, options?: Partial<SearchOptions> ): Promise<DeltaChatOverJsonRpcServer>
+export function startDeltaChat(directory: string, options?: Partial<SearchOptions & StartOptions> ): Promise<DeltaChatOverJsonRpcServer>
 
 
 export namespace FnTypes {

deltachat-rpc-server/npm-package/index.js

@@ -22,10 +22,6 @@ import {
 import package_json from "./package.json" with { type: "json" };
 import { createRequire } from "node:module";
 
-// exports
-// - [ ] a raw starter that has a stdin/out handle thingie like desktop uses
-// - [X] a function that already wraps the stdio handle from above into the deltachat jsonrpc bindings
-
 function findRPCServerInNodeModules() {
   const arch = os.arch();
   const operating_system = process.platform;
@@ -43,6 +39,32 @@ function findRPCServerInNodeModules() {
   }
 }
 
+/**
+ * @returns {Promise<string>}
+ */
+async function getLocationInPath() {
+  const exec = promisify(execFile);
+
+  if (os.platform() === "win32") {
+    const { stdout: executable } = await exec("where", [PATH_EXECUTABLE_NAME], {
+      shell: true,
+    });
+    return executable;
+  }
+
+  try {
+    const { stdout: executable } = await exec(
+      "command",
+      ["-v", PATH_EXECUTABLE_NAME],
+      { shell: true }
+    );
+    return executable;
+  } catch (error) {
+    if (error.code > 0) return "";
+    else throw error;
+  }
+}
+
 /** @type {import("./index").FnTypes.getRPCServerPath} */
 export async function getRPCServerPath(
   options = { skipSearchInPath: false, disableEnvPath: false }
@@ -65,19 +87,18 @@ export async function getRPCServerPath(
 
   // 2. check if it can be found in PATH
   if (!process.env[SKIP_SEARCH_IN_PATH] && !skipSearchInPath) {
-    const exec = promisify(execFile);
-
-    const { stdout: executable } =
-      os.platform() !== "win32"
-        ? await exec("command", ["-v", PATH_EXECUTABLE_NAME])
-        : await exec("where", [PATH_EXECUTABLE_NAME]);
+    const executable = await getLocationInPath();
 
     // by just trying to execute it and then use "command -v deltachat-rpc-server" (unix) or "where deltachat-rpc-server" (windows) to get the path to the executable
     if (executable.length > 1) {
       // test if it is the right version
       try {
         // for some unknown reason it is in stderr and not in stdout
-        const { stderr } = await promisify(execFile)(executable, ["--version"]);
+        const { stderr } = await promisify(execFile)(
+          executable,
+          ["--version"],
+          { shell: true }
+        );
         const version = stderr.slice(0, stderr.indexOf("\n"));
         if (package_json.version !== version) {
           throw new Error(
@@ -109,6 +130,7 @@ export async function startDeltaChat(directory, options) {
       RUST_LOG: process.env.RUST_LOG || "info",
       DC_ACCOUNTS_PATH: directory,
     },
+    stdio: ["pipe", "pipe", options.muteStdErr ? "ignore" : "inherit"],
   });
 
   server.on("error", (err) => {
@@ -123,13 +145,11 @@ export async function startDeltaChat(directory, options) {
     throw new Error("Server quit");
   });
 
-  server.stderr.pipe(process.stderr);
-
   /** @type {import('./index').DeltaChatOverJsonRpcServer} */
   //@ts-expect-error
   const dc = new StdioDeltaChat(server.stdin, server.stdout, true);
 
-  dc.shutdown = async () => {
+  dc.close = () => {
     shouldClose = true;
     if (!server.kill()) {
       console.log("server termination failed");

deltachat-rpc-server/npm-package/scripts/make_local_dev_version.py

@@ -29,3 +29,6 @@ subprocess.run(["python", "scripts/build_platform_package.py", host_target], cap
 
 # run update_optional_dependencies_and_version.js to adjust the package / make it installable locally
 subprocess.run(["node", "scripts/update_optional_dependencies_and_version.js", "--local"], capture_output=False, check=True)
+
+# typescript / npm local package installing/linking needs that this package has it's own node_modules folder
+subprocess.run(["npm", "i"], capture_output=False, check=True)

deltachat-rpc-server/npm-package/scripts/src/convert_platform.py

@@ -2,7 +2,7 @@ def convert_cpu_arch_to_npm_cpu_arch(arch):
     if arch == "x86_64":
         return "x64"
     if arch == "i686":
-        return "i32"
+        return "ia32"
     if arch == "aarch64":
         return "arm64"
     if arch == "armv7" or arch == "arm":

deltachat-rpc-server/npm-package/scripts/update_optional_dependencies_and_version.js

@@ -54,4 +54,10 @@ for (const { folder_name, package_name } of platform_package_names) {
     : version;
 }
 
+if (is_local) {
+  package_json.peerDependencies["@deltachat/jsonrpc-client"] = 'file:../../deltachat-jsonrpc/typescript'
+} else {
+  package_json.peerDependencies["@deltachat/jsonrpc-client"] = "*"
+}
+
 await fs.writeFile("./package.json", JSON.stringify(package_json, null, 4));

deltachat-rpc-server/src/main.rs

@@ -1,3 +1,4 @@
+#![recursion_limit = "256"]
 //! Delta Chat core RPC server.
 //!
 //! It speaks JSON Lines over stdio.
@@ -10,6 +11,7 @@ use deltachat::constants::DC_VERSION_STR;
 use deltachat_jsonrpc::api::{Accounts, CommandApi};
 use futures_lite::stream::StreamExt;
 use tokio::io::{self, AsyncBufReadExt, BufReader};
+use tracing_subscriber::{prelude::*, EnvFilter};
 use yerpc::RpcServer as _;
 
 #[cfg(target_family = "unix")]
@@ -61,6 +63,12 @@ async fn main_impl() -> Result<()> {
 
     env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init();
 
+    tracing_subscriber::registry()
+        .with(tracing_subscriber::fmt::layer().with_writer(std::io::stderr))
+        .with(EnvFilter::builder().from_env_lossy())
+        .try_init()
+        .ok();
+
     let path = std::env::var("DC_ACCOUNTS_PATH").unwrap_or_else(|_| "accounts".to_string());
     log::info!("Starting with accounts directory `{}`.", path);
     let writable = true;

deny.toml

@@ -23,6 +23,9 @@ ignore = [
 # when upgrading.
 # Please keep this list alphabetically sorted.
 skip = [
+     { name = "asn1-rs-derive", version = "0.4.0" },
+     { name = "asn1-rs-impl", version = "0.1.0" },
+     { name = "asn1-rs", version = "0.5.2" },
      { name = "async-channel", version = "1.9.0" },
      { name = "base16ct", version = "0.1.1" },
      { name = "base64", version = "<0.21" },
@@ -34,20 +37,38 @@ skip = [
      { name = "darling_core", version = "<0.14" },
      { name = "darling_macro", version = "<0.14" },
      { name = "darling", version = "<0.14" },
+     { name = "der_derive", version = "0.6.1" },
+     { name = "derive_more", version = "0.99.17" },
+     { name = "der-parser", version = "8.2.0" },
      { name = "der", version = "0.6.1" },
      { name = "digest", version = "<0.10" },
+     { name = "dlopen2", version = "0.4.1" },
      { name = "ed25519-dalek", version = "1.0.1" },
      { name = "ed25519", version = "1.5.3" },
      { name = "env_logger", version = "0.10.2" },
      { name = "event-listener", version = "2.5.3" },
+     { name = "event-listener", version = "4.0.3" },
+     { name = "fastrand", version = "1.9.0" },
+     { name = "futures-lite", version = "1.13.0" },
      { name = "getrandom", version = "<0.2" },
+     { name = "http-body", version = "0.4.6" },
+     { name = "http", version = "0.2.12" },
+     { name = "hyper", version = "0.14.28" },
      { name = "idna", version = "0.4.0" },
+     { name = "netlink-packet-core", version = "0.5.0" },
+     { name = "netlink-packet-route", version = "0.15.0" },
+     { name = "nix", version = "0.26.4" },
+     { name = "oid-registry", version = "0.6.1" },
      { name = "pem-rfc7468", version = "0.6.0" },
+     { name = "pem", version = "1.1.1" },
      { name = "pkcs8", version = "0.9.0" },
+     { name = "proc-macro-error-attr", version = "0.4.12" },
+     { name = "proc-macro-error", version = "0.4.12" },
      { name = "quick-error", version = "<2.0" },
      { name = "rand_chacha", version = "<0.3" },
      { name = "rand_core", version = "<0.6" },
      { name = "rand", version = "<0.8" },
+     { name = "rcgen", version = "<0.12.1" },
      { name = "redox_syscall", version = "0.3.5" },
      { name = "regex-automata", version = "0.1.10" },
      { name = "regex-syntax", version = "0.6.29" },
@@ -57,23 +78,31 @@ skip = [
      { name = "signature", version = "1.6.4" },
      { name = "spin", version = "<0.9.6" },
      { name = "spki", version = "0.6.0" },
+     { name = "ssh-encoding", version = "0.1.0" },
+     { name = "ssh-key", version = "0.5.1" },
      { name = "sync_wrapper", version = "0.1.2" },
+     { name = "synstructure", version = "0.12.6" },
      { name = "syn", version = "1.0.109" },
+     { name = "system-configuration-sys", version = "0.5.0" },
+     { name = "system-configuration", version = "0.5.1" },
      { name = "time", version = "<0.3" },
      { name = "toml_edit", version = "0.21.1" },
      { name = "untrusted", version = "0.7.1" },
      { name = "wasi", version = "<0.11" },
      { name = "windows_aarch64_gnullvm", version = "<0.52" },
      { name = "windows_aarch64_msvc", version = "<0.52" },
+     { name = "windows-core", version = "<0.54.0" },
      { name = "windows_i686_gnu", version = "<0.52" },
      { name = "windows_i686_msvc", version = "<0.52" },
      { name = "windows-sys", version = "<0.52" },
      { name = "windows-targets", version = "<0.52" },
      { name = "windows", version = "0.32.0" },
+     { name = "windows", version = "<0.54.0" },
      { name = "windows_x86_64_gnullvm", version = "<0.52" },
      { name = "windows_x86_64_gnu", version = "<0.52" },
      { name = "windows_x86_64_msvc", version = "<0.52" },
      { name = "winnow", version = "0.5.40" },
+     { name = "x509-parser", version = "<0.16.0" },
 ]
 
 

flake.lock

@@ -48,11 +48,11 @@
         "rust-analyzer-src": "rust-analyzer-src"
       },
       "locked": {
-        "lastModified": 1713421495,
-        "narHash": "sha256-5vVF9W1tJT+WdfpWAEG76KywktKDAW/71mVmNHEHjac=",
+        "lastModified": 1714112748,
+        "narHash": "sha256-jq6Cpf/pQH85p+uTwPPrGG8Ky/zUOTwMJ7mcqc5M4So=",
         "owner": "nix-community",
         "repo": "fenix",
-        "rev": "fd47b1f9404fae02a4f38bd9f4b12bad7833c96b",
+        "rev": "3ae4b908a795b6a3824d401a0702e11a7157d7e1",
         "type": "github"
       },
       "original": {
@@ -166,11 +166,11 @@
     },
     "nixpkgs_2": {
       "locked": {
-        "lastModified": 1713248628,
-        "narHash": "sha256-NLznXB5AOnniUtZsyy/aPWOk8ussTuePp2acb9U+ISA=",
+        "lastModified": 1713895582,
+        "narHash": "sha256-cfh1hi+6muQMbi9acOlju3V1gl8BEaZBXBR9jQfQi4U=",
         "owner": "nixos",
         "repo": "nixpkgs",
-        "rev": "5672bc9dbf9d88246ddab5ac454e82318d094bb8",
+        "rev": "572af610f6151fd41c212f897c71f7056e3fb518",
         "type": "github"
       },
       "original": {
@@ -182,12 +182,11 @@
     },
     "nixpkgs_3": {
       "locked": {
-        "lastModified": 1713562564,
-        "narHash": "sha256-NQpYhgoy0M89g9whRixSwsHb8RFIbwlxeYiVSDwSXJg=",
-        "owner": "NixOS",
-        "repo": "nixpkgs",
-        "rev": "92d295f588631b0db2da509f381b4fb1e74173c5",
-        "type": "github"
+        "lastModified": 1711668574,
+        "narHash": "sha256-u1dfs0ASQIEr1icTVrsKwg2xToIpn7ZXxW3RHfHxshg=",
+        "path": "/nix/store/9fpv0kjq9a80isa1wkkvrdqsh9dpcn05-source",
+        "rev": "219951b495fc2eac67b1456824cc1ec1fd2ee659",
+        "type": "path"
       },
       "original": {
         "id": "nixpkgs",
@@ -196,11 +195,11 @@
     },
     "nixpkgs_4": {
       "locked": {
-        "lastModified": 1713537308,
-        "narHash": "sha256-XtTSSIB2DA6tOv+l0FhvfDMiyCmhoRbNB+0SeInZkbk=",
+        "lastModified": 1714076141,
+        "narHash": "sha256-Drmja/f5MRHZCskS6mvzFqxEaZMeciScCTFxWVLqWEY=",
         "owner": "nixos",
         "repo": "nixpkgs",
-        "rev": "5c24cf2f0a12ad855f444c30b2421d044120c66f",
+        "rev": "7bb2ccd8cdc44c91edba16c48d2c8f331fb3d856",
         "type": "github"
       },
       "original": {
@@ -223,11 +222,11 @@
     "rust-analyzer-src": {
       "flake": false,
       "locked": {
-        "lastModified": 1713373173,
-        "narHash": "sha256-octd9BFY9G/Gbr4KfwK4itZp4Lx+qvJeRRcYnN+dEH8=",
+        "lastModified": 1714031783,
+        "narHash": "sha256-xS/niQsq1CQPOe4M4jvVPO2cnXS/EIeRG5gIopUbk+Q=",
         "owner": "rust-lang",
         "repo": "rust-analyzer",
-        "rev": "46702ffc1a02a2ac153f1d1ce619ec917af8f3a6",
+        "rev": "56bee2ddafa6177b19c631eedc88d43366553223",
         "type": "github"
       },
       "original": {

flake.nix

@@ -525,15 +525,25 @@
               };
           };
 
-        devShells.default = pkgs.mkShell {
+        devShells.default = let 
+          pkgs = import nixpkgs {
+            system = system;
+            overlays = [ fenix.overlays.default ];
+          };
+          in pkgs.mkShell {
+
           buildInputs = with pkgs; [
-            cargo
-            clippy
-            rustc
-            rustfmt
-            rust-analyzer
+            (fenix.packages.${system}.complete.withComponents [
+              "cargo"
+              "clippy"
+              "rust-src"
+              "rustc"
+              "rustfmt"
+            ])
             cargo-deny
+            rust-analyzer-nightly
             perl # needed to build vendored OpenSSL
+            git-cliff
           ];
         };
       }

src/config.rs

@@ -9,7 +9,7 @@ use base64::Engine as _;
 use deltachat_contact_tools::addr_cmp;
 use serde::{Deserialize, Serialize};
 use strum::{EnumProperty, IntoEnumIterator};
-use strum_macros::{AsRefStr, Display, EnumIter, EnumProperty, EnumString};
+use strum_macros::{AsRefStr, Display, EnumIter, EnumString};
 use tokio::fs;
 
 use crate::blob::BlobObject;
@@ -362,6 +362,9 @@ pub enum Config {
 
     /// MsgId of webxdc map integration.
     WebxdcIntegration,
+
+    /// Iroh secret key.
+    IrohSecretKey,
 }
 
 impl Config {

src/context.rs

@@ -12,7 +12,7 @@ use anyhow::{bail, ensure, Context as _, Result};
 use async_channel::{self as channel, Receiver, Sender};
 use pgp::SignedPublicKey;
 use ratelimit::Ratelimit;
-use tokio::sync::{Mutex, Notify, RwLock};
+use tokio::sync::{Mutex, Notify, OnceCell, RwLock};
 
 use crate::aheader::EncryptPreference;
 use crate::chat::{get_chat_cnt, ChatId, ProtectionStatus};
@@ -30,6 +30,7 @@ use crate::key::{load_self_public_key, load_self_secret_key, DcKey as _};
 use crate::login_param::LoginParam;
 use crate::message::{self, Message, MessageState, MsgId, Viewtype};
 use crate::param::{Param, Params};
+use crate::peer_channels::Iroh;
 use crate::peerstate::Peerstate;
 use crate::push::PushSubscriber;
 use crate::quota::QuotaInfo;
@@ -288,6 +289,9 @@ pub struct InnerContext {
 
     /// True if account has subscribed to push notifications via IMAP.
     pub(crate) push_subscribed: AtomicBool,
+
+    /// Iroh for realtime peer channels.
+    pub(crate) iroh: OnceCell<Iroh>,
 }
 
 /// The state of ongoing process.
@@ -335,7 +339,6 @@ impl Context {
     ) -> Result<Context> {
         let context =
             Self::new_closed(dbfile, id, events, stock_strings, Default::default()).await?;
-
         // Open the database if is not encrypted.
         if context.check_passphrase("".to_string()).await? {
             context.sql.open(&context, "".to_string()).await?;
@@ -445,6 +448,7 @@ impl Context {
             debug_logging: std::sync::RwLock::new(None),
             push_subscriber,
             push_subscribed: AtomicBool::new(false),
+            iroh: OnceCell::new(),
         };
 
         let ctx = Context {
@@ -482,6 +486,9 @@ impl Context {
 
     /// Indicate that the network likely has come back.
     pub async fn maybe_network(&self) {
+        if let Some(iroh) = self.iroh.get() {
+            iroh.network_change().await;
+        }
         self.scheduler.maybe_network().await;
     }
 
@@ -1368,6 +1375,43 @@ pub fn get_version_str() -> &'static str {
     &DC_VERSION_STR
 }
 
+#[derive(Default, Debug)]
+struct CollectVisitor(HashMap<String, String>);
+
+impl tracing::field::Visit for CollectVisitor {
+    fn record_f64(&mut self, field: &tracing::field::Field, value: f64) {
+        self.0.insert(field.to_string(), value.to_string());
+    }
+
+    fn record_i64(&mut self, field: &tracing::field::Field, value: i64) {
+        self.0.insert(field.to_string(), value.to_string());
+    }
+
+    fn record_u64(&mut self, field: &tracing::field::Field, value: u64) {
+        self.0.insert(field.to_string(), value.to_string());
+    }
+
+    fn record_bool(&mut self, field: &tracing::field::Field, value: bool) {
+        self.0.insert(field.to_string(), value.to_string());
+    }
+
+    fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
+        self.0.insert(field.to_string(), value.to_string());
+    }
+
+    fn record_error(
+        &mut self,
+        field: &tracing::field::Field,
+        value: &(dyn std::error::Error + 'static),
+    ) {
+        self.0.insert(field.to_string(), value.to_string());
+    }
+
+    fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
+        self.0.insert(field.to_string(), format!("{:?}", value));
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use anyhow::Context as _;
@@ -1648,6 +1692,7 @@ mod tests {
             "socks5_password",
             "key_id",
             "webxdc_integration",
+            "iroh_secret_key",
         ];
         let t = TestContext::new().await;
         let info = t.get_info().await.unwrap();

src/events/payload.rs

@@ -279,6 +279,15 @@ pub enum EventType {
         status_update_serial: StatusUpdateSerial,
     },
 
+    /// Data received over an ephemeral peer channel.
+    WebxdcRealtimeData {
+        /// Message ID.
+        msg_id: MsgId,
+
+        /// Realtime data.
+        data: Vec<u8>,
+    },
+
     /// Inform that a message containing a webxdc instance has been deleted.
     WebxdcInstanceDeleted {
         /// ID of the deleted message.

src/headerdef.rs

@@ -93,6 +93,12 @@ pub enum HeaderDef {
     /// See <https://datatracker.ietf.org/doc/html/rfc8601>
     AuthenticationResults,
 
+    /// Node address from iroh where direct addresses have been removed.
+    IrohNodeAddr,
+
+    /// Advertised gossip topic for one webxdc.
+    IrohGossipTopic,
+
     #[cfg(test)]
     TestHeader,
 }

src/imap.rs

@@ -22,6 +22,7 @@ use futures_lite::FutureExt;
 use num_traits::FromPrimitive;
 use rand::Rng;
 use ratelimit::Ratelimit;
+use url::Url;
 
 use crate::chat::{self, ChatId, ChatIdBlocked};
 use crate::chatlist_events;
@@ -111,6 +112,8 @@ pub(crate) struct ServerMetadata {
     /// IMAP METADATA `/shared/admin` as defined in
     /// <https://www.rfc-editor.org/rfc/rfc5464#section-6.2.2>.
     pub admin: Option<String>,
+
+    pub iroh_relay: Option<Url>,
 }
 
 impl async_imap::Authenticator for OAuth2 {
@@ -1449,11 +1452,16 @@ impl Session {
 
         let mut comment = None;
         let mut admin = None;
+        let mut iroh_relay = None;
 
         let mailbox = "";
         let options = "";
         let metadata = self
-            .get_metadata(mailbox, options, "(/shared/comment /shared/admin)")
+            .get_metadata(
+                mailbox,
+                options,
+                "(/shared/comment /shared/admin /shared/vendor/deltachat/irohrelay)",
+            )
             .await?;
         for m in metadata {
             match m.entry.as_ref() {
@@ -1463,10 +1471,24 @@ impl Session {
                 "/shared/admin" => {
                     admin = m.value;
                 }
+                "/shared/vendor/deltachat/irohrelay" => {
+                    if let Some(url) = m.value.as_deref().and_then(|s| Url::parse(s).ok()) {
+                        iroh_relay = Some(url);
+                    } else {
+                        warn!(
+                            context,
+                            "Got invalid URL from iroh relay metadata: {:?}.", m.value
+                        );
+                    }
+                }
                 _ => {}
             }
         }
-        *lock = Some(ServerMetadata { comment, admin });
+        *lock = Some(ServerMetadata {
+            comment,
+            admin,
+            iroh_relay,
+        });
         Ok(())
     }
 

src/imex/transfer.rs

@@ -38,6 +38,7 @@ use iroh::progress::ProgressEmitter;
 use iroh::protocol::AuthToken;
 use iroh::provider::{DataSource, Event, Provider, Ticket};
 use iroh::Hash;
+use iroh_old as iroh;
 use tokio::fs::{self, File};
 use tokio::io::{self, AsyncWriteExt, BufWriter};
 use tokio::sync::broadcast::error::RecvError;

src/lib.rs

@@ -106,6 +106,7 @@ pub mod receive_imf;
 pub mod tools;
 
 pub mod accounts;
+pub mod peer_channels;
 pub mod reaction;
 
 /// If set IMAP/incoming and SMTP/outgoing MIME messages will be printed.

src/message.rs

@@ -4,6 +4,7 @@ use std::collections::BTreeSet;
 use std::path::{Path, PathBuf};
 
 use anyhow::{ensure, format_err, Context as _, Result};
+use deltachat_contact_tools::{parse_vcard, VcardContact};
 use deltachat_derive::{FromSql, ToSql};
 use serde::{Deserialize, Serialize};
 use tokio::{fs, io};
@@ -607,6 +608,20 @@ impl Message {
         self.param.get_path(Param::File, context).unwrap_or(None)
     }
 
+    /// Returns vector of vcards if the file has a vCard attachment.
+    pub async fn vcard_contacts(&self, context: &Context) -> Result<Vec<VcardContact>> {
+        if self.viewtype != Viewtype::Vcard {
+            return Ok(Vec::new());
+        }
+
+        let path = self
+            .get_file(context)
+            .context("vCard message does not have an attachment")?;
+        let bytes = tokio::fs::read(path).await?;
+        let vcard_contents = std::str::from_utf8(&bytes).context("vCard is not a valid UTF-8")?;
+        Ok(parse_vcard(vcard_contents))
+    }
+
     /// Save file copy at the user-provided path.
     pub async fn save_file(&self, context: &Context, path: &Path) -> Result<()> {
         let path_src = self.get_file(context).context("No file")?;

src/mimefactory.rs

@@ -17,11 +17,12 @@ use crate::contact::Contact;
 use crate::context::Context;
 use crate::e2ee::EncryptHelper;
 use crate::ephemeral::Timer as EphemeralTimer;
+use crate::headerdef::HeaderDef;
 use crate::html::new_html_mimepart;
-use crate::location;
 use crate::message::{self, Message, MsgId, Viewtype};
 use crate::mimeparser::SystemMessage;
 use crate::param::Param;
+use crate::peer_channels::create_iroh_header;
 use crate::peerstate::Peerstate;
 use crate::simplify::escape_message_footer_marks;
 use crate::stock_str;
@@ -29,6 +30,7 @@ use crate::tools::IsNoneOrEmpty;
 use crate::tools::{
     create_outgoing_rfc724_mid, create_smeared_timestamp, remove_subject_prefix, time,
 };
+use crate::{location, peer_channels};
 
 // attachments of 25 mb brutto should work on the majority of providers
 // (brutto examples: web.de=50, 1&1=40, t-online.de=32, gmail=25, posteo=50, yahoo=25, all-inkl=100).
@@ -1148,6 +1150,18 @@ impl<'a> MimeFactory<'a> {
                     "protection-disabled".to_string(),
                 ));
             }
+            SystemMessage::IrohNodeAddr => {
+                headers.protected.push(Header::new(
+                    HeaderDef::IrohNodeAddr.get_headername().to_string(),
+                    serde_json::to_string(
+                        &context
+                            .get_or_try_init_peer_channel()
+                            .await?
+                            .get_node_addr()
+                            .await?,
+                    )?,
+                ));
+            }
             _ => {}
         }
 
@@ -1314,6 +1328,10 @@ impl<'a> MimeFactory<'a> {
             let json = self.msg.param.get(Param::Arg).unwrap_or_default();
             parts.push(context.build_status_update_part(json));
         } else if self.msg.viewtype == Viewtype::Webxdc {
+            let topic = peer_channels::create_random_topic();
+            headers
+                .protected
+                .push(create_iroh_header(context, topic, self.msg.id).await?);
             if let Some(json) = context
                 .render_webxdc_status_update_object(self.msg.id, None)
                 .await?

src/mimeparser.rs

@@ -199,6 +199,9 @@ pub enum SystemMessage {
 
     /// Webxdc info added with `info` set in `send_webxdc_status_update()`.
     WebxdcInfoMessage = 32,
+
+    /// This message contains a users iroh node address.
+    IrohNodeAddr = 40,
 }
 
 const MIME_AC_SETUP_FILE: &str = "application/autocrypt-setup";

src/peer_channels.rs

@@ -0,0 +1,801 @@
+//! Peer channels for realtime communication in webxdcs.
+//!
+//! We use Iroh as an ephemeral peer channels provider to create direct communication
+//! channels between webxdcs. See [here](https://webxdc.org/docs/spec/joinRealtimeChannel.html) for the webxdc specs.
+//!
+//! Ephemeral channels should be established lazily, to avoid bootstrapping p2p connectivity
+//! when it's not required. Only when a webxdc subscribes to realtime data or when a reatlime message is sent,
+//! the p2p machinery should be started.
+//!
+//! Adding peer channels to webxdc needs upfront negotation of a topic and sharing of public keys so that
+//! nodes can connect to each other. The explicit approach is as follows:
+//!
+//! 1. We introduce a new [GossipTopic](crate::headerdef::HeaderDef::IrohGossipTopic) message header with a random 32-byte TopicId,
+//!    securely generated on the initial webxdc sender's device. This message header is encrypted
+//!    and sent in the same message as the webxdc application.
+//! 2. Whenever `joinRealtimeChannel().setListener()` or `joinRealtimeChannel().send()` is called by the webxdc application,
+//!    we start a routine to establish p2p connectivity and join the gossip swarm with Iroh.
+//! 3. The first step of this routine is to introduce yourself with a regular message containing the `IrohPublicKey`.
+//!    This message contains the users relay-server and public key.
+//!    Direct IP address is not included as this information can be persisted by email providers.
+//! 4. After the announcement, the sending peer joins the gossip swarm with an empty list of peer IDs (as they don't know anyone yet).
+//! 5. Upon receiving an announcement message, other peers store the sender's [NodeAddr] in the database
+//!    (scoped per WebXDC app instance/message-id). The other peers can then join the gossip with `joinRealtimeChannel().setListener()`
+//!    and `joinRealtimeChannel().send()` just like the other peers.
+
+use anyhow::{anyhow, Context as _, Result};
+use email::Header;
+use futures_lite::StreamExt;
+use iroh_gossip::net::{Gossip, JoinTopicFut, GOSSIP_ALPN};
+use iroh_gossip::proto::{Event as IrohEvent, TopicId};
+use iroh_net::relay::{RelayMap, RelayUrl};
+use iroh_net::{key::SecretKey, relay::RelayMode, MagicEndpoint};
+use iroh_net::{NodeAddr, NodeId};
+use std::collections::{BTreeSet, HashMap};
+use std::env;
+use tokio::sync::RwLock;
+use tokio::task::JoinHandle;
+use url::Url;
+
+use crate::chat::send_msg;
+use crate::context::Context;
+use crate::headerdef::HeaderDef;
+use crate::message::{Message, MsgId, Viewtype};
+use crate::mimeparser::SystemMessage;
+use crate::EventType;
+
+/// The length of an ed25519 `PublicKey`, in bytes.
+const PUBLIC_KEY_LENGTH: usize = 32;
+const PUBLIC_KEY_STUB: &[u8] = "static_string".as_bytes();
+
+/// Store iroh peer channels for the context.
+#[derive(Debug)]
+pub struct Iroh {
+    /// [MagicEndpoint] needed for iroh peer channels.
+    pub(crate) endpoint: MagicEndpoint,
+
+    /// [Gossip] needed for iroh peer channels.
+    pub(crate) gossip: Gossip,
+
+    /// Topics for which an advertisement has already been sent.
+    pub(crate) iroh_channels: RwLock<HashMap<TopicId, ChannelState>>,
+
+    /// Currently used Iroh secret key
+    pub(crate) secret_key: SecretKey,
+}
+
+impl Iroh {
+    /// Notify the endpoint that the network has changed.
+    pub(crate) async fn network_change(&self) {
+        self.endpoint.network_change().await
+    }
+
+    /// Join a topic and create the subscriber loop for it.
+    ///
+    /// If there is no gossip, create it.
+    ///
+    /// The returned future resolves when the swarm becomes operational.
+    async fn join_and_subscribe_gossip(
+        &self,
+        ctx: &Context,
+        msg_id: MsgId,
+    ) -> Result<Option<JoinTopicFut>> {
+        let topic = get_iroh_topic_for_msg(ctx, msg_id).await?;
+        let seq = if let Some(channel_state) = self.iroh_channels.read().await.get(&topic) {
+            if channel_state.subscribe_loop.is_some() {
+                return Ok(None);
+            }
+            channel_state.seq_number
+        } else {
+            0
+        };
+
+        let peers = get_iroh_gossip_peers(ctx, msg_id).await?;
+        info!(
+            ctx,
+            "IROH_REALTIME: Joining gossip with peers: {:?}",
+            peers.iter().map(|p| p.node_id).collect::<Vec<_>>()
+        );
+
+        // Connect to all peers
+        for peer in &peers {
+            self.endpoint.add_node_addr(peer.clone())?;
+        }
+
+        let ctx = ctx.clone();
+        let gossip = self.gossip.clone();
+        let subscribe_loop = tokio::spawn(async move {
+            if let Err(e) = subscribe_loop(&ctx, gossip, topic, msg_id).await {
+                warn!(ctx, "subscribe_loop failed: {e}")
+            }
+        });
+
+        let connect_future = self
+            .gossip
+            .join(topic, peers.into_iter().map(|addr| addr.node_id).collect())
+            .await?;
+
+        self.iroh_channels
+            .write()
+            .await
+            .insert(topic, ChannelState::new(seq, subscribe_loop));
+
+        Ok(Some(connect_future))
+    }
+
+    /// Add gossip peers to realtime channel if it is already active.
+    pub async fn maybe_add_gossip_peers(&self, topic: TopicId, peers: Vec<NodeAddr>) -> Result<()> {
+        if let Some(state) = self.iroh_channels.read().await.get(&topic) {
+            if state.subscribe_loop.is_some() {
+                for peer in &peers {
+                    self.endpoint.add_node_addr(peer.clone())?;
+                }
+                self.gossip
+                    .join(topic, peers.into_iter().map(|peer| peer.node_id).collect())
+                    .await?;
+            }
+        }
+        Ok(())
+    }
+
+    /// Send realtime data to the gossip swarm.
+    pub async fn send_webxdc_realtime_data(
+        &self,
+        ctx: &Context,
+        msg_id: MsgId,
+        mut data: Vec<u8>,
+    ) -> Result<()> {
+        let topic = get_iroh_topic_for_msg(ctx, msg_id).await?;
+        self.join_and_subscribe_gossip(ctx, msg_id).await?;
+
+        let seq_num = self.get_and_incr(&topic).await;
+        data.extend(seq_num.to_le_bytes());
+        data.extend(self.secret_key.public().as_bytes());
+
+        self.gossip.broadcast(topic, data.into()).await?;
+
+        if env::var("REALTIME_DEBUG").is_ok() {
+            info!(ctx, "Sent realtime data");
+        }
+
+        Ok(())
+    }
+
+    async fn get_and_incr(&self, topic: &TopicId) -> i32 {
+        let mut seq = 0;
+        if let Some(state) = self.iroh_channels.write().await.get_mut(topic) {
+            seq = state.seq_number;
+            state.seq_number = state.seq_number.wrapping_add(1)
+        }
+        seq
+    }
+
+    /// Get the iroh [NodeAddr] without direct IP addresses.
+    pub(crate) async fn get_node_addr(&self) -> Result<NodeAddr> {
+        let mut addr = self.endpoint.my_addr().await?;
+        addr.info.direct_addresses = BTreeSet::new();
+        Ok(addr)
+    }
+
+    /// Leave the realtime channel for a given topic.
+    pub(crate) async fn leave_realtime(&self, topic: TopicId) -> Result<()> {
+        if let Some(channel) = &mut self.iroh_channels.write().await.get_mut(&topic) {
+            if let Some(subscribe_loop) = channel.subscribe_loop.take() {
+                subscribe_loop.abort();
+            }
+        }
+        self.gossip.quit(topic).await?;
+        Ok(())
+    }
+}
+
+/// Single gossip channel state.
+#[derive(Debug)]
+pub(crate) struct ChannelState {
+    /// Sequence number for the gossip channel.
+    seq_number: i32,
+    /// The subscribe loop handle.
+    subscribe_loop: Option<JoinHandle<()>>,
+}
+
+impl ChannelState {
+    fn new(seq_number: i32, subscribe_loop: JoinHandle<()>) -> Self {
+        Self {
+            seq_number,
+            subscribe_loop: Some(subscribe_loop),
+        }
+    }
+}
+
+impl Context {
+    /// Create magic endpoint and gossip.
+    async fn init_peer_channels(&self) -> Result<Iroh> {
+        let secret_key: SecretKey = SecretKey::generate();
+
+        let relay_mode = if let Some(relay_url) = self
+            .metadata
+            .read()
+            .await
+            .as_ref()
+            .and_then(|conf| conf.iroh_relay.clone())
+        {
+            RelayMode::Custom(RelayMap::from_url(RelayUrl::from(relay_url)))
+        } else {
+            // FIXME: this should be RelayMode::Disabled instead.
+            // Currently using default relays because otherwise Rust tests fail.
+            RelayMode::Default
+        };
+
+        let endpoint = MagicEndpoint::builder()
+            .secret_key(secret_key.clone())
+            .alpns(vec![GOSSIP_ALPN.to_vec()])
+            .relay_mode(relay_mode)
+            .bind(0)
+            .await?;
+
+        // create gossip
+        let my_addr = endpoint.my_addr().await?;
+        let gossip = Gossip::from_endpoint(endpoint.clone(), Default::default(), &my_addr.info);
+
+        // spawn endpoint loop that forwards incoming connections to the gossiper
+        let context = self.clone();
+
+        // Shuts down on deltachat shutdown
+        tokio::spawn(endpoint_loop(context, endpoint.clone(), gossip.clone()));
+        let endp = endpoint.clone();
+        let gsp = gossip.clone();
+        tokio::spawn(async move {
+            let mut stream = endp.local_endpoints();
+            while let Some(endpoints) = stream.next().await {
+                gsp.update_endpoints(&endpoints)?;
+            }
+            anyhow::Ok(())
+        });
+        Ok(Iroh {
+            endpoint,
+            gossip,
+            iroh_channels: RwLock::new(HashMap::new()),
+            secret_key,
+        })
+    }
+
+    /// Get or initialize the iroh peer channel.
+    pub async fn get_or_try_init_peer_channel(&self) -> Result<&Iroh> {
+        let ctx = self.clone();
+        self.iroh
+            .get_or_try_init(|| async { ctx.init_peer_channels().await })
+            .await
+    }
+}
+
+/// Cache a peers [NodeId] for one topic.
+pub(crate) async fn iroh_add_peer_for_topic(
+    ctx: &Context,
+    msg_id: MsgId,
+    topic: TopicId,
+    peer: NodeId,
+    relay_server: Option<&str>,
+) -> Result<()> {
+    ctx.sql
+        .execute(
+            "INSERT OR REPLACE INTO iroh_gossip_peers (msg_id, public_key, topic, relay_server) VALUES (?, ?, ?, ?)",
+            (msg_id, peer.as_bytes(), topic.as_bytes(), relay_server),
+        )
+        .await?;
+    Ok(())
+}
+
+/// Insert topicId into the database so that we can use it to retrieve the topic.
+pub(crate) async fn insert_topic_stub(ctx: &Context, msg_id: MsgId, topic: TopicId) -> Result<()> {
+    ctx.sql
+        .execute(
+            "INSERT OR REPLACE INTO iroh_gossip_peers (msg_id, public_key, topic, relay_server) VALUES (?, ?, ?, ?)",
+            (msg_id, PUBLIC_KEY_STUB, topic.as_bytes(), Option::<&str>::None),
+        )
+        .await?;
+    Ok(())
+}
+
+/// Get a list of [NodeAddr]s for one webxdc.
+async fn get_iroh_gossip_peers(ctx: &Context, msg_id: MsgId) -> Result<Vec<NodeAddr>> {
+    ctx.sql
+        .query_map(
+            "SELECT public_key, relay_server FROM iroh_gossip_peers WHERE msg_id = ? AND public_key != ?",
+            (msg_id, PUBLIC_KEY_STUB),
+            |row| {
+                let key:  Vec<u8> = row.get(0)?;
+                let server: Option<String> = row.get(1)?;
+                Ok((key, server))
+            },
+            |g| {
+                g.map(|data| {
+                    let (key, server) = data?;
+                    let server = server.map(|data| Ok::<_, url::ParseError>(RelayUrl::from(Url::parse(&data)?))).transpose()?;
+                    let id = NodeId::from_bytes(&key.try_into()
+                    .map_err(|_| anyhow!("Can't convert sql data to [u8; 32]"))?)?;
+                    Ok::<_, anyhow::Error>(NodeAddr::from_parts(
+                        id, server, vec![]
+                    ))
+                })
+                .collect::<std::result::Result<Vec<_>, _>>()
+                .map_err(Into::into)
+            },
+        )
+        .await
+}
+
+/// Get the topic for a given [MsgId].
+pub(crate) async fn get_iroh_topic_for_msg(ctx: &Context, msg_id: MsgId) -> Result<TopicId> {
+    let bytes: Vec<u8> = ctx
+        .sql
+        .query_get_value(
+            "SELECT topic FROM iroh_gossip_peers WHERE msg_id = ? LIMIT 1",
+            (msg_id,),
+        )
+        .await?
+        .context("couldn't restore topic from db")?;
+    Ok(TopicId::from_bytes(bytes.try_into().unwrap()))
+}
+
+/// Send a gossip advertisement to the chat that [MsgId] belongs to.
+/// This method should be called from the frontend when `joinRealtimeChannel` is called.
+pub async fn send_webxdc_realtime_advertisement(
+    ctx: &Context,
+    msg_id: MsgId,
+) -> Result<Option<JoinTopicFut>> {
+    let iroh = ctx.get_or_try_init_peer_channel().await?;
+    let conn = iroh.join_and_subscribe_gossip(ctx, msg_id).await?;
+
+    let webxdc = Message::load_from_db(ctx, msg_id).await?;
+    let mut msg = Message::new(Viewtype::Text);
+    msg.hidden = true;
+    msg.param.set_cmd(SystemMessage::IrohNodeAddr);
+    msg.in_reply_to = Some(webxdc.rfc724_mid.clone());
+    send_msg(ctx, webxdc.chat_id, &mut msg).await?;
+    info!(ctx, "IROH_REALTIME: Sent realtime advertisement");
+    Ok(conn)
+}
+
+/// Send realtime data to the gossip swarm.
+pub async fn send_webxdc_realtime_data(ctx: &Context, msg_id: MsgId, data: Vec<u8>) -> Result<()> {
+    let iroh = ctx.get_or_try_init_peer_channel().await?;
+    iroh.send_webxdc_realtime_data(ctx, msg_id, data).await?;
+    Ok(())
+}
+
+/// Leave the gossip of the webxdc with given [MsgId].
+pub async fn leave_webxdc_realtime(ctx: &Context, msg_id: MsgId) -> Result<()> {
+    let iroh = ctx.get_or_try_init_peer_channel().await?;
+    iroh.leave_realtime(get_iroh_topic_for_msg(ctx, msg_id).await?)
+        .await?;
+    info!(ctx, "IROH_REALTIME: Left gossip for message {msg_id}");
+
+    Ok(())
+}
+
+pub(crate) fn create_random_topic() -> TopicId {
+    TopicId::from_bytes(rand::random())
+}
+
+pub(crate) async fn create_iroh_header(
+    ctx: &Context,
+    topic: TopicId,
+    msg_id: MsgId,
+) -> Result<Header> {
+    insert_topic_stub(ctx, msg_id, topic).await?;
+    Ok(Header::new(
+        HeaderDef::IrohGossipTopic.get_headername().to_string(),
+        topic.to_string(),
+    ))
+}
+
+async fn endpoint_loop(context: Context, endpoint: MagicEndpoint, gossip: Gossip) {
+    while let Some(conn) = endpoint.accept().await {
+        info!(context, "IROH_REALTIME: accepting iroh connection");
+        let gossip = gossip.clone();
+        let context = context.clone();
+        tokio::spawn(async move {
+            if let Err(err) = handle_connection(&context, conn, gossip).await {
+                warn!(context, "IROH_REALTIME: iroh connection error: {err}");
+            }
+        });
+    }
+}
+
+async fn handle_connection(
+    context: &Context,
+    mut conn: iroh_net::magic_endpoint::Connecting,
+    gossip: Gossip,
+) -> anyhow::Result<()> {
+    let alpn = conn.alpn().await?;
+    let conn = conn.await?;
+    let peer_id = iroh_net::magic_endpoint::get_remote_node_id(&conn)?;
+
+    match alpn.as_bytes() {
+        GOSSIP_ALPN => gossip
+            .handle_connection(conn)
+            .await
+            .context(format!("Connection to {peer_id} with ALPN {alpn} failed"))?,
+        _ => warn!(
+            context,
+            "Ignoring connection from {peer_id}: unsupported ALPN protocol"
+        ),
+    }
+    Ok(())
+}
+
+async fn subscribe_loop(
+    context: &Context,
+    gossip: Gossip,
+    topic: TopicId,
+    msg_id: MsgId,
+) -> Result<()> {
+    let mut stream = gossip.subscribe(topic).await?;
+    loop {
+        let event = stream.recv().await?;
+        match event {
+            IrohEvent::NeighborUp(node) => {
+                info!(context, "IROH_REALTIME: NeighborUp: {}", node.to_string());
+                iroh_add_peer_for_topic(context, msg_id, topic, node, None).await?;
+            }
+            IrohEvent::Received(event) => {
+                info!(context, "IROH_REALTIME: Received realtime data");
+                context.emit_event(EventType::WebxdcRealtimeData {
+                    msg_id,
+                    data: event
+                        .content
+                        .get(0..event.content.len() - 4 - PUBLIC_KEY_LENGTH)
+                        .context("too few bytes in iroh message")?
+                        .into(),
+                });
+            }
+            _ => (),
+        };
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{
+        chat::send_msg,
+        message::{Message, Viewtype},
+        test_utils::TestContextManager,
+        EventType,
+    };
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_can_communicate() {
+        let mut tcm = TestContextManager::new();
+        let alice = &mut tcm.alice().await;
+        let bob = &mut tcm.bob().await;
+
+        // Alice sends webxdc to bob
+        let alice_chat = alice.create_chat(bob).await;
+        let mut instance = Message::new(Viewtype::File);
+        instance
+            .set_file_from_bytes(
+                alice,
+                "minimal.xdc",
+                include_bytes!("../test-data/webxdc/minimal.xdc"),
+                None,
+            )
+            .await
+            .unwrap();
+
+        send_msg(alice, alice_chat.id, &mut instance).await.unwrap();
+        let alice_webxdc = alice.get_last_msg().await;
+        assert_eq!(alice_webxdc.get_viewtype(), Viewtype::Webxdc);
+
+        let webxdc = alice.pop_sent_msg().await;
+        let bob_webdxc = bob.recv_msg(&webxdc).await;
+        assert_eq!(bob_webdxc.get_viewtype(), Viewtype::Webxdc);
+
+        bob_webdxc.chat_id.accept(bob).await.unwrap();
+
+        // Alice advertises herself.
+        send_webxdc_realtime_advertisement(alice, alice_webxdc.id)
+            .await
+            .unwrap();
+
+        bob.recv_msg_trash(&alice.pop_sent_msg().await).await;
+        let bob_iroh = bob.get_or_try_init_peer_channel().await.unwrap();
+
+        // Bob adds alice to gossip peers.
+        let members = get_iroh_gossip_peers(bob, bob_webdxc.id)
+            .await
+            .unwrap()
+            .into_iter()
+            .map(|addr| addr.node_id)
+            .collect::<Vec<_>>();
+
+        let alice_iroh = alice.get_or_try_init_peer_channel().await.unwrap();
+        assert_eq!(
+            members,
+            vec![alice_iroh.get_node_addr().await.unwrap().node_id]
+        );
+
+        bob_iroh
+            .join_and_subscribe_gossip(bob, bob_webdxc.id)
+            .await
+            .unwrap()
+            .unwrap()
+            .await
+            .unwrap();
+
+        // Alice sends ephemeral message
+        alice_iroh
+            .send_webxdc_realtime_data(alice, alice_webxdc.id, "alice -> bob".as_bytes().to_vec())
+            .await
+            .unwrap();
+
+        loop {
+            let event = bob.evtracker.recv().await.unwrap();
+            if let EventType::WebxdcRealtimeData { data, .. } = event.typ {
+                if data == "alice -> bob".as_bytes() {
+                    break;
+                } else {
+                    panic!(
+                        "Unexpected status update: {}",
+                        String::from_utf8_lossy(&data)
+                    );
+                }
+            }
+        }
+        // Bob sends ephemeral message
+        bob_iroh
+            .send_webxdc_realtime_data(bob, bob_webdxc.id, "bob -> alice".as_bytes().to_vec())
+            .await
+            .unwrap();
+
+        loop {
+            let event = alice.evtracker.recv().await.unwrap();
+            if let EventType::WebxdcRealtimeData { data, .. } = event.typ {
+                if data == "bob -> alice".as_bytes() {
+                    break;
+                } else {
+                    panic!(
+                        "Unexpected status update: {}",
+                        String::from_utf8_lossy(&data)
+                    );
+                }
+            }
+        }
+
+        // Alice adds bob to gossip peers.
+        let members = get_iroh_gossip_peers(alice, alice_webxdc.id)
+            .await
+            .unwrap()
+            .into_iter()
+            .map(|addr| addr.node_id)
+            .collect::<Vec<_>>();
+
+        assert_eq!(
+            members,
+            vec![bob_iroh.get_node_addr().await.unwrap().node_id]
+        );
+
+        bob_iroh
+            .send_webxdc_realtime_data(bob, bob_webdxc.id, "bob -> alice 2".as_bytes().to_vec())
+            .await
+            .unwrap();
+
+        loop {
+            let event = alice.evtracker.recv().await.unwrap();
+            if let EventType::WebxdcRealtimeData { data, .. } = event.typ {
+                if data == "bob -> alice 2".as_bytes() {
+                    break;
+                } else {
+                    panic!(
+                        "Unexpected status update: {}",
+                        String::from_utf8_lossy(&data)
+                    );
+                }
+            }
+        }
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_can_reconnect() {
+        let mut tcm = TestContextManager::new();
+        let alice = &mut tcm.alice().await;
+        let bob = &mut tcm.bob().await;
+
+        // Alice sends webxdc to bob
+        let alice_chat = alice.create_chat(bob).await;
+        let mut instance = Message::new(Viewtype::File);
+        instance
+            .set_file_from_bytes(
+                alice,
+                "minimal.xdc",
+                include_bytes!("../test-data/webxdc/minimal.xdc"),
+                None,
+            )
+            .await
+            .unwrap();
+
+        send_msg(alice, alice_chat.id, &mut instance).await.unwrap();
+        let alice_webxdc = alice.get_last_msg().await;
+        assert_eq!(alice_webxdc.get_viewtype(), Viewtype::Webxdc);
+
+        let webxdc = alice.pop_sent_msg().await;
+        let bob_webdxc = bob.recv_msg(&webxdc).await;
+        assert_eq!(bob_webdxc.get_viewtype(), Viewtype::Webxdc);
+
+        bob_webdxc.chat_id.accept(bob).await.unwrap();
+
+        // Alice advertises herself.
+        send_webxdc_realtime_advertisement(alice, alice_webxdc.id)
+            .await
+            .unwrap();
+
+        bob.recv_msg_trash(&alice.pop_sent_msg().await).await;
+        let bob_iroh = bob.get_or_try_init_peer_channel().await.unwrap();
+
+        // Bob adds alice to gossip peers.
+        let members = get_iroh_gossip_peers(bob, bob_webdxc.id)
+            .await
+            .unwrap()
+            .into_iter()
+            .map(|addr| addr.node_id)
+            .collect::<Vec<_>>();
+
+        let alice_iroh = alice.get_or_try_init_peer_channel().await.unwrap();
+        assert_eq!(
+            members,
+            vec![alice_iroh.get_node_addr().await.unwrap().node_id]
+        );
+
+        bob_iroh
+            .join_and_subscribe_gossip(bob, bob_webdxc.id)
+            .await
+            .unwrap()
+            .unwrap()
+            .await
+            .unwrap();
+
+        // Alice sends ephemeral message
+        alice_iroh
+            .send_webxdc_realtime_data(alice, alice_webxdc.id, "alice -> bob".as_bytes().to_vec())
+            .await
+            .unwrap();
+
+        loop {
+            let event = bob.evtracker.recv().await.unwrap();
+            if let EventType::WebxdcRealtimeData { data, .. } = event.typ {
+                if data == "alice -> bob".as_bytes() {
+                    break;
+                } else {
+                    panic!(
+                        "Unexpected status update: {}",
+                        String::from_utf8_lossy(&data)
+                    );
+                }
+            }
+        }
+
+        // TODO: check that seq number is persisted
+        leave_webxdc_realtime(bob, bob_webdxc.id).await.unwrap();
+
+        bob_iroh
+            .join_and_subscribe_gossip(bob, bob_webdxc.id)
+            .await
+            .unwrap()
+            .unwrap()
+            .await
+            .unwrap();
+
+        bob_iroh
+            .send_webxdc_realtime_data(bob, bob_webdxc.id, "bob -> alice".as_bytes().to_vec())
+            .await
+            .unwrap();
+
+        loop {
+            let event = alice.evtracker.recv().await.unwrap();
+            if let EventType::WebxdcRealtimeData { data, .. } = event.typ {
+                if data == "bob -> alice".as_bytes() {
+                    break;
+                } else {
+                    panic!(
+                        "Unexpected status update: {}",
+                        String::from_utf8_lossy(&data)
+                    );
+                }
+            }
+        }
+
+        // channel is only used to remeber if an advertisement has been sent
+        // bob for example does not change the channels because he never sends an
+        // advertisement
+        assert_eq!(
+            alice.iroh.get().unwrap().iroh_channels.read().await.len(),
+            1
+        );
+        leave_webxdc_realtime(alice, alice_webxdc.id).await.unwrap();
+        let topic = get_iroh_topic_for_msg(alice, alice_webxdc.id)
+            .await
+            .unwrap();
+        assert!(if let Some(state) = alice
+            .iroh
+            .get()
+            .unwrap()
+            .iroh_channels
+            .read()
+            .await
+            .get(&topic)
+        {
+            state.subscribe_loop.is_none()
+        } else {
+            false
+        });
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_parallel_connect() {
+        let mut tcm = TestContextManager::new();
+        let alice = &mut tcm.alice().await;
+        let bob = &mut tcm.bob().await;
+
+        // Alice sends webxdc to bob
+        let alice_chat = alice.create_chat(bob).await;
+        let mut instance = Message::new(Viewtype::File);
+        instance
+            .set_file_from_bytes(
+                alice,
+                "minimal.xdc",
+                include_bytes!("../test-data/webxdc/minimal.xdc"),
+                None,
+            )
+            .await
+            .unwrap();
+        send_msg(alice, alice_chat.id, &mut instance).await.unwrap();
+        let alice_webxdc = alice.get_last_msg().await;
+
+        let webxdc = alice.pop_sent_msg().await;
+        let bob_webxdc = bob.recv_msg(&webxdc).await;
+        assert_eq!(bob_webxdc.get_viewtype(), Viewtype::Webxdc);
+
+        bob_webxdc.chat_id.accept(bob).await.unwrap();
+
+        eprintln!("Sending advertisements");
+        // Alice advertises herself.
+        let alice_advertisement_future = send_webxdc_realtime_advertisement(alice, alice_webxdc.id)
+            .await
+            .unwrap()
+            .unwrap();
+        let alice_advertisement = alice.pop_sent_msg().await;
+
+        send_webxdc_realtime_advertisement(bob, bob_webxdc.id)
+            .await
+            .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");
+        alice_advertisement_future.await;
+
+        // Alice sends ephemeral message
+        eprintln!("Sending ephemeral message");
+        send_webxdc_realtime_data(alice, alice_webxdc.id, b"alice -> bob".into())
+            .await
+            .unwrap();
+
+        eprintln!("Waiting for ephemeral message");
+        loop {
+            let event = bob.evtracker.recv().await.unwrap();
+            if let EventType::WebxdcRealtimeData { data, .. } = event.typ {
+                if data == b"alice -> bob" {
+                    break;
+                } else {
+                    panic!(
+                        "Unexpected status update: {}",
+                        String::from_utf8_lossy(&data)
+                    );
+                }
+            }
+        }
+    }
+}

src/qr.rs

@@ -23,6 +23,7 @@ use crate::peerstate::Peerstate;
 use crate::socks::Socks5Config;
 use crate::token;
 use crate::tools::validate_id;
+use iroh_old as iroh;
 
 const OPENPGP4FPR_SCHEME: &str = "OPENPGP4FPR:"; // yes: uppercase
 const IDELTACHAT_SCHEME: &str = "https://i.delta.chat/#";

src/receive_imf.rs

@@ -1,11 +1,13 @@
 //! Internet Message Format reception pipeline.
 
 use std::collections::HashSet;
+use std::str::FromStr;
 
 use anyhow::{Context as _, Result};
 use deltachat_contact_tools::{
     addr_cmp, may_be_valid_addr, normalize_name, strip_rtlo_characters, ContactAddress,
 };
+use iroh_gossip::proto::TopicId;
 use mailparse::{parse_mail, SingleInfo};
 use num_traits::FromPrimitive;
 use once_cell::sync::Lazy;
@@ -30,6 +32,7 @@ use crate::message::{
 };
 use crate::mimeparser::{parse_message_ids, AvatarAction, MimeMessage, SystemMessage};
 use crate::param::{Param, Params};
+use crate::peer_channels::{get_iroh_topic_for_msg, insert_topic_stub, iroh_add_peer_for_topic};
 use crate::peerstate::Peerstate;
 use crate::reaction::{set_msg_reaction, Reaction};
 use crate::securejoin::{self, handle_securejoin_handshake, observe_securejoin_on_other_device};
@@ -40,6 +43,7 @@ use crate::sync::Sync::*;
 use crate::tools::{self, buf_compress, extract_grpid_from_rfc724_mid};
 use crate::{chatlist_events, location};
 use crate::{contact, imap};
+use iroh_net::NodeAddr;
 
 /// This is the struct that is returned after receiving one email (aka MIME message).
 ///
@@ -1220,7 +1224,7 @@ async fn add_parts(
     }
 
     let orig_chat_id = chat_id;
-    let chat_id = if is_mdn || is_reaction {
+    let mut chat_id = if is_mdn || is_reaction {
         DC_CHAT_ID_TRASH
     } else {
         chat_id.unwrap_or_else(|| {
@@ -1430,6 +1434,25 @@ async fn add_parts(
         .await?;
     }
 
+    if let Some(node_addr) = mime_parser.get_header(HeaderDef::IrohNodeAddr) {
+        match serde_json::from_str::<NodeAddr>(node_addr).context("Failed to parse node address") {
+            Ok(node_addr) => {
+                info!(context, "Adding iroh peer with address {node_addr:?}.");
+                let instance_id = parent.context("Failed to get parent message")?.id;
+                let node_id = node_addr.node_id;
+                let relay_server = node_addr.relay_url().map(|relay| relay.as_str());
+                let topic = get_iroh_topic_for_msg(context, instance_id).await?;
+                iroh_add_peer_for_topic(context, instance_id, topic, node_id, relay_server).await?;
+                let iroh = context.get_or_try_init_peer_channel().await?;
+                iroh.maybe_add_gossip_peers(topic, vec![node_addr]).await?;
+                chat_id = DC_CHAT_ID_TRASH;
+            }
+            Err(err) => {
+                warn!(context, "Couldn't parse NodeAddr: {err:#}.");
+            }
+        }
+    }
+
     for part in &mime_parser.parts {
         if part.is_reaction {
             let reaction_str = simplify::remove_footers(part.msg.as_str());
@@ -1597,6 +1620,16 @@ RETURNING id
 
     // check all parts whether they contain a new logging webxdc
     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 {
+            if let Some(topic) = mime_parser.get_header(HeaderDef::IrohGossipTopic) {
+                let topic = TopicId::from_str(topic).context("wrong gossip topic header")?;
+                insert_topic_stub(context, *msg_id, topic).await?;
+            } else {
+                warn!(context, "webxdc doesn't have a gossip topic")
+            }
+        }
+
         maybe_set_logging_xdc_inner(
             context,
             part.typ,

src/sql/migrations.rs

@@ -912,6 +912,30 @@ CREATE INDEX msgs_status_updates_index2 ON msgs_status_updates (uid);
         .await?;
     }
 
+    if dbversion < 111 {
+        sql.execute_migration(
+            "CREATE TABLE iroh_gossip_peers (msg_id TEXT not NULL, topic TEXT NOT NULL, public_key TEXT NOT NULL)",
+            111,
+        )
+        .await?;
+    }
+
+    if dbversion < 112 {
+        sql.execute_migration(
+            "DROP TABLE iroh_gossip_peers; CREATE TABLE iroh_gossip_peers (msg_id INTEGER not NULL, topic BLOB NOT NULL, public_key BLOB NOT NULL, relay_server TEXT, UNIQUE (public_key, topic)) STRICT",
+            112,
+        )
+        .await?;
+    }
+
+    if dbversion < 113 {
+        sql.execute_migration(
+            "DROP TABLE iroh_gossip_peers; CREATE TABLE iroh_gossip_peers (msg_id INTEGER not NULL, topic BLOB NOT NULL, public_key BLOB NOT NULL, relay_server TEXT, UNIQUE (topic, public_key), PRIMARY KEY(topic, public_key)) STRICT",
+            113,
+        )
+        .await?;
+    }
+
     let new_version = sql
         .get_raw_config_int(VERSION_CFG)
         .await?