aurabase 0.1.1

use futures::{SinkExt, StreamExt};
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::RwLock;
use tokio::sync::broadcast;
use tokio::sync::mpsc;
use tokio::sync::oneshot;
use tokio_tungstenite::tungstenite::Message;

use crate::client::AuraClient;
use crate::error::AuraError;
use crate::types::*;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConnectionState {
    Disconnected,
    Connecting,
    Connected,
    Reconnecting,
}

/// Erreur remontée par le serveur realtime (N29).
///
/// Le SDK ignorait purement et simplement `ServerMessage::Error` (`_ => {}` dans
/// `dispatch_message`) : un refus de policy RLS à l'abonnement, un refus d'écriture ou une
/// déconnexion pour violations répétées étaient **invisibles** pour l'appelant. Ces erreurs
/// sont désormais diffusées sur [`RealtimeService::on_error`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RealtimeError {
    /// Message du serveur, tel quel.
    pub message: String,
    /// Canal concerné, en nom COURT — celui passé à [`RealtimeService::channel`], sans le
    /// préfixe `<project_id>:` que le serveur ajoute (même normalisation que pour `Subscribed`
    /// et `Event`).
    ///
    /// - `Some(nom)` → l'erreur ne concerne QUE ce canal ; les autres canaux vont bien.
    /// - `None` → erreur GLOBALE de connexion (heartbeat, token expiré, quota de violations).
    ///
    /// Un serveur antérieur à B9/N23 n'émettait pas ce champ : toutes ses erreurs arrivent donc
    /// en `None`, ce qui préserve son comportement (« tout le monde est concerné »).
    pub channel: Option<String>,
}

pub enum ChannelListener {
    PostgresChanges {
        filter: PostgresChangesFilter,
        callback: Box<dyn Fn(RealtimeEvent) + Send + Sync + 'static>,
    },
    Broadcast {
        filter: BroadcastFilter,
        callback: Box<dyn Fn(RealtimeEvent) + Send + Sync + 'static>,
    },
    Presence {
        filter: PresenceFilter,
        callback: Box<dyn Fn(serde_json::Value) + Send + Sync + 'static>,
    },
}

// ----------------------------------------------------------------------------
// AuraChannel
// ----------------------------------------------------------------------------

pub struct AuraChannel {
    pub name: String,
    /// Client d'origine — SEULE source du préfixe de canal depuis que le SDK ne porte plus
    /// d'identifiant de projet : il est lu dans le claim `project_id` du jeton de session
    /// (`AuraClient::project_prefix`), le même jeton que celui envoyé au serveur.
    client: AuraClient,
    pub listeners: Vec<ChannelListener>,
    pub subscribed: bool,
    pub is_private: bool,
    pub broadcast_ack: bool,
    pub broadcast_self: bool,
    pub presence_key: String,
    pub subscription_id: Option<String>,
    pub last_event_id: Option<String>,
}

impl AuraChannel {
    pub fn new(name: &str, client: AuraClient, options: Option<ChannelConfig>) -> Self {
        let opts = options.unwrap_or_default();
        let mut broadcast_ack = false;
        let mut broadcast_self = false;
        let mut presence_key = String::new();

        if let Some(ref bc) = opts.broadcast {
            broadcast_ack = bc.ack.unwrap_or(false);
            broadcast_self = bc.self_broadcast.unwrap_or(false);
        }

        if let Some(ref pcw) = opts.config {
            if let Some(ref pc) = pcw.presence {
                presence_key = pc.key.clone().unwrap_or_default();
            }
        }

        Self {
            name: name.to_string(),
            client,
            listeners: Vec::new(),
            subscribed: false,
            is_private: opts.private,
            broadcast_ack,
            broadcast_self,
            presence_key,
            subscription_id: None,
            last_event_id: None,
        }
    }

    /// Nom du canal SUR LE FIL : `<project_id>:<nom>`.
    ///
    /// `aura-realtime` refuse tout canal qui ne porte pas le préfixe du projet du JETON
    /// (`handlers/websocket.rs`, `expected_prefix`) ; le préfixe est donc résolu à chaque
    /// appel depuis le jeton courant, et vaut `""` tant qu'il n'y a pas de session — parité
    /// `aurabase-js` (`AuraChannel.fullName`, alimenté par un thunk `_projectPrefix`).
    pub fn full_name(&self) -> String {
        format!("{}{}", self.client.project_prefix(), self.name)
    }

