use std::{
    borrow::Cow,
    cell::RefCell,
    collections::HashMap,
    fmt,
    os::fd::OwnedFd,
    sync::atomic::{AtomicUsize, Ordering},
    time::Duration,
};

use anyhow::{Context, Result, anyhow, bail};
use bitflags::bitflags;
use futures_channel::oneshot;
use futures_util::future::{self, Either};
use gio::prelude::*;
use glib::{
    self,
    variant::{Handle, ObjectPath},
};

const DESKTOP_BUS_NAME: &str = "org.freedesktop.portal.Desktop";
const DESKTOP_OBJECT_PATH: &str = "/org/freedesktop/portal/desktop";
const SESSION_IFACE_NAME: &str = "org.freedesktop.portal.Session";
const REQUEST_IFACE_NAME: &str = "org.freedesktop.portal.Request";
const SCREENCAST_IFACE_NAME: &str = "org.freedesktop.portal.ScreenCast";
const PROXY_CALL_TIMEOUT: Duration = Duration::from_secs(10);
const REQUEST_CLOSE_TIMEOUT: Duration = Duration::from_secs(2);

bitflags! {
    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
    pub struct CursorMode: u32 {
        const HIDDEN = 1;
        const EMBEDDED = 2;
        const METADATA = 4;
    }
}

bitflags! {
    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
    pub struct SourceType: u32 {
        const MONITOR = 1;
        const WINDOW = 2;
        const VIRTUAL = 4;
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u32)]
pub enum PersistMode {
    None = 0,
}

#[derive(Debug, Clone, Copy)]
pub struct StartOptions {
    pub cursor_mode: CursorMode,
    pub source_type: SourceType,
    pub multiple: bool,
}

#[derive(Debug)]
pub struct PortalCapture {
    pub session: Session,
    pub streams: Vec<Stream>,
    pub fd: OwnedFd,
}

#[derive(Debug, Clone)]
pub struct Stream {
    node_id: u32,
    position: Option<(i32, i32)>,
    size: Option<(i32, i32)>,
}

impl Stream {
    pub fn node_id(&self) -> u32 {
        self.node_id
    }

    pub fn position(&self) -> Option<(i32, i32)> {
        self.position
    }

    pub fn size(&self) -> Option<(i32, i32)> {
        self.size
    }
}

type StreamVariantType = (u32, VariantDict);

impl StaticVariantType for Stream {
    fn static_variant_type() -> Cow<'static, glib::VariantTy> {
        <StreamVariantType>::static_variant_type()
    }
}

impl FromVariant for Stream {
    fn from_variant(variant: &glib::Variant) -> Option<Self> {
        let (node_id, props) = variant.get::<StreamVariantType>()?;
        Some(Self {
            node_id,
            position: props.get_flatten("position").ok(),
            size: props.get_flatten("size").ok(),
        })
    }
}

pub async fn start_screencast(options: StartOptions) -> Result<PortalCapture> {
    let proxy = Proxy::new()
        .await
        .context("failed to connect to xdg-desktop-portal screencast API")?;
    let cursor_mode = proxy.supported_cursor_mode(options.cursor_mode)?;
    if cursor_mode != options.cursor_mode {
        eprintln!(
            "warning: portal does not support {} cursor mode; using {} cursor mode",
            cursor_mode_name(options.cursor_mode),
            cursor_mode_name(cursor_mode),
        );
    }

    let session = proxy
        .create_session()
        .await
        .context("failed to create portal screencast session")?;

    let setup_result = async {
        session
            .select_sources(
                options.source_type,
                options.multiple,
                cursor_mode,
                PersistMode::None,
            )
            .await
            .context("failed to select portal screencast sources")?;

        let streams = session
            .start()
            .await
            .context("failed to start portal screencast session")?;

        if streams.is_empty() {
            bail!("portal returned no streams");
        }

        let fd = session
            .open_pipe_wire_remote()
            .await
            .context("failed to open portal PipeWire remote")?;

        Ok((streams, fd))
    }
    .await;

    let (streams, fd) = match setup_result {
        Ok(capture) => capture,
        Err(err) => {
            if let Err(close_err) = session.close().await {
                eprintln!("warning: failed to close failed portal session: {close_err:#}");
            }
            return Err(err);
        }
    };

    Ok(PortalCapture {
        session,
        streams,
        fd,
    })
}

