Implement ArchiveVersionDownloads background job

.env.sample

@@ -52,6 +52,13 @@ export TEST_DATABASE_URL=
 # export CDN_LOG_QUEUE_URL=
 # export CDN_LOG_QUEUE_REGION=
 
+# Configuration for the version downloads data archive.
+# You can leave these commented out if you're not using the archival process.
+# export DOWNLOADS_ARCHIVE_ACCESS_KEY=
+# export DOWNLOADS_ARCHIVE_SECRET_KEY=
+# export DOWNLOADS_ARCHIVE_REGION=
+# export DOWNLOADS_ARCHIVE_BUCKET=
+
 # Upstream location of the registry index. Background jobs will push to
 # this URL. The default points to a local index for development.
 # Run `./script/init-local-index.sh` to initialize this repo.

Cargo.toml

@@ -65,6 +65,7 @@ crates_io_index = { path = "crates/crates_io_index" }
 crates_io_markdown = { path = "crates/crates_io_markdown" }
 crates_io_tarball = { path = "crates/crates_io_tarball" }
 crates_io_worker = { path = "crates/crates_io_worker" }
+csv = "=1.3.0"
 chrono = { version = "=0.4.38", default-features = false, features = ["serde"] }
 clap = { version = "=4.5.4", features = ["derive", "env", "unicode", "wrap_help"] }
 cookie = { version = "=0.18.1", features = ["secure"] }
@@ -110,7 +111,7 @@ spdx = "=0.10.4"
 tar = "=0.4.40"
 tempfile = "=3.10.1"
 thiserror = "=1.0.60"
-tokio = { version = "=1.37.0", features = ["net", "signal", "io-std", "io-util", "rt-multi-thread", "macros"]}
+tokio = { version = "=1.37.0", features = ["net", "signal", "io-std", "io-util", "rt-multi-thread", "macros", "process"]}
 toml = "=0.8.12"
 tower = "=0.4.13"
 tower-http = { version = "=0.5.2", features = ["add-extension", "fs", "catch-panic", "timeout", "compression-full"] }

src/admin/enqueue_job.rs

@@ -2,6 +2,7 @@ use crate::db;
 use crate::schema::{background_jobs, crates};
 use crate::worker::jobs;
 use anyhow::Result;
+use chrono::NaiveDate;
 use crates_io_worker::BackgroundJob;
 use diesel::dsl::exists;
 use diesel::prelude::*;
@@ -14,6 +15,11 @@ use secrecy::{ExposeSecret, SecretString};
     rename_all = "snake_case"
 )]
 pub enum Command {
+    ArchiveVersionDownloads {
+        #[arg(long)]
+        /// The date before which to archive version downloads (default: 90 days ago)
+        before: Option<NaiveDate>,
+    },
     UpdateDownloads,
     CleanProcessedLogFiles,
     DumpDb {
@@ -45,6 +51,12 @@ pub fn run(command: Command) -> Result<()> {
     println!("Enqueueing background job: {command:?}");
 
     match command {
+        Command::ArchiveVersionDownloads { before } => {
+            before
+                .map(jobs::ArchiveVersionDownloads::before)
+                .unwrap_or_default()
+                .enqueue(conn)?;
+        }
         Command::UpdateDownloads => {
             let count: i64 = background_jobs::table
                 .filter(background_jobs::job_type.eq(jobs::UpdateDownloads::JOB_NAME))

src/bin/background-worker.rs

@@ -18,6 +18,7 @@ use crates_io::cloudfront::CloudFront;
 use crates_io::fastly::Fastly;
 use crates_io::storage::Storage;
 use crates_io::team_repo::TeamRepoImpl;
+use crates_io::worker::jobs::ArchiveVersionDownloads;
 use crates_io::worker::{Environment, RunnerExt};
 use crates_io::{config, Emails};
 use crates_io::{db, ssh};
@@ -69,6 +70,7 @@ fn main() -> anyhow::Result<()> {
 
     let cloudfront = CloudFront::from_environment();
     let storage = Arc::new(Storage::from_config(&config.storage));
+    let downloads_archive_store = ArchiveVersionDownloads::store_from_environment()?;
 
     let client = Client::builder()
         .timeout(Duration::from_secs(45))
@@ -88,6 +90,7 @@ fn main() -> anyhow::Result<()> {
         .cloudfront(cloudfront)
         .fastly(fastly)
         .storage(storage)
+        .downloads_archive_store(downloads_archive_store)
         .deadpool(deadpool.clone())
         .emails(emails)
         .team_repo(Box::new(team_repo))

src/worker/environment.rs

@@ -8,6 +8,7 @@ use crates_io_index::{Repository, RepositoryConfig};
 use deadpool_diesel::postgres::Pool as DeadpoolPool;
 use derive_builder::Builder;
 use diesel::PgConnection;
+use object_store::ObjectStore;
 use parking_lot::{Mutex, MutexGuard};
 use std::ops::{Deref, DerefMut};
 use std::sync::{Arc, OnceLock};
@@ -26,6 +27,8 @@ pub struct Environment {
     #[builder(default)]
     fastly: Option<Fastly>,
     pub storage: Arc<Storage>,
+    #[builder(default)]
+    pub downloads_archive_store: Option<Box<dyn ObjectStore>>,
     pub deadpool: DeadpoolPool,
     pub emails: Emails,
     pub team_repo: Box<dyn TeamRepo + Send + Sync>,