use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Duration;

use serde_json::json;
use tokio::sync::Mutex;
use zbus::Connection;
use zbus::message::Header;
use zbus::object_server::SignalEmitter;
use zbus::zvariant::{OwnedFd, OwnedObjectPath, Value};

use crate::backend::{Cancel, Options, authorize};
use crate::bridge::Bridge;
use crate::desktop::Requests;
use crate::eis::Worker;

const PATH: &str = "/org/freedesktop/portal/desktop";

#[derive(Default)]
struct State {
    starting: bool,
    started: bool,
    id: u64,
    capabilities: u32,
    keymap: String,
    bridge: Option<Bridge>,
    watch: Option<Bridge>,
    worker: Option<Worker>,
}

struct Session {
    owner: String,
    app: String,
    cancel: Arc<Cancel>,
    state: Mutex<State>,
}

#[derive(Clone, Default)]
pub(crate) struct InputCapture {
    sessions: Arc<Mutex<HashMap<String, Arc<Session>>>>,
    requests: Requests,
}

impl InputCapture {
    pub(crate) fn new(requests: Requests) -> Self {
        Self {
            requests,
            ..Self::default()
        }
    }

    async fn session(
        &self,
        connection: &Connection,
        header: &Header<'_>,
        path: &OwnedObjectPath,
        app: &str,
    ) -> zbus::fdo::Result<Arc<Session>> {
        let owner = authorize(connection, header).await?;
        self.sessions
            .lock()
            .await
            .get(path.as_str())
            .filter(|session| session.owner == owner && session.app == app)
            .cloned()
            .ok_or_else(|| invalid("Unknown input capture session"))
    }

    async fn end(&self, connection: &Connection, path: &str) {
        let session = self.sessions.lock().await.remove(path);
        if let Some(session) = session {
            session.cancel.stop();
            let state = session.state.lock().await;
            if let Some(worker) = &state.worker {
                worker.stop();
            }
            if let Some(bridge) = &state.bridge {
                bridge.close();
            }
            if let Some(watch) = &state.watch {
                watch.close();
            }
            if let Ok(emitter) = SignalEmitter::new(connection, path) {
                let _ = SessionObject::closed(&emitter).await;
            }
            let connection = connection.clone();
            let path = path.to_owned();
            tokio::spawn(async move {
                let _ = connection.object_server().remove::<SessionObject, _>(path).await;
            });
        }
    }

    pub(crate) async fn revoke_stale(&self, connection: &Connection, owner: Option<&str>) {
        let stale = self
            .sessions
            .lock()
            .await
            .iter()
            .filter_map(|(path, session)| (Some(session.owner.as_str()) != owner).then_some(path.clone()))
            .collect::<Vec<_>>();
        for path in stale {
            self.end(connection, &path).await;
        }
    }

    async fn native(session: &Session) -> zbus::fdo::Result<(Bridge, u64)> {
        let state = session.state.lock().await;
        if !state.started || session.cancel.stopped.load(Ordering::SeqCst) {
            return Err(invalid("Start the session first"));
        }
        Ok((
            state
                .bridge
                .clone()
                .ok_or_else(|| invalid("Missing capture connection"))?,
            state.id,
        ))
    }

    async fn watch(
        self,
        connection: Connection,
        path: OwnedObjectPath,
        session: Arc<Session>,
        bridge: Bridge,
        id: u64,
    ) {
        loop {
            let response = tokio::select! {
                _ = session.cancel.wait() => break,
                result = bridge.capture_watch(id) => result,
            };
            let Ok(response) = response else {
                break;
            };
            let Some(events) = response["events"].as_array() else {
                break;
            };
            let state = session.state.lock().await;
            if session.cancel.stopped.load(Ordering::SeqCst) {
                break;
            }
            let transport = state.worker.as_ref().map(|worker| worker.send(events.clone()));
            if transport.is_some_and(|result| result.is_err()) {
                break;
            }
            let mut closed = false;
            let Ok(emitter) = SignalEmitter::new(&connection, PATH) else {
                break;
            };
            for event in events {
                let result = match event["type"].as_str() {
                    Some("activated") => {
                        let mut options = Options::new();
                        options.insert(
                            "activation_id".into(),
                            (event["activation_id"].as_u64().unwrap_or(0) as u32).into(),
                        );
                        options.insert(
                            "barrier_id".into(),
                            (event["barrier_id"].as_u64().unwrap_or(0) as u32).into(),
                        );
                        if let Some(position) = event["cursor_position"]
                            .as_array()
                            .filter(|position| position.len() == 2)
                            && let Ok(position) =
                                Value::from((position[0].as_f64().unwrap_or(0.0), position[1].as_f64().unwrap_or(0.0)))
                                    .try_to_owned()
                        {
                            options.insert("cursor_position".into(), position);
                        }
                        Self::activated(&emitter, path.clone(), options).await
                    }
                    Some("deactivated") => {
                        Self::deactivated(
                            &emitter,
                            path.clone(),
                            Options::from([(
                                "activation_id".into(),
                                (event["activation_id"].as_u64().unwrap_or(0) as u32).into(),
                            )]),
                        )
                        .await
                    }
                    Some("disabled") => Self::disabled(&emitter, path.clone(), Options::new()).await,
                    Some("zones") => Self::zones_changed(&emitter, path.clone(), Options::new()).await,
                    Some("closed") => {
                        closed = true;
                        Ok(())
                    }
                    _ => Ok(()),
                };
                if result.is_err() {
                    closed = true;
                }
            }
            drop(state);
            if closed {
                break;
            }
        }
        self.end(&connection, path.as_str()).await;
    }
}

