//! An OCI registry of the images repositories publish as releases
//! (`knotless release publish --oci`, docs/design/packages.md):
//!
//! ```text
//! docker pull knotless.dev/<owner>/<repo>/<package>:<version>
//! docker pull knotless.dev/<owner>/<repo>:latest   # the package named as the repository
//! docker login knotless.dev -u <handle>            # an app password
//! docker push knotless.dev/<owner>/<repo>/<package>:<version>
//! ```
//!
//! A tag is a version, or `latest`, the highest without a pre-release
//! part; its manifest is the one its release's `index.json` names. Every
//! blob is a file of a release, named for its digest, read from the
//! owner's PDS once, checked against its record, and kept.
//!
//! A push is the owner's alone, with a token `/v2/token` gives for their
//! app password, which logs in to their PDS and is kept nowhere. Each blob
//! pushed is checked against its digest and uploaded to their PDS as it
//! lands; a manifest pushed by a tag, a version, makes the release of it
//! and every blob it reaches, with an `index.json` naming it. A version is
//! never pushed twice, and `latest` is never pushed: it is computed.

// A handler's error is the answer it sends, made once per request, so a
// large one costs nothing to move.
#![allow(clippy::result_large_err)]

use crate::releases::{Records, base_url, kept_dir};
use axum::Router;
use axum::body::Body;
use axum::extract::{Path as Params, Query, State};
use axum::http::{HeaderMap, HeaderName, StatusCode, header};
use axum::response::{IntoResponse, Response};
use axum::routing::get;
use futures::StreamExt;
use knotless::env::Ctx;
use knotless::oci;
use knotless::push;
use knotless::records::{RELEASE_MIME, ReleaseFile, ReleasePart, ReleaseRecord};
use knotless::release::{self, Made, Target};
use knotless::remote::Remote;
use knotless::url::RepoUrl;
use knotless::xrpc::Auth;
use serde_json::{Value, json};
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::io::AsyncWriteExt;

/// The most packages whose releases are kept listed at once, and the most
/// blobs pushed and not yet released: past either, all are forgotten.
const MOST_KEPT: usize = 1024;

/// How long a token for pushing lasts.
const TOKEN_FOR: Duration = Duration::from_secs(3600);

/// The largest manifest taken.
const MOST_MANIFEST: usize = 4 << 20;

const API_VERSION: HeaderName = HeaderName::from_static("docker-distribution-api-version");
const DIGEST: HeaderName = HeaderName::from_static("docker-content-digest");
const UPLOAD: HeaderName = HeaderName::from_static("docker-upload-uuid");

/// An image's name: an owner, a repository of theirs, and a package.
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
struct Name {
    owner: String,
    repo: String,
    package: String,
}

impl std::fmt::Display for Name {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}/{}/{}", self.owner, self.repo, self.package)
    }
}

/// What a path under `/v2/` asks of an image.
#[derive(Debug, PartialEq)]
enum Ask {
    /// A manifest, by tag or digest.
    Manifest(String),
    /// A blob, by digest.
    Blob(String),
    /// A blob's upload: a new one, or one under way by its id.
    Upload(Option<String>),
    /// The tags.
    Tags,
}

/// Someone pushing: their account, and their session's authority there.
struct Pusher {
    did: String,
    auth: Auth,
}

struct Upload {
    did: String,
    path: PathBuf,
}

struct Registry {
    ctx: Ctx,
    public_url: Option<String>,
    /// Each package's releases, by name.
    records: Mutex<HashMap<Name, Arc<Records>>>,
    /// Who each token for pushing is, and when it was given.
    tokens: Mutex<HashMap<String, (Arc<Pusher>, Instant)>>,
    /// The uploads under way, by id.
    uploads: Mutex<HashMap<String, Upload>>,
    /// The blobs pushed and uploaded, not yet in a release.
    staged: Mutex<Staged>,
}

/// The most blobs one account has staged at once: past it, its oldest is
/// forgotten, so one account's pushes never cost another's.
const STAGED_EACH: usize = 256;

/// The blobs pushed and uploaded, not yet in a release, by account and
/// digest, each with when it was staged, counting up. Past [`STAGED_EACH`]
/// for an account, or [`MOST_KEPT`] in all, the oldest is forgotten: a push
/// still under way stages its blobs last, so it keeps them.
#[derive(Default)]
struct Staged {
    files: HashMap<(String, String), (u64, ReleaseFile)>,
    next: u64,
}