    pub fn on_postgres_changes<F>(mut self, filter: PostgresChangesFilter, callback: F) -> Self
    where
        F: Fn(RealtimeEvent) + Send + Sync + 'static,
    {
        self.listeners.push(ChannelListener::PostgresChanges {
            filter,
            callback: Box::new(callback),
        });
        self
    }

    pub fn on_broadcast<F>(mut self, filter: BroadcastFilter, callback: F) -> Self
    where
        F: Fn(RealtimeEvent) + Send + Sync + 'static,
    {
        self.listeners.push(ChannelListener::Broadcast {
            filter,
            callback: Box::new(callback),
        });
        self
    }

    pub fn on_presence<F>(mut self, filter: PresenceFilter, callback: F) -> Self
    where
        F: Fn(serde_json::Value) + Send + Sync + 'static,
    {
        self.listeners.push(ChannelListener::Presence {
            filter,
            callback: Box::new(callback),
        });
        self
    }

    pub fn build_subscribe_message(&self, ref_id: &str) -> serde_json::Value {
        let pg_filters: Vec<PostgresChangesFilter> = self
            .listeners
            .iter()
            .filter_map(|l| match l {
                ChannelListener::PostgresChanges { filter, .. } => Some(filter.clone()),
                _ => None,
            })
            .collect();

        serde_json::json!({
            "type": "subscribe",
            "channel": self.full_name(),
            "ref_id": ref_id,
            "since_id": self.last_event_id,
            "config": {
                "postgres_changes": pg_filters,
                "broadcast": {
                    "self_broadcast": self.broadcast_self,
                    "ack": self.broadcast_ack,
                },
                "presence": { "key": self.presence_key },
                "private": self.is_private,
            }
        })
    }
}

// ----------------------------------------------------------------------------
// RealtimeService
// ----------------------------------------------------------------------------

pub struct RealtimeState {
    pub connection_state: ConnectionState,
    pub channels: HashMap<String, AuraChannel>,
    pub pending_acks: HashMap<String, oneshot::Sender<()>>,
    pub writer_tx: Option<mpsc::Sender<Message>>,
    pub reconnect_attempts: u32,
    pub auto_reconnect: bool,
    pub last_ref_id: u64,
}

pub struct RealtimeService {
    client: AuraClient,
    pub state: Arc<RwLock<RealtimeState>>,
    status_tx: broadcast::Sender<ConnectionState>,
    /// Diffusion des erreurs serveur (N29). `broadcast` et non `mpsc` : plusieurs consommateurs
    /// peuvent observer, et l'absence d'abonné ne doit jamais bloquer la boucle de lecture.
    error_tx: broadcast::Sender<RealtimeError>,
}

impl RealtimeService {
    pub fn new(client: AuraClient) -> Self {
        let state = Arc::new(RwLock::new(RealtimeState {
            connection_state: ConnectionState::Disconnected,
            channels: HashMap::new(),
            pending_acks: HashMap::new(),
            writer_tx: None,
            reconnect_attempts: 0,
            auto_reconnect: true,
            last_ref_id: 0,
        }));

        let (status_tx, _) = broadcast::channel(16);
        let (error_tx, _) = broadcast::channel(64);

        Self {
            client,
            state,
            status_tx,
            error_tx,
        }
    }

    pub fn connection_state(&self) -> ConnectionState {
        self.state.read().unwrap().connection_state
    }

    pub fn on_state_change(&self) -> broadcast::Receiver<ConnectionState> {
        self.status_tx.subscribe()
    }

    /// S'abonne aux erreurs remontées par le serveur (N29).
    ///
    /// À appeler AVANT [`Self::subscribe_channel`] : `tokio::sync::broadcast` ne conserve rien
    /// pour un abonné qui n'existait pas encore au moment de l'envoi.
    ///
    /// ⚠️ Sur la MÊME instance de service. `AuraClient::realtime()` construit un service NEUF à
    /// chaque appel (état, canaux et flux d'erreurs propres) : conservez l'instance, comme
    /// [`Self::subscribe_channel`] et [`Self::send_broadcast`] l'exigent déjà.
    pub fn on_error(&self) -> broadcast::Receiver<RealtimeError> {
        self.error_tx.subscribe()
    }

