Mirrors/chatmail-core
1
0
Fork 0
mirror of https://github.com/chatmail/core.git synced 2026-05-03 13:26:28 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into flub/send-backup

Browse Source
This commit is contained in:
Floris Bruynooghe
2023-02-22 15:54:23 +01:00
parent a1e19e2c41 c7b7fbaf78
commit 1b39be8a42
53 changed files with 1053 additions and 776 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/chat.rs
View File

@@ -907,7 +907,8 @@ impl ChatId {
async fn parent_query<T, F>(self, context: &Context, fields: &str, f: F) -> Result<Option<T>>
where
F: FnOnce(&rusqlite::Row) -> rusqlite::Result<T>,
F: Send + FnOnce(&rusqlite::Row) -> rusqlite::Result<T>,
T: Send + 'static,
{
let sql = &context.sql;
let query = format!(

37
src/config.rs
View File

@@ -1,5 +1,7 @@
//! # Key-value configuration management.
use std::str::FromStr;
use anyhow::{ensure, Context as _, Result};
use strum::{EnumProperty, IntoEnumIterator};
use strum_macros::{AsRefStr, Display, EnumIter, EnumProperty, EnumString};
@@ -173,6 +175,10 @@ pub enum Config {
#[strum(props(default = "0"))]
DeleteDeviceAfter,
/// Move messages to the Trash folder instead of marking them "\Deleted". Overrides
/// `ProviderOptions::delete_to_trash`.
DeleteToTrash,
/// Save raw MIME messages with headers in the database if true.
SaveMimeHeaders,
@@ -227,6 +233,9 @@ pub enum Config {
/// Configured "Sent" folder.
ConfiguredSentboxFolder,
/// Configured "Trash" folder.
ConfiguredTrashFolder,
/// Unix timestamp of the last successful configuration.
ConfiguredTimestamp,
@@ -327,30 +336,37 @@ impl Context {
}
}
/// Returns 32-bit signed integer configuration value for the given key.
pub async fn get_config_int(&self, key: Config) -> Result<i32> {
/// Returns Some(T) if a value for the given key exists and was successfully parsed.
/// Returns None if could not parse.
pub async fn get_config_parsed<T: FromStr>(&self, key: Config) -> Result<Option<T>> {
self.get_config(key)
.await
.map(|s: Option<String>| s.and_then(|s| s.parse().ok()).unwrap_or_default())
.map(|s: Option<String>| s.and_then(|s| s.parse().ok()))
}
/// Returns 32-bit signed integer configuration value for the given key.
pub async fn get_config_int(&self, key: Config) -> Result<i32> {
Ok(self.get_config_parsed(key).await?.unwrap_or_default())
}
/// Returns 64-bit signed integer configuration value for the given key.
pub async fn get_config_i64(&self, key: Config) -> Result<i64> {
self.get_config(key)
.await
.map(|s: Option<String>| s.and_then(|s| s.parse().ok()).unwrap_or_default())
Ok(self.get_config_parsed(key).await?.unwrap_or_default())
}
/// Returns 64-bit unsigned integer configuration value for the given key.
pub async fn get_config_u64(&self, key: Config) -> Result<u64> {
self.get_config(key)
.await
.map(|s: Option<String>| s.and_then(|s| s.parse().ok()).unwrap_or_default())
Ok(self.get_config_parsed(key).await?.unwrap_or_default())
}
/// Returns boolean configuration value (if any) for the given key.
pub async fn get_config_bool_opt(&self, key: Config) -> Result<Option<bool>> {
Ok(self.get_config_parsed::<i32>(key).await?.map(|x| x != 0))
}
/// Returns boolean configuration value for the given key.
pub async fn get_config_bool(&self, key: Config) -> Result<bool> {
Ok(self.get_config_int(key).await? != 0)
Ok(self.get_config_bool_opt(key).await?.unwrap_or_default())
}
/// Returns true if movebox ("DeltaChat" folder) should be watched.
@@ -550,7 +566,6 @@ fn get_config_keys_string() -> String {
#[cfg(test)]
mod tests {
use std::str::FromStr;
use std::string::ToString;
use num_traits::FromPrimitive;

4
src/configure.rs
View File

@@ -249,7 +249,7 @@ async fn configure(ctx: &Context, param: &mut LoginParam) -> Result<()> {
}
}
},
strict_tls: Some(provider.strict_tls),
strict_tls: Some(provider.opt.strict_tls),
})
.collect();
@@ -338,7 +338,7 @@ async fn configure(ctx: &Context, param: &mut LoginParam) -> Result<()> {
.collect();
let provider_strict_tls = param
.provider
.map_or(socks5_config.is_some(), |provider| provider.strict_tls);
.map_or(socks5_config.is_some(), |provider| provider.opt.strict_tls);
let smtp_config_task = task::spawn(async move {
let mut smtp_configured = false;

2
src/constants.rs
View File

@@ -201,7 +201,7 @@ pub const BALANCED_IMAGE_SIZE: u32 = 1280;
pub const WORSE_IMAGE_SIZE: u32 = 640;
// this value can be increased if the folder configuration is changed and must be redone on next program start
pub(crate) const DC_FOLDERS_CONFIGURED_VERSION: i32 = 3;
pub(crate) const DC_FOLDERS_CONFIGURED_VERSION: i32 = 4;
#[cfg(test)]
mod tests {

74
src/contact.rs
View File

@@ -869,47 +869,45 @@ impl Contact {
Ok(ret)
}
// add blocked mailinglists as contacts
// to allow unblocking them as if they are contacts
// (this way, only one unblock-ffi is needed and only one set of ui-functions,
// from the users perspective,
// there is not much difference in an email- and a mailinglist-address)
/// Adds blocked mailinglists as contacts
/// to allow unblocking them as if they are contacts
/// (this way, only one unblock-ffi is needed and only one set of ui-functions,
/// from the users perspective,
/// there is not much difference in an email- and a mailinglist-address)
async fn update_blocked_mailinglist_contacts(context: &Context) -> Result<()> {
let blocked_mailinglists = context
context
.sql
.query_map(
"SELECT name, grpid FROM chats WHERE type=? AND blocked=?;",
paramsv![Chattype::Mailinglist, Blocked::Yes],
|row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
|rows| {
rows.collect::<std::result::Result<Vec<_>, _>>()
.map_err(Into::into)
},
)
.transaction(move |transaction| {
let mut stmt = transaction
.prepare("SELECT name, grpid FROM chats WHERE type=? AND blocked=?")?;
let rows = stmt.query_map(params![Chattype::Mailinglist, Blocked::Yes], |row| {
let name: String = row.get(0)?;
let grpid: String = row.get(1)?;
Ok((name, grpid))
})?;
let blocked_mailinglists = rows.collect::<std::result::Result<Vec<_>, _>>()?;
for (name, grpid) in blocked_mailinglists {
let count = transaction.query_row(
"SELECT COUNT(id) FROM contacts WHERE addr=?",
[&grpid],
|row| {
let count: isize = row.get(0)?;
Ok(count)
},
)?;
if count == 0 {
transaction.execute("INSERT INTO contacts (addr) VALUES (?)", [&grpid])?;
}
// Always do an update in case the blocking is reset or name is changed.
transaction.execute(
"UPDATE contacts SET name=?, origin=?, blocked=1 WHERE addr=?",
params![&name, Origin::MailinglistAddress, &grpid],
)?;
}
Ok(())
})
.await?;
for (name, grpid) in blocked_mailinglists {
if !context
.sql
.exists(
"SELECT COUNT(id) FROM contacts WHERE addr=?;",
paramsv![grpid],
)
.await?
{
context
.sql
.execute("INSERT INTO contacts (addr) VALUES (?);", paramsv![grpid])
.await?;
}
// always do an update in case the blocking is reset or name is changed
context
.sql
.execute(
"UPDATE contacts SET name=?, origin=?, blocked=1 WHERE addr=?;",
paramsv![name, Origin::MailinglistAddress, grpid],
)
.await?;
}
Ok(())
}

40
src/context.rs
View File

@@ -7,7 +7,7 @@ use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime};
use anyhow::{bail, ensure, Result};
use anyhow::{bail, ensure, Context as _, Result};
use async_channel::{self as channel, Receiver, Sender};
use ratelimit::Ratelimit;
use tokio::sync::{Mutex, RwLock};
@@ -655,6 +655,10 @@ impl Context {
.get_config(Config::ConfiguredMvboxFolder)
.await?
.unwrap_or_else(|| "<unset>".to_string());
let configured_trash_folder = self
.get_config(Config::ConfiguredTrashFolder)
.await?
.unwrap_or_else(|| "<unset>".to_string());
let mut res = get_info();
@@ -721,6 +725,7 @@ impl Context {
res.insert("configured_inbox_folder", configured_inbox_folder);
res.insert("configured_sentbox_folder", configured_sentbox_folder);
res.insert("configured_mvbox_folder", configured_mvbox_folder);
res.insert("configured_trash_folder", configured_trash_folder);
res.insert("mdns_enabled", mdns_enabled.to_string());
res.insert("e2ee_enabled", e2ee_enabled.to_string());
res.insert(
@@ -754,6 +759,12 @@ impl Context {
.await?
.to_string(),
);
res.insert(
"delete_to_trash",
self.get_config(Config::DeleteToTrash)
.await?
.unwrap_or_else(|| "<unset>".to_string()),
);
res.insert(
"last_housekeeping",
self.get_config_int(Config::LastHousekeeping)
@@ -919,6 +930,33 @@ impl Context {
Ok(mvbox.as_deref() == Some(folder_name))
}
/// Returns true if given folder name is the name of the trash folder.
pub async fn is_trash(&self, folder_name: &str) -> Result<bool> {
let trash = self.get_config(Config::ConfiguredTrashFolder).await?;
Ok(trash.as_deref() == Some(folder_name))
}
pub(crate) async fn should_delete_to_trash(&self) -> Result<bool> {
if let Some(v) = self.get_config_bool_opt(Config::DeleteToTrash).await? {
return Ok(v);
}
if let Some(provider) = self.get_configured_provider().await? {
return Ok(provider.opt.delete_to_trash);
}
Ok(false)
}
/// Returns `target` for deleted messages as per `imap` table. Empty string means "delete w/o
/// moving to trash".
pub(crate) async fn get_delete_msgs_target(&self) -> Result<String> {
if !self.should_delete_to_trash().await? {
return Ok("".into());
}
self.get_config(Config::ConfiguredTrashFolder)
.await?
.context("No configured trash folder")
}
pub(crate) fn derive_blobdir(dbfile: &Path) -> PathBuf {
let mut blob_fname = OsString::new();
blob_fname.push(dbfile.file_name().unwrap_or_default());

2
src/download.rs
View File

@@ -138,7 +138,7 @@ impl Job {
context
.sql
.query_row_optional(
"SELECT uid, folder FROM imap WHERE rfc724_mid=? AND target!=''",
"SELECT uid, folder FROM imap WHERE rfc724_mid=? AND target=folder",
paramsv![msg.rfc724_mid],
|row| {
let server_uid: u32 = row.get(0)?;

10
src/ephemeral.rs
View File

@@ -588,19 +588,25 @@ pub(crate) async fn delete_expired_imap_messages(context: &Context) -> Result<()
now - max(delete_server_after, MIN_DELETE_SERVER_AFTER),
),
};
let target = context.get_delete_msgs_target().await?;
context
.sql
.execute(
"UPDATE imap
SET target=''
SET target=?
WHERE rfc724_mid IN (
SELECT rfc724_mid FROM msgs
WHERE ((download_state = 0 AND timestamp < ?) OR
(download_state != 0 AND timestamp < ?) OR
(ephemeral_timestamp != 0 AND ephemeral_timestamp <= ?))
)",
paramsv![threshold_timestamp, threshold_timestamp_extended, now],
paramsv![
target,
threshold_timestamp,
threshold_timestamp_extended,
now,
],
)
.await?;

224
src/imap.rs
View File

@@ -113,13 +113,15 @@ impl async_imap::Authenticator for OAuth2 {
}
}
#[derive(Debug, PartialEq, Clone, Copy)]
enum FolderMeaning {
#[derive(Debug, Display, PartialEq, Eq, Clone, Copy)]
pub enum FolderMeaning {
Unknown,
Spam,
Inbox,
Mvbox,
Sent,
Trash,
Drafts,
Other,
/// Virtual folders.
///
@@ -131,13 +133,15 @@ enum FolderMeaning {
}
impl FolderMeaning {
fn to_config(self) -> Option<Config> {
pub fn to_config(self) -> Option<Config> {
match self {
FolderMeaning::Unknown => None,
FolderMeaning::Spam => None,
FolderMeaning::Inbox => Some(Config::ConfiguredInboxFolder),
FolderMeaning::Mvbox => Some(Config::ConfiguredMvboxFolder),
FolderMeaning::Sent => Some(Config::ConfiguredSentboxFolder),
FolderMeaning::Trash => Some(Config::ConfiguredTrashFolder),
FolderMeaning::Drafts => None,
FolderMeaning::Other => None,
FolderMeaning::Virtual => None,
}
}
@@ -270,7 +274,7 @@ impl Imap {
param
.provider
.map_or(param.socks5_config.is_some(), |provider| {
provider.strict_tls
provider.opt.strict_tls
}),
idle_interrupt_receiver,
)?;
@@ -449,7 +453,7 @@ impl Imap {
&mut self,
context: &Context,
watch_folder: &str,
is_spam_folder: bool,
folder_meaning: FolderMeaning,
) -> Result<()> {
if !context.sql.is_open().await {
// probably shutdown
@@ -458,7 +462,7 @@ impl Imap {
self.prepare(context).await?;
let msgs_fetched = self
.fetch_new_messages(context, watch_folder, is_spam_folder, false)
.fetch_new_messages(context, watch_folder, folder_meaning, false)
.await
.context("fetch_new_messages")?;
if msgs_fetched && context.get_config_delete_device_after().await?.is_some() {
@@ -490,49 +494,60 @@ impl Imap {
pub(crate) async fn resync_folder_uids(
&mut self,
context: &Context,
folder: String,
folder: &str,
folder_meaning: FolderMeaning,
) -> Result<()> {
// Collect pairs of UID and Message-ID.
let mut msg_ids = BTreeMap::new();
let mut msgs = BTreeMap::new();
let session = self
.session
.as_mut()
.context("IMAP No connection established")?;
session.select_folder(context, Some(&folder)).await?;
session.select_folder(context, Some(folder)).await?;
let mut list = session
.uid_fetch("1:*", RFC724MID_UID)
.await
.with_context(|| format!("can't resync folder {folder}"))?;
while let Some(fetch) = list.next().await {
let msg = fetch?;
let fetch = fetch?;
let headers = match get_fetch_headers(&fetch) {
Ok(headers) => headers,
Err(err) => {
warn!(context, "Failed to parse FETCH headers: {}", err);
continue;
}
};
let message_id = prefetch_get_message_id(&headers);
// Get Message-ID
let message_id =
get_fetch_headers(&msg).map_or(None, |headers| prefetch_get_message_id(&headers));
if let (Some(uid), Some(rfc724_mid)) = (msg.uid, message_id) {
msg_ids.insert(uid, rfc724_mid);
if let (Some(uid), Some(rfc724_mid)) = (fetch.uid, message_id) {
msgs.insert(
uid,
(
rfc724_mid,
target_folder(context, folder, folder_meaning, &headers).await?,
),
);
}
}
info!(
context,
"Resync: collected {} message IDs in folder {}",
msg_ids.len(),
&folder
msgs.len(),
folder,
);
let uid_validity = get_uidvalidity(context, &folder).await?;
let uid_validity = get_uidvalidity(context, folder).await?;
// Write collected UIDs to SQLite database.
context
.sql
.transaction(move |transaction| {
transaction.execute("DELETE FROM imap WHERE folder=?", params![folder])?;
for (uid, rfc724_mid) in &msg_ids {
for (uid, (rfc724_mid, target)) in &msgs {
// This may detect previously undetected moved
// messages, so we update server_folder too.
transaction.execute(
@@ -541,7 +556,7 @@ impl Imap {
ON CONFLICT(folder, uid, uidvalidity)
DO UPDATE SET rfc724_mid=excluded.rfc724_mid,
target=excluded.target",
params![rfc724_mid, folder, uid, uid_validity, folder],
params![rfc724_mid, folder, uid, uid_validity, target],
)?;
}
Ok(())
@@ -683,10 +698,10 @@ impl Imap {
&mut self,
context: &Context,
folder: &str,
is_spam_folder: bool,
folder_meaning: FolderMeaning,
fetch_existing_msgs: bool,
) -> Result<bool> {
if should_ignore_folder(context, folder, is_spam_folder).await? {
if should_ignore_folder(context, folder, folder_meaning).await? {
info!(context, "Not fetching from {}", folder);
return Ok(false);
}
@@ -713,8 +728,6 @@ impl Imap {
};
let read_cnt = msgs.len();
let show_emails = ShowEmails::from_i32(context.get_config_int(Config::ShowEmails).await?)
.unwrap_or_default();
let download_limit = context.download_limit().await?;
let mut uids_fetch = Vec::<(_, bool /* partially? */)>::with_capacity(msgs.len() + 1);
let mut uid_message_ids = BTreeMap::new();
@@ -732,14 +745,7 @@ impl Imap {
// Get the Message-ID or generate a fake one to identify the message in the database.
let message_id = prefetch_get_or_create_message_id(&headers);
let target = match target_folder(context, folder, is_spam_folder, &headers).await? {
Some(config) => match context.get_config(config).await? {
Some(target) => target,
None => folder.to_string(),
},
None => folder.to_string(),
};
let target = target_folder(context, folder, folder_meaning, &headers).await?;
context
.sql
@@ -763,14 +769,13 @@ impl Imap {
// Never download messages directly from the spam folder.
// If the sender is known, the message will be moved to the Inbox or Mvbox
// and then we download the message from there.
// Also see `spam_target_folder()`.
&& !is_spam_folder
// Also see `spam_target_folder_cfg()`.
&& folder_meaning != FolderMeaning::Spam
&& prefetch_should_download(
context,
&headers,
&message_id,
fetch_response.flags(),
show_emails,
)
.await.context("prefetch_should_download")?
{
@@ -870,17 +875,21 @@ impl Imap {
.context("failed to get recipients from the inbox")?;
if context.get_config_bool(Config::FetchExistingMsgs).await? {
for config in &[
Config::ConfiguredMvboxFolder,
Config::ConfiguredInboxFolder,
Config::ConfiguredSentboxFolder,
for meaning in [
FolderMeaning::Mvbox,
FolderMeaning::Inbox,
FolderMeaning::Sent,
] {
if let Some(folder) = context.get_config(*config).await? {
let config = match meaning.to_config() {
Some(c) => c,
None => continue,
};
if let Some(folder) = context.get_config(config).await? {
info!(
context,
"Fetching existing messages from folder \"{}\"", folder
);
self.fetch_new_messages(context, &folder, false, true)
self.fetch_new_messages(context, &folder, meaning, true)
.await
.context("could not fetch existing messages")?;
}
@@ -952,44 +961,60 @@ impl Session {
return Ok(());
}
Err(err) => {
if context.should_delete_to_trash().await? {
error!(
context,
"Cannot move messages {} to {}, no fallback to COPY/DELETE because \
delete_to_trash is set. Error: {:#}",
set,
target,
err,
);
return Err(err.into());
}
warn!(
context,
"Cannot move message, fallback to COPY/DELETE {} to {}: {}",
"Cannot move messages, fallback to COPY/DELETE {} to {}: {}",
set,
target,
err
);
}
}
} else {
}
// Server does not support MOVE or MOVE failed.
// Copy messages to the destination folder if needed and mark records for deletion.
let copy = !context.is_trash(target).await?;
if copy {
info!(
context,
"Server does not support MOVE, fallback to COPY/DELETE {} to {}", set, target
);
self.uid_copy(&set, &target).await?;
} else {
error!(
context,
"Server does not support MOVE, fallback to DELETE {} to {}", set, target,
);
}
// Server does not support MOVE or MOVE failed.
// Copy the message to the destination folder and mark the record for deletion.
match self.uid_copy(&set, &target).await {
Ok(()) => {
context
.sql
.execute(
&format!(
"UPDATE imap SET target='' WHERE id IN ({})",
sql::repeat_vars(row_ids.len())
),
rusqlite::params_from_iter(row_ids),
)
.await
.context("cannot plan deletion of copied messages")?;
context.emit_event(EventType::ImapMessageMoved(format!(
"IMAP messages {set} copied to {target}"
)));
Ok(())
}
Err(err) => Err(err.into()),
context
.sql
.execute(
&format!(
"UPDATE imap SET target='' WHERE id IN ({})",
sql::repeat_vars(row_ids.len())
),
rusqlite::params_from_iter(row_ids),
)
.await
.context("cannot plan deletion of messages")?;
if copy {
context.emit_event(EventType::ImapMessageMoved(format!(
"IMAP messages {set} copied to {target}"
)));
}
Ok(())
}
/// Moves and deletes messages as planned in the `imap` table.
@@ -1644,7 +1669,7 @@ impl Imap {
}
}
let folder_meaning = get_folder_meaning(&folder);
let folder_meaning = get_folder_meaning_by_attrs(folder.attributes());
let folder_name_meaning = get_folder_meaning_by_name(folder.name());
if let Some(config) = folder_meaning.to_config() {
// Always takes precedence
@@ -1776,7 +1801,7 @@ async fn should_move_out_of_spam(
/// If this returns None, the message will not be moved out of the
/// Spam folder, and as `fetch_new_messages()` doesn't download
/// messages from the Spam folder, the message will be ignored.
async fn spam_target_folder(
async fn spam_target_folder_cfg(
context: &Context,
headers: &[mailparse::MailHeader<'_>],
) -> Result<Option<Config>> {
@@ -1797,18 +1822,18 @@ async fn spam_target_folder(
/// Returns `ConfiguredInboxFolder`, `ConfiguredMvboxFolder` or `ConfiguredSentboxFolder` if
/// the message needs to be moved from `folder`. Otherwise returns `None`.
pub async fn target_folder(
pub async fn target_folder_cfg(
context: &Context,
folder: &str,
is_spam_folder: bool,
folder_meaning: FolderMeaning,
headers: &[mailparse::MailHeader<'_>],
) -> Result<Option<Config>> {
if context.is_mvbox(folder).await? {
return Ok(None);
}
if is_spam_folder {
spam_target_folder(context, headers).await
if folder_meaning == FolderMeaning::Spam {
spam_target_folder_cfg(context, headers).await
} else if needs_move_to_mvbox(context, headers).await? {
Ok(Some(Config::ConfiguredMvboxFolder))
} else {
@@ -1816,6 +1841,21 @@ pub async fn target_folder(
}
}
pub async fn target_folder(
context: &Context,
folder: &str,
folder_meaning: FolderMeaning,
headers: &[mailparse::MailHeader<'_>],
) -> Result<String> {
match target_folder_cfg(context, folder, folder_meaning, headers).await? {
Some(config) => match context.get_config(config).await? {
Some(target) => Ok(target),
None => Ok(folder.to_string()),
},
None => Ok(folder.to_string()),
}
}
async fn needs_move_to_mvbox(
context: &Context,
headers: &[mailparse::MailHeader<'_>],
@@ -1940,10 +1980,10 @@ fn get_folder_meaning_by_name(folder_name: &str) -> FolderMeaning {
}
}
fn get_folder_meaning(folder_name: &Name) -> FolderMeaning {
for attr in folder_name.attributes() {
fn get_folder_meaning_by_attrs(folder_attrs: &[NameAttribute]) -> FolderMeaning {
for attr in folder_attrs {
match attr {
NameAttribute::Trash => return FolderMeaning::Other,
NameAttribute::Trash => return FolderMeaning::Trash,
NameAttribute::Sent => return FolderMeaning::Sent,
NameAttribute::Junk => return FolderMeaning::Spam,
NameAttribute::Drafts => return FolderMeaning::Drafts,
@@ -1961,6 +2001,13 @@ fn get_folder_meaning(folder_name: &Name) -> FolderMeaning {
FolderMeaning::Unknown
}
pub(crate) fn get_folder_meaning(folder: &Name) -> FolderMeaning {
match get_folder_meaning_by_attrs(folder.attributes()) {
FolderMeaning::Unknown => get_folder_meaning_by_name(folder.name()),
meaning => meaning,
}
}
/// Parses the headers from the FETCH result.
fn get_fetch_headers(prefetch_msg: &Fetch) -> Result<Vec<mailparse::MailHeader>> {
match prefetch_msg.header() {
@@ -2005,7 +2052,6 @@ pub(crate) async fn prefetch_should_download(
headers: &[mailparse::MailHeader<'_>],
message_id: &str,
mut flags: impl Iterator<Item = Flag<'_>>,
show_emails: ShowEmails,
) -> Result<bool> {
if message::rfc724_mid_exists(context, message_id)
.await?
@@ -2065,6 +2111,9 @@ pub(crate) async fn prefetch_should_download(
})
.unwrap_or_default();
let show_emails =
ShowEmails::from_i32(context.get_config_int(Config::ShowEmails).await?).unwrap_or_default();
let show = is_autocrypt_setup_message
|| match show_emails {
ShowEmails::Off => is_chat_message || is_reply_to_chat_message,
@@ -2272,7 +2321,7 @@ pub async fn get_config_last_seen_uid(context: &Context, folder: &str) -> Result
async fn should_ignore_folder(
context: &Context,
folder: &str,
is_spam_folder: bool,
folder_meaning: FolderMeaning,
) -> Result<bool> {
if !context.get_config_bool(Config::OnlyFetchMvbox).await? {
return Ok(false);
@@ -2281,7 +2330,7 @@ async fn should_ignore_folder(
// Still respect the SentboxWatch setting.
return Ok(!context.get_config_bool(Config::SentboxWatch).await?);
}
Ok(!(context.is_mvbox(folder).await? || is_spam_folder))
Ok(!(context.is_mvbox(folder).await? || folder_meaning == FolderMeaning::Spam))
}
/// Builds a list of sequence/uid sets. The returned sets have each no more than around 1000
@@ -2564,14 +2613,13 @@ mod tests {
};
let (headers, _) = mailparse::parse_headers(bytes)?;
let is_spam_folder = folder == "Spam";
let actual =
if let Some(config) = target_folder(&t, folder, is_spam_folder, &headers).await? {
t.get_config(config).await?
} else {
None
};
let actual = if let Some(config) =
target_folder_cfg(&t, folder, get_folder_meaning_by_name(folder), &headers).await?
{
t.get_config(config).await?
} else {
None
};
let expected = if expected_destination == folder {
None

5
src/imap/idle.rs
View File

@@ -7,7 +7,7 @@ use futures_lite::FutureExt;
use super::session::Session;
use super::Imap;
use crate::imap::client::IMAP_TIMEOUT;
use crate::imap::{client::IMAP_TIMEOUT, FolderMeaning};
use crate::{context::Context, scheduler::InterruptInfo};
const IDLE_TIMEOUT: Duration = Duration::from_secs(23 * 60);
@@ -113,6 +113,7 @@ impl Imap {
&mut self,
context: &Context,
watch_folder: Option<String>,
folder_meaning: FolderMeaning,
) -> InterruptInfo {
// Idle using polling. This is also needed if we're not yet configured -
// in this case, we're waiting for a configure job (and an interrupt).
@@ -173,7 +174,7 @@ impl Imap {
// will have already fetched the messages so perform_*_fetch
// will not find any new.
match self
.fetch_new_messages(context, &watch_folder, false, false)
.fetch_new_messages(context, &watch_folder, folder_meaning, false)
.await
{
Ok(res) => {

39
src/imap/scan_folders.rs
View File

@@ -3,7 +3,7 @@ use std::{collections::BTreeMap, time::Instant};
use anyhow::{Context as _, Result};
use futures::stream::StreamExt;
use super::{get_folder_meaning, get_folder_meaning_by_name};
use super::{get_folder_meaning_by_attrs, get_folder_meaning_by_name};
use crate::config::Config;
use crate::imap::Imap;
use crate::log::LogExt;
@@ -33,7 +33,7 @@ impl Imap {
let mut folder_configs = BTreeMap::new();
for folder in folders {
let folder_meaning = get_folder_meaning(&folder);
let folder_meaning = get_folder_meaning_by_attrs(folder.attributes());
if folder_meaning == FolderMeaning::Virtual {
// Gmail has virtual folders that should be skipped. For example,
// emails appear in the inbox and under "All Mail" as soon as it is
@@ -53,21 +53,22 @@ impl Imap {
.or_insert_with(|| folder.name().to_string());
}
let is_drafts = folder_meaning == FolderMeaning::Drafts
|| (folder_meaning == FolderMeaning::Unknown
&& folder_name_meaning == FolderMeaning::Drafts);
let is_spam_folder = folder_meaning == FolderMeaning::Spam
|| (folder_meaning == FolderMeaning::Unknown
&& folder_name_meaning == FolderMeaning::Spam);
let folder_meaning = match folder_meaning {
FolderMeaning::Unknown => folder_name_meaning,
_ => folder_meaning,
};
// Don't scan folders that are watched anyway
if !watched_folders.contains(&folder.name().to_string()) && !is_drafts {
if !watched_folders.contains(&folder.name().to_string())
&& folder_meaning != FolderMeaning::Drafts
&& folder_meaning != FolderMeaning::Trash
{
let session = self.session.as_mut().context("no session")?;
// Drain leftover unsolicited EXISTS messages
session.server_sent_unsolicited_exists(context)?;
loop {
self.fetch_move_delete(context, folder.name(), is_spam_folder)
self.fetch_move_delete(context, folder.name(), folder_meaning)
.await
.ok_or_log_msg(context, "Can't fetch new msgs in scanned folder");
@@ -80,15 +81,15 @@ impl Imap {
}
}
// Set the `ConfiguredSentboxFolder` or set it to `None` if the folder was deleted.
context
.set_config(
Config::ConfiguredSentboxFolder,
folder_configs
.get(&Config::ConfiguredSentboxFolder)
.map(|s| s.as_str()),
)
.await?;
// Set configs for necessary folders. Or reset if the folder was deleted.
for conf in [
Config::ConfiguredSentboxFolder,
Config::ConfiguredTrashFolder,
] {
context
.set_config(conf, folder_configs.get(&conf).map(|s| s.as_str()))
.await?;
}
last_scan.replace(Instant::now());
Ok(true)

83
src/imex.rs
View File

@@ -431,8 +431,6 @@ async fn import_backup(
context.get_dbfile().display()
);
context.sql.config_cache.write().await.clear();
let mut archive = Archive::new(backup_file);
let mut entries = archive.entries()?;
@@ -760,29 +758,34 @@ async fn export_database(context: &Context, dest: &Path, passphrase: String) ->
context.sql.set_raw_config_int("backup_time", now).await?;
sql::housekeeping(context).await.ok_or_log(context);
let conn = context.sql.get_conn().await?;
tokio::task::block_in_place(move || {
conn.execute("VACUUM;", params![])
.map_err(|err| warn!(context, "Vacuum failed, exporting anyway {err}"))
.ok();
conn.execute(
"ATTACH DATABASE ? AS backup KEY ?",
paramsv![dest, passphrase],
)
.context("failed to attach backup database")?;
let res = conn
.query_row("SELECT sqlcipher_export('backup')", [], |_row| Ok(()))
.context("failed to export to attached backup database");
conn.execute("DETACH DATABASE backup", [])
.context("failed to detach backup database")?;
res?;
Ok(())
})
context
.sql
.call(|conn| {
conn.execute("VACUUM;", params![])
.map_err(|err| warn!(context, "Vacuum failed, exporting anyway {err}"))
.ok();
conn.execute(
"ATTACH DATABASE ? AS backup KEY ?",
paramsv![dest, passphrase],
)
.context("failed to attach backup database")?;
let res = conn
.query_row("SELECT sqlcipher_export('backup')", [], |_row| Ok(()))
.context("failed to export to attached backup database");
conn.execute("DETACH DATABASE backup", [])
.context("failed to detach backup database")?;
res?;
Ok(())
})
.await
}
#[cfg(test)]
mod tests {
use std::time::Duration;
use ::pgp::armor::BlockType;
use tokio::task;
use super::*;
use crate::pgp::{split_armored_data, HEADER_AUTOCRYPT, HEADER_SETUPCODE};
@@ -930,6 +933,46 @@ mod tests {
Ok(())
}
/// This is a regression test for
/// https://github.com/deltachat/deltachat-android/issues/2263
/// where the config cache wasn't reset properly after a backup.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_import_backup_reset_config_cache() -> Result<()> {
let backup_dir = tempfile::tempdir()?;
let context1 = TestContext::new_alice().await;
let context2 = TestContext::new().await;
assert!(!context2.is_configured().await?);
// export from context1
imex(&context1, ImexMode::ExportBackup, backup_dir.path(), None).await?;
// import to context2
let backup = has_backup(&context2, backup_dir.path()).await?;
let context2_cloned = context2.clone();
let handle = task::spawn(async move {
imex(
&context2_cloned,
ImexMode::ImportBackup,
backup.as_ref(),
None,
)
.await
.unwrap();
});
while !handle.is_finished() {
// The database is still unconfigured;
// fill the config cache with the old value.
context2.is_configured().await.ok();
tokio::time::sleep(Duration::from_micros(1)).await;
}
// Assert that the config cache has the new value now.
assert!(context2.is_configured().await?);
Ok(())
}
#[test]
fn test_normalize_setup_code() {
let norm = normalize_setup_code("123422343234423452346234723482349234");

8
src/job.rs
View File

@@ -12,7 +12,7 @@ use deltachat_derive::{FromSql, ToSql};
use rand::{thread_rng, Rng};
use crate::context::Context;
use crate::imap::Imap;
use crate::imap::{get_folder_meaning, FolderMeaning, Imap};
use crate::param::Params;
use crate::scheduler::InterruptInfo;
use crate::tools::time;
@@ -172,8 +172,12 @@ impl Job {
let mut any_failed = false;
for folder in all_folders {
let folder_meaning = get_folder_meaning(&folder);
if folder_meaning == FolderMeaning::Virtual {
continue;
}
if let Err(e) = imap
.resync_folder_uids(context, folder.name().to_string())
.resync_folder_uids(context, folder.name(), folder_meaning)
.await
{
warn!(context, "{:#}", e);

60
src/key.rs
View File

@@ -289,39 +289,41 @@ pub async fn store_self_keypair(
keypair: &KeyPair,
default: KeyPairUse,
) -> Result<()> {
let mut conn = context.sql.get_conn().await?;
let transaction = conn.transaction()?;
context
.sql
.transaction(|transaction| {
let public_key = DcKey::to_bytes(&keypair.public);
let secret_key = DcKey::to_bytes(&keypair.secret);
transaction
.execute(
"DELETE FROM keypairs WHERE public_key=? OR private_key=?;",
paramsv![public_key, secret_key],
)
.context("failed to remove old use of key")?;
if default == KeyPairUse::Default {
transaction
.execute("UPDATE keypairs SET is_default=0;", paramsv![])
.context("failed to clear default")?;
}
let is_default = match default {
KeyPairUse::Default => i32::from(true),
KeyPairUse::ReadOnly => i32::from(false),
};
let public_key = DcKey::to_bytes(&keypair.public);
let secret_key = DcKey::to_bytes(&keypair.secret);
transaction
.execute(
"DELETE FROM keypairs WHERE public_key=? OR private_key=?;",
paramsv![public_key, secret_key],
)
.context("failed to remove old use of key")?;
if default == KeyPairUse::Default {
transaction
.execute("UPDATE keypairs SET is_default=0;", paramsv![])
.context("failed to clear default")?;
}
let is_default = match default {
KeyPairUse::Default => i32::from(true),
KeyPairUse::ReadOnly => i32::from(false),
};
let addr = keypair.addr.to_string();
let t = time();
let addr = keypair.addr.to_string();
let t = time();
transaction
.execute(
"INSERT INTO keypairs (addr, is_default, public_key, private_key, created)
transaction
.execute(
"INSERT INTO keypairs (addr, is_default, public_key, private_key, created)
VALUES (?,?,?,?,?);",
paramsv![addr, is_default, public_key, secret_key, t],
)
.context("failed to insert keypair")?;
paramsv![addr, is_default, public_key, secret_key, t],
)
.context("failed to insert keypair")?;
transaction.commit()?;
Ok(())
})
.await?;
Ok(())
}

52
src/location.rs
View File

@@ -601,32 +601,38 @@ pub(crate) async fn save(
..
} = location;
let conn = context.sql.get_conn().await?;
let mut stmt_test =
conn.prepare_cached("SELECT id FROM locations WHERE timestamp=? AND from_id=?")?;
let mut stmt_insert = conn.prepare_cached(stmt_insert)?;
context
.sql
.call(|conn| {
let mut stmt_test = conn
.prepare_cached("SELECT id FROM locations WHERE timestamp=? AND from_id=?")?;
let mut stmt_insert = conn.prepare_cached(stmt_insert)?;
let exists = stmt_test.exists(paramsv![timestamp, contact_id])?;
let exists = stmt_test.exists(paramsv![timestamp, contact_id])?;
if independent || !exists {
stmt_insert.execute(paramsv![
timestamp,
contact_id,
chat_id,
latitude,
longitude,
accuracy,
independent,
])?;
if independent || !exists {
stmt_insert.execute(paramsv![
timestamp,
contact_id,
chat_id,
latitude,
longitude,
accuracy,
independent,
])?;
if timestamp > newest_timestamp {
// okay to drop, as we use cached prepared statements
drop(stmt_test);
drop(stmt_insert);
newest_timestamp = timestamp;
newest_location_id = Some(u32::try_from(conn.last_insert_rowid())?);
}
}
if timestamp > newest_timestamp {
// okay to drop, as we use cached prepared statements
drop(stmt_test);
drop(stmt_insert);
newest_timestamp = timestamp;
newest_location_id = Some(u32::try_from(conn.last_insert_rowid())?);
}
}
Ok(())
})
.await?;
}
Ok(newest_location_id)

5
src/message.rs
View File

@@ -1386,11 +1386,12 @@ pub async fn delete_msgs(context: &Context, msg_ids: &[MsgId]) -> Result<()> {
context.emit_event(EventType::WebxdcInstanceDeleted { msg_id: *msg_id });
}
let target = context.get_delete_msgs_target().await?;
context
.sql
.execute(
"UPDATE imap SET target='' WHERE rfc724_mid=?",
paramsv![msg.rfc724_mid],
"UPDATE imap SET target=? WHERE rfc724_mid=?",
paramsv![target, msg.rfc724_mid],
)
.await?;

2
src/peerstate.rs
View File

@@ -186,7 +186,7 @@ impl Peerstate {
async fn from_stmt(
context: &Context,
query: &str,
params: impl rusqlite::Params,
params: impl rusqlite::Params + Send,
) -> Result<Option<Peerstate>> {
let peerstate = context
.sql

24
src/provider.rs
View File

@@ -129,6 +129,16 @@ pub struct Provider {
/// Default configuration values to set when provider is configured.
pub config_defaults: Option<Vec<ConfigDefault>>,
/// Type of OAuth 2 authorization if provider supports it.
pub oauth2_authorizer: Option<Oauth2Authorizer>,
/// Options with good defaults.
pub opt: ProviderOptions,
}
/// Provider options with good defaults.
#[derive(Debug, PartialEq, Eq)]
pub struct ProviderOptions {
/// True if provider is known to use use proper,
/// not self-signed certificates.
pub strict_tls: bool,
@@ -136,8 +146,18 @@ pub struct Provider {
/// Maximum number of recipients the provider allows to send a single email to.
pub max_smtp_rcpt_to: Option<u16>,
/// Type of OAuth 2 authorization if provider supports it.
pub oauth2_authorizer: Option<Oauth2Authorizer>,
/// Move messages to the Trash folder instead of marking them "\Deleted".
pub delete_to_trash: bool,
}
impl Default for ProviderOptions {
fn default() -> Self {
Self {
strict_tls: true,
max_smtp_rcpt_to: None,
delete_to_trash: false,
}
}
}
/// Get resolver to query MX records.

238
src/provider/data.rs
View File

@@ -1,13 +1,14 @@
// file generated by src/provider/update.py
use std::collections::HashMap;
use once_cell::sync::Lazy;
use crate::provider::Protocol::*;
use crate::provider::Socket::*;
use crate::provider::UsernamePattern::*;
use crate::provider::{Config, ConfigDefault, Oauth2Authorizer, Provider, Server, Status};
use crate::provider::{
Config, ConfigDefault, Oauth2Authorizer, Provider, ProviderOptions, Server, Status,
};
use std::collections::HashMap;
use once_cell::sync::Lazy;
// 163.md: 163.com
static P_163: Lazy<Provider> = Lazy::new(|| Provider {
@@ -32,9 +33,8 @@ static P_163: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -61,9 +61,8 @@ static P_AKTIVIX_ORG: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -79,9 +78,8 @@ static P_AOL: Lazy<Provider> = Lazy::new(|| {
Server { protocol: Imap, socket: Ssl, hostname: "imap.aol.com", port: 993, username_pattern: Email },
Server { protocol: Smtp, socket: Ssl, hostname: "smtp.aol.com", port: 465, username_pattern: Email },
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
}
});
@@ -109,9 +107,8 @@ static P_ARCOR_DE: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -138,9 +135,8 @@ static P_AUTISTICI_ORG: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -167,9 +163,8 @@ static P_BLINDZELN_ORG: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -196,9 +191,8 @@ static P_BLUEWIN_CH: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -225,9 +219,8 @@ static P_BUZON_UY: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -254,9 +247,8 @@ static P_CHELLO_AT: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -268,9 +260,8 @@ static P_COMCAST: Lazy<Provider> = Lazy::new(|| Provider {
after_login_hint: "",
overview_page: "https://providers.delta.chat/comcast",
server: vec![],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -282,9 +273,8 @@ static P_DISMAIL_DE: Lazy<Provider> = Lazy::new(|| Provider {
after_login_hint: "",
overview_page: "https://providers.delta.chat/dismail-de",
server: vec![],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -311,9 +301,8 @@ static P_DISROOT: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -340,9 +329,8 @@ static P_E_EMAIL: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -354,9 +342,8 @@ static P_ESPIV_NET: Lazy<Provider> = Lazy::new(|| Provider {
after_login_hint: "",
overview_page: "https://providers.delta.chat/espiv-net",
server: vec![],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -372,9 +359,8 @@ static P_EXAMPLE_COM: Lazy<Provider> = Lazy::new(|| {
Server { protocol: Imap, socket: Ssl, hostname: "imap.example.com", port: 1337, username_pattern: Email },
Server { protocol: Smtp, socket: Starttls, hostname: "smtp.example.com", port: 1337, username_pattern: Email },
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
}
});
@@ -403,9 +389,8 @@ static P_FASTMAIL: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -419,9 +404,8 @@ static P_FIREMAIL_DE: Lazy<Provider> = Lazy::new(|| {
overview_page: "https://providers.delta.chat/firemail-de",
server: vec![
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
}
});
@@ -434,6 +418,7 @@ static P_FIVE_CHAT: Lazy<Provider> = Lazy::new(|| Provider {
after_login_hint: "",
overview_page: "https://providers.delta.chat/five-chat",
server: vec![],
opt: Default::default(),
config_defaults: Some(vec![
ConfigDefault {
key: Config::BccSelf,
@@ -448,8 +433,6 @@ static P_FIVE_CHAT: Lazy<Provider> = Lazy::new(|| Provider {
value: "0",
},
]),
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -465,9 +448,8 @@ static P_FREENET_DE: Lazy<Provider> = Lazy::new(|| {
Server { protocol: Imap, socket: Ssl, hostname: "mx.freenet.de", port: 993, username_pattern: Email },
Server { protocol: Smtp, socket: Starttls, hostname: "mx.freenet.de", port: 587, username_pattern: Email },
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
}
});
@@ -484,9 +466,11 @@ static P_GMAIL: Lazy<Provider> = Lazy::new(|| {
Server { protocol: Imap, socket: Ssl, hostname: "imap.gmail.com", port: 993, username_pattern: Email },
Server { protocol: Smtp, socket: Ssl, hostname: "smtp.gmail.com", port: 465, username_pattern: Email },
],
opt: ProviderOptions {
delete_to_trash: true,
..Default::default()
},
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: Some(Oauth2Authorizer::Gmail),
}
});
@@ -521,9 +505,8 @@ static P_GMX_NET: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -535,6 +518,10 @@ static P_HERMES_RADIO: Lazy<Provider> = Lazy::new(|| Provider {
after_login_hint: "",
overview_page: "https://providers.delta.chat/hermes-radio",
server: vec![],
opt: ProviderOptions {
strict_tls: false,
..Default::default()
},
config_defaults: Some(vec![
ConfigDefault {
key: Config::MdnsEnabled,
@@ -549,8 +536,6 @@ static P_HERMES_RADIO: Lazy<Provider> = Lazy::new(|| Provider {
value: "2",
},
]),
strict_tls: false,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -564,9 +549,8 @@ static P_HEY_COM: Lazy<Provider> = Lazy::new(|| {
overview_page: "https://providers.delta.chat/hey-com",
server: vec![
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
}
});
@@ -579,9 +563,8 @@ static P_I_UA: Lazy<Provider> = Lazy::new(|| Provider {
after_login_hint: "",
overview_page: "https://providers.delta.chat/i-ua",
server: vec![],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -593,9 +576,8 @@ static P_I3_NET: Lazy<Provider> = Lazy::new(|| Provider {
after_login_hint: "",
overview_page: "https://providers.delta.chat/i3-net",
server: vec![],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -622,9 +604,8 @@ static P_ICLOUD: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -651,9 +632,11 @@ static P_INFOMANIAK_COM: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: ProviderOptions {
max_smtp_rcpt_to: Some(10),
..Default::default()
},
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: Some(10),
oauth2_authorizer: None,
});
@@ -665,9 +648,8 @@ static P_KOLST_COM: Lazy<Provider> = Lazy::new(|| Provider {
after_login_hint: "",
overview_page: "https://providers.delta.chat/kolst-com",
server: vec![],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -679,9 +661,8 @@ static P_KONTENT_COM: Lazy<Provider> = Lazy::new(|| Provider {
after_login_hint: "",
overview_page: "https://providers.delta.chat/kontent-com",
server: vec![],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -708,9 +689,8 @@ static P_MAIL_DE: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -726,9 +706,8 @@ static P_MAIL_RU: Lazy<Provider> = Lazy::new(|| {
Server { protocol: Imap, socket: Ssl, hostname: "imap.mail.ru", port: 993, username_pattern: Email },
Server { protocol: Smtp, socket: Ssl, hostname: "smtp.mail.ru", port: 465, username_pattern: Email },
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
}
});
@@ -756,9 +735,8 @@ static P_MAIL2TOR: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -785,9 +763,8 @@ static P_MAILBOX_ORG: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -814,9 +791,8 @@ static P_MAILO_COM: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -843,6 +819,11 @@ static P_NAUTA_CU: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: ProviderOptions {
max_smtp_rcpt_to: Some(20),
strict_tls: false,
..Default::default()
},
config_defaults: Some(vec![
ConfigDefault {
key: Config::DeleteServerAfter,
@@ -869,8 +850,6 @@ static P_NAUTA_CU: Lazy<Provider> = Lazy::new(|| Provider {
value: "0",
},
]),
strict_tls: false,
max_smtp_rcpt_to: Some(20),
oauth2_authorizer: None,
});
@@ -897,9 +876,8 @@ static P_NAVER: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -926,9 +904,8 @@ static P_NUBO_COOP: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -955,9 +932,8 @@ static P_OUTLOOK_COM: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -984,9 +960,8 @@ static P_OUVATON_COOP: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -998,6 +973,13 @@ static P_POSTEO: Lazy<Provider> = Lazy::new(|| Provider {
after_login_hint: "",
overview_page: "https://providers.delta.chat/posteo",
server: vec![
Server {
protocol: Imap,
socket: Ssl,
hostname: "posteo.de",
port: 993,
username_pattern: Email,
},
Server {
protocol: Imap,
socket: Starttls,
@@ -1005,6 +987,13 @@ static P_POSTEO: Lazy<Provider> = Lazy::new(|| Provider {
port: 143,
username_pattern: Email,
},
Server {
protocol: Smtp,
socket: Ssl,
hostname: "posteo.de",
port: 465,
username_pattern: Email,
},
Server {
protocol: Smtp,
socket: Starttls,
@@ -1013,9 +1002,8 @@ static P_POSTEO: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -1029,9 +1017,8 @@ static P_PROTONMAIL: Lazy<Provider> = Lazy::new(|| {
overview_page: "https://providers.delta.chat/protonmail",
server: vec![
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
}
});
@@ -1048,9 +1035,8 @@ static P_QQ: Lazy<Provider> = Lazy::new(|| {
Server { protocol: Imap, socket: Ssl, hostname: "imap.qq.com", port: 993, username_pattern: Emaillocalpart },
Server { protocol: Smtp, socket: Ssl, hostname: "smtp.qq.com", port: 465, username_pattern: Email },
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
}
});
@@ -1078,9 +1064,8 @@ static P_RISEUP_NET: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -1092,9 +1077,21 @@ static P_ROGERS_COM: Lazy<Provider> = Lazy::new(|| Provider {
after_login_hint: "",
overview_page: "https://providers.delta.chat/rogers-com",
server: vec![],
opt: Default::default(),
config_defaults: None,
oauth2_authorizer: None,
});
// sonic.md: sonic.net
static P_SONIC: Lazy<Provider> = Lazy::new(|| Provider {
id: "sonic",
status: Status::Ok,
before_login_hint: "",
after_login_hint: "",
overview_page: "https://providers.delta.chat/sonic",
server: vec![],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -1121,9 +1118,8 @@ static P_SYSTEMAUSFALL_ORG: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -1150,9 +1146,8 @@ static P_SYSTEMLI_ORG: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -1168,9 +1163,8 @@ static P_T_ONLINE: Lazy<Provider> = Lazy::new(|| {
Server { protocol: Imap, socket: Ssl, hostname: "secureimap.t-online.de", port: 993, username_pattern: Email },
Server { protocol: Smtp, socket: Ssl, hostname: "securesmtp.t-online.de", port: 465, username_pattern: Email },
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
}
});
@@ -1205,6 +1199,7 @@ static P_TESTRUN: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: Some(vec![
ConfigDefault {
key: Config::BccSelf,
@@ -1219,8 +1214,6 @@ static P_TESTRUN: Lazy<Provider> = Lazy::new(|| Provider {
value: "0",
},
]),
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -1247,9 +1240,8 @@ static P_TISCALI_IT: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -1263,9 +1255,8 @@ static P_TUTANOTA: Lazy<Provider> = Lazy::new(|| {
overview_page: "https://providers.delta.chat/tutanota",
server: vec![
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
}
});
@@ -1278,9 +1269,8 @@ static P_UKR_NET: Lazy<Provider> = Lazy::new(|| Provider {
after_login_hint: "",
overview_page: "https://providers.delta.chat/ukr-net",
server: vec![],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -1307,9 +1297,8 @@ static P_UNDERNET_UY: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -1321,9 +1310,8 @@ static P_VFEMAIL: Lazy<Provider> = Lazy::new(|| Provider {
after_login_hint: "",
overview_page: "https://providers.delta.chat/vfemail",
server: vec![],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -1350,9 +1338,8 @@ static P_VIVALDI: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -1379,9 +1366,8 @@ static P_VODAFONE_DE: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -1398,9 +1384,8 @@ static P_WEB_DE: Lazy<Provider> = Lazy::new(|| {
Server { protocol: Imap, socket: Starttls, hostname: "imap.web.de", port: 143, username_pattern: Emaillocalpart },
Server { protocol: Smtp, socket: Starttls, hostname: "smtp.web.de", port: 587, username_pattern: Emaillocalpart },
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
}
});
@@ -1417,9 +1402,8 @@ static P_YAHOO: Lazy<Provider> = Lazy::new(|| {
Server { protocol: Imap, socket: Ssl, hostname: "imap.mail.yahoo.com", port: 993, username_pattern: Email },
Server { protocol: Smtp, socket: Ssl, hostname: "smtp.mail.yahoo.com", port: 465, username_pattern: Email },
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
}
});
@@ -1447,9 +1431,8 @@ static P_YANDEX_RU: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: Some(Oauth2Authorizer::Yandex),
});
@@ -1465,11 +1448,10 @@ static P_YGGMAIL: Lazy<Provider> = Lazy::new(|| {
Server { protocol: Imap, socket: Plain, hostname: "localhost", port: 1143, username_pattern: Email },
Server { protocol: Smtp, socket: Plain, hostname: "localhost", port: 1025, username_pattern: Email },
],
opt: Default::default(),
config_defaults: Some(vec![
ConfigDefault { key: Config::MvboxMove, value: "0" },
]),
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
}
});
@@ -1497,9 +1479,8 @@ static P_ZIGGO_NL: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -1526,9 +1507,8 @@ static P_ZOHO: Lazy<Provider> = Lazy::new(|| Provider {
username_pattern: Email,
},
],
opt: Default::default(),
config_defaults: None,
strict_tls: true,
max_smtp_rcpt_to: None,
oauth2_authorizer: None,
});
@@ -1822,6 +1802,7 @@ pub(crate) static PROVIDER_DATA: Lazy<HashMap<&'static str, &'static Provider>>
("foxmail.com", &*P_QQ),
("riseup.net", &*P_RISEUP_NET),
("rogers.com", &*P_ROGERS_COM),
("sonic.net", &*P_SONIC),
("systemausfall.org", &*P_SYSTEMAUSFALL_ORG),
("solidaris.me", &*P_SYSTEMAUSFALL_ORG),
("systemli.org", &*P_SYSTEMLI_ORG),
@@ -1946,6 +1927,7 @@ pub(crate) static PROVIDER_IDS: Lazy<HashMap<&'static str, &'static Provider>> =
("qq", &*P_QQ),
("riseup.net", &*P_RISEUP_NET),
("rogers.com", &*P_ROGERS_COM),
("sonic", &*P_SONIC),
("systemausfall.org", &*P_SYSTEMAUSFALL_ORG),
("systemli.org", &*P_SYSTEMLI_ORG),
("t-online", &*P_T_ONLINE),
@@ -1970,4 +1952,4 @@ pub(crate) static PROVIDER_IDS: Lazy<HashMap<&'static str, &'static Provider>> =
});
pub static PROVIDER_UPDATED: Lazy<chrono::NaiveDate> =
Lazy::new(|| chrono::NaiveDate::from_ymd_opt(2023, 1, 6).unwrap());
Lazy::new(|| chrono::NaiveDate::from_ymd_opt(2023, 2, 21).unwrap());

31
src/provider/update.py
View File

@@ -40,6 +40,23 @@ def file2url(f):
return "https://providers.delta.chat/" + f
def process_opt(data):
if not "opt" in data:
return "Default::default()"
opt = "ProviderOptions {\n"
opt_data = data.get("opt", "")
for key in opt_data:
value = str(opt_data[key])
if key == "max_smtp_rcpt_to":
value = "Some(" + value + ")"
if value in {"True", "False"}:
value = value.lower()
opt += " " + key + ": " + value + ",\n"
opt += " ..Default::default()\n"
opt += " }"
return opt
def process_config_defaults(data):
if not "config_defaults" in data:
return "None"
@@ -106,14 +123,9 @@ def process_data(data, file):
server += (" Server { protocol: " + protocol.capitalize() + ", socket: " + socket.capitalize() + ", hostname: \""
+ hostname + "\", port: " + str(port) + ", username_pattern: " + username_pattern.capitalize() + " },\n")
opt = process_opt(data)
config_defaults = process_config_defaults(data)
strict_tls = data.get("strict_tls", True)
strict_tls = "true" if strict_tls else "false"
max_smtp_rcpt_to = data.get("max_smtp_rcpt_to", 0)
max_smtp_rcpt_to = "Some(" + str(max_smtp_rcpt_to) + ")" if max_smtp_rcpt_to != 0 else "None"
oauth2 = data.get("oauth2", "")
oauth2 = "Some(Oauth2Authorizer::" + camel(oauth2) + ")" if oauth2 != "" else "None"
@@ -128,9 +140,8 @@ def process_data(data, file):
provider += " after_login_hint: \"" + after_login_hint + "\",\n"
provider += " overview_page: \"" + file2url(file) + "\",\n"
provider += " server: vec![\n" + server + " ],\n"
provider += " opt: " + opt + ",\n"
provider += " config_defaults: " + config_defaults + ",\n"
provider += " strict_tls: " + strict_tls + ",\n"
provider += " max_smtp_rcpt_to: " + max_smtp_rcpt_to + ",\n"
provider += " oauth2_authorizer: " + oauth2 + ",\n"
provider += "});\n\n"
else:
@@ -174,7 +185,9 @@ if __name__ == "__main__":
"use crate::provider::Protocol::*;\n"
"use crate::provider::Socket::*;\n"
"use crate::provider::UsernamePattern::*;\n"
"use crate::provider::{Config, ConfigDefault, Oauth2Authorizer, Provider, Server, Status};\n"
"use crate::provider::{\n"
" Config, ConfigDefault, Oauth2Authorizer, Provider, ProviderOptions, Server, Status,\n"
"};\n"
"use std::collections::HashMap;\n\n"
"use once_cell::sync::Lazy;\n\n")

78
src/receive_imf.rs
View File

@@ -342,11 +342,12 @@ pub(crate) async fn receive_imf_inner(
if received_msg.needs_delete_job
|| (delete_server_after == Some(0) && is_partial_download.is_none())
{
let target = context.get_delete_msgs_target().await?;
context
.sql
.execute(
"UPDATE imap SET target='' WHERE rfc724_mid=?",
paramsv![rfc724_mid],
"UPDATE imap SET target=? WHERE rfc724_mid=?",
paramsv![target, rfc724_mid],
)
.await?;
} else if !mime_parser.mdn_reports.is_empty() && mime_parser.has_chat_version() {
@@ -1084,8 +1085,6 @@ async fn add_parts(
let mut created_db_entries = Vec::with_capacity(mime_parser.parts.len());
let conn = context.sql.get_conn().await?;
for part in &mime_parser.parts {
if part.is_reaction {
set_msg_reaction(
@@ -1117,39 +1116,6 @@ async fn add_parts(
}
let mut txt_raw = "".to_string();
let mut stmt = conn.prepare_cached(
r#"
INSERT INTO msgs
(
id,
rfc724_mid, chat_id,
from_id, to_id, timestamp, timestamp_sent,
timestamp_rcvd, type, state, msgrmsg,
txt, subject, txt_raw, param,
bytes, mime_headers, mime_in_reply_to,
mime_references, mime_modified, error, ephemeral_timer,
ephemeral_timestamp, download_state, hop_info
)
VALUES (
?,
?, ?, ?, ?,
?, ?, ?, ?,
?, ?, ?, ?,
?, ?, ?, ?,
?, ?, ?, ?,
?, ?, ?, ?
)
ON CONFLICT (id) DO UPDATE
SET rfc724_mid=excluded.rfc724_mid, chat_id=excluded.chat_id,
from_id=excluded.from_id, to_id=excluded.to_id, timestamp=excluded.timestamp, timestamp_sent=excluded.timestamp_sent,
timestamp_rcvd=excluded.timestamp_rcvd, type=excluded.type, state=excluded.state, msgrmsg=excluded.msgrmsg,
txt=excluded.txt, subject=excluded.subject, txt_raw=excluded.txt_raw, param=excluded.param,
bytes=excluded.bytes, mime_headers=excluded.mime_headers, mime_in_reply_to=excluded.mime_in_reply_to,
mime_references=excluded.mime_references, mime_modified=excluded.mime_modified, error=excluded.error, ephemeral_timer=excluded.ephemeral_timer,
ephemeral_timestamp=excluded.ephemeral_timestamp, download_state=excluded.download_state, hop_info=excluded.hop_info
"#,
)?;
let (msg, typ): (&str, Viewtype) = if let Some(better_msg) = &better_msg {
(better_msg, Viewtype::Text)
} else {
@@ -1183,7 +1149,38 @@ SET rfc724_mid=excluded.rfc724_mid, chat_id=excluded.chat_id,
// also change `MsgId::trash()` and `delete_expired_messages()`
let trash = chat_id.is_trash() || (is_location_kml && msg.is_empty());
stmt.execute(paramsv![
let row_id = context.sql.insert(
r#"
INSERT INTO msgs
(
id,
rfc724_mid, chat_id,
from_id, to_id, timestamp, timestamp_sent,
timestamp_rcvd, type, state, msgrmsg,
txt, subject, txt_raw, param,
bytes, mime_headers, mime_in_reply_to,
mime_references, mime_modified, error, ephemeral_timer,
ephemeral_timestamp, download_state, hop_info
)
VALUES (
?,
?, ?, ?, ?,
?, ?, ?, ?,
?, ?, ?, ?,
?, ?, ?, ?,
?, ?, ?, ?,
?, ?, ?, ?
)
ON CONFLICT (id) DO UPDATE
SET rfc724_mid=excluded.rfc724_mid, chat_id=excluded.chat_id,
from_id=excluded.from_id, to_id=excluded.to_id, timestamp=excluded.timestamp, timestamp_sent=excluded.timestamp_sent,
timestamp_rcvd=excluded.timestamp_rcvd, type=excluded.type, state=excluded.state, msgrmsg=excluded.msgrmsg,
txt=excluded.txt, subject=excluded.subject, txt_raw=excluded.txt_raw, param=excluded.param,
bytes=excluded.bytes, mime_headers=excluded.mime_headers, mime_in_reply_to=excluded.mime_in_reply_to,
mime_references=excluded.mime_references, mime_modified=excluded.mime_modified, error=excluded.error, ephemeral_timer=excluded.ephemeral_timer,
ephemeral_timestamp=excluded.ephemeral_timestamp, download_state=excluded.download_state, hop_info=excluded.hop_info
"#,
paramsv![
replace_msg_id,
rfc724_mid,
if trash { DC_CHAT_ID_TRASH } else { chat_id },
@@ -1222,17 +1219,14 @@ SET rfc724_mid=excluded.rfc724_mid, chat_id=excluded.chat_id,
DownloadState::Done
},
mime_parser.hop_info
])?;
]).await?;
// We only replace placeholder with a first part,
// afterwards insert additional parts.
replace_msg_id = None;
let row_id = conn.last_insert_rowid();
drop(stmt);
created_db_entries.push(MsgId::new(u32::try_from(row_id)?));
}
drop(conn);
// check all parts whether they contain a new logging webxdc
for (part, msg_id) in mime_parser.parts.iter().zip(&created_db_entries) {

16
src/receive_imf/tests.rs
View File

@@ -5,7 +5,7 @@ use crate::aheader::EncryptPreference;
use crate::chat::get_chat_contacts;
use crate::chat::{get_chat_msgs, ChatItem, ChatVisibility};
use crate::chatlist::Chatlist;
use crate::constants::{ShowEmails, DC_GCL_NO_SPECIALS};
use crate::constants::DC_GCL_NO_SPECIALS;
use crate::imap::prefetch_should_download;
use crate::message::Message;
use crate::test_utils::{get_chat_msg, TestContext, TestContextManager};
@@ -647,15 +647,11 @@ async fn test_parse_ndn(
// Check that the ndn would be downloaded:
let headers = mailparse::parse_mail(raw_ndn).unwrap().headers;
assert!(prefetch_should_download(
&t,
&headers,
"some-other-message-id",
std::iter::empty(),
ShowEmails::Off,
)
.await
.unwrap());
assert!(
prefetch_should_download(&t, &headers, "some-other-message-id", std::iter::empty(),)
.await
.unwrap()
);
receive_imf(&t, raw_ndn, false).await.unwrap();
let msg = Message::load_from_db(&t, msg_id).await.unwrap();

217
src/scheduler.rs
View File

@@ -1,6 +1,8 @@
use std::iter::{self, once};
use anyhow::{bail, Context as _, Result};
use async_channel::{self as channel, Receiver, Sender};
use futures::try_join;
use futures::future::try_join_all;
use futures_lite::FutureExt;
use tokio::task;
@@ -9,7 +11,7 @@ use crate::config::Config;
use crate::contact::{ContactId, RecentlySeenLoop};
use crate::context::Context;
use crate::ephemeral::{self, delete_expired_imap_messages};
use crate::imap::Imap;
use crate::imap::{FolderMeaning, Imap};
use crate::job;
use crate::location;
use crate::log::LogExt;
@@ -20,15 +22,19 @@ use crate::tools::{duration_to_str, maybe_add_time_based_warnings};
pub(crate) mod connectivity;
#[derive(Debug)]
struct SchedBox {
meaning: FolderMeaning,
conn_state: ImapConnectionState,
handle: task::JoinHandle<()>,
}
/// Job and connection scheduler.
#[derive(Debug)]
pub(crate) struct Scheduler {
inbox: ImapConnectionState,
inbox_handle: task::JoinHandle<()>,
mvbox: ImapConnectionState,
mvbox_handle: Option<task::JoinHandle<()>>,
sentbox: ImapConnectionState,
sentbox_handle: Option<task::JoinHandle<()>>,
inbox: SchedBox,
/// Optional boxes -- mvbox, sentbox.
oboxes: Vec<SchedBox>,
smtp: SmtpConnectionState,
smtp_handle: task::JoinHandle<()>,
ephemeral_handle: task::JoinHandle<()>,
@@ -161,7 +167,7 @@ async fn inbox_loop(ctx: Context, started: Sender<()>, inbox_handlers: ImapConne
}
}
info = fetch_idle(&ctx, &mut connection, Config::ConfiguredInboxFolder).await;
info = fetch_idle(&ctx, &mut connection, FolderMeaning::Inbox).await;
}
}
}
@@ -182,7 +188,20 @@ async fn inbox_loop(ctx: Context, started: Sender<()>, inbox_handlers: ImapConne
/// handling all the errors. In case of an error, it is logged, but not propagated upwards. If
/// critical operation fails such as fetching new messages fails, connection is reset via
/// `trigger_reconnect`, so a fresh one can be opened.
async fn fetch_idle(ctx: &Context, connection: &mut Imap, folder_config: Config) -> InterruptInfo {
async fn fetch_idle(
ctx: &Context,
connection: &mut Imap,
folder_meaning: FolderMeaning,
) -> InterruptInfo {
let folder_config = match folder_meaning.to_config() {
Some(c) => c,
None => {
error!(ctx, "Bad folder meaning: {}", folder_meaning);
return connection
.fake_idle(ctx, None, FolderMeaning::Unknown)
.await;
}
};
let folder = match ctx.get_config(folder_config).await {
Ok(folder) => folder,
Err(err) => {
@@ -190,7 +209,9 @@ async fn fetch_idle(ctx: &Context, connection: &mut Imap, folder_config: Config)
ctx,
"Can not watch {} folder, failed to retrieve config: {:#}", folder_config, err
);
return connection.fake_idle(ctx, None).await;
return connection
.fake_idle(ctx, None, FolderMeaning::Unknown)
.await;
}
};
@@ -199,7 +220,9 @@ async fn fetch_idle(ctx: &Context, connection: &mut Imap, folder_config: Config)
} else {
connection.connectivity.set_not_configured(ctx).await;
info!(ctx, "Can not watch {} folder, not set", folder_config);
return connection.fake_idle(ctx, None).await;
return connection
.fake_idle(ctx, None, FolderMeaning::Unknown)
.await;
};
// connect and fake idle if unable to connect
@@ -210,7 +233,9 @@ async fn fetch_idle(ctx: &Context, connection: &mut Imap, folder_config: Config)
{
warn!(ctx, "{:#}", err);
connection.trigger_reconnect(ctx);
return connection.fake_idle(ctx, Some(watch_folder)).await;
return connection
.fake_idle(ctx, Some(watch_folder), folder_meaning)
.await;
}
if folder_config == Config::ConfiguredInboxFolder {
@@ -227,7 +252,7 @@ async fn fetch_idle(ctx: &Context, connection: &mut Imap, folder_config: Config)
// Fetch the watched folder.
if let Err(err) = connection
.fetch_move_delete(ctx, &watch_folder, false)
.fetch_move_delete(ctx, &watch_folder, folder_meaning)
.await
.context("fetch_move_delete")
{
@@ -265,7 +290,7 @@ async fn fetch_idle(ctx: &Context, connection: &mut Imap, folder_config: Config)
// no new messages. We want to select the watched folder anyway before going IDLE
// there, so this does not take additional protocol round-trip.
if let Err(err) = connection
.fetch_move_delete(ctx, &watch_folder, false)
.fetch_move_delete(ctx, &watch_folder, folder_meaning)
.await
.context("fetch_move_delete after scan_folders")
{
@@ -293,7 +318,9 @@ async fn fetch_idle(ctx: &Context, connection: &mut Imap, folder_config: Config)
ctx,
"IMAP session does not support IDLE, going to fake idle."
);
return connection.fake_idle(ctx, Some(watch_folder)).await;
return connection
.fake_idle(ctx, Some(watch_folder), folder_meaning)
.await;
}
info!(ctx, "IMAP session supports IDLE, using it.");
@@ -318,7 +345,9 @@ async fn fetch_idle(ctx: &Context, connection: &mut Imap, folder_config: Config)
}
} else {
warn!(ctx, "No IMAP session, going to fake idle.");
connection.fake_idle(ctx, Some(watch_folder)).await
connection
.fake_idle(ctx, Some(watch_folder), folder_meaning)
.await
}
}
@@ -326,11 +355,11 @@ async fn simple_imap_loop(
ctx: Context,
started: Sender<()>,
inbox_handlers: ImapConnectionHandlers,
folder_config: Config,
folder_meaning: FolderMeaning,
) {
use futures::future::FutureExt;
info!(ctx, "starting simple loop for {}", folder_config);
info!(ctx, "starting simple loop for {}", folder_meaning);
let ImapConnectionHandlers {
mut connection,
stop_receiver,
@@ -346,7 +375,7 @@ async fn simple_imap_loop(
}
loop {
fetch_idle(&ctx, &mut connection, folder_config).await;
fetch_idle(&ctx, &mut connection, folder_meaning).await;
}
};
@@ -443,75 +472,56 @@ async fn smtp_loop(ctx: Context, started: Sender<()>, smtp_handlers: SmtpConnect
impl Scheduler {
/// Start the scheduler.
pub async fn start(ctx: Context) -> Result<Self> {
let (mvbox, mvbox_handlers) = ImapConnectionState::new(&ctx).await?;
let (sentbox, sentbox_handlers) = ImapConnectionState::new(&ctx).await?;
let (smtp, smtp_handlers) = SmtpConnectionState::new();
let (inbox, inbox_handlers) = ImapConnectionState::new(&ctx).await?;
let (inbox_start_send, inbox_start_recv) = channel::bounded(1);
let (mvbox_start_send, mvbox_start_recv) = channel::bounded(1);
let mut mvbox_handle = None;
let (sentbox_start_send, sentbox_start_recv) = channel::bounded(1);
let mut sentbox_handle = None;
let (smtp_start_send, smtp_start_recv) = channel::bounded(1);
let (ephemeral_interrupt_send, ephemeral_interrupt_recv) = channel::bounded(1);
let (location_interrupt_send, location_interrupt_recv) = channel::bounded(1);
let inbox_handle = {
let mut oboxes = Vec::new();
let mut start_recvs = Vec::new();
let (conn_state, inbox_handlers) = ImapConnectionState::new(&ctx).await?;
let (inbox_start_send, inbox_start_recv) = channel::bounded(1);
let handle = {
let ctx = ctx.clone();
task::spawn(async move { inbox_loop(ctx, inbox_start_send, inbox_handlers).await })
};
let inbox = SchedBox {
meaning: FolderMeaning::Inbox,
conn_state,
handle,
};
start_recvs.push(inbox_start_recv);
if ctx.should_watch_mvbox().await? {
let ctx = ctx.clone();
mvbox_handle = Some(task::spawn(async move {
simple_imap_loop(
ctx,
mvbox_start_send,
mvbox_handlers,
Config::ConfiguredMvboxFolder,
)
.await
}));
} else {
mvbox_start_send
.send(())
.await
.context("mvbox start send, missing receiver")?;
mvbox_handlers
.connection
.connectivity
.set_not_configured(&ctx)
.await
}
if ctx.get_config_bool(Config::SentboxWatch).await? {
let ctx = ctx.clone();
sentbox_handle = Some(task::spawn(async move {
simple_imap_loop(
ctx,
sentbox_start_send,
sentbox_handlers,
Config::ConfiguredSentboxFolder,
)
.await
}));
} else {
sentbox_start_send
.send(())
.await
.context("sentbox start send, missing receiver")?;
sentbox_handlers
.connection
.connectivity
.set_not_configured(&ctx)
.await
for (meaning, should_watch) in [
(FolderMeaning::Mvbox, ctx.should_watch_mvbox().await),
(
FolderMeaning::Sent,
ctx.get_config_bool(Config::SentboxWatch).await,
),
] {
if should_watch? {
let (conn_state, handlers) = ImapConnectionState::new(&ctx).await?;
let (start_send, start_recv) = channel::bounded(1);
let ctx = ctx.clone();
let handle = task::spawn(async move {
simple_imap_loop(ctx, start_send, handlers, meaning).await
});
oboxes.push(SchedBox {
meaning,
conn_state,
handle,
});
start_recvs.push(start_recv);
}
}
let smtp_handle = {
let ctx = ctx.clone();
task::spawn(async move { smtp_loop(ctx, smtp_start_send, smtp_handlers).await })
};
start_recvs.push(smtp_start_recv);
let ephemeral_handle = {
let ctx = ctx.clone();
@@ -531,12 +541,8 @@ impl Scheduler {
let res = Self {
inbox,
mvbox,
sentbox,
oboxes,
smtp,
inbox_handle,
mvbox_handle,
sentbox_handle,
smtp_handle,
ephemeral_handle,
ephemeral_interrupt_send,
@@ -546,12 +552,7 @@ impl Scheduler {
};
// wait for all loops to be started
if let Err(err) = try_join!(
inbox_start_recv.recv(),
mvbox_start_recv.recv(),
sentbox_start_recv.recv(),
smtp_start_recv.recv()
) {
if let Err(err) = try_join_all(start_recvs.iter().map(|r| r.recv())).await {
bail!("failed to start scheduler: {}", err);
}
@@ -559,30 +560,26 @@ impl Scheduler {
Ok(res)
}
fn boxes(&self) -> iter::Chain<iter::Once<&SchedBox>, std::slice::Iter<'_, SchedBox>> {
once(&self.inbox).chain(self.oboxes.iter())
}
fn maybe_network(&self) {
self.interrupt_inbox(InterruptInfo::new(true));
self.interrupt_mvbox(InterruptInfo::new(true));
self.interrupt_sentbox(InterruptInfo::new(true));
for b in self.boxes() {
b.conn_state.interrupt(InterruptInfo::new(true));
}
self.interrupt_smtp(InterruptInfo::new(true));
}
fn maybe_network_lost(&self) {
self.interrupt_inbox(InterruptInfo::new(false));
self.interrupt_mvbox(InterruptInfo::new(false));
self.interrupt_sentbox(InterruptInfo::new(false));
for b in self.boxes() {
b.conn_state.interrupt(InterruptInfo::new(false));
}
self.interrupt_smtp(InterruptInfo::new(false));
}
fn interrupt_inbox(&self, info: InterruptInfo) {
self.inbox.interrupt(info);
}
fn interrupt_mvbox(&self, info: InterruptInfo) {
self.mvbox.interrupt(info);
}
fn interrupt_sentbox(&self, info: InterruptInfo) {
self.sentbox.interrupt(info);
self.inbox.conn_state.interrupt(info);
}
fn interrupt_smtp(&self, info: InterruptInfo) {
@@ -605,29 +602,17 @@ impl Scheduler {
///
/// It consumes the scheduler and never fails to stop it. In the worst case, long-running tasks
/// are forcefully terminated if they cannot shutdown within the timeout.
pub(crate) async fn stop(mut self, context: &Context) {
pub(crate) async fn stop(self, context: &Context) {
// Send stop signals to tasks so they can shutdown cleanly.
self.inbox.stop().await.ok_or_log(context);
if self.mvbox_handle.is_some() {
self.mvbox.stop().await.ok_or_log(context);
}
if self.sentbox_handle.is_some() {
self.sentbox.stop().await.ok_or_log(context);
for b in self.boxes() {
b.conn_state.stop().await.ok_or_log(context);
}
self.smtp.stop().await.ok_or_log(context);
// Actually shutdown tasks.
let timeout_duration = std::time::Duration::from_secs(30);
tokio::time::timeout(timeout_duration, self.inbox_handle)
.await
.ok_or_log(context);
if let Some(mvbox_handle) = self.mvbox_handle.take() {
tokio::time::timeout(timeout_duration, mvbox_handle)
.await
.ok_or_log(context);
}
if let Some(sentbox_handle) = self.sentbox_handle.take() {
tokio::time::timeout(timeout_duration, sentbox_handle)
for b in once(self.inbox).chain(self.oboxes.into_iter()) {
tokio::time::timeout(timeout_duration, b.handle)
.await
.ok_or_log(context);
}

100
src/scheduler/connectivity.rs
View File

@@ -1,18 +1,18 @@
use core::fmt;
use std::{ops::Deref, sync::Arc};
use std::{iter::once, ops::Deref, sync::Arc};
use anyhow::{anyhow, Result};
use humansize::{format_size, BINARY};
use tokio::sync::{Mutex, RwLockReadGuard};
use crate::events::EventType;
use crate::imap::scan_folders::get_watched_folder_configs;
use crate::imap::{scan_folders::get_watched_folder_configs, FolderMeaning};
use crate::quota::{
QUOTA_ERROR_THRESHOLD_PERCENTAGE, QUOTA_MAX_AGE_SECONDS, QUOTA_WARN_THRESHOLD_PERCENTAGE,
};
use crate::tools::time;
use crate::{config::Config, scheduler::Scheduler, stock_str, tools};
use crate::{context::Context, log::LogExt};
use crate::{scheduler::Scheduler, stock_str, tools};
#[derive(Debug, Clone, Copy, PartialEq, Eq, EnumProperty, PartialOrd, Ord)]
pub enum Connectivity {
@@ -157,17 +157,14 @@ impl ConnectivityStore {
/// Called during `dc_maybe_network()` to make sure that `dc_accounts_all_work_done()`
/// returns false immediately after `dc_maybe_network()`.
pub(crate) async fn idle_interrupted(scheduler: RwLockReadGuard<'_, Option<Scheduler>>) {
let [inbox, mvbox, sentbox] = match &*scheduler {
Some(Scheduler {
inbox,
mvbox,
sentbox,
..
}) => [
inbox.state.connectivity.clone(),
mvbox.state.connectivity.clone(),
sentbox.state.connectivity.clone(),
],
let (inbox, oboxes) = match &*scheduler {
Some(Scheduler { inbox, oboxes, .. }) => (
inbox.conn_state.state.connectivity.clone(),
oboxes
.iter()
.map(|b| b.conn_state.state.connectivity.clone())
.collect::<Vec<_>>(),
),
None => return,
};
drop(scheduler);
@@ -185,7 +182,7 @@ pub(crate) async fn idle_interrupted(scheduler: RwLockReadGuard<'_, Option<Sched
}
drop(connectivity_lock);
for state in &[&mvbox, &sentbox] {
for state in oboxes {
let mut connectivity_lock = state.0.lock().await;
if *connectivity_lock == DetailedConnectivity::Connected {
*connectivity_lock = DetailedConnectivity::InterruptingIdle;
@@ -202,17 +199,11 @@ pub(crate) async fn maybe_network_lost(
context: &Context,
scheduler: RwLockReadGuard<'_, Option<Scheduler>>,
) {
let stores = match &*scheduler {
Some(Scheduler {
inbox,
mvbox,
sentbox,
..
}) => [
inbox.state.connectivity.clone(),
mvbox.state.connectivity.clone(),
sentbox.state.connectivity.clone(),
],
let stores: Vec<_> = match &*scheduler {
Some(sched) => sched
.boxes()
.map(|b| b.conn_state.state.connectivity.clone())
.collect(),
None => return,
};
drop(scheduler);
@@ -260,14 +251,9 @@ impl Context {
pub async fn get_connectivity(&self) -> Connectivity {
let lock = self.scheduler.read().await;
let stores: Vec<_> = match &*lock {
Some(Scheduler {
inbox,
mvbox,
sentbox,
..
}) => [&inbox.state, &mvbox.state, &sentbox.state]
.iter()
.map(|state| state.connectivity.clone())
Some(sched) => sched
.boxes()
.map(|b| b.conn_state.state.connectivity.clone())
.collect(),
None => return Connectivity::NotConnected,
};
@@ -348,28 +334,12 @@ impl Context {
let lock = self.scheduler.read().await;
let (folders_states, smtp) = match &*lock {
Some(Scheduler {
inbox,
mvbox,
sentbox,
smtp,
..
}) => (
[
(
Config::ConfiguredInboxFolder,
inbox.state.connectivity.clone(),
),
(
Config::ConfiguredMvboxFolder,
mvbox.state.connectivity.clone(),
),
(
Config::ConfiguredSentboxFolder,
sentbox.state.connectivity.clone(),
),
],
smtp.state.connectivity.clone(),
Some(sched) => (
sched
.boxes()
.map(|b| (b.meaning, b.conn_state.state.connectivity.clone()))
.collect::<Vec<_>>(),
sched.smtp.state.connectivity.clone(),
),
None => {
return Err(anyhow!("Not started"));
@@ -390,8 +360,8 @@ impl Context {
for (folder, state) in &folders_states {
let mut folder_added = false;
if watched_folders.contains(folder) {
let f = self.get_config(*folder).await.ok_or_log(self).flatten();
if let Some(config) = folder.to_config().filter(|c| watched_folders.contains(c)) {
let f = self.get_config(config).await.ok_or_log(self).flatten();
if let Some(foldername) = f {
let detailed = &state.get_detailed().await;
@@ -407,7 +377,7 @@ impl Context {
}
}
if !folder_added && folder == &Config::ConfiguredInboxFolder {
if !folder_added && folder == &FolderMeaning::Inbox {
let detailed = &state.get_detailed().await;
if let DetailedConnectivity::Error(_) = detailed {
// On the inbox thread, we also do some other things like scan_folders and run jobs
@@ -535,14 +505,10 @@ impl Context {
pub async fn all_work_done(&self) -> bool {
let lock = self.scheduler.read().await;
let stores: Vec<_> = match &*lock {
Some(Scheduler {
inbox,
mvbox,
sentbox,
smtp,
..
}) => [&inbox.state, &mvbox.state, &sentbox.state, &smtp.state]
.iter()
Some(sched) => sched
.boxes()
.map(|b| &b.conn_state.state)
.chain(once(&sched.smtp.state))
.map(|state| state.connectivity.clone())
.collect(),
None => return false,

5
src/smtp.rs
View File

@@ -101,8 +101,9 @@ impl Smtp {
&lp.smtp,
&lp.socks5_config,
&lp.addr,
lp.provider
.map_or(lp.socks5_config.is_some(), |provider| provider.strict_tls),
lp.provider.map_or(lp.socks5_config.is_some(), |provider| {
provider.opt.strict_tls
}),
)
.await
}

2
src/smtp/send.rs
View File

@@ -47,7 +47,7 @@ impl Smtp {
let chunk_size = context
.get_configured_provider()
.await?
.and_then(|provider| provider.max_smtp_rcpt_to)
.and_then(|provider| provider.opt.max_smtp_rcpt_to)
.map_or(DEFAULT_MAX_SMTP_RCPT_TO, usize::from);
for recipients_chunk in recipients.chunks(chunk_size) {

199
src/sql.rs
View File

@@ -2,8 +2,7 @@
use std::collections::{HashMap, HashSet};
use std::convert::TryFrom;
use std::path::Path;
use std::path::PathBuf;
use std::path::{Path, PathBuf};
use anyhow::{bail, Context as _, Result};
use rusqlite::{self, config::DbConfig, Connection, OpenFlags, TransactionBehavior};
@@ -49,7 +48,7 @@ pub(crate) fn params_iter(iter: &[impl crate::ToSql]) -> impl Iterator<Item = &d
mod migrations;
mod pool;
use pool::{Pool, PooledConnection};
use pool::Pool;
/// A wrapper around the underlying Sqlite3 object.
#[derive(Debug)]
@@ -128,45 +127,51 @@ impl Sql {
pub(crate) async fn import(&self, path: &Path, passphrase: String) -> Result<()> {
let path_str = path
.to_str()
.with_context(|| format!("path {path:?} is not valid unicode"))?;
let conn = self.get_conn().await?;
.with_context(|| format!("path {path:?} is not valid unicode"))?
.to_string();
let res = self
.call(move |conn| {
// Check that backup passphrase is correct before resetting our database.
conn.execute(
"ATTACH DATABASE ? AS backup KEY ?",
paramsv![path_str, passphrase],
)
.context("failed to attach backup database")?;
if let Err(err) = conn
.query_row("SELECT count(*) FROM sqlite_master", [], |_row| Ok(()))
.context("backup passphrase is not correct")
{
conn.execute("DETACH DATABASE backup", [])
.context("failed to detach backup database")?;
return Err(err);
}
tokio::task::block_in_place(move || {
// Check that backup passphrase is correct before resetting our database.
conn.execute(
"ATTACH DATABASE ? AS backup KEY ?",
paramsv![path_str, passphrase],
)
.context("failed to attach backup database")?;
if let Err(err) = conn
.query_row("SELECT count(*) FROM sqlite_master", [], |_row| Ok(()))
.context("backup passphrase is not correct")
{
// Reset the database without reopening it. We don't want to reopen the database because we
// don't have main database passphrase at this point.
// See <https://sqlite.org/c3ref/c_dbconfig_enable_fkey.html> for documentation.
// Without resetting import may fail due to existing tables.
conn.set_db_config(DbConfig::SQLITE_DBCONFIG_RESET_DATABASE, true)
.context("failed to set SQLITE_DBCONFIG_RESET_DATABASE")?;
conn.execute("VACUUM", [])
.context("failed to vacuum the database")?;
conn.set_db_config(DbConfig::SQLITE_DBCONFIG_RESET_DATABASE, false)
.context("failed to unset SQLITE_DBCONFIG_RESET_DATABASE")?;
let res = conn
.query_row("SELECT sqlcipher_export('main', 'backup')", [], |_row| {
Ok(())
})
.context("failed to import from attached backup database");
conn.execute("DETACH DATABASE backup", [])
.context("failed to detach backup database")?;
return Err(err);
}
res?;
Ok(())
})
.await;
// Reset the database without reopening it. We don't want to reopen the database because we
// don't have main database passphrase at this point.
// See <https://sqlite.org/c3ref/c_dbconfig_enable_fkey.html> for documentation.
// Without resetting import may fail due to existing tables.
conn.set_db_config(DbConfig::SQLITE_DBCONFIG_RESET_DATABASE, true)
.context("failed to set SQLITE_DBCONFIG_RESET_DATABASE")?;
conn.execute("VACUUM", [])
.context("failed to vacuum the database")?;
conn.set_db_config(DbConfig::SQLITE_DBCONFIG_RESET_DATABASE, false)
.context("failed to unset SQLITE_DBCONFIG_RESET_DATABASE")?;
let res = conn
.query_row("SELECT sqlcipher_export('main', 'backup')", [], |_row| {
Ok(())
})
.context("failed to import from attached backup database");
conn.execute("DETACH DATABASE backup", [])
.context("failed to detach backup database")?;
res?;
Ok(())
})
// The config cache is wrong now that we have a different database
self.config_cache.write().await.clear();
res
}
/// Creates a new connection pool.
@@ -294,22 +299,41 @@ impl Sql {
}
}
/// Allocates a connection and calls given function with the connection.
///
/// Returns the result of the function.
pub async fn call<'a, F, R>(&'a self, function: F) -> Result<R>
where
F: 'a + FnOnce(&mut Connection) -> Result<R> + Send,
R: Send + 'static,
{
let lock = self.pool.read().await;
let pool = lock.as_ref().context("no SQL connection")?;
let mut conn = pool.get().await?;
let res = tokio::task::block_in_place(move || function(&mut conn))?;
Ok(res)
}
/// Execute the given query, returning the number of affected rows.
pub async fn execute(&self, query: &str, params: impl rusqlite::Params) -> Result<usize> {
let conn = self.get_conn().await?;
tokio::task::block_in_place(move || {
pub async fn execute(
&self,
query: &str,
params: impl rusqlite::Params + Send,
) -> Result<usize> {
self.call(move |conn| {
let res = conn.execute(query, params)?;
Ok(res)
})
.await
}
/// Executes the given query, returning the last inserted row ID.
pub async fn insert(&self, query: &str, params: impl rusqlite::Params) -> Result<i64> {
let conn = self.get_conn().await?;
tokio::task::block_in_place(move || {
pub async fn insert(&self, query: &str, params: impl rusqlite::Params + Send) -> Result<i64> {
self.call(move |conn| {
conn.execute(query, params)?;
Ok(conn.last_insert_rowid())
})
.await
}
/// Prepares and executes the statement and maps a function over the resulting rows.
@@ -318,40 +342,32 @@ impl Sql {
pub async fn query_map<T, F, G, H>(
&self,
sql: &str,
params: impl rusqlite::Params,
params: impl rusqlite::Params + Send,
f: F,
mut g: G,
) -> Result<H>
where
F: FnMut(&rusqlite::Row) -> rusqlite::Result<T>,
G: FnMut(rusqlite::MappedRows<F>) -> Result<H>,
F: Send + FnMut(&rusqlite::Row) -> rusqlite::Result<T>,
G: Send + FnMut(rusqlite::MappedRows<F>) -> Result<H>,
H: Send + 'static,
{
let conn = self.get_conn().await?;
tokio::task::block_in_place(move || {
self.call(move |conn| {
let mut stmt = conn.prepare(sql)?;
let res = stmt.query_map(params, f)?;
g(res)
})
}
/// Allocates a connection from the connection pool and returns it.
pub(crate) async fn get_conn(&self) -> Result<PooledConnection> {
let lock = self.pool.read().await;
let pool = lock.as_ref().context("no SQL connection")?;
let conn = pool.get().await?;
Ok(conn)
.await
}
/// Used for executing `SELECT COUNT` statements only. Returns the resulting count.
pub async fn count(&self, query: &str, params: impl rusqlite::Params) -> Result<usize> {
pub async fn count(&self, query: &str, params: impl rusqlite::Params + Send) -> Result<usize> {
let count: isize = self.query_row(query, params, |row| row.get(0)).await?;
Ok(usize::try_from(count)?)
}
/// Used for executing `SELECT COUNT` statements only. Returns `true`, if the count is at least
/// one, `false` otherwise.
pub async fn exists(&self, sql: &str, params: impl rusqlite::Params) -> Result<bool> {
pub async fn exists(&self, sql: &str, params: impl rusqlite::Params + Send) -> Result<bool> {
let count = self.count(sql, params).await?;
Ok(count > 0)
}
@@ -360,17 +376,18 @@ impl Sql {
pub async fn query_row<T, F>(
&self,
query: &str,
params: impl rusqlite::Params,
params: impl rusqlite::Params + Send,
f: F,
) -> Result<T>
where
F: FnOnce(&rusqlite::Row) -> rusqlite::Result<T>,
F: FnOnce(&rusqlite::Row) -> rusqlite::Result<T> + Send,
T: Send + 'static,
{
let conn = self.get_conn().await?;
tokio::task::block_in_place(move || {
self.call(move |conn| {
let res = conn.query_row(query, params, f)?;
Ok(res)
})
.await
}
/// Execute the function inside a transaction.
@@ -388,8 +405,7 @@ impl Sql {
H: Send + 'static,
G: Send + FnOnce(&mut rusqlite::Transaction<'_>) -> Result<H>,
{
let mut conn = self.get_conn().await?;
tokio::task::block_in_place(move || {
self.call(move |conn| {
let mut transaction = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
let ret = callback(&mut transaction);
@@ -404,12 +420,12 @@ impl Sql {
}
}
})
.await
}
/// Query the database if the requested table already exists.
pub async fn table_exists(&self, name: &str) -> Result<bool> {
let conn = self.get_conn().await?;
tokio::task::block_in_place(move || {
self.call(move |conn| {
let mut exists = false;
conn.pragma(None, "table_info", name.to_string(), |_row| {
// will only be executed if the info was found
@@ -419,12 +435,12 @@ impl Sql {
Ok(exists)
})
.await
}
/// Check if a column exists in a given table.
pub async fn col_exists(&self, table_name: &str, col_name: &str) -> Result<bool> {
let conn = self.get_conn().await?;
tokio::task::block_in_place(move || {
self.call(move |conn| {
let mut exists = false;
// `PRAGMA table_info` returns one row per column,
// each row containing 0=cid, 1=name, 2=type, 3=notnull, 4=dflt_value
@@ -438,29 +454,27 @@ impl Sql {
Ok(exists)
})
.await
}
/// Execute a query which is expected to return zero or one row.
pub async fn query_row_optional<T, F>(
&self,
sql: &str,
params: impl rusqlite::Params,
params: impl rusqlite::Params + Send,
f: F,
) -> Result<Option<T>>
where
F: FnOnce(&rusqlite::Row) -> rusqlite::Result<T>,
F: Send + FnOnce(&rusqlite::Row) -> rusqlite::Result<T>,
T: Send + 'static,
{
let conn = self.get_conn().await?;
let res =
tokio::task::block_in_place(move || match conn.query_row(sql.as_ref(), params, f) {
Ok(res) => Ok(Some(res)),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(rusqlite::Error::InvalidColumnType(_, _, rusqlite::types::Type::Null)) => {
Ok(None)
}
Err(err) => Err(err),
})?;
Ok(res)
self.call(move |conn| match conn.query_row(sql.as_ref(), params, f) {
Ok(res) => Ok(Some(res)),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(rusqlite::Error::InvalidColumnType(_, _, rusqlite::types::Type::Null)) => Ok(None),
Err(err) => Err(err.into()),
})
.await
}
/// Executes a query which is expected to return one row and one
@@ -469,10 +483,10 @@ impl Sql {
pub async fn query_get_value<T>(
&self,
query: &str,
params: impl rusqlite::Params,
params: impl rusqlite::Params + Send,
) -> Result<Option<T>>
where
T: rusqlite::types::FromSql,
T: rusqlite::types::FromSql + Send + 'static,
{
self.query_row_optional(query, params, |row| row.get::<_, T>(0))
.await
@@ -935,11 +949,16 @@ mod tests {
async fn test_auto_vacuum() -> Result<()> {
let t = TestContext::new().await;
let conn = t.sql.get_conn().await?;
let auto_vacuum = conn.pragma_query_value(None, "auto_vacuum", |row| {
let auto_vacuum: i32 = row.get(0)?;
Ok(auto_vacuum)
})?;
let auto_vacuum = t
.sql
.call(|conn| {
let auto_vacuum = conn.pragma_query_value(None, "auto_vacuum", |row| {
let auto_vacuum: i32 = row.get(0)?;
Ok(auto_vacuum)
})?;
Ok(auto_vacuum)
})
.await?;
// auto_vacuum=2 is the same as auto_vacuum=INCREMENTAL
assert_eq!(auto_vacuum, 2);

31
src/sql/pool.rs
View File

@@ -1,10 +1,18 @@
//! Connection pool.
//! # SQLite connection pool.
//!
//! The connection pool holds a number of SQLite connections and allows to allocate them.
//! When allocated connection is dropped, underlying connection is returned back to the pool.
//!
//! The pool is organized as a stack. It always allocates the most recently used connection.
//! Each SQLite connection has its own page cache, so allocating recently used connections
//! improves the performance compared to, for example, organizing the pool as a queue
//! and returning the least recently used connection each time.
use std::ops::{Deref, DerefMut};
use std::sync::{Arc, Weak};
use anyhow::{Context, Result};
use crossbeam_queue::ArrayQueue;
use parking_lot::Mutex;
use rusqlite::Connection;
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
@@ -12,7 +20,7 @@ use tokio::sync::{OwnedSemaphorePermit, Semaphore};
#[derive(Debug)]
struct InnerPool {
/// Available connections.
connections: ArrayQueue<Connection>,
connections: Mutex<Vec<Connection>>,
/// Counts the number of available connections.
semaphore: Arc<Semaphore>,
@@ -23,7 +31,9 @@ impl InnerPool {
///
/// The connection could be new or returned back.
fn put(&self, connection: Connection) {
self.connections.force_push(connection);
let mut connections = self.connections.lock();
connections.push(connection);
drop(connections);
}
}
@@ -74,22 +84,19 @@ pub struct Pool {
impl Pool {
/// Creates a new connection pool.
pub fn new(connections: Vec<Connection>) -> Self {
let semaphore = Arc::new(Semaphore::new(connections.len()));
let inner = Arc::new(InnerPool {
connections: ArrayQueue::new(connections.len()),
semaphore: Arc::new(Semaphore::new(connections.len())),
connections: Mutex::new(connections),
semaphore,
});
for connection in connections {
inner.connections.force_push(connection);
}
Pool { inner }
}
/// Retrieves a connection from the pool.
pub async fn get(&self) -> Result<PooledConnection> {
let permit = self.inner.semaphore.clone().acquire_owned().await?;
let conn = self
.inner
.connections
let mut connections = self.inner.connections.lock();
let conn = connections
.pop()
.context("got a permit when there are no connections in the pool")?;
let conn = PooledConnection {