From 0a7e697f50041c0f772e0f832e23e58a3ce6236a Mon Sep 17 00:00:00 2001 From: link2xt Date: Thu, 17 Sep 2026 18:03:13 +0000 Subject: [PATCH] refactor: add smtp::queue module No changes except that smtp::is_queue_empty is renamed into smtp::queue::is_empty, mimefactory::QueuedEncryption is renamed into smtp::queue::Encryption and mimefactory::QueueSideEffects is renamed to smtp::queue::SideEffects. mimefactory module does not know anything about smtp2 table anymore. --- src/accounts.rs | 2 +- src/chat.rs | 101 +-------- src/mimefactory.rs | 109 +-------- src/receive_imf/receive_imf_tests.rs | 2 +- src/smtp.rs | 132 +---------- src/smtp/queue.rs | 323 +++++++++++++++++++++++++++ src/test_utils.rs | 4 +- 7 files changed, 339 insertions(+), 334 deletions(-) create mode 100644 src/smtp/queue.rs diff --git a/src/accounts.rs b/src/accounts.rs index 66f3b32f9..131013147 100644 --- a/src/accounts.rs +++ b/src/accounts.rs @@ -528,7 +528,7 @@ impl Accounts { pub async fn is_sending_finished(&self) -> Result { let accounts: Vec = self.accounts.values().cloned().collect(); for account in accounts { - if !smtp::is_queue_empty(&account).await? { + if !smtp::queue::is_empty(&account).await? { return Ok(false); } } diff --git a/src/chat.rs b/src/chat.rs index 2a838a218..51a21e7b0 100644 --- a/src/chat.rs +++ b/src/chat.rs @@ -30,17 +30,17 @@ use crate::download::{DownloadState, PRE_MSG_ATTACHMENT_SIZE_THRESHOLD}; use crate::ensure_and_debug_assert_eq; use crate::ephemeral::{Timer as EphemeralTimer, start_chat_ephemeral_timers}; use crate::events::EventType; -use crate::key::{DcKey as _, Fingerprint, self_fingerprint}; +use crate::key::{Fingerprint, self_fingerprint}; use crate::log::{LogExt, warn}; use crate::logged_debug_assert; use crate::message::{self, Message, MessageState, MsgId, Viewtype}; -use crate::mimefactory; -use crate::mimefactory::{MimeFactory, QueueSideEffects, QueuedMail, ToBeQueuedMail}; +use crate::mimefactory::MimeFactory; use crate::mimeparser::SystemMessage; use crate::param::{Param, Params}; use crate::pgp::addresses_from_public_key; use crate::reaction::broadcast_reactions; use crate::receive_imf::ReceivedMsg; +use crate::smtp::queue::{ToBeQueuedMail, enqueue_mail}; use crate::smtp::send_msg_to_smtp; use crate::stock_str; use crate::sync::{self, Sync::*, SyncData}; @@ -331,7 +331,7 @@ impl ChatId { Ok(chat_id) } - fn set_selfavatar_timestamp( + pub(crate) fn set_selfavatar_timestamp( self, transaction: &mut rusqlite::Transaction<'_>, timestamp: i64, @@ -2815,99 +2815,6 @@ async fn render_mime_message_and_pre_message( } } -/// Process side effects and store queued mail. -pub(crate) fn enqueue_mail( - transaction: &mut rusqlite::Transaction<'_>, - now: i64, - msg_id: MsgId, - queued_mail: &QueuedMail, - side_effects: Option<&QueueSideEffects>, -) -> Result { - if let Some(side_effects) = side_effects { - if let Some(last_added_location_timestamp) = side_effects.last_added_location_timestamp { - transaction.execute( - "UPDATE chats SET locations_last_sent=? WHERE id=?;", - (last_added_location_timestamp, side_effects.chat_id), - )?; - } - - if side_effects.avatar_is_attached { - side_effects - .chat_id - .set_selfavatar_timestamp(transaction, now) - .context("Failed to set selfavatar timestamp")?; - } - - if let Some(ref sync_ids) = side_effects.sync_ids_to_delete { - transaction.execute( - &format!("DELETE FROM multi_device_sync WHERE id IN ({sync_ids})"), - (), - )?; - } - } - - // Store mail into queue. - let all_recipients = queued_mail.recipients.join(" "); - let is_encrypted = queued_mail.encryption.is_encrypted(); - - transaction - .execute( - " - INSERT INTO smtp2 ( - display_name, - rfc724_mid, - mime, - should_attach_pubkey, - should_compress, - should_sign, - msg_id, - recipients, - bcc_self, - is_encrypted, - shared_secret, - encryption_fingerprints - ) - VALUES ( - ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? - ) - ", - ( - &queued_mail.display_name, - &queued_mail.rfc724_mid, - &queued_mail.raw_message, - queued_mail.should_attach_pubkey, - queued_mail.should_compress, - queued_mail.should_sign, - msg_id, - &all_recipients, - queued_mail.bcc_self, - is_encrypted, - if let mimefactory::QueuedEncryption::Symmetric { ref shared_secret } = - queued_mail.encryption - { - shared_secret - } else { - "" - }, - if let mimefactory::QueuedEncryption::Asymmetric { - ref encryption_pubkeys, - } = queued_mail.encryption - { - let res: Vec = encryption_pubkeys - .iter() - .map(|pubkey| pubkey.dc_fingerprint().hex()) - .collect(); - res.join(" ") - } else { - "".to_string() - }, - ), - ) - .context("Failed to insert a row into smtp2 table")?; - let row_id = transaction.last_insert_rowid(); - Ok(row_id) -} - /// Constructs jobs for sending a message and inserts them into the `smtp` table. /// /// Updates the message `GuaranteeE2ee` parameter and persists it diff --git a/src/mimefactory.rs b/src/mimefactory.rs index bb28db8b9..ae79fc908 100644 --- a/src/mimefactory.rs +++ b/src/mimefactory.rs @@ -17,7 +17,7 @@ use tokio::fs; use crate::aheader::{Aheader, EncryptPreference}; use crate::blob::BlobObject; -use crate::chat::{self, Chat, ChatId, PARAM_BROADCAST_SECRET, load_broadcast_secret}; +use crate::chat::{self, Chat, PARAM_BROADCAST_SECRET, load_broadcast_secret}; use crate::config::Config; use crate::constants::{Chattype, DC_FROM_HANDSHAKE}; use crate::contact::{Contact, ContactId, Origin}; @@ -36,6 +36,9 @@ use crate::param::Param; use crate::peer_channels::{create_iroh_header, get_iroh_topic_for_msg}; use crate::pgp::{SeipdVersion, addresses_from_public_key, pubkey_supports_seipdv2, relay_addrs}; use crate::simplify::escape_message_footer_marks; +use crate::smtp::queue::{ + Encryption as QueuedEncryption, QueuedMail, SideEffects as QueueSideEffects, ToBeQueuedMail, +}; use crate::stock_str; use crate::tools::{IsNoneOrEmpty, create_outgoing_rfc724_mid, remove_subject_prefix, time}; use crate::webxdc::StatusUpdateSerial; @@ -221,110 +224,6 @@ pub struct RenderedMessage { sync_ids_to_delete: Option, } -#[derive(Debug, Clone)] -pub(crate) enum QueuedEncryption { - /// Unencrypted message. - No, - - /// The message is encrypted asymmetrically to public keys. - Asymmetric { - /// OpenPGP keys to use for encryption. - /// - /// The message is always encrypted to self, - /// no need to include own key here. - encryption_pubkeys: Vec, - }, - - /// Symmetrically encrypted message with a shared secret. - Symmetric { shared_secret: String }, -} - -impl QueuedEncryption { - pub(crate) fn is_encrypted(&self) -> bool { - match self { - Self::No => false, - Self::Asymmetric { .. } => true, - Self::Symmetric { .. } => true, - } - } -} - -/// Email message queued, but not sent yet. -/// -/// It is stored unencrypted to -/// make it possible to change protected headers -/// like the From address and Autocrypt header later. -#[derive(Debug, Clone)] -pub(crate) struct QueuedMail { - /// Unencrypted queued message. - /// - /// This message has both the headers and the body, - /// but without the From, Autocrypt and Message-ID headers. - /// - /// For encrypted messages this is the OpenPGP payload. - pub(crate) raw_message: Vec, - - /// Display name to put in the `From:` field. - /// - /// Email address is not determined yet here. - pub(crate) display_name: String, - - /// Message-ID. - pub(crate) rfc724_mid: String, - - /// Whether the message is encrypted and encryption keys. - pub(crate) encryption: QueuedEncryption, - - /// If true, Autocrypt header should be added before sending. - pub(crate) should_attach_pubkey: bool, - - /// If true, OpenPGP compression may be used. - pub(crate) should_compress: bool, - - /// If true, encrypted message should be signed. - pub(crate) should_sign: bool, - - /// Recipient addresses. - pub(crate) recipients: Vec, - - /// Addresses the messages was already sent to. - pub(crate) sent_to: Vec, - - /// If true, own addresses should be added to the list of recipients. - /// - /// For unencrypted messages, only the sending addresses should be added. - /// For encrypted messages, all published addresses should be added. - pub(crate) bcc_self: bool, -} - -/// Side effects that should be applied at the same time -/// as the message is persisted in the queue. -#[derive(Debug, Clone, Default)] -pub struct QueueSideEffects { - /// ID of the chat side effects should be applied to. - pub chat_id: ChatId, - - /// Largest timestamp of the location sent in `location.kml` in this message. - pub last_added_location_timestamp: Option, - - /// True if the message has the avatar attached. - /// - /// Timestamp of the last time avatar was gossiped should be updated. - pub avatar_is_attached: bool, - - /// A comma-separated string of sync-IDs that are used by the rendered email and must be deleted - /// from `multi_device_sync` once the message is actually queued for sending. - pub sync_ids_to_delete: Option, - - /// Subject that was rendered into the message. - /// - /// Used to update the subject on the sent message object. - pub subject: String, -} - -/// Email message ready to be queued with the side effects that should be applied at the same time. -pub(crate) type ToBeQueuedMail = (QueuedMail, Option); - /// Renders [`QueuedMail`]. /// /// Adds headers: diff --git a/src/receive_imf/receive_imf_tests.rs b/src/receive_imf/receive_imf_tests.rs index 8283680b6..5f7eaa32e 100644 --- a/src/receive_imf/receive_imf_tests.rs +++ b/src/receive_imf/receive_imf_tests.rs @@ -5763,7 +5763,7 @@ pub(crate) async fn first_row_in_smtp_queue(context: &TestContext) -> (MsgId, St let queued_mail = context .sql .transaction_ext(query_only, |transaction| { - smtp::load_queued_mail(transaction, rowid) + smtp::queue::load_queued_mail(transaction, rowid) }) .await .unwrap(); diff --git a/src/smtp.rs b/src/smtp.rs index d0352f70e..285d9a841 100644 --- a/src/smtp.rs +++ b/src/smtp.rs @@ -1,6 +1,7 @@ //! # SMTP transport module. mod connect; +pub(crate) mod queue; pub mod send; use std::collections::BTreeSet; @@ -8,27 +9,24 @@ use std::collections::BTreeSet; use anyhow::{Context as _, Error, Result, bail, format_err}; use async_smtp::response::{Category, Code, Detail}; use async_smtp::{EmailAddress, SmtpTransport}; -use pgp::composed::SignedPublicKey; use rusqlite::OptionalExtension as _; use tokio::task; -use crate::chat; use crate::chat::{ChatId, add_info_msg_with_cmd}; use crate::config::Config; use crate::contact::{Contact, ContactId}; use crate::context::Context; use crate::events::EventType; use crate::key; -use crate::key::DcKey; use crate::log::{LogExt, warn}; use crate::message::Message; use crate::message::{self, MsgId}; use crate::mimefactory; use crate::mimefactory::MimeFactory; -use crate::mimefactory::QueuedMail; use crate::net::proxy::ProxyConfig; use crate::net::session::SessionBufStream; use crate::scheduler::connectivity::ConnectivityStore; +use crate::smtp::queue::QueuedMail; use crate::stock_str::unencrypted_email; use crate::tools::{self, time_elapsed}; use crate::transport::{ @@ -364,7 +362,7 @@ pub(crate) async fn insert_into_smtp( let msg_id = message::insert_tombstone(context, rfc724_mid).await?; context .sql - .transaction(|transaction| chat::enqueue_mail(transaction, now, msg_id, queued_msg, None)) + .transaction(|transaction| queue::enqueue_mail(transaction, now, msg_id, queued_msg, None)) .await?; Ok(()) } @@ -413,7 +411,7 @@ pub(crate) async fn send_msg_to_smtp( else { return Ok(None); }; - let queued_mail = load_queued_mail(transaction, rowid) + let queued_mail = queue::load_queued_mail(transaction, rowid) .with_context(|| format!("Failed to load queued mail for {rowid}"))?; Ok(Some((queued_mail, msg_id, retries))) }) @@ -825,125 +823,3 @@ pub(crate) async fn add_self_recipients( Ok(()) } - -/// Returns true if SMTP queue is empty. -pub(crate) async fn is_queue_empty(context: &Context) -> Result { - let sending_finished = !context.sql.exists("SELECT COUNT(*) FROM smtp2", ()).await?; - Ok(sending_finished) -} - -/// Loads the queued mail from `smtp2` table and the list of recipients. -pub(crate) fn load_queued_mail( - transaction: &mut rusqlite::Transaction<'_>, - row_id: i64, -) -> Result { - let (mut queued_mail, encryption_fingerprints) = transaction - .query_row_and_then( - " -SELECT display_name, - rfc724_mid, - mime, - should_attach_pubkey, - should_compress, - should_sign, - is_encrypted, - shared_secret, - encryption_fingerprints, - recipients, - sent_to, - bcc_self -FROM smtp2 WHERE id = ? -", - (row_id,), - |row| { - let display_name: String = row.get(0)?; - let rfc724_mid: String = row.get(1)?; - let raw_message: Vec = row.get(2)?; - let should_attach_pubkey: bool = row.get(3)?; - let should_compress: bool = row.get(4)?; - let should_sign: bool = row.get(5)?; - let is_encrypted: bool = row.get(6)?; - let shared_secret: String = row.get(7)?; - let encryption_fingerprints: String = row.get(8)?; - let encryption_fingerprints: Vec = if encryption_fingerprints.is_empty() { - Vec::new() - } else { - encryption_fingerprints - .split(' ') - .map(|s| s.to_string()) - .collect() - }; - let recipients: String = row.get(9)?; - let recipients: Vec = if recipients.is_empty() { - Vec::new() - } else { - recipients.split(' ').map(|s| s.to_string()).collect() - }; - debug_assert!(!recipients.iter().any(|s| s.is_empty())); - let sent_to: String = row.get(10)?; - let sent_to: Vec = if sent_to.is_empty() { - Vec::new() - } else { - sent_to.split(' ').map(|s| s.to_string()).collect() - }; - let bcc_self: bool = row.get(11)?; - - let encryption = match ( - is_encrypted, - shared_secret.is_empty(), - encryption_fingerprints.is_empty(), - ) { - (false, true, true) => mimefactory::QueuedEncryption::No, - (true, false, true) => { - mimefactory::QueuedEncryption::Symmetric { shared_secret } - } - (true, true, _) => mimefactory::QueuedEncryption::Asymmetric { - // Public keys are loaded below based on the encryption fingerprints. - encryption_pubkeys: Vec::new(), - }, - _ => bail!("Invalid encryption in smtp2 row"), - }; - Ok::<_, anyhow::Error>(( - QueuedMail { - raw_message, - display_name, - rfc724_mid, - encryption, - should_attach_pubkey, - should_compress, - should_sign, - recipients, - sent_to, - bcc_self, - }, - encryption_fingerprints, - )) - }, - ) - .with_context(|| format!("Failed to select row {row_id} from smtp2 table"))?; - - if let mimefactory::QueuedEncryption::Asymmetric { - ref mut encryption_pubkeys, - } = queued_mail.encryption - { - for fingerprint in encryption_fingerprints { - let public_key_bytes: Option> = transaction - .query_row( - "SELECT public_key FROM public_keys WHERE fingerprint=?", - (fingerprint,), - |row| { - let bytes: Vec = row.get(0)?; - Ok(bytes) - }, - ) - .optional() - .context("Failed to select public key by fingerprint")?; - if let Some(public_key_bytes) = public_key_bytes { - let public_key = SignedPublicKey::from_slice(&public_key_bytes)?; - encryption_pubkeys.push(public_key); - } - } - } - - Ok(queued_mail) -} diff --git a/src/smtp/queue.rs b/src/smtp/queue.rs new file mode 100644 index 000000000..6ea9978ba --- /dev/null +++ b/src/smtp/queue.rs @@ -0,0 +1,323 @@ +//! # SMTP queue module. +use anyhow::{Context as _, Result, bail}; + +use crate::chat::ChatId; +use crate::context::Context; +use crate::key::{DcKey, SignedPublicKey}; +use crate::message::MsgId; +use rusqlite::OptionalExtension as _; + +#[derive(Debug, Clone)] +pub(crate) enum Encryption { + /// Unencrypted message. + No, + + /// The message is encrypted asymmetrically to public keys. + Asymmetric { + /// OpenPGP keys to use for encryption. + /// + /// The message is always encrypted to self, + /// no need to include own key here. + encryption_pubkeys: Vec, + }, + + /// Symmetrically encrypted message with a shared secret. + Symmetric { shared_secret: String }, +} + +impl Encryption { + pub(crate) fn is_encrypted(&self) -> bool { + match self { + Self::No => false, + Self::Asymmetric { .. } => true, + Self::Symmetric { .. } => true, + } + } +} + +/// Email message queued, but not sent yet. +/// +/// It is stored unencrypted to +/// make it possible to change protected headers +/// like the From address and Autocrypt header later. +#[derive(Debug, Clone)] +pub(crate) struct QueuedMail { + /// Unencrypted queued message. + /// + /// This message has both the headers and the body, + /// but without the From, Autocrypt and Message-ID headers. + /// + /// For encrypted messages this is the OpenPGP payload. + pub(crate) raw_message: Vec, + + /// Display name to put in the `From:` field. + /// + /// Email address is not determined yet here. + pub(crate) display_name: String, + + /// Message-ID. + pub(crate) rfc724_mid: String, + + /// Whether the message is encrypted and encryption keys. + pub(crate) encryption: Encryption, + + /// If true, Autocrypt header should be added before sending. + pub(crate) should_attach_pubkey: bool, + + /// If true, OpenPGP compression may be used. + pub(crate) should_compress: bool, + + /// If true, encrypted message should be signed. + pub(crate) should_sign: bool, + + /// Recipient addresses. + pub(crate) recipients: Vec, + + /// Addresses the messages was already sent to. + pub(crate) sent_to: Vec, + + /// If true, own addresses should be added to the list of recipients. + /// + /// For unencrypted messages, only the sending addresses should be added. + /// For encrypted messages, all published addresses should be added. + pub(crate) bcc_self: bool, +} + +/// Side effects that should be applied at the same time +/// as the message is persisted in the queue. +#[derive(Debug, Clone, Default)] +pub struct SideEffects { + /// ID of the chat side effects should be applied to. + pub chat_id: ChatId, + + /// Largest timestamp of the location sent in `location.kml` in this message. + pub last_added_location_timestamp: Option, + + /// True if the message has the avatar attached. + /// + /// Timestamp of the last time avatar was gossiped should be updated. + pub avatar_is_attached: bool, + + /// A comma-separated string of sync-IDs that are used by the rendered email and must be deleted + /// from `multi_device_sync` once the message is actually queued for sending. + pub sync_ids_to_delete: Option, + + /// Subject that was rendered into the message. + /// + /// Used to update the subject on the sent message object. + pub subject: String, +} + +/// Email message ready to be queued with the side effects that should be applied at the same time. +pub(crate) type ToBeQueuedMail = (QueuedMail, Option); + +/// Process side effects and store queued mail. +pub(crate) fn enqueue_mail( + transaction: &mut rusqlite::Transaction<'_>, + now: i64, + msg_id: MsgId, + queued_mail: &QueuedMail, + side_effects: Option<&SideEffects>, +) -> Result { + if let Some(side_effects) = side_effects { + if let Some(last_added_location_timestamp) = side_effects.last_added_location_timestamp { + transaction.execute( + "UPDATE chats SET locations_last_sent=? WHERE id=?;", + (last_added_location_timestamp, side_effects.chat_id), + )?; + } + + if side_effects.avatar_is_attached { + side_effects + .chat_id + .set_selfavatar_timestamp(transaction, now) + .context("Failed to set selfavatar timestamp")?; + } + + if let Some(ref sync_ids) = side_effects.sync_ids_to_delete { + transaction.execute( + &format!("DELETE FROM multi_device_sync WHERE id IN ({sync_ids})"), + (), + )?; + } + } + + // Store mail into queue. + let all_recipients = queued_mail.recipients.join(" "); + let is_encrypted = queued_mail.encryption.is_encrypted(); + + transaction + .execute( + " + INSERT INTO smtp2 ( + display_name, + rfc724_mid, + mime, + should_attach_pubkey, + should_compress, + should_sign, + msg_id, + recipients, + bcc_self, + is_encrypted, + shared_secret, + encryption_fingerprints + ) + VALUES ( + ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? + ) + ", + ( + &queued_mail.display_name, + &queued_mail.rfc724_mid, + &queued_mail.raw_message, + queued_mail.should_attach_pubkey, + queued_mail.should_compress, + queued_mail.should_sign, + msg_id, + &all_recipients, + queued_mail.bcc_self, + is_encrypted, + if let Encryption::Symmetric { ref shared_secret } = queued_mail.encryption { + shared_secret + } else { + "" + }, + if let Encryption::Asymmetric { + ref encryption_pubkeys, + } = queued_mail.encryption + { + let res: Vec = encryption_pubkeys + .iter() + .map(|pubkey| pubkey.dc_fingerprint().hex()) + .collect(); + res.join(" ") + } else { + "".to_string() + }, + ), + ) + .context("Failed to insert a row into smtp2 table")?; + let row_id = transaction.last_insert_rowid(); + Ok(row_id) +} + +/// Loads the queued mail from `smtp2` table and the list of recipients. +pub(crate) fn load_queued_mail( + transaction: &mut rusqlite::Transaction<'_>, + row_id: i64, +) -> Result { + let (mut queued_mail, encryption_fingerprints) = transaction + .query_row_and_then( + " +SELECT display_name, + rfc724_mid, + mime, + should_attach_pubkey, + should_compress, + should_sign, + is_encrypted, + shared_secret, + encryption_fingerprints, + recipients, + sent_to, + bcc_self +FROM smtp2 WHERE id = ? +", + (row_id,), + |row| { + let display_name: String = row.get(0)?; + let rfc724_mid: String = row.get(1)?; + let raw_message: Vec = row.get(2)?; + let should_attach_pubkey: bool = row.get(3)?; + let should_compress: bool = row.get(4)?; + let should_sign: bool = row.get(5)?; + let is_encrypted: bool = row.get(6)?; + let shared_secret: String = row.get(7)?; + let encryption_fingerprints: String = row.get(8)?; + let encryption_fingerprints: Vec = if encryption_fingerprints.is_empty() { + Vec::new() + } else { + encryption_fingerprints + .split(' ') + .map(|s| s.to_string()) + .collect() + }; + let recipients: String = row.get(9)?; + let recipients: Vec = if recipients.is_empty() { + Vec::new() + } else { + recipients.split(' ').map(|s| s.to_string()).collect() + }; + debug_assert!(!recipients.iter().any(|s| s.is_empty())); + let sent_to: String = row.get(10)?; + let sent_to: Vec = if sent_to.is_empty() { + Vec::new() + } else { + sent_to.split(' ').map(|s| s.to_string()).collect() + }; + let bcc_self: bool = row.get(11)?; + + let encryption = match ( + is_encrypted, + shared_secret.is_empty(), + encryption_fingerprints.is_empty(), + ) { + (false, true, true) => Encryption::No, + (true, false, true) => Encryption::Symmetric { shared_secret }, + (true, true, _) => Encryption::Asymmetric { + // Public keys are loaded below based on the encryption fingerprints. + encryption_pubkeys: Vec::new(), + }, + _ => bail!("Invalid encryption in smtp2 row"), + }; + Ok::<_, anyhow::Error>(( + QueuedMail { + raw_message, + display_name, + rfc724_mid, + encryption, + should_attach_pubkey, + should_compress, + should_sign, + recipients, + sent_to, + bcc_self, + }, + encryption_fingerprints, + )) + }, + ) + .with_context(|| format!("Failed to select row {row_id} from smtp2 table"))?; + + if let Encryption::Asymmetric { + ref mut encryption_pubkeys, + } = queued_mail.encryption + { + for fingerprint in encryption_fingerprints { + let public_key_bytes: Option> = transaction + .query_row( + "SELECT public_key FROM public_keys WHERE fingerprint=?", + (fingerprint,), + |row| { + let bytes: Vec = row.get(0)?; + Ok(bytes) + }, + ) + .optional() + .context("Failed to select public key by fingerprint")?; + if let Some(public_key_bytes) = public_key_bytes { + let public_key = SignedPublicKey::from_slice(&public_key_bytes)?; + encryption_pubkeys.push(public_key); + } + } + } + + Ok(queued_mail) +} + +/// Returns true if SMTP queue is empty. +pub(crate) async fn is_empty(context: &Context) -> Result { + let sending_finished = !context.sql.exists("SELECT COUNT(*) FROM smtp2", ()).await?; + Ok(sending_finished) +} diff --git a/src/test_utils.rs b/src/test_utils.rs index 52e7c29fb..4e6ca6150 100644 --- a/src/test_utils.rs +++ b/src/test_utils.rs @@ -627,7 +627,7 @@ ORDER BY id" .ctx .sql .transaction_ext(query_only, |transaction| { - smtp::load_queued_mail(transaction, rowid) + smtp::queue::load_queued_mail(transaction, rowid) }) .await .expect("Failed to load queued mail"); @@ -724,7 +724,7 @@ ORDER BY id" .ctx .sql .transaction_ext(query_only, |transaction| { - smtp::load_queued_mail(transaction, rowid) + smtp::queue::load_queued_mail(transaction, rowid) }) .await .expect("Failed to load queued mail");