//! The HTTP side: the router, the token check, and `/v1`, where ipx servers send snapshots and read the
//! catalogue.

use crate::config::Access;
use crate::directory::Shared;
use crate::wire::{Blocked, FeedPage, Instance, Listing, MAX_SNAPSHOT, Me, SnapshotEntry, SnapshotResult};
use crate::{otel, site};
use axum::extract::{Path, Query, Request, State};
use axum::http::{StatusCode, header};
use axum::middleware::Next;
use axum::response::{IntoResponse, Response};
use axum::routing::{delete, get, post};
use axum::{Extension, Json, Router};
use serde::Deserialize;
use std::time::Instant;
use tower_http::compression::CompressionLayer;
use tower_http::decompression::RequestDecompressionLayer;
use tracing::Instrument;

/// Every route: the API under `/v1`, behind a token, and the public pages.
pub fn router(dir: Shared) -> Router {
    let v1 = Router::new()
        .route("/v1/me", get(me))
        .route("/v1/feeds", get(list))
        .route("/v1/popular", get(popular))
        .route("/v1/feeds/{id}", get(feed).delete(remove))
        .route("/v1/feeds/{id}/refresh", post(refresh))
        .route("/v1/snapshot", post(snapshot))
        .route("/v1/blocked", get(blocked))
        .route("/v1/blocked/{id}", delete(unblock))
        .route("/v1/instances", get(instances))
        .route("/v1/instances/{name}", delete(drop_instance))
        .layer(axum::middleware::from_fn_with_state(dir.clone(), authenticate));
    Router::new()
        .route("/healthz", get(healthz))
        .merge(v1)
        // Anyone may browse, no token, read only (#215).
        .merge(site::router())
        .route_layer(axum::middleware::from_fn(remember_route))
        // Brotli, zstd or gzip as the caller takes them: the feed list went to ipx as 603 KB of
        // JSON, most of a Directory page's time (#253). Inside the access log, which records what
        // was sent.
        .layer(CompressionLayer::new().br(true).zstd(true).gzip(true))
        // And takes a request's body gzipped: ipx sends its snapshot so, about 100 KB as JSON.
        .layer(RequestDecompressionLayer::new().gzip(true))
        .layer(axum::middleware::from_fn(access_log))
        .with_state(dir)
}

pub(crate) struct ApiError(StatusCode, String);

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        (self.0, Json(serde_json::json!({ "error": self.1 }))).into_response()
    }
}

impl From<anyhow::Error> for ApiError {
    fn from(e: anyhow::Error) -> Self {
        tracing::error!(error = %format!("{e:#}"), "request failed");
        ApiError(StatusCode::INTERNAL_SERVER_ERROR, "something went wrong; the directory's log says what".into())
    }
}

/// Who made the request, for the access log, set by `authenticate`.
#[derive(Clone)]
pub(crate) struct Who(String);

/// Puts the route matched on the response for `access_log`, which runs outside routing: by path,
/// each feed's page was a row of its own in the request metric.
async fn remember_route(route: axum::extract::MatchedPath, req: Request, next: Next) -> Response {
    let mut resp = next.run(req).await;
    resp.extensions_mut().insert(route);
    resp
}