impl Staged {
    fn get(&self, did: &str, digest: &str) -> Option<&ReleaseFile> {
        let key = (did.to_string(), digest.to_string());
        self.files.get(&key).map(|(_, file)| file)
    }

    fn insert(&mut self, did: &str, digest: String, file: ReleaseFile) {
        let own = self.files.keys().filter(|(d, _)| d == did).count();
        if own >= STAGED_EACH {
            self.forget_oldest(Some(did));
        }
        if self.files.len() >= MOST_KEPT {
            self.forget_oldest(None);
        }
        self.next += 1;
        self.files
            .insert((did.to_string(), digest), (self.next, file));
    }

    /// Forgets the oldest staged, of `did` if given.
    fn forget_oldest(&mut self, did: Option<&str>) {
        let oldest = self
            .files
            .iter()
            .filter(|((d, _), _)| did.is_none_or(|did| d == did))
            .min_by_key(|(_, (at, _))| *at)
            .map(|(key, _)| key.clone());
        if let Some(key) = oldest {
            self.files.remove(&key);
        }
    }
}

/// The registry's routes, reading as `ctx`, its tokens given out at
/// `public_url` (or else the host it is asked for).
pub fn routes(ctx: Ctx, public_url: Option<String>) -> Router {
    let registry = Arc::new(Registry {
        ctx,
        public_url,
        records: Mutex::default(),
        tokens: Mutex::default(),
        uploads: Mutex::default(),
        staged: Mutex::default(),
    });
    Router::new()
        .route("/v2/", get(base))
        .route("/v2/token", get(token))
        .route(
            "/v2/{*rest}",
            get(asked).post(posted).patch(patched).put(put),
        )
        .with_state(registry)
}

/// An error as the distribution spec has them.
fn error(status: StatusCode, code: &str, message: String) -> Response {
    let body = json!({"errors": [{"code": code, "message": message}]});
    let json = [(header::CONTENT_TYPE, "application/json")];
    (
        status,
        [(API_VERSION, "registry/2.0")],
        json,
        body.to_string(),
    )
        .into_response()
}

fn unknown(code: &str, message: String) -> Response {
    error(StatusCode::NOT_FOUND, code, message)
}

fn invalid(code: &str, message: String) -> Response {
    error(StatusCode::BAD_REQUEST, code, message)
}

fn failed(e: anyhow::Error) -> Response {
    error(StatusCode::BAD_GATEWAY, "UNKNOWN", format!("{e:#}"))
}

/// What a path under `/v2/` asks for, of which image.
fn parse(rest: &str) -> Option<(Name, Ask)> {
    let (name, ask) = if let Some(name) = rest.strip_suffix("/tags/list") {
        (name, Ask::Tags)
    } else if let Some((name, id)) = rest.split_once("/blobs/uploads") {
        let id = id.trim_start_matches('/');
        (
            name,
            Ask::Upload(Some(id.to_string()).filter(|i| !i.is_empty())),
        )
    } else if let Some((name, reference)) = rest.rsplit_once("/manifests/") {
        (name, Ask::Manifest(reference.into()))
    } else {
        let (name, digest) = rest.rsplit_once("/blobs/")?;
        (name, Ask::Blob(digest.into()))
    };
    let parts: Vec<&str> = name.split('/').collect();
    let (owner, repo, package) = match parts[..] {
        [owner, repo] => (owner, repo, repo),
        [owner, repo, package] => (owner, repo, package),
        _ => return None,
    };
    let name = Name {
        owner: owner.into(),
        repo: repo.into(),
        package: package.into(),
    };
    Some((name, ask))
}

/// A fresh value no one can guess, as hex.
fn random() -> String {
    use rand_core::{OsRng, RngCore};
    let mut bytes = [0u8; 16];
    OsRng.fill_bytes(&mut bytes);
    bytes.iter().map(|b| format!("{b:02x}")).collect()
}

/// What a handler answers, either way.
type Answer = Result<Response, Response>;

/// Empties `map` once it holds [`MOST_KEPT`].
fn bounded<K, V>(map: &mut HashMap<K, V>) {
    if map.len() >= MOST_KEPT {
        map.clear();
    }
}

impl Registry {
    fn url(name: &Name) -> Option<RepoUrl> {
        RepoUrl::parse(&format!("knotless://{}/{}", name.owner, name.repo)).ok()
    }

