//! This device's node in one share, and the guest API it speaks.

use crate::Observer;
use crate::storage;
use crate::transports::{
    ConnectionCheck, InviteRefusedError, Stream, TerminalState, Transport, TransportError,
};
use anyhow::{Context, Result};
use http_body_util::{BodyExt, Full};
use hyper::StatusCode;
use hyper::client::conn::http1 as http1_client;
use hyper_util::rt::TokioIo;
use std::sync::{Arc, Mutex};
use tokio::time::Instant;
use tracing::{debug, warn};
use wire::{ConfigHash, ShareInfo};
use wispers_access_wire as wire;

/// This device's guest node in one share: its transport to the host node,
/// its row in the store, and the guest API calls it makes. Restored on the
/// share's first use, by `Client::restore_guest_node`.
pub struct GuestNode {
    /// The `<share>` label in `<app>.<share>.wa.localhost`.
    label: String,
    row: storage::Row,
    transport: Box<dyn Transport>,
    observer: Arc<dyn Observer>,
    /// Set once the transport reports a terminal failure mid-session. From
    /// then on the share answers without dialing.
    dead: Mutex<Option<TerminalState>>,
    /// Held while `check_connection` probes the cached connection, so a
    /// stream asked for meanwhile waits for the verdict rather than going
    /// out on a connection that may be dead.
    probe_gate: tokio::sync::Mutex<()>,
    /// Time when the last QUIC stream was opened, used to determine whether or
    /// not to reestablish the connection if it drops.
    last_used: Mutex<Option<Instant>>,
}

impl GuestNode {
    pub fn new(
        label: String,
        row: storage::Row,
        transport: Box<dyn Transport>,
        observer: Arc<dyn Observer>,
    ) -> Self {
        Self {
            label,
            row,
            transport,
            observer,
            dead: Mutex::new(None),
            probe_gate: tokio::sync::Mutex::new(()),
            last_used: Mutex::new(None),
        }
    }

    /// Check connection health and drop/redial the connection if necessary.
    /// Dead connections get redialed if they had been used recently (within
    /// `RECENT_USE`). Otherwise, we just drop them. The probe delay is capped
    /// by `PROBE_DEADLINE`.
    pub async fn check_connection(&self) {
        let _gate = self.probe_gate.lock().await;
        match self.transport.check_connection(PROBE_DEADLINE).await {
            Ok(ConnectionCheck::Alive) => {
                debug!(share = self.label, "cached connection answered");
                return;
            }
            Ok(ConnectionCheck::NoConnection) => {}
            Ok(ConnectionCheck::Dead) => {
                warn!(share = self.label, "cached connection is dead; dropped");
            }
            Err(TransportError::Terminal(state)) => {
                self.record_terminal_error(state);
                return;
            }
            Err(TransportError::Transient(e)) => {
                warn!(
                    share = self.label,
                    error = format!("{e:#}"),
                    "could not check the connection"
                );
            }
        }
        if self.recently_used() {
            self.reconnect().await;
        }
    }

    fn recently_used(&self) -> bool {
        self.last_used
            .lock()
            .expect("unpoisoned")
            .is_some_and(|at| at.elapsed() < RECENT_USE)
    }

    /// Dials the host node now, in anticipation of future requests, with the
    /// aim to cut down on user-visible latency. This ignores errors — followup
    /// requests will deal with those.
    async fn reconnect(&self) {
        match self.open_stream_ungated().await {
            Ok(_) => debug!(share = self.label, "reconnected ahead of the next request"),
            Err(TransportError::Terminal(_)) => {}
            Err(TransportError::Transient(e)) => warn!(
                share = self.label,
                error = format!("{e:#}"),
                "could not reconnect ahead of the next request"
            ),
        }
    }

    /// Forget the cached connection.
    pub fn drop_connection(&self) {
        self.transport.drop_connection();
    }

    /// Shut the transport down. Used when leaving a share.
    pub async fn close(&self) {
        self.transport.close().await;
    }

    pub fn label(&self) -> &str {
        &self.label
    }

    /// Opens a REVERSE_PROXY stream for one app, ready for the HTTP request.
    pub async fn open_reverse_proxy_stream(&self, app_id: &str) -> Result<Stream, TransportError> {
        let mut stream = self.open_stream().await?;
        wire::open_reverse_proxy_stream(&mut stream, app_id)
            .await
            .map_err(|e| TransportError::Transient(e.into()))?;
        Ok(stream)
    }