    /// Crée ou récupère un canal par nom.
    pub fn channel(&self, name: &str, options: Option<ChannelConfig>) -> AuraChannel {
        AuraChannel::new(name, self.client.clone(), options)
    }

    /// Ajoute ou met à jour un canal et lance la connexion.
    pub async fn subscribe_channel(&self, mut channel: AuraChannel) -> Result<(), AuraError> {
        channel.subscribed = true;
        let name = channel.name.clone();

        {
            let mut state = self.state.write().unwrap();
            state.channels.insert(name.clone(), channel);
        }

        self.connect();

        // Si déjà connecté, envoyer le message de souscription immédiatement
        let tx_and_msg = {
            let mut state = self.state.write().unwrap();
            if state.connection_state == ConnectionState::Connected && state.writer_tx.is_some() {
                state.last_ref_id += 1;
                let ref_id = format!("ref_{}", state.last_ref_id);
                let tx = state.writer_tx.as_ref().unwrap().clone();
                state.channels.get(&name).map(|ch| {
                    let msg = ch.build_subscribe_message(&ref_id);
                    (tx, Message::Text(msg.to_string()))
                })
            } else {
                None
            }
        };

        if let Some((tx, msg)) = tx_and_msg {
            let _ = tx.send(msg).await;
        }

        Ok(())
    }

    /// Supprime un canal.
    pub async fn remove_channel(&self, name: &str) {
        let tx_and_msg = {
            let mut state = self.state.write().unwrap();
            if let Some(ch) = state.channels.remove(name) {
                state.writer_tx.as_ref().map(|tx| {
                    let msg = serde_json::json!({
                        "type": "unsubscribe",
                        "channel": ch.full_name()
                    });
                    (tx.clone(), Message::Text(msg.to_string()))
                })
            } else {
                None
            }
        };

        if let Some((tx, msg)) = tx_and_msg {
            let _ = tx.send(msg).await;
        }
    }

