use std::collections::HashMap;
use std::time::Instant;

use tokio::sync::mpsc;
use tokio::time::{MissedTickBehavior, interval};

use super::io::{Input, Output};
use super::state::Session;
use crate::SocketHub;
use crate::engine::{self, Packet as EnginePacket};

pub(crate) async fn run(
    hub: SocketHub,
    sid: String,
    user_id: i64,
    access_token: String,
    input: mpsc::Sender<Input>,
    mut receiver: mpsc::Receiver<Input>,
    output: Output,
    send_open: bool,
) {
    let options = hub.options().clone();
    let mut shutdown = hub.shutdown_receiver();
    let mut session = Session {
        hub,
        sid,
        user_id,
        access_token,
        device_id: String::new(),
        deliveries: HashMap::new(),
        input,
        output,
        namespaces: HashMap::new(),
        pending_binary: None,
        pending_acks: HashMap::new(),
        next_ack: 0,
        last_pong: Instant::now(),
        last_actor_heartbeat: Instant::now(),
        max_payload: options.max_payload,
        auth_deadline: (user_id == 0).then(|| tokio::time::Instant::now() + options.auth_timeout),
    };

    if send_open {
        let open = engine::open(
            &session.sid,
            options.heartbeat_interval,
            options.heartbeat_timeout,
            options.max_payload,
            &[],
        );
        if !session.output.send(vec![open]).await {
            session.cleanup().await;
            return;
        }
    }

    let mut heartbeat = interval(options.heartbeat_interval);
    heartbeat.set_missed_tick_behavior(MissedTickBehavior::Delay);
    heartbeat.tick().await;

    loop {
        if *shutdown.borrow() {
            break;
        }
        let auth_deadline = session.auth_deadline;
        let ack_deadline = session
            .pending_acks
            .values()
            .map(|(_, deadline, _)| *deadline)
            .min();
        tokio::select! {
            _ = shutdown.changed() => break,
            _ = wait_until(auth_deadline) => break,
            _ = wait_until(ack_deadline) => {
                let now = tokio::time::Instant::now();
                session.pending_acks.retain(|_, (_, deadline, sender)| *deadline > now && !sender.is_closed());
            }
            incoming = receiver.recv() => {
                let Some(input) = incoming else {
                    break;
                };
                if !session.handle(input).await {
                    break;
                }
            }
            _ = heartbeat.tick() => {
                let elapsed = session.last_pong.elapsed();
                if elapsed >= options.heartbeat_interval + options.heartbeat_timeout {
                    break;
                }
                if !session
                    .output
                    .send(vec![EnginePacket::Ping(String::new())])
                    .await
                {
                    break;
                }
            }
        }
    }

    session.cleanup().await;
}

async fn wait_until(deadline: Option<tokio::time::Instant>) {
    match deadline {
        Some(deadline) => tokio::time::sleep_until(deadline).await,
        None => std::future::pending().await,
    }
}