// atlibs serve <ref> [<file arg>] [--port 8787] [--host 127.0.0.1] [--origin URL | ATLIBS_ORIGIN] [--oauth-path oauth/]
//
// The no-server option, and the module author's loop: an HTTP server (hyper) that
//   GET  /                                       the module's actions (a small form page in a browser)
//   POST /run/<entry>                            body: the input JSON; answers the run's result JSON
//   POST /run/<instance>/<entry>                 the same, for a composition's instance
//   GET  /oauth/<session>/client-metadata.json   ClientMetadataHandler (atlibs_host::oauth), the
//                                                client_id on --origin / ATLIBS_ORIGIN (unset: only
//                                                loopback requests, on their own origin)
//   GET  /.well-known/did.json, /.well-known/atproto-did, /.well-known/<file>, /xrpc/com.atproto.repo.{getRecord,
//        listRecords,describeRepo}           with --origin and a session record declaring `app`: the app's own
//                                                identity did:web:<origin host> and read-only repo (AppRepo)
//   GET  /oauth/login?handle=<handle|did>        sign in (atlibs_host::oauth_client): this server's own OAuth
//   GET  /oauth/callback                         client, confidential on --origin (/oauth/client-metadata.json,
//   POST /oauth/logout                           /oauth/jwks.json), else the loopback client on 127.0.0.1:<port>;
//                                                the callback sets the atlibs_session cookie (also a bearer)
// A POST /run/* with a session (cookie or `Authorization: Bearer <it>`) runs as that user, with their OAuth
// session alone; without one, as --as / --app-password if given, else signed out. A local module ref (a module
// directory, a lockfile name) is loaded again for every run, so `cargo wasm` while it serves is picked
// up; remote refs and compositions are loaded once at start. Loopback, IP or --origin Hosts only; a
// POST /run/* needs content-type: application/json and no foreign Origin (no cross-site runs).

use std::collections::HashMap;
use std::net::SocketAddr;
use std::rc::Rc;
use std::sync::{Arc, Mutex};

use atlibs_host::host::Session;
use atlibs_host::oauth::{AppRepo, ClientMetadataHandler, scope_list};
use atlibs_host::oauth_client::{self, Client};
use atlibs_host::signin::{SignIn, handle_of};
use atlibs_host::{Signal, pretty};
use http_body_util::{BodyExt, Full};
use hyper::body::{Bytes, Incoming};
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper_util::rt::TokioIo;
use serde_json::{Map, Value, json};

use crate::registry::{self, is_remote};
use crate::run::{Host, Target, input_of, target, target_of};
use crate::{Args, Cli};

pub struct Served {
    pub cli: Cli,
    pub host: Host,
    /// the ref words (`<ref> [<file arg>]`)
    pub words: Vec<String>,
    /// loaded once for remote refs; None: load per run
    pub fixed: Option<Target>,
    pub oauth: ClientMetadataHandler,
    /// the app's identity and read-only repo (--origin, and a session record that declares `app`)
    pub app: Option<AppRepo>,
    /// entry names by module key, for the index
    pub actions: Map<String, Value>,
    /// this server's OAuth client (None until main knows the loopback port, without --origin)
    pub client: Option<Client>,
    pub scope: String,
    /// signed-in users' sessions by DID: one per user in this process (one refresh for all their runs)
    pub sessions: Mutex<HashMap<String, Arc<Session>>>,
}

fn reply(status: u16, content_type: &str, body: String) -> http::Response<String> {
    http::Response::builder().status(status).header("content-type", content_type).body(body).expect("valid response")
}

fn json_reply(status: u16, value: &Value) -> http::Response<String> {
    reply(status, "application/json", format!("{}\n", pretty(value)))
}

