use std::collections::{BTreeMap, BTreeSet, HashSet};
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use bytes::Bytes;
use cap_fs_ext::OpenOptionsFollowExt;
use futures::{
    FutureExt,
    channel::oneshot,
    future::{BoxFuture, Shared},
};

type Acquisition = Shared<BoxFuture<'static, Result<Arc<Vec<u8>>, RegistryError>>>;
type RemoteAcquisition = Shared<BoxFuture<'static, Result<Arc<RemoteArtifact>, RegistryError>>>;
pub(super) type NpmAcquisition =
    Shared<BoxFuture<'static, Result<Arc<npm::NpmArtifact>, npm::NpmError>>>;

use http_body_util::{BodyExt, Empty};
use hyper::Request;
use hyper_rustls::{HttpsConnector, HttpsConnectorBuilder};
use hyper_util::{
    client::legacy::{Client, connect::HttpConnector},
    rt::TokioExecutor,
};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use url::Url;

mod error;
mod http;
mod jsr;
pub mod npm;

pub use error::RegistryError;
use http::{acquire_remote, http_url, remote_blob, remote_vendor_path, validate_remote_entry};
pub use jsr::{JsrRequest, parse_jsr_request};
use jsr::{PackageIndex, VersionMetadata, manifest_path, vendor_path};

pub(crate) use jsr::validate_name;

pub const MAX_ARTIFACT_SIZE: usize = 32 * 1024 * 1024;

/// Progress during package acquisition. Callbacks must not block or call
/// back into the store. Most callbacks run on worker threads; npm solver
/// callbacks run on the caller thread. No callback runs while in-memory
/// store or resolver state locks are held.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ProgressEvent {
    NpmMetadata { name: String },
    NpmArchive { name: String, version: String },
    NpmMaterialized { name: String, version: String },
    SourceDownload { url: String },
    NpmMaterializing { name: String, version: String },
    NpmResolving,
}

pub type ProgressSink = Arc<dyn Fn(&ProgressEvent) + Send + Sync>;

struct AcquisitionContext {
    client: Client<HttpsConnector<HttpConnector>, Empty<Bytes>>,
    options: Options,
    authority: Option<crate::authority::SourceAuthority>,
    progress: Option<ProgressSink>,
}

#[derive(Clone, Debug)]
pub struct Options {
    pub jsr_registry: Url,
    pub npm_registry: Url,
    pub cache: PathBuf,
    pub lock: PathBuf,
    pub project_root: PathBuf,
    pub offline: bool,
    pub frozen: bool,
    pub update: bool,
    /// Read and resolve against the lock in memory, but never write it back.
    pub ephemeral_lock: bool,
    pub vendor: Option<PathBuf>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct Lockfile {
    pub version: u32,
    pub jsr_registry: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub npm_registry: Option<String>,
    pub specifiers: BTreeMap<String, String>,
    pub packages: BTreeMap<String, String>,
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub remote: BTreeMap<String, RemoteEntry>,
    #[serde(default, skip_serializing_if = "NpmLock::is_empty")]
    pub npm: NpmLock,
}

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct NpmLock {
    #[serde(default)]
    pub specifiers: BTreeMap<String, String>,
    #[serde(default)]
    pub packages: BTreeMap<String, NpmPin>,
    #[serde(default)]
    pub roots: BTreeMap<String, String>,
    #[serde(default)]
    pub instances: BTreeMap<String, NpmInstancePin>,
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub types: BTreeMap<String, Option<String>>,
}

impl NpmLock {
    fn is_empty(&self) -> bool {
        self.specifiers.is_empty()
            && self.packages.is_empty()
            && self.roots.is_empty()
            && self.instances.is_empty()
            && self.types.is_empty()
    }
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct NpmInstancePin {
    pub package: String,
    #[serde(default)]
    pub deps: BTreeMap<String, String>,
    #[serde(default)]
    pub peers: BTreeMap<String, String>,
}

// Keep unvisited legacy file maps inline until their sidecars are persisted.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct NpmPin {
    pub tarball: String,
    pub integrity: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub manifest: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub files: Option<BTreeMap<String, String>>,
}

impl NpmPin {
    fn is_legacy(&self) -> bool {
        self.files.is_some()
    }

    // Conflict checks accept migration; structural equality still detects lock edits.
    fn same_content(&self, other: &Self) -> bool {
        self.tarball == other.tarball
            && self.integrity == other.integrity
            && matches!((self.digest(), other.digest()), (Some(a), Some(b)) if a == b)
    }

    fn digest(&self) -> Option<String> {
        if let Some(manifest) = &self.manifest {
            return Some(manifest.clone());
        }
        self.files
            .as_ref()
            .and_then(|files| npm::storage::manifest_digest(files).ok())
    }
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RemoteEntry {
    pub checksum: String,
    pub url: String,
}

struct RemoteArtifact {
    bytes: Vec<u8>,
    entry: RemoteEntry,
}

#[derive(Default, Clone)]
struct Used {
    specifiers: BTreeSet<String>,
    packages: BTreeSet<String>,
    sources: BTreeSet<String>,
    pending_specifiers: BTreeSet<String>,
    pending_packages: BTreeSet<String>,
    remote: BTreeSet<String>,
    pending_remote: BTreeSet<String>,
    configs: BTreeSet<String>,
    npm_specifiers: BTreeSet<String>,
    npm_packages: BTreeSet<String>,
    pending_npm_specifiers: BTreeSet<String>,
    pending_npm_packages: BTreeSet<String>,
    npm_roots: BTreeSet<String>,
    npm_instances: BTreeSet<String>,
    pending_npm_roots: BTreeSet<String>,
    pending_npm_instances: BTreeSet<String>,
    npm_types: BTreeSet<String>,
    pending_npm_types: BTreeSet<String>,
    npm_types_discovered: BTreeSet<String>,
}

/// Acquires packages on a caller-owned Tokio runtime, which must stay alive and driven.
/// Synchronous consumers need a runtime whose workers run independently.
pub struct PackageStore {
    options: Options,
    authority: Option<crate::authority::SourceAuthority>,
    progress: Option<ProgressSink>,
    npm_packuments: Arc<Mutex<BTreeMap<String, npm::SharedPackument>>>,
    npm_downloads: Arc<tokio::sync::Semaphore>,
    npm_types_gate: tokio::sync::Mutex<()>,
    lock: Mutex<Lockfile>,
    used: Mutex<Used>,
    metadata: Mutex<BTreeMap<String, Arc<VersionMetadata>>>,
    indexes: Mutex<BTreeMap<String, Arc<PackageIndex>>>,
    inflight: Mutex<BTreeMap<String, Acquisition>>,
    remote_acquisitions: Mutex<BTreeMap<String, RemoteAcquisition>>,
    remote_bases: Mutex<BTreeMap<String, String>>,
    npm_inflight: Mutex<BTreeMap<String, NpmAcquisition>>,
    runtime: tokio::runtime::Handle,
    client: Client<HttpsConnector<HttpConnector>, Empty<Bytes>>,
}

impl PackageStore {
    pub fn new(runtime: tokio::runtime::Handle, options: Options) -> Result<Self, RegistryError> {
        Self::construct(runtime, options, None)
    }

