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;
    let mut maintenance = interval(std::time::Duration::from_secs(1));

    loop {
        if *shutdown.borrow() {
            break;
        }
        tokio::select! {
            _ = shutdown.changed() => break,
            _ = maintenance.tick() => {
                let now = tokio::time::Instant::now();
                if session.auth_deadline.is_some_and(|deadline| now >= deadline) {
                    break;
                }
                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;
}