struct Proxy(gio::DBusProxy);

impl Proxy {
    async fn new() -> Result<Self> {
        let proxy = gio::DBusProxy::for_bus_future(
            gio::BusType::Session,
            gio::DBusProxyFlags::NONE,
            None,
            DESKTOP_BUS_NAME,
            DESKTOP_OBJECT_PATH,
            SCREENCAST_IFACE_NAME,
        )
        .await?;

        Ok(Self(proxy))
    }

    async fn create_session(&self) -> Result<Session> {
        let session_handle_token = HandleToken::new();
        let handle_token = HandleToken::new();
        let options = VariantDict::builder()
            .entry("handle_token", &handle_token)
            .entry("session_handle_token", &session_handle_token)
            .build();

        let response =
            screencast_request_call(&self.0, &handle_token, "CreateSession", &(options,)).await?;
        let session_handle = response.get_flatten::<String>("session_handle")?;

        Ok(Session {
            proxy: self.0.clone(),
            session_handle: ObjectPath::try_from(session_handle)?,
        })
    }

    fn supported_cursor_mode(&self, requested: CursorMode) -> Result<CursorMode> {
        let Some(available) = self.available_cursor_modes()? else {
            return Ok(requested);
        };

        choose_cursor_mode(requested, available)
    }

    fn available_cursor_modes(&self) -> Result<Option<CursorMode>> {
        let Some(variant) = self.0.cached_property("AvailableCursorModes") else {
            return Ok(None);
        };
        let raw = variant.get::<u32>().ok_or_else(|| {
            anyhow!(
                "portal property `AvailableCursorModes` expected type `u`, got `{}` with value `{}`",
                variant.type_(),
                variant
            )
        })?;

        Ok(Some(CursorMode::from_bits_truncate(raw)))
    }
}

fn choose_cursor_mode(requested: CursorMode, available: CursorMode) -> Result<CursorMode> {
    if available.contains(requested) {
        return Ok(requested);
    }

    if requested == CursorMode::EMBEDDED {
        if available.contains(CursorMode::METADATA) {
            return Ok(CursorMode::METADATA);
        }
        if available.contains(CursorMode::HIDDEN) {
            return Ok(CursorMode::HIDDEN);
        }
    }

    bail!(
        "portal does not support {} cursor mode; available cursor modes: {}",
        cursor_mode_name(requested),
        available_cursor_mode_names(available)
    )
}

fn cursor_mode_name(mode: CursorMode) -> &'static str {
    if mode == CursorMode::HIDDEN {
        "hidden"
    } else if mode == CursorMode::EMBEDDED {
        "embedded"
    } else if mode == CursorMode::METADATA {
        "metadata"
    } else {
        "unknown"
    }
}

fn available_cursor_mode_names(available: CursorMode) -> String {
    let mut names = Vec::new();
    for mode in [
        CursorMode::HIDDEN,
        CursorMode::EMBEDDED,
        CursorMode::METADATA,
    ] {
        if available.contains(mode) {
            names.push(cursor_mode_name(mode));
        }
    }

    if names.is_empty() {
        "none".to_string()
    } else {
        names.join(", ")
    }
}

#[derive(Debug)]
pub struct Session {
    proxy: gio::DBusProxy,
    session_handle: ObjectPath,
}

