mod guide;
mod worker;

use guide::agent_guide;
use worker::{PendingRequest, RequestProgress, Worker};

use std::{
    io::Read,
    ops::{Deref, DerefMut},
    panic::{AssertUnwindSafe, catch_unwind},
    sync::{
        Arc, RwLock,
        mpsc::{Receiver, SyncSender, TrySendError, sync_channel},
    },
    sync::{
        Condvar, Mutex,
        atomic::{AtomicBool, Ordering},
    },
    thread,
};

use anyhow::{Context as _, Result};
use tiny_http::{Header, Method, Request, Response, Server, StatusCode};
use winit::event_loop::EventLoopProxy;

use crate::{
    activity::{Activity, ActivityLog},
    app::AppEvent,
    document::NamespaceState,
    endor_host,
};

pub const API_ADDRESS: &str = "127.0.0.1:4737";
const MAX_CODE_SIZE: u64 = endor_host::MAX_SOURCE_BYTES as u64;
type Notify = Arc<dyn Fn() + Send + Sync>;

#[derive(Clone, Debug)]
pub struct ShutdownOutcome {
    pub summary: String,
    pub error: Option<String>,
}
type ShutdownCompletion = Arc<(Mutex<Option<ShutdownOutcome>>, Condvar)>;
#[derive(Clone)]
pub struct RuntimeControl {
    sender: SyncSender<()>,
    requested: Arc<AtomicBool>,
    completion: ShutdownCompletion,
}
impl RuntimeControl {
    pub fn request_shutdown(&self) -> bool {
        if self.requested.swap(true, Ordering::AcqRel) {
            return true;
        }
        if self.sender.try_send(()).is_ok() {
            true
        } else {
            self.requested.store(false, Ordering::Release);
            false
        }
    }
    pub fn is_exiting(&self) -> bool {
        self.requested.load(Ordering::Acquire)
    }
    pub fn wait_shutdown(&self, timeout: std::time::Duration) -> Option<ShutdownOutcome> {
        let (lock, ready) = &*self.completion;
        let result = lock.lock().ok()?;
        let (result, _) = ready
            .wait_timeout_while(result, timeout, |result| result.is_none())
            .ok()?;
        result.clone()
    }
}
struct Shutdown {
    receiver: Receiver<()>,
    control: RuntimeControl,
    proxy: EventLoopProxy<AppEvent>,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum RequestKind {
    Event,
    Code,
    Expression,
    Repl,
    ResetRepl,
}

struct BuildRequest {
    request: ObservedRequest,
    kind: RequestKind,
    source: String,
    namespace: String,
    submission: u64,
}

pub fn start(
    namespace: Arc<RwLock<NamespaceState>>,
    proxy: EventLoopProxy<AppEvent>,
) -> Result<(Activity, RuntimeControl)> {
    let (sender, receiver) = sync_channel(1);
    let control = RuntimeControl {
        sender,
        requested: Arc::new(AtomicBool::new(false)),
        completion: Arc::new((Mutex::new(None), Condvar::new())),
    };
    let shutdown = Shutdown {
        receiver,
        control: control.clone(),
        proxy: proxy.clone(),
    };
    let server = Server::http(API_ADDRESS)
        .map_err(|error| anyhow::anyhow!(error))
        .with_context(|| format!("bind Tessera API to {API_ADDRESS}"))?;
    let notify: Notify = Arc::new(move || {
        let _ = proxy.send_event(AppEvent::DocumentUpdated);
    });
    let activity = Arc::new(Mutex::new(ActivityLog::default()));
    start_server_observed(
        Arc::new(server),
        namespace,
        notify.clone(),
        activity.clone(),
        notify,
        Some(shutdown),
    )?;
    Ok((activity, control))
}

#[cfg(test)]
fn start_server(
    server: Arc<Server>,
    namespace: Arc<RwLock<NamespaceState>>,
    notify: Notify,
) -> Result<thread::JoinHandle<()>> {
    start_server_observed(
        server,
        namespace,
        notify,
        Arc::new(Mutex::new(ActivityLog::default())),
        Arc::new(|| {}),
        None,
    )
}

fn start_server_observed(
    server: Arc<Server>,
    namespace: Arc<RwLock<NamespaceState>>,
    notify: Notify,
    activity: Activity,
    observer_notify: Notify,
    shutdown: Option<Shutdown>,
) -> Result<thread::JoinHandle<()>> {
    let (sender, receiver) = sync_channel::<BuildRequest>(8);
    let worker_namespace = namespace.clone();
    let worker_notify = notify.clone();
    let exiting = shutdown
        .as_ref()
        .map(|shutdown| shutdown.control.requested.clone());
    thread::Builder::new()
        .name("tessera-endor-worker".into())
        .spawn(move || {
            let mut worker = Worker::default();
            let mut pending: Option<(BuildRequest, PendingRequest)> = None;
            loop {
                if let Some(shutdown) = &shutdown
                    && shutdown.receiver.try_recv().is_ok()
                {
                    if let Some((job, input)) = pending.take() {
                        drop(input);
                        let _ = job
                            .request
                            .respond(error_response(StatusCode(503), "application is exiting"));
                    }
                    let outcome = worker.shutdown_runtime();
                    while let Ok(job) = receiver.try_recv() {
                        let _ = job
                            .request
                            .respond(error_response(StatusCode(503), "application is exiting"));
                    }
                    if let Ok(mut current) = worker_namespace.write() {
                        current.runtime_session = 0;
                        current.window_commands.clear();
                        current.document.frame.session = 0;
                        current.document.frame.focus = None;
                        for view in current.document.windows.values_mut() {
                            view.frame.session = 0;
                            view.frame.focus = None;
                        }
                    }
                    eprintln!("{}", outcome.summary);
                    let (lock, ready) = &*shutdown.control.completion;
                    if let Ok(mut result) = lock.lock() {
                        *result = Some(outcome.clone());
                        ready.notify_all();
                    }
                    let _ = shutdown.proxy.send_event(AppEvent::ShutdownComplete {
                        error: outcome.error,
                    });
                    break;
                }
                if let Some((job, input)) = pending.take() {
                    match worker.resume_request(&job, input, &worker_namespace, &worker_notify) {
                        RequestProgress::Pending(input) => {
                            pending = Some((job, input));
                            // A soft VM quantum is a control boundary, not a
                            // minimum-duration time slice. Yield to other ready
                            // threads without charging a timer delay per quantum.
                            thread::yield_now();
                        }
                        RequestProgress::Complete(response) => {
                            let _ = job.request.respond(response);
                        }
                    }
                    continue;
                }
                worker.refresh_viewport(&worker_namespace, &worker_notify);
                let wait = if worker.repl.as_ref().is_some_and(|(_, runtime)| {
                    runtime.has_pending_refresh() || runtime.has_cooperative_tasks()
                }) {
                    std::time::Duration::ZERO
                } else {
                    std::time::Duration::from_millis(100)
                };
                match receiver.recv_timeout(wait) {
                    Ok(job) => {
                        match worker.begin_request(&job, &worker_namespace, &worker_notify) {
                            RequestProgress::Pending(input) => {
                                if matches!(&input, PendingRequest::BeforeInput) {
                                    eprintln!(
                                        "Tessera worker: submission {} waits for active refresh",
                                        job.submission
                                    );
                                }
                                pending = Some((job, input));
                            }
                            RequestProgress::Complete(response) => {
                                let _ = job.request.respond(response);
                            }
                        }
                    }
                    Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
                    Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
                        if worker.repl.as_ref().is_some_and(|(epoch, runtime)| {
                            worker_namespace.read().is_ok_and(|current| {
                                current.submission != *epoch
                                    && current.runtime_session != runtime.id
                            })
                        }) {
                            worker.repl = None;
                        }
                        if wait.is_zero() {
                            thread::yield_now();
                        }
                    }
                }
            }
        })
        .context("start Aster execution worker")?;
    thread::Builder::new()
        .name("tessera-agent-api".to_owned())
        .spawn(move || {
            for request in server.incoming_requests() {
                if exiting
                    .as_ref()
                    .is_some_and(|flag| flag.load(Ordering::Acquire))
                    && !matches!(request.method(), Method::Get | Method::Options)
                {
                    let _ =
                        request.respond(error_response(StatusCode(503), "application is exiting"));
                    continue;
                }
                handle(
                    request,
                    &namespace,
                    &notify,
                    &sender,
                    &activity,
                    &observer_notify,
                );
            }
        })
        .context("start Tessera agent API thread")
}

