//! HTTP API for programmatic ticket management and peer sync.
//!
//! The API runs on the tailnet, but tailnet membership is not sufficient on its
//! own: anything that can reach the port could otherwise forge tickets. So the
//! server fails closed -- it never serves without a shared bearer token. The
//! token is resolved in order from `--token`, the `WAM_TOKEN` env var, a token
//! persisted beside the database, or, failing all of those, a freshly generated
//! one that is persisted and printed once at startup. Every request (including
//! peer sync) must carry `Authorization: Bearer <token>`.

use std::sync::Arc;

use color_eyre::eyre::WrapErr;
use tokio::sync::Mutex;

use axum::{
    Json, Router,
    extract::{Path, Query, Request, State},
    http::StatusCode,
    middleware::{self, Next},
    response::{IntoResponse, Response},
    routing::{get, patch, post},
};
use rusqlite::Connection;
use serde::Deserialize;

use crate::db::{self, ListFilter};
use crate::types::{Channel, NewTicket, Priority, Status, Ticket};

/// Shared state: SQLite connection + node identity + shared bearer token.
#[derive(Clone)]
pub(crate) struct AppState {
    pub db: Arc<Mutex<Connection>>,
    pub node_id: String,
    /// Shared bearer token required on every request and presented to peers.
    /// Always present -- the server refuses to start without one.
    pub token: Arc<str>,
}

/// Start the HTTP server, optionally syncing with peers.
///
/// `token` is the CLI-supplied shared secret. Token resolution fails closed:
/// see [`resolve_or_create_token`]. Every request (including peer sync) must
/// present the resolved token as a bearer token.
pub(crate) async fn serve(
    conn: Connection,
    port: u16,
    peers: Vec<String>,
    token: Option<String>,
) -> color_eyre::eyre::Result<()> {
    let node_id = db::get_or_create_node_id(&conn)?;
    eprintln!("node: {}", &node_id[..8]);

    let token = resolve_or_create_token(token)?;

    let app_state = AppState {
        db: Arc::new(Mutex::new(conn)),
        node_id,
        token: Arc::clone(&token),
    };

    // Spawn sync loop if peers are configured
    if !peers.is_empty() {
        let sync_state = app_state.clone();
        let sync_peers = peers.clone();
        tokio::spawn(async move {
            sync_loop(sync_state, sync_peers).await;
        });
    }

    let app = Router::new()
        .route("/tickets", post(create_ticket))
        .route("/tickets", get(list_tickets))
        .route("/tickets/{id}", get(get_ticket))
        .route("/tickets/{id}", patch(update_ticket))
        .route("/sync/pull", get(sync_pull))
        .route("/sync/push", post(sync_push))
        .route("/sync/node", get(sync_node_info))
        .route_layer(middleware::from_fn_with_state(
            app_state.clone(),
            require_auth,
        ))
        .with_state(app_state);

    let addr = format!("0.0.0.0:{port}");
    let listener = tokio::net::TcpListener::bind(&addr).await?;
    eprintln!("wam serving on {addr}");
    eprintln!("auth: bearer token required on every request");
    if !peers.is_empty() {
        eprintln!("syncing with {} peer(s)", peers.len());
    }
    axum::serve(listener, app).await?;
    Ok(())
}

/// Resolve the shared bearer token, failing closed so the API is never open.
///
/// Precedence: the explicit `--token` flag, then the `WAM_TOKEN` env var, then a
/// token persisted beside the database from a previous run, then a freshly
/// generated one (persisted with `0600` perms and printed once). The result is
/// always a usable token; there is no unauthenticated path.
fn resolve_or_create_token(cli_token: Option<String>) -> color_eyre::eyre::Result<Arc<str>> {
    // An explicit flag or environment variable wins, in that order.
    let supplied = cli_token
        .or_else(|| std::env::var("WAM_TOKEN").ok())
        .map(|t| t.trim().to_owned())
        .filter(|t| !t.is_empty());
    if let Some(tok) = supplied {
        return Ok(Arc::from(tok));
    }

    token_from_dir(&db::data_dir()?)
}

