use_prelude!();
use std::{
    any::Any,
    collections::{HashMap, HashSet},
    path::PathBuf,
    sync::Arc,
};

use ffi_sdk::BoxedDitto;

use crate::error::{DittoError, ErrorKind};

pub(crate) mod peers_observer;
pub(crate) mod presence_manager_v2;

/// A configuration object specifying which network transports Ditto should use
/// to sync data.
///
/// A `Ditto` object comes with a default transport configuration where all
/// available peer-to-peer transports are enabled. You can customize this by
/// initializing a `TransportConfig`, adjusting its properties, and supplying it
/// to `set_transport_config()` on `Ditto`.
///
/// When you initialize a `TransportConfig` yourself it starts with all
/// transports disabled. You must enable each one directly.
///
/// Peer-to-peer transports will automatically discover peers in the vicinity
/// and create connections without any configuration. These are configured
/// inside the `peer_to_peer` property. To turn each one on, set its `enabled`
/// property to `true`.
///
/// To connect to a peer at a known location, such as a Ditto Big Peer, add its
/// address inside the `connect` configuration. These are either `host:port`
/// strings for raw TCP sync, or a `wss://…` URL for websockets.
///
/// The `listen` configurations are for specific less common data sync
/// scenarios. Please read the documentation on the Ditto website for examples.
/// Incorrect use of `listen` can result in insecure configurations.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct TransportConfig {
    pub peer_to_peer: PeerToPeer,
    pub connect: Connect,
    pub listen: Listen,
    pub global: Global,
}

impl TransportConfig {
    pub fn new() -> Self {
        Self {
            peer_to_peer: PeerToPeer {
                bluetooth_le: BluetoothConfig::new(),
                lan: LanConfig::new(),
            },
            connect: Connect {
                tcp_servers: HashSet::new(),
                websocket_urls: HashSet::new(),
            },
            listen: Listen {
                tcp: TcpListenConfig::new(),
                http: HttpListenConfig::new(),
            },
            global: Global { sync_group: 0 },
        }
    }

    pub fn enable_all_peer_to_peer(&mut self) {
        self.peer_to_peer.bluetooth_le.enabled = true;
        self.peer_to_peer.lan.enabled = true;
    }
}

#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct PeerToPeer {
    pub bluetooth_le: BluetoothConfig,
    pub lan: LanConfig,
}

#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Connect {
    pub tcp_servers: HashSet<String>,
    pub websocket_urls: HashSet<String>,
}

#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Listen {
    pub tcp: TcpListenConfig,
    pub http: HttpListenConfig,
}

/// Settings not associated with any specific type of transport.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Global {
    /// The sync group for this device.
    ///
    /// When peer-to-peer transports are enabled, all devices with the same App
    /// ID will normally form an interconnected mesh network. In some
    /// situations it may be  desirable to have distinct groups of devices
    /// within the same app, so that connections will only be formed within
    /// each group. The `sync_group` parameter changes that group
    /// membership. A device can only ever be in one sync group, which
    /// by default is group 0. Up to 2^32 distinct group numbers can be used in
    /// an app.
    ///
    /// This is an optimization, not a security control. If a connection is
    /// created manually, such as by specifying a `connect` transport, then
    /// devices from different sync groups will still sync as normal. If
    /// two groups of devices are intended to have access to different data
    /// sets, this must be enforced using  Ditto's permissions system.
    pub sync_group: u32,
}

#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct HttpListenConfig {
    pub enabled: bool,
    pub interface_ip: String,
    pub port: u16,
    pub static_content_path: Option<PathBuf>,
    pub websocket_sync: bool,
    pub tls_key_path: Option<PathBuf>,
    pub tls_certificate_path: Option<PathBuf>,
}

impl HttpListenConfig {
    pub fn new() -> Self {
        Self {
            enabled: false,
            interface_ip: "[::]".to_string(),
            port: 80,
            static_content_path: None,
            websocket_sync: true,
            tls_key_path: None,
            tls_certificate_path: None,
        }
    }
}

#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct TcpListenConfig {
    pub enabled: bool,
    pub interface_ip: String,
    pub port: u16,
}

