ai-cortex-sdk 0.1.1

//! 服务端推送的更新通知 SSE 订阅器。
//!
//! 服务端通过 `GET /api/softwares/sdk-update-events`（text/event-stream，PAT 鉴权）下发更新事件。
//! 帧格式：
//! - update 事件：`id: <u64>\nevent: update\ndata: <UpdateEvent JSON>\n\n`
//! - keepalive 注释：`: keepalive\n\n`（忽略）
//! - lagged 注释：`: lagged\n\n`（忽略）
//!
//! 支持可选查询参数 `?software_id=<id>`（仅转发该软件的事件）和可选请求头
//! `Last-Event-ID: <u64>`（断线重连，服务端补播 id > last 的事件）。

use std::collections::VecDeque;
use std::pin::Pin;
use std::task::{Context, Poll};

use futures_util::Stream;

use crate::error::{SdkError, SdkResult};
use crate::models::UpdateEvent;

/// 底层字节 chunk 流：每个 item 是一个 chunk 的字节（已从 `bytes::Bytes` 转为 `Vec<u8>`，
/// 避免在本 crate 显式依赖 bytes）。
type ByteChunkStream = Pin<Box<dyn Stream<Item = Result<Vec<u8>, reqwest::Error>> + Send>>;

/// SSE 更新事件流。实现 [`futures_util::Stream`]，逐个产出解析后的 [`UpdateEvent`]。
///
/// 用法：
/// ```ignore
/// use futures_util::StreamExt;
/// let mut stream = client.open_update_events(None, None).await?;
/// while let Some(ev) = stream.next().await {
///     match ev {
///         Ok(e) => println!("update: {:?}", e),
///         Err(e) => eprintln!("stream error: {e}"),
///     }
/// }
/// ```
pub struct UpdateEventStream {
    /// 底层 chunk 流（已 pin + type-erase）
    inner: ByteChunkStream,
    /// 尚未凑成完整 dispatch 的字节（lossy UTF-8 累积）
    line_buf: String,
    /// 已解析但尚未 yield 的事件队列
    pending: VecDeque<UpdateEvent>,
}

impl UpdateEventStream {
    /// 从一个 chunk 流构造。`stream` 的 item 为 `Result<Vec<u8>, reqwest::Error>`。
    pub fn new<S>(stream: S) -> Self
    where
        S: Stream<Item = Result<Vec<u8>, reqwest::Error>> + Send + 'static,
    {
        Self {
            inner: Box::pin(stream),
            line_buf: String::new(),
            pending: VecDeque::new(),
        }
    }

    /// 尝试从 buffer 中切出所有完整 dispatch（以 `\n\n` 分隔），逐个解析并入队。
    fn drain_complete_dispatches(&mut self) {
        // 反复切出第一个 `\n\n` 之前的内容（含分隔符）
        while let Some(idx) = self.line_buf.find("\n\n") {
            // 取出 dispatch（含末尾 `\n\n`），剩余留在 buffer
            let dispatch: String = self.line_buf.drain(..idx + 2).collect();
            if let Some(ev) = parse_dispatch(&dispatch) {
                self.pending.push_back(ev);
            }
        }
    }
}

impl Stream for UpdateEventStream {
    type Item = SdkResult<UpdateEvent>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        // 所有字段均为 Unpin（Pin<Box<..>>/String/VecDeque 都是 Unpin），
        // 可安全 get_mut
        let this = self.get_mut();

        loop {
            // 1. 优先吐出已解析的事件
            if let Some(ev) = this.pending.pop_front() {
                return Poll::Ready(Some(Ok(ev)));
            }

            // 2. 从底层流拉取下一个 chunk
            match this.inner.as_mut().poll_next(cx) {
                Poll::Ready(Some(Ok(chunk))) => {
                    // lossy UTF-8 追加（SSE 帧边界不会切在多字节字符中间，lossy 足够）
                    this.line_buf.push_str(&String::from_utf8_lossy(&chunk));
                    // 切出所有完整 dispatch 并入队
                    this.drain_complete_dispatches();
                    // 循环：若有 pending 则下一轮吐出，否则继续拉取
                }
                Poll::Ready(Some(Err(e))) => {
                    return Poll::Ready(Some(Err(SdkError::RequestError(e))));
                }
                Poll::Ready(None) => {
                    // 连接关闭：尝试解析 buffer 中残留的最后一段（无尾随 `\n\n`）
                    if !this.line_buf.is_empty() {
                        let rest: String = this.line_buf.drain(..).collect();
                        if let Some(ev) = parse_dispatch(&rest) {
                            this.pending.push_back(ev);
                        }
                    }
                    if let Some(ev) = this.pending.pop_front() {
                        return Poll::Ready(Some(Ok(ev)));
                    }
                    return Poll::Ready(None);
                }
                Poll::Pending => {
                    // 底层未就绪：若有 pending 则吐出，否则挂起
                    if let Some(ev) = this.pending.pop_front() {
                        return Poll::Ready(Some(Ok(ev)));
                    }
                    return Poll::Pending;
                }
            }
        }
    }
}

