use std::{
    env, fs, io,
    path::{Path, PathBuf},
    time::Duration,
};

use serde::{Deserialize, Serialize};
use tokio::{
    io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
    net::{UnixListener, UnixStream},
    sync::{mpsc::UnboundedSender, oneshot},
    task::JoinHandle,
};

use crate::domain::ResourceId;

pub const GZG_RUNTIME_HOME_ENV: &str = "GZG_RUNTIME_HOME";
const CONTROL_SCHEMA_VERSION: u32 = 1;
const CONTROL_TIMEOUT: Duration = Duration::from_millis(1200);

#[derive(Debug)]
pub struct RuntimeRequest {
    pub command: RuntimeCommand,
    pub reply: oneshot::Sender<ControlReply>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RuntimeCommand {
    Open {
        request_id: String,
        resource: ResourceId,
    },
    Set {
        request_id: String,
        key: String,
        value: String,
    },
}

impl RuntimeCommand {
    fn request_id(&self) -> &str {
        match self {
            Self::Open { request_id, .. } | Self::Set { request_id, .. } => request_id,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ControlReply {
    pub ok: bool,
    pub result: Option<String>,
    pub error: Option<String>,
}

impl ControlReply {
    pub fn ok(result: impl Into<String>) -> Self {
        Self {
            ok: true,
            result: Some(result.into()),
            error: None,
        }
    }

    pub fn error(error: impl Into<String>) -> Self {
        Self {
            ok: false,
            result: None,
            error: Some(error.into()),
        }
    }
}

#[derive(Debug, Serialize, Deserialize)]
struct WireCommand {
    schema_version: u32,
    id: String,
    method: String,
    #[serde(default)]
    resource: Option<String>,
    #[serde(default)]
    key: Option<String>,
    #[serde(default)]
    value: Option<String>,
}

#[derive(Debug, Serialize, Deserialize)]
struct WireReply {
    schema_version: u32,
    id: String,
    ok: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    result: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    error: Option<String>,
}

pub fn runtime_dir() -> PathBuf {
    runtime_dir_from_env(
        env::var_os(GZG_RUNTIME_HOME_ENV),
        env::var_os("XDG_RUNTIME_DIR"),
        env::var_os("UID"),
    )
}

fn runtime_dir_from_env(
    override_dir: Option<std::ffi::OsString>,
    xdg_runtime: Option<std::ffi::OsString>,
    uid: Option<std::ffi::OsString>,
) -> PathBuf {
    if let Some(path) = override_dir {
        return PathBuf::from(path);
    }
    if let Some(path) = xdg_runtime {
        return PathBuf::from(path).join("ghzinga");
    }
    let suffix = uid
        .and_then(|value| value.into_string().ok())
        .filter(|value| !value.trim().is_empty())
        .unwrap_or_else(|| std::process::id().to_string());
    PathBuf::from("/tmp").join(format!("ghzinga-{suffix}"))
}

pub fn socket_path(session_id: &str) -> PathBuf {
    runtime_dir().join(format!("{session_id}.sock"))
}

pub struct ControlServer {
    path: PathBuf,
    task: JoinHandle<()>,
}

impl Drop for ControlServer {
    fn drop(&mut self) {
        self.task.abort();
        let _ = fs::remove_file(&self.path);
    }
}

pub fn start_server(
    session_id: &str,
    tx: UnboundedSender<RuntimeRequest>,
) -> io::Result<ControlServer> {
    let path = socket_path(session_id);
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent)?;
    }
    remove_stale_socket(&path)?;
    let listener = UnixListener::bind(&path)?;
    set_owner_only_permissions(&path);
    let task = tokio::spawn(server_loop(listener, tx));
    Ok(ControlServer { path, task })
}

fn remove_stale_socket(path: &Path) -> io::Result<()> {
    if !path.exists() {
        return Ok(());
    }
    match std::os::unix::net::UnixStream::connect(path) {
        Ok(_) => Err(io::Error::new(
            io::ErrorKind::AddrInUse,
            format!("ghzinga session socket already active: {}", path.display()),
        )),
        Err(_) => fs::remove_file(path),
    }
}

fn set_owner_only_permissions(path: &Path) {
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;

        if let Ok(metadata) = fs::metadata(path) {
            let mut permissions = metadata.permissions();
            permissions.set_mode(0o600);
            let _ = fs::set_permissions(path, permissions);
        }
    }
}

async fn server_loop(listener: UnixListener, tx: UnboundedSender<RuntimeRequest>) {
    loop {
        let Ok((stream, _)) = listener.accept().await else {
            break;
        };
        let tx = tx.clone();
        tokio::spawn(async move {
            let _ = handle_connection(stream, tx).await;
        });
    }
}

async fn handle_connection(
    stream: UnixStream,
    tx: UnboundedSender<RuntimeRequest>,
) -> io::Result<()> {
    let mut reader = BufReader::new(stream);
    let mut line = String::new();
    reader.read_line(&mut line).await?;
    let command = parse_wire_command(&line);
    let reply = match command {
        Ok(command) => {
            let request_id = command.request_id().to_string();
            let (reply_tx, reply_rx) = oneshot::channel();
            if tx
                .send(RuntimeRequest {
                    command,
                    reply: reply_tx,
                })
                .is_err()
            {
                wire_reply(
                    &request_id,
                    ControlReply::error("ghzinga UI is not accepting commands"),
                )
            } else {
                match tokio::time::timeout(CONTROL_TIMEOUT, reply_rx).await {
                    Ok(Ok(reply)) => wire_reply(&request_id, reply),
                    Ok(Err(_)) => wire_reply(
                        &request_id,
                        ControlReply::error("ghzinga UI dropped command"),
                    ),
                    Err(_) => wire_reply(
                        &request_id,
                        ControlReply::error("ghzinga UI command timed out"),
                    ),
                }
            }
        }
        Err((request_id, error)) => wire_reply(&request_id, ControlReply::error(error)),
    };
    let stream = reader.get_mut();
    stream
        .write_all(serde_json::to_string(&reply)?.as_bytes())
        .await?;
    stream.write_all(b"\n").await?;
    stream.flush().await?;
    Ok(())
}

fn parse_wire_command(line: &str) -> Result<RuntimeCommand, (String, String)> {
    let value = serde_json::from_str::<WireCommand>(line).map_err(|error| {
        (
            "unknown".to_string(),
            format!("invalid control command: {error}"),
        )
    })?;
    if value.schema_version != CONTROL_SCHEMA_VERSION {
        return Err((
            value.id,
            format!("unsupported control schema {}", value.schema_version),
        ));
    }
    match value.method.as_str() {
        "open" => {
            let resource = value.resource.ok_or_else(|| {
                (
                    value.id.clone(),
                    "open command requires resource".to_string(),
                )
            })?;
            let resource = ResourceId::parse(&resource)
                .map_err(|error| (value.id.clone(), error.to_string()))?;
            Ok(RuntimeCommand::Open {
                request_id: value.id,
                resource,
            })
        }
        "set" => {
            let key = value
                .key
                .ok_or_else(|| (value.id.clone(), "set command requires key".to_string()))?;
            let value_text = value
                .value
                .ok_or_else(|| (value.id.clone(), "set command requires value".to_string()))?;
            Ok(RuntimeCommand::Set {
                request_id: value.id,
                key,
                value: value_text,
            })
        }
        other => Err((value.id, format!("unknown control method `{other}`"))),
    }
}

fn wire_reply(id: &str, reply: ControlReply) -> WireReply {
    WireReply {
        schema_version: CONTROL_SCHEMA_VERSION,
        id: id.to_string(),
        ok: reply.ok,
        result: reply.result,
        error: reply.error,
    }
}

pub async fn send_open(session_id: &str, resource: &ResourceId) -> io::Result<ControlReply> {
    send_command(
        session_id,
        WireCommand {
            schema_version: CONTROL_SCHEMA_VERSION,
            id: command_id(),
            method: "open".into(),
            resource: Some(resource.canonical_name()),
            key: None,
            value: None,
        },
    )
    .await
}

pub async fn send_set(session_id: &str, key: &str, value: &str) -> io::Result<ControlReply> {
    send_command(
        session_id,
        WireCommand {
            schema_version: CONTROL_SCHEMA_VERSION,
            id: command_id(),
            method: "set".into(),
            resource: None,
            key: Some(key.into()),
            value: Some(value.into()),
        },
    )
    .await
}

async fn send_command(session_id: &str, command: WireCommand) -> io::Result<ControlReply> {
    let path = socket_path(session_id);
    let mut stream = tokio::time::timeout(CONTROL_TIMEOUT, UnixStream::connect(&path))
        .await
        .map_err(|_| {
            io::Error::new(io::ErrorKind::TimedOut, "control socket connect timed out")
        })??;
    stream
        .write_all(serde_json::to_string(&command)?.as_bytes())
        .await?;
    stream.write_all(b"\n").await?;
    stream.flush().await?;
    let mut reader = BufReader::new(stream);
    let mut line = String::new();
    tokio::time::timeout(CONTROL_TIMEOUT, reader.read_line(&mut line))
        .await
        .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "control reply timed out"))??;
    let reply = serde_json::from_str::<WireReply>(&line)
        .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?;
    Ok(ControlReply {
        ok: reply.ok,
        result: reply.result,
        error: reply.error,
    })
}