impl Session {
    async fn select_sources(
        &self,
        source_type: SourceType,
        multiple: bool,
        cursor_mode: CursorMode,
        persist_mode: PersistMode,
    ) -> Result<()> {
        let handle_token = HandleToken::new();
        let options = VariantDict::builder()
            .entry("handle_token", &handle_token)
            .entry("types", source_type.bits())
            .entry("multiple", multiple)
            .entry("cursor_mode", cursor_mode.bits())
            .entry("persist_mode", persist_mode as u32)
            .build();

        let response = screencast_request_call(
            &self.proxy,
            &handle_token,
            "SelectSources",
            &(&self.session_handle, options),
        )
        .await?;
        debug_assert!(response.is_empty());

        Ok(())
    }

    async fn start(&self) -> Result<Vec<Stream>> {
        let handle_token = HandleToken::new();
        let options = VariantDict::builder()
            .entry("handle_token", &handle_token)
            .build();
        let response = screencast_request_call(
            &self.proxy,
            &handle_token,
            "Start",
            &(&self.session_handle, "", options),
        )
        .await?;

        response.get_flatten("streams")
    }

    async fn open_pipe_wire_remote(&self) -> Result<OwnedFd> {
        let (response, fd_list) = self
            .proxy
            .call_with_unix_fd_list_future(
                "OpenPipeWireRemote",
                Some(&(&self.session_handle, VariantDict::default()).to_variant()),
                gio::DBusCallFlags::NONE,
                PROXY_CALL_TIMEOUT.as_millis() as i32,
                gio::UnixFDList::NONE,
            )
            .await?;
        let fd_list = fd_list.context("portal did not return a Unix FD list")?;
        let (fd_index,) = variant_get::<(Handle,)>(&response)?;
        fd_list
            .get(fd_index.0)
            .with_context(|| format!("failed to get PipeWire FD at index {}", fd_index.0))
    }

    pub async fn close(self) -> Result<()> {
        let response = self
            .proxy
            .connection()
            .call_future(
                Some(DESKTOP_BUS_NAME),
                self.session_handle.as_str(),
                SESSION_IFACE_NAME,
                "Close",
                None,
                None,
                gio::DBusCallFlags::NONE,
                PROXY_CALL_TIMEOUT.as_millis() as i32,
            )
            .await
            .context("failed to close portal screencast session")?;
        variant_get::<()>(&response)?;
        Ok(())
    }
}