fn invalid(message: &str) -> zbus::fdo::Error {
    zbus::fdo::Error::InvalidArgs(message.into())
}

fn failed(error: String) -> zbus::fdo::Error {
    zbus::fdo::Error::Failed(error)
}

fn capabilities(options: &Options) -> zbus::fdo::Result<u32> {
    let requested = options
        .get("capabilities")
        .ok_or_else(|| invalid("Missing requested capabilities"))?;
    let requested = u32::try_from(requested).map_err(|_| invalid("Invalid requested capabilities"))?;
    if requested == 0 || requested & 3 == 0 {
        return Err(invalid("No supported capabilities requested"));
    }
    Ok(requested & 3)
}

#[zbus::interface(name = "org.freedesktop.impl.portal.InputCapture")]
impl InputCapture {
    #[zbus(property, name = "version")]
    fn version(&self) -> u32 {
        2
    }

    #[zbus(property)]
    fn supported_capabilities(&self) -> u32 {
        3
    }

    async fn create_session2(
        &self,
        session_handle: OwnedObjectPath,
        app_id: String,
        _options: Options,
        #[zbus(connection)] connection: &Connection,
        #[zbus(header)] header: Header<'_>,
    ) -> zbus::fdo::Result<Options> {
        let owner = authorize(connection, &header).await?;
        if !session_handle.as_str().starts_with(&format!("{PATH}/session/"))
            || session_handle.as_str().len() > 512
            || app_id.len() > 512
        {
            return Err(invalid("Invalid session"));
        }
        let mut sessions = self.sessions.lock().await;
        if sessions.len() >= 8 || sessions.contains_key(session_handle.as_str()) {
            return Err(invalid("Session limit reached or duplicate session"));
        }
        if !connection
            .object_server()
            .at(
                &session_handle,
                SessionObject {
                    backend: self.clone(),
                    path: session_handle.to_string(),
                    owner: owner.clone(),
                },
            )
            .await?
        {
            return Err(invalid("Session object already exists"));
        }
        sessions.insert(
            session_handle.to_string(),
            Arc::new(Session {
                owner,
                app: app_id,
                cancel: Arc::default(),
                state: Mutex::default(),
            }),
        );
        Ok(Options::new())
    }

    #[allow(clippy::too_many_arguments)]
    async fn create_session(
        &self,
        handle: OwnedObjectPath,
        session_handle: OwnedObjectPath,
        app_id: String,
        parent_window: String,
        options: Options,
        #[zbus(connection)] connection: &Connection,
        #[zbus(header)] header: Header<'_>,
    ) -> zbus::fdo::Result<(u32, Options)> {
        capabilities(&options)?;
        self.create_session2(
            session_handle.clone(),
            app_id.clone(),
            Options::new(),
            connection,
            header.clone(),
        )
        .await?;
        let result = self
            .start(
                handle,
                session_handle.clone(),
                app_id,
                parent_window,
                options,
                connection,
                header,
            )
            .await;
        let (response, mut results) = match result {
            Ok(result) => result,
            Err(error) => {
                self.end(connection, session_handle.as_str()).await;
                return Err(error);
            }
        };
        if response == 0 {
            results.insert(
                "session_id".into(),
                Value::from(session_handle.to_string())
                    .try_to_owned()
                    .map_err(|e| failed(e.to_string()))?,
            );
        }
        Ok((response, results))
    }