/// One line a request, with the token's name and never the token: what the Grafana panels read.
pub(crate) async fn access_log(req: Request, next: Next) -> Response {
    let (method, path) = (req.method().clone(), req.uri().path().to_owned());
    let start = Instant::now();
    // A span for each request but the healthcheck's, a child of the caller's when it sent a
    // traceparent, as ipx does on its calls here (#213).
    let span = if path == "/healthz" {
        tracing::Span::none()
    } else {
        tracing::info_span!(
            "http",
            otel.name = %format!("{method} {path}"),
            http.request.method = %method,
            url.path = %path,
            http.response.status_code = tracing::field::Empty,
        )
    };
    if let Some(parent) = otel::remote_parent(req.headers(), None) {
        use tracing_opentelemetry::OpenTelemetrySpanExt;
        let _ = span.set_parent(parent);
    }
    let resp = Instrument::instrument(next.run(req), span.clone()).await;
    span.record("http.response.status_code", resp.status().as_u16());
    let _in = span.enter();
    let who = resp.extensions().get::<Who>().map_or("", |w| w.0.as_str()).to_owned();
    let status = resp.status().as_u16();
    if path != "/healthz" {
        let route = resp.extensions().get::<axum::extract::MatchedPath>().map_or("", |r| r.as_str());
        crate::metrics::REQUESTS.observe(
            &format!(
                "method=\"{}\",route=\"{}\",status=\"{status}\"",
                crate::metrics::method(method.as_str()),
                crate::metrics::label(route)
            ),
            start.elapsed().as_secs_f64(),
        );
        tracing::info!(
            target: "ipx_directory::http",
            method = %method, path = %path, status, duration_ms = start.elapsed().as_millis() as u64, token = %who,
            "{method} {path} -> {status}"
        );
    }
    resp
}

pub(crate) async fn authenticate(State(dir): State<Shared>, mut req: Request, next: Next) -> Response {
    let presented = req
        .headers()
        .get(header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.strip_prefix("Bearer "))
        .map(str::trim)
        .filter(|t| !t.is_empty());
    let Some(me) = presented.and_then(|t| dir.caller(t)) else {
        return ApiError(StatusCode::UNAUTHORIZED, "a valid token is needed: Authorization: Bearer <token>".into())
            .into_response();
    };
    let who = Who(me.name.clone());
    req.extensions_mut().insert(me);
    let mut resp = next.run(req).await;
    resp.extensions_mut().insert(who);
    resp
}

pub(crate) fn need(me: &Me, access: Access) -> Result<(), ApiError> {
    if me.access >= access {
        Ok(())
    } else {
        let what = match access {
            Access::Ro => "browse",
            Access::Rw => "send snapshots",
            Access::Rwa => "administer the directory",
        };
        Err(ApiError(StatusCode::FORBIDDEN, format!("this token may not {what}")))
    }
}

pub(crate) async fn healthz(State(dir): State<Shared>) -> Result<&'static str, ApiError> {
    dir.store.ping().await?;
    Ok("ok")
}

pub(crate) async fn me(Extension(me): Extension<Me>) -> Json<Me> {
    Json(me)
}

pub(crate) async fn list(State(dir): State<Shared>) -> Result<Json<Vec<Listing>>, ApiError> {
    Ok(Json(dir.store.list().await?))
}

#[derive(Deserialize)]
pub(crate) struct Limit {
    limit: Option<u64>,
}

pub(crate) async fn popular(State(dir): State<Shared>, Query(q): Query<Limit>) -> Result<Json<Vec<Listing>>, ApiError> {
    Ok(Json(dir.store.popular(q.limit.unwrap_or(10).clamp(1, 100)).await?))
}

pub(crate) async fn feed(State(dir): State<Shared>, Path(id): Path<String>) -> Result<Json<FeedPage>, ApiError> {
    dir.store
        .feed(&id)
        .await?
        .map(Json)
        .ok_or_else(|| ApiError(StatusCode::NOT_FOUND, format!("{id:?} is not in the directory")))
}

pub(crate) async fn snapshot(
    State(dir): State<Shared>,
    Extension(me): Extension<Me>,
    Json(entries): Json<Vec<SnapshotEntry>>,
) -> Result<Json<SnapshotResult>, ApiError> {
    need(&me, Access::Rw)?;
    if entries.len() > MAX_SNAPSHOT {
        return Err(ApiError(StatusCode::PAYLOAD_TOO_LARGE, format!("a snapshot lists at most {MAX_SNAPSHOT} feeds")));
    }
    let r = dir.store.snapshot(&me.name, &entries).await?;
    dir.wake.notify_one();
    crate::metrics::SNAPSHOTS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    tracing::info!(
        instance = %me.name, accepted = r.accepted, refused_private = r.refused_private,
        refused_blocked = r.refused_blocked, refused_invalid = r.refused_invalid, "snapshot"
    );
    Ok(Json(r))
}