    fn records(&self, name: &Name) -> Option<Arc<Records>> {
        let url = Registry::url(name)?;
        Some(self.records_at(name, url))
    }

    /// `name`'s releases, kept, of the repository at `url`.
    fn records_at(&self, name: &Name, url: RepoUrl) -> Arc<Records> {
        let mut kept = self.records.lock().unwrap();
        bounded(&mut kept);
        let records = kept
            .entry(name.clone())
            .or_insert_with(|| Arc::new(Records::new(self.ctx.clone(), url, name.package.clone())));
        records.clone()
    }

    /// Asks for a token, for `scope` if given.
    fn challenge(&self, headers: &HeaderMap, scope: Option<String>) -> Response {
        let realm = format!("{}/v2/token", base_url(self.public_url.as_deref(), headers));
        let mut asked = format!("Bearer realm=\"{realm}\",service=\"knotless\"");
        if let Some(scope) = scope {
            asked.push_str(&format!(",scope=\"{scope}\""));
        }
        let says = "log in with an app password: docker login <host> -u <handle>";
        let mut res = error(StatusCode::UNAUTHORIZED, "UNAUTHORIZED", says.into());
        let value = asked.parse().expect("a header's value");
        res.headers_mut().insert(header::WWW_AUTHENTICATE, value);
        res
    }

    /// Who `headers`' token is, if it is one for pushing still.
    fn bearer(&self, headers: &HeaderMap) -> Option<Arc<Pusher>> {
        let said = headers.get(header::AUTHORIZATION)?.to_str().ok()?;
        let token = said.strip_prefix("Bearer ")?;
        let mut tokens = self.tokens.lock().unwrap();
        tokens.retain(|_, (_, at)| at.elapsed() < TOKEN_FOR);
        tokens.get(token).map(|(pusher, _)| pusher.clone())
    }

    /// The owner of `name`, pushing with `headers`' token, and the
    /// repository: else what to answer instead.
    async fn pusher(
        &self,
        headers: &HeaderMap,
        name: &Name,
    ) -> Result<(Arc<Pusher>, RepoUrl), Response> {
        let scope = format!("repository:{name}:push,pull");
        let Some(pusher) = self.bearer(headers) else {
            return Err(self.challenge(headers, Some(scope)));
        };
        let Some(url) = Registry::url(name) else {
            let says = format!("{}/{}: no such repository", name.owner, name.repo);
            return Err(unknown("NAME_INVALID", says));
        };
        let owner = knotless::identity::resolve(&self.ctx, &name.owner).await;
        let owner = owner.map_err(|e| unknown("NAME_UNKNOWN", format!("{e:#}")))?;
        if owner.did != pusher.did {
            let only = format!("only {} pushes to {name}", name.owner);
            return Err(error(StatusCode::FORBIDDEN, "DENIED", only));
        }
        Ok((pusher, url))
    }

    /// The blob of `digest` that `did` pushed and no release holds yet.
    fn staged(&self, did: &str, digest: &str) -> Option<ReleaseFile> {
        let staged = self.staged.lock().unwrap();
        staged.get(did, digest).cloned()
    }

    /// Uploads the file at `path`, of `size` bytes and SHA-256 `sha`, to
    /// `pusher`'s PDS, keeps it among the release files, and stages it.
    async fn stage(
        &self,
        pusher: &Pusher,
        url: &RepoUrl,
        path: &PathBuf,
        size: u64,
        sha: &str,
    ) -> anyhow::Result<()> {
        let remote = Remote::open(&self.ctx, url).await?;
        let auth = &pusher.auth;
        let blobs = push::upload_file(&self.ctx, &remote, auth, path, size, RELEASE_MIME).await?;
        let kept = kept_dir(&self.ctx);
        tokio::fs::create_dir_all(&kept).await?;
        tokio::fs::rename(path, kept.join(sha)).await?;
        let file = ReleaseFile {
            name: format!("sha256-{sha}"),
            size,
            sha256: sha.to_string(),
            parts: blobs.into_iter().map(|blob| ReleasePart { blob }).collect(),
        };
        let mut staged = self.staged.lock().unwrap();
        staged.insert(&pusher.did, format!("sha256:{sha}"), file);
        Ok(())
    }

    /// A new file for an upload's bytes, until it is done.
    async fn uploading(&self) -> anyhow::Result<PathBuf> {
        let dir = self.ctx.env.cache_dir.join("serve").join("uploads");
        tokio::fs::create_dir_all(&dir).await?;
        let path = dir.join(random());
        tokio::fs::write(&path, b"").await?;
        Ok(path)
    }
}

