feat(cli): add re delete bulk-emails to bulk-delete emails by time range

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`

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<DateTime<Utc>>,
+
+        #[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<DateTime<Utc>>,
+
+        #[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<DateTime<Utc>>,
+    to_timestamp: Option<DateTime<Utc>>,
+    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<EmailId>| -> 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<dyn Iterator<Item = reinfer_client::Result<Vec<Email>>> + '_> =
+            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<EmailId> = 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<Statistics>) -> Progress {
         ProgressOptions { bytes_units: false },
     )
 }
+
+fn delete_emails_progress_bar(statistics: &Arc<Statistics>) -> 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 },
+    )
+}

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::<Vec<_>>()
+        .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<String> {
+        let mut values: Vec<String> = output
+            .lines()
+            .filter(|line| !line.trim().is_empty())
+            .map(|line| {
+                serde_json::from_str::<serde_json::Value>(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::<Vec<_>>()
+        .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();