    /// Opens a FORWARD_PROXY stream to `host:port`.
    pub async fn open_forward_proxy_stream(
        &self,
        host: &str,
        port: u16,
    ) -> Result<Stream, TransportError> {
        let mut stream = self.open_stream().await?;
        wire::open_forward_proxy_stream(&mut stream, host, port)
            .await
            .map_err(|e| TransportError::Transient(e.into()))?;
        Ok(stream)
    }

    /// Asks the host node for the share if it changed since the stored copy,
    /// and stores the answer. `None` = unchanged. This may reply with an error
    /// indicating that the guest node's status has changed, e.g. it got revoked
    /// or its SDK version is too old.
    pub async fn refresh(&self) -> Result<Option<ShareInfo>, RefreshError> {
        let stream = self.open_stream().await.map_err(|e| match e {
            TransportError::Terminal(_) => RefreshError::Terminal,
            TransportError::Transient(e) => RefreshError::Transient(e),
        })?;
        let known = self.row.read_share_config_hash()?;
        let info = match fetch_info(stream, known).await {
            Ok(info) => info,
            Err(e) if e.is::<ClientTooOldError>() => {
                if self.row.write_outdated(true)? {
                    warn!(
                        share = self.label,
                        "the host node needs a newer version of this app"
                    );
                    self.observer.on_share_changed(self.row.read_share()?);
                }
                return Err(RefreshError::Outdated);
            }
            Err(e) => return Err(RefreshError::Transient(e)),
        };
        let previously_outdated = self.row.write_outdated(false)?;
        if let Some(info) = &info {
            self.row.write_share_info(info)?;
        }
        if previously_outdated || info.is_some() {
            self.observer.on_share_changed(self.row.read_share()?);
        }
        Ok(info)
    }

    /// Open a QUIC stream to the host node on the share's transport, opening a
    /// new QUIC connection if necessary. This will wait for in-flight probes to
    /// finish.
    async fn open_stream(&self) -> Result<Stream, TransportError> {
        drop(self.probe_gate.lock().await);
        self.open_stream_ungated().await
    }

    /// Open a QUIC stream to the host node like `open_stream`, except without
    /// blocking on in-flight probes. Any terminal errors (e.g. if this guest
    /// node has been revoked) get recorded, to prevent futile retries in the
    /// future.
    async fn open_stream_ungated(&self) -> Result<Stream, TransportError> {
        if let Some(state) = *self.dead.lock().expect("unpoisoned") {
            return Err(TransportError::Terminal(state));
        }
        match self.transport.open_stream().await {
            Err(TransportError::Terminal(state)) => {
                self.record_terminal_error(state);
                Err(TransportError::Terminal(state))
            }
            Ok(stream) => {
                *self.last_used.lock().expect("unpoisoned") = Some(Instant::now());
                Ok(stream)
            }
            other => other,
        }
    }

    /// Record that we've received a terminal error from the host node,
    /// preventing retries in the future.
    fn record_terminal_error(&self, state: TerminalState) {
        warn!(
            share = self.label,
            "share is no longer available: {}",
            state.describe()
        );
        *self.dead.lock().expect("unpoisoned") = Some(state);
        match self
            .row
            .write_terminal_state(state.as_str())
            .and_then(|()| self.row.read_share())
        {
            Ok(share) => self.observer.on_share_changed(share),
            Err(e) => warn!(
                share = self.label,
                error = format!("{e:#}"),
                "could not record the share's state"
            ),
        }
    }
}

/// How recently a share must have been used for `check_connection` to
/// redial it unasked: long enough for a trip to another app and back, short
/// enough that this morning's share is not dialled at noon.
const RECENT_USE: std::time::Duration = std::time::Duration::from_secs(10 * 60);

/// How long a checked connection gets to answer. A live one answers in one
/// round trip, well under a second even on cellular; this only ever waits
/// on a dead one whose death the transport has not noticed yet.
const PROBE_DEADLINE: std::time::Duration = std::time::Duration::from_secs(2);

pub enum RefreshError {
    /// The host node has turned this device away for good. The store has
    /// the state.
    Terminal,
    /// This guest node's SDK is too old to talk to the host node.
    Outdated,
    Transient(anyhow::Error),
}