    /// Establish source authority before any lockfile or source IO, so a
    /// Read graph never parses an unreadable lock. Execute skips the gate
    /// to preserve existing host-side lock handling. Use one authority for
    /// the graph, packages, and store.
    pub fn new_with_authority(
        runtime: tokio::runtime::Handle,
        options: Options,
        authority: crate::authority::SourceAuthority,
    ) -> Result<Self, RegistryError> {
        Self::construct(runtime, options, Some(authority))
    }

    fn construct(
        runtime: tokio::runtime::Handle,
        options: Options,
        authority: Option<crate::authority::SourceAuthority>,
    ) -> Result<Self, RegistryError> {
        if options.update && options.frozen {
            return Err(RegistryError::UpdateAndFrozen);
        }
        validate_registry(&options.jsr_registry)?;
        validate_registry(&options.npm_registry)?;
        if !options.project_root.is_absolute() {
            return Err(RegistryError::NpmProjectRootNotAbsolute);
        }
        let mut lock = match read_lock(&options.lock, authority.as_ref())? {
            Some(lock) => lock,
            None if options.frozen => {
                return Err(RegistryError::FrozenLockfileDoesNotExist {
                    path: options.lock.clone(),
                });
            }
            None => Lockfile {
                version: 1,
                jsr_registry: options.jsr_registry.to_string(),
                npm_registry: None,
                specifiers: BTreeMap::new(),
                packages: BTreeMap::new(),
                remote: BTreeMap::new(),
                npm: NpmLock::default(),
            },
        };
        validate_lock(&lock, &options.jsr_registry, &options.npm_registry)?;
        if options.update {
            lock.specifiers.clear();
            lock.npm.specifiers.clear();
            lock.npm.roots.clear();
            lock.npm.instances.clear();
        }
        let connector = HttpsConnectorBuilder::new()
            .with_webpki_roots()
            .https_or_http()
            .enable_http1()
            .enable_http2()
            .build();
        let client = Client::builder(TokioExecutor::new()).build(connector);
        Ok(Self {
            options,
            authority,
            progress: None,
            npm_packuments: Arc::new(Mutex::new(BTreeMap::new())),
            npm_downloads: Arc::new(tokio::sync::Semaphore::new(
                npm::MAX_NPM_CONCURRENT_DOWNLOADS,
            )),
            npm_types_gate: tokio::sync::Mutex::new(()),
            lock: Mutex::new(lock),
            used: Mutex::default(),
            metadata: Mutex::default(),
            indexes: Mutex::default(),
            inflight: Mutex::default(),
            remote_acquisitions: Mutex::default(),
            remote_bases: Mutex::default(),
            npm_inflight: Mutex::default(),
            runtime,
            client,
        })
    }

    pub fn with_capabilities(mut self, caps: Arc<crate::caps::Capabilities>) -> Self {
        self.authority = Some(crate::authority::SourceAuthority::execute(caps));
        self
    }

    pub fn with_progress(mut self, sink: ProgressSink) -> Self {
        self.progress = Some(sink);
        self
    }

    pub fn options(&self) -> &Options {
        &self.options
    }

    pub async fn remote_source(&self, name: &str) -> Result<String, RegistryError> {
        if name.starts_with(self.options.jsr_registry.as_str()) {
            return Ok(self.source(name).await?);
        }
        self.remote_artifact(name, |_| Ok(())).await
    }

    pub async fn remote_config(&self, name: &str) -> Result<String, RegistryError> {
        if name.starts_with(self.options.jsr_registry.as_str()) {
            return Err(RegistryError::HttpConfigUsesJsrRegistry { name: name.to_owned() });
        }
        let source = self
            .remote_artifact(name, |entry| {
                if entry.url.starts_with(self.options.jsr_registry.as_str()) {
                    Err(RegistryError::HttpConfigRedirectsIntoJsr { url: entry.url.clone() })
                } else {
                    Ok(())
                }
            })
            .await?;
        self.used.lock().unwrap().configs.insert(name.to_owned());
        Ok(source)
    }

    async fn remote_artifact(
        &self,
        name: &str,
        validate: impl FnOnce(&RemoteEntry) -> Result<(), RegistryError>,
    ) -> Result<String, RegistryError> {
        let url = http_url(name)?;
        let access = Access::Module(url.clone());
        access.check(self.authority.as_ref())?;
        let expected = self.lock.lock().unwrap().remote.get(name).cloned();
        if expected.is_none() && self.options.frozen {
            return Err(RegistryError::FrozenLockfileMissingRemote { name: name.to_owned() });
        }
        let future = {
            let mut acquisitions = self.remote_acquisitions.lock().unwrap();
            acquisitions
                .entry(name.to_owned())
                .or_insert_with(|| {
                    let context = AcquisitionContext {
                        options: self.options.clone(),
                        client: self.client.clone(),
                        authority: self.authority.clone(),
                        progress: self.progress.clone(),
                    };
                    let access = access.clone();
                    let name = name.to_owned();
                    let (send, receive) = oneshot::channel();
                    self.runtime.spawn(async move {
                        let result = acquire_remote(context, name, url, expected, access)
                            .await
                            .map(Arc::new);
                        let _ = send.send(result);
                    });
                    async move {
                        receive
                            .await
                            .map_err(|_| RegistryError::HttpAcquisitionWorkerStopped)?
                            .map_err(Into::into)
                    }
                    .boxed()
                    .shared()
                })
                .clone()
        };
        let result = future.clone().await;
        if result.is_err() {
            let mut acquisitions = self.remote_acquisitions.lock().unwrap();
            if acquisitions
                .get(name)
                .is_some_and(|current| current.ptr_eq(&future))
            {
                acquisitions.remove(name);
            }
        }
        access.check(self.authority.as_ref())?;
        let artifact = result?;
        validate(&artifact.entry)?;
        if let Some(authority) = self.authority.as_ref() {
            authority.check_remote_final(&artifact.entry.url)?;
        }
        let source = String::from_utf8(artifact.bytes.clone())
            .map_err(|_| RegistryError::ModuleNotUtf8 { name: name.to_owned() })?;
        let mut lock = self.lock.lock().unwrap();
        if !self.options.update
            && let Some(existing) = lock.remote.get(name)
            && existing != &artifact.entry
        {
            return Err(RegistryError::RemoteIntegrityMismatch { name: name.to_owned() });
        }
        let mut used = self.used.lock().unwrap();
        used.remote.insert(name.to_owned());
        if lock.remote.get(name) != Some(&artifact.entry) || self.options.update {
            used.pending_remote.insert(name.to_owned());
        }
        lock.remote.insert(name.to_owned(), artifact.entry.clone());
        self.remote_bases
            .lock()
            .unwrap()
            .insert(name.to_owned(), artifact.entry.url.clone());
        Ok(source)
    }