    #[allow(clippy::too_many_arguments)]
    async fn start(
        &self,
        handle: OwnedObjectPath,
        session_handle: OwnedObjectPath,
        app_id: String,
        parent_window: String,
        options: Options,
        #[zbus(connection)] connection: &Connection,
        #[zbus(header)] header: Header<'_>,
    ) -> zbus::fdo::Result<(u32, Options)> {
        let session = self.session(connection, &header, &session_handle, &app_id).await?;
        let caps = capabilities(&options)?;
        crate::restore::persist_mode(&options)?;
        let cancel = self.requests.begin(connection, &header, &handle).await?;
        {
            let mut state = session.state.lock().await;
            if state.starting || state.started {
                self.requests.end(connection, &handle).await;
                return Err(invalid("Input capture can only be started once"));
            }
            state.starting = true;
        }
        let title = match caps {
            1 => "Share keyboard input?",
            2 => "Share pointer input?",
            _ => "Share keyboard and pointer?",
        };
        let consent = crate::desktop::consent(
            &app_id,
            &parent_window,
            title,
            "Allow input capture at the screen edges this app configures. Captured input is sent to this app until it releases control. Press Ctrl+Alt+Escape to return control to Ferese.",
            "Allow input capture",
            None,
        );
        let approved = tokio::select! {
            _ = cancel.wait() => Ok(false),
            _ = session.cancel.wait() => Ok(false),
            result = consent => result,
        };
        let result = if approved == Ok(true)
            && !cancel.stopped.load(Ordering::SeqCst)
            && !session.cancel.stopped.load(Ordering::SeqCst)
        {
            async {
                let bridge = Bridge::connect()?;
                let reply = bridge
                    .call("input-capture-register", json!({"capabilities":caps}))
                    .await?;
                let id = reply["session"].as_u64().ok_or("Missing capture session ID")?;
                let keymap = reply["keymap"].as_str().ok_or("Missing keyboard map")?.to_owned();
                let watch = Bridge::connect()?;
                let mut state = session.state.lock().await;
                if cancel.stopped.load(Ordering::SeqCst) || session.cancel.stopped.load(Ordering::SeqCst) {
                    bridge.close();
                    return Err("Input capture approval was cancelled".to_owned());
                }
                state.started = true;
                state.starting = false;
                state.id = id;
                state.capabilities = caps;
                state.keymap = keymap;
                state.bridge = Some(bridge);
                state.watch = Some(watch.clone());
                tokio::spawn(self.clone().watch(
                    connection.clone(),
                    session_handle.clone(),
                    session.clone(),
                    watch,
                    id,
                ));
                Ok(Options::from([
                    ("capabilities".into(), caps.into()),
                    ("clipboard_enabled".into(), false.into()),
                    ("persist_mode".into(), 0u32.into()),
                ]))
            }
            .await
        } else {
            Err(approved.err().unwrap_or_else(|| "Cancelled".into()))
        };
        self.requests.end(connection, &handle).await;
        match result {
            Ok(results) => Ok((0, results)),
            Err(error) => {
                let response = if error == "Cancelled"
                    || cancel.stopped.load(Ordering::SeqCst)
                    || session.cancel.stopped.load(Ordering::SeqCst)
                {
                    1
                } else {
                    eprintln!("ferese input capture: {error}");
                    2
                };
                self.end(connection, session_handle.as_str()).await;
                Ok((response, Options::new()))
            }
        }
    }

    async fn get_zones(
        &self,
        handle: OwnedObjectPath,
        session_handle: OwnedObjectPath,
        app_id: String,
        _options: Options,
        #[zbus(connection)] connection: &Connection,
        #[zbus(header)] header: Header<'_>,
    ) -> zbus::fdo::Result<(u32, Options)> {
        let session = self.session(connection, &header, &session_handle, &app_id).await?;
        let (bridge, id) = Self::native(&session).await?;
        let cancel = self.requests.begin(connection, &header, &handle).await?;
        let result = tokio::select! {
            biased;
            _ = cancel.wait() => None,
            _ = session.cancel.wait() => None,
            result = bridge.call("input-capture-zones", json!({"session":id})) => Some(result),
        };
        self.requests.end(connection, &handle).await;
        if cancel.stopped.load(Ordering::SeqCst) || session.cancel.stopped.load(Ordering::SeqCst) || result.is_none() {
            return Ok((1, Options::new()));
        }
        let result = result.unwrap().map_err(failed)?;
        let zones: Vec<(u32, u32, i32, i32)> =
            serde_json::from_value(result["zones"].clone()).map_err(|_| invalid("Invalid native capture zones"))?;
        Ok((
            0,
            Options::from([
                (
                    "zones".into(),
                    Value::from(zones).try_to_owned().map_err(|e| failed(e.to_string()))?,
                ),
                (
                    "zone_set".into(),
                    (result["zone_set"].as_u64().unwrap_or(0) as u32).into(),
                ),
            ]),
        ))
    }

