use std::sync::Arc;

#[cfg(feature = "wrapped-transport")]
use easytier_core::gateway::proxy::wrapped_transport::WrappedTransportEngines;
#[cfg(feature = "wireguard")]
use easytier_core::gateway::vpn_portal::PortalHost;
#[cfg(test)]
use easytier_core::host::packet::{HostPacket, PacketSink};
#[cfg(feature = "web-client")]
use easytier_core::{
    connectivity::manual::ManualTunnelConnector,
    host::dns::{DnsRecordResolver, DnsResolver},
};
use easytier_core::{
    events::{CoreEvent, CoreEventSink},
    instance::{CoreHostAdapters, CoreInstance, CoreInstanceConfig, PacketEgressHost},
    process_runtime::CoreProcessRuntime,
};

use crate::common::global_ctx::GlobalCtxEvent;
#[cfg(feature = "public-ipv6-provider")]
use crate::instance::public_ipv6_provider::runtime_public_ipv6_provider_platform;
use crate::{
    common::global_ctx::ArcGlobalCtx,
    common::{config::TomlConfig, global_ctx::GlobalCtx},
    host_runtime::native_host_runtime,
    instance::config::{
        compact_runtime_core_host_config, runtime_core_host_config, runtime_peer_credential_storage,
    },
    instance::listeners::RuntimeExternalListenerFactory,
    instance::runtime_host::NativeInstanceRuntimeHost,
};

#[cfg(feature = "tun")]
use super::shared_virtual_nic::ArcSharedVirtualNicRegistry;
use super::{
    host::{NativeInstanceHost, native_instance_host},
    runtime_executor::NativeInstanceExecutor,
};
#[cfg(feature = "kcp")]
use crate::gateway::kcp_proxy::KcpProxyService;
#[cfg(feature = "quic")]
use crate::gateway::quic_proxy::QuicProxyService;
use crate::tunnel::protocol::{runtime_client_protocol_upgrader, runtime_server_protocol_upgrader};
#[cfg(any(feature = "kcp", feature = "quic"))]
use easytier_core::gateway::proxy::wrapped_transport::WrappedTransportEngine;

pub(crate) type NativeCoreInstance = CoreInstance<NativeInstanceHost>;

pub(super) fn compose_native_core_instance(
    toml_config: TomlConfig,
    process_runtime: Arc<CoreProcessRuntime>,
    #[cfg(feature = "tun")] shared_virtual_nic_registry: ArcSharedVirtualNicRegistry,
    compact_runtime: bool,
    executor: Option<NativeInstanceExecutor>,
) -> anyhow::Result<Arc<NativeCoreInstance>> {
    let host_config = if compact_runtime {
        compact_runtime_core_host_config()
    } else {
        runtime_core_host_config()
    };
    let normalized = CoreInstanceConfig::from_toml_with_host(&toml_config, &host_config)?;
    let global_ctx = Arc::new(GlobalCtx::new_with_runtime_config(
        toml_config.clone(),
        &normalized,
        &host_config,
    ));
    let runtime_host = NativeInstanceRuntimeHost::new_with_executor(
        global_ctx.clone(),
        #[cfg(feature = "tun")]
        shared_virtual_nic_registry,
        executor,
    );
    let mut adapters = runtime_core_host_adapters_with_packet_egress_and_config(
        global_ctx.clone(),
        process_runtime,
        runtime_host.clone(),
        host_config,
    );
    adapters.instance_runtime = runtime_host;
    NativeCoreInstance::from_toml(toml_config, adapters)
}

