use std::collections::HashMap;
use std::time::Duration;

use actor::{ActorHost, PushStore};
use async_trait::async_trait;
use socket::{
    ConnectedSession, ConnectionPhase, DeliveryContext, SocketBackend, SocketError, SocketPayload,
};

use crate::gateway::platform::{PlatformClient, PlatformError};
use crate::gateway::subscriptions::Subscriptions;

const IDLE_FALLBACK: Duration = Duration::from_secs(30);
const STANDBY_WAIT: Duration = Duration::from_secs(2);
/// 单条订阅的投递授权结论缓存时长；与身份复核的 60 秒节奏一致。
const AUTHORIZATION_TTL: Duration = Duration::from_secs(60);

/// socket 通道与事件业务之间的适配层：身份、订阅、投递策略都落在这里。
#[derive(Clone)]
pub struct RealtimeBackend {
    host: ActorHost,
    platform: PlatformClient,
    subscriptions: Subscriptions,
    push_store: PushStore,
}

impl RealtimeBackend {
    pub fn new(
        host: ActorHost,
        platform: PlatformClient,
        subscriptions: Subscriptions,
        push_store: PushStore,
    ) -> Self {
        Self {
            host,
            platform,
            subscriptions,
            push_store,
        }
    }
}

#[async_trait]
impl SocketBackend for RealtimeBackend {
    async fn authenticate(&self, token: &str) -> Result<i64, SocketError> {
        self.platform
            .authenticate(token)
            .await
            .map_err(|error| match error {
                PlatformError::Rejected(message) => SocketError::Rejected(message),
                PlatformError::Unavailable(message) => SocketError::Unavailable(message),
                PlatformError::Config(message) => SocketError::Config(message),
            })
    }

    async fn on_connect(&self, session: &ConnectedSession) -> Result<(), SocketError> {
        self.push_store
            .register_device(session.user_id, &session.device_id, &session.namespace)
            .await
            .map_err(SocketError::Unavailable)?;
        if session.first_namespace {
            self.host
                .socket_heartbeat(session.user_id)
                .await
                .map_err(|error| SocketError::Unavailable(error.to_string()))?;
        }
        Ok(())
    }

    async fn heartbeat(&self, user_id: i64) {
        let _ = self.host.socket_heartbeat(user_id).await;
    }

    async fn report_connection(&self, phase: ConnectionPhase, user_id: i64, session: &str) {
        let phase = match phase {
            ConnectionPhase::Open => "open",
            ConnectionPhase::Heartbeat => "heartbeat",
            ConnectionPhase::Close => "close",
        };
        if let Err(error) = self
            .platform
            .report_connection(phase, user_id, session)
            .await
        {
            tracing::warn!(%error, "chat connection report failed");
        }
    }