    #[allow(clippy::too_many_arguments)]
    async fn set_pointer_barriers(
        &self,
        handle: OwnedObjectPath,
        session_handle: OwnedObjectPath,
        app_id: String,
        _options: Options,
        barriers: Vec<Options>,
        zone_set: u32,
        #[zbus(connection)] connection: &Connection,
        #[zbus(header)] header: Header<'_>,
    ) -> zbus::fdo::Result<(u32, Options)> {
        let session = self.session(connection, &header, &session_handle, &app_id).await?;
        let (bridge, id) = Self::native(&session).await?;
        if barriers.len() > 64 {
            return Err(invalid("Too many pointer barriers"));
        }
        let barriers = barriers
            .into_iter()
            .map(|barrier| {
                let id = u32::try_from(barrier.get("barrier_id").ok_or_else(|| invalid("Missing barrier ID"))?)
                    .map_err(|_| invalid("Invalid barrier ID"))?;
                let position = barrier
                    .get("position")
                    .ok_or_else(|| invalid("Missing barrier position"))?
                    .try_clone()
                    .map_err(|_| invalid("Invalid position"))?;
                let position: (i32, i32, i32, i32) =
                    position.try_into().map_err(|_| invalid("Invalid barrier position"))?;
                Ok(json!({"id":id,"position":[position.0,position.1,position.2,position.3]}))
            })
            .collect::<zbus::fdo::Result<Vec<_>>>()?;
        let cancel = self.requests.begin(connection, &header, &handle).await?;
        let result = tokio::select! {
            biased;
            _ = cancel.wait() => None,
            _ = session.cancel.wait() => None,
            result = bridge.call("input-capture-barriers", json!({"session":id,"zone_set":zone_set,"barriers":barriers})) => Some(result),
        };
        self.requests.end(connection, &handle).await;
        if cancel.stopped.load(Ordering::SeqCst) || session.cancel.stopped.load(Ordering::SeqCst) || result.is_none() {
            self.end(connection, session_handle.as_str()).await;
            return Ok((1, Options::new()));
        }
        let result = result.unwrap().map_err(failed)?;
        let failed_barriers: Vec<u32> = serde_json::from_value(result["failed_barriers"].clone())
            .map_err(|_| invalid("Invalid native barrier response"))?;
        Ok((
            0,
            Options::from([(
                "failed_barriers".into(),
                Value::from(failed_barriers)
                    .try_to_owned()
                    .map_err(|e| failed(e.to_string()))?,
            )]),
        ))
    }

    #[zbus(name = "ConnectToEIS")]
    async fn connect_to_eis(
        &self,
        session_handle: OwnedObjectPath,
        app_id: String,
        _options: Options,
        #[zbus(connection)] connection: &Connection,
        #[zbus(header)] header: Header<'_>,
    ) -> zbus::fdo::Result<OwnedFd> {
        let session = self.session(connection, &header, &session_handle, &app_id).await?;
        Self::native(&session).await?;
        let mut state = session.state.lock().await;
        if state.worker.is_some() {
            return Err(invalid("EIS has already been connected"));
        }
        let capabilities = state.capabilities;
        let keymap = state.keymap.clone();
        let cancel = session.cancel.clone();
        let (worker, fd) = tokio::task::spawn_blocking(move || Worker::start(capabilities, &keymap, cancel))
            .await
            .map_err(|e| failed(e.to_string()))?
            .map_err(failed)?;
        state.worker = Some(worker);
        Ok(fd.into())
    }