pub(crate) async fn remove(
    State(dir): State<Shared>,
    Extension(me): Extension<Me>,
    Path(id): Path<String>,
) -> Result<StatusCode, ApiError> {
    need(&me, Access::Rwa)?;
    dir.store.block(&id).await?;
    tracing::info!(feed = %id, by = %me.name, "feed removed and blocked");
    Ok(StatusCode::NO_CONTENT)
}

pub(crate) async fn blocked(
    State(dir): State<Shared>,
    Extension(me): Extension<Me>,
) -> Result<Json<Vec<Blocked>>, ApiError> {
    need(&me, Access::Rwa)?;
    Ok(Json(dir.store.blocked().await?))
}

pub(crate) async fn unblock(
    State(dir): State<Shared>,
    Extension(me): Extension<Me>,
    Path(id): Path<String>,
) -> Result<StatusCode, ApiError> {
    need(&me, Access::Rwa)?;
    if !dir.store.unblock(&id).await? {
        return Err(ApiError(StatusCode::NOT_FOUND, format!("{id:?} is not blocked")));
    }
    tracing::info!(feed = %id, by = %me.name, "feed unblocked");
    Ok(StatusCode::NO_CONTENT)
}

pub(crate) async fn instances(
    State(dir): State<Shared>,
    Extension(me): Extension<Me>,
) -> Result<Json<Vec<Instance>>, ApiError> {
    need(&me, Access::Rwa)?;
    Ok(Json(dir.store.instances().await?))
}

pub(crate) async fn drop_instance(
    State(dir): State<Shared>,
    Extension(me): Extension<Me>,
    Path(name): Path<String>,
) -> Result<StatusCode, ApiError> {
    need(&me, Access::Rwa)?;
    let n = dir.store.drop_instance(&name).await?;
    tracing::info!(instance = %name, feeds = n, by = %me.name, "snapshot dropped");
    Ok(StatusCode::NO_CONTENT)
}

pub(crate) async fn refresh(
    State(dir): State<Shared>,
    Extension(me): Extension<Me>,
    Path(id): Path<String>,
) -> Result<Json<FeedPage>, ApiError> {
    need(&me, Access::Rwa)?;
    let due = dir
        .store
        .due_one(&id)
        .await?
        .ok_or_else(|| ApiError(StatusCode::NOT_FOUND, format!("{id:?} is not in the directory")))?;
    dir.refresh(&due).await?;
    feed(State(dir), Path(id)).await
}

#[cfg(test)]
mod tests {

    use super::*;
    use crate::config;
    use crate::config::Token;
    use crate::directory::Directory;
    use crate::store;
    use crate::store::Store;
    use axum::body::Body;
    use tower::ServiceExt;

    async fn app() -> (Router, Shared) {
        let tokens = ["ro", "rw", "rwa"]
            .iter()
            .map(|n| Token {
                name: format!("{n}-token"),
                access: serde_json::from_value(serde_json::json!(n)).unwrap(),
                sha256: config::hash(&format!("secret-{n}")),
            })
            .collect();
        let dir = Directory::new(Store::scratch().await, tokens, Default::default()).unwrap();
        (router(dir.clone()), dir)
    }