// Capture only allowlisted headers: credentials and cookies never enter the inspector.
struct ObservedRequest {
    inner: Request,
    activity: Activity,
    id: u64,
    notify: Notify,
}
impl ObservedRequest {
    fn new(inner: Request, activity: Activity, notify: Notify) -> Self {
        let headers = inner
            .headers()
            .iter()
            .filter(|h| h.field.equiv("Content-Type") || h.field.equiv("Content-Length"))
            .map(ToString::to_string)
            .collect::<Vec<_>>()
            .join("\n");
        let origin = if inner
            .headers()
            .iter()
            .any(|h| h.field.equiv("X-Tessera-Origin") && h.value.as_str() == "console")
        {
            "Console"
        } else {
            "HTTP"
        }
        .to_owned();
        let id = activity.lock().unwrap().begin(
            inner.method().to_string(),
            inner.url().to_owned(),
            origin,
            headers,
        );
        notify();
        Self {
            inner,
            activity,
            id,
            notify,
        }
    }
    fn record_body(&self, body: &str) {
        self.activity.lock().unwrap().body(self.id, body);
        (self.notify)();
    }
    fn respond(self, response: Response<std::io::Cursor<Vec<u8>>>) -> std::io::Result<()> {
        let status = response.status_code();
        let headers = response.headers().to_vec();
        let length = response.data_length();
        let reader = response.into_reader();
        self.activity.lock().unwrap().finish(
            self.id,
            status.0,
            &String::from_utf8_lossy(reader.get_ref()),
            headers
                .iter()
                .map(ToString::to_string)
                .collect::<Vec<_>>()
                .join("\n"),
        );
        (self.notify)();
        self.inner
            .respond(Response::new(status, headers, reader, length, None))
    }
}
impl Deref for ObservedRequest {
    type Target = Request;
    fn deref(&self) -> &Request {
        &self.inner
    }
}
impl DerefMut for ObservedRequest {
    fn deref_mut(&mut self) -> &mut Request {
        &mut self.inner
    }
}