    pub fn remote_base(&self, name: &str) -> Result<String, RegistryError> {
        http_url(name)?;
        if name.starts_with(self.options.jsr_registry.as_str()) {
            return Ok(name.to_owned());
        }
        self.remote_bases
            .lock()
            .unwrap()
            .get(name)
            .cloned()
            .ok_or_else(|| RegistryError::HttpModuleNotAcquired { name: name.to_owned() })
    }

    /// Reauthorize remote access by acquired identity under the caller's
    /// authority: the requested URL always, plus the recorded final URL
    /// under Read. Execute keeps requested-only redirect semantics.
    pub fn check_acquired_remote(
        &self,
        name: &str,
        authority: &crate::authority::SourceAuthority,
    ) -> Result<(), RegistryError> {
        let url = http_url(name)?;
        Access::Module(url).check(Some(authority))?;
        if let Some(final_url) = self.remote_bases.lock().unwrap().get(name).cloned() {
            authority.check_remote_final(&final_url)?;
        }
        Ok(())
    }

    async fn bytes(
        &self,
        url: Url,
        expected: Option<String>,
        refresh: bool,
        access: Access,
    ) -> Result<Vec<u8>, RegistryError> {
        access.check(self.authority.as_ref())?;
        let key = url.to_string();
        let future = {
            let mut inflight = self.inflight.lock().unwrap();
            inflight
                .entry(key.clone())
                .or_insert_with(|| {
                    let context = AcquisitionContext {
                        options: self.options.clone(),
                        client: self.client.clone(),
                        authority: self.authority.clone(),
                        progress: self.progress.clone(),
                    };
                    let access = access.clone();
                    let expected = expected.clone();
                    let (send, receive) = oneshot::channel();
                    self.runtime.spawn(async move {
                        let result = acquire(context, url, expected, refresh, access)
                            .await
                            .map(Arc::new);
                        let _ = send.send(result);
                    });
                    async move {
                        receive
                            .await
                            .map_err(|_| RegistryError::PackageAcquisitionWorkerStopped)?
                    }
                    .boxed()
                    .shared()
                })
                .clone()
        };
        let result = future.clone().await;
        {
            let mut inflight = self.inflight.lock().unwrap();
            if inflight
                .get(&key)
                .is_some_and(|current| current.ptr_eq(&future))
            {
                inflight.remove(&key);
            }
        }
        let bytes = result?;
        access.check(self.authority.as_ref())?;
        if let Some(expected) = expected {
            verify(&bytes, &expected)?;
        }
        Ok((*bytes).clone())
    }

    pub fn save(&self) -> Result<(), RegistryError> {
        let (lock, mut used) = {
            let lock = self.lock.lock().unwrap();
            (lock.clone(), self.used.lock().unwrap().clone())
        };
        let pending_types = std::mem::take(&mut used.pending_npm_types);
        used.npm_types.retain(|key| !pending_types.contains(key));
        self.save_used(lock, used, false)
    }

    pub fn save_graph(
        &self,
        specifiers: &BTreeSet<String>,
        modules: &HashSet<String>,
    ) -> Result<(), RegistryError> {
        let packages: BTreeSet<_> = modules
            .iter()
            .filter(|name| name.starts_with(self.options.jsr_registry.as_str()))
            .map(|name| {
                self.module_base(name)
                    .map(|base| format!("{}_meta.json", base.as_str().trim_end_matches('/')))
            })
            .collect::<Result<_, _>>()?;
        let (lock, mut used) = {
            let lock = self.lock.lock().unwrap();
            (lock.clone(), self.used.lock().unwrap().clone())
        };
        used.specifiers.retain(|key| specifiers.contains(key));
        used.packages.retain(|url| packages.contains(url));
        used.pending_specifiers
            .retain(|key| specifiers.contains(key));
        used.pending_packages.retain(|url| packages.contains(url));
        used.remote
            .retain(|url| modules.contains(url) || used.configs.contains(url));
        used.pending_remote
            .retain(|url| modules.contains(url) || used.configs.contains(url));
        // npm entries persist only for this successful walk: acquisitions
        // record usage while preparing, but a resolver success the module
        // graph never completes (bad syntax, builtins, CJS) must not leak
        // into the lock. Exact tracked roots stay, plus declared roots
        // sharing a successful batch context so unused dependencies replay
        // under frozen; unrelated or failed contexts are dropped, as are
        // the instances, pins, and selections only those reference.
        let mut kept_roots: BTreeSet<String> = specifiers
            .iter()
            .filter(|key| used.npm_roots.contains(*key))
            .cloned()
            .collect();
        let suffixes: BTreeSet<String> = kept_roots
            .iter()
            .filter_map(|key| key.split_once('\0').map(|(_, suffix)| suffix.to_owned()))
            .collect();
        for key in &used.npm_roots {
            if let Some((_, suffix)) = key.split_once('\0')
                && suffixes.contains(suffix)
            {
                kept_roots.insert(key.clone());
            }
        }
        let mut kept_instances: BTreeSet<String> = kept_roots
            .iter()
            .filter_map(|key| lock.npm.roots.get(key))
            .cloned()
            .collect();
        let mut frontier: Vec<String> = kept_instances.iter().cloned().collect();
        while let Some(id) = frontier.pop() {
            if let Some(pin) = lock.npm.instances.get(&id) {
                for next in pin.deps.values().chain(pin.peers.values()) {
                    if kept_instances.insert(next.clone()) {
                        frontier.push(next.clone());
                    }
                }
            }
        }
        let kept_packages: BTreeSet<String> = kept_instances
            .iter()
            .filter_map(|id| lock.npm.instances.get(id))
            .map(|pin| pin.package.clone())
            .collect();
        used.npm_roots.retain(|key| kept_roots.contains(key));
        used.pending_npm_roots
            .retain(|key| kept_roots.contains(key));
        used.npm_instances
            .retain(|key| kept_instances.contains(key));
        used.pending_npm_instances
            .retain(|key| kept_instances.contains(key));
        used.npm_packages.retain(|key| kept_packages.contains(key));
        used.pending_npm_packages
            .retain(|key| kept_packages.contains(key));
        used.npm_specifiers.retain(|key| {
            npm::parse_npm_request(key).is_ok_and(|request| {
                lock.npm.specifiers.get(key).is_some_and(|version| {
                    kept_packages.iter().any(|package| {
                        package.as_str()
                            == format!(
                                "{}|{}|{version}",
                                self.options.npm_registry.as_str(),
                                request.name
                            )
                    })
                })
            })
        });
        used.pending_npm_specifiers
            .retain(|key| used.npm_specifiers.contains(key));
        used.npm_types.retain(|key| {
            kept_packages.contains(key)
                && lock.npm.types.get(key).is_some_and(|request| {
                    request.as_ref().is_none_or(|request| {
                        kept_roots.iter().any(|root| {
                            root == request || root.starts_with(&format!("{request}\0ctx:"))
                        })
                    })
                })
        });
        used.pending_npm_types
            .retain(|key| used.npm_types.contains(key));
        self.save_used(lock, used, false)
    }

