feltdb-client 0.6.1

use serde::de::DeserializeOwned;

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

/// One server-sent event.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ServerEvent {
    /// The opaque cursor after this event. Pass it back to resume from here.
    pub id: Option<String>,
    /// The event name; `state.committed` for a committed change.
    pub event: String,
    /// The event payload, a JSON document for `state.committed`.
    pub data: String,
}

impl ServerEvent {
    /// Decode the payload.
    pub fn json<T: DeserializeOwned>(&self) -> Result<T> {
        serde_json::from_str(&self.data).map_err(|source| Error::Decode {
            status: 200,
            source,
        })
    }
}

/// A live event stream: a delivery adapter over committed history, not the history itself.
///
/// It does not reconnect on its own. When it ends or fails, open a new one from
/// [`EventStream::last_event_id`] and nothing committed in between is skipped. For a complete,
/// bounded read of history use the application's event replay instead.
pub struct EventStream {
    response: reqwest::Response,
    buffer: Vec<u8>,
    last_event_id: Option<String>,
    finished: bool,
}

impl EventStream {
    pub(crate) fn new(response: reqwest::Response, resumed_from: Option<String>) -> Self {
        Self {
            response,
            buffer: Vec::new(),
            last_event_id: resumed_from,
            finished: false,
        }
    }

    /// The cursor of the last event delivered (or the one this stream resumed from).
    pub fn last_event_id(&self) -> Option<&str> {
        self.last_event_id.as_deref()
    }

    /// The next event; `None` when the server closed the stream.
    ///
    /// An `authority.error` event from the server ends the stream with [`Error::Stream`].
    pub async fn next(&mut self) -> Option<Result<ServerEvent>> {
        loop {
            if self.finished {
                return None;
            }
            while let Some(frame) = take_frame(&mut self.buffer) {
                let Some(event) = parse_frame(&frame) else {
                    continue;
                };
                if event.event == "authority.error" {
                    self.finished = true;
                    return Some(Err(Error::Stream(event.data)));
                }
                if event.id.is_some() {
                    self.last_event_id.clone_from(&event.id);
                }
                return Some(Ok(event));
            }
            match self.response.chunk().await {
                // Bytes, not text: a chunk can end inside a multi-byte character, and only a
                // complete frame is decoded. Carriage returns are dropped so CRLF framing parses
                // as LF; JSON payloads escape theirs.
                Ok(Some(bytes)) => self
                    .buffer
                    .extend(bytes.iter().copied().filter(|byte| *byte != b'\r')),
                Ok(None) => {
                    self.finished = true;
                    return None;
                }
                Err(source) => {
                    self.finished = true;
                    return Some(Err(Error::transport(source, false)));
                }
            }
        }
    }
}

fn take_frame(buffer: &mut Vec<u8>) -> Option<String> {
    let end = buffer.windows(2).position(|pair| pair == b"\n\n")?;
    let frame = String::from_utf8_lossy(&buffer[..end]).into_owned();
    buffer.drain(..end + 2);
    Some(frame)
}

/// Parse one SSE frame. Comments (keep-alives) and frames without data are skipped.
fn parse_frame(frame: &str) -> Option<ServerEvent> {
    let mut id = None;
    let mut event = None;
    let mut data: Vec<&str> = Vec::new();
    for line in frame.split('\n') {
        if line.starts_with(':') {
            continue;
        }
        let (field, value) = line.split_once(':').unwrap_or((line, ""));
        let value = value.strip_prefix(' ').unwrap_or(value);
        match field {
            "id" => id = Some(value.to_owned()),
            "event" => event = Some(value.to_owned()),
            "data" => data.push(value),
            _ => {}
        }
    }
    if data.is_empty() {
        return None;
    }
    Some(ServerEvent {
        id,
        event: event.unwrap_or_else(|| "message".to_owned()),
        data: data.join("\n"),
    })
}

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

    #[test]
    fn frames_are_split_and_parsed() {
        let mut buffer =
            b": keep-alive\n\nid: c-2\nevent: state.committed\ndata: {\"a\":\ndata: 1}\n\nid: partial"
                .to_vec();
        assert_eq!(parse_frame(&take_frame(&mut buffer).unwrap()), None);
        let event = parse_frame(&take_frame(&mut buffer).unwrap()).unwrap();
        assert_eq!(event.id.as_deref(), Some("c-2"));
        assert_eq!(event.event, "state.committed");
        assert_eq!(event.data, "{\"a\":\n1}");
        assert_eq!(take_frame(&mut buffer), None);
        assert_eq!(buffer, b"id: partial");
    }
}