    /// Établit la connexion WebSocket de manière asynchrone (si elle n'est pas déjà active).
    pub fn connect(&self) {
        let mut state = self.state.write().unwrap();
        if state.connection_state == ConnectionState::Connected
            || state.connection_state == ConnectionState::Connecting
        {
            return;
        }

        state.connection_state = ConnectionState::Connecting;
        let _ = self.status_tx.send(ConnectionState::Connecting);

        let base_ws = self.client.base_url().replace("http", "ws");
        let token = self.client.inner.auth_store.token();
        let api_key = self.client.api_key();

        let mut query = Vec::new();
        if let Some(tok) = token {
            query.push(format!("token={}", tok));
        }
        if let Some(key) = api_key {
            query.push(format!("apikey={}", key));
        }

        let qs = if query.is_empty() {
            "".to_string()
        } else {
            format!("?{}", query.join("&"))
        };

        let ws_url = format!("{}/v1/realtime/ws{}", base_ws, qs);

        let state_clone = self.state.clone();
        let status_tx_clone = self.status_tx.clone();
        let error_tx_clone = self.error_tx.clone();
        let client_clone = self.client.clone();

        tokio::spawn(async move {
            let res = tokio_tungstenite::connect_async(&ws_url).await;

            match res {
                Ok((ws_stream, _)) => {
                    let (mut write, mut read) = ws_stream.split();
                    let (tx, mut rx) = mpsc::channel::<Message>(64);

                    let subscribe_msgs = {
                        let mut st = state_clone.write().unwrap();
                        st.connection_state = ConnectionState::Connected;
                        st.writer_tx = Some(tx.clone());
                        st.reconnect_attempts = 0;
                        let _ = status_tx_clone.send(ConnectionState::Connected);

                        let mut msgs = Vec::new();
                        let mut ref_id_counter = st.last_ref_id;
                        // Envoyer les messages de souscription pour tous les canaux déjà abonnés
                        for ch in st.channels.values() {
                            ref_id_counter += 1;
                            let ref_id = format!("ref_{}", ref_id_counter);
                            let msg = ch.build_subscribe_message(&ref_id);
                            msgs.push(Message::Text(msg.to_string()));
                        }
                        st.last_ref_id = ref_id_counter;
                        msgs
                    };

                    for msg in subscribe_msgs {
                        let _ = write.send(msg).await;
                    }

                    // Task 1: Writer task
                    let state_clone_writer = state_clone.clone();
                    let writer_handle = tokio::spawn(async move {
                        while let Some(msg) = rx.recv().await {
                            if write.send(msg).await.is_err() {
                                break;
                            }
                        }
                        let mut st = state_clone_writer.write().unwrap();
                        st.writer_tx = None;
                    });

                    // Task 2: Heartbeat task
                    let tx_heartbeat = tx.clone();
                    let state_clone_hb = state_clone.clone();
                    let hb_handle = tokio::spawn(async move {
                        loop {
                            tokio::time::sleep(tokio::time::Duration::from_secs(25)).await;
                            {
                                let st = state_clone_hb.read().unwrap();
                                if st.connection_state != ConnectionState::Connected {
                                    break;
                                }
                            }
                            let ping_msg = serde_json::json!({ "type": "ping" });
                            if tx_heartbeat
                                .send(Message::Text(ping_msg.to_string()))
                                .await
                                .is_err()
                            {
                                break;
                            }
                        }
                    });

                    // Task 3: Reader task
                    while let Some(msg_res) = read.next().await {
                        match msg_res {
                            Ok(Message::Text(txt)) => {
                                if let Ok(msg) = serde_json::from_str::<ServerMessage>(&txt) {
                                    Self::dispatch_message(msg, &state_clone, &error_tx_clone);
                                }
                            }
                            Ok(Message::Close(_)) | Err(_) => {
                                break;
                            }
                            _ => {}
                        }
                    }

                    // Nettoyage en cas de déconnexion
                    writer_handle.abort();
                    hb_handle.abort();

                    let do_reconnect = {
                        let mut st = state_clone.write().unwrap();
                        st.connection_state = ConnectionState::Disconnected;
                        st.writer_tx = None;
                        let _ = status_tx_clone.send(ConnectionState::Disconnected);
                        st.auto_reconnect
                    };

                    if do_reconnect {
                        Self::trigger_reconnect(
                            &state_clone,
                            &status_tx_clone,
                            &error_tx_clone,
                            &client_clone,
                        )
                        .await;
                    }
                }
                Err(_) => {
                    let do_reconnect = {
                        let mut st = state_clone.write().unwrap();
                        st.connection_state = ConnectionState::Disconnected;
                        let _ = status_tx_clone.send(ConnectionState::Disconnected);
                        st.auto_reconnect
                    };
                    if do_reconnect {
                        Self::trigger_reconnect(
                            &state_clone,
                            &status_tx_clone,
                            &error_tx_clone,
                            &client_clone,
                        )
                        .await;
                    }
                }
            }
        });
    }

    /// Déclenche la reconnexion avec backoff exponentiel.
    async fn trigger_reconnect(
        state_lock: &Arc<RwLock<RealtimeState>>,
        status_tx: &broadcast::Sender<ConnectionState>,
        error_tx: &broadcast::Sender<RealtimeError>,
        client: &AuraClient,
    ) {
        let attempts = {
            let mut state = state_lock.write().unwrap();
            if state.connection_state == ConnectionState::Connecting
                || state.connection_state == ConnectionState::Connected
            {
                return;
            }
            state.connection_state = ConnectionState::Reconnecting;
            let _ = status_tx.send(ConnectionState::Reconnecting);
            state.reconnect_attempts += 1;
            state.reconnect_attempts
        };

        // Calcul du délai: 1s, 2s, 4s, 8s, 16s, 30s max
        let base_delays = [1000, 2000, 4000, 8000, 16000, 30000];
        let idx = std::cmp::min((attempts - 1) as usize, base_delays.len() - 1);
        let delay_ms = base_delays[idx];

        tokio::time::sleep(tokio::time::Duration::from_millis(delay_ms)).await;

        let svc = RealtimeService {
            client: client.clone(),
            state: state_lock.clone(),
            status_tx: status_tx.clone(),
            error_tx: error_tx.clone(),
        };

        svc.connect();
    }

