//! Images by hash over the API (oxygenflow#39): `flow rootfs push` sends

//! an image's missing chunks and then its manifest; runners, and `flow

//! rootfs pull`, fetch a manifest and the chunks they lack.

use super::{caller, not_found, ApiError};

use crate::Shared;

use axum::body::Bytes;

use axum::extract::{Path, State};

use axum::http::{header, HeaderMap, StatusCode};

use axum::response::{IntoResponse, Response};

use axum::Json;

use jet::images::{is_hash, Manifest};

use serde::{Deserialize, Serialize};

/// Most chunk names one `missing` question may ask about.

const MAX_ASK: usize = 200_000;

/// Who sends images: an account whose token may write repositories.

async fn sender(app: &Shared, headers: &HeaderMap) -> Result<String, ApiError> {

    let c = caller(app, headers).await?;

    Ok(c.require("repo:write")?.username.clone())

}

/// Who fetches images: a runner, or any account (for `flow rootfs pull`).

async fn fetcher(app: &Shared, headers: &HeaderMap) -> Result<(), ApiError> {

    if crate::flow::runner_auth(app, headers).await.is_ok() {

        return Ok(());

    }

    let c = caller(app, headers).await?;

    c.account.as_ref().map(|_| ()).ok_or_else(|| ApiError(StatusCode::UNAUTHORIZED, "a runner or personal access token is required".into()))

}

#[derive(Deserialize)]

pub struct Ask {

    chunks: Vec<String>,

}

#[derive(Serialize)]

pub struct Missing {

    missing: Vec<String>,

}

/// `POST /flow/images/missing {"chunks": [..]}`: which of these chunks the

/// hub lacks.

pub async fn missing(State(app): State<Shared>, headers: HeaderMap, Json(ask): Json<Ask>) -> Result<Json<Missing>, ApiError> {

    sender(&app, &headers).await?;

    if ask.chunks.len() > MAX_ASK {

        return Err(ApiError(StatusCode::BAD_REQUEST, format!("ask about at most {MAX_ASK} chunks at once")));

    }

    let store = app.images();

    let missing = tokio::task::spawn_blocking(move || {

        let mut seen = std::collections::HashSet::new();

        ask.chunks.into_iter().filter(|h| is_hash(h) && seen.insert(h.clone()) && !store.has_chunk(h)).collect()

    })

    .await

    .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    Ok(Json(Missing { missing }))

}

/// `PUT /flow/images/chunks/{hash}`: one chunk, checked against its name.

pub async fn put_chunk(

    State(app): State<Shared>,

    headers: HeaderMap,

    Path(hash): Path<String>,

    body: Bytes,

) -> Result<StatusCode, ApiError> {

    sender(&app, &headers).await?;

    if !is_hash(&hash) || body.is_empty() || body.len() > oxy_core::chunk::MAX_SIZE {

        return Err(ApiError(StatusCode::BAD_REQUEST, "not a chunk: a name is 64 hex digits, a chunk 1 byte to 256 KiB".into()));

    }

    if blake3::hash(&body).to_hex().as_str() != hash {

        return Err(ApiError(StatusCode::BAD_REQUEST, format!("those bytes are not chunk {}", &hash[..12])));

    }

    let store = app.images();

    tokio::task::spawn_blocking(move || store.put_chunk(&hash, &body))

        .await

        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?

        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("keep the chunk: {e}")))?;

    Ok(StatusCode::NO_CONTENT)

}

/// `PUT /flow/images/{hash}`: the manifest, accepted once every chunk is

/// here and the image they make has this hash.

pub async fn put_manifest(

    State(app): State<Shared>,

    headers: HeaderMap,

    Path(hash): Path<String>,

    Json(m): Json<Manifest>,

) -> Result<(StatusCode, Json<serde_json::Value>), ApiError> {

    let who = sender(&app, &headers).await?;

    if m.hash != hash {

        return Err(ApiError(StatusCode::BAD_REQUEST, format!("the manifest is for {}, not {}", m.hash, hash)));

    }

    let store = app.images();

    let kept = m.clone();

    tokio::task::spawn_blocking(move || store.verify_and_keep(&kept))

        .await

        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?

        .map_err(|e| ApiError(StatusCode::CONFLICT, e))?;

    crate::images::record(&app.db, &m, &who).await?;

    crate::db::accounts::audit(&app.db, None, "flow.image", &who, &format!("{} ({} MiB)", &m.hash[..12], m.size >> 20)).await.ok();

    Ok((StatusCode::CREATED, Json(serde_json::json!({ "hash": m.hash, "size": m.size, "chunks": m.chunks.len() }))))

}

/// `GET /flow/images/{hash}`: an image's manifest.

pub async fn get_manifest(State(app): State<Shared>, headers: HeaderMap, Path(hash): Path<String>) -> Result<Json<Manifest>, ApiError> {

    fetcher(&app, &headers).await?;

    app.images().manifest(&hash).map(Json).ok_or_else(|| not_found("this hub does not hold that image; send it with flow rootfs push"))

}

/// `GET /flow/images/chunks/{hash}`: one chunk's bytes.

pub async fn get_chunk(State(app): State<Shared>, headers: HeaderMap, Path(hash): Path<String>) -> Result<Response, ApiError> {

    fetcher(&app, &headers).await?;

    let store = app.images();

    let bytes = tokio::task::spawn_blocking(move || store.chunk(&hash))

        .await

        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?

        .map_err(|_| not_found("no such chunk"))?;

    Ok(([(header::CONTENT_TYPE, "application/octet-stream"), (header::CACHE_CONTROL, "private, max-age=31536000, immutable")], bytes)

        .into_response())

}