/// Says this is a registry, which clients ask first: with a token, any
/// token, as a pull's is anyone's; without one, asking for one.
async fn base(State(registry): State<Arc<Registry>>, headers: HeaderMap) -> Response {
    let said = headers
        .get(header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok());
    if !said.is_some_and(|s| s.starts_with("Bearer ")) {
        return registry.challenge(&headers, None);
    }
    let json = [(header::CONTENT_TYPE, "application/json")];
    ([(API_VERSION, "registry/2.0")], json, "{}").into_response()
}

/// `value` as a JSON answer.
fn json_answer(value: Value) -> Response {
    (
        [(header::CONTENT_TYPE, "application/json")],
        value.to_string(),
    )
        .into_response()
}

/// A token: anyone's, for pulling, or for an owner's app password, given
/// as Basic, one to push with.
async fn token(State(registry): State<Arc<Registry>>, headers: HeaderMap) -> Answer {
    use base64::Engine;
    let basic = headers
        .get(header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.strip_prefix("Basic "))
        .and_then(|b| base64::engine::general_purpose::STANDARD.decode(b).ok())
        .and_then(|b| String::from_utf8(b).ok());
    let Some(basic) = basic else {
        return Ok(json_answer(
            json!({"token": "anonymous", "access_token": "anonymous"}),
        ));
    };
    let (user, password) = basic.split_once(':').unwrap_or((&basic, ""));
    let session = knotless::session::create(&registry.ctx, user, password).await;
    let refused = |e| {
        let says = format!("{user} could not log in: {e:#}");
        error(StatusCode::UNAUTHORIZED, "UNAUTHORIZED", says)
    };
    let session = session.map_err(refused)?;
    let token = random();
    let pusher = Arc::new(Pusher {
        did: session.did.clone(),
        auth: session.auth(),
    });
    let mut tokens = registry.tokens.lock().unwrap();
    tokens.insert(token.clone(), (pusher, Instant::now()));
    let lasts = TOKEN_FOR.as_secs();
    Ok(json_answer(
        json!({"token": token, "access_token": token, "expires_in": lasts}),
    ))
}

/// The file of `digest` among `releases`.
fn by_digest<'a>(releases: &'a [ReleaseRecord], digest: &str) -> Option<&'a ReleaseFile> {
    let name = oci::file_name(digest)?;
    releases
        .iter()
        .flat_map(|r| &r.files)
        .find(|f| f.name == name)
}

/// What a blob, or a manifest of `media_type`, is handed out with.
fn headers(digest: &str, media_type: &str) -> [(HeaderName, String); 3] {
    [
        (header::CONTENT_TYPE, media_type.to_string()),
        (DIGEST, digest.to_string()),
        (API_VERSION, "registry/2.0".to_string()),
    ]
}

/// `name`'s releases, and where to read their files: else what to answer.
async fn opened(
    registry: &Registry,
    name: &Name,
) -> Result<(Arc<Records>, Arc<Vec<ReleaseRecord>>), Response> {
    let Some(records) = registry.records(name) else {
        let invalid = format!("{}/{}: no such repository", name.owner, name.repo);
        return Err(unknown("NAME_INVALID", invalid));
    };
    let listed = records.list().await;
    let releases = listed.map_err(|e| unknown("NAME_UNKNOWN", format!("{e:#}")))?;
    Ok((records, releases))
}

/// That nothing at `rest` is done by `method`.
fn unsupported(rest: &str, method: &str) -> Response {
    let says = format!("{rest}: nothing is {method} here");
    error(StatusCode::METHOD_NOT_ALLOWED, "UNSUPPORTED", says)
}

/// `name` and what is asked of it, from a path under `/v2/`, else that
/// nothing is done there by `method`.
fn asking(rest: &str, method: &str) -> Result<(Name, Ask), Response> {
    parse(rest).ok_or_else(|| unsupported(rest, method))
}