fn handle(
    request: Request,
    namespace: &Arc<RwLock<NamespaceState>>,
    notify: &Notify,
    sender: &SyncSender<BuildRequest>,
    activity: &Activity,
    observer_notify: &Notify,
) {
    let request = ObservedRequest::new(request, activity.clone(), observer_notify.clone());
    let method = request.method().clone();
    let path = request
        .url()
        .split('?')
        .next()
        .unwrap_or(request.url())
        .to_owned();

    let evaluation = namespace_from_route(&path, "/evaluate");
    let repl = namespace_from_route(&path, "/repl");
    let event = namespace_from_route(&path, "/events");
    let route = if method == Method::Put {
        namespace_from_path(&path).map(|name| (name, RequestKind::Code))
    } else if method == Method::Post {
        event
            .map(|name| (name, RequestKind::Event))
            .or_else(|| repl.map(|name| (name, RequestKind::Repl)))
            .or(evaluation.map(|name| (name, RequestKind::Expression)))
    } else if method == Method::Delete {
        repl.map(|name| (name, RequestKind::ResetRepl))
    } else {
        None
    };
    if let Some((addressed, kind)) = route {
        submit(request, addressed, namespace, sender, kind);
        return;
    }
    let response = if method == Method::Get && path == "/agent" {
        match namespace.read() {
            Ok(namespace) => markdown(StatusCode(200), agent_guide(&namespace)),
            Err(_) => error_response(StatusCode(500), "namespace lock is poisoned"),
        }
    } else if let Some(addressed_namespace) = namespace_from_path(&path).map(str::to_owned) {
        code_request(method, &addressed_namespace, namespace, notify)
    } else if method == Method::Options {
        text(StatusCode(204), "", "text/plain")
    } else {
        error_response(StatusCode(404), "route not found; start with GET /agent")
    };

    let _ = request.respond(response);
}

