add CLI command to repackage file-storage to archive-storage

crates/bin/docs_rs_admin/Cargo.toml

@@ -17,10 +17,12 @@ docs_rs_database = { path = "../../lib/docs_rs_database" }
 docs_rs_fastly = { path = "../../lib/docs_rs_fastly" }
 docs_rs_headers = { path = "../../lib/docs_rs_headers" }
 docs_rs_logging = { path = "../../lib/docs_rs_logging" }
+docs_rs_storage = { path = "../../lib/docs_rs_storage" }
 docs_rs_types = { path = "../../lib/docs_rs_types" }
 docs_rs_utils = { path = "../../lib/docs_rs_utils" }
 futures-util = { workspace = true }
 sqlx = { workspace = true }
+tempfile = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
 
@@ -34,3 +36,4 @@ docs_rs_storage = { path = "../../lib/docs_rs_storage", features = ["testing"] }
 docs_rs_test_fakes = { path = "../../lib/docs_rs_test_fakes" }
 docs_rs_types = { path = "../../lib/docs_rs_types", features = ["testing"] }
 pretty_assertions = { workspace = true }
+test-case = { workspace = true }

crates/bin/docs_rs_admin/src/main.rs

@@ -1,4 +1,5 @@
 mod rebuilds;
+mod repackage;
 #[cfg(test)]
 pub(crate) mod testing;
 
@@ -17,7 +18,7 @@ use docs_rs_database::{
 };
 use docs_rs_fastly::CdnBehaviour as _;
 use docs_rs_headers::SurrogateKey;
-use docs_rs_types::{CrateId, KrateName, Version};
+use docs_rs_types::{CrateId, KrateName, ReleaseId, Version};
 use futures_util::StreamExt;
 use rebuilds::queue_rebuilds_faulty_rustdoc;
 use std::iter;
@@ -73,6 +74,8 @@ impl CommandLine {
             .with_meter_provider()?
             .with_pool()
             .await?
+            .with_storage()
+            .await?
             .with_build_queue()?
             .with_repository_stats()?
             .with_registry_api()?
@@ -283,6 +286,14 @@ enum DatabaseSubcommand {
         version: Option<i64>,
     },
 
+    /// temporary command to repackage missing crates into archive storage.
+    /// starts at the earliest release and works forwards.
+    Repackage {
+        /// process at most this amount of releases
+        #[arg(long)]
+        limit: Option<u32>,
+    },
+
     /// temporary command to update the `crates.latest_version_id` field
     UpdateLatestVersionId,
 
@@ -320,6 +331,48 @@ impl DatabaseSubcommand {
             }
             .context("Failed to run database migrations")?,
 
+            Self::Repackage { limit } => {
+                let pool = ctx.pool()?;
+                let storage = ctx.storage()?;
+                let mut list_conn = pool.get_async().await?;
+                let mut update_conn = pool.get_async().await?;
+
+                let limit = limit.unwrap_or(2_000_000u32);
+
+                let mut stream = sqlx::query!(
+                    r#"SELECT
+                           r.id as "rid: ReleaseId",
+                           c.name as "name: KrateName",
+                           r.version as "version: Version"
+                       FROM
+                            crates as c
+                            INNER JOIN releases as r ON c.id = r.crate_id
+                       WHERE
+                            r.archive_storage = FALSE
+                       ORDER BY r.id
+                       LIMIT $1
+                    "#,
+                    limit as i64,
+                )
+                .fetch(&mut *list_conn);
+
+                while let Some(row) = stream.next().await {
+                    let row = row?;
+
+                    crate::repackage::repackage(
+                        &mut update_conn,
+                        storage,
+                        row.rid,
+                        &row.name,
+                        &row.version,
+                    )
+                    .await?;
+                }
+
+                Ok::<(), anyhow::Error>(())
+            }
+            .context("Failed to repackage storage")?,
+
             Self::UpdateLatestVersionId => {
                 let pool = ctx.pool()?;
                 let mut list_conn = pool.get_async().await?;