const PAGE: &str = r#"<!doctype html><meta charset="utf-8"><title>atlibs serve</title>
<style>body{font:14px system-ui;max-width:52rem;margin:2rem auto;padding:0 1rem}textarea{width:100%;font:13px monospace}pre{background:#8881;padding:.5rem;overflow:auto}</style>
<h1>atlibs serve</h1><p id="ref"></p><form id="login"><input name="handle" placeholder="handle.bsky.social"> <button>Sign in</button> <button type="button" id="logout">Sign out</button></form><div id="forms"></div>
<script type="module">
const index = INDEX;
document.getElementById("ref").textContent = index.ref;
const login = document.getElementById("login");
login.action = index.login.split("?")[0];
document.getElementById("logout").onclick = async () => alert(await (await fetch(index.login.replace(/login\?.*/, "logout"), { method: "POST" })).text());
for (const path of index.run) {
	const form = document.createElement("form");
	form.innerHTML = `<h3></h3><textarea rows="3">{}</textarea><button>Run</button><pre></pre>`;
	form.querySelector("h3").textContent = `POST ${path}`;
	form.onsubmit = async (event) => {
		event.preventDefault();
		const out = form.querySelector("pre");
		out.textContent = "running...";
		const res = await fetch(path, { method: "POST", headers: { "content-type": "application/json" }, body: form.querySelector("textarea").value });
		out.textContent = await res.text();
	};
	document.getElementById("forms").append(form);
}
</script>"#;

impl Served {
    fn index(&self) -> Value {
        let mut run = vec![];
        let comp = match &self.fixed {
            Some(Target::Composition { comp, .. }) => Some(comp.manifest.clone()),
            _ => None,
        };
        for (key, entries) in &self.actions {
            for entry in entries.as_array().into_iter().flatten().filter_map(Value::as_str) {
                match &comp {
                    Some(m) => {
                        for inst in m["instances"].as_array().into_iter().flatten().filter(|i| i["module"] == key.as_str()) {
                            run.push(format!("/run/{}/{entry}", inst["name"].as_str().unwrap_or_default()));
                        }
                    }
                    None => run.push(format!("/run/{entry}")),
                }
            }
        }
        json!({ "ref": self.words.join(" "), "run": run, "login": format!("{}login?handle=<handle>", self.prefix()), "oauth": self.oauth.records.keys().map(|s| format!("{}{s}/client-metadata.json", self.oauth.path.as_deref().unwrap_or("/oauth/"))).collect::<Vec<_>>() })
    }

    /// Why a request is refused (403): a Host that isn't loopback, an IP literal or --origin's (DNS
    /// rebinding), an Origin other than the one it reached (a cross-site page), or, for POST /run/*,
    /// a body that isn't application/json (which a cross-site page can't send without a preflight,
    /// and serve answers no preflight). A missing or non-ASCII Host or Origin fails closed. A run acts
    /// as the request's signed-in user, else as --as / --app-password if given.
    fn refused(&self, req: &http::Request<Bytes>) -> Option<&'static str> {
        let header = |name: &str| req.headers().get(name).and_then(|v| v.to_str().ok());
        if req.headers().get("origin").is_some() && header("origin").is_none() {
            return Some("Origin is not ASCII");
        }
        let Some(host) = header("host").map(str::to_ascii_lowercase) else {
            return Some("no Host (or not ASCII)");
        };
        let name = host.rsplit_once(':').filter(|(_, p)| p.bytes().all(|b| b.is_ascii_digit())).map_or(host.as_str(), |(h, _)| h);
        let public = self.oauth.origin.as_deref().and_then(|o| url::Url::parse(o).ok());
        let host_ok = name == "localhost"
            || name.trim_start_matches('[').trim_end_matches(']').parse::<std::net::IpAddr>().is_ok()
            || public.as_ref().and_then(|u| u.host_str()).is_some_and(|h| h.eq_ignore_ascii_case(name));
        if !host_ok {
            return Some("Host is not this server's");
        }
        if let Some(origin) = header("origin")
            && origin != format!("http://{host}")
            && Some(origin.to_string()) != public.as_ref().map(|u| u.origin().ascii_serialization())
        {
            return Some("cross-origin requests are refused");
        }
        let json = header("content-type").is_some_and(|c| c.split(';').next().unwrap_or_default().trim().eq_ignore_ascii_case("application/json"));
        (req.method() == http::Method::POST && req.uri().path().starts_with("/run/") && !json).then_some("POST the input as content-type: application/json")
    }

    fn prefix(&self) -> String {
        self.oauth.path.clone().unwrap_or_else(|| "/oauth/".into())
    }

    /// The OAuth session `did` signed in with, one per user in this process.
    async fn session(&self, did: &str) -> Result<Arc<Session>, String> {
        if let Some(s) = self.sessions.lock().unwrap().get(did) {
            return Ok(s.clone());
        }
        let opened = Session::oauth(did, self.cli.fetch.clone(), self.cli.store()?.clone()).await?;
        Ok(self.sessions.lock().unwrap().entry(did.into()).or_insert(opened).clone())
    }

    /// /oauth/login, /oauth/callback, /oauth/logout, and the confidential client's documents (None: not
    /// one of them): atlibs_host::signin.
    async fn sign_in(&self, req: &http::Request<Bytes>) -> Option<http::Response<String>> {
        let route = req.uri().path().strip_prefix(&self.prefix())?;
        let (client, store) = match (&self.client, self.cli.store()) {
            (Some(c), Ok(s)) => (c.clone(), s.clone()),
            (None, _) => return matches!(route, "login" | "callback" | "logout").then(|| json_reply(503, &json!({ "error": "NoClient", "message": "this server has no OAuth client yet" }))),
            (_, Err(e)) => return Some(json_reply(500, &json!({ "error": "AuthFailed", "message": e }))),
        };
        let routes = SignIn { client, store, net: self.cli.net.clone(), prefix: self.prefix(), name: format!("atlibs serve {}", self.words.join(" ")) };
        let reply = routes.handle(req).await?;
        // a new sign-in, or a sign-out, replaces this process's session of that user
        if let Some(did) = &reply.changed {
            self.sessions.lock().unwrap().remove(did);
        }
        Some(reply.response)
    }

    pub async fn respond(&self, req: http::Request<Bytes>, signal: Signal) -> http::Response<String> {
        if let Some(res) = self.oauth.handle(&req) {
            return res;
        }
        if let Some(res) = self.app.as_ref().and_then(|a| a.handle(&req)) {
            return res;
        }
        // this server's confidential client, public as the per-session documents are (the authorization server reads them)
        if req.method() == http::Method::GET && matches!(req.uri().path().strip_prefix(&self.prefix()), Some("client-metadata.json" | "jwks.json")) && self.client.as_ref().is_some_and(|c| c.key.is_some()) {
            if let Some(res) = self.sign_in(&req).await {
                return res;
            }
        }
        if let Some(why) = self.refused(&req) {
            return json_reply(403, &json!({ "error": "Forbidden", "message": why }));
        }
        if let Some(res) = self.sign_in(&req).await {
            return res;
        }
        let path = req.uri().path().to_string();
        if path == "/" && req.method() == http::Method::GET {
            let html = req.headers().get("accept").and_then(|a| a.to_str().ok()).is_some_and(|a| a.contains("text/html"));
            return if html { reply(200, "text/html; charset=utf-8", PAGE.replace("INDEX", &self.index().to_string().replace('<', "\\u003c"))) } else { json_reply(200, &self.index()) };
        }
        let Some(rest) = path.strip_prefix("/run/") else {
            return json_reply(404, &json!({ "error": "NotFound", "message": format!("{path}: try GET / or POST /run/<entry>") }));
        };
        if req.method() != http::Method::POST {
            return json_reply(405, &json!({ "error": "InvalidRequest", "message": "POST the input JSON" }));
        }
        let body = String::from_utf8_lossy(req.body()).to_string();
        let input = match input_of(Some(&body).filter(|b| !b.trim().is_empty())) {
            Ok(input) => input,
            Err(e) => return json_reply(400, &json!({ "error": "InvalidRequest", "message": e })),
        };
        let loaded;
        let target = match &self.fixed {
            Some(t) => t,
            None => match target(&self.cli.net, &self.words).await {
                Ok((t, _)) => {
                    loaded = t;
                    &loaded
                }
                Err(e) => return json_reply(500, &json!({ "error": "LoadFailed", "message": e })),
            },
        };
        let (instance, entry) = match (target.is_composition(), rest.split_once('/')) {
            (true, Some((i, e))) => (Some(i), e),
            (false, None) => (None, rest),
            _ => return json_reply(404, &json!({ "error": "NotFound", "message": if target.is_composition() { "POST /run/<instance>/<entry>" } else { "POST /run/<entry>" } })),
        };
        if let Some(no) = self.host.no_entry(target, instance, entry) {
            return json_reply(404, &json!({ "error": "NotFound", "message": no }));
        }
        // who this run acts as: the request's signed-in user (their session alone), else the server's own (--as / --app-password)
        let host = match handle_of(&req) {
            None => self.host.clone(),
            Some(handle) => {
                let did = self.cli.store().and_then(|store| oauth_client::did_of_handle(store, handle));
                let session = match did {
                    Ok(Some(did)) => self.session(&did).await,
                    Ok(None) => Err("unknown or expired session: sign in again (GET /oauth/login?handle=...)".into()),
                    Err(e) => Err(e),
                };
                match session {
                    Ok(session) => Host { session: Some(session), ..self.host.clone() },
                    Err(e) => return json_reply(401, &json!({ "error": "AuthRequired", "message": e })),
                }
            }
        };
        match host.run(target, instance, entry, input, signal).await {
            Ok(ran) => {
                if let Some(e) = ran.get("error").and_then(Value::as_str) {
                    eprintln!("{path}: {e}");
                }
                json_reply(200, &ran)
            }
            Err(e) => json_reply(400, &json!({ "error": "InvalidRequest", "message": e })),
        }
    }
}

