//! SSE push notification stream for real-time sync notifications.
//!
//! The server sends zero-data `event: changed` events over SSE whenever
//! another device pushes changes. The client should pull on each event.
//! The stream auto-reconnects with backoff across transient drops; it only
//! ends (`next_change` returns `None`) on a fatal condition, the server
//! rejecting the reconnect for auth reasons, or a protocol violation.

use std::sync::Arc;
use std::time::Duration;

use rand::RngExt;

use crate::error::{Result, SyncKitError};

use super::{SecretToken, SyncKitClient};

/// A stream of SSE "changed" notifications from the SyncKit server.
///
/// Created by [`SyncKitClient::subscribe`]. Wraps a raw HTTP response byte
/// stream and parses SSE events line by line, yielding `Some(())` for each
/// `event: changed`. When the underlying connection drops, it transparently
/// reconnects (exponential backoff, capped, jittered) rather than ending,
/// SSE `changed` events are contentless pokes, so a reconnect needs no cursor.
/// `next_change` returns `None` only when reconnection is futile: the server
/// rejects the request with `401`/`403` (the caller must re-authenticate), or
/// the stream violates the protocol (invalid UTF-8, oversized block).
pub struct SyncNotifyStream {
    buffer: String,
    /// `None` between a drop and the next successful reconnect.
    response: Option<reqwest::Response>,
    http: reqwest::Client,
    url: String,
    token: Arc<SecretToken>,
    reconnect_attempt: u32,
}

/// Maximum SSE buffer size (1 MB). If the server sends data without a
/// block terminator (`\n\n`) exceeding this limit, we treat it as an error
/// and close the stream to prevent unbounded memory growth.
const MAX_SSE_BUFFER: usize = 1024 * 1024;

/// Cap on the reconnect backoff delay.
const MAX_RECONNECT_DELAY: Duration = Duration::from_mins(1);

/// Maximum *consecutive* failed reconnect attempts before the stream gives up.
/// A single success resets the counter, so this only fires when the server has
/// been unreachable or erroring for a sustained stretch (with the 60s backoff
/// cap, several minutes). Without it, a permanently-degraded non-auth server
/// (persistent 500s, connection refused) is retried silently forever, the
/// caller is better told the stream is dead so it can re-subscribe later.
const MAX_RECONNECT_ATTEMPTS: u32 = 10;

/// Reconnect backoff: 1s, 2s, 4s, ... capped at [`MAX_RECONNECT_DELAY`], with
/// +/-20% jitter so a fleet of clients doesn't reconnect in lockstep.
fn reconnect_delay(attempt: u32) -> Duration {
    let base = Duration::from_secs(1)
        .saturating_mul(2u32.saturating_pow(attempt.min(6)))
        .min(MAX_RECONNECT_DELAY);
    let millis = base.as_millis() as u64;
    let span = millis / 5;
    let delta = rand::rng().random_range(0..=2 * span.max(1));
    Duration::from_millis(millis.saturating_sub(span).saturating_add(delta))
}

impl SyncNotifyStream {
    pub(crate) fn new(
        http: reqwest::Client,
        url: String,
        token: Arc<SecretToken>,
        response: reqwest::Response,
    ) -> Self {
        Self {
            buffer: String::new(),
            response: Some(response),
            http,
            url,
            token,
            reconnect_attempt: 0,
        }
    }

    /// Re-establish the SSE connection after a drop. Loops with backoff on
    /// transient failures; returns `false` (fatal) if the server rejects the
    /// request for auth reasons (caller re-authenticates) or after
    /// [`MAX_RECONNECT_ATTEMPTS`] consecutive failures (caller re-subscribes
    /// later) rather than retrying a dead server forever.
    async fn reconnect(&mut self) -> bool {
        loop {
            self.reconnect_attempt += 1;
            if self.reconnect_attempt > MAX_RECONNECT_ATTEMPTS {
                tracing::warn!(
                    attempts = self.reconnect_attempt - 1,
                    "SSE reconnect gave up after too many consecutive failures; stream closing"
                );
                return false;
            }
            tokio::time::sleep(reconnect_delay(self.reconnect_attempt)).await;

            match self
                .http
                .get(&self.url)
                .bearer_auth(&*self.token)
                .send()
                .await
            {
                Ok(resp) => {
                    let status = resp.status().as_u16();
                    if status == 401 || status == 403 {
                        tracing::warn!(status, "SSE reconnect rejected (auth); stream closing");
                        return false;
                    }
                    if status >= 400 {
                        tracing::debug!(
                            status,
                            attempt = self.reconnect_attempt,
                            "SSE reconnect got error status; backing off"
                        );
                        continue;
                    }
                    tracing::debug!(attempt = self.reconnect_attempt, "SSE reconnected");
                    self.reconnect_attempt = 0;
                    self.buffer.clear();
                    self.response = Some(resp);
                    return true;
                }
                Err(e) => {
                    tracing::debug!(
                        error = %e,
                        attempt = self.reconnect_attempt,
                        "SSE reconnect failed; backing off"
                    );
                }
            }
        }
    }

