//! `oxy witness serve` (Oxygen spec §14.8, oxygenhub docs/transparency.md):
//! a small transparency-log witness anyone can run. It cosigns the
//! checkpoints of the logs it is told about (C2SP tlog-witness), each only
//! when it extends the last one this witness cosigned, with its own key
//! and clock, so a log cannot show two histories without the witness
//! seeing it.
//!
//! State is one file per log in the state directory: the size and root of
//! the last checkpoint cosigned, replaced atomically.

use axum::extract::State;
use axum::http::{header, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::post;
use axum::Router;
use oxy_core::identity::{Identity, SecretKey};
use oxy_core::tlog::{self, AddCheckpoint, Checkpoint, Note, WitnessError, WitnessKey, Witnessed};
use std::path::PathBuf;
use std::sync::{Arc, Mutex};

/// What a witness serves.
pub struct WitnessConfig {
    /// Its name in cosignatures (a host name, no spaces).
    pub name: String,
    pub identity: Identity,
    pub sk: SecretKey,
    /// The logs it witnesses: their origins and keys.
    pub logs: Vec<(String, Identity)>,
    /// Where it keeps what it cosigned.
    pub state: PathBuf,
    pub tls: Option<crate::TlsFiles>,
}

struct Witness {
    cfg: WitnessConfig,
    /// One add-checkpoint at a time: each reads and moves its log's state.
    lock: Mutex<()>,
}

fn state_file(w: &Witness, origin: &str) -> PathBuf {
    let safe: String = origin
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '.' || c == '-' {
                c
            } else {
                '_'
            }
        })
        .collect();
    w.cfg.state.join(safe)
}

fn read_state(w: &Witness, origin: &str) -> Option<(u64, tlog::Hash)> {
    let text = std::fs::read_to_string(state_file(w, origin)).ok()?;
    let (size, root) = text.trim().split_once(' ')?;
    let root = oxy_core::hash::ObjectRef::from_hex(root)?;
    Some((size.parse().ok()?, root.0))
}

fn plain(status: StatusCode, why: &str) -> Response {
    (
        status,
        [(header::CONTENT_TYPE, "text/plain; charset=utf-8")],
        format!("{why}\n"),
    )
        .into_response()
}

async fn add_checkpoint(State(w): State<Arc<Witness>>, body: String) -> Response {
    let Ok(req) = AddCheckpoint::parse(&body) else {
        return plain(StatusCode::BAD_REQUEST, "not an add-checkpoint request");
    };
    let Some(origin) = Note::parse(&req.note)
        .ok()
        .and_then(|n| Checkpoint::parse(&n.text).ok())
        .map(|c| c.origin)
    else {
        return plain(StatusCode::BAD_REQUEST, "the checkpoint does not parse");
    };
    let Some(key) = w
        .cfg
        .logs
        .iter()
        .find(|(o, _)| *o == origin)
        .map(|(_, k)| k.clone())
    else {
        return plain(
            StatusCode::NOT_FOUND,
            &format!("this witness does not witness {origin}"),
        );
    };
    let _held = w.lock.lock().unwrap_or_else(|p| p.into_inner());
    let latest = read_state(&w, &origin);
    let me = WitnessKey {
        name: &w.cfg.name,
        identity: &w.cfg.identity,
        sk: &w.cfg.sk,
    };
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    let conflict = |n: u64| {
        (
            StatusCode::CONFLICT,
            [(header::CONTENT_TYPE, "text/x.tlog.size")],
            format!("{n}\n"),
        )
            .into_response()
    };
    match tlog::witness(&req, &origin, &key, latest, &me, now) {
        Ok(Witnessed::Cosigned { checkpoint, lines }) => {
            let text = format!(
                "{} {}\n",
                checkpoint.size,
                oxy_core::hash::ObjectRef(checkpoint.root).to_hex()
            );
            if let Err(e) = std::fs::create_dir_all(&w.cfg.state).and_then(|_| {
                oxy_core::durable::write_atomic(&state_file(&w, &origin), text.as_bytes())
            }) {
                return plain(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    &format!("keeping the witness's state: {e}"),
                );
            }
            let body = Note {
                text: String::new(),
                sigs: lines,
            }
            .encode();
            (
                [(header::CONTENT_TYPE, "text/plain; charset=utf-8")],
                body.trim_start_matches('\n').to_string(),
            )
                .into_response()
        }
        Ok(Witnessed::Conflict(n)) => conflict(n),
        Err(WitnessError::Unsigned(e)) => plain(StatusCode::FORBIDDEN, &e),
        Err(WitnessError::Inconsistent(e)) => plain(StatusCode::UNPROCESSABLE_ENTITY, &e),
        Err(WitnessError::Bad(e)) => plain(StatusCode::BAD_REQUEST, &e),
    }
}