    /// Route les messages reçus du serveur vers le bon canal et les callbacks correspondants.
    fn dispatch_message(
        msg: ServerMessage,
        state_lock: &Arc<RwLock<RealtimeState>>,
        error_tx: &broadcast::Sender<RealtimeError>,
    ) {
        match msg {
            ServerMessage::Subscribed(sub) => {
                let prefix = format!("{}:", sub.channel.split(':').next().unwrap_or(""));
                let name = sub.channel.trim_start_matches(&prefix).to_string();

                let mut state = state_lock.write().unwrap();
                if let Some(ch) = state.channels.get_mut(&name) {
                    ch.subscription_id = sub.subscription_id;
                }
            }
            ServerMessage::Event(evt) => {
                let prefix = format!("{}:", evt.topic.split(':').next().unwrap_or(""));
                let name = evt.topic.trim_start_matches(&prefix).to_string();

                let mut state = state_lock.write().unwrap();
                if let Some(ch) = state.channels.get_mut(&name) {
                    ch.last_event_id = Some(evt.id.clone());

                    // Vérifier le filtre RLS
                    if let Some(ref allowed) = evt.allowed_sub_ids {
                        if let Some(ref sub_id) = ch.subscription_id {
                            if !allowed.contains(sub_id) {
                                return;
                            }
                        }
                    }

                    // Dispatcher aux écouteurs du canal
                    let event = RealtimeEvent {
                        id: evt.id,
                        event: evt.event.clone(),
                        topic: evt.topic,
                        payload: evt.payload.clone(),
                        timestamp: evt.timestamp,
                        allowed_sub_ids: evt.allowed_sub_ids,
                        ref_id: None,
                    };

                    for listener in &ch.listeners {
                        match listener {
                            ChannelListener::PostgresChanges { filter, callback } => {
                                if (evt.event == "insert"
                                    || evt.event == "update"
                                    || evt.event == "delete")
                                    && (filter.event == "*"
                                        || filter.event.to_lowercase() == evt.event)
                                {
                                    if let Some(ref schema) = filter.schema {
                                        if evt.payload.get("schema").and_then(|v| v.as_str())
                                            != Some(schema)
                                        {
                                            continue;
                                        }
                                    }
                                    if let Some(ref table) = filter.table {
                                        if evt.payload.get("table").and_then(|v| v.as_str())
                                            != Some(table)
                                        {
                                            continue;
                                        }
                                    }
                                    callback(event.clone());
                                }
                            }
                            ChannelListener::Broadcast { filter, callback } => {
                                if evt.event == "broadcast"
                                    && (filter.event.is_empty()
                                        || evt.payload.get("event").and_then(|v| v.as_str())
                                            == Some(&filter.event))
                                {
                                    callback(event.clone());
                                }
                            }
                            ChannelListener::Presence { filter, callback } => {
                                if evt.event == "presence" {
                                    let action = evt
                                        .payload
                                        .get("action")
                                        .and_then(|v| v.as_str())
                                        .unwrap_or("");
                                    if (filter.event == "join" && action == "join")
                                        || (filter.event == "leave" && action == "leave")
                                    {
                                        callback(evt.payload.clone());
                                    }
                                }
                            }
                        }
                    }
                }
            }
            ServerMessage::BroadcastAck(ack) => {
                if let Some(ref_id) = ack.ref_id {
                    let mut state = state_lock.write().unwrap();
                    if let Some(tx) = state.pending_acks.remove(&ref_id) {
                        let _ = tx.send(());
                    }
                }
            }
            // N29 — l'ancien `_ => {}` avalait CECI : refus de policy RLS à l'abonnement, refus
            // d'écriture, déconnexion pour violations répétées, token invalide… L'appelant n'en
            // voyait rien. On diffuse désormais l'erreur, en normalisant le nom du canal comme
            // pour `Subscribed`/`Event` (le serveur émet `<project_id>:<nom>`, l'appelant
            // manipule `<nom>`).
            ServerMessage::Error(err) => {
                let channel = err.channel.as_deref().map(Self::nom_court);
                // `send` échoue seulement quand personne n'écoute : ce n'est pas une erreur.
                let _ = error_tx.send(RealtimeError {
                    message: err.message,
                    channel,
                });
            }
            ServerMessage::Pong
            | ServerMessage::TokenRefreshed
            | ServerMessage::PresenceState(_) => {}
        }
    }