impl From<anyhow::Error> for RefreshError {
    fn from(e: anyhow::Error) -> Self {
        RefreshError::Transient(e)
    }
}

/// The host node answered 426 Upgrade Required: it serves no SDK this old.
/// Any guest API call can get it, so it is one error, found by downcast.
#[derive(Debug, thiserror::Error)]
#[error("this app is too old for the share's host; update Wispers Access and try again")]
pub struct ClientTooOldError;

/// `GET /v1/share` over a fresh stream. `None` when the host node says the
/// share is unchanged since `known`. Also what `join` uses, before there is a
/// share to hang it on.
pub async fn fetch_info(stream: Stream, known: Option<ConfigHash>) -> Result<Option<ShareInfo>> {
    let mut req = hyper::Request::builder()
        .method(hyper::Method::GET)
        .uri(wire::SHARE_PATH);
    if let Some(known) = known {
        req = req.header(hyper::header::IF_NONE_MATCH, known.etag());
    }
    let req = req
        .body(Full::new(bytes::Bytes::new()))
        .expect("static request is valid");
    let resp = guest_api_request(stream, req)
        .await
        .context("GET /v1/share (is waserver up to date?)")?;
    match resp.status() {
        StatusCode::NOT_MODIFIED => Ok(None),
        StatusCode::OK => Ok(Some(read_share(resp).await?)),
        StatusCode::UPGRADE_REQUIRED => Err(ClientTooOldError.into()),
        status => anyhow::bail!("the host node answered {} to GET /v1/share", status),
    }
}

/// `POST /v1/activation` over a fresh stream on a connection whose key the host
/// node does not know yet: binds that key to the invite and returns the share.
/// A refusal carries the contract's reason.
pub async fn activate(stream: Stream, secret: &wire::InviteSecret) -> Result<ShareInfo> {
    let body = serde_json::to_vec(&wire::Activation { secret: *secret })?;
    let req = hyper::Request::builder()
        .method(hyper::Method::POST)
        .uri(wire::ACTIVATION_PATH)
        .header(hyper::header::CONTENT_TYPE, "application/json")
        .body(Full::new(bytes::Bytes::from(body)))
        .expect("static request is valid");
    let resp = guest_api_request(stream, req)
        .await
        .context("POST /v1/activation")?;
    if resp.status() == StatusCode::OK {
        return read_share(resp).await;
    }
    if resp.status() == StatusCode::UPGRADE_REQUIRED {
        return Err(ClientTooOldError.into());
    }
    let status = resp.status();
    let body = resp.into_body().collect().await?.to_bytes();
    match serde_json::from_slice::<wire::ApiError>(&body) {
        Ok(err) => match wire::ActivationError::parse(&err.error).and_then(refusal) {
            Some(refused) => Err(refused.into()),
            None => anyhow::bail!("activation refused: {}", err.error),
        },
        Err(_) => anyhow::bail!("the host node answered {} to the activation", status),
    }
}

/// What to tell the user about an invite the host node refused. `None` for
/// refusals about this device's key rather than the invite, which a join's
/// fresh key never gets.
fn refusal(why: wire::ActivationError) -> Option<InviteRefusedError> {
    use wire::ActivationError::*;
    match why {
        InviteConsumed => Some(InviteRefusedError(
            "This invite has been used already. Ask for a new one.",
        )),
        InviteExpired => Some(InviteRefusedError(
            "This invite has expired. Ask for a new one.",
        )),
        InviteUnknown => Some(InviteRefusedError(
            "This invite is no longer valid. Ask for a new one.",
        )),
        Malformed | Revoked | AlreadyGuest => None,
    }
}

/// `DELETE /v1/guest` over a fresh stream: this device leaves the share.
pub async fn leave(stream: Stream) -> Result<()> {
    let req = hyper::Request::builder()
        .method(hyper::Method::DELETE)
        .uri(wire::GUEST_PATH)
        .body(Full::new(bytes::Bytes::new()))
        .expect("static request is valid");
    let resp = guest_api_request(stream, req)
        .await
        .context("DELETE /v1/guest")?;
    match resp.status() {
        StatusCode::NO_CONTENT => Ok(()),
        StatusCode::UPGRADE_REQUIRED => Err(ClientTooOldError.into()),
        status => anyhow::bail!("the host node answered {} to DELETE /v1/guest", status),
    }
}