/// The witness's routes: `POST /add-checkpoint`.
pub fn router(cfg: WitnessConfig) -> Router {
    Router::new()
        .route("/add-checkpoint", post(add_checkpoint))
        .with_state(Arc::new(Witness {
            cfg,
            lock: Mutex::new(()),
        }))
}

/// Serve a witness on `listener` until stopped.
pub fn run_blocking(
    listener: std::net::TcpListener,
    mut cfg: WitnessConfig,
) -> std::io::Result<()> {
    let rt = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()?;
    let tls = cfg.tls.take();
    rt.block_on(async move {
        listener.set_nonblocking(true)?;
        match tls {
            Some(tls) => {
                let _ = rustls::crypto::ring::default_provider().install_default();
                let rustls_cfg =
                    axum_server::tls_rustls::RustlsConfig::from_pem_file(&tls.cert, &tls.key)
                        .await
                        .map_err(|e| std::io::Error::other(format!("loading TLS files: {e}")))?;
                axum_server::from_tcp_rustls(listener, rustls_cfg)?
                    .serve(router(cfg).into_make_service())
                    .await
            }
            None => axum::serve(tokio::net::TcpListener::from_std(listener)?, router(cfg)).await,
        }
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use axum::body::Body;
    use axum::http::Request;
    use tower::ServiceExt;

    #[tokio::test]
    async fn the_witness_cosigns_and_keeps_what_it_cosigned() {
        let dir = tempfile::tempdir().unwrap();
        let log_sk = SecretKey::generate();
        let log = log_sk.identity("log", "log@example.org", 1);
        let w_sk = SecretKey::generate();
        let w = w_sk.identity("witness", "w@example.org", 1);
        let cfg = || WitnessConfig {
            name: "witness.example".into(),
            identity: w.clone(),
            sk: w_sk.clone(),
            logs: vec![("hub.example/log".into(), log.clone())],
            state: dir.path().join("state"),
            tls: None,
        };
        let mut stored: Vec<tlog::Hash> = Vec::new();
        for n in 0..9u64 {
            stored.extend(
                tlog::stored_hashes_for_leaf(n, tlog::leaf_hash(&[n as u8]), &stored).unwrap(),
            );
        }
        let note = |n: u64| {
            Checkpoint {
                origin: "hub.example/log".into(),
                size: n,
                root: tlog::tree_hash(n, &stored).unwrap(),
                extensions: vec![],
            }
            .sign(&log, &log_sk)
            .unwrap()
            .encode()
        };
        let post = |body: String| async {
            let res = router(cfg())
                .oneshot(
                    Request::post("/add-checkpoint")
                        .body(Body::from(body))
                        .unwrap(),
                )
                .await
                .unwrap();
            let st = res.status();
            let b = axum::body::to_bytes(res.into_body(), 1 << 20)
                .await
                .unwrap();
            (st, String::from_utf8(b.to_vec()).unwrap())
        };
        let (st, lines) = post(
            AddCheckpoint {
                old_size: 0,
                proof: vec![],
                note: note(4),
            }
            .encode(),
        )
        .await;
        assert_eq!(st, StatusCode::OK, "{lines}");
        let cosigned =
            Note::parse(&format!("{}\n{lines}", Note::parse(&note(4)).unwrap().text)).unwrap();
        assert!(cosigned.cosigned_by("witness.example", &w).is_ok());
        // Kept across a restart (a new router): a wrong old size is told.
        let (st, size) = post(
            AddCheckpoint {
                old_size: 0,
                proof: vec![],
                note: note(9),
            }
            .encode(),
        )
        .await;
        assert_eq!((st, size.trim()), (StatusCode::CONFLICT, "4"));
        let proof = tlog::prove_consistency(9, 4, &stored).unwrap();
        let (st, _) = post(
            AddCheckpoint {
                old_size: 4,
                proof,
                note: note(9),
            }
            .encode(),
        )
        .await;
        assert_eq!(st, StatusCode::OK);
    }
}