use google_gmail1::api::{BatchDeleteMessagesRequest, BatchModifyMessagesRequest};
use crate::{EolAction, Error, GmailClient, Result, message_list::MessageList, rules::EolRule};
const TRASH_LABEL: &str = "TRASH";
const INBOX_LABEL: &str = "INBOX";
const GMAIL_MODIFY_SCOPE: &str = "https://www.googleapis.com/auth/gmail.modify";
const GMAIL_DELETE_SCOPE: &str = "https://mail.google.com/";
#[doc(hidden)]
pub(crate) trait MailOperations {
fn add_labels(&mut self, labels: &[String]) -> Result<()>;
fn label_ids(&self) -> Vec<String>;
fn set_query(&mut self, query: &str);
fn prepare(&mut self, pages: u32) -> impl std::future::Future<Output = Result<()>> + Send;
fn batch_trash(&mut self) -> impl std::future::Future<Output = Result<()>> + Send;
}
async fn process_label_with_rule<T: MailOperations>(
client: &mut T,
rule: &EolRule,
label: &str,
pages: u32,
execute: bool,
) -> Result<()> {
client.add_labels(&[label.to_owned()])?;
if client.label_ids().is_empty() {
return Err(Error::LabelNotFoundInMailbox(label.to_owned()));
}
let Some(query) = rule.eol_query() else {
return Err(Error::NoQueryStringCalculated(rule.id()));
};
client.set_query(&query);
log::info!("Ready to process messages for label: {label}");
client.prepare(pages).await?;
if execute {
log::info!("Execute mode: applying rule action to messages");
client.batch_trash().await
} else {
log::info!("Dry-run mode: no changes made to messages");
Ok(())
}
}
impl MailOperations for GmailClient {
fn add_labels(&mut self, labels: &[String]) -> Result<()> {
MessageList::add_labels(self, labels)
}
fn label_ids(&self) -> Vec<String> {
MessageList::label_ids(self)
}
fn set_query(&mut self, query: &str) {
MessageList::set_query(self, query);
}
async fn prepare(&mut self, pages: u32) -> Result<()> {
self.get_messages(pages).await
}
async fn batch_trash(&mut self) -> Result<()> {
RuleProcessor::batch_trash(self).await
}
}
pub trait RuleProcessor {
fn find_rule_and_messages_for_label(
&mut self,
label: &str,
) -> impl std::future::Future<Output = Result<()>> + Send;
fn set_execute(&mut self, value: bool);
fn initialise_lists(&mut self);
fn set_rule(&mut self, rule: EolRule);
fn action(&self) -> Option<EolAction>;
fn prepare(&mut self, pages: u32) -> impl std::future::Future<Output = Result<()>> + Send;
fn batch_delete(&mut self) -> impl std::future::Future<Output = Result<()>> + Send;
fn batch_trash(&mut self) -> impl std::future::Future<Output = Result<()>> + Send;
fn process_in_chunks(
&self,
message_ids: Vec<String>,
action: EolAction,
) -> impl std::future::Future<Output = Result<()>> + Send;
fn call_batch_delete(
&self,
ids: &[String],
) -> impl std::future::Future<Output = Result<()>> + Send;
fn call_batch_trash(
&self,
ids: &[String],
) -> impl std::future::Future<Output = Result<()>> + Send;
}
impl RuleProcessor for GmailClient {
fn initialise_lists(&mut self) {
self.messages = Vec::new();
self.label_ids = Vec::new();
}
fn set_rule(&mut self, value: EolRule) {
self.rule = Some(value);
}
fn set_execute(&mut self, value: bool) {
self.execute = value;
}
fn action(&self) -> Option<EolAction> {
if let Some(rule) = &self.rule {
return rule.action();
}
None
}
async fn find_rule_and_messages_for_label(&mut self, label: &str) -> Result<()> {
let Some(rule) = self.rule.clone() else {
return Err(Error::RuleNotFound(0));
};
let execute = self.execute;
process_label_with_rule(self, &rule, label, 0, execute).await
}
async fn prepare(&mut self, pages: u32) -> Result<()> {
self.get_messages(pages).await
}
async fn batch_delete(&mut self) -> Result<()> {
let message_ids = MessageList::message_ids(self);
if message_ids.is_empty() {
log::info!("No messages to delete - skipping batch delete operation");
return Ok(());
}
self.log_messages("Message with subject `", "` permanently deleted")
.await?;
self.process_in_chunks(message_ids, EolAction::Delete)
.await?;
Ok(())
}
async fn batch_trash(&mut self) -> Result<()> {
let message_ids = MessageList::message_ids(self);
if message_ids.is_empty() {
log::info!("No messages to trash - skipping batch trash operation");
return Ok(());
}
self.log_messages("Message with subject `", "` moved to trash")
.await?;
self.process_in_chunks(message_ids, EolAction::Trash)
.await?;
Ok(())
}
async fn process_in_chunks(&self, message_ids: Vec<String>, action: EolAction) -> Result<()> {
let (chunks, remainder) = message_ids.as_chunks::<1000>();
log::info!(
"Message list chopped into {} chunks with {} ids in the remainder",
chunks.len(),
remainder.len()
);
let act = async |action, list| match action {
EolAction::Trash => self.call_batch_trash(list).await,
EolAction::Delete => self.call_batch_delete(list).await,
};
if !chunks.is_empty() {
for (i, chunk) in chunks.iter().enumerate() {
log::info!("Processing chunk {i}");
act(action, chunk).await?;
}
}
if !remainder.is_empty() {
log::info!("Processing remainder.");
act(action, remainder).await?;
}
Ok(())
}
async fn call_batch_delete(&self, ids: &[String]) -> Result<()> {
let ids = Some(Vec::from(ids));
let batch_request = BatchDeleteMessagesRequest { ids };
log::trace!("{batch_request:#?}");
let res = self
.hub()
.users()
.messages_batch_delete(batch_request, "me")
.add_scope(GMAIL_DELETE_SCOPE)
.doit()
.await
.map_err(Box::new);
log::trace!("Batch delete response {res:?}");
res?;
Ok(())
}
async fn call_batch_trash(&self, ids: &[String]) -> Result<()> {
let ids = Some(Vec::from(ids));
let add_label_ids = Some(vec![TRASH_LABEL.to_string()]);
let remove_label_ids = Some(vec![INBOX_LABEL.to_string()]);
let batch_request = BatchModifyMessagesRequest {
add_label_ids,
ids,
remove_label_ids,
};
log::trace!("{batch_request:#?}");
let _res = self
.hub()
.users()
.messages_batch_modify(batch_request, "me")
.add_scope(GMAIL_MODIFY_SCOPE)
.doit()
.await
.map_err(Box::new)?;
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{EolAction, Error, MessageSummary, rules::EolRule};
use std::sync::{Arc, Mutex};
fn create_test_rule(id: usize, has_query: bool) -> EolRule {
use crate::{MessageAge, Retention};
let mut rule = EolRule::new(id);
if has_query {
let retention = Retention::new(MessageAge::Days(30), false);
rule.set_retention(retention);
rule.add_label("test-label");
}
rule
}
struct FakeClient {
labels: Vec<String>,
label_ids: Vec<String>,
query: String,
messages_prepared: bool,
prepare_call_count: u32,
batch_trash_call_count: Arc<Mutex<u32>>, should_fail_add_labels: bool,
should_fail_prepare: bool,
should_fail_batch_trash: bool,
simulate_missing_labels: bool, }
impl Default for FakeClient {
fn default() -> Self {
Self {
labels: Vec::new(),
label_ids: Vec::new(),
query: String::new(),
messages_prepared: false,
prepare_call_count: 0,
batch_trash_call_count: Arc::new(Mutex::new(0)),
should_fail_add_labels: false,
should_fail_prepare: false,
should_fail_batch_trash: false,
simulate_missing_labels: false,
}
}
}
impl FakeClient {
fn new() -> Self {
Self::default()
}
fn with_missing_labels() -> Self {
Self {
simulate_missing_labels: true,
..Default::default()
}
}
fn with_labels(label_ids: Vec<String>) -> Self {
Self {
label_ids,
..Default::default()
}
}
fn with_failure(failure_type: &str) -> Self {
match failure_type {
"add_labels" => Self {
should_fail_add_labels: true,
..Default::default()
},
"prepare" => Self {
should_fail_prepare: true,
..Default::default()
},
"batch_trash" => Self {
should_fail_batch_trash: true,
..Default::default()
},
_ => Self::default(),
}
}
fn get_batch_trash_call_count(&self) -> u32 {
*self.batch_trash_call_count.lock().unwrap()
}
}
impl MailOperations for FakeClient {
fn add_labels(&mut self, labels: &[String]) -> Result<()> {
if self.should_fail_add_labels {
return Err(Error::DirectoryUnset); }
self.labels.extend(labels.iter().cloned());
if !self.simulate_missing_labels && !labels.is_empty() {
self.label_ids = labels.to_vec();
}
Ok(())
}
fn label_ids(&self) -> Vec<String> {
self.label_ids.clone()
}
fn set_query(&mut self, query: &str) {
self.query = query.to_owned();
}
async fn prepare(&mut self, _pages: u32) -> Result<()> {
self.prepare_call_count += 1;
if self.should_fail_prepare {
return Err(Error::NoLabelsFound); }
self.messages_prepared = true;
Ok(())
}
async fn batch_trash(&mut self) -> Result<()> {
*self.batch_trash_call_count.lock().unwrap() += 1;
if self.should_fail_batch_trash {
return Err(Error::InvalidPagingMode); }
Ok(())
}
}
#[tokio::test]
async fn test_errors_when_label_missing() {
let mut client = FakeClient::with_missing_labels(); let rule = create_test_rule(1, true);
let label = "missing-label";
let result = process_label_with_rule(&mut client, &rule, label, 0, false).await;
assert!(matches!(result, Err(Error::LabelNotFoundInMailbox(_))));
assert_eq!(client.prepare_call_count, 0);
assert_eq!(client.get_batch_trash_call_count(), 0);
}
#[tokio::test]
async fn test_errors_when_rule_has_no_query() {
let mut client = FakeClient::with_labels(vec!["test-label".to_string()]);
let rule = create_test_rule(2, false); let label = "test-label";
let result = process_label_with_rule(&mut client, &rule, label, 0, false).await;
assert!(matches!(result, Err(Error::NoQueryStringCalculated(2))));
assert_eq!(client.prepare_call_count, 0);
assert_eq!(client.get_batch_trash_call_count(), 0);
}
#[tokio::test]
async fn test_dry_run_does_not_trash() {
let mut client = FakeClient::with_labels(vec!["test-label".to_string()]);
let rule = create_test_rule(3, true);
let label = "test-label";
let result = process_label_with_rule(&mut client, &rule, label, 0, false).await;
assert!(result.is_ok());
assert_eq!(client.prepare_call_count, 1);
assert_eq!(client.get_batch_trash_call_count(), 0); assert!(client.messages_prepared);
assert!(!client.query.is_empty()); }
#[tokio::test]
async fn test_execute_trashes_messages_once() {
let mut client = FakeClient::with_labels(vec!["test-label".to_string()]);
let rule = create_test_rule(4, true);
let label = "test-label";
let result = process_label_with_rule(&mut client, &rule, label, 0, true).await;
assert!(result.is_ok());
assert_eq!(client.prepare_call_count, 1);
assert_eq!(client.get_batch_trash_call_count(), 1); assert!(client.messages_prepared);
assert!(!client.query.is_empty());
}
#[tokio::test]
async fn test_propagates_prepare_error() {
let mut client = FakeClient::with_labels(vec!["test-label".to_string()]);
client.should_fail_prepare = true;
let rule = create_test_rule(5, true);
let label = "test-label";
let result = process_label_with_rule(&mut client, &rule, label, 0, true).await;
assert!(result.is_err());
assert_eq!(client.prepare_call_count, 1); assert_eq!(client.get_batch_trash_call_count(), 0); }
#[tokio::test]
async fn test_propagates_batch_trash_error() {
let mut client = FakeClient::with_labels(vec!["test-label".to_string()]);
client.should_fail_batch_trash = true;
let rule = create_test_rule(6, true);
let label = "test-label";
let result = process_label_with_rule(&mut client, &rule, label, 0, true).await;
assert!(result.is_err());
assert_eq!(client.prepare_call_count, 1);
assert_eq!(client.get_batch_trash_call_count(), 1); }
#[tokio::test]
async fn test_pages_parameter_passed_correctly() {
let mut client = FakeClient::with_labels(vec!["test-label".to_string()]);
let rule = create_test_rule(7, true);
let label = "test-label";
let pages = 5;
let result = process_label_with_rule(&mut client, &rule, label, pages, false).await;
assert!(result.is_ok());
assert_eq!(client.prepare_call_count, 1);
}
#[test]
fn test_rule_processor_setters_and_getters() {
struct MockProcessor {
messages: Vec<MessageSummary>,
rule: Option<EolRule>,
execute: bool,
labels: Vec<String>,
}
impl RuleProcessor for MockProcessor {
fn initialise_lists(&mut self) {
self.messages = Vec::new();
self.labels = Vec::new();
}
fn set_rule(&mut self, rule: EolRule) {
self.rule = Some(rule);
}
fn set_execute(&mut self, value: bool) {
self.execute = value;
}
fn action(&self) -> Option<EolAction> {
self.rule.as_ref().and_then(|r| r.action())
}
async fn find_rule_and_messages_for_label(&mut self, _label: &str) -> Result<()> {
Ok(())
}
async fn prepare(&mut self, _pages: u32) -> Result<()> {
Ok(())
}
async fn batch_delete(&mut self) -> Result<()> {
Ok(())
}
async fn call_batch_delete(&self, _ids: &[String]) -> Result<()> {
Ok(())
}
async fn batch_trash(&mut self) -> Result<()> {
Ok(())
}
async fn call_batch_trash(&self, _ids: &[String]) -> Result<()> {
Ok(())
}
async fn process_in_chunks(
&self,
_message_ids: Vec<String>,
_action: EolAction,
) -> Result<()> {
Ok(())
}
}
let mut processor = MockProcessor {
rule: None,
execute: false,
messages: Vec::new(),
labels: Vec::new(),
};
assert!(processor.action().is_none());
assert!(!processor.execute);
let rule = create_test_rule(8, true);
processor.set_rule(rule);
assert!(processor.action().is_some());
assert_eq!(processor.action(), Some(EolAction::Trash));
processor.set_execute(true);
assert!(processor.execute);
processor.set_execute(false);
assert!(!processor.execute);
}
}