fn command_id() -> String {
    use std::time::{SystemTime, UNIX_EPOCH};

    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos();
    format!("c_{}_{}", std::process::id(), nanos)
}

#[cfg(test)]
mod tests {
    use tempfile::tempdir;
    use tokio::sync::Mutex;

    use super::*;

    static ENV_LOCK: Mutex<()> = Mutex::const_new(());

    #[test]
    fn runtime_dir_uses_override_then_xdg_then_tmp() {
        assert_eq!(
            runtime_dir_from_env(Some("/tmp/custom".into()), None, None),
            PathBuf::from("/tmp/custom")
        );
        assert_eq!(
            runtime_dir_from_env(None, Some("/run/user/1000".into()), None),
            PathBuf::from("/run/user/1000/ghzinga")
        );
        assert_eq!(
            runtime_dir_from_env(None, None, Some("1000".into())),
            PathBuf::from("/tmp/ghzinga-1000")
        );
    }

    #[test]
    fn parses_open_wire_command() {
        let parsed = parse_wire_command(
            r#"{"schema_version":1,"id":"c_1","method":"open","resource":"owner/repo#12"}"#,
        )
        .unwrap();

        assert!(matches!(
            parsed,
            RuntimeCommand::Open {
                request_id,
                resource
            } if request_id == "c_1" && resource.canonical_name() == "owner/repo#12"
        ));
    }