    async fn enable(
        &self,
        session_handle: OwnedObjectPath,
        app_id: String,
        _options: Options,
        #[zbus(connection)] connection: &Connection,
        #[zbus(header)] header: Header<'_>,
    ) -> zbus::fdo::Result<(u32, Options)> {
        let session = self.session(connection, &header, &session_handle, &app_id).await?;
        let (bridge, id) = Self::native(&session).await?;
        let ready = session
            .state
            .lock()
            .await
            .worker
            .as_ref()
            .map(|worker| worker.ready.clone())
            .ok_or_else(|| invalid("Connect to EIS before enabling input capture"))?;
        let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
        while !ready.load(Ordering::SeqCst) {
            if session.cancel.stopped.load(Ordering::SeqCst) || tokio::time::Instant::now() >= deadline {
                return Err(invalid("EIS receiver has not bound an input device"));
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        bridge
            .call("input-capture-enable", json!({"session":id}))
            .await
            .map_err(failed)?;
        Ok((0, Options::new()))
    }

    async fn disable(
        &self,
        session_handle: OwnedObjectPath,
        app_id: String,
        _options: Options,
        #[zbus(connection)] connection: &Connection,
        #[zbus(header)] header: Header<'_>,
    ) -> zbus::fdo::Result<(u32, Options)> {
        let session = self.session(connection, &header, &session_handle, &app_id).await?;
        let (bridge, id) = Self::native(&session).await?;
        bridge
            .call("input-capture-disable", json!({"session":id}))
            .await
            .map_err(failed)?;
        Ok((0, Options::new()))
    }

    async fn release(
        &self,
        session_handle: OwnedObjectPath,
        app_id: String,
        options: Options,
        #[zbus(connection)] connection: &Connection,
        #[zbus(header)] header: Header<'_>,
    ) -> zbus::fdo::Result<(u32, Options)> {
        let session = self.session(connection, &header, &session_handle, &app_id).await?;
        let (bridge, id) = Self::native(&session).await?;
        let activation = options
            .get("activation_id")
            .map(u32::try_from)
            .transpose()
            .map_err(|_| invalid("Invalid activation ID"))?;
        let position = options
            .get("cursor_position")
            .map(|value| {
                let position: (f64, f64) = value
                    .try_clone()
                    .map_err(|_| invalid("Invalid cursor position"))?
                    .try_into()
                    .map_err(|_| invalid("Invalid cursor position"))?;
                if !position.0.is_finite() || !position.1.is_finite() {
                    return Err(invalid("Invalid cursor position"));
                }
                Ok(position)
            })
            .transpose()?;
        bridge
            .call(
                "input-capture-release",
                json!({"session":id,"activation_id":activation,"cursor_position":position}),
            )
            .await
            .map_err(failed)?;
        Ok((0, Options::new()))
    }

    #[zbus(signal)]
    async fn activated(
        emitter: &SignalEmitter<'_>,
        session_handle: OwnedObjectPath,
        options: Options,
    ) -> zbus::Result<()>;

    #[zbus(signal)]
    async fn deactivated(
        emitter: &SignalEmitter<'_>,
        session_handle: OwnedObjectPath,
        options: Options,
    ) -> zbus::Result<()>;

    #[zbus(signal)]
    async fn disabled(
        emitter: &SignalEmitter<'_>,
        session_handle: OwnedObjectPath,
        options: Options,
    ) -> zbus::Result<()>;

    #[zbus(signal)]
    async fn zones_changed(
        emitter: &SignalEmitter<'_>,
        session_handle: OwnedObjectPath,
        options: Options,
    ) -> zbus::Result<()>;
}

struct SessionObject {
    backend: InputCapture,
    path: String,
    owner: String,
}

#[zbus::interface(name = "org.freedesktop.impl.portal.Session")]
impl SessionObject {
    #[zbus(property, name = "version")]
    fn version(&self) -> u32 {
        1
    }

    async fn close(
        &self,
        #[zbus(connection)] connection: &Connection,
        #[zbus(header)] header: Header<'_>,
    ) -> zbus::fdo::Result<()> {
        if header.sender().map(|owner| owner.as_str()) != Some(self.owner.as_str()) {
            return Err(zbus::fdo::Error::AccessDenied("Wrong session owner".into()));
        }
        self.backend.end(connection, &self.path).await;
        Ok(())
    }

    #[zbus(signal)]
    async fn closed(emitter: &SignalEmitter<'_>) -> zbus::Result<()>;
}

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

    #[test]
    fn requested_capabilities_are_required_and_supported_subset_is_returned() {
        assert!(capabilities(&Options::new()).is_err());
        for requested in [0u32, 4, 8] {
            assert!(capabilities(&Options::from([("capabilities".into(), requested.into())])).is_err());
        }
        for (requested, supported) in [(1u32, 1), (2, 2), (3, 3), (7, 3), (9, 1)] {
            assert_eq!(
                capabilities(&Options::from([("capabilities".into(), requested.into())])).unwrap(),
                supported
            );
        }
    }
}