    /// Retire le préfixe `<project_id>:` d'un nom de canal émis par le serveur.
    /// Même normalisation que pour `Subscribed` et `Event` : `proj:room` → `room`,
    /// `proj:db:public.t` → `db:public.t`.
    fn nom_court(canal_complet: &str) -> String {
        let prefix = format!("{}:", canal_complet.split(':').next().unwrap_or(""));
        canal_complet.trim_start_matches(&prefix).to_string()
    }

    /// Envoie un message de broadcast sur un canal.
    pub async fn send_broadcast(
        &self,
        channel_name: &str,
        event: &str,
        payload: serde_json::Value,
    ) -> Result<&'static str, AuraError> {
        let (full_name, ack_enabled) = {
            let state = self.state.read().unwrap();
            let ch = state.channels.get(channel_name).ok_or_else(|| {
                AuraError::new(0, "channel_not_found", "Channel not subscribed", None)
            })?;
            (ch.full_name(), ch.broadcast_ack)
        };

        let mut rx = None;
        let ref_id = {
            let mut state = self.state.write().unwrap();
            state.last_ref_id += 1;
            let ref_id = format!("ref_{}", state.last_ref_id);

            if ack_enabled {
                let (tx, r) = oneshot::channel();
                state.pending_acks.insert(ref_id.clone(), tx);
                rx = Some(r);
            }
            ref_id
        };

        let msg = serde_json::json!({
            "type": "broadcast",
            "channel": full_name,
            "ref_id": ref_id,
            "payload": {
                "event": event,
                "payload": payload
            }
        });

        let tx_to_send = {
            let state = self.state.read().unwrap();
            state.writer_tx.clone()
        };

        if let Some(tx) = tx_to_send {
            tx.send(Message::Text(msg.to_string()))
                .await
                .map_err(|e| AuraError::network(&e.to_string()))?;
        } else {
            return Err(AuraError::new(
                0,
                "disconnected",
                "WebSocket not connected",
                None,
            ));
        }

        if let Some(receiver) = rx {
            match tokio::time::timeout(tokio::time::Duration::from_secs(5), receiver).await {
                Ok(Ok(())) => Ok("ok"),
                _ => {
                    let mut state = self.state.write().unwrap();
                    state.pending_acks.remove(&ref_id);
                    Ok("timeout")
                }
            }
        } else {
            Ok("ok")
        }
    }

    /// Active la présence d'un utilisateur sur un canal.
    pub async fn track_presence(
        &self,
        channel_name: &str,
        metadata: serde_json::Value,
    ) -> Result<(), AuraError> {
        let full_name = {
            let state = self.state.read().unwrap();
            let ch = state.channels.get(channel_name).ok_or_else(|| {
                AuraError::new(0, "channel_not_found", "Channel not subscribed", None)
            })?;
            ch.full_name()
        };

        let msg = serde_json::json!({
            "type": "presence_update",
            "channel": full_name,
            "meta": metadata
        });

        let tx_opt = {
            let state = self.state.read().unwrap();
            state.writer_tx.clone()
        };

        if let Some(tx) = tx_opt {
            tx.send(Message::Text(msg.to_string()))
                .await
                .map_err(|e| AuraError::network(&e.to_string()))?;
            Ok(())
        } else {
            Err(AuraError::new(
                0,
                "disconnected",
                "WebSocket not connected",
                None,
            ))
        }
    }

    /// Arrête le tracking de la présence.
    pub async fn untrack_presence(&self, channel_name: &str) -> Result<(), AuraError> {
        self.track_presence(channel_name, serde_json::json!({}))
            .await
    }

    /// Coupe la connexion WebSocket.
    pub async fn disconnect(&self) {
        let tx_to_close = {
            let mut state = self.state.write().unwrap();
            state.auto_reconnect = false;
            state.connection_state = ConnectionState::Disconnected;
            state.writer_tx.take()
        };

        if let Some(tx) = tx_to_close {
            let _ = tx.send(Message::Close(None)).await;
        }
        let _ = self.status_tx.send(ConnectionState::Disconnected);
    }
}