/// Reuse a token persisted in `dir/token`, or mint, persist, and print a new
/// one. Split out from [`resolve_or_create_token`] so the disk paths are
/// testable against a scratch directory.
fn token_from_dir(dir: &std::path::Path) -> color_eyre::eyre::Result<Arc<str>> {
    let path = dir.join("token");

    // Reuse a token persisted by an earlier run.
    if let Ok(saved) = std::fs::read_to_string(&path) {
        let saved = saved.trim().to_owned();
        if !saved.is_empty() {
            eprintln!("auth: using persisted token at {}", path.display());
            return Ok(Arc::from(saved));
        }
    }

    // Nothing configured: mint one, persist it, and print it once so the
    // operator can propagate it to producers and peers.
    let token = format!(
        "{}{}",
        uuid::Uuid::new_v4().simple(),
        uuid::Uuid::new_v4().simple()
    );
    write_token_file(&path, &token)?;
    eprintln!(
        "auth: no token supplied -- generated one and saved it to {}",
        path.display()
    );
    eprintln!("      token: {token}");
    eprintln!("      set WAM_TOKEN to this value on every producer and peer.");
    Ok(Arc::from(token))
}

/// Write the token to `path`, restricting it to owner-only (`0600`) on unix.
fn write_token_file(path: &std::path::Path, token: &str) -> color_eyre::eyre::Result<()> {
    std::fs::write(path, format!("{token}\n"))
        .wrap_err_with(|| format!("write token file: {}", path.display()))?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))
            .wrap_err_with(|| format!("restrict token file perms: {}", path.display()))?;
    }
    Ok(())
}

/// Reject any request lacking a valid `Authorization: Bearer <token>` header.
/// The token is always set, so every route behind this layer is authenticated.
async fn require_auth(State(state): State<AppState>, req: Request, next: Next) -> Response {
    let expected = state.token.as_ref();
    let presented = req
        .headers()
        .get(axum::http::header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.strip_prefix("Bearer "));
    match presented {
        Some(tok) if ct_eq(tok.as_bytes(), expected.as_bytes()) => next.run(req).await,
        _ => (
            StatusCode::UNAUTHORIZED,
            Json(serde_json::json!({"error": "unauthorized"})),
        )
            .into_response(),
    }
}

/// Constant-time byte comparison, so a wrong token can't be recovered by timing
/// how far it matched.
fn ct_eq(a: &[u8], b: &[u8]) -> bool {
    if a.len() != b.len() {
        return false;
    }
    let mut diff = 0u8;
    for (x, y) in a.iter().zip(b) {
        diff |= x ^ y;
    }
    diff == 0
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn ct_eq_matches_identical() {
        assert!(ct_eq(b"s3cr3t-token", b"s3cr3t-token"));
        assert!(ct_eq(b"", b""));
    }

    #[test]
    fn ct_eq_rejects_different_content() {
        assert!(!ct_eq(b"s3cr3t-token", b"s3cr3t-toker"));
        assert!(!ct_eq(b"abc", b"xyz"));
    }

    #[test]
    fn ct_eq_rejects_length_mismatch() {
        // A correct prefix must not pass — the guard is exact-match only.
        assert!(!ct_eq(b"s3cr3t", b"s3cr3t-token"));
        assert!(!ct_eq(b"s3cr3t-token", b"s3cr3t"));
    }

    #[test]
    fn with_token_attaches_only_when_set() {
        let client = reqwest::Client::new();
        // Smoke: both branches build a valid request without panicking.
        let _ = with_token(client.get("http://127.0.0.1/x"), Some("tok"));
        let _ = with_token(client.get("http://127.0.0.1/x"), None);
    }

    /// A unique scratch directory under the system temp dir, removed on drop.
    struct ScratchDir(std::path::PathBuf);

    impl ScratchDir {
        fn new() -> Self {
            let dir = std::env::temp_dir().join(format!("wam-test-{}", uuid::Uuid::new_v4()));
            std::fs::create_dir_all(&dir).unwrap();
            Self(dir)
        }
    }

    impl Drop for ScratchDir {
        fn drop(&mut self) {
            let _ = std::fs::remove_dir_all(&self.0);
        }
    }

    #[test]
    fn explicit_token_wins_and_is_trimmed() {
        // A supplied token short-circuits before any env or disk lookup.
        let tok = resolve_or_create_token(Some("  s3cr3t  ".to_string())).unwrap();
        assert_eq!(&*tok, "s3cr3t");
    }

    #[test]
    fn token_from_dir_generates_persists_and_reuses() {
        let scratch = ScratchDir::new();

        // First call mints a token and writes it to dir/token.
        let first = token_from_dir(&scratch.0).unwrap();
        assert!(!first.is_empty());
        let path = scratch.0.join("token");
        assert!(path.exists());
        assert_eq!(std::fs::read_to_string(&path).unwrap().trim(), &*first);

        // On unix the file is owner-only.
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let mode = std::fs::metadata(&path).unwrap().permissions().mode();
            assert_eq!(mode & 0o777, 0o600);
        }

        // A second call reuses the persisted token rather than minting anew.
        let second = token_from_dir(&scratch.0).unwrap();
        assert_eq!(&*first, &*second);
    }

    #[test]
    fn token_from_dir_ignores_blank_file() {
        let scratch = ScratchDir::new();
        std::fs::write(scratch.0.join("token"), "   \n").unwrap();
        // A whitespace-only file is treated as absent: a real token is minted.
        let tok = token_from_dir(&scratch.0).unwrap();
        assert!(!tok.trim().is_empty());
        assert_eq!(tok.len(), 64); // two hyphen-free v4 UUIDs
    }
}