fn code_request(
    method: Method,
    addressed_namespace: &str,
    namespace: &Arc<RwLock<NamespaceState>>,
    notify: &Notify,
) -> Response<std::io::Cursor<Vec<u8>>> {
    let Ok(current) = namespace.read() else {
        return error_response(StatusCode(500), "namespace lock is poisoned");
    };
    if addressed_namespace != current.name {
        return error_response(
            StatusCode(404),
            format!(
                "namespace `{addressed_namespace}` is not active; current namespace is `{}`",
                current.name
            ),
        );
    }

    if method == Method::Get {
        let mut response = text(
            StatusCode(200),
            &current.document.source,
            "text/plain; charset=utf-8",
        );
        add_header(
            &mut response,
            "X-Tessera-Revision",
            &current.revision.to_string(),
        );
        add_header(
            &mut response,
            "X-Tessera-Session",
            &current.document.frame.session.to_string(),
        );
        add_header(
            &mut response,
            "X-Tessera-Window",
            &current.document.frame.window.to_string(),
        );
        add_header(
            &mut response,
            "X-Tessera-Layout-Version",
            &current.document.frame.layout_version.to_string(),
        );
        if let Some(viewport) = current.viewport {
            add_header(
                &mut response,
                "X-Tessera-Viewport-Width",
                &viewport.width.to_string(),
            );
            add_header(
                &mut response,
                "X-Tessera-Viewport-Height",
                &viewport.height.to_string(),
            );
        }
        add_header(
            &mut response,
            "X-Tessera-Source-Kind",
            if current
                .document
                .root
                .as_ref()
                .is_some_and(|root| root.kind == "repl-session")
            {
                "repl-transcript"
            } else {
                "source-snapshot"
            },
        );
        return response;
    }
    drop(current);

    match method {
        Method::Delete => match namespace.write() {
            Ok(mut current) => {
                let revision = match current.clear() {
                    Ok(revision) => revision,
                    Err(error) => return error_response(StatusCode(500), error.to_string()),
                };
                drop(current);
                notify();
                json(
                    StatusCode(200),
                    serde_json::json!({
                        "ok": true,
                        "namespace": addressed_namespace,
                        "revision": revision
                    }),
                )
            }
            Err(_) => error_response(StatusCode(500), "namespace lock is poisoned"),
        },
        Method::Options => text(StatusCode(204), "", "text/plain"),
        _ => error_response(StatusCode(405), "code supports GET, PUT, and DELETE"),
    }
}

fn submit(
    mut request: ObservedRequest,
    addressed: &str,
    namespace: &Arc<RwLock<NamespaceState>>,
    sender: &SyncSender<BuildRequest>,
    kind: RequestKind,
) {
    let source = match read_code(&mut request) {
        Ok(source) => source,
        Err(error) => {
            let _ = request.respond(error_response(StatusCode(400), error.to_string()));
            return;
        }
    };
    request.record_body(&source);
    let Ok(mut current) = namespace.write() else {
        let _ = request.respond(error_response(
            StatusCode(500),
            "namespace lock is poisoned",
        ));
        return;
    };
    if current.name != addressed {
        let _ = request.respond(error_response(StatusCode(404), "namespace is not active"));
        return;
    }
    let submission = match if kind == RequestKind::Code {
        current.next_submission()
    } else {
        Ok(current.submission)
    } {
        Ok(id) => id,
        Err(error) => {
            let _ = request.respond(error_response(StatusCode(500), error.to_string()));
            return;
        }
    };
    let job = BuildRequest {
        request,
        kind,
        source,
        namespace: addressed.into(),
        submission,
    };
    match sender.try_send(job) {
        // A rejected queue entry never invalidates accepted work. The worker
        // uses this same lock when committing, so publication is atomic.
        Ok(()) => {
            if kind == RequestKind::Code {
                current.submission = submission;
            }
        }
        Err(TrySendError::Full(job)) => {
            let _ = job
                .request
                .respond(error_response(StatusCode(503), "build queue is full"));
        }
        Err(TrySendError::Disconnected(job)) => {
            let _ = job.request.respond(error_response(
                StatusCode(503),
                "execution worker is unavailable",
            ));
        }
    }
}