    pub fn prune(&self) -> Result<(), RegistryError> {
        if self.options.frozen {
            return Err(RegistryError::PruneFrozen);
        }
        let (lock, mut used) = {
            let lock = self.lock.lock().unwrap();
            (lock.clone(), self.used.lock().unwrap().clone())
        };
        // Prune the npm graph to its full closure: saved roots, every
        // instance reachable from them, and the specifiers and pins those
        // reference. Roots only a failed walk prepared are pending without
        // a file entry, so they (and their pending writes) drop out here;
        // every CLI prune follows the walk save that files successes.
        // Computed includes from earlier walks stay when reached.
        let disk = read_lock(&self.options.lock, self.authority.as_ref())?;
        let disk_roots: BTreeSet<String> = disk
            .iter()
            .flat_map(|disk| disk.npm.roots.keys().cloned())
            .collect();
        let disk_packages: BTreeSet<String> = disk
            .iter()
            .flat_map(|disk| disk.npm.packages.keys().cloned())
            .collect();
        let mut reachable = BTreeSet::new();
        let mut frontier: Vec<String> = used
            .npm_roots
            .iter()
            .filter(|key| disk_roots.contains(key.as_str()))
            .filter_map(|key| lock.npm.roots.get(key).cloned())
            .collect();
        while let Some(instance) = frontier.pop() {
            if !reachable.insert(instance.clone()) {
                continue;
            }
            if let Some(pin) = lock.npm.instances.get(&instance) {
                frontier.extend(pin.deps.values().cloned());
                frontier.extend(pin.peers.values().cloned());
            }
        }
        used.npm_roots.retain(|key| {
            lock.npm
                .roots
                .get(key)
                .is_some_and(|instance| reachable.contains(instance))
        });
        used.pending_npm_roots
            .retain(|key| used.npm_roots.contains(key));
        used.npm_instances.retain(|key| reachable.contains(key));
        used.pending_npm_instances
            .retain(|key| reachable.contains(key));
        // Direct acquisitions record packages without graph roots; ones an
        // earlier save persisted stay, while pins no successful walk kept
        // never land. Acquisitions alone cannot tell those apart, but the
        // lockfile can: only file-backed pins join the closure here.
        let mut packages: BTreeSet<String> = used
            .npm_packages
            .iter()
            .filter(|key| disk_packages.contains(key.as_str()))
            .cloned()
            .collect();
        packages.extend(
            reachable
                .iter()
                .filter_map(|instance| lock.npm.instances.get(instance))
                .map(|pin| pin.package.clone()),
        );
        used.npm_packages.retain(|key| packages.contains(key));
        used.pending_npm_packages
            .retain(|key| packages.contains(key));
        used.npm_specifiers.retain(|key| {
            npm::parse_npm_request(key).is_ok_and(|request| {
                lock.npm.specifiers.get(key).is_some_and(|version| {
                    packages.iter().any(|package| {
                        package.as_str()
                            == format!(
                                "{}|{}|{version}",
                                self.options.npm_registry.as_str(),
                                request.name
                            )
                    })
                })
            })
        });
        used.pending_npm_specifiers
            .retain(|key| used.npm_specifiers.contains(key));
        used.npm_types.retain(|key| {
            !used.pending_npm_types.contains(key)
                && packages.contains(key)
                && lock.npm.types.get(key).is_some_and(|request| {
                    request.as_ref().is_none_or(|request| {
                        let Ok((registry, _, _)) = npm::parse_npm_key(key) else {
                            return false;
                        };
                        let Ok(request) = npm::parse_npm_request(request) else {
                            return false;
                        };
                        lock.npm
                            .specifiers
                            .get(&request.request_key())
                            .is_some_and(|version| {
                                packages
                                    .contains(&format!("{}|{}|{version}", registry, request.name))
                            })
                    })
                })
        });
        used.pending_npm_types
            .retain(|key| used.npm_types.contains(key));
        self.save_used(lock, used, true)
    }