impl TcpListenConfig {
    pub fn new() -> Self {
        Self {
            enabled: false,
            interface_ip: "[::]".to_string(),
            port: 4040,
        }
    }
}

#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct BluetoothConfig {
    pub enabled: bool,
}

impl BluetoothConfig {
    pub fn new() -> Self {
        Self { enabled: false }
    }
}

#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct LanConfig {
    pub enabled: bool,
    pub multicast_enabled: bool,
}

impl LanConfig {
    pub fn new() -> Self {
        Self {
            enabled: false,
            multicast_enabled: true,
        }
    }
}

/// Handles for the various Transports
/// operating on the other side of the FFI interface
pub struct Transports {
    ditto: Arc<BoxedDitto>,
    config: TransportConfig,
    sync_active: bool,
    web_identity_valid: bool,
    x509_identity_valid: bool,
    tcp_clients: HashMap<String, Box<dyn Any + Send + Sync>>,
    ws_clients: HashMap<String, Box<dyn Any + Send + Sync>>,
    ble_client_transport: Option<Box<dyn Any + Send + Sync>>,
    ble_server_transport: Option<Box<dyn Any + Send + Sync>>,
}

// We make all transport-modifying methods here mutable since either the main
// executable thread or an automated process like the ValidityListener may try
// to change the transports at any time. This forces both to first acquire a
// WriteLock on the Transports struct before mutating the current state of the
// Transports or TransportConfig
impl Transports {
    pub(crate) fn from_config(config: TransportConfig, ditto: Arc<BoxedDitto>) -> Transports {
        let t = Transports {
            ditto,
            config,
            sync_active: false, // off until try_start_sync called
            web_identity_valid: false,
            x509_identity_valid: false,
            tcp_clients: HashMap::with_capacity(0), // no need to allocate until actually configured
            ws_clients: HashMap::with_capacity(0),
            ble_client_transport: None,
            ble_server_transport: None,
        };
        t.apply_transport_global_config(&t.config.global, &TransportConfig::new().global);
        t
    }

    pub(crate) fn try_start_sync(&mut self) -> Result<(), DittoError> {
        if !self.sync_active {
            self.sync_active = true;
            let all_disabled = TransportConfig::new();
            self.apply_transport_config(&self.config.clone(), &all_disabled);
        }
        Ok(())
    }

    pub(crate) fn stop_sync(&mut self) {
        if self.sync_active {
            self.sync_active = false;
            let all_disabled = TransportConfig::new();
            self.apply_transport_config(&all_disabled, &self.config.clone());
        }
    }

    pub(crate) fn set_transport_config(&mut self, config: TransportConfig) {
        if self.sync_active {
            self.apply_transport_config(&config, &self.config.clone());
        }
        self.apply_transport_global_config(&config.global, &self.config.global);
        self.config = config;
    }

    pub(crate) fn current_config(&self) -> &TransportConfig {
        &self.config
    }

    fn apply_transport_config(&mut self, config: &TransportConfig, old_config: &TransportConfig) {
        // Diff the two configs transport-by-transport

        // TCP and LAN are interrelated
        if config.listen.tcp != old_config.listen.tcp
            || config.peer_to_peer.lan != old_config.peer_to_peer.lan
        {
            let lan_stops_server = !old_config.listen.tcp.enabled;
            self.stop_lan(lan_stops_server);
            if config.peer_to_peer.lan.enabled {
                let lan_starts_server = !config.listen.tcp.enabled;
                self.start_lan(lan_starts_server);
            }
        }

        if config.listen.tcp != old_config.listen.tcp {
            self.stop_tcp_listen();
            if config.listen.tcp.enabled {
                self.start_tcp_listen(&config.listen.tcp);
            }
        }

        if config.listen.http != old_config.listen.http {
            self.stop_http_listen();
            if config.listen.http.enabled {
                let _ = self.start_http_listen(&config.listen.http);
            }
        }

        let tcp_connects_to_stop = old_config
            .connect
            .tcp_servers
            .difference(&config.connect.tcp_servers);
        for addr in tcp_connects_to_stop {
            self.stop_tcp_connect(addr);
        }

        let tcp_connects_to_start = config
            .connect
            .tcp_servers
            .difference(&old_config.connect.tcp_servers);
        for addr in tcp_connects_to_start {
            self.start_tcp_connect(addr.clone());
        }

        let ws_connects_to_stop = old_config
            .connect
            .websocket_urls
            .difference(&config.connect.websocket_urls);
        for url in ws_connects_to_stop {
            self.stop_ws_connect(url);
        }

        let ws_connects_to_start = config
            .connect
            .websocket_urls
            .difference(&old_config.connect.websocket_urls);
        for url in ws_connects_to_start {
            self.start_ws_connect(url.clone());
        }

        if config.peer_to_peer.bluetooth_le != old_config.peer_to_peer.bluetooth_le {
            self.stop_bluetooth();
            if config.peer_to_peer.bluetooth_le.enabled {
                self.start_bluetooth();
            }
        }
    }

