diff --git a/CHANGELOG.md b/CHANGELOG.md index 2c27f453..5ed243f9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,6 @@ # Unreleased - Add `re delete emails` to delete emails by id from a bucket +- Add `re delete bulk-emails` to bulk-delete emails from a bucket in a given time range # v0.40.0 - Allow `re get emails` to filter by `--mailbox`, `--from-timestamp`, and `--to-timestamp` diff --git a/cli/src/commands/delete.rs b/cli/src/commands/delete.rs index 6227f759..82c66dac 100644 --- a/cli/src/commands/delete.rs +++ b/cli/src/commands/delete.rs @@ -9,9 +9,10 @@ use std::sync::{ use structopt::StructOpt; use reinfer_client::{ - resources::{bucket::GetKeyedSyncStateIdsRequest, project::ForceDeleteProject}, - BucketIdentifier, Client, CommentId, CommentsIter, CommentsIterTimerange, DatasetIdentifier, - EmailId, ProjectName, Source, SourceIdentifier, UserIdentifier, + resources::{bucket::GetKeyedSyncStateIdsRequest, email::Email, project::ForceDeleteProject}, + Bucket, BucketIdentifier, Client, CommentId, CommentsIter, CommentsIterTimerange, + DatasetIdentifier, EmailId, EmailsQueryFilter, ProjectName, Source, SourceIdentifier, + UserIdentifier, }; use crate::progress::{Options as ProgressOptions, Progress}; @@ -79,6 +80,29 @@ pub enum DeleteArgs { no_progress: bool, }, + #[structopt(name = "bulk-emails")] + /// Delete all emails in a bucket in a given time range. With no time range, + /// deletes every email in the bucket. + BulkEmails { + #[structopt(short = "b", long = "bucket")] + /// Name or id of the bucket to delete emails from + bucket: BucketIdentifier, + + #[structopt(long = "from-timestamp")] + /// Starting timestamp for emails to delete (inclusive). Should be in + /// RFC 3339 format, e.g. 2024-01-01T00:00:00Z + from_timestamp: Option>, + + #[structopt(long = "to-timestamp")] + /// Ending timestamp for emails to delete (exclusive). Should be in + /// RFC 3339 format, e.g. 2024-02-01T00:00:00Z + to_timestamp: Option>, + + #[structopt(long)] + /// Don't display a progress bar + no_progress: bool, + }, + #[structopt(name = "bucket")] /// Delete a bucket Bucket { @@ -173,6 +197,22 @@ pub fn run(delete_args: &DeleteArgs, client: Client) -> Result<()> { ) .context("Operation to delete comments has failed.")?; } + DeleteArgs::BulkEmails { + bucket, + from_timestamp, + to_timestamp, + no_progress, + } => { + let bucket = client.get_bucket(bucket.clone())?; + delete_emails_in_period( + &client, + bucket, + *from_timestamp, + *to_timestamp, + !no_progress, + ) + .context("Operation to delete emails has failed.")?; + } DeleteArgs::Dataset { dataset } => { client .delete_dataset(dataset.clone()) @@ -306,6 +346,78 @@ fn delete_comments_in_period( Ok(()) } +fn delete_emails_in_period( + client: &Client, + bucket: Bucket, + from_timestamp: Option>, + to_timestamp: Option>, + show_progress: bool, +) -> Result<()> { + log::info!( + "Deleting emails in bucket `{}`{}", + bucket.full_name().0, + match (from_timestamp, to_timestamp) { + (None, None) => "".into(), + (Some(start), None) => format!(" after {start}"), + (None, Some(end)) => format!(" before {end}"), + (Some(start), Some(end)) => format!(" in range {start} -> {end}"), + }, + ); + let filter = EmailsQueryFilter { + from_timestamp, + to_timestamp, + mailbox_name: None, + }; + let statistics = Arc::new(Statistics::new()); + let bucket_name = bucket.full_name(); + { + let _progress = if show_progress { + Some(delete_emails_progress_bar(&statistics)) + } else { + None + }; + + // The maximum number of emails the API permits deleting in a single call. + const DELETION_BATCH_SIZE: usize = 32; + let mut emails_to_delete = Vec::with_capacity(DELETION_BATCH_SIZE); + + let delete_batch = |email_ids: Vec| -> Result<()> { + client + .delete_emails(bucket_name.clone(), &email_ids) + .context("Operation to delete emails failed")?; + statistics.increment_deleted(email_ids.len()); + Ok(()) + }; + + // An empty filter deletes every email in the bucket via the listing + // endpoint; any filter narrows the set via the query endpoint. + let mut pages: Box>> + '_> = + if filter.is_empty() { + Box::new(client.get_emails_iter(&bucket_name, None)) + } else { + Box::new(client.query_emails_iter(&bucket_name, filter, None)) + }; + + pages.try_for_each(|page| -> Result<()> { + let page = page.context("Operation to get emails failed")?; + emails_to_delete.extend(page.into_iter().map(|email| email.id)); + while emails_to_delete.len() >= DELETION_BATCH_SIZE { + let batch: Vec = emails_to_delete.drain(..DELETION_BATCH_SIZE).collect(); + delete_batch(batch)?; + } + Ok(()) + })?; + + // Delete any emails left over in the final partial batch. + if !emails_to_delete.is_empty() { + assert!(emails_to_delete.len() < DELETION_BATCH_SIZE); + delete_batch(emails_to_delete)?; + } + } + log::info!("Deleted {} emails.", statistics.deleted()); + Ok(()) +} + #[derive(Debug)] pub struct Statistics { deleted: AtomicUsize, @@ -362,3 +474,18 @@ fn delete_comments_progress_bar(statistics: &Arc) -> Progress { ProgressOptions { bytes_units: false }, ) } + +fn delete_emails_progress_bar(statistics: &Arc) -> Progress { + Progress::new( + move |statistics| { + let num_deleted = statistics.deleted() as u64; + ( + num_deleted, + format!("{} {}", num_deleted.to_string().bold(), "deleted".dimmed()), + ) + }, + statistics, + None, + ProgressOptions { bytes_units: false }, + ) +} diff --git a/cli/tests/test_buckets.rs b/cli/tests/test_buckets.rs index 23f28721..6c4fc17f 100644 --- a/cli/tests/test_buckets.rs +++ b/cli/tests/test_buckets.rs @@ -219,6 +219,145 @@ fn test_delete_emails() { cli.run(["delete", "bucket", &bucket]); } +#[test] +fn test_bulk_delete_emails() { + let cli = TestCli::get(); + let owner = TestCli::project(); + + let bucket = format!("{}/test-bulk-delete-emails-{}", owner, Uuid::new_v4()); + cli.run(["create", "bucket", &bucket]); + + // Four emails across two mailboxes and a spread of timestamps. + let emails = [ + ("alice-1", "alice@reinfer.io", "2020-01-01T00:00:00Z"), + ("alice-2", "alice@reinfer.io", "2020-02-01T00:00:00Z"), + ("bob-1", "bob@reinfer.io", "2020-01-15T00:00:00Z"), + ("bob-2", "bob@reinfer.io", "2020-03-01T00:00:00Z"), + ]; + let jsonl = emails + .iter() + .map(|(id, mailbox, timestamp)| { + serde_json::json!({ + "id": id, + "mailbox": mailbox, + "timestamp": timestamp, + "mime_content": format!( + "Date: {timestamp}\r\nFrom: {mailbox}\r\nTo: support@reinfer.io\r\n\ + Subject: {id}\r\nContent-Type: text/plain\r\n\r\nHello from {id}\r\n" + ), + }) + .to_string() + }) + .collect::>() + .join("\n"); + cli.run_with_stdin(["create", "emails", "-y", "-b", &bucket], jsonl.as_bytes()); + + // Sorted list of email ids currently in the bucket. + let ids = |output: &str| -> Vec { + let mut values: Vec = output + .lines() + .filter(|line| !line.trim().is_empty()) + .map(|line| { + serde_json::from_str::(line).unwrap()["id"] + .as_str() + .unwrap() + .to_owned() + }) + .collect(); + values.sort(); + values + }; + + // Bulk delete by time range [from inclusive, to exclusive): removes alice-2 and bob-1. + cli.run([ + "delete", + "bulk-emails", + "-b", + &bucket, + "--from-timestamp", + "2020-01-10T00:00:00Z", + "--to-timestamp", + "2020-02-15T00:00:00Z", + ]); + assert_eq!( + ids(&cli.run(["get", "emails", &bucket])), + vec!["alice-1", "bob-2"], + "time-range bulk delete should remove only emails in [from, to)" + ); + + // Bulk delete with no time range: removes everything remaining. + cli.run(["delete", "bulk-emails", "-b", &bucket]); + assert!( + ids(&cli.run(["get", "emails", &bucket])).is_empty(), + "unfiltered bulk delete should remove all emails" + ); + + cli.run(["delete", "bucket", &bucket]); +} + +#[test] +fn test_bulk_delete_emails_across_batches() { + let cli = TestCli::get(); + let owner = TestCli::project(); + + let bucket = format!("{}/test-bulk-delete-batches-{}", owner, Uuid::new_v4()); + cli.run(["create", "bucket", &bucket]); + + // More emails than both the deletion batch size (32) and the iterator page + // size (64), so the bulk delete must drain several full batches across more + // than one fetched page plus a final partial batch. + const COUNT: usize = 70; + let jsonl = (0..COUNT) + .map(|i| { + serde_json::json!({ + "id": format!("email-{i:03}"), + "mailbox": "alice@reinfer.io", + "timestamp": "2020-01-01T00:00:00Z", + "mime_content": format!( + "Date: 2020-01-01T00:00:00Z\r\nFrom: alice@reinfer.io\r\nTo: support@reinfer.io\r\n\ + Subject: email-{i:03}\r\nContent-Type: text/plain\r\n\r\nHello {i}\r\n" + ), + }) + .to_string() + }) + .collect::>() + .join("\n"); + cli.run_with_stdin(["create", "emails", "-y", "-b", &bucket], jsonl.as_bytes()); + + let count = |output: &str| { + output + .lines() + .filter(|line| !line.trim().is_empty()) + .count() + }; + + assert_eq!( + count(&cli.run(["get", "emails", &bucket])), + COUNT, + "all emails should be present before deletion" + ); + + // Time-range bulk delete covering every email: exercises the query iterator + // across multiple pages and multiple deletion batches. + cli.run([ + "delete", + "bulk-emails", + "-b", + &bucket, + "--from-timestamp", + "2020-01-01T00:00:00Z", + "--to-timestamp", + "2020-01-02T00:00:00Z", + ]); + assert_eq!( + count(&cli.run(["get", "emails", &bucket])), + 0, + "bulk delete should remove every email across all batches" + ); + + cli.run(["delete", "bucket", &bucket]); +} + #[test] fn test_create_without_org_fails() { let cli = TestCli::get();