    #[test]
    fn rejects_unknown_schema() {
        let error =
            parse_wire_command(r#"{"schema_version":2,"id":"c_1","method":"open"}"#).unwrap_err();

        assert_eq!(error.0, "c_1");
        assert!(error.1.contains("unsupported control schema"));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn server_receives_open_command_and_replies() {
        let _guard = ENV_LOCK.lock().await;
        let dir = tempdir().unwrap();
        let previous_runtime = env::var_os(GZG_RUNTIME_HOME_ENV);
        env::set_var(GZG_RUNTIME_HOME_ENV, dir.path());
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let _server = start_server("work", tx).unwrap();
        let id = ResourceId::parse("owner/repo#12").unwrap();
        let client = tokio::spawn(async move { send_open("work", &id).await.unwrap() });

        let request = rx.recv().await.unwrap();
        assert!(matches!(
            request.command,
            RuntimeCommand::Open { resource, .. } if resource.canonical_name() == "owner/repo#12"
        ));
        request.reply.send(ControlReply::ok("opened")).unwrap();

        let reply = client.await.unwrap();
        assert!(reply.ok);
        assert_eq!(reply.result.as_deref(), Some("opened"));

        if let Some(value) = previous_runtime {
            env::set_var(GZG_RUNTIME_HOME_ENV, value);
        } else {
            env::remove_var(GZG_RUNTIME_HOME_ENV);
        }
    }
}