async fn asked(State(registry): State<Arc<Registry>>, Params(rest): Params<String>) -> Answer {
    let Some((name, ask)) = parse(&rest) else {
        return Err(unknown(
            "NAME_UNKNOWN",
            format!("{rest}: not <owner>/<repo>[/<package>]"),
        ));
    };
    let (records, releases) = opened(&registry, &name).await?;
    match ask {
        Ask::Tags => {
            let images: Vec<ReleaseRecord> = releases
                .iter()
                .filter(|r| r.files.iter().any(|f| f.name == oci::INDEX))
                .cloned()
                .collect();
            let mut tags: Vec<&str> = images.iter().map(|r| r.version.as_str()).collect();
            if release::latest(&images, &name.package).is_some() {
                tags.push("latest");
            }
            Ok(json_answer(json!({"name": name.to_string(), "tags": tags})))
        }
        Ask::Blob(digest) => {
            let file = match by_digest(&releases, &digest) {
                Some(file) => Some(file.clone()),
                None => {
                    let owner = knotless::identity::resolve(&registry.ctx, &name.owner).await;
                    owner.ok().and_then(|o| registry.staged(&o.did, &digest))
                }
            };
            let Some(file) = file else {
                return Err(unknown("BLOB_UNKNOWN", format!("no blob {digest}")));
            };
            let octets = headers(&digest, "application/octet-stream");
            records.send(&file, octets).await.map_err(failed)
        }
        Ask::Manifest(reference) => manifest(&records, &releases, &name, &reference).await,
        Ask::Upload(_) => Err(unknown(
            "BLOB_UPLOAD_UNKNOWN",
            format!("{rest}: not read here"),
        )),
    }
}

/// A manifest by its digest, or by a tag, the one its release names.
async fn manifest(
    records: &Records,
    releases: &[ReleaseRecord],
    name: &Name,
    reference: &str,
) -> Answer {
    let digest = match reference.starts_with("sha256:") {
        true => reference.to_string(),
        false => {
            let tagged = match reference {
                "latest" => release::latest(releases, &name.package),
                version => releases.iter().find(|r| r.version == version),
            };
            let index = tagged.and_then(|r| r.files.iter().find(|f| f.name == oci::INDEX));
            let Some(index) = index else {
                return Err(unknown("MANIFEST_UNKNOWN", format!("no image {reference}")));
            };
            let read = records.read(index).await.and_then(|i| oci::manifest(&i));
            read.map_err(failed)?.digest
        }
    };
    let Some(file) = by_digest(releases, &digest) else {
        return Err(unknown("MANIFEST_UNKNOWN", format!("no manifest {digest}")));
    };
    let bytes = records.read(file).await.map_err(failed)?;
    let media_type = oci::media_type(&bytes);
    Ok((headers(&digest, &media_type), bytes).into_response())
}

/// Appends `body` to the file at `path`, and says its size now.
async fn append(path: &PathBuf, body: Body) -> anyhow::Result<u64> {
    let mut options = tokio::fs::OpenOptions::new();
    let mut file = options.append(true).open(path).await?;
    let mut chunks = body.into_data_stream();
    while let Some(chunk) = chunks.next().await {
        file.write_all(&chunk?).await?;
    }
    file.flush().await?;
    Ok(file.metadata().await?.len())
}

/// Where an upload under way goes on, as it is answered.
fn going(name: &Name, id: &str, size: u64) -> Response {
    let at = format!("/v2/{name}/blobs/uploads/{id}");
    let range = format!("0-{}", size.saturating_sub(1));
    let headers = [
        (header::LOCATION, at),
        (header::RANGE, range),
        (UPLOAD, id.to_string()),
        (API_VERSION, "registry/2.0".to_string()),
    ];
    (StatusCode::ACCEPTED, headers).into_response()
}

/// Where a blob or manifest pushed is now, as it is answered.
fn made(name: &Name, kind: &str, digest: &str) -> Response {
    let headers = [
        (header::LOCATION, format!("/v2/{name}/{kind}/{digest}")),
        (DIGEST, digest.to_string()),
        (API_VERSION, "registry/2.0".to_string()),
    ];
    (StatusCode::CREATED, headers).into_response()
}

/// Begins a blob's upload, and with `?digest=` ends it with the body.
async fn posted(
    State(registry): State<Arc<Registry>>,
    Params(rest): Params<String>,
    Query(query): Query<HashMap<String, String>>,
    headers: HeaderMap,
    body: Body,
) -> Answer {
    let (name, ask) = asking(&rest, "posted")?;
    if ask != Ask::Upload(None) {
        return Err(unsupported(&rest, "posted"));
    }
    let (pusher, url) = registry.pusher(&headers, &name).await?;
    let path = registry.uploading().await.map_err(failed)?;
    let id = path
        .file_name()
        .expect("a name")
        .to_string_lossy()
        .into_owned();
    let upload = Upload {
        did: pusher.did.clone(),
        path,
    };
    registry.uploads.lock().unwrap().insert(id.clone(), upload);
    match query.get("digest") {
        Some(digest) => finish(&registry, &pusher, &url, &name, &id, digest, body).await,
        None => Ok(going(&name, &id, 0)),
    }
}