async fn screencast_request_call(
    proxy: &gio::DBusProxy,
    handle_token: &HandleToken,
    method: &str,
    params: impl ToVariant,
) -> Result<VariantDict> {
    let connection = proxy.connection();
    let unique_identifier = connection
        .unique_name()
        .context("D-Bus connection has no unique name")?
        .trim_start_matches(':')
        .replace('.', "_");
    let request_path = ObjectPath::try_from(
        format!(
            "/org/freedesktop/portal/desktop/request/{}/{}",
            unique_identifier,
            handle_token.as_str()
        )
        .as_str(),
    )?;

    let request_proxy = gio::DBusProxy::for_bus_future(
        gio::BusType::Session,
        gio::DBusProxyFlags::DO_NOT_AUTO_START
            | gio::DBusProxyFlags::DO_NOT_CONNECT_SIGNALS
            | gio::DBusProxyFlags::DO_NOT_LOAD_PROPERTIES,
        None,
        DESKTOP_BUS_NAME,
        request_path.as_str(),
        REQUEST_IFACE_NAME,
    )
    .await?;

    let (owner_lost_tx, owner_lost_rx) = oneshot::channel();
    let owner_lost_tx = RefCell::new(Some(owner_lost_tx));
    let handler_id = request_proxy.connect_notify_local(Some("g-name-owner"), move |proxy, _| {
        if proxy.name_owner().is_none()
            && let Some(tx) = owner_lost_tx.take()
        {
            let _ = tx.send(());
        }
    });

    let (response_tx, response_rx) = oneshot::channel();
    let response_tx = RefCell::new(Some(response_tx));
    let subscription = connection.subscribe_to_signal(
        Some(DESKTOP_BUS_NAME),
        Some(REQUEST_IFACE_NAME),
        Some("Response"),
        Some(request_path.as_str()),
        None,
        gio::DBusSignalFlags::NONE,
        move |signal| {
            if let Some(tx) = response_tx.take() {
                let _ = tx.send(signal.parameters.clone());
            }
        },
    );

    let params = params.to_variant();
    let returned_path = proxy
        .call_future(
            method,
            Some(&params),
            gio::DBusCallFlags::NONE,
            PROXY_CALL_TIMEOUT.as_millis() as i32,
        )
        .await
        .with_context(|| format!("failed to call portal method `{method}`"))?;
    let (returned_path,) = variant_get::<(ObjectPath,)>(&returned_path)?;
    let path_result = ensure_returned_request_path(method, &returned_path, &request_path);
    if let Err(err) = path_result {
        close_portal_request(&request_proxy, method).await;
        request_proxy.disconnect(handler_id);
        drop(subscription);
        return Err(err);
    }

    let response_result = match future::select(
        future::select(response_rx, owner_lost_rx),
        glib::timeout_future(PROXY_CALL_TIMEOUT),
    )
    .await
    {
        Either::Left((Either::Left((Ok(response), _)), _)) => Ok(response),
        Either::Left((Either::Left((Err(_), _)), _)) => {
            Err(anyhow!("portal response sender dropped for `{method}`"))
        }
        Either::Left((Either::Right(_), _)) => {
            Err(anyhow!("lost portal request owner for `{method}`"))
        }
        Either::Right(_) => Err(anyhow!(
            "portal request `{method}` timed out waiting for response"
        )),
    };

    if response_result.is_err() {
        close_portal_request(&request_proxy, method).await;
    }

    request_proxy.disconnect(handler_id);
    drop(subscription);

    let response = response_result?;

    let (response_no, response) = variant_get::<(u32, VariantDict)>(&response)?;
    match response_no {
        0 => Ok(response),
        1 => bail!("portal request `{method}` was cancelled by the user"),
        2 => bail!("portal request `{method}` ended unexpectedly with {response:?}"),
        other => bail!("portal request `{method}` returned unknown response code {other}"),
    }
}

async fn close_portal_request(request_proxy: &gio::DBusProxy, method: &str) {
    if let Err(err) = request_proxy
        .call_future(
            "Close",
            None,
            gio::DBusCallFlags::NONE,
            REQUEST_CLOSE_TIMEOUT.as_millis() as i32,
        )
        .await
    {
        eprintln!("warning: failed to close timed-out portal request `{method}`: {err:#}");
    }
}

fn ensure_returned_request_path(
    method: &str,
    returned_path: &ObjectPath,
    expected_path: &ObjectPath,
) -> Result<()> {
    if returned_path == expected_path {
        return Ok(());
    }

    bail!(
        "portal method `{method}` returned request path `{}`, expected `{}`",
        returned_path.as_str(),
        expected_path.as_str()
    )
}

fn variant_get<T: FromVariant>(variant: &glib::Variant) -> Result<T> {
    variant.get::<T>().ok_or_else(|| {
        anyhow!(
            "expected variant type `{}`, got `{}` with value `{}`",
            T::static_variant_type(),
            variant.type_(),
            variant
        )
    })
}

static HANDLE_COUNTER: AtomicUsize = AtomicUsize::new(0);

#[derive(Debug)]
struct HandleToken(String);

impl HandleToken {
    fn new() -> Self {
        Self(format!(
            "ocula_{}",
            HANDLE_COUNTER.fetch_add(1, Ordering::Relaxed)
        ))
    }

    fn as_str(&self) -> &str {
        &self.0
    }
}

impl ToVariant for &HandleToken {
    fn to_variant(&self) -> glib::Variant {
        self.0.to_variant()
    }
}

#[derive(Default)]
struct VariantDict(HashMap<String, glib::Variant>);

impl VariantDict {
    fn builder() -> VariantDictBuilder {
        VariantDictBuilder {
            values: HashMap::new(),
        }
    }

    fn is_empty(&self) -> bool {
        self.0.is_empty()
    }