    fn save_used(&self, lock: Lockfile, used: Used, prune: bool) -> Result<(), RegistryError> {
        if self.options.frozen
            || self.options.ephemeral_lock
            || (!prune
                && used.packages.is_empty()
                && used.specifiers.is_empty()
                && used.remote.is_empty()
                && used.npm_specifiers.is_empty()
                && used.npm_packages.is_empty()
                && used.npm_roots.is_empty()
                && used.npm_instances.is_empty()
                && used.npm_types.is_empty())
        {
            return Ok(());
        }
        if !prune
            && used.pending_packages.is_empty()
            && used.pending_specifiers.is_empty()
            && used.pending_remote.is_empty()
            && used.pending_npm_specifiers.is_empty()
            && used.pending_npm_packages.is_empty()
            && used.pending_npm_roots.is_empty()
            && used.pending_npm_instances.is_empty()
            && used.pending_npm_types.is_empty()
        {
            if let Some(disk) = read_lock(&self.options.lock, self.authority.as_ref())? {
                self.merge(&lock, &used, disk)?;
            }
            return Ok(());
        }
        let path = &self.options.lock;
        let parent = path.parent().ok_or(RegistryError::LockfileNeedsParent)?;
        std::fs::create_dir_all(parent)?;
        update_file(
            path,
            || read_lock_bytes(path, self.authority.as_ref()),
            |disk| {
                let mut merged = match disk {
                    Some(bytes) => parse_lock(path, bytes)?,
                    None => Lockfile {
                        version: 1,
                        jsr_registry: lock.jsr_registry.clone(),
                        npm_registry: None,
                        specifiers: BTreeMap::new(),
                        packages: BTreeMap::new(),
                        remote: BTreeMap::new(),
                        npm: NpmLock::default(),
                    },
                };
                let previous = merged.clone();
                merged = self.merge(&lock, &used, merged)?;
                if prune {
                    merged
                        .specifiers
                        .retain(|key, _| used.specifiers.contains(key));
                    merged.packages.retain(|url, _| used.packages.contains(url));
                    merged.remote.retain(|url, _| used.remote.contains(url));
                    merged
                        .npm
                        .specifiers
                        .retain(|key, _| used.npm_specifiers.contains(key));
                    merged
                        .npm
                        .packages
                        .retain(|key, _| used.npm_packages.contains(key));
                    merged
                        .npm
                        .roots
                        .retain(|key, _| used.npm_roots.contains(key));
                    merged
                        .npm
                        .instances
                        .retain(|key, _| used.npm_instances.contains(key));
                    merged
                        .npm
                        .types
                        .retain(|key, _| used.npm_types.contains(key));
                }
                if merged == previous {
                    return Ok(None);
                }
                let value = serde_json::to_value(&merged)?;
                let mut bytes = serde_fjson::stringify(&value)?.into_bytes();
                bytes.push(b'\n');
                Ok(Some(bytes))
            },
        )?;
        let mut current = self.used.lock().unwrap();
        for key in used.pending_specifiers {
            current.pending_specifiers.remove(&key);
        }
        for url in used.pending_packages {
            current.pending_packages.remove(&url);
        }
        for url in used.pending_remote {
            current.pending_remote.remove(&url);
        }
        for key in used.pending_npm_specifiers {
            current.pending_npm_specifiers.remove(&key);
        }
        for key in used.pending_npm_packages {
            current.pending_npm_packages.remove(&key);
        }
        for key in used.pending_npm_roots {
            current.pending_npm_roots.remove(&key);
        }
        for key in used.pending_npm_instances {
            current.pending_npm_instances.remove(&key);
        }
        for key in used.pending_npm_types {
            current.pending_npm_types.remove(&key);
        }
        Ok(())
    }

    fn merge(
        &self,
        lock: &Lockfile,
        used: &Used,
        mut merged: Lockfile,
    ) -> Result<Lockfile, RegistryError> {
        validate_lock(
            &merged,
            &self.options.jsr_registry,
            &self.options.npm_registry,
        )?;
        for key in &used.specifiers {
            if !self.options.update
                && merged
                    .specifiers
                    .get(key)
                    .is_some_and(|value| value != &lock.specifiers[key])
            {
                return Err(RegistryError::LockfileChanged { key: key.to_owned() });
            }
        }
        for url in &used.packages {
            if merged
                .packages
                .get(url)
                .is_some_and(|value| value != &lock.packages[url])
            {
                return Err(RegistryError::LockedMetadataChanged { url: url.to_owned() });
            }
        }
        for url in &used.remote {
            if !self.options.update
                && merged
                    .remote
                    .get(url)
                    .is_some_and(|value| value != &lock.remote[url])
            {
                return Err(RegistryError::LockfileChanged { key: url.to_owned() });
            }
        }
        for url in &used.pending_remote {
            merged.remote.insert(url.clone(), lock.remote[url].clone());
        }
        for key in &used.pending_specifiers {
            let version = &lock.specifiers[key];
            merged.specifiers.insert(key.clone(), version.clone());
        }
        for url in &used.pending_packages {
            let checksum = &lock.packages[url];
            merged.packages.insert(url.clone(), checksum.clone());
        }
        for key in &used.npm_specifiers {
            if !self.options.update
                && merged
                    .npm
                    .specifiers
                    .get(key)
                    .is_some_and(|value| value != &lock.npm.specifiers[key])
            {
                return Err(RegistryError::LockfileChanged { key: key.to_owned() });
            }
        }
        for key in &used.npm_packages {
            if merged
                .npm
                .packages
                .get(key)
                .is_some_and(|value| !value.same_content(&lock.npm.packages[key]))
            {
                return Err(RegistryError::LockedNpmPackageChanged { key: key.to_owned() });
            }
        }
        for key in &used.pending_npm_specifiers {
            let version = &lock.npm.specifiers[key];
            merged.npm.specifiers.insert(key.clone(), version.clone());
        }
        for key in &used.pending_npm_packages {
            let pin = &lock.npm.packages[key];
            merged.npm.packages.insert(key.clone(), pin.clone());
        }
        for key in &used.npm_roots {
            if !self.options.update
                && merged
                    .npm
                    .roots
                    .get(key)
                    .is_some_and(|value| value != &lock.npm.roots[key])
            {
                return Err(RegistryError::LockfileChanged { key: key.to_owned() });
            }
        }
        for key in &used.npm_instances {
            if merged
                .npm
                .instances
                .get(key)
                .is_some_and(|value| value != &lock.npm.instances[key])
            {
                return Err(RegistryError::LockedNpmInstanceChanged { id: key.to_owned() });
            }
        }
        for key in &used.pending_npm_roots {
            let instance = &lock.npm.roots[key];
            merged.npm.roots.insert(key.clone(), instance.clone());
        }
        for key in &used.pending_npm_instances {
            let pin = &lock.npm.instances[key];
            merged.npm.instances.insert(key.clone(), pin.clone());
        }
        for key in &used.npm_types {
            if !self.options.update
                && !used.pending_npm_types.contains(key)
                && merged.npm.types.get(key) != lock.npm.types.get(key)
            {
                return Err(RegistryError::LockfileChanged { key: key.clone() });
            }
        }
        for key in &used.pending_npm_types {
            let association = lock.npm.types[key].clone();
            if !self.options.update
                && merged
                    .npm
                    .types
                    .get(key)
                    .is_some_and(|existing| existing != &association)
            {
                return Err(RegistryError::LockfileChanged { key: key.clone() });
            }
            merged.npm.types.insert(key.clone(), association);
        }
        if merged.npm.is_empty() {
            merged.version = 1;
            merged.npm_registry = None;
        } else {
            merged.version = 2;
            merged.npm_registry = Some(self.options.npm_registry.to_string());
        }
        Ok(merged)
    }