/// Goes on with an upload, with the body as its next part.
async fn patched(
    State(registry): State<Arc<Registry>>,
    Params(rest): Params<String>,
    headers: HeaderMap,
    body: Body,
) -> Answer {
    let (name, ask) = asking(&rest, "patched")?;
    let Ask::Upload(Some(id)) = ask else {
        return Err(unsupported(&rest, "patched"));
    };
    let (pusher, _) = registry.pusher(&headers, &name).await?;
    let path = upload_of(&registry, &pusher, &id)?;
    let size = append(&path, body).await.map_err(failed)?;
    Ok(going(&name, &id, size))
}

/// Where `pusher`'s upload `id` is, if it is theirs.
fn upload_of(registry: &Registry, pusher: &Pusher, id: &str) -> Result<PathBuf, Response> {
    let uploads = registry.uploads.lock().unwrap();
    match uploads.get(id).filter(|u| u.did == pusher.did) {
        Some(upload) => Ok(upload.path.clone()),
        None => Err(unknown("BLOB_UPLOAD_UNKNOWN", format!("no upload {id}"))),
    }
}

/// Ends an upload, with the body as its last part, or pushes a manifest.
async fn put(
    State(registry): State<Arc<Registry>>,
    Params(rest): Params<String>,
    Query(query): Query<HashMap<String, String>>,
    headers: HeaderMap,
    body: Body,
) -> Answer {
    let (name, ask) = asking(&rest, "put")?;
    let (reference, id) = match ask {
        Ask::Manifest(reference) => (Some(reference), None),
        Ask::Upload(Some(id)) => (None, Some(id)),
        _ => return Err(unsupported(&rest, "put")),
    };
    let (pusher, url) = registry.pusher(&headers, &name).await?;
    if let Some(reference) = reference {
        let media_type = headers
            .get(header::CONTENT_TYPE)
            .and_then(|v| v.to_str().ok());
        let media_type = media_type.map(String::from);
        let pushing = Pushing {
            registry: &registry,
            pusher: &pusher,
            url: &url,
            name: &name,
        };
        return pushing.manifest(&reference, media_type, body).await;
    }
    let id = id.unwrap_or_default();
    let Some(digest) = query.get("digest") else {
        return Err(invalid(
            "DIGEST_INVALID",
            "an upload ends with ?digest=".into(),
        ));
    };
    finish(&registry, &pusher, &url, &name, &id, digest, body).await
}

/// Ends upload `id` with `body`: checked against `digest`, uploaded and
/// staged.
async fn finish(
    registry: &Registry,
    pusher: &Pusher,
    url: &RepoUrl,
    name: &Name,
    id: &str,
    digest: &str,
    body: Body,
) -> Answer {
    let path = upload_of(registry, pusher, id)?;
    registry.uploads.lock().unwrap().remove(id);
    let appended = append(&path, body).await;
    let measured = appended.and_then(|_| release::sha256_file(&path));
    let (size, sha) = measured.map_err(failed)?;
    if format!("sha256:{sha}") != digest {
        let _ = tokio::fs::remove_file(&path).await;
        let says = format!("what was uploaded is sha256:{sha}, not {digest}");
        return Err(invalid("DIGEST_INVALID", says));
    }
    let staged = registry.stage(pusher, url, &path, size, &sha).await;
    staged.map_err(failed)?;
    Ok(made(name, "blobs", digest))
}

/// The digests `manifest` points at: its blobs, and the manifests an index
/// names, apart.
fn pointed(manifest: &Value) -> (Vec<String>, Vec<String>) {
    let digests = |v: &Value| -> Vec<String> {
        let all = v.as_array().into_iter().flatten();
        all.filter_map(|d| d["digest"].as_str().map(String::from))
            .collect()
    };
    let mut blobs = digests(&manifest["layers"]);
    blobs.extend(digests(&manifest["blobs"]));
    blobs.extend(manifest["config"]["digest"].as_str().map(String::from));
    (blobs, digests(&manifest["manifests"]))
}