impl CoreEventSink for GlobalCtx {
    fn emit(&self, event: CoreEvent) {
        let event = match event {
            CoreEvent::PeerAdded(peer_id) => GlobalCtxEvent::PeerAdded(peer_id),
            CoreEvent::PeerRemoved(peer_id) => GlobalCtxEvent::PeerRemoved(peer_id),
            CoreEvent::PeerConnAdded(info) => GlobalCtxEvent::PeerConnAdded(info.into()),
            CoreEvent::PeerConnRemoved(info) => GlobalCtxEvent::PeerConnRemoved(info.into()),
            CoreEvent::CredentialChanged => GlobalCtxEvent::CredentialChanged,
            CoreEvent::ManualConnecting { url } => GlobalCtxEvent::Connecting(url),
            CoreEvent::ManualConnectError {
                url,
                ip_version,
                error,
            } => GlobalCtxEvent::ConnectError(url.to_string(), format!("{ip_version:?}"), error),
            CoreEvent::ListenerPlanFailed { url, error } => {
                GlobalCtxEvent::ListenerAddFailed(url, error)
            }
            CoreEvent::ListenerAdded { url, .. } => GlobalCtxEvent::ListenerAdded(url),
            CoreEvent::ListenerRemoved { .. } | CoreEvent::ListenerSocketAccepted { .. } => return,
            CoreEvent::ListenerAddFailed {
                url,
                error,
                will_retry,
                ..
            } => {
                let message = if will_retry {
                    format!("error: {error}, retry listen later...")
                } else {
                    format!("error: {error}")
                };
                GlobalCtxEvent::ListenerAddFailed(url, message)
            }
            CoreEvent::ListenerAcceptFailed { url, error } => GlobalCtxEvent::ListenerAcceptFailed(
                url,
                format!("error: {error}, retry listen later..."),
            ),
            CoreEvent::ListenerAcceptedSocketHandleFailed { url, error } => {
                tracing::error!(%url, %error, "accepted socket handler failed");
                return;
            }
            CoreEvent::TunnelAccepted {
                local_url,
                remote_url,
            } => GlobalCtxEvent::ConnectionAccepted(local_url, remote_url),
            CoreEvent::TunnelAdmissionFailed {
                local_url,
                remote_url,
                error,
            } => GlobalCtxEvent::ConnectionError(local_url, remote_url, error),
            CoreEvent::UdpPortMappingEstablished {
                local_listener,
                mapped_listener,
                backend,
            } => GlobalCtxEvent::ListenerPortMappingEstablished {
                local_listener,
                mapped_listener,
                backend,
            },
            CoreEvent::ProxyCidrsUpdated { added, removed } => {
                GlobalCtxEvent::ProxyCidrsUpdated(added, removed)
            }
            CoreEvent::PublicIpv6LeaseChanged { old, new } => {
                GlobalCtxEvent::PublicIpv6Changed(old, new)
            }
            CoreEvent::PublicIpv6RoutesChanged { added, removed } => {
                GlobalCtxEvent::PublicIpv6RoutesUpdated(added, removed)
            }
            CoreEvent::VpnPortalStarted(portal) => GlobalCtxEvent::VpnPortalStarted(portal),
            CoreEvent::VpnPortalClientConnected { portal, client } => {
                GlobalCtxEvent::VpnPortalClientConnected(portal, client)
            }
            CoreEvent::VpnPortalClientDisconnected { portal, client } => {
                GlobalCtxEvent::VpnPortalClientDisconnected(portal, client)
            }
            CoreEvent::GatewayPortForwardAdded(config) => {
                GlobalCtxEvent::PortForwardAdded(config.into())
            }
        };
        self.issue_event(event);
    }
}

#[cfg(feature = "wrapped-transport")]
fn runtime_wrapped_transport_engines(
    config: &easytier_core::instance::CoreInstanceHostConfig,
) -> WrappedTransportEngines {
    #[cfg(feature = "kcp")]
    let kcp = config
        .kcp_enabled
        .then(|| Arc::new(KcpProxyService::new()) as Arc<dyn WrappedTransportEngine>);
    #[cfg(not(feature = "kcp"))]
    let kcp = None;
    #[cfg(feature = "quic")]
    let quic = config
        .quic_enabled
        .then(|| Arc::new(QuicProxyService::new()) as Arc<dyn WrappedTransportEngine>);
    #[cfg(not(feature = "quic"))]
    let quic = None;

    WrappedTransportEngines { kcp, quic }
}