    /// Wait for the next "changed" notification.
    ///
    /// Transparently reconnects across transient stream drops. Returns `Some(())`
    /// when the server signals new changes. Returns `None` only on a fatal
    /// condition: the reconnect is rejected for auth, or the stream violates the
    /// SSE protocol (invalid UTF-8 / oversized block). Keepalive comments (lines
    /// starting with `:`) are silently ignored.
    pub async fn next_change(&mut self) -> Option<()> {
        loop {
            if self.response.is_none() && !self.reconnect().await {
                return None;
            }

            // Try to extract a complete SSE block from the buffer
            if let Some(pos) = self.buffer.find("\n\n") {
                let block = self.buffer[..pos].to_string();
                self.buffer = self.buffer[pos + 2..].to_string();

                if parse_sse_block_is_changed(&block) {
                    return Some(());
                }
                // Not a "changed" event, skip and try next block
                continue;
            }

            // Need more data from the stream
            let chunk = self
                .response
                .as_mut()
                .expect("connected above")
                .chunk()
                .await;
            match chunk {
                Ok(Some(chunk)) => {
                    match std::str::from_utf8(&chunk) {
                        Ok(text) => self.buffer.push_str(text),
                        Err(_) => {
                            tracing::warn!("SSE stream contained invalid UTF-8, closing");
                            return None;
                        }
                    }
                    if self.buffer.len() > MAX_SSE_BUFFER {
                        tracing::warn!(
                            "SSE buffer exceeded {MAX_SSE_BUFFER} bytes, closing stream"
                        );
                        return None;
                    }
                }
                // Stream ended or errored: drop it and reconnect on the next loop
                // instead of ending, this is the auto-reconnect.
                Ok(None) => {
                    tracing::debug!("SSE stream ended; will reconnect");
                    self.response = None;
                }
                Err(e) => {
                    tracing::debug!(error = %e, "SSE stream error; will reconnect");
                    self.response = None;
                }
            }
        }
    }
}

/// Parse an SSE block and return `true` if it contains `event: changed`.
///
/// An SSE block is one or more lines separated by `\n`, terminated by `\n\n`.
/// Lines starting with `:` are comments (keepalive). The `event:` field
/// specifies the event type, `data:` the payload (ignored here).
fn parse_sse_block_is_changed(block: &str) -> bool {
    for line in block.lines() {
        let trimmed = line.trim();
        // Skip comments
        if trimmed.starts_with(':') {
            continue;
        }
        if let Some(value) = trimmed.strip_prefix("event:")
            && value.trim() == "changed"
        {
            return true;
        }
    }
    false
}

impl SyncKitClient {
    /// Open an SSE connection for real-time push notifications.
    ///
    /// Returns a [`SyncNotifyStream`] that yields `Some(())` each time the
    /// server signals new changes are available. The caller should pull after
    /// each notification to get the actual changes.
    ///
    /// The returned stream auto-reconnects with backoff across transient drops;
    /// `next_change()` returns `None` only on a fatal condition (auth rejected on
    /// reconnect, or a protocol violation), at which point the caller should
    /// re-authenticate and subscribe again.
    #[tracing::instrument(skip(self))]
    pub async fn subscribe(&self) -> Result<SyncNotifyStream> {
        let token = self.require_token()?;
        let (app_id, _user_id) = self.require_session_ids()?;
        let url = format!("{}?app_id={}", self.endpoints.subscribe, app_id);

        let resp = self
            .http_stream
            .get(&url)
            .bearer_auth(&*token)
            .send()
            .await
            .map_err(SyncKitError::Http)?;

        let status = resp.status().as_u16();
        if status >= 400 {
            let message = crate::client::helpers::read_text_capped(
                resp,
                crate::client::helpers::MAX_CONTROL_BODY_BYTES,
            )
            .await;
            return Err(SyncKitError::Server {
                status,
                message,
                retry_after_secs: None,
            });
        }

        // Hand the stream everything it needs to reconnect itself.
        Ok(SyncNotifyStream::new(
            self.http_stream.clone(),
            url,
            token,
            resp,
        ))
    }
}

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

    #[test]
    fn parse_changed_event() {
        let block = "event: changed\ndata: {}";
        assert!(parse_sse_block_is_changed(block));
    }

    #[test]
    fn parse_ignores_keepalive_comments() {
        let block = ": keepalive";
        assert!(!parse_sse_block_is_changed(block));
    }

    #[test]
    fn parse_handles_malformed_lines() {
        let block = "garbled nonsense";
        assert!(!parse_sse_block_is_changed(block));

        let block2 = "";
        assert!(!parse_sse_block_is_changed(block2));

        let block3 = "event:other\ndata: test";
        assert!(!parse_sse_block_is_changed(block3));
    }

    #[test]
    fn parse_changed_with_extra_whitespace() {
        let block = "event:  changed \ndata: {}";
        assert!(parse_sse_block_is_changed(block));
    }

    #[test]
    fn reconnect_delay_grows_then_caps() {
        // Within +/-20% jitter, the delay grows with the attempt and never
        // exceeds the cap plus jitter headroom.
        let d1 = reconnect_delay(1);
        let d6 = reconnect_delay(6);
        assert!(d1 >= Duration::from_millis(1600) && d1 <= Duration::from_millis(2400));
        assert!(d6 <= MAX_RECONNECT_DELAY + Duration::from_secs(12));
        // Deep attempts stay capped (exponent is clamped).
        assert!(reconnect_delay(100) <= MAX_RECONNECT_DELAY + Duration::from_secs(12));
    }
}