/// The served ref: target words, the run host, OAuth records and actions (registry::resolve_ref's libs;
/// a <module.wasm> <module.json> pair, its records/<session>.json if any).
pub async fn prepare(cli: Cli, args: &Args) -> Result<Served, String> {
    let (found, rest) = registry::resolve_ref(&cli.net, std::path::Path::new("."), &args.pos).await?;
    if !rest.is_empty() {
        return Err(format!("serve takes one ref; extra: {}\n{}", rest.join(" "), crate::usage("serve")));
    }
    let words: Vec<String> = args.pos.clone();
    let mut records = Map::new();
    let mut actions = Map::new();
    for lib in found.libs() {
        if let Some(c) = &lib.client {
            records.insert(c.rkey.clone(), c.record.clone());
        }
        actions.insert(lib.key.clone(), json!(crate::run::actions(lib)));
    }
    let loaded = target_of(found)?;
    let oauth = ClientMetadataHandler {
        records,
        path: Some(format!("/{}/", args.flag("oauth-path").unwrap_or("oauth/").trim_matches('/'))),
        origin: match args.flag("origin").map(String::from).or_else(|| crate::env("ATLIBS_ORIGIN")) {
            Some(o) => Some(url::Url::parse(&o).map_err(|_| format!("--origin / ATLIBS_ORIGIN {o} is not a URL"))?.origin().ascii_serialization()),
            None => None,
        },
        ..Default::default()
    };
    let mut scope: Vec<String> = vec!["atproto".into()];
    for record in oauth.records.values() {
        for s in scope_list(record["clientMetadata"]["scope"].as_str().unwrap_or_default()) {
            if !scope.iter().any(|x| x == s) {
                scope.push(s.into());
            }
        }
    }
    let scope = args.flag("scope").map(String::from).unwrap_or_else(|| scope.join(" "));
    let client = match oauth.origin.as_deref().filter(|o| !is_loopback(o)) {
        Some(origin) => Some(Client::confidential(origin, oauth.path.as_deref().unwrap_or("/oauth/"), &scope, oauth_client::client_key(cli.store()?).await?)),
        None => None,
    };
    // the key behind did:web:<origin host>: kept in the store under app:<origin> (losing it is a new identity in practice)
    let app = match oauth.origin.as_deref().filter(|o| !is_loopback(o)) {
        Some(origin) if oauth.records.values().any(|r| r.get("app").is_some()) => {
            let key = cli.store()?.didkey(&format!("app:{origin}")).await?;
            AppRepo::new(&oauth.records, origin, key.did.trim_start_matches("did:key:"))?
        }
        _ => None,
    };
    let mut host = Host::new(&cli).await?;
    host.handlers = crate::handlers::parse(&cli, &args.all("handler"), args.has("allow-exec")).await?;
    let fixed = (is_remote(&words[0]) || loaded.is_composition()).then_some(loaded);
    Ok(Served { cli, host, words, fixed, oauth, app, actions, client, scope, sessions: Mutex::new(HashMap::new()) })
}