    fn get_flatten<T: FromVariant>(&self, key: &str) -> Result<T> {
        let variant = self
            .0
            .get(key)
            .ok_or_else(|| anyhow!("portal response key `{key}` is missing"))?;
        variant.get::<T>().ok_or_else(|| {
            anyhow!(
                "portal response key `{key}` expected type `{}`, got `{}` with value `{}`",
                T::static_variant_type(),
                variant.type_(),
                variant
            )
        })
    }
}

impl fmt::Debug for VariantDict {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt::Debug::fmt(&self.to_variant(), f)
    }
}

impl StaticVariantType for VariantDict {
    fn static_variant_type() -> Cow<'static, glib::VariantTy> {
        glib::VariantDict::static_variant_type()
    }
}

impl FromVariant for VariantDict {
    fn from_variant(value: &glib::Variant) -> Option<Self> {
        Some(Self(value.get::<HashMap<String, glib::Variant>>()?))
    }
}

impl ToVariant for VariantDict {
    fn to_variant(&self) -> glib::Variant {
        self.0.to_variant()
    }
}

struct VariantDictBuilder {
    values: HashMap<String, glib::Variant>,
}

impl VariantDictBuilder {
    fn entry(mut self, key: &str, value: impl ToVariant) -> Self {
        self.values.insert(key.to_string(), value.to_variant());
        self
    }

    fn build(self) -> VariantDict {
        VariantDict(self.values)
    }
}

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

    #[test]
    fn handle_tokens_are_unique() {
        let a = HandleToken::new();
        let b = HandleToken::new();
        assert_ne!(a.as_str(), b.as_str());
    }

    #[test]
    fn stream_variant_type_matches_portal_api() {
        assert_eq!(
            Stream::static_variant_type(),
            glib::VariantTy::new("(ua{sv})").unwrap()
        );
    }

    #[test]
    fn returned_request_path_must_match_subscription_path() {
        let expected =
            ObjectPath::try_from("/org/freedesktop/portal/desktop/request/ocula/a").unwrap();
        let same = ObjectPath::try_from("/org/freedesktop/portal/desktop/request/ocula/a").unwrap();
        let other =
            ObjectPath::try_from("/org/freedesktop/portal/desktop/request/ocula/b").unwrap();

        assert!(ensure_returned_request_path("Start", &same, &expected).is_ok());
        assert!(ensure_returned_request_path("Start", &other, &expected).is_err());
    }

    #[test]
    fn cursor_mode_uses_requested_mode_when_available() {
        assert_eq!(
            choose_cursor_mode(
                CursorMode::EMBEDDED,
                CursorMode::HIDDEN | CursorMode::EMBEDDED,
            )
            .unwrap(),
            CursorMode::EMBEDDED
        );
        assert_eq!(
            choose_cursor_mode(
                CursorMode::HIDDEN,
                CursorMode::HIDDEN | CursorMode::METADATA
            )
            .unwrap(),
            CursorMode::HIDDEN
        );
    }

    #[test]
    fn embedded_cursor_mode_falls_back_to_supported_modes() {
        assert_eq!(
            choose_cursor_mode(CursorMode::EMBEDDED, CursorMode::METADATA).unwrap(),
            CursorMode::METADATA
        );
        assert_eq!(
            choose_cursor_mode(CursorMode::EMBEDDED, CursorMode::HIDDEN).unwrap(),
            CursorMode::HIDDEN
        );
    }

    #[test]
    fn hidden_cursor_mode_is_not_silently_replaced() {
        let err = choose_cursor_mode(CursorMode::HIDDEN, CursorMode::EMBEDDED)
            .unwrap_err()
            .to_string();

        assert!(err.contains("hidden cursor mode"));
        assert!(err.contains("embedded"));
    }

    #[test]
    fn available_cursor_mode_names_are_human_readable() {
        assert_eq!(available_cursor_mode_names(CursorMode::empty()), "none");
        assert_eq!(
            available_cursor_mode_names(CursorMode::HIDDEN | CursorMode::METADATA),
            "hidden, metadata"
        );
    }
}