/// One request to the guest API, on a CTRL stream that carries nothing else.
async fn guest_api_request(
    mut stream: Stream,
    mut req: hyper::Request<Full<bytes::Bytes>>,
) -> Result<hyper::Response<hyper::body::Incoming>> {
    wire::open_ctrl_stream(&mut stream).await?;
    let (mut sender, conn) = http1_client::handshake(TokioIo::new(stream))
        .await
        .context("guest API handshake (is waserver up to date?)")?;
    tokio::spawn(async move {
        let _ = conn.await;
    });
    let headers = req.headers_mut();
    headers.insert(hyper::header::HOST, "waserver".parse().expect("valid"));
    headers.insert(hyper::header::CONNECTION, "close".parse().expect("valid"));
    headers.insert(
        hyper::header::USER_AGENT,
        wire::client_user_agent(env!("CARGO_PKG_VERSION"))
            .parse()
            .expect("a semver is a valid header value"),
    );
    Ok(sender.send_request(req).await?)
}

async fn read_share(resp: hyper::Response<hyper::body::Incoming>) -> Result<ShareInfo> {
    let body = resp
        .into_body()
        .collect()
        .await
        .context("reading the share")?
        .to_bytes();
    serde_json::from_slice(&body).context("parsing the share")
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::NoObserver;
    use crate::transports::BoxFuture;
    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};

    /// A transport whose host node never answers: streams open, then hang,
    /// and a check finds the connection dead.
    struct Silent {
        connected: Arc<AtomicBool>,
        dials: Arc<AtomicUsize>,
        /// The far ends, kept so the streams stay open rather than see EOF.
        far_ends: Mutex<Vec<tokio::io::DuplexStream>>,
    }

    impl Transport for Silent {
        fn open_stream(&self) -> BoxFuture<'_, Result<Stream, TransportError>> {
            Box::pin(async {
                if !self.connected.swap(true, Ordering::SeqCst) {
                    self.dials.fetch_add(1, Ordering::SeqCst);
                }
                let (near, far) = tokio::io::duplex(4096);
                self.far_ends.lock().expect("unpoisoned").push(far);
                Ok(Box::new(near) as Stream)
            })
        }

        fn check_connection(
            &self,
            _deadline: std::time::Duration,
        ) -> BoxFuture<'_, Result<ConnectionCheck, TransportError>> {
            Box::pin(async {
                // Nobody ever answers: a cached connection is found dead.
                if self.connected.swap(false, Ordering::SeqCst) {
                    Ok(ConnectionCheck::Dead)
                } else {
                    Ok(ConnectionCheck::NoConnection)
                }
            })
        }

        fn drop_connection(&self) {
            self.connected.store(false, Ordering::SeqCst);
        }
    }

    /// A transport whose host node answers every guest API request with
    /// `status` and the JSON `body`, after noting the version the request
    /// named.
    struct Answering {
        status: &'static str,
        body: &'static str,
        user_agents: Arc<Mutex<Vec<String>>>,
    }

    impl Transport for Answering {
        fn open_stream(&self) -> BoxFuture<'_, Result<Stream, TransportError>> {
            Box::pin(async {
                let (near, far) = tokio::io::duplex(4096);
                let (status, body) = (self.status, self.body);
                let user_agents = self.user_agents.clone();
                tokio::spawn(async move {
                    use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
                    let mut far = BufReader::new(far);
                    // The CTRL stream's type byte, then HTTP/1.1 headers.
                    let mut first = [0u8; 1];
                    tokio::io::AsyncReadExt::read_exact(&mut far, &mut first)
                        .await
                        .unwrap();
                    let mut line = String::new();
                    loop {
                        line.clear();
                        far.read_line(&mut line).await.unwrap();
                        if let Some(ua) = line.strip_prefix("user-agent: ") {
                            user_agents.lock().unwrap().push(ua.trim().to_owned());
                        }
                        if line == "\r\n" {
                            break;
                        }
                    }
                    let response = format!(
                        "HTTP/1.1 {status}\r\ncontent-type: application/json\r\n\
                         content-length: {}\r\nconnection: close\r\n\r\n{body}",
                        body.len()
                    );
                    far.get_mut().write_all(response.as_bytes()).await.unwrap();
                    far.get_mut().shutdown().await.unwrap();
                });
                Ok(Box::new(near) as Stream)
            })
        }

        fn check_connection(
            &self,
            _deadline: std::time::Duration,
        ) -> BoxFuture<'_, Result<ConnectionCheck, TransportError>> {
            Box::pin(async { Ok(ConnectionCheck::NoConnection) })
        }

        fn drop_connection(&self) {}
    }

    #[tokio::test]
    async fn a_426_marks_the_share_outdated_until_the_host_node_serves_again() {
        let db = storage::DB::in_memory();
        let row = db.new_row().unwrap();
        row.write_deduped_hostname("rt").unwrap();
        row.mark_complete().unwrap();
        let user_agents = Arc::new(Mutex::new(Vec::new()));
        // A node per answer; the row outlives them in the store.
        let stored_row = || db.find_row("rt").unwrap().expect("the row");
        let node = |status| {
            GuestNode::new(
                "rt".into(),
                stored_row(),
                Box::new(Answering {
                    status,
                    body: r#"{"error":"client-too-old"}"#,
                    user_agents: user_agents.clone(),
                }),
                Arc::new(NoObserver),
            )
        };

        assert!(matches!(
            node("426 Upgrade Required").refresh().await,
            Err(RefreshError::Outdated)
        ));
        assert!(stored_row().read_outdated().unwrap());
        assert_eq!(
            user_agents.lock().unwrap().as_slice(),
            [wire::client_user_agent(env!("CARGO_PKG_VERSION"))]
        );

        // Updated and served again: back to live, without a share body.
        assert!(matches!(node("304 Not Modified").refresh().await, Ok(None)));
        assert!(!stored_row().read_outdated().unwrap());
    }

    #[tokio::test]
    async fn an_invite_the_host_node_refuses_says_why() {
        let activate = |status, body| async move {
            let host = Answering {
                status,
                body,
                user_agents: Default::default(),
            };
            let stream = host.open_stream().await.ok().expect("a stream");
            activate(stream, &wire::InviteSecret([7; 16])).await
        };

        let used = activate("409 Conflict", r#"{"error":"invite-consumed"}"#)
            .await
            .expect_err("a used invite fails");
        assert_eq!(
            used.downcast_ref::<InviteRefusedError>()
                .map(ToString::to_string),
            Some("This invite has been used already. Ask for a new one.".into())
        );
        // About the key, not the invite, so no sentence for the user.
        let revoked = activate("403 Forbidden", r#"{"error":"revoked"}"#)
            .await
            .expect_err("a revoked key fails");
        assert!(!revoked.is::<InviteRefusedError>());
    }

    #[tokio::test(start_paused = true)]
    async fn a_connection_that_stays_silent_is_dropped_by_the_check() {
        let db = storage::DB::in_memory();
        let row = db.new_row().unwrap();
        row.write_deduped_hostname("rt").unwrap();
        row.mark_complete().unwrap();
        let connected = Arc::new(AtomicBool::new(false));
        let dials = Arc::new(AtomicUsize::new(0));
        let node = GuestNode::new(
            "rt".into(),
            row,
            Box::new(Silent {
                connected: connected.clone(),
                dials: dials.clone(),
                far_ends: Mutex::new(Vec::new()),
            }),
            Arc::new(NoObserver),
        );

        // Never in use: nothing to probe, no dial.
        node.check_connection().await;
        assert_eq!(dials.load(Ordering::SeqCst), 0);

        // A cached connection that never answers is dropped by the check
        // and, since the share was in use, redialled at once.
        drop(node.open_stream().await.ok().expect("a stream"));
        assert_eq!(dials.load(Ordering::SeqCst), 1);
        node.check_connection().await;
        assert_eq!(dials.load(Ordering::SeqCst), 2);
        assert!(connected.load(Ordering::SeqCst));

        // Dropped by the app (a network change) and checked on resume:
        // redialled without a probe.
        node.drop_connection();
        node.check_connection().await;
        assert_eq!(dials.load(Ordering::SeqCst), 3);

        // Long unused: left for the next request to dial.
        node.drop_connection();
        tokio::time::advance(RECENT_USE + std::time::Duration::from_secs(1)).await;
        node.check_connection().await;
        assert_eq!(dials.load(Ordering::SeqCst), 3);
    }
}