fn superseded(submission: u64) -> Response<std::io::Cursor<Vec<u8>>> {
    json(
        StatusCode(409),
        serde_json::json!({ "ok": false, "submission": submission,
        "error": "submission superseded by a newer PUT or DELETE" }),
    )
}

fn namespace_from_route<'a>(path: &'a str, suffix: &str) -> Option<&'a str> {
    path.strip_prefix("/v1/namespaces/")?
        .strip_suffix(suffix)
        .filter(|namespace| !namespace.is_empty() && !namespace.contains('/'))
}
fn namespace_from_path(path: &str) -> Option<&str> {
    namespace_from_route(path, "/code")
}

fn read_code(request: &mut Request) -> Result<String> {
    if request
        .body_length()
        .is_some_and(|length| length as u64 > MAX_CODE_SIZE)
    {
        anyhow::bail!("code exceeds the 1 MiB limit");
    }
    let mut code = String::new();
    request
        .as_reader()
        .take(MAX_CODE_SIZE + 1)
        .read_to_string(&mut code)
        .context("read code")?;
    if code.len() as u64 > MAX_CODE_SIZE {
        anyhow::bail!("code exceeds the 1 MiB limit");
    }
    Ok(code)
}

fn compile_error(error: anyhow::Error) -> Response<std::io::Cursor<Vec<u8>>> {
    let mut diagnostic =
        serde_json::json!({ "severity": "error", "message": format!("{error:#}") });
    if let Some(source) = error.downcast_ref::<frontend::SourceDiagnostic>() {
        diagnostic["line"] = source.location.line.into();
        diagnostic["column"] = source.location.column.into();
        diagnostic["start_byte"] = source.diagnostic.span.start.into();
        diagnostic["end_byte"] = source.diagnostic.span.end.into();
    }
    json(
        StatusCode(422),
        serde_json::json!({
            "ok": false,
            "diagnostics": [diagnostic]
        }),
    )
}

fn error_response(
    status: StatusCode,
    message: impl Into<String>,
) -> Response<std::io::Cursor<Vec<u8>>> {
    json(status, serde_json::json!({ "error": message.into() }))
}

fn json(status: StatusCode, value: serde_json::Value) -> Response<std::io::Cursor<Vec<u8>>> {
    text(
        status,
        &value.to_string(),
        "application/json; charset=utf-8",
    )
}

fn markdown(status: StatusCode, value: String) -> Response<std::io::Cursor<Vec<u8>>> {
    text(status, &value, "text/markdown; charset=utf-8")
}

fn text(status: StatusCode, value: &str, content_type: &str) -> Response<std::io::Cursor<Vec<u8>>> {
    let mut response = Response::from_string(value).with_status_code(status);
    add_header(&mut response, "Content-Type", content_type);
    add_header(
        &mut response,
        "Access-Control-Allow-Origin",
        "http://127.0.0.1",
    );
    add_header(
        &mut response,
        "Access-Control-Allow-Methods",
        "GET, PUT, POST, DELETE, OPTIONS",
    );
    add_header(
        &mut response,
        "Access-Control-Allow-Headers",
        "Content-Type, X-Tessera-Origin",
    );
    response
}

fn add_header(response: &mut Response<std::io::Cursor<Vec<u8>>>, name: &str, value: &str) {
    if let Ok(header) = Header::from_bytes(name, value) {
        response.add_header(header);
    }
}

#[cfg(test)]
mod tests;