    fn apply_transport_global_config(&self, config: &Global, old_config: &Global) {
        if config.sync_group != old_config.sync_group {
            unsafe { ffi_sdk::ditto_set_sync_group(&self.ditto, config.sync_group) };
        }
    }

    /// Start a TCP Server which can listen for connections from other Peers
    /// Generally also requires a update to the relevant DNS zone file to be
    /// discoverable by other peers
    fn start_tcp_listen(&mut self, config: &crate::transport::TcpListenConfig) {
        let bind_ip = format!("{}:{}", config.interface_ip, config.port);
        let c_addr = char_p::new(bind_ip);
        // Convert to a public Error type
        let _result =
            unsafe { ffi_sdk::ditto_start_tcp_server(&self.ditto, Some(c_addr.as_ref())) };
    }

    fn stop_tcp_listen(&mut self) {
        unsafe { ffi_sdk::ditto_stop_tcp_server(&self.ditto) };
    }

    /// Starts an HTTP server that other devices will be able to connect to.
    /// * `bind_ip` - IP Address:Port on which to listen for connections
    /// * `enable_websocket` - If true, enable websocket transport
    /// * `static_path` - Optional root of web content directory
    /// * `tls_cert_path` - Optional x509 certificate for web server (also
    ///   requires key)
    /// * `tls_key_path` - Optional TLS private key, required if cert is
    ///   provided
    fn start_http_listen(
        &mut self,
        config: &crate::transport::HttpListenConfig,
    ) -> Result<(), DittoError> {
        let enable_ws = if config.websocket_sync {
            ffi_sdk::WebSocketMode::Enabled
        } else {
            ffi_sdk::WebSocketMode::Disabled
        };

        let bind_ip = format!("{}:{}", config.interface_ip, config.port);
        let c_addr = char_p::new(bind_ip);
        let c_static_path = config
            .static_content_path
            .as_ref()
            .map(|x| char_p::new(x.to_string_lossy().to_string()));
        let c_tls_cert_path = config
            .tls_certificate_path
            .as_ref()
            .map(|x| char_p::new(x.to_string_lossy().to_string()));
        let c_tls_key_path = config
            .tls_key_path
            .as_ref()
            .map(|x| char_p::new(x.to_string_lossy().to_string()));

        let status = unsafe {
            ffi_sdk::ditto_start_http_server(
                &self.ditto,
                Some(c_addr.as_ref()),
                c_static_path.as_ref().map(|x| x.as_ref()),
                enable_ws,
                c_tls_cert_path.as_ref().map(|x| x.as_ref()), // TLS cert path
                c_tls_key_path.as_ref().map(|x| x.as_ref()),  // TLS key path
            )
        };
        if status != 0 {
            Err(DittoError::from_ffi(ErrorKind::InvalidInput))
        } else {
            Ok(())
        }
    }

    fn stop_http_listen(&mut self) {
        unsafe { ffi_sdk::ditto_stop_http_server(&self.ditto) };
    }