    async fn run_delivery(&self, context: DeliveryContext) {
        let store = self.push_store.clone();
        let subscriptions = self.subscriptions.clone();
        let platform = self.platform.clone();
        let hub = context.hub.clone();
        let input = context.handle.clone();
        let user = context.user_id;
        let device = context.device_id.clone();
        let namespace = context.namespace.clone();
        let token = context.access_token.clone();
        let owner = format!("{}:{}", context.session_sid, context.session_id);
        let mut stopped = context.stop;
        let mut shutdown = hub.shutdown_receiver();
        let notification = hub.notification(user).await;
        let work = async {
            let mut verified: Option<tokio::time::Instant> = None;
            let mut authorized: HashMap<String, tokio::time::Instant> = HashMap::new();
            loop {
                let changed = notification.notified();
                tokio::pin!(changed);
                changed.as_mut().enable();
                if verified.is_none_or(|verified: tokio::time::Instant| {
                    verified.elapsed() >= Duration::from_secs(60)
                }) {
                    match platform.authenticate(&token).await {
                        Ok(current) if current == user => {
                            verified = Some(tokio::time::Instant::now())
                        }
                        Err(PlatformError::Unavailable(_)) => {
                            tokio::time::sleep(Duration::from_secs(1)).await;
                            continue;
                        }
                        _ => {
                            input.close().await;
                            break;
                        }
                    }
                }
                if !matches!(
                    store.claim(user, &device, &namespace, &owner).await,
                    Ok(true)
                ) {
                    tokio::time::sleep(STANDBY_WAIT).await;
                    continue;
                }
                let (cursor, next) = match store.next(user, &device, &namespace).await {
                    Ok(next) => next,
                    Err(_) => {
                        tokio::time::sleep(Duration::from_secs(1)).await;
                        continue;
                    }
                };
                let Some((delivery, mut payload)) = next else {
                    tokio::select! {
                        _ = &mut changed => {}
                        _ = tokio::time::sleep(IDLE_FALLBACK) => {}
                    }
                    continue;
                };
                let authorization = async {
                    if let Some(destinations) = payload.get("destinations") {
                        let destinations: Vec<crate::gateway::Subscription> =
                            serde_json::from_value(destinations.clone())
                                .map_err(|error| PlatformError::Unavailable(error.to_string()))?;
                        for subscription in destinations {
                            if subscription.device_id != device
                                || subscription.namespace != namespace
                            {
                                continue;
                            }
                            if authorized
                                .get(&subscription.id)
                                .is_some_and(|at| at.elapsed() < AUTHORIZATION_TTL)
                            {
                                return Ok(true);
                            }
                            if !subscriptions
                                .contains(&subscription)
                                .await
                                .map_err(PlatformError::Unavailable)?
                            {
                                continue;
                            }
                            if !platform
                                .authorize(user, &subscription.filter, "DELIVER")
                                .await?
                            {
                                continue;
                            }
                            if let Some(org) = subscription.filter.org_id.as_deref() {
                                if !platform.is_member(user, org).await? {
                                    continue;
                                }
                            }
                            authorized.insert(subscription.id.clone(), tokio::time::Instant::now());
                            return Ok(true);
                        }
                        return Ok(false);
                    }
                    if payload.get("source").and_then(serde_json::Value::as_str) == Some("socket") {
                        return Ok(payload.get("namespace").and_then(serde_json::Value::as_str)
                            == Some(namespace.as_str()));
                    }
                    if payload.get("control").and_then(serde_json::Value::as_bool) != Some(true) {
                        if let Some(org) = payload
                            .pointer("/properties/org_id")
                            .and_then(serde_json::Value::as_str)
                        {
                            return platform.is_member(user, org).await;
                        }
                    }
                    Ok(true)
                }
                .await;
                let allowed = match authorization {
                    Ok(allowed) => allowed,
                    Err(PlatformError::Rejected(_)) => {
                        input.close().await;
                        break;
                    }
                    Err(_) => {
                        tokio::time::sleep(Duration::from_secs(1)).await;
                        continue;
                    }
                };
                if allowed {
                    let previous_sequence = cursor
                        .split_once('-')
                        .and_then(|(number, _)| number.parse::<u64>().ok());
                    let sequence = delivery
                        .split_once('-')
                        .and_then(|(number, _)| number.parse::<u64>().ok());
                    if previous_sequence.and_then(|previous| previous.checked_add(1)) != sequence {
                        payload["resync_required"] = serde_json::Value::Bool(true);
                    }
                    let resync = payload
                        .get("resync_required")
                        .and_then(serde_json::Value::as_bool)
                        == Some(true);
                    let binaries: Vec<Vec<u8>> = payload
                        .get("binaries")
                        .cloned()
                        .and_then(|value| serde_json::from_value(value).ok())
                        .unwrap_or_default();
                    if let Some(object) = payload.as_object_mut() {
                        object.remove("binaries");
                        object.remove("destinations");
                    }
                    let packet = crate::gateway::EmitPacket {
                        namespace: namespace.clone(),
                        event: if resync { "resync" } else { "event" }.to_string(),
                        args: vec![payload],
                        binaries,
                    };
                    if !input.deliver(packet, Duration::from_secs(15)).await {
                        tokio::time::sleep(Duration::from_secs(1)).await;
                        continue;
                    }
                }
                if !matches!(
                    store
                        .acknowledge(user, &device, &namespace, &owner, &cursor, &delivery)
                        .await,
                    Ok(true)
                ) {
                    tokio::time::sleep(Duration::from_secs(1)).await;
                }
            }
        };
        if !*shutdown.borrow() {
            tokio::select! {
                _ = &mut stopped => {}
                _ = shutdown.changed() => {}
                _ = input.closed() => {}
                _ = work => {}
            }
        }
        store.release(user, &device, &namespace, &owner).await;
        let _ = SocketPayload::default();
    }
}