diff --git a/rust/src/api/messages/incoming/client2client/messages.rs b/rust/src/api/messages/incoming/client2client/messages.rs index 63993400..b48eaa96 100644 --- a/rust/src/api/messages/incoming/client2client/messages.rs +++ b/rust/src/api/messages/incoming/client2client/messages.rs @@ -26,12 +26,12 @@ static ALREADY_QUEUED_RECEIPTS: LazyLock, + t: &mut Transaction<'_, Sqlite>, target_user_id: i64, content: proto::EncryptedContent, contact_will_send_receipt: bool, ) -> Result { - let Ok(Some(contact)) = Contact::get_contact_by_id(transaction, target_user_id).await else { + let Some(contact) = Contact::get_contact_by_id(t, target_user_id).await? else { return Err(twonly_error!("missing contact")); }; @@ -48,9 +48,10 @@ pub(crate) async fn queue_encrypted_content( .encode_to_vec(); let receipt_id = new_uuid_v4(); + NewReceipt::new(&receipt_id, target_user_id, &message) .contact_will_send_receipt(contact_will_send_receipt) - .insert(transaction) + .insert(t) .await?; Ok(receipt_id) @@ -58,21 +59,22 @@ pub(crate) async fn queue_encrypted_content( pub(crate) async fn process_encrypted_or_queue_error( ctx: &Arc, - tr: &mut Transaction<'_, Sqlite>, + t: &mut Transaction<'_, Sqlite>, from_user_id: i64, receipt_id: &str, content: proto::EncryptedContent, ) -> Result> { let group_id = content.group_id.clone(); - match handle_encrypted(ctx, tr, from_user_id, receipt_id, content).await { + match handle_encrypted(ctx, t, from_user_id, receipt_id, content).await { Ok(()) => Ok(None), Err(error) => { let description = error.to_string(); if description.contains("group join arrived before") { - queue_retry_control(tr, from_user_id, receipt_id).await?; + queue_retry_control(t, from_user_id, receipt_id).await?; return Ok(Some(receipt_id.to_owned())); } + let error_type = if description.contains("not a member") { Some(Type::GroupNotFoundOrNotAMember) } else if description.contains("not implemented in Rust") { @@ -80,9 +82,11 @@ pub(crate) async fn process_encrypted_or_queue_error( } else { None }; + let Some(error_type) = error_type else { return Err(error); }; + let outgoing_receipt_id = new_uuid_v4(); let response_content = proto::EncryptedContent { group_id, @@ -92,6 +96,7 @@ pub(crate) async fn process_encrypted_or_queue_error( }), ..Default::default() }; + let response = proto::Message { r#type: proto::message::Type::Ciphertext as i32, receipt_id: String::new(), @@ -99,6 +104,7 @@ pub(crate) async fn process_encrypted_or_queue_error( plaintext_content: None, } .encode_to_vec(); + sqlx::query!( r#" INSERT INTO receipts(receipt_id, contact_id, message, contact_will_sends_receipt) @@ -108,15 +114,16 @@ pub(crate) async fn process_encrypted_or_queue_error( from_user_id, response, ) - .execute(&mut **tr) + .execute(&mut **t) .await?; + Ok(Some(outgoing_receipt_id)) } } } async fn queue_retry_control( - transaction: &mut Transaction<'_, Sqlite>, + t: &mut Transaction<'_, Sqlite>, from_user_id: i64, receipt_id: &str, ) -> Result<()> { @@ -130,6 +137,7 @@ async fn queue_retry_control( }), } .encode_to_vec(); + sqlx::query!( r#" INSERT OR REPLACE INTO receipts(receipt_id, contact_id, message, contact_will_sends_receipt) @@ -139,8 +147,9 @@ async fn queue_retry_control( from_user_id, response, ) - .execute(&mut **transaction) + .execute(&mut **t) .await?; + Ok(()) } @@ -175,6 +184,7 @@ pub(crate) async fn ensure_contact_exists(ctx: &Arc, from_user_id: i64) .map(String::from_utf8) .transpose()? .unwrap_or_else(|| "[Unknown]".into()); + let signal_version = if user.pqc_bundle.is_some() { "v2" } else { @@ -215,7 +225,7 @@ pub(crate) async fn ensure_contact_exists(ctx: &Arc, from_user_id: i64) } pub(crate) async fn queue_sender_delivery_receipt( - transaction: &mut Transaction<'_, Sqlite>, + t: &mut Transaction<'_, Sqlite>, from_user_id: i64, receipt_id: &str, ) -> Result<()> { @@ -226,6 +236,7 @@ pub(crate) async fn queue_sender_delivery_receipt( plaintext_content: None, } .encode_to_vec(); + sqlx::query!( r#" INSERT OR IGNORE INTO receipts( @@ -236,8 +247,9 @@ pub(crate) async fn queue_sender_delivery_receipt( from_user_id, response, ) - .execute(&mut **transaction) + .execute(&mut **t) .await?; + Ok(()) } @@ -275,9 +287,11 @@ async fn encrypt_v2_with_session_recovery( contact_id, "Signal session missing; rebuilding it from the server prekey bundle" ); + ContactService::new(ctx) .establish_signal_session(contact_id) .await?; + encrypt(plaintext).await } result => result, diff --git a/rust/src/api/messages/incoming/client2client/mod.rs b/rust/src/api/messages/incoming/client2client/mod.rs index 1fd4d59b..24430bd5 100644 --- a/rust/src/api/messages/incoming/client2client/mod.rs +++ b/rust/src/api/messages/incoming/client2client/mod.rs @@ -68,7 +68,7 @@ pub(crate) async fn handle_encrypted( } if let Some(update) = content.message_update { - return text_message::handle_message_update(t, from_user_id, update).await; + return text_message::handle_message_update(ctx, t, from_user_id, update).await; } if let Some(update) = content.media_update { diff --git a/rust/src/api/messages/incoming/client2client/recovery.rs b/rust/src/api/messages/incoming/client2client/recovery.rs index a0f7c552..343b78c2 100644 --- a/rust/src/api/messages/incoming/client2client/recovery.rs +++ b/rust/src/api/messages/incoming/client2client/recovery.rs @@ -63,7 +63,7 @@ pub(crate) async fn handle_passwordless_recovery( } pub(crate) async fn perform_heartbeat(ctx: &Arc) -> Result<()> { - let now = chrono::Utc::now(); + let now = crate::utils::current_time().with_timezone(&chrono::Utc); let base_config = UserConfig::load_required_from(ctx)?; let mut config = base_config.clone(); @@ -164,11 +164,15 @@ pub(crate) async fn perform_heartbeat(ctx: &Arc) -> Result<()> { } if config != base_config { - UserConfig::update_json( - ctx, - &serde_json::to_string(&base_config)?, - &serde_json::to_string(&config)?, - )?; + let updated_recovery = config.password_less_recovery.as_ref(); + let last_server_heartbeat = updated_recovery.and_then(|r| r.last_server_heartbeat); + let last_contact_heartbeat = updated_recovery.and_then(|r| r.last_contact_heartbeat); + let config = UserConfig::update(ctx, |current| { + if let Some(recovery) = current.password_less_recovery.as_mut() { + recovery.last_server_heartbeat = last_server_heartbeat; + recovery.last_contact_heartbeat = last_contact_heartbeat; + } + })?; if let Ok(callbacks) = crate::bridge::callbacks::get_callbacks() { (callbacks.api.user_config_changed)(config).await; } diff --git a/rust/src/api/messages/incoming/client2client/text_message.rs b/rust/src/api/messages/incoming/client2client/text_message.rs index df21aaaf..b4e0b808 100644 --- a/rust/src/api/messages/incoming/client2client/text_message.rs +++ b/rust/src/api/messages/incoming/client2client/text_message.rs @@ -4,9 +4,11 @@ */ use crate::api::proto::client::encrypted_content; +use crate::context::Context; use crate::database::app::tables::{Group, Message, MessageType, NewMessage}; use crate::error::Result; -use crate::utils::milliseconds_to_seconds; +use crate::services::mediafiles::MediaFileService; +use crate::utils::{current_time, milliseconds_to_seconds}; use encrypted_content::message_update::Type; use sqlx::{Sqlite, Transaction}; @@ -28,7 +30,7 @@ pub(crate) async fn handle_text_message( .sender_id(from_user_id) .content(&message.text) .maybe_quotes_message_id(message.quote_message_id.as_deref()) - .ack_by_server(chrono::Utc::now().timestamp()) + .ack_by_server(current_time().timestamp()) .build() .insert(t) .await?; @@ -39,6 +41,7 @@ pub(crate) async fn handle_text_message( } pub(crate) async fn handle_message_update( + ctx: &std::sync::Arc, t: &mut Transaction<'_, Sqlite>, from_user_id: i64, update: encrypted_content::MessageUpdate, @@ -51,146 +54,32 @@ pub(crate) async fn handle_message_update( match update_type { Type::Opened => { for message_id in update.multiple_target_message_ids { - let action_at = sqlx::query_scalar::<_, i64>( - "SELECT MAX(created_at, ?) FROM messages WHERE message_id = ?", - ) - .bind(timestamp) - .bind(&message_id) - .fetch_optional(&mut **t) - .await?; - - let Some(action_at) = action_at else { continue }; - - sqlx::query!( - r#" - INSERT INTO message_actions(message_id, contact_id, type, action_at) - VALUES (?, ?, 'openedAt', ?) - ON CONFLICT(message_id, contact_id, type) - DO UPDATE SET action_at = excluded.action_at - "#, - message_id, - from_user_id, - action_at, - ) - .execute(&mut **t) - .await?; - - sqlx::query!( - r#"UPDATE messages SET opened_at = ?, opened_by_all = CASE WHEN NOT EXISTS( - SELECT 1 FROM group_members gm - WHERE gm.group_id = messages.group_id AND NOT EXISTS( - SELECT 1 FROM message_actions ma - WHERE ma.message_id = messages.message_id - AND ma.contact_id = gm.contact_id AND ma.type = 'openedAt' - ) - ) THEN ? ELSE NULL END - WHERE message_id = ?"#, - action_at, - action_at, - message_id, - ) - .execute(&mut **t) - .await?; + Message::record_opened(t, &message_id, from_user_id, timestamp).await?; } } Type::Delete => { - let media_id = sqlx::query_scalar!( - "SELECT media_id FROM messages WHERE message_id = ? AND sender_id = ?", - update.sender_message_id, + let deleted_media = Message::delete_from_sender( + t, + update.sender_message_id.as_deref(), from_user_id, - ) - .fetch_optional(&mut **t) - .await? - .flatten(); - - sqlx::query!( - "DELETE FROM message_histories WHERE message_id = ?", - update.sender_message_id - ) - .execute(&mut **t) - .await?; - - sqlx::query!( - "DELETE FROM receipts WHERE message_id = ?", - update.sender_message_id - ) - .execute(&mut **t) - .await?; - - sqlx::query!( - r#" - UPDATE messages - SET is_deleted_from_sender = 1, content = NULL, media_id = NULL, modified_at = ? - WHERE message_id = ? AND sender_id = ? - "#, timestamp, - update.sender_message_id, - from_user_id, ) - .execute(&mut **t) .await?; - if let Some(media_id) = media_id { - let references = sqlx::query_scalar!( - "SELECT COUNT(*) FROM messages WHERE media_id = ?", - media_id, - ) - .fetch_one(&mut **t) - .await?; - - if references == 0 { - let media_type = sqlx::query_scalar!( - "SELECT type FROM media_files WHERE media_id = ?", - media_id, - ) - .fetch_optional(&mut **t) + if let Some(media) = deleted_media { + MediaFileService::new(ctx) + .remove_files_if_deleted(t, &media.media_id, &media.media_type) .await?; - sqlx::query!("DELETE FROM media_files WHERE media_id = ?", media_id) - .execute(&mut **t) - .await?; - if let Some(media_type) = media_type { - let ctx = crate::context::Context::get_static()?; - let ctx = ctx.clone(); - tokio::spawn(async move { - // Let the surrounding message transaction commit before - // applying its corresponding filesystem side effect. - tokio::time::sleep(std::time::Duration::from_millis(50)).await; - if let Err(error) = - crate::services::mediafiles::MediaFileService::new(&ctx) - .remove_files_if_deleted(&media_id, &media_type) - .await - { - tracing::warn!(media_id, %error, "could not remove media files"); - } - }); - } - } } } Type::EditText => { - sqlx::query!( - r#"INSERT INTO message_histories(message_id, content, created_at) - SELECT message_id, content, ? FROM messages - WHERE message_id = ? AND sender_id = ? AND content IS NOT NULL"#, - timestamp, - update.sender_message_id, + Message::edit_text( + t, + update.sender_message_id.as_deref(), from_user_id, - ) - .execute(&mut **t) - .await?; - - sqlx::query!( - r#" - UPDATE messages - SET content = ?, modified_at = ? - WHERE message_id = ? AND sender_id = ? AND content IS NOT NULL - "#, - update.text, + update.text.as_deref(), timestamp, - update.sender_message_id, - from_user_id, ) - .execute(&mut **t) .await?; } } diff --git a/rust/src/database/app/tables/message.rs b/rust/src/database/app/tables/message.rs index aa07cd2f..21ee0100 100644 --- a/rust/src/database/app/tables/message.rs +++ b/rust/src/database/app/tables/message.rs @@ -9,6 +9,11 @@ use crate::error::{Result, TwonlyError}; pub struct Message; +pub struct DeletedMessageMedia { + pub media_id: String, + pub media_type: String, +} + pub enum MessageType<'a> { Text, Media, @@ -37,6 +42,142 @@ impl<'a> NewMessage<'a> { } impl Message { + pub async fn record_opened( + t: &mut Transaction<'_, Sqlite>, + message_id: &str, + contact_id: i64, + timestamp: i64, + ) -> Result<()> { + let action_at = sqlx::query_scalar::<_, i64>( + "SELECT MAX(created_at, ?) FROM messages WHERE message_id = ?", + ) + .bind(timestamp) + .bind(message_id) + .fetch_optional(&mut **t) + .await?; + let Some(action_at) = action_at else { + return Ok(()); + }; + + sqlx::query!( + r#"INSERT INTO message_actions(message_id, contact_id, type, action_at) + VALUES (?, ?, 'openedAt', ?) + ON CONFLICT(message_id, contact_id, type) + DO UPDATE SET action_at = excluded.action_at"#, + message_id, + contact_id, + action_at, + ) + .execute(&mut **t) + .await?; + + sqlx::query!( + r#"UPDATE messages SET opened_at = ?, opened_by_all = CASE WHEN NOT EXISTS( + SELECT 1 FROM group_members gm + WHERE gm.group_id = messages.group_id AND NOT EXISTS( + SELECT 1 FROM message_actions ma + WHERE ma.message_id = messages.message_id + AND ma.contact_id = gm.contact_id AND ma.type = 'openedAt' + ) + ) THEN ? ELSE NULL END + WHERE message_id = ?"#, + action_at, + action_at, + message_id, + ) + .execute(&mut **t) + .await?; + Ok(()) + } + + pub async fn delete_from_sender( + t: &mut Transaction<'_, Sqlite>, + message_id: Option<&str>, + sender_id: i64, + timestamp: i64, + ) -> Result> { + let media_id = sqlx::query_scalar!( + "SELECT media_id FROM messages WHERE message_id = ? AND sender_id = ?", + message_id, + sender_id, + ) + .fetch_optional(&mut **t) + .await? + .flatten(); + + sqlx::query!( + "DELETE FROM message_histories WHERE message_id = ?", + message_id + ) + .execute(&mut **t) + .await?; + sqlx::query!("DELETE FROM receipts WHERE message_id = ?", message_id) + .execute(&mut **t) + .await?; + sqlx::query!( + r#"UPDATE messages + SET is_deleted_from_sender = 1, content = NULL, media_id = NULL, modified_at = ? + WHERE message_id = ? AND sender_id = ?"#, + timestamp, + message_id, + sender_id, + ) + .execute(&mut **t) + .await?; + + let Some(media_id) = media_id else { + return Ok(None); + }; + let references = + sqlx::query_scalar!("SELECT COUNT(*) FROM messages WHERE media_id = ?", media_id,) + .fetch_one(&mut **t) + .await?; + if references != 0 { + return Ok(None); + } + let media_type = + sqlx::query_scalar!("SELECT type FROM media_files WHERE media_id = ?", media_id,) + .fetch_optional(&mut **t) + .await?; + sqlx::query!("DELETE FROM media_files WHERE media_id = ?", media_id) + .execute(&mut **t) + .await?; + Ok(media_type.map(|media_type| DeletedMessageMedia { + media_id, + media_type, + })) + } + + pub async fn edit_text( + t: &mut Transaction<'_, Sqlite>, + message_id: Option<&str>, + sender_id: i64, + text: Option<&str>, + timestamp: i64, + ) -> Result<()> { + sqlx::query!( + r#"INSERT INTO message_histories(message_id, content, created_at) + SELECT message_id, content, ? FROM messages + WHERE message_id = ? AND sender_id = ? AND content IS NOT NULL"#, + timestamp, + message_id, + sender_id, + ) + .execute(&mut **t) + .await?; + sqlx::query!( + r#"UPDATE messages SET content = ?, modified_at = ? + WHERE message_id = ? AND sender_id = ? AND content IS NOT NULL"#, + text, + timestamp, + message_id, + sender_id, + ) + .execute(&mut **t) + .await?; + Ok(()) + } + pub async fn insert(tr: &mut Transaction<'_, Sqlite>, message: NewMessage<'_>) -> Result<()> { sqlx::query!( r#" diff --git a/rust/src/error.rs b/rust/src/error.rs index 0faac60b..8168f82b 100644 --- a/rust/src/error.rs +++ b/rust/src/error.rs @@ -125,6 +125,8 @@ pub enum TwonlyError { InvalidParams(#[from] InvalidParams), #[error("{0}")] InvalidOutputLen(#[from] InvalidOutputLen), + #[error("{0}")] + InvalidDigestLength(#[from] sha2::digest::InvalidLength), #[error("AES-GCM error")] AesGcm, diff --git a/rust/src/services/contacts.rs b/rust/src/services/contacts.rs index 9fc8e4e4..802c8aae 100644 --- a/rust/src/services/contacts.rs +++ b/rust/src/services/contacts.rs @@ -5,12 +5,14 @@ use crate::api::messages::outgoing::send_c2c_message_to_contact; use crate::api::proto::client::{self as proto, encrypted_content}; +use crate::api::proto::server_to_client; use crate::api::Server; use crate::bridge::api::ServerResult; use crate::context::Context; use crate::database::app::tables::{Contact, Group, UpdateContact}; use crate::error::{Result, TwonlyError}; use crate::signal::engine::FrbPreKeyBundle; +use crate::user_config::UserConfig; use prost::Message as _; use std::io::Write as _; use std::sync::Arc; @@ -73,7 +75,7 @@ impl ContactService { async fn process_user_prekey_bundle( &self, - user: &crate::api::proto::server_to_client::response::UserData, + user: &server_to_client::response::UserData, ) -> Result<()> { let missing = TwonlyError::ApiResponseMissingField; let pqc_bundle = user.pqc_bundle.as_ref().ok_or(missing("pqc_bundle"))?; @@ -81,6 +83,7 @@ impl ContactService { .public_identity_key .clone() .ok_or(missing("public_identity_key"))?; + let registration_id = user.registration_id.ok_or(missing("registration_id"))?; let (pre_key_id, pre_key_public, kyber_pre_key_id, kyber_pre_key_public, kyber_signature) = @@ -136,6 +139,7 @@ impl ContactService { let contact = Contact::get_contact_by_id(&mut transaction, contact_id) .await? .ok_or_else(|| TwonlyError::Generic("contact request does not exist".into()))?; + if contact.requested == 0 { return Err(TwonlyError::Generic( "contact has no pending request".into(), @@ -150,6 +154,7 @@ impl ContactService { .build() .update(&mut transaction) .await?; + Group::create_direct_chat(&self.ctx, &mut transaction, contact).await?; transaction.commit().await?; database.notify_committed(["contacts", "groups"]); @@ -198,7 +203,7 @@ impl ContactService { } pub async fn send_profile(&self, contact_id: i64) -> Result<()> { - let config = crate::user_config::UserConfig::load_required_from(&self.ctx)?; + let config = UserConfig::load_required_from(&self.ctx)?; let avatar_svg_compressed = config .avatar_svg diff --git a/rust/src/services/groups/mod.rs b/rust/src/services/groups/mod.rs index a8d28247..e5a0c5d4 100644 --- a/rust/src/services/groups/mod.rs +++ b/rust/src/services/groups/mod.rs @@ -51,13 +51,16 @@ impl GroupService { let Some(best_friend) = groups.iter().max_by_key(|group| group.total_media_counter) else { return Ok(()); }; + let best_friend_id = best_friend.group_id.clone(); let now = current_time().timestamp(); let start_today = now - now.rem_euclid(86_400); + for group in groups { let Some(changed) = group.last_flame_counter_change else { continue; }; + if changed < start_today || group .last_flame_sync @@ -65,9 +68,11 @@ impl GroupService { { continue; } + if group.flame_counter <= 2 && group.group_id != best_friend_id { continue; } + MessageService::new(&self.ctx) .send_to_group( group.group_id.clone(), @@ -85,6 +90,7 @@ impl GroupService { false, ) .await?; + Group::set_last_flame_sync(&db.pool, &group.group_id, now).await?; } db.notify_committed(["groups"]); diff --git a/rust/src/services/mediafiles.rs b/rust/src/services/mediafiles.rs index 5d8a79b3..ed250dec 100644 --- a/rust/src/services/mediafiles.rs +++ b/rust/src/services/mediafiles.rs @@ -3,6 +3,7 @@ * */ +use crate::api::messages::outgoing::send_c2c_message_to_contact; use crate::api::proto::client::{encrypted_content, EncryptedContent}; use crate::bridge::api::RustApi; use crate::context::Context; @@ -11,7 +12,7 @@ use chacha20poly1305::aead::{AeadInPlace, KeyInit}; use chacha20poly1305::{ChaCha20Poly1305, Nonce, Tag}; use prost::Message as _; use sha2::{Digest, Sha256}; -use sqlx::FromRow; +use sqlx::{FromRow, Sqlite, Transaction}; use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::Duration; @@ -240,26 +241,28 @@ impl MediaFileService { let mac = media.encryption_mac.as_deref().ok_or_else(|| { TwonlyError::Generic(format!("media {} has no encryption MAC", media.media_id)) })?; - let cipher = ChaCha20Poly1305::new_from_slice(key) - .map_err(|_| TwonlyError::Generic("invalid media encryption key".into()))?; + + let cipher = ChaCha20Poly1305::new_from_slice(key)?; + if nonce.len() != 12 { return Err(TwonlyError::Generic( "invalid media encryption nonce".into(), )); } + if mac.len() != 16 { return Err(TwonlyError::Generic("invalid media encryption MAC".into())); } + let nonce = Nonce::from_slice(nonce); let tag = Tag::from_slice(mac); let mut bytes = std::fs::read(encrypted_path)?; - cipher - .decrypt_in_place_detached(nonce, b"", &mut bytes, tag) - .map_err(|_| TwonlyError::Generic("media authentication failed".into()))?; + cipher.decrypt_in_place_detached(nonce, b"", &mut bytes, tag)?; let temp_path = self.temp_path(&media.media_id, &media.media_type); Self::ensure_parent(&temp_path)?; std::fs::write(&temp_path, &bytes)?; + let hash = Sha256::digest(&bytes).to_vec(); // Keep the file update and state transition ordered: ready is only @@ -273,6 +276,7 @@ impl MediaFileService { ) .execute(&database.pool) .await?; + std::fs::remove_file(encrypted_path)?; Ok(()) } @@ -285,6 +289,7 @@ impl MediaFileService { ) .execute(&database.pool) .await?; + let targets = sqlx::query_as::<_, ReuploadTarget>( r#"SELECT message_id, sender_id FROM messages WHERE media_id = ? AND opened_at IS NULL AND sender_id IS NOT NULL"#, @@ -292,6 +297,7 @@ impl MediaFileService { .bind(media_id) .fetch_all(&database.pool) .await?; + database.notify_committed(["media_files"]); drop(database); @@ -303,7 +309,8 @@ impl MediaFileService { }), ..Default::default() }; - crate::api::messages::outgoing::send_c2c_message_to_contact() + + send_c2c_message_to_contact() .ctx(&self.ctx) .contact_id(target.sender_id) .encrypted_content(content.encode_to_vec()) @@ -324,15 +331,20 @@ impl MediaFileService { Ok(()) } - pub async fn remove_files_if_deleted(&self, media_id: &str, media_type: &str) -> Result<()> { - let database = self.ctx.get_app_database().await; + pub async fn remove_files_if_deleted( + &self, + t: &mut Transaction<'_, Sqlite>, + media_id: &str, + media_type: &str, + ) -> Result<()> { let still_exists = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM media_files WHERE media_id = ?)", media_id, ) - .fetch_one(&database.pool) + .fetch_one(&mut **t) .await? != 0; + if !still_exists { self.remove_files(media_id, media_type)?; } diff --git a/rust/src/user_config.rs b/rust/src/user_config.rs index 13fb8533..16da3129 100644 --- a/rust/src/user_config.rs +++ b/rust/src/user_config.rs @@ -375,7 +375,7 @@ impl UserConfig { /// Atomically updates the latest persisted configuration with a typed Rust /// mutation. Unlike `update_json`, this does not need a caller snapshot: /// loading, mutation, and saving all happen while holding the write lock. - pub(crate) fn update(context: &Context, mutate: impl FnOnce(&mut Self)) -> Result<()> { + pub(crate) fn update(context: &Context, mutate: impl FnOnce(&mut Self)) -> Result { let _guard = config_lock() .write() .map_err(|_| twonly_error!("user configuration lock was poisoned"))?; @@ -383,7 +383,7 @@ impl UserConfig { .ok_or_else(|| twonly_error!("user configuration is unavailable"))?; mutate(&mut config); Self::save_unlocked(context, &config)?; - Ok(()) + Ok(config) } /// Applies only fields changed relative to the caller's original snapshot.