    pub fn vendor_relative(&self, name: &str) -> Result<PathBuf, RegistryError> {
        if !name.starts_with(self.options.jsr_registry.as_str()) {
            return Ok(remote_vendor_path(name, &self.remote_base(name)?)?);
        }
        let url = Url::parse(name)?;
        Ok(vendor_path(&self.options.jsr_registry, &url)?)
    }

    pub fn vendor(&self, destination: &Path) -> Result<(), RegistryError> {
        let (lock, used) = {
            let lock = self.lock.lock().unwrap();
            (lock.clone(), self.used.lock().unwrap().clone())
        };
        for name in &used.remote {
            let entry = &lock.remote[name];
            let relative = self.vendor_relative(name)?;
            let bytes = match read_artifact(&remote_blob(&self.options.cache, &entry.checksum))? {
                Some(bytes) => Some(bytes),
                None => match &self.options.vendor {
                    Some(vendor) => read_artifact(&vendor.join(&relative))?,
                    None => None,
                },
            }
            .ok_or_else(|| RegistryError::MissingVendoredArtifact { name: name.to_owned() })?;
            verify(&bytes, &entry.checksum)?;
            let target = destination.join(relative);
            std::fs::create_dir_all(target.parent().unwrap())?;
            atomic_write(&target, &bytes)?;
        }
        for name in used.packages.iter().chain(&used.sources) {
            let url = Url::parse(name)?;
            let relative = vendor_path(&self.options.jsr_registry, &url)?;
            let bytes = match &self.options.vendor {
                Some(vendor) => match read_artifact(&vendor.join(&relative))? {
                    Some(bytes) => Some(bytes),
                    None => {
                        read_artifact(&self.options.cache.join("files").join(artifact_key(name)))?
                    }
                },
                None => read_artifact(&self.options.cache.join("files").join(artifact_key(name)))?,
            }
            .ok_or_else(|| RegistryError::MissingVendoredArtifact { name: name.to_owned() })?;
            let expected = if let Some(hash) = lock.packages.get(name) {
                hash.clone()
            } else {
                let base = self.module_base(name)?;
                let metadata_url = format!("{}_meta.json", base.as_str().trim_end_matches('/'));
                let metadata = self.metadata.lock().unwrap();
                metadata[&metadata_url].manifest[&manifest_path(&base, &url)?]
                    .checksum
                    .clone()
            };
            verify(&bytes, &expected)?;
            let target = destination.join(relative);
            std::fs::create_dir_all(target.parent().unwrap())?;
            atomic_write(&target, &bytes)?;
        }
        let targets: BTreeSet<String> = used
            .npm_instances
            .iter()
            .filter_map(|id| lock.npm.instances.get(id))
            .map(|pin| pin.package.clone())
            .collect();
        for package in &targets {
            let Some(pin_entry) = lock.npm.packages.get(package) else {
                return Err(RegistryError::MissingVendoredArtifact { name: package.to_owned() });
            };
            let (registry, name, version) = npm::parse_npm_key(package)?;
            if registry != self.options.npm_registry {
                return Err(RegistryError::NpmInstanceRegistryMismatch {
                    package: package.to_owned(),
                });
            }
            let source = npm::storage::npm_package_dir(
                &self.options,
                npm::NPM_DEFAULT_INSTANCE,
                &name,
                &version,
            );
            let relative = npm::storage::npm_vendor_relative(
                &self.options,
                npm::NPM_DEFAULT_INSTANCE,
                &self.options.npm_registry,
                &name,
                &version,
            )?;
            // The file map comes from the digest-authenticated sidecar,
            // never from an unverified tree listing.
            let files = npm::storage::load_vendor_source_files(&self.options, package, pin_entry)?;
            for (rel, expected) in &files {
                let bytes = npm::storage::read_npm_project_file(&self.options, &source, rel)?;
                if checksum(&bytes) != *expected {
                    return Err(RegistryError::NpmFileIntegrityMismatch { rel: rel.to_owned() });
                }
                npm::storage::write_vendor_file(destination, &relative.join(rel), &bytes)?;
            }
            npm::storage::write_vendor_manifest(destination, &files)?;
        }
        Ok(())
    }
}

async fn acquire(
    context: AcquisitionContext,
    url: Url,
    expected: Option<String>,
    refresh: bool,
    access: Access,
) -> Result<Vec<u8>, RegistryError> {
    let AcquisitionContext {
        client,
        options,
        authority,
        progress,
    } = context;
    access.check(authority.as_ref())?;
    let key = artifact_key(url.as_str());
    let file = options.cache.join("files").join(&key);
    let cached = || -> Result<Option<Vec<u8>>, RegistryError> {
        if refresh && !options.offline {
            return Ok(None);
        }
        let bytes = match &options.vendor {
            Some(vendor) => {
                match read_artifact(&vendor.join(vendor_path(&options.jsr_registry, &url)?))? {
                    Some(bytes) => Some(bytes),
                    None => read_artifact(&file)?,
                }
            }
            None => read_artifact(&file)?,
        };
        if let (Some(bytes), Some(expected)) = (&bytes, &expected) {
            verify(bytes, expected)
                .map_err(|_| RegistryError::VerifyMismatch { name: url.to_string() })?;
        }
        Ok(bytes)
    };
    if let Some(bytes) = cached()? {
        return Ok(bytes);
    }
    let lock_path = options.cache.join("locks").join(&key);
    std::fs::create_dir_all(lock_path.parent().unwrap())?;
    let _guard = tokio::task::spawn_blocking(move || {
        let mut guard = fslock::LockFile::open(&lock_path)?;
        guard.lock()?;
        Ok::<_, RegistryError>(guard)
    })
    .await
    .map_err(|detail| RegistryError::WorkerJoin { detail: detail.to_string() })??;
    // Recheck after locking because another process may have filled the cache.
    if let Some(bytes) = cached()? {
        return Ok(bytes);
    }
    access.check(authority.as_ref())?;
    if options.offline {
        return Err(RegistryError::OfflineCacheMiss { url: url.to_string() });
    }
    if let Some(sink) = &progress {
        sink(&ProgressEvent::SourceDownload { url: url.to_string() });
        access.check(authority.as_ref())?;
    }
    let download = async {
        let request = Request::get(url.as_str())
            .header("Accept", "application/json, application/typescript, */*")
            .header("User-Agent", concat!("avril/", env!("CARGO_PKG_VERSION")))
            .body(Empty::new())
            .map_err(|detail| RegistryError::HttpRequest { detail: detail.to_string() })?;
        let response = client
            .request(request)
            .await
            .map_err(|detail| RegistryError::HttpRequest { detail: detail.to_string() })?;
        if !response.status().is_success() {
            return Err(RegistryError::RegistryStatus {
                status: response.status().to_string(),
                url: url.to_string(),
            });
        }
        if response
            .headers()
            .get("content-type")
            .and_then(|value| value.to_str().ok())
            .is_some_and(|value| value.starts_with("text/html"))
        {
            return Err(RegistryError::RegistryHtmlArtifact { url: url.to_string() });
        }
        let mut body = response.into_body();
        let mut bytes = Vec::new();
        while let Some(frame) = body.frame().await {
            if let Ok(data) = frame
                .map_err(|error| RegistryError::Hyper(Arc::new(error)))?
                .into_data()
            {
                if bytes.len() + data.len() > MAX_ARTIFACT_SIZE {
                    return Err(RegistryError::RegistryArtifactTooLarge { url: url.to_string() });
                }
                bytes.extend_from_slice(&data);
            }
        }
        Ok::<_, RegistryError>(bytes)
    };
    let bytes = tokio::time::timeout(Duration::from_secs(30), download)
        .await
        .map_err(|_| RegistryError::RegistryTimeout { url: url.to_string() })??;
    if let Some(expected) = &expected {
        verify(&bytes, expected)
            .map_err(|_| RegistryError::VerifyMismatch { name: url.to_string() })?;
    }
    std::fs::create_dir_all(file.parent().unwrap())?;
    access.check(authority.as_ref())?;
    atomic_write(&file, &bytes)?;
    Ok(bytes)
}

#[derive(Clone)]
enum Access {
    Package(Url),
    Module(Url),
}

impl Access {
    fn check(
        &self,
        authority: Option<&crate::authority::SourceAuthority>,
    ) -> Result<(), crate::caps::CapabilitiesError> {
        let Some(authority) = authority else {
            return Ok(());
        };
        match self {
            Self::Package(base) => authority.check_remote_package(base),
            Self::Module(url) => authority.check_remote_module(url),
        }
    }
}

pub fn checksum(bytes: &[u8]) -> String {
    format!("sha256-{:x}", Sha256::digest(bytes))
}

pub fn artifact_key(url: &str) -> String {
    format!("{:x}", Sha256::digest(url.as_bytes()))
}

fn valid_checksum(checksum: &str) -> bool {
    checksum
        .strip_prefix("sha256-")
        .is_some_and(|hex| hex.len() == 64 && hex.bytes().all(|ch| ch.is_ascii_hexdigit()))
}

fn verify(bytes: &[u8], expected: &str) -> Result<(), RegistryError> {
    if checksum(bytes) != expected {
        return Err(RegistryError::IntegrityMismatch);
    }
    Ok(())
}

fn read_artifact(path: &Path) -> Result<Option<Vec<u8>>, RegistryError> {
    let mut options = std::fs::OpenOptions::new();
    options.read(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        options.custom_flags(libc::O_NONBLOCK | libc::O_NOFOLLOW);
    }
    let file = match options.open(path) {
        Ok(file) => file,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(error) => return Err(error.into()),
    };
    let metadata = file.metadata()?;
    if !metadata.is_file() {
        return Err(RegistryError::ArtifactNotRegularFile);
    }
    let capacity = usize::try_from(metadata.len().min(MAX_ARTIFACT_SIZE as u64)).unwrap_or(0);
    let mut bytes = Vec::with_capacity(capacity);
    file.take(MAX_ARTIFACT_SIZE as u64 + 1)
        .read_to_end(&mut bytes)?;
    if bytes.len() > MAX_ARTIFACT_SIZE {
        return Err(RegistryError::CachedArtifactTooLarge);
    }
    Ok(Some(bytes))
}

fn read_lock(
    path: &Path,
    authority: Option<&crate::authority::SourceAuthority>,
) -> Result<Option<Lockfile>, RegistryError> {
    read_lock_bytes(path, authority)?
        .map(|bytes| parse_lock(path, &bytes))
        .transpose()
}

fn read_lock_bytes(
    path: &Path,
    authority: Option<&crate::authority::SourceAuthority>,
) -> Result<Option<Vec<u8>>, RegistryError> {
    let Some(authority) = authority.filter(|authority| authority.is_read()) else {
        return read_artifact(path);
    };
    let target = crate::caps::PathTarget::for_entry(path)?;
    authority.check_local(&target)?;
    let bytes = read_lock_nofollow(&target)?;
    authority.check_local(&target)?;
    Ok(bytes)
}

fn read_lock_nofollow(target: &crate::caps::PathTarget) -> Result<Option<Vec<u8>>, RegistryError> {
    let (directory, name) = crate::caps::fs::parent(target.as_path())?;
    let mut options = cap_std::fs::OpenOptions::new();
    options.read(true).follow(cap_fs_ext::FollowSymlinks::No);
    #[cfg(unix)]
    {
        use cap_fs_ext::OpenOptionsExt;
        options.custom_flags(libc::O_NONBLOCK);
    }
    let file = match directory.open_with(Path::new(&name), &options) {
        Ok(file) => file,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(error) => return Err(error.into()),
    };
    let metadata = file.metadata()?;
    if !metadata.is_file() {
        return Err(RegistryError::LockfileNotRegularFile);
    }
    let capacity = usize::try_from(metadata.len().min(MAX_ARTIFACT_SIZE as u64)).unwrap_or(0);
    let mut bytes = Vec::with_capacity(capacity);
    file.take(MAX_ARTIFACT_SIZE as u64 + 1)
        .read_to_end(&mut bytes)?;
    if bytes.len() > MAX_ARTIFACT_SIZE {
        return Err(RegistryError::LockfileTooLarge);
    }
    Ok(Some(bytes))
}

fn parse_lock(path: &Path, bytes: &[u8]) -> Result<Lockfile, RegistryError> {
    let name = path.to_string_lossy();
    let source = std::str::from_utf8(bytes)
        .map_err(|_| RegistryError::LockfileNotUtf8 { name: name.to_string() })?;
    let value = serde_fjson::parse(source).map_err(|error| RegistryError::LockfileParse {
        detail: error.render(&name, source),
    })?;
    serde_json::from_value(value).map_err(|source| RegistryError::LockfileJson {
        name: name.to_string(),
        source: Arc::new(source),
    })
}

fn validate_registry(registry: &Url) -> Result<(), RegistryError> {
    if !registry.as_str().ends_with('/')
        || registry.query().is_some()
        || registry.fragment().is_some()
        || !registry.username().is_empty()
        || registry.password().is_some()
    {
        return Err(RegistryError::InvalidRegistryBaseUrl);
    }
    let loopback = matches!(
        registry.host_str(),
        Some("localhost" | "127.0.0.1" | "[::1]")
    );
    if registry.scheme() != "https" && !(registry.scheme() == "http" && loopback) {
        return Err(RegistryError::RegistryRequiresHttps);
    }
    Ok(())
}

fn validate_lock(
    lock: &Lockfile,
    jsr_registry: &Url,
    npm_registry: &Url,
) -> Result<(), RegistryError> {
    if lock.version != 1 && lock.version != 2 {
        return Err(RegistryError::UnsupportedLockfileVersion { version: lock.version });
    }
    if lock.jsr_registry != jsr_registry.as_str() {
        return Err(RegistryError::LockfileJsrRegistryMismatch);
    }
    if lock.version == 1 {
        if lock.npm_registry.is_some() || !lock.npm.is_empty() {
            return Err(RegistryError::LockfileV1NpmPins);
        }
    } else if let Some(npm_registry_name) = &lock.npm_registry {
        if npm_registry_name != npm_registry.as_str() {
            return Err(RegistryError::LockfileNpmRegistryMismatch);
        }
    } else if !lock.npm.is_empty() {
        return Err(RegistryError::LockfileNpmPackagesWithoutRegistry);
    }
    for (name, entry) in &lock.remote {
        http_url(name)?;
        validate_remote_entry(entry)?;
        if name.starts_with(jsr_registry.as_str()) {
            return Err(RegistryError::RegistryModuleRemoteLockEntry { name: name.to_owned() });
        }
    }
    npm::validate_lock_npm(&lock.npm, npm_registry)?;
    Ok(())
}

pub fn atomic_write(path: &Path, bytes: &[u8]) -> Result<(), RegistryError> {
    stage(path, bytes)?
        .persist(path)
        .map_err(|detail| RegistryError::Persist { detail: detail.to_string() })?;
    Ok(())
}

/// Atomically rewrite a file that other processes may update concurrently: `edit` derives the
/// new contents (or `None` to leave it alone) from what `read` returns, and is run again
/// whenever the file has changed by the time the rewrite is ready. A writer that lands between
/// that final check and the rename can still be overwritten.
pub fn update_file<T: PartialEq, E>(
    path: &Path,
    read: impl Fn() -> Result<T, E>,
    mut edit: impl FnMut(&T) -> Result<Option<Vec<u8>>, E>,
) -> Result<(), E>
where
    E: From<RegistryError>,
{
    for _ in 0..16 {
        let current = read()?;
        let Some(bytes) = edit(&current)? else {
            return Ok(());
        };
        let staged = stage(path, &bytes)?;
        if read()? == current {
            staged
                .persist(path)
                .map_err(|error| RegistryError::Persist { detail: error.to_string() })?;
            return Ok(());
        }
    }
    Err(RegistryError::FileKeptChanging(path.display().to_string()).into())
}

fn stage(path: &Path, bytes: &[u8]) -> Result<tempfile::NamedTempFile, RegistryError> {
    let parent = path.parent().ok_or(RegistryError::OutputNeedsParent)?;
    let mut builder = tempfile::Builder::new();
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        builder.permissions(std::fs::Permissions::from_mode(0o644));
    }
    let mut file = builder.tempfile_in(parent)?;
    match std::fs::symlink_metadata(path) {
        Ok(metadata) => {
            if !metadata.is_file() || metadata.is_symlink() {
                return Err(RegistryError::OutputNotRegularFile);
            }
            file.as_file().set_permissions(metadata.permissions())?;
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
        Err(error) => return Err(error.into()),
    }
    file.write_all(bytes)?;
    file.as_file().sync_all()?;
    Ok(file)
}

#[cfg(test)]
mod tests {
    use super::{MAX_ARTIFACT_SIZE, RegistryError, read_artifact, read_lock_nofollow, update_file};

