use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;

use eyre::Context;
use mm1_address::address::Address;
use mm1_common::log::{info, warn};
use mm1_common::types::{AnyError, Never};
use mm1_core::envelope::dispatch;
use mm1_core::tracing::WithTraceIdExt;
use mm1_proto::message;
use mm1_proto_network_management::protocols::ProtocolResolved;
use mm1_proto_network_management::{self as nm, protocols};
use mm1_proto_sup::uniform as uni_sup;
use mm1_proto_system as sys;
use mm1_proto_well_known::MULTINODE_MANAGER;
use mm1_timer::v1::OneshotTimer;
use tokio::net::TcpStream;

use crate::actors::context::ActorContext;
use crate::codec::Protocol;

const RECONNECT_INTERVAL: Duration = Duration::from_secs(1);
const PROTOCOL_WAIT_TIMEOUT: Duration = Duration::from_secs(60);
const CONNECTION_START_TIMEOUT: Duration = Duration::from_secs(5);

pub async fn run<Ctx>(
    ctx: &mut Ctx,
    connection_sup: Address,
    dst_addr: SocketAddr,
    protocol_names: Vec<nm::ProtocolName>,
    options: nm::Options,
) -> Result<Never, AnyError>
where
    Ctx: ActorContext,
{
    ctx.init_done(ctx.address()).await;

    let mut timer_api = OneshotTimer::create(ctx)
        .await
        .wrap_err("OneshotTimer::create")?;

    let mut protocols = vec![];
    for protocol_name in protocol_names {
        let protocol = wait_for_protocol(ctx, protocol_name)
            .await
            .wrap_err("wait_for_protocol")?;
        protocols.push(protocol);
    }

    ctx.tell(ctx.address(), Connect)
        .await
        .wrap_err("ctx.tell")?;
    event_loop(
        ctx,
        &mut timer_api,
        connection_sup,
        dst_addr,
        Arc::new(options),
        protocols.into_boxed_slice().into(),
    )
    .await
}

async fn event_loop<Ctx>(
    ctx: &mut Ctx,
    timer_api: &mut OneshotTimer<Ctx>,
    connection_sup: Address,
    dst_addr: SocketAddr,
    options: Arc<nm::Options>,
    protocols: Arc<[ProtocolResolved<Protocol>]>,
) -> Result<Never, AnyError>
where
    Ctx: ActorContext,
{
    loop {
        let envelope = ctx.recv().await.wrap_err("ctx.recv")?;
        let trace_id = envelope.header().trace_id();
        dispatch!(match envelope {
            Connect =>
                handle_connect(
                    ctx,
                    timer_api,
                    connection_sup,
                    dst_addr,
                    options.clone(),
                    protocols.clone()
                )
                .with_trace_id(trace_id)
                .await
                .wrap_err("handle_connect")?,

            sys::Down { normal_exit, .. } if *normal_exit =>
                break trace_id
                    .scope_async(async {
                        info!("connection terminated normally [dst: {}]", dst_addr);
                        Ok(ctx.quit_ok().await)
                    })
                    .await,
            sys::Down { normal_exit, .. } => {
                trace_id
                    .scope_async(async {
                        assert!(!normal_exit);

                        warn!(
                            "connection terminated abnormally. Reconnecting in {:?} [dst: {}]",
                            RECONNECT_INTERVAL, dst_addr
                        );
                        timer_api
                            .schedule_once_after(RECONNECT_INTERVAL, Connect)
                            .await
                            .wrap_err("timer_api.schedule_once_after")
                    })
                    .await?;
            },
        })
    }
}

async fn handle_connect<Ctx>(
    ctx: &mut Ctx,
    timer_api: &mut OneshotTimer<Ctx>,
    connection_sup: Address,
    dst_addr: SocketAddr,
    options: Arc<nm::Options>,
    protocols: Arc<[ProtocolResolved<Protocol>]>,
) -> Result<(), AnyError>
where
    Ctx: ActorContext,
{
    info!("connecting to {} from {:?}", dst_addr, dst_addr);

    let tcp_stream = match TcpStream::connect(dst_addr).await {
        Ok(tcp_stream) => tcp_stream,
        Err(reason) => {
            warn!("could not connect to {}: {}", dst_addr, reason);
            timer_api
                .schedule_once_after(RECONNECT_INTERVAL, Connect)
                .await
                .wrap_err("timer_api.schedule_once_after")?;
            return Ok(())
        },
    };

    let connection_addr = ctx
        .fork_ask::<_, uni_sup::StartResponse>(
            connection_sup,
            uni_sup::StartRequest {
                args: (tcp_stream, options, protocols),
            },
            CONNECTION_START_TIMEOUT,
        )
        .await
        .wrap_err("ctx.fork_ask")?
        .wrap_err("uni_sup::Start")?;

    let _watch_ref = ctx.watch(connection_addr).await;

    Ok(())
}

async fn wait_for_protocol<Ctx>(
    ctx: &mut Ctx,
    name: nm::ProtocolName,
) -> Result<ProtocolResolved<Protocol>, AnyError>
where
    Ctx: ActorContext,
{
    let resolved = ctx
        .fork_ask::<_, protocols::GetProtocolByNameResponse<Protocol>>(
            MULTINODE_MANAGER,
            protocols::GetProtocolByNameRequest {
                name,
                timeout: Some(PROTOCOL_WAIT_TIMEOUT),
            },
            PROTOCOL_WAIT_TIMEOUT + Duration::from_secs(1),
        )
        .await
        .wrap_err("fork_ask")?
        .wrap_err("GetProtocolByName")?;

    Ok(resolved)
}

#[message(base_path = ::mm1_proto)]
struct Connect;
