use crate::error::Result;
use crate::types::*;
use futures_util::StreamExt;
use serde::{Deserialize, Serialize};
use tokio::time::{sleep, Duration};
use tokio_tungstenite::{connect_async, tungstenite::protocol::Message};

#[derive(Debug, Deserialize, Serialize)]
pub struct JetstreamCommit {
    pub rev: String,
    pub operation: String,
    pub collection: String,
    pub rkey: String,
    pub record: serde_json::Value,
    pub cid: String,
}

#[derive(Debug, Deserialize, Serialize)]
pub struct JetstreamEvent {
    pub did: String,
    pub time_us: u64,
    pub kind: String,
    pub commit: Option<JetstreamCommit>,
}

pub enum EgregoreEvent {
    Signal {
        did: String,
        uri: String,
        signal: Signal,
    },
    Ack {
        did: String,
        uri: String,
        ack: Ack,
    },
}

pub struct JetstreamListener {
    url: String,
    cursor: Option<u64>,
    ws_stream: Option<
        tokio_tungstenite::WebSocketStream<
            tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
        >,
    >,
}

impl JetstreamListener {
    pub async fn connect(jetstream_url: &str) -> Result<Self> {
        Ok(Self {
            url: jetstream_url.to_string(),
            cursor: None,
            ws_stream: None,
        })
    }

    pub fn set_cursor(&mut self, cursor: u64) {
        self.cursor = Some(cursor);
    }

    async fn ensure_connected(&mut self) -> Result<()> {
        if self.ws_stream.is_some() {
            return Ok(());
        }

        let mut backoff = Duration::from_secs(1);
        let max_backoff = Duration::from_secs(30);

        loop {
            let mut connect_url = format!(
                "{}?wantedCollections=system.egregore.signal&wantedCollections=system.egregore.ack",
                self.url
            );
            if let Some(cursor) = self.cursor {
                connect_url.push_str(&format!("&cursor={}", cursor));
            }

            match connect_async(&connect_url).await {
                Ok((stream, _)) => {
                    self.ws_stream = Some(stream);
                    return Ok(());
                }
                Err(e) => {
                    eprintln!(
                        "[jetstream] connection failed: {}. retrying in {:?}...",
                        e, backoff
                    );
                    sleep(backoff).await;
                    backoff = std::cmp::min(backoff * 2, max_backoff);
                }
            }
        }
    }

    pub fn parse_event(text: &str) -> Option<EgregoreEvent> {
        let event: JetstreamEvent = serde_json::from_str(text).ok()?;
        if event.kind != "commit" {
            return None;
        }
        let commit = event.commit?;
        if commit.operation != "create" {
            return None;
        }
        let uri = format!("at://{}/{}/{}", event.did, commit.collection, commit.rkey);

        if commit.collection == "system.egregore.signal" {
            let mut signal: Signal = serde_json::from_value(commit.record).ok()?;
            signal.author_did = Some(event.did.clone());
            signal.uri = Some(uri.clone());
            Some(EgregoreEvent::Signal {
                did: event.did,
                uri,
                signal,
            })
        } else if commit.collection == "system.egregore.ack" {
            let mut ack: Ack = serde_json::from_value(commit.record).ok()?;
            ack.author_did = Some(event.did.clone());
            ack.uri = Some(uri.clone());
            Some(EgregoreEvent::Ack {
                did: event.did,
                uri,
                ack,
            })
        } else {
            None
        }
    }

    pub async fn next_event(&mut self) -> Result<EgregoreEvent> {
        loop {
            self.ensure_connected().await?;

            let stream = self.ws_stream.as_mut().unwrap();

            match stream.next().await {
                Some(Ok(Message::Text(text))) => {
                    if let Ok(event) = serde_json::from_str::<JetstreamEvent>(&text) {
                        self.cursor = Some(event.time_us);
                    }
                    if let Some(egregore_event) = Self::parse_event(&text) {
                        return Ok(egregore_event);
                    }
                }
                Some(Ok(Message::Close(_))) | Some(Err(_)) | None => {
                    self.ws_stream = None;
                    eprintln!("[jetstream] connection lost. reconnecting...");
                }
                _ => {} // Ignore non-text messages
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_parse_signal_event() {
        let json = r#"{
            "did": "did:plc:abc123",
            "time_us": 1700000000000000,
            "kind": "commit",
            "commit": {
                "rev": "rev1",
                "operation": "create",
                "collection": "system.egregore.signal",
                "rkey": "3abc",
                "record": {
                    "$type": "system.egregore.signal",
                    "signalType": "interrupt",
                    "socket": "arena",
                    "content": { "$type": "interrupt", "lens": "test", "body": "hello", "scope": ["general"] },
                    "priority": 5,
                    "createdAt": "2026-02-06T04:00:00Z"
                },
                "cid": "bafyrei123"
            }
        }"#;
        let event = JetstreamListener::parse_event(json);
        assert!(event.is_some());
        match event.unwrap() {
            EgregoreEvent::Signal { did, uri, signal } => {
                assert_eq!(did, "did:plc:abc123");
                assert!(uri.contains("system.egregore.signal"));
                assert_eq!(signal.socket, "arena");
                assert_eq!(signal.signal_type, SignalType::Interrupt);
            }
            _ => panic!("Expected Signal event"),
        }
    }

    #[test]
    fn test_parse_non_create_ignored() {
        let json = r#"{
            "did": "did:plc:abc123",
            "time_us": 1700000000000000,
            "kind": "commit",
            "commit": {
                "rev": "rev1",
                "operation": "delete",
                "collection": "system.egregore.signal",
                "rkey": "3abc",
                "record": {},
                "cid": "bafyrei123"
            }
        }"#;
        assert!(JetstreamListener::parse_event(json).is_none());
    }

    #[test]
    fn test_parse_irrelevant_collection_ignored() {
        let json = r#"{
            "did": "did:plc:abc123",
            "time_us": 1700000000000000,
            "kind": "identity",
            "commit": null
        }"#;
        assert!(JetstreamListener::parse_event(json).is_none());
    }
}