fn is_loopback(origin: &str) -> bool {
    url::Url::parse(origin).ok().and_then(|u| u.host_str().map(|h| matches!(h, "localhost" | "127.0.0.1" | "[::1]"))).unwrap_or(false)
}

pub async fn main(cli: Cli, argv: &[String], signal: Signal) -> Result<(), String> {
    let args = Args::parse(argv, &["allow-exec"])?;
    args.only(&["port", "host", "origin", "oauth-path", "scope", "handler", "allow-exec"], "serve")?;
    if args.pos.is_empty() {
        return Err(crate::usage("serve"));
    }
    let mut served = prepare(cli, &args).await?;
    let addr: SocketAddr = format!("{}:{}", args.flag("host").unwrap_or("127.0.0.1"), args.flag("port").unwrap_or("8787")).parse().map_err(|e| format!("--host/--port: {e}"))?;
    let listener = tokio::net::TcpListener::bind(addr).await.map_err(|e| format!("{addr}: {e}"))?;
    let bound = listener.local_addr().map_err(|e| e.to_string())?;
    if served.client.is_none() {
        // no public origin: the loopback client, signing in on http://127.0.0.1:<this port>
        let redirect = format!("http://127.0.0.1:{}{}callback", bound.port(), served.prefix());
        served.client = Some(Client::loopback(&redirect, &served.scope));
    }
    let served = Rc::new(served);
    if !bound.ip().is_loopback() {
        eprintln!("atlibs serve: warning: {bound} is not loopback; anyone who can reach it runs entries as this account");
    }
    eprintln!("atlibs serve: http://{bound}/  ({} ; GET /oauth/<session>/client-metadata.json)", served.index()["run"].as_array().map(|r| r.iter().filter_map(Value::as_str).collect::<Vec<_>>().join(", ")).filter(|r| !r.is_empty()).unwrap_or_else(|| "POST /run/<entry>".into()));
    // Runs are not Send (one thread, like the JS event loop): connections are local tasks.
    let local = tokio::task::LocalSet::new();
    local
        .run_until(async move {
            loop {
                let (stream, _) = tokio::select! {
                    accepted = listener.accept() => accepted.map_err(|e| e.to_string())?,
                    _ = signal.aborted() => return Ok(()),
                };
                let (served, signal) = (served.clone(), signal.clone());
                tokio::task::spawn_local(async move {
                    let service = service_fn(move |req: http::Request<Incoming>| {
                        let (served, signal) = (served.clone(), signal.clone());
                        async move {
                            let (parts, body) = req.into_parts();
                            let bytes = body.collect().await.map(|b| b.to_bytes()).unwrap_or_default();
                            let res = served.respond(http::Request::from_parts(parts, bytes), signal).await;
                            Ok::<_, std::convert::Infallible>(res.map(|b| Full::new(Bytes::from(b))))
                        }
                    });
                    let _ = http1::Builder::new().serve_connection(TokioIo::new(stream), service).await;
                });
            }
        })
        .await
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::fake::Pds;
    use crate::registry::tests::tempdir;

    #[tokio::test]
    async fn index_oauth_and_errors_for_a_published_module() {
        let pds = Pds::new();
        pds.publish_module("alice.test", "echo", b"\0asm", Some("echo"));
        let dir = tempdir("serve");
        let served = prepare(pds.cli(&dir), &Args::parse(&["alice.test/echo".into()], &[]).unwrap()).await.unwrap();
        let get = |path: &str| http::Request::get(path).header("host", "127.0.0.1:8787").body(Bytes::new()).unwrap();
        let index = served.respond(get("/"), Signal::new()).await;
        assert_eq!(index.status(), 200);
        assert!(index.body().contains("/oauth/echo/client-metadata.json"));
        let doc: Value = serde_json::from_str(served.respond(get("/oauth/echo/client-metadata.json"), Signal::new()).await.body()).unwrap();
        assert_eq!(doc["client_id"], "http://127.0.0.1:8787/oauth/echo/client-metadata.json");
        let public = |host: &str| http::Request::get("/oauth/echo/client-metadata.json").header("host", host).body(Bytes::new()).unwrap();
        assert_eq!(served.respond(public("evil.example"), Signal::new()).await.status(), 500, "a public Host needs --origin");
        let fixed = prepare(pds.cli(&dir), &Args::parse(&["alice.test/echo".into(), "--origin".into(), "https://pub.example/x".into()], &[]).unwrap()).await.unwrap();
        let doc: Value = serde_json::from_str(fixed.respond(public("evil.example"), Signal::new()).await.body()).unwrap();
        assert_eq!(doc["client_id"], "https://pub.example/oauth/echo/client-metadata.json");
        assert_eq!(served.respond(get("/run/x"), Signal::new()).await.status(), 405);
        assert_eq!(served.respond(get("/nope"), Signal::new()).await.status(), 404);
        let bad = http::Request::post("/run/x").header("host", "127.0.0.1:8787").header("content-type", "application/json").body(Bytes::from("[1]")).unwrap();
        assert_eq!(served.respond(bad, Signal::new()).await.status(), 400);
    }

    #[tokio::test]
    async fn runs_refuse_foreign_origins_hosts_and_non_json_bodies() {
        let pds = Pds::new();
        pds.publish_module("alice.test", "echo", b"\0asm", Some("echo"));
        let dir = tempdir("serve-guard");
        let served = prepare(pds.cli(&dir), &Args::parse(&["alice.test/echo".into(), "--origin".into(), "https://pub.example".into()], &[]).unwrap()).await.unwrap();
        let post = |host: &str, origin: Option<&str>, ct: &str| {
            let mut b = http::Request::post("/run/x").header("host", host).header("content-type", ct);
            if let Some(o) = origin {
                b = b.header("origin", o);
            }
            b.body(Bytes::from("{}")).unwrap()
        };
        let status = |req| async { served.respond(req, Signal::new()).await.status().as_u16() };
        const JSON: &str = "application/json";
        assert_eq!(status(post("127.0.0.1:8787", Some("https://evil.example"), "text/plain")).await, 403);
        assert_eq!(status(post("127.0.0.1:8787", Some("https://evil.example"), JSON)).await, 403, "foreign Origin");
        assert_eq!(status(post("127.0.0.1:8787", None, "text/plain")).await, 403, "a no-cors body");
        assert_eq!(status(post("evil.example:8787", None, JSON)).await, 403, "DNS rebinding");
        assert_eq!(status(post("evil.example", Some("http://evil.example"), JSON)).await, 403);
        assert_eq!(served.respond(http::Request::get("/").header("host", "evil.example").body(Bytes::new()).unwrap(), Signal::new()).await.status(), 403);
        // review 4a L1: no Host (an HTTP/1.0 request), or one or an Origin that isn't ASCII, fails closed
        assert_eq!(status(http::Request::post("/run/x").header("content-type", "text/plain").body(Bytes::from("{}")).unwrap()).await, 403, "no Host");
        let bytes = |b: &[u8]| http::HeaderValue::from_bytes(b).unwrap();
        assert_eq!(status(http::Request::post("/run/x").header("host", bytes(b"127.0.0.1:8787\xff")).header("content-type", JSON).body(Bytes::from("{}")).unwrap()).await, 403, "non-ASCII Host");
        assert_eq!(status(http::Request::post("/run/x").header("host", "127.0.0.1:8787").header("origin", bytes(b"http://127.0.0.1:8787\xff")).header("content-type", JSON).body(Bytes::from("{}")).unwrap()).await, 403, "non-ASCII Origin");
        // same origin (the form page), the bind IP, --origin's host, and curl (no Origin) get through to the run
        for (host, origin) in [("127.0.0.1:8787", Some("http://127.0.0.1:8787")), ("localhost:8787", None), ("[::1]:8787", Some("http://[::1]:8787")), ("192.168.1.5:8787", None), ("pub.example", Some("https://pub.example"))] {
            assert_ne!(status(post(host, origin, "application/json; charset=utf-8")).await, 403, "{host} {origin:?}");
        }
    }
}