// -- Ticket handlers ----------------------------------------------------------

/// POST /tickets
async fn create_ticket(
    State(state): State<AppState>,
    Json(new): Json<NewTicket>,
) -> impl IntoResponse {
    let conn = state.db.lock().await;
    match db::create_ticket(&conn, &new, &state.node_id) {
        Ok(ticket) => (StatusCode::CREATED, Json(serde_json::json!(ticket))).into_response(),
        Err(e) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({"error": e.to_string()})),
        )
            .into_response(),
    }
}

#[derive(Debug, Deserialize, Default)]
pub(crate) struct ListQuery {
    pub status: Option<String>,
    pub priority: Option<String>,
    pub channel: Option<String>,
    pub source: Option<String>,
    pub search: Option<String>,
}

/// GET /tickets
async fn list_tickets(
    State(state): State<AppState>,
    Query(q): Query<ListQuery>,
) -> impl IntoResponse {
    let status = q.status.as_deref().and_then(|s| s.parse::<Status>().ok());
    let priority = q
        .priority
        .as_deref()
        .and_then(|s| s.parse::<Priority>().ok());
    let channel = q.channel.as_deref().and_then(|s| s.parse::<Channel>().ok());

    let conn = state.db.lock().await;
    let filter = ListFilter {
        status,
        priority,
        channel,
        source: q.source.as_deref(),
        search: q.search.as_deref(),
    };
    match db::list_tickets(&conn, &filter) {
        Ok(tickets) => {
            Json(serde_json::json!({"data": tickets, "count": tickets.len()})).into_response()
        }
        Err(e) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({"error": e.to_string()})),
        )
            .into_response(),
    }
}

/// GET /tickets/:id
async fn get_ticket(State(state): State<AppState>, Path(id): Path<String>) -> impl IntoResponse {
    let conn = state.db.lock().await;
    match db::get_ticket(&conn, &id) {
        Ok(ticket) => Json(serde_json::json!(ticket)).into_response(),
        Err(_) => (
            StatusCode::NOT_FOUND,
            Json(serde_json::json!({"error": "ticket not found"})),
        )
            .into_response(),
    }
}

#[derive(Debug, Deserialize)]
pub(crate) struct UpdateBody {
    pub status: Option<String>,
}

/// PATCH /tickets/:id
async fn update_ticket(
    State(state): State<AppState>,
    Path(id): Path<String>,
    Json(body): Json<UpdateBody>,
) -> impl IntoResponse {
    let conn = state.db.lock().await;

    let Ok(ticket) = db::get_ticket(&conn, &id) else {
        return (
            StatusCode::NOT_FOUND,
            Json(serde_json::json!({"error": "ticket not found"})),
        )
            .into_response();
    };

    if let Some(status_str) = body.status {
        let Ok(status) = status_str.parse::<Status>() else {
            return (
                StatusCode::BAD_REQUEST,
                Json(serde_json::json!({"error": format!("invalid status: {status_str}")})),
            )
                .into_response();
        };
        if let Err(e) = db::update_status(&conn, &ticket.id, status) {
            return (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(serde_json::json!({"error": e.to_string()})),
            )
                .into_response();
        }
    }

    match db::get_ticket(&conn, &ticket.id) {
        Ok(t) => Json(serde_json::json!(t)).into_response(),
        Err(e) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({"error": e.to_string()})),
        )
            .into_response(),
    }
}

// -- Sync endpoints -----------------------------------------------------------

#[derive(Debug, Deserialize)]
pub(crate) struct SyncPullQuery {
    /// RFC3339 timestamp. Returns tickets updated after this time.
    pub since: String,
}