    #[test]
    fn bounded_readers_preserve_bytes_through_the_size_limit() {
        let directory = tempfile::tempdir().unwrap();
        for size in [0, 1, MAX_ARTIFACT_SIZE - 1, MAX_ARTIFACT_SIZE] {
            let bytes: Vec<u8> = (0..size).map(|index| (index % 251) as u8).collect();
            let path = directory.path().join("artifact");
            std::fs::write(&path, &bytes).unwrap();
            assert_eq!(read_artifact(&path).unwrap(), Some(bytes.clone()));

            let target = crate::caps::PathTarget::for_entry(&path).unwrap();
            assert_eq!(read_lock_nofollow(&target).unwrap(), Some(bytes));
        }
    }

    #[test]
    fn bounded_readers_reject_oversized_sparse_files() {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("oversized");
        let file = std::fs::File::create(&path).unwrap();
        file.set_len(MAX_ARTIFACT_SIZE as u64 + 1).unwrap();
        assert!(matches!(
            read_artifact(&path),
            Err(RegistryError::CachedArtifactTooLarge)
        ));

        let target = crate::caps::PathTarget::for_entry(&path).unwrap();
        assert!(matches!(
            read_lock_nofollow(&target),
            Err(RegistryError::LockfileTooLarge)
        ));
    }

