//! `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", "[email protected]", 1);

        let w_sk = SecretKey::generate();

        let w = w_sk.identity("witness", "[email protected]", 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);

    }

}