Import: copy an npm registry's packages (Nexus, Verdaccio, another Forge) into a hosted npm repository

.github/workflows/ci.yml

@@ -725,6 +725,7 @@ jobs:
           docker pull mirror.gcr.io/versity/versitygw:v1.9.0
           docker pull mirror.gcr.io/library/redis:5.0
           docker pull mirror.gcr.io/sonatype/nexus3:3.70.1
+          docker pull mirror.gcr.io/verdaccio/verdaccio:6.11.1
 
       - name: Run e2e tests (forge-e2e feature)
         id: tests

README.md

@@ -239,8 +239,8 @@ The guides below are also a site, with navigation and search:
 - **[Format plugins](./docs/format-plugins.md)** — further package formats,
   built into the server: design, operation, and writing one.
 - **[Importing repositories](./docs/import.md)** — copy a hosted repository
-  of Nexus, Artifactory, a registry or another Forge (NuGet, Docker) into
-  Forge through the format's own protocol: dry run, conflicts, resuming, and
+  of Nexus, Artifactory, Verdaccio, a registry or another Forge (NuGet, npm,
+  Docker) into Forge through the format's own protocol: dry run, conflicts, resuming, and
   the cutover.
 
 Run `cargo xtask help` for the complete command list.

crates/core/src/net/upstream.rs

@@ -786,8 +786,11 @@ impl UpstreamHttp {
     }
 
     /// GET a metadata body, size-capped and spooled; a gzip-encoded body is
-    /// inflated (on the blocking pool) under the same cap.
-    async fn get_metadata_body(
+    /// inflated (on the blocking pool) under the same cap. `Ok(None)` on a
+    /// 404; any other non-2xx is an error. A large document (an npm
+    /// packument) is read from the spool as it streams by, rather than held
+    /// in memory whole.
+    pub async fn get_metadata_body(
         &self,
         url: &str,
         auth: Option<&UpstreamAuth<'_>>,

crates/core/src/repo/hosted.rs

@@ -827,6 +827,22 @@ impl HostedWrite {
             ))
         };
 
+        // An import of a package the source named otherwise stores nothing.
+        if let Some((id, version)) = import.and_then(|options| options.expected.as_ref()) {
+            let same_version = self
+                .format
+                .normalize_version(version)
+                .is_ok_and(|v| v == parsed.version);
+            if self.format.lower_id(id) != parsed.id || !same_version {
+                return Err(WriteError::InvalidPackage(
+                    crate::domain::FormatError::InvalidPackage(format!(
+                        "the package is {} {}, not {id} {version}",
+                        parsed.original_id, parsed.version
+                    )),
+                ));
+            }
+        }
+
         let mut publish_context =
             self.publish_policy
                 .as_ref()

crates/core/src/repo/mod.rs

@@ -214,6 +214,12 @@ pub struct ImportOptions {
     pub uploader: Option<String>,
     /// Where it came from, for the audit trail (the source's URL).
     pub source: String,
+    /// The package the source says this is, `(id, version)` as the format
+    /// writes them: a package that turns out to be another one is refused
+    /// before anything is written ([`crate::domain::FormatError`]). An npm
+    /// registry names a tarball in a package's packument, and the tarball's
+    /// `package.json` must agree.
+    pub expected: Option<(String, String)>,
 }
 
 /// What an import did with one package version.

crates/db/src/component_dao.rs

@@ -614,6 +614,30 @@ pub async fn set_imported_times(
     Ok(())
 }
 
+/// Why version `version` of `name` is deprecated: `None` when there is no
+/// such version, `Some(None)` when it is not deprecated.
+pub async fn deprecation<'e, E>(
+    executor: E,
+    repository_id: i64,
+    name: &str,
+    version: &str,
+) -> sqlx::Result<Option<Option<String>>>
+where
+    E: sqlx::PgExecutor<'e>,
+{
+    sqlx::query_scalar!(
+        r#"
+        SELECT deprecation FROM component
+        WHERE repository_id = $1 AND name = $2 AND version = $3
+        "#,
+        repository_id,
+        name,
+        version,
+    )
+    .fetch_optional(executor)
+    .await
+}
+
 /// Deprecate version `version` of `name` with `message`, or (`None`) no
 /// longer. Returns whether the version exists.
 pub async fn set_deprecation<'e, E>(

crates/db/src/import_dao.rs

@@ -47,7 +47,8 @@ pub struct RepositoryImport {
     pub state: String,
     /// `scan` or `copy`.
     pub phase: String,
-    /// What the scan walks: `search`, `catalog` or `names`.
+    /// What the scan walks: `search`, `catalog`, `all` (npm's `/-/all`) or
+    /// `names`.
     pub scan_listing: String,
     #[serde(skip)]
     pub cursor: Option<String>,
@@ -81,7 +82,7 @@ pub struct NewImport<'a> {
     pub on_conflict: &'a str,
     pub concurrency: i32,
     pub dry_run: bool,
-    /// Where the scan starts: `search`, `catalog` or `names`.
+    /// Where the scan starts: `search`, `catalog`, `all` or `names`.
     pub scan_listing: &'a str,
     pub source_server: Option<&'a str>,
     pub source_total: Option<i64>,
@@ -442,7 +443,9 @@ pub struct ImportItem {
     pub id: i64,
     #[serde(skip)]
     pub import_id: i64,
-    /// `component` or `manifest`.
+    /// `component` (a version, a Docker tag), `manifest` (a Docker referrer
+    /// held by digest) or `tag` (an npm dist-tag: `name` the package,
+    /// `version` the tag).
     pub kind: String,
     pub name: String,
     pub version: String,
@@ -556,6 +559,40 @@ pub async fn finish_item(
     Ok(())
 }
 
+/// What the newest earlier import into repository `repository_id` (before
+/// import `import_id`, a dry run none of them) recorded of the item `kind`
+/// `name` `version` it copied or found identical: its `source`. Where Forge
+/// still holds what that import set (a dist-tag where it pointed it, a
+/// deprecation it gave), a later import follows the source.
+pub async fn earlier_source(
+    pool: &PgPool,
+    repository_id: i64,
+    import_id: i64,
+    kind: &str,
+    name: &str,
+    version: &str,
+) -> sqlx::Result<Option<JsonValue>> {
+    sqlx::query_scalar!(
+        r#"
+        SELECT i.source
+        FROM repository_import_item i
+        JOIN repository_import r ON r.id = i.import_id
+        WHERE r.repository_id = $1 AND r.id < $2 AND NOT r.dry_run
+          AND i.kind = $3 AND i.name = $4 AND i.version = $5
+          AND i.state IN ('imported', 'replaced', 'identical')
+        ORDER BY r.id DESC
+        LIMIT 1
+        "#,
+        repository_id,
+        import_id,
+        kind,
+        name,
+        version,
+    )
+    .fetch_optional(pool)
+    .await
+}
+
 /// Forget an item: a manifest the source holds by digest that turned out not
 /// to be a referrer of an imported image.
 pub async fn drop_item(pool: &PgPool, item_id: i64) -> sqlx::Result<()> {