#[cfg(test)]
pub(crate) fn runtime_core_host_adapters(
    global_ctx: ArcGlobalCtx,
    process_runtime: Arc<CoreProcessRuntime>,
    packet_sink: Arc<dyn PacketSink>,
) -> CoreHostAdapters<NativeInstanceHost> {
    let host = native_instance_host(global_ctx.clone());
    let runtime_dns = native_host_runtime();
    let adapters = CoreHostAdapters::new(host, runtime_dns, packet_sink, process_runtime);
    configure_runtime_core_host_adapters(global_ctx, adapters, runtime_core_host_config())
}

#[cfg(test)]
pub(crate) fn runtime_core_host_adapters_with_packet_egress(
    global_ctx: ArcGlobalCtx,
    process_runtime: Arc<CoreProcessRuntime>,
    packet_egress: Arc<dyn PacketEgressHost>,
) -> CoreHostAdapters<NativeInstanceHost> {
    let host = native_instance_host(global_ctx.clone());
    let runtime_dns = native_host_runtime();
    let adapters =
        CoreHostAdapters::new_with_packet_egress(host, runtime_dns, packet_egress, process_runtime);
    configure_runtime_core_host_adapters(global_ctx, adapters, runtime_core_host_config())
}

fn runtime_core_host_adapters_with_packet_egress_and_config(
    global_ctx: ArcGlobalCtx,
    process_runtime: Arc<CoreProcessRuntime>,
    packet_egress: Arc<dyn PacketEgressHost>,
    host_config: easytier_core::instance::CoreInstanceHostConfig,
) -> CoreHostAdapters<NativeInstanceHost> {
    let host = native_instance_host(global_ctx.clone());
    let runtime_dns = native_host_runtime();
    let adapters =
        CoreHostAdapters::new_with_packet_egress(host, runtime_dns, packet_egress, process_runtime);
    configure_runtime_core_host_adapters(global_ctx, adapters, host_config)
}

fn configure_runtime_core_host_adapters(
    global_ctx: ArcGlobalCtx,
    mut adapters: CoreHostAdapters<NativeInstanceHost>,
    host_config: easytier_core::instance::CoreInstanceHostConfig,
) -> CoreHostAdapters<NativeInstanceHost> {
    #[cfg(test)]
    adapters.replace_stun_provider(Arc::new(crate::common::stun::MockStunInfoCollector {
        udp_nat_type: crate::proto::common::NatType::Unknown,
    }));
    adapters.config = host_config.clone();
    adapters.credential_storage = (!host_config.ignore_unsupported_config)
        .then(|| runtime_peer_credential_storage(&global_ctx))
        .flatten();
    adapters.events = global_ctx.clone();
    #[cfg(feature = "wrapped-transport")]
    {
        adapters.wrapped_transports = runtime_wrapped_transport_engines(&host_config);
    }
    adapters.protocol = Some(runtime_client_protocol_upgrader(global_ctx.clone()));
    adapters.external_listener_factory = Some(Arc::new(RuntimeExternalListenerFactory));
    adapters.server_protocol = Some(runtime_server_protocol_upgrader(global_ctx.clone()));
    #[cfg(feature = "upnp")]
    if host_config.upnp_enabled {
        adapters.udp_hole_punch_platform = Some(
            crate::instance::udp_hole_punch::runtime_udp_hole_punch_platform(
                global_ctx.net_ns.clone(),
            ),
        );
    }
    #[cfg(feature = "icmp-proxy")]
    {
        adapters.icmp_proxy_host = Some(Arc::new(crate::gateway::icmp_proxy::RuntimeIcmpProxyHost));
    }
    #[cfg(feature = "proxy-cidr-monitor")]
    {
        adapters.proxy_cidr_monitor_enabled = true;
    }
    #[cfg(feature = "public-ipv6-provider")]
    if host_config.public_ipv6_provider_supported {
        adapters.public_ipv6_host = Some(global_ctx.clone());
        adapters.public_ipv6_provider = Some(runtime_public_ipv6_provider_platform(&global_ctx));
    }
    #[cfg(feature = "wireguard")]
    if host_config.vpn_portal_enabled {
        use crate::common::config::ConfigLoader as _;

        if let Some(config) = global_ctx
            .config
            .get_vpn_portal_config()
            .filter(|config| config.enabled != Some(false))
        {
            adapters.vpn_portal = Some(crate::vpn_portal::wireguard::WireGuardPortalHost::new(
                global_ctx.clone(),
                config,
            ) as Arc<dyn PortalHost>);
        }
    }
    adapters
}