/// `file`, or else that `digest` was never pushed.
fn pushed_already(file: Option<ReleaseFile>, digest: &str) -> Result<ReleaseFile, Response> {
    let says = format!("{digest} was not pushed");
    file.ok_or_else(|| invalid("MANIFEST_BLOB_UNKNOWN", says))
}

/// Adds `file` to `files` unless one of its name is there.
fn add(files: &mut Vec<ReleaseFile>, file: ReleaseFile) {
    if !files.iter().any(|f| f.name == file.name) {
        files.push(file);
    }
}

/// A manifest being pushed, by the owner of a repository.
struct Pushing<'a> {
    registry: &'a Registry,
    pusher: &'a Pusher,
    url: &'a RepoUrl,
    name: &'a Name,
}

impl Pushing<'_> {
    /// The blob of `digest`, pushed or released.
    fn find(&self, releases: &[ReleaseRecord], digest: &str) -> Result<ReleaseFile, Response> {
        let staged = self.registry.staged(&self.pusher.did, digest);
        pushed_already(
            staged.or_else(|| by_digest(releases, digest).cloned()),
            digest,
        )
    }

    /// Pushes a manifest: by digest, staged; by tag, a version, released,
    /// with every blob it reaches.
    async fn manifest(&self, reference: &str, media_type: Option<String>, body: Body) -> Answer {
        let (registry, name) = (self.registry, self.name);
        let says = format!("a manifest is at most {MOST_MANIFEST} bytes, of JSON");
        let read = axum::body::to_bytes(body, MOST_MANIFEST).await.ok();
        let value = read
            .as_ref()
            .and_then(|b| serde_json::from_slice::<Value>(b).ok());
        let (Some(bytes), Some(value)) = (read, value) else {
            return Err(invalid("MANIFEST_INVALID", says));
        };
        let sha = release::sha256(&bytes);
        let digest = format!("sha256:{sha}");
        let tag = !reference.starts_with("sha256:");
        if !tag && reference != digest {
            let says = format!("the manifest is {digest}, not {reference}");
            return Err(invalid("DIGEST_INVALID", says));
        }
        if tag && release::version(reference).is_err() {
            let says =
                format!("{reference}: a tag is a version, such as 1.0.0; latest is the highest");
            return Err(invalid("TAG_INVALID", says));
        }
        let path = registry.uploading().await.map_err(failed)?;
        let size = bytes.len() as u64;
        let staged = async {
            tokio::fs::write(&path, &bytes).await?;
            registry
                .stage(self.pusher, self.url, &path, size, &sha)
                .await
        };
        staged.await.map_err(failed)?;
        if !tag {
            return Ok(made(name, "manifests", &digest));
        }
        let (records, releases) = opened(registry, name).await?;
        // The manifest, and what it reaches: an index's manifests, and theirs.
        let mut files: Vec<ReleaseFile> = Vec::new();
        let mut manifests = vec![(digest.clone(), value)];
        while let Some((at, manifest)) = manifests.pop() {
            add(&mut files, self.find(&releases, &at)?);
            let (blobs, children) = pointed(&manifest);
            for blob in blobs {
                add(&mut files, self.find(&releases, &blob)?);
            }
            for child in children {
                let file = self.find(&releases, &child)?;
                let read = records.read(&file).await.ok();
                let value = read.and_then(|b| serde_json::from_slice(&b).ok());
                manifests.push((child, value.unwrap_or(Value::Null)));
            }
        }
        // As the client said, else as the manifest says.
        let said = oci::media_type(&bytes);
        let media_type = media_type.unwrap_or(said);
        let index = json!({
            "schemaVersion": 2,
            "mediaType": "application/vnd.oci.image.index.v1+json",
            "manifests": [{"mediaType": media_type, "digest": digest, "size": size}],
        });
        self.release(reference, index, files).await?;
        Ok(made(name, "manifests", &digest))
    }

    /// Makes the release of `version`: its `index`, uploaded, and `files`.
    async fn release(
        &self,
        version: &str,
        index: Value,
        mut files: Vec<ReleaseFile>,
    ) -> Result<(), Response> {
        let (ctx, auth) = (&self.registry.ctx, &self.pusher.auth);
        let denied = |e| error(StatusCode::FORBIDDEN, "DENIED", format!("{e:#}"));
        let target = Target::open(ctx, self.url, &self.name.package, version).await;
        let target = target.map_err(denied)?;
        let index = index.to_string().into_bytes();
        let blobs = push::upload(ctx, &target.remote, auth, &index, RELEASE_MIME).await;
        let parts = blobs.map_err(failed)?;
        let file = ReleaseFile {
            name: oci::INDEX.to_string(),
            size: index.len() as u64,
            sha256: release::sha256(&index),
            parts: parts.into_iter().map(|blob| ReleasePart { blob }).collect(),
        };
        files.insert(0, file);
        let made = target.create(auth, files, Made::default()).await;
        made.map_err(failed)?;
        // Pulled at once, it is among those listed.
        let records = self.registry.records_at(self.name, self.url.clone());
        records.forget().await;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn name(o: &str, r: &str, p: &str) -> Name {
        Name {
            owner: o.into(),
            repo: r.into(),
            package: p.into(),
        }
    }

    #[test]
    fn names() {
        assert_eq!(
            parse("alice.test/tools/manifests/latest"),
            Some((
                name("alice.test", "tools", "tools"),
                Ask::Manifest("latest".into())
            ))
        );
        assert_eq!(
            parse("alice.test/tools/serve/blobs/sha256:ab"),
            Some((
                name("alice.test", "tools", "serve"),
                Ask::Blob("sha256:ab".into())
            ))
        );
        assert_eq!(
            parse("alice.test/tools/blobs/uploads/"),
            Some((name("alice.test", "tools", "tools"), Ask::Upload(None)))
        );
        assert_eq!(
            parse("alice.test/tools/blobs/uploads/ab12"),
            Some((
                name("alice.test", "tools", "tools"),
                Ask::Upload(Some("ab12".into()))
            ))
        );
        assert_eq!(
            parse("alice.test/tools/tags/list"),
            Some((name("alice.test", "tools", "tools"), Ask::Tags))
        );
        for bad in [
            "alice.test/manifests/1.0.0",
            "a/b/c/d/manifests/1",
            "a/b/other",
        ] {
            assert_eq!(parse(bad), None, "{bad}");
        }
    }

    #[test]
    fn what_a_manifest_points_at() {
        let image = json!({
            "config": {"digest": "sha256:c"},
            "layers": [{"digest": "sha256:l1"}, {"digest": "sha256:l2"}],
        });
        let blobs = vec![
            "sha256:l1".to_string(),
            "sha256:l2".into(),
            "sha256:c".into(),
        ];
        assert_eq!(pointed(&image), (blobs, vec![]));
        let index = json!({"manifests": [{"digest": "sha256:m"}]});
        assert_eq!(pointed(&index), (vec![], vec!["sha256:m".to_string()]));
    }

    #[test]
    fn what_is_kept_is_bounded() {
        let ctx = Ctx::new(knotless::env::Env::from_vars(|_| None));
        let registry = Registry {
            ctx,
            public_url: None,
            records: Mutex::default(),
            tokens: Mutex::default(),
            uploads: Mutex::default(),
            staged: Mutex::default(),
        };
        for i in 0..=MOST_KEPT {
            assert!(
                registry
                    .records(&name("alice.test", &format!("r{i}"), "p"))
                    .is_some()
            );
        }
        assert_eq!(registry.records.lock().unwrap().len(), 1);
        assert!(
            registry
                .records(&name("alice.test", "no/such", "p"))
                .is_none()
        );
    }

    #[test]
    fn staging_forgets_only_the_oldest() {
        let file = |name: &str| ReleaseFile {
            name: name.into(),
            size: 1,
            sha256: name.into(),
            parts: Vec::new(),
        };
        let mut staged = Staged::default();
        staged.insert("did:a", "a0".into(), file("a0"));
        // One account past its share loses only its own oldest.
        for i in 0..STAGED_EACH {
            staged.insert("did:b", format!("b{i}"), file("b"));
        }
        staged.insert("did:b", "last".into(), file("b"));
        assert!(staged.get("did:b", "b0").is_none());
        assert!(staged.get("did:b", "b1").is_some());
        assert!(staged.get("did:b", "last").is_some());
        assert!(staged.get("did:a", "a0").is_some());
        // Many accounts past the whole lose the oldest of all.
        for i in 0..MOST_KEPT {
            staged.insert(&format!("did:{i}"), "x".into(), file("x"));
        }
        assert_eq!(staged.files.len(), MOST_KEPT);
        assert!(staged.get("did:a", "a0").is_none());
        assert!(staged.get(&format!("did:{}", MOST_KEPT - 1), "x").is_some());
    }
}