    fn start_tcp_connect(&mut self, address: String) {
        let addr = char_p::new(address.clone());
        // this handle stores a tx entangled with an rx across the FFI boundary which
        // will drop if all tx elements drop
        let tcp_client_handle =
            unsafe { ffi_sdk::ditto_add_static_tcp_client(&self.ditto, addr.as_ref()) };
        ::log::info!("Static TCP client transport {:?} started", &address);
        self.tcp_clients
            .insert(address, Box::new(tcp_client_handle));
    }

    fn stop_tcp_connect(&mut self, address: &str) {
        let _ = self.tcp_clients.remove(address);
    }

    fn start_ws_connect(&mut self, url: String) {
        let c_url = char_p::new(url.clone());
        let ws_client_handle =
            unsafe { ffi_sdk::ditto_add_websocket_client(&self.ditto, c_url.as_ref()) };
        ::log::info!("Websocket client transport {:?} started", &url);
        self.ws_clients.insert(url, Box::new(ws_client_handle));
    }

    fn stop_ws_connect(&mut self, url: &str) {
        let _ = self.ws_clients.remove(url);
    }

    fn start_bluetooth(&mut self) {
        // BlueZ cfg guard
        #[cfg(all(target_os = "linux", not(target_env = "musl")))]
        {
            let ble_client_handle =
                unsafe { ffi_sdk::ditto_add_internal_ble_client_transport(&self.ditto) };
            ::log::info!("BLE client transport started");
            self.ble_client_transport = Some(Box::new(ble_client_handle));
            let ble_server_handle =
                unsafe { ffi_sdk::ditto_add_internal_ble_server_transport(&self.ditto) };
            ::log::info!("BLE server transport started");
            self.ble_server_transport = Some(Box::new(ble_server_handle));
        }
        ::log::info!("handling BLE transport")
    }

    fn stop_bluetooth(&mut self) {
        let _to_drop = self.ble_client_transport.take();
        let _to_drop = self.ble_server_transport.take();
    }

    fn start_lan(&mut self, lan_controls_server: bool) {
        unsafe {
            if lan_controls_server {
                ffi_sdk::ditto_start_tcp_server(&self.ditto, None);
            }
            ffi_sdk::ditto_add_multicast_transport(&self.ditto);
        }
    }

    fn stop_lan(&mut self, lan_controls_server: bool) {
        unsafe {
            if lan_controls_server {
                ffi_sdk::ditto_stop_tcp_server(&self.ditto);
            }
            ffi_sdk::ditto_remove_multicast_transport(&self.ditto);
        }
    }

    pub(crate) fn validity_updated(&mut self, web_valid: bool, x509_valid: bool) {
        self.apply_validity_changed(
            web_valid,
            self.web_identity_valid,
            x509_valid,
            self.x509_identity_valid,
        );
        self.web_identity_valid = web_valid;
        self.x509_identity_valid = x509_valid;
    }

    fn apply_validity_changed(&mut self, web: bool, old_web: bool, x509: bool, old_x509: bool) {
        if web && !old_web {
            // copy the URLs so we don't hold a reference to the list while modifying Ditto
            let urls = self.config.connect.websocket_urls.clone();
            for url in urls {
                self.start_ws_connect(url);
            }
        }

        if old_web && !web {
            log::debug!("Web Auth has become invalid, shutting down WS Transport");
            let urls = self.config.connect.websocket_urls.clone();
            for url in urls {
                self.stop_ws_connect(&url);
            }
        }
        if x509 && !old_x509 {
            if self.config.peer_to_peer.bluetooth_le.enabled {
                self.start_bluetooth()
            }
            if self.config.listen.tcp.enabled {
                self.start_tcp_listen(&self.config.listen.tcp.clone())
            }
            let addrs = self.config.connect.tcp_servers.clone();
            for addr in addrs {
                self.start_tcp_connect(addr)
            }
        }
        if old_x509 && !x509 {
            log::debug!("BLE Transport shutting down due to invalid x509 certificate");
            self.stop_bluetooth();
            log::debug!("TCP Transport shutting down due to invalid x509 certificate");
            self.stop_tcp_listen();
            let addrs = self.config.connect.tcp_servers.clone();
            for addr in addrs {
                self.stop_tcp_connect(&addr);
            }
        }
    }
}