/// 解析单个 SSE dispatch（可能包含 `id:`/`event:`/`data:` 行及 `:` 注释行）。
///
/// 规则：
/// - `:` 开头为注释行（keepalive/lagged），忽略；
/// - 多个 `data:` 行按 SSE 规范以 `\n` 拼接；
/// - `id:` 行解析为 u64，覆盖事件 id（用于 Last-Event-ID 重连）；
/// - `event:` 行记录类型（当前不按类型过滤，data 能反序列化即转发）；
/// - data 为空或反序列化失败则跳过该 dispatch（不产出事件）。
fn parse_dispatch(dispatch: &str) -> Option<UpdateEvent> {
    let mut data_lines: Vec<String> = Vec::new();
    let mut id: Option<u64> = None;
    let mut _event_type: Option<String> = None;

    for line in dispatch.lines() {
        if line.is_empty() {
            continue;
        }
        // 注释行（: keepalive / : lagged 等）
        if let Some(_comment) = line.strip_prefix(':') {
            continue;
        }
        if let Some(rest) = line.strip_prefix("data:") {
            // 去掉一个可选前导空格（SSE 规范：`data: xxx` 与 `data:xxx` 都合法）
            let rest = rest.strip_prefix(' ').unwrap_or(rest);
            data_lines.push(rest.to_string());
        } else if let Some(rest) = line.strip_prefix("id:") {
            let rest = rest.strip_prefix(' ').unwrap_or(rest);
            if let Ok(parsed) = rest.trim().parse::<u64>() {
                id = Some(parsed);
            }
        } else if let Some(rest) = line.strip_prefix("event:") {
            let rest = rest.strip_prefix(' ').unwrap_or(rest);
            _event_type = Some(rest.to_string());
        }
        // 其余字段（如 retry:）忽略
    }

    if data_lines.is_empty() {
        return None;
    }

    let data = data_lines.join("\n");
    match serde_json::from_str::<UpdateEvent>(&data) {
        Ok(mut ev) => {
            // SSE 帧 id: 行优先于 JSON 内 id（id: 是 Last-Event-ID 的权威来源）
            if let Some(id) = id {
                ev.id = id;
            }
            Some(ev)
        }
        // 反序列化失败：跳过畸形帧（按 spec 选择 skip-with-warn，这里静默跳过）
        Err(_) => None,
    }
}

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

    #[test]
    fn parse_single_update_event() {
        let frame = "id: 42\nevent: update\ndata: {\"software_id\":\"s1\",\"version\":\"1.2.0\",\"platform\":\"linux\",\"channel\":\"stable\",\"force_update\":false}\n\n";
        let ev = parse_dispatch(frame).expect("should parse");
        assert_eq!(ev.id, 42);
        assert_eq!(ev.software_id, "s1");
        assert_eq!(ev.version, "1.2.0");
        assert_eq!(ev.platform, "linux");
        assert_eq!(ev.channel, "stable");
        assert!(!ev.force_update);
    }

    #[test]
    fn parse_ignores_comments() {
        // keepalive / lagged 注释应被忽略，不产出事件
        assert!(parse_dispatch(": keepalive\n\n").is_none());
        assert!(parse_dispatch(": lagged\n\n").is_none());
    }

    #[test]
    fn parse_multi_line_data() {
        let frame = "data: {\"software_id\":\"s2\",\ndata: \"version\":\"2.0\",\"platform\":\"win\",\"channel\":\"beta\",\"force_update\":true}\n\n";
        let ev = parse_dispatch(frame).expect("should parse");
        assert_eq!(ev.software_id, "s2");
        assert_eq!(ev.version, "2.0");
        assert!(ev.force_update);
    }

    #[test]
    fn parse_skips_malformed() {
        // data 不是合法 JSON → 跳过（返回 None）
        assert!(parse_dispatch("data: not-json\n\n").is_none());
    }

    #[test]
    fn parse_id_overrides_json() {
        // SSE id: 行优先于 JSON 内的 id
        let frame = "id: 99\ndata: {\"id\":5,\"software_id\":\"s\",\"version\":\"1\",\"platform\":\"p\",\"channel\":\"c\",\"force_update\":false}\n\n";
        let ev = parse_dispatch(frame).expect("should parse");
        assert_eq!(ev.id, 99);
    }
}