    async fn call(
        app: &Router,
        method: &str,
        path: &str,
        token: Option<&str>,
        body: Option<serde_json::Value>,
    ) -> (u16, serde_json::Value) {
        let mut req = axum::http::Request::builder().method(method).uri(path);
        if let Some(t) = token {
            req = req.header("authorization", format!("Bearer secret-{t}"));
        }
        let req = match body {
            Some(b) => req.header("content-type", "application/json").body(Body::from(b.to_string())),
            None => req.body(Body::empty()),
        }
        .unwrap();
        let resp = app.clone().oneshot(req).await.unwrap();
        let status = resp.status().as_u16();
        let bytes = axum::body::to_bytes(resp.into_body(), 1 << 20).await.unwrap();
        (status, serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null))
    }

    #[tokio::test]
    async fn a_snapshot_may_come_gzipped() {
        use std::io::Write as _;
        let (app, _) = app().await;
        let json = serde_json::json!([{"url": "https://example.com/feed.xml", "subscribers": 2}]).to_string();
        let mut gz = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
        gz.write_all(json.as_bytes()).unwrap();
        // As ipx sends it (#253).
        let req = axum::http::Request::builder()
            .method("POST")
            .uri("/v1/snapshot")
            .header("authorization", "Bearer secret-rw")
            .header("content-type", "application/json")
            .header("content-encoding", "gzip")
            .body(Body::from(gz.finish().unwrap()))
            .unwrap();
        let resp = app.clone().oneshot(req).await.unwrap();
        assert_eq!(resp.status(), 200);
        let got: serde_json::Value =
            serde_json::from_slice(&axum::body::to_bytes(resp.into_body(), 1 << 20).await.unwrap()).unwrap();
        assert_eq!(got["accepted"], 1);
    }

    #[tokio::test]
    async fn answers_come_compressed_as_the_caller_asks() {
        let (app, _) = app().await;
        for (asked, sent) in [("br", Some("br")), ("zstd", Some("zstd")), ("gzip", Some("gzip")), ("identity", None)] {
            // A page, not an empty feed list: under 32 bytes is sent as it is.
            let req =
                axum::http::Request::builder().uri("/").header("accept-encoding", asked).body(Body::empty()).unwrap();
            let resp = app.clone().oneshot(req).await.unwrap();
            assert_eq!(resp.status(), 200);
            let got = resp.headers().get("content-encoding").map(|v| v.to_str().unwrap().to_owned());
            assert_eq!(got.as_deref(), sent, "asked for {asked} (#253)");
        }
    }

    async fn get_page(app: &Router, path: &str) -> (u16, String, Option<String>) {
        let req = axum::http::Request::builder().uri(path).body(Body::empty()).unwrap();
        let resp = app.clone().oneshot(req).await.unwrap();
        let status = resp.status().as_u16();
        let csp = resp.headers().get("content-security-policy").map(|v| v.to_str().unwrap().to_owned());
        let bytes = axum::body::to_bytes(resp.into_body(), 1 << 22).await.unwrap();
        (status, String::from_utf8(bytes.to_vec()).unwrap(), csp)
    }

    #[tokio::test]
    async fn anyone_browses_the_pages_and_the_api_still_needs_a_token() {
        let (app, dir) = app().await;
        let snap = serde_json::json!([{ "url": "https://x.test/a.xml", "subscribers": 3 }]);
        call(&app, "POST", "/v1/snapshot", Some("rw"), Some(snap)).await;
        store::tests::fetch_all(&dir.store).await;
        let id = dir.store.list().await.unwrap()[0].id.clone();

        let (s, home, csp) = get_page(&app, "/").await;
        assert_eq!(s, 200);
        assert!(csp.unwrap().contains("script-src 'self'"));
        assert!(home.contains("Most subscribed") && home.contains(&format!("/feed/{id}")), "{home}");
        assert!(get_page(&app, "/all").await.1.contains("3 people subscribe"));
        let (s, page, _) = get_page(&app, &format!("/feed/{id}")).await;
        assert_eq!(s, 200);
        // Its address, to subscribe to anywhere.
        assert!(page.contains("value=\"https://x.test/a.xml\""), "{page}");
        assert!(!page.contains("<script>"), "the publisher's HTML is text here");
        assert_eq!(get_page(&app, "/feed/nope").await.0, 404);
        assert_eq!(get_page(&app, "/category/Nothing").await.0, 404);
        assert_eq!(get_page(&app, "/search?q=").await.0, 303);
        assert_eq!(get_page(&app, "/site.js").await.0, 200);
        // Crawlers are pointed at every page (#380).
        let (s, robots, _) = get_page(&app, "/robots.txt").await;
        assert!(s == 200 && robots.contains("Disallow: /v1/") && robots.contains("/sitemap.xml"), "{robots}");
        let (s, map, _) = get_page(&app, "/sitemap.xml").await;
        assert_eq!(s, 200);
        assert!(map.contains(&format!("/feed/{id}</loc>")) && map.contains("/category/Podcasting</loc>"), "{map}");
        // Reading the pages takes no token; the API still does.
        assert_eq!(call(&app, "GET", "/v1/feeds", None, None).await.0, 401);
        assert_eq!(call(&app, "DELETE", &format!("/v1/feeds/{id}"), None, None).await.0, 401);
        assert_eq!(call(&app, "DELETE", &format!("/feed/{id}"), None, None).await.0, 405);
    }

    /// What the log layer wrote, kept for the test to read.
    struct Lines(std::sync::Arc<std::sync::Mutex<Vec<u8>>>);

    impl std::io::Write for Lines {
        fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
            self.0.lock().unwrap().extend_from_slice(b);
            Ok(b.len())
        }
        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    #[tokio::test]
    async fn a_snapshots_log_line_has_the_fields_grafana_reads() {
        use tracing_subscriber::layer::SubscriberExt;
        let buf = std::sync::Arc::new(std::sync::Mutex::new(vec![]));
        let b = buf.clone();
        // Global, as the directory's own is; other tests' snapshots land here too, and each of
        // their lines must carry the same fields.
        tracing::subscriber::set_global_default(
            tracing_subscriber::registry().with(crate::otel::json_layer(move || Lines(b.clone()), false)),
        )
        .unwrap();
        let (app, _dir) = app().await;
        let snap = serde_json::json!([{ "url": "https://x.test/log.xml", "subscribers": 1 }]);
        assert_eq!(call(&app, "POST", "/v1/snapshot", Some("rw"), Some(snap)).await.0, 200);

        let text = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
        let lines: Vec<serde_json::Value> = text.lines().map(|l| serde_json::from_str(l).expect(l)).collect();
        let snaps: Vec<&serde_json::Value> = lines.iter().filter(|l| l["message"] == "snapshot").collect();
        assert!(!snaps.is_empty(), "no snapshot line in {text}");
        for l in snaps {
            assert!(l["instance"].is_string(), "{l}");
            for n in ["accepted", "refused_private", "refused_blocked", "refused_invalid"] {
                assert!(l[n].is_u64(), "{n} in {l}");
            }
        }
    }

    #[tokio::test]
    async fn dropping_a_servers_snapshot_takes_only_the_feeds_nobody_else_lists() {
        let (app, dir) = app().await;
        let feed = |url: &str| serde_json::json!({ "url": url, "subscribers": 1 });
        let mine = serde_json::json!([feed("https://only-rw.test/feed"), feed("https://both.test/feed")]);
        assert_eq!(call(&app, "POST", "/v1/snapshot", Some("rw"), Some(mine)).await.0, 200);
        let theirs = serde_json::json!([feed("https://both.test/feed")]);
        assert_eq!(call(&app, "POST", "/v1/snapshot", Some("rwa"), Some(theirs)).await.0, 200);

        assert_eq!(call(&app, "DELETE", "/v1/instances/rw-token", Some("rw"), None).await.0, 403, "admins only");
        assert_eq!(call(&app, "DELETE", "/v1/instances/rw-token", Some("rwa"), None).await.0, 204);
        let left: Vec<(String, i64)> =
            dir.store.instances().await.unwrap().into_iter().map(|i| (i.name, i.feeds)).collect();
        assert_eq!(left, [("rwa-token".to_owned(), 1)]);
        let gone = dir.store.due_one(&store::feed_id("https://only-rw.test/feed")).await.unwrap();
        assert!(gone.is_none(), "nobody lists it now");
        let kept = dir.store.due_one(&store::feed_id("https://both.test/feed")).await.unwrap();
        assert!(kept.is_some(), "another server still does");
    }

    #[tokio::test]
    async fn each_access_level_does_what_it_may_and_no_more() {
        let (app, dir) = app().await;
        let snap = serde_json::json!([{ "url": "https://x.test/a.xml", "subscribers": 2, "category": "Comedy" }]);

        assert_eq!(call(&app, "GET", "/v1/feeds", None, None).await.0, 401);
        assert_eq!(call(&app, "GET", "/v1/feeds", Some("wrong"), None).await.0, 401);
        assert_eq!(call(&app, "GET", "/healthz", None, None).await.0, 200);

        let (s, me) = call(&app, "GET", "/v1/me", Some("ro"), None).await;
        assert_eq!((s, me["name"].as_str(), me["access"].as_str()), (200, Some("ro-token"), Some("ro")));
        assert_eq!(call(&app, "POST", "/v1/snapshot", Some("ro"), Some(snap.clone())).await.0, 403);

        let (s, r) = call(&app, "POST", "/v1/snapshot", Some("rw"), Some(snap.clone())).await;
        assert_eq!((s, r["accepted"].as_u64()), (200, Some(1)));
        store::tests::fetch_all(&dir.store).await;
        let (_, list) = call(&app, "GET", "/v1/feeds", Some("ro"), None).await;
        let id = list[0]["id"].as_str().unwrap().to_owned();
        assert_eq!(list[0]["subscribers"], 2);
        assert_ne!(list[0]["category"], "Comedy", "the feed's own category wins over the sender's");
        assert_eq!(call(&app, "GET", &format!("/v1/feeds/{id}"), Some("ro"), None).await.0, 200);
        assert_eq!(call(&app, "GET", "/v1/feeds/nope", Some("ro"), None).await.0, 404);
        assert_eq!(call(&app, "GET", "/v1/popular?limit=5", Some("ro"), None).await.1[0]["id"], id.as_str());

        for (m, p) in [
            ("DELETE", format!("/v1/feeds/{id}")),
            ("GET", "/v1/blocked".into()),
            ("DELETE", "/v1/instances/rw-token".into()),
        ] {
            assert_eq!(call(&app, m, &p, Some("rw"), None).await.0, 403, "{m} {p}");
        }
        assert_eq!(call(&app, "DELETE", &format!("/v1/feeds/{id}"), Some("rwa"), None).await.0, 204);
        assert_eq!(call(&app, "GET", "/v1/blocked", Some("rwa"), None).await.1[0]["id"], id.as_str());
        assert_eq!(call(&app, "POST", "/v1/snapshot", Some("rw"), Some(snap.clone())).await.1["refused_blocked"], 1);
        assert_eq!(call(&app, "DELETE", &format!("/v1/blocked/{id}"), Some("rwa"), None).await.0, 204);
        assert_eq!(call(&app, "DELETE", &format!("/v1/blocked/{id}"), Some("rwa"), None).await.0, 404);
        call(&app, "POST", "/v1/snapshot", Some("rw"), Some(snap)).await;
        assert_eq!(call(&app, "DELETE", "/v1/instances/rw-token", Some("rwa"), None).await.0, 204);
        assert_eq!(call(&app, "GET", "/v1/feeds", Some("ro"), None).await.1, serde_json::json!([]));
    }

    #[tokio::test]
    async fn a_revoked_token_loses_its_access_and_its_counts() {
        let dir = Directory::new(Store::scratch().await, vec![], Default::default()).unwrap();
        let t = |n: &str| Token { name: n.into(), access: Access::Rw, sha256: config::hash(&format!("secret-{n}")) };
        dir.set_tokens(vec![t("a"), t("b")]).await.unwrap();
        let app = router(dir.clone());
        let snap = serde_json::json!([{ "url": "https://x.test/a.xml", "subscribers": 1 }]);
        call(&app, "POST", "/v1/snapshot", Some("a"), Some(snap.clone())).await;
        call(&app, "POST", "/v1/snapshot", Some("b"), Some(snap)).await;
        store::tests::fetch_all(&dir.store).await;
        assert_eq!(call(&app, "GET", "/v1/feeds", Some("a"), None).await.1[0]["subscribers"], 2);

        dir.set_tokens(vec![t("a")]).await.unwrap();
        assert_eq!(call(&app, "GET", "/v1/feeds", Some("b"), None).await.0, 401);
        assert_eq!(call(&app, "GET", "/v1/feeds", Some("a"), None).await.1[0]["subscribers"], 1);
    }
}