#[cfg(feature = "web-client")]
pub(crate) fn runtime_one_shot_manual_connector(
    global_ctx: ArcGlobalCtx,
    config: &TomlConfig,
    process_runtime: Arc<CoreProcessRuntime>,
) -> anyhow::Result<ManualTunnelConnector<NativeInstanceHost>> {
    let normalized = CoreInstanceConfig::from_toml_with_host(config, &runtime_core_host_config())?;
    let host = native_instance_host(global_ctx.clone());
    let runtime_dns = native_host_runtime();
    let dns: Arc<dyn DnsResolver> = runtime_dns.clone();
    let dns_records: Arc<dyn DnsRecordResolver> = runtime_dns;
    Ok(process_runtime.manual_connector(
        host,
        dns,
        dns_records,
        runtime_client_protocol_upgrader(global_ctx.clone()),
        normalized.connectivity.endpoint_discovery,
        normalized.connectivity.manual,
    ))
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    #[cfg(feature = "kcp")]
    use easytier_core::gateway::proxy::wrapped_transport::{
        WrappedTransportConnect, WrappedTransportEngine,
    };
    use easytier_core::listener::plan::ListenerRuntimeConfig;
    use smoltcp::wire::{IpAddress, IpProtocol, Ipv4Packet, UdpPacket};
    #[cfg(feature = "kcp")]
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    #[cfg(feature = "kcp")]
    use crate::gateway::kcp_proxy::KcpProxyService;
    use crate::{
        common::{config::NetworkIdentity, global_ctx::tests::get_mock_global_ctx_with_network},
        instance::config::test_core_instance_config,
    };

    use super::*;

    fn create_host_packet_channel() -> (
        tokio::sync::mpsc::Sender<Vec<u8>>,
        tokio::sync::mpsc::Receiver<Vec<u8>>,
    ) {
        tokio::sync::mpsc::channel(16)
    }

    #[cfg(feature = "kcp")]
    fn build_native_kcp_test_instance(
        global_ctx: ArcGlobalCtx,
        packet_sink: tokio::sync::mpsc::Sender<Vec<u8>>,
        listeners: Option<ListenerRuntimeConfig>,
    ) -> anyhow::Result<(Arc<NativeCoreInstance>, Arc<KcpProxyService>)> {
        let mut adapters = runtime_core_host_adapters(
            global_ctx.clone(),
            CoreProcessRuntime::new(),
            Arc::new(packet_sink),
        );
        adapters.proxy_cidr_monitor_enabled = false;
        let service = Arc::new(KcpProxyService::new());
        adapters.wrapped_transports = WrappedTransportEngines {
            kcp: Some(service.clone()),
            quic: None,
        };

        let mut config = test_core_instance_config(&global_ctx);
        config.connectivity.listeners = listeners;
        config.connectivity.startup_plan.gateway = false;
        config.connectivity.stun.udp_servers.clear();
        config.connectivity.stun.tcp_servers.clear();
        config.connectivity.stun.udp_v6_servers.clear();
        config.connectivity.manual = Default::default();
        config.connectivity.direct.testing = true;

        let instance = NativeCoreInstance::new(config, adapters)?;
        Ok((instance, service))
    }

    #[cfg(feature = "kcp")]
    #[tokio::test]
    async fn native_kcp_engine_round_trips_through_portable_cores() {
        tokio::time::timeout(std::time::Duration::from_secs(20), async {
            const REQUEST: &[u8] = b"native-kcp-request";
            const REPLY: &[u8] = b"native-kcp-reply";

            let global_a = get_mock_global_ctx_with_network(Some(NetworkIdentity::new(
                "native-kcp-round-trip".to_owned(),
                "shared-secret".to_owned(),
            )));
            let global_b = get_mock_global_ctx_with_network(Some(NetworkIdentity::new(
                "native-kcp-round-trip".to_owned(),
                "shared-secret".to_owned(),
            )));
            global_a.set_ipv4(Some("10.250.0.1/24".parse().unwrap()));
            global_b.set_ipv4(Some("10.250.0.2/24".parse().unwrap()));

            let mut flags_a = global_a.get_flags();
            flags_a.enable_kcp_proxy = true;
            flags_a.disable_kcp_input = true;
            flags_a.disable_tcp_hole_punching = true;
            flags_a.disable_udp_hole_punching = true;
            flags_a.disable_sym_hole_punching = true;
            flags_a.disable_upnp = true;
            global_a.set_flags(flags_a);

            let mut flags_b = global_b.get_flags();
            flags_b.enable_kcp_proxy = false;
            flags_b.disable_kcp_input = false;
            flags_b.disable_tcp_hole_punching = true;
            flags_b.disable_udp_hole_punching = true;
            flags_b.disable_sym_hole_punching = true;
            flags_b.disable_upnp = true;
            global_b.set_flags(flags_b);

            let (packet_sink_a, _packet_receiver_a) = create_host_packet_channel();
            let (packet_sink_b, _packet_receiver_b) = create_host_packet_channel();
            let (instance_a, kcp_a) = build_native_kcp_test_instance(
                global_a.clone(),
                packet_sink_a,
                Some(ListenerRuntimeConfig::new(
                    vec!["tcp://127.0.0.1:0".parse().unwrap()],
                    false,
                    test_core_instance_config(&global_a)
                        .connectivity
                        .direct
                        .tcp_bind
                        .context,
                )),
            )
            .unwrap();
            let (instance_b, _kcp_b) =
                build_native_kcp_test_instance(global_b, packet_sink_b, None).unwrap();

            let (start_a, start_b) = tokio::join!(instance_a.start(), instance_b.start());
            start_a.unwrap();
            start_b.unwrap();

            let listener = instance_a.running_listeners().pop().unwrap();
            instance_b.add_connector(listener).unwrap();
            let peer_a_id = instance_a.peer_id();
            let peer_b_id = instance_b.peer_id();
            loop {
                let a_peers = instance_a.connected_peers().await;
                let b_peers = instance_b.connected_peers().await;
                if a_peers.contains(&peer_b_id) && b_peers.contains(&peer_a_id) {
                    break;
                }
                tokio::time::sleep(std::time::Duration::from_millis(20)).await;
            }

            let echo_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let echo_addr = echo_listener.local_addr().unwrap();
            let responder = tokio::spawn(async move {
                let (mut socket, _) = echo_listener.accept().await.unwrap();
                let mut request = [0; REQUEST.len()];
                socket.read_exact(&mut request).await.unwrap();
                assert_eq!(&request, REQUEST);
                socket.write_all(REPLY).await.unwrap();
            });

            let mut stream = kcp_a
                .connect_source(WrappedTransportConnect {
                    my_peer_id: peer_a_id,
                    dst_peer_id: peer_b_id,
                    src: "10.250.0.1:40000".parse().unwrap(),
                    dst: echo_addr,
                })
                .await
                .unwrap();
            stream.write_all(REQUEST).await.unwrap();
            let mut reply = [0; REPLY.len()];
            stream.read_exact(&mut reply).await.unwrap();
            assert_eq!(&reply, REPLY);
            responder.await.unwrap();

            instance_b.stop().await;
            instance_a.stop().await;
        })
        .await
        .expect("native KCP round trip timed out");
    }

    #[tokio::test]
    async fn portable_core_instances_connect_through_core_tcp_listener() {
        let global_a = get_mock_global_ctx_with_network(Some(NetworkIdentity::new(
            "portable-connect-listen".to_owned(),
            "shared-secret".to_owned(),
        )));
        let global_b = get_mock_global_ctx_with_network(Some(NetworkIdentity::new(
            "portable-connect-listen".to_owned(),
            "shared-secret".to_owned(),
        )));
        global_a.set_ipv4(Some("10.250.0.1/24".parse().unwrap()));
        global_b.set_ipv4(Some("10.250.0.2/24".parse().unwrap()));
        let (packet_sink_a, _packet_receiver_a) = create_host_packet_channel();
        let (packet_sink_b, mut packet_receiver_b) = create_host_packet_channel();
        let mut config_a = test_core_instance_config(&global_a);
        config_a.connectivity.initial_peers.clear();
        config_a.connectivity.listeners = Some(ListenerRuntimeConfig::new(
            vec!["tcp://127.0.0.1:0".parse().unwrap()],
            false,
            config_a.connectivity.direct.tcp_bind.context.clone(),
        ));
        config_a.connectivity.runtime = Default::default();
        config_a.connectivity.stun.udp_servers.clear();
        config_a.connectivity.stun.tcp_servers.clear();
        config_a.connectivity.stun.udp_v6_servers.clear();
        config_a.connectivity.manual = Default::default();
        config_a.connectivity.direct.testing = true;
        let instance_a = NativeCoreInstance::new(
            config_a,
            runtime_core_host_adapters(
                global_a.clone(),
                CoreProcessRuntime::new(),
                Arc::new(packet_sink_a),
            ),
        )
        .unwrap();

        let mut config_b = test_core_instance_config(&global_b);
        config_b.connectivity.initial_peers.clear();
        config_b.connectivity.listeners = None;
        config_b.connectivity.runtime = Default::default();
        config_b.connectivity.stun.udp_servers.clear();
        config_b.connectivity.stun.tcp_servers.clear();
        config_b.connectivity.stun.udp_v6_servers.clear();
        config_b.connectivity.manual = Default::default();
        config_b.connectivity.direct.testing = true;
        let instance_b = NativeCoreInstance::new(
            config_b,
            runtime_core_host_adapters(
                global_b.clone(),
                CoreProcessRuntime::new(),
                Arc::new(packet_sink_b),
            ),
        )
        .unwrap();

        let (start_a, start_b) = tokio::join!(instance_a.start(), instance_b.start());
        start_a.unwrap();
        start_b.unwrap();
        let listener = instance_a.running_listeners().pop().unwrap();
        instance_b.add_connector(listener).unwrap();

        let peer_a_id = instance_a.peer_id();
        let peer_b_id = instance_b.peer_id();
        tokio::time::timeout(std::time::Duration::from_secs(10), async {
            loop {
                let a_peers = instance_a.connected_peers().await;
                let b_peers = instance_b.connected_peers().await;
                if a_peers.contains(&peer_b_id) && b_peers.contains(&peer_a_id) {
                    break;
                }
                tokio::time::sleep(std::time::Duration::from_millis(20)).await;
            }
        })
        .await
        .expect("portable core instances did not connect through the core listener");

        let source_ip = "10.250.0.1".parse().unwrap();
        let destination_ip = "10.250.0.2".parse().unwrap();
        let mut ip_packet = vec![0u8; 28];
        {
            let mut ipv4 = Ipv4Packet::new_unchecked(&mut ip_packet);
            ipv4.set_version(4);
            ipv4.set_header_len(20);
            ipv4.set_total_len(28);
            ipv4.set_hop_limit(64);
            ipv4.set_next_header(IpProtocol::Udp);
            ipv4.set_src_addr(source_ip);
            ipv4.set_dst_addr(destination_ip);
        }
        {
            let mut udp = UdpPacket::new_unchecked(&mut ip_packet[20..]);
            udp.set_src_port(10000);
            udp.set_dst_port(10001);
            udp.set_len(8);
            udp.fill_checksum(
                &IpAddress::Ipv4(source_ip),
                &IpAddress::Ipv4(destination_ip),
            );
        }
        Ipv4Packet::new_unchecked(&mut ip_packet).fill_checksum();
        let received = tokio::time::timeout(std::time::Duration::from_secs(10), async {
            loop {
                instance_a
                    .packet_plane()
                    .send_ip_packet(HostPacket::copy_from_payload(&ip_packet))
                    .await
                    .unwrap();
                match tokio::time::timeout(
                    std::time::Duration::from_millis(100),
                    packet_receiver_b.recv(),
                )
                .await
                {
                    Ok(Some(packet)) => break packet,
                    Ok(None) => panic!("portable host packet sink closed"),
                    Err(_) => {}
                }
            }
        })
        .await
        .expect("portable host packet sink did not receive the IP packet");
        assert_eq!(received, ip_packet);

        instance_b.stop().await;
        instance_a.stop().await;
    }
}