    #[test]
    fn bounded_readers_reject_nonregular_files() {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("directory");
        std::fs::create_dir(&path).unwrap();
        // Windows can reject directories at open, before the regular-file check.
        let error = read_artifact(&path).unwrap_err();
        assert!(
            matches!(error, RegistryError::ArtifactNotRegularFile)
                || cfg!(windows)
                    && matches!(&error, RegistryError::Io(error)
                        if error.kind() == std::io::ErrorKind::PermissionDenied),
            "{error}"
        );

        let target = crate::caps::PathTarget::for_entry(&path).unwrap();
        let error = read_lock_nofollow(&target).unwrap_err();
        assert!(
            matches!(error, RegistryError::LockfileNotRegularFile)
                || cfg!(windows)
                    && matches!(&error, RegistryError::Io(error)
                        if error.kind() == std::io::ErrorKind::PermissionDenied),
            "{error}"
        );
    }

    #[test]
    fn updates_are_recomputed_when_another_writer_lands_first() {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("entries");
        let mut raced = false;
        update_file(
            &path,
            || read_artifact(&path),
            |current| {
                if !raced {
                    raced = true;
                    std::fs::write(&path, "theirs\n").unwrap();
                }
                let mut bytes = current.clone().unwrap_or_default();
                bytes.extend_from_slice(b"ours\n");
                Ok(Some(bytes))
            },
        )
        .unwrap();
        assert_eq!(std::fs::read(&path).unwrap(), b"theirs\nours\n");
    }
}