/// GET /sync/pull?since=<rfc3339> — peer pulls tickets updated after timestamp
async fn sync_pull(
    State(state): State<AppState>,
    Query(q): Query<SyncPullQuery>,
) -> impl IntoResponse {
    let conn = state.db.lock().await;
    match db::tickets_since(&conn, &q.since) {
        Ok(tickets) => Json(serde_json::json!({
            "tickets": tickets,
            "count": tickets.len(),
            "node_id": state.node_id,
        }))
        .into_response(),
        Err(e) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({"error": e.to_string()})),
        )
            .into_response(),
    }
}

/// POST /sync/push — peer pushes tickets to us
async fn sync_push(
    State(state): State<AppState>,
    Json(tickets): Json<Vec<Ticket>>,
) -> impl IntoResponse {
    let conn = state.db.lock().await;
    let mut accepted = 0u32;
    let mut rejected = 0u32;

    for ticket in &tickets {
        match db::upsert_synced_ticket(&conn, ticket) {
            Ok(true) => accepted += 1,
            Ok(false) => rejected += 1,
            Err(e) => {
                eprintln!("sync upsert error for {}: {e}", ticket.short_id());
                rejected += 1;
            }
        }
    }

    Json(serde_json::json!({
        "accepted": accepted,
        "rejected": rejected,
    }))
}

/// GET /sync/node — returns this node's identity
async fn sync_node_info(State(state): State<AppState>) -> impl IntoResponse {
    Json(serde_json::json!({
        "node_id": state.node_id,
    }))
}

/// Attach the shared bearer token to an outbound peer request when one is set.
fn with_token(req: reqwest::RequestBuilder, token: Option<&str>) -> reqwest::RequestBuilder {
    match token {
        Some(t) => req.bearer_auth(t),
        None => req,
    }
}

// -- Background sync loop -----------------------------------------------------

/// Periodically pull from all peers and push local changes.
async fn sync_loop(state: AppState, peers: Vec<String>) {
    let client = reqwest::Client::builder()
        .timeout(std::time::Duration::from_secs(10))
        .connect_timeout(std::time::Duration::from_secs(5))
        .build()
        .unwrap();

    loop {
        tokio::time::sleep(std::time::Duration::from_secs(30)).await;

        for peer in &peers {
            if let Err(e) = sync_with_peer(&state, &client, peer).await {
                eprintln!("sync with {peer}: {e}");
            }
        }
    }
}

/// Pull new tickets from a peer, then push our new tickets to them.
async fn sync_with_peer(
    state: &AppState,
    client: &reqwest::Client,
    peer_url: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let conn = state.db.lock().await;

    // Get our cursor for this peer (default to epoch)
    let cursor =
        db::get_sync_cursor(&conn, peer_url)?.unwrap_or_else(|| "1970-01-01T00:00:00Z".to_string());

    drop(conn); // Release lock before HTTP

    // Pull from peer
    let pull_url = format!(
        "{peer_url}/sync/pull?since={}",
        urlencoding::encode(&cursor)
    );
    let resp: serde_json::Value = with_token(client.get(&pull_url), Some(state.token.as_ref()))
        .send()
        .await?
        .json()
        .await?;

    let tickets: Vec<Ticket> = serde_json::from_value(
        resp.get("tickets")
            .cloned()
            .unwrap_or(serde_json::json!([])),
    )?;

    if !tickets.is_empty() {
        let conn = state.db.lock().await;
        let mut latest_updated = cursor.clone();

        for ticket in &tickets {
            db::upsert_synced_ticket(&conn, ticket)?;
            let ts = ticket.updated_at.to_rfc3339();
            if ts > latest_updated {
                latest_updated = ts;
            }
        }

        db::set_sync_cursor(&conn, peer_url, &latest_updated)?;
        drop(conn);

        eprintln!("sync: pulled {} ticket(s) from {peer_url}", tickets.len());
    }

    // Push our changes to peer (tickets updated since their last pull from us)
    // We use the same cursor — they'll filter by last-writer-wins
    let conn = state.db.lock().await;
    let our_tickets = db::tickets_since(&conn, &cursor)?;
    drop(conn);

    if !our_tickets.is_empty() {
        let push_url = format!("{peer_url}/sync/push");
        let resp: serde_json::Value = with_token(
            client.post(&push_url).json(&our_tickets),
            Some(state.token.as_ref()),
        )
        .send()
        .await?
        .json()
        .await?;
        let accepted = resp
            .get("accepted")
            .and_then(serde_json::Value::as_u64)
            .unwrap_or(0);
        if accepted > 0 {
            eprintln!("sync: pushed {accepted} ticket(s) to {peer_url}");
        }
    }

    Ok(())
}