mecha10-core 0.8.0

//! Streaming RPC for Long-Running Operations
//!
//! Provides bidirectional streaming RPC on top of the pub/sub messaging system.
//! Enables long-running operations where both client and server can send multiple
//! messages over a persistent stream.
//!
//! # Use Cases
//!
//! - **Video streaming**: Continuous frame delivery
//! - **Sensor data streaming**: Real-time sensor feeds
//! - **Interactive sessions**: Chat-like bidirectional communication
//! - **Progress updates**: Long-running tasks with incremental updates
//! - **File transfer**: Chunked file upload/download
//! - **Telemetry**: Continuous monitoring data
//!
//! # Stream Types
//!
//! - **Server Streaming**: Client sends one request, server sends multiple responses
//! - **Client Streaming**: Client sends multiple requests, server sends one response
//! - **Bidirectional Streaming**: Both client and server send multiple messages
//!
//! # Example: Server Streaming
//!
//! ```rust
//! use mecha10_core::prelude::*;
//! use mecha10_core::rpc_streaming::StreamingRpcExt;
//! use serde::{Serialize, Deserialize};
//! use std::time::Duration;
//!
//! #[derive(Debug, Serialize, Deserialize, Clone)]
//! struct CameraRequest { camera_id: u32 }
//! impl Message for CameraRequest {}
//!
//! #[derive(Debug, Serialize, Deserialize, Clone)]
//! struct Frame { data: Vec<u8> }
//! impl Message for Frame {}
//!
//! # fn process_frame(_frame: &Frame) -> Result<()> { Ok(()) }
//! # async fn capture_frame(_camera_id: u32) -> Result<Frame> { Ok(Frame { data: vec![] }) }
//!
//! // Client: Subscribe to video stream
//! # async fn client_example(ctx: &Context) -> Result<()> {
//! let mut stream = ctx
//!     .stream_request("/camera/stream", &CameraRequest { camera_id: 0 })
//!     .await?;
//!
//! while let Some(frame) = stream.recv().await {
//!     process_frame(&frame)?;
//! }
//! # Ok(())
//! # }
//!
//! // Server: Stream video frames
//! # async fn server_example(ctx: &Context) -> Result<()> {
//! ctx.stream_respond("/camera/stream", |req: CameraRequest, sender| async move {
//!     loop {
//!         let frame = capture_frame(req.camera_id).await?;
//!         sender.send(&frame).await?;
//!         tokio::time::sleep(Duration::from_millis(33)).await; // 30 FPS
//!     }
//! }).await?;
//! # Ok(())
//! # }
//! ```
//!
//! # Example: Bidirectional Streaming
//!
//! ```rust
//! use mecha10_core::prelude::*;
//! use mecha10_core::rpc_streaming::StreamingRpcExt;
//! use serde::{Serialize, Deserialize};
//!
//! #[derive(Debug, Serialize, Deserialize, Clone)]
//! struct ChatMessage { text: String }
//! impl Message for ChatMessage {}
//!
//! // Interactive chat session
//! # async fn bidirectional_example(ctx: &Context) -> Result<()> {
//! let (tx, mut rx) = ctx
//!     .bidirectional_stream::<ChatMessage>("/chat/session")
//!     .await?;
//!
//! // Send and receive concurrently
//! tokio::spawn(async move {
//!     while let Some(msg) = rx.recv().await {
//!         println!("Received: {}", msg.text);
//!     }
//! });
//!
//! tx.send(&ChatMessage { text: "Hello!".into() }).await?;
//! # Ok(())
//! # }
//! ```

use crate::context::{Context, Receiver};
use crate::error::{Mecha10Error, Result};
use crate::messages::Message;
use crate::prelude::now_micros;
use serde::{de::DeserializeOwned, Deserialize, Serialize};
use std::future::Future;
use std::marker::PhantomData;
use std::sync::Arc;
use tokio::sync::Mutex;
use tracing::{debug, error, info, warn};
use uuid::Uuid;

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamMessage<T> {
    pub stream_id: String,

    pub sequence: u64,

    pub payload: StreamPayload<T>,

    pub timestamp: u64,
}

impl<T: Message> Message for StreamMessage<T> {}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum StreamPayload<T> {
    Data(T),

    Start,

    End,

    Error(StreamError),

    Cancel,

    Heartbeat,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamError {
    pub message: String,
    pub code: Option<String>,
}

pub struct StreamReceiver<T> {
    stream_id: String,
    receiver: Receiver<StreamMessage<T>>,
    sequence: u64,
    ended: bool,
}

impl<T: Message + DeserializeOwned + Send + 'static> StreamReceiver<T> {
    /// Create a new stream receiver
    pub(crate) fn new(stream_id: String, receiver: Receiver<StreamMessage<T>>) -> Self {
        Self {
            stream_id,
            receiver,
            sequence: 0,
            ended: false,
        }
    }

    /// Receive the next message from the stream
    ///
    /// Returns:
    /// - `Some(T)` if a data message was received
    /// - `None` if the stream has ended
    pub async fn recv(&mut self) -> Option<T> {
        if self.ended {
            return None;
        }

        loop {
            let msg = self.receiver.recv().await?;

            if msg.stream_id != self.stream_id {
                continue;
            }

            if msg.sequence < self.sequence {
                warn!(
                    "Out of order message in stream {} (expected {}, got {})",
                    self.stream_id, self.sequence, msg.sequence
                );
            }
            self.sequence = msg.sequence + 1;

            match msg.payload {
                StreamPayload::Data(data) => {
                    debug!("Stream {} received message #{}", self.stream_id, msg.sequence);
                    return Some(data);
                }
                StreamPayload::End => {
                    info!("Stream {} ended normally", self.stream_id);
                    self.ended = true;
                    return None;
                }
                StreamPayload::Error(err) => {
                    error!("Stream {} error: {}", self.stream_id, err.message);
                    self.ended = true;
                    return None;
                }
                StreamPayload::Cancel => {
                    info!("Stream {} cancelled", self.stream_id);
                    self.ended = true;
                    return None;
                }
                StreamPayload::Start => {
                    debug!("Stream {} started", self.stream_id);
                }
                StreamPayload::Heartbeat => {
                    debug!("Stream {} heartbeat", self.stream_id);
                }
            }
        }
    }

    /// Check if the stream has ended
    pub fn is_ended(&self) -> bool {
        self.ended
    }

    /// Get the stream ID
    pub fn stream_id(&self) -> &str {
        &self.stream_id
    }
}

#[derive(Clone)]
pub struct StreamSender<T: Message + Serialize + Clone> {
    stream_id: String,
    topic: String,
    ctx: Arc<Context>,
    sequence: Arc<Mutex<u64>>,
    ended: Arc<Mutex<bool>>,
    _phantom: PhantomData<T>,
}

impl<T: Message + Serialize + Clone> StreamSender<T> {
    /// Create a new stream sender
    pub(crate) fn new(stream_id: String, topic: String, ctx: Arc<Context>) -> Self {
        Self {
            stream_id,
            topic,
            ctx,
            sequence: Arc::new(Mutex::new(0)),
            ended: Arc::new(Mutex::new(false)),
            _phantom: PhantomData,
        }
    }

    /// Send a data message
    pub async fn send(&self, data: &T) -> Result<()> {
        if *self.ended.lock().await {
            return Err(Mecha10Error::MessagingError {
                message: "Stream already ended".to_string(),
                suggestion: "Cannot send after stream end".to_string(),
            });
        }

        let mut seq = self.sequence.lock().await;
        let message = StreamMessage {
            stream_id: self.stream_id.clone(),
            sequence: *seq,
            payload: StreamPayload::Data(data.clone()),
            timestamp: now_micros(),
        };

        *seq += 1;
        drop(seq);

        self.ctx.publish_raw(&self.topic, &message).await?;
        debug!("Stream {} sent message #{}", self.stream_id, message.sequence);

        Ok(())
    }

    /// Send stream end signal
    pub async fn end(&self) -> Result<()> {
        let mut ended = self.ended.lock().await;
        if *ended {
            return Ok(());
        }

        let seq = self.sequence.lock().await;
        let message = StreamMessage {
            stream_id: self.stream_id.clone(),
            sequence: *seq,
            payload: StreamPayload::<T>::End,
            timestamp: now_micros(),
        };

        self.ctx.publish_raw(&self.topic, &message).await?;
        *ended = true;

        info!("Stream {} ended", self.stream_id);
        Ok(())
    }

    /// Send error and end stream
    pub async fn error(&self, error_message: &str) -> Result<()> {
        let mut ended = self.ended.lock().await;
        if *ended {
            return Ok(());
        }

        let seq = self.sequence.lock().await;
        let message = StreamMessage {
            stream_id: self.stream_id.clone(),
            sequence: *seq,
            payload: StreamPayload::<T>::Error(StreamError {
                message: error_message.to_string(),
                code: None,
            }),
            timestamp: now_micros(),
        };

        self.ctx.publish_raw(&self.topic, &message).await?;
        *ended = true;

        error!("Stream {} error: {}", self.stream_id, error_message);
        Ok(())
    }

    /// Send heartbeat (keep-alive)
    pub async fn heartbeat(&self) -> Result<()> {
        if *self.ended.lock().await {
            return Ok(());
        }

        let seq = self.sequence.lock().await;
        let message = StreamMessage {
            stream_id: self.stream_id.clone(),
            sequence: *seq,
            payload: StreamPayload::<T>::Heartbeat,
            timestamp: now_micros(),
        };

        self.ctx.publish_raw(&self.topic, &message).await?;
        debug!("Stream {} heartbeat", self.stream_id);

        Ok(())
    }

    /// Check if stream has ended
    pub async fn is_ended(&self) -> bool {
        *self.ended.lock().await
    }

    /// Get stream ID
    pub fn stream_id(&self) -> &str {
        &self.stream_id
    }
}

impl<T: Message + Serialize + Clone> Drop for StreamSender<T> {
    fn drop(&mut self) {
        debug!("StreamSender dropped for stream {}", self.stream_id);
    }
}

pub trait StreamingRpcExt {
    /// Create a server-streaming RPC (client sends one request, server streams responses)
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10_core::prelude::*;
    /// # use mecha10_core::rpc_streaming::StreamingRpcExt;
    /// # use serde::{Serialize, Deserialize};
    /// # #[derive(Debug, Serialize, Deserialize, Clone)]
    /// # struct CameraRequest { camera_id: u32 }
    /// # impl Message for CameraRequest {}
    /// # #[derive(Debug, Serialize, Deserialize, Clone)]
    /// # struct Frame { data: Vec<u8> }
    /// # impl Message for Frame {}
    /// # fn process_frame(_frame: &Frame) -> Result<()> { Ok(()) }
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// let mut stream = ctx
    ///     .stream_request("/camera/feed", &CameraRequest { camera_id: 0 })
    ///     .await?;
    ///
    /// while let Some(frame) = stream.recv().await {
    ///     process_frame(&frame)?;
    /// }
    /// # Ok(())
    /// # }
    /// ```
    fn stream_request<Req, Resp>(
        &self,
        topic: &str,
        request: &Req,
    ) -> impl Future<Output = Result<StreamReceiver<Resp>>> + Send
    where
        Req: Message + Serialize + Clone,
        Resp: Message + DeserializeOwned + Send + 'static;

    /// Handle server-streaming RPC requests
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10_core::prelude::*;
    /// # use mecha10_core::rpc_streaming::StreamingRpcExt;
    /// # use serde::{Serialize, Deserialize};
    /// # use std::time::Duration;
    /// # #[derive(Debug, Serialize, Deserialize, Clone)]
    /// # struct CameraRequest { camera_id: u32 }
    /// # impl Message for CameraRequest {}
    /// # #[derive(Debug, Serialize, Deserialize, Clone)]
    /// # struct Frame { data: Vec<u8> }
    /// # impl Message for Frame {}
    /// # async fn capture_frame(_camera_id: u32) -> Result<Frame> { Ok(Frame { data: vec![] }) }
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// ctx.stream_respond("/camera/feed", |req: CameraRequest, sender| async move {
    ///     loop {
    ///         let frame = capture_frame(req.camera_id).await?;
    ///         sender.send(&frame).await?;
    ///         tokio::time::sleep(Duration::from_millis(33)).await;
    ///     }
    /// }).await?;
    /// # Ok(())
    /// # }
    /// ```
    fn stream_respond<Req, Resp, F, Fut>(&self, topic: &str, handler: F) -> impl Future<Output = Result<()>> + Send
    where
        Req: Message + DeserializeOwned + Send + 'static,
        Resp: Message + Serialize + Clone + Send + 'static,
        F: Fn(Req, StreamSender<Resp>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<()>> + Send + 'static;

    /// Create a bidirectional stream (both sides can send/receive)
    ///
    /// # Example
    ///
    /// ```rust
    /// # use mecha10_core::prelude::*;
    /// # use mecha10_core::rpc_streaming::StreamingRpcExt;
    /// # use serde::{Serialize, Deserialize};
    /// # #[derive(Debug, Serialize, Deserialize, Clone)]
    /// # struct ChatMessage { text: String }
    /// # impl Message for ChatMessage {}
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// let (tx, mut rx) = ctx.bidirectional_stream::<ChatMessage>("/chat").await?;
    ///
    /// // Send
    /// tx.send(&ChatMessage { text: "Hello".into() }).await?;
    ///
    /// // Receive
    /// while let Some(msg) = rx.recv().await {
    ///     println!("Received: {}", msg.text);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    fn bidirectional_stream<T>(
        &self,
        topic: &str,
    ) -> impl Future<Output = Result<(StreamSender<T>, StreamReceiver<T>)>> + Send
    where
        T: Message + Serialize + DeserializeOwned + Clone + Send + 'static;
}

impl StreamingRpcExt for Context {
    async fn stream_request<Req, Resp>(&self, topic: &str, request: &Req) -> Result<StreamReceiver<Resp>>
    where
        Req: Message + Serialize + Clone,
        Resp: Message + DeserializeOwned + Send + 'static,
    {
        let stream_id = Uuid::new_v4().to_string();
        let request_topic = format!("{}/stream/request", topic);
        let response_topic = format!("{}/stream/response", topic);

        let receiver = self.subscribe_raw::<StreamMessage<Resp>>(&response_topic).await?;

        let start_msg = StreamMessage {
            stream_id: stream_id.clone(),
            sequence: 0,
            payload: StreamPayload::Data(request.clone()),
            timestamp: now_micros(),
        };

        self.publish_raw(&request_topic, &start_msg).await?;

        info!("Stream request started: {} (stream_id: {})", topic, stream_id);

        Ok(StreamReceiver::new(stream_id, receiver))
    }

    async fn stream_respond<Req, Resp, F, Fut>(&self, topic: &str, handler: F) -> Result<()>
    where
        Req: Message + DeserializeOwned + Send + 'static,
        Resp: Message + Serialize + Clone + Send + 'static,
        F: Fn(Req, StreamSender<Resp>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<()>> + Send + 'static,
    {
        let request_topic = format!("{}/stream/request", topic);
        let response_topic = format!("{}/stream/response", topic);

        let mut requests = self.subscribe_raw::<StreamMessage<Req>>(&request_topic).await?;
        let ctx = Arc::new(self.clone());
        let handler = Arc::new(handler);

        info!("Stream responder registered for: {}", topic);

        tokio::spawn(async move {
            while let Some(msg) = requests.recv().await {
                match msg.payload {
                    StreamPayload::Data(request) => {
                        let stream_id = msg.stream_id.clone();
                        let sender = StreamSender::new(stream_id.clone(), response_topic.clone(), Arc::clone(&ctx));

                        let handler = Arc::clone(&handler);

                        tokio::spawn(async move {
                            debug!("Handling stream request: {}", stream_id);

                            if let Err(e) = handler(request, sender.clone()).await {
                                error!("Stream handler error ({}): {}", stream_id, e);
                                let _ = sender.error(&e.to_string()).await;
                            } else {
                                let _ = sender.end().await;
                            }
                        });
                    }
                    StreamPayload::Cancel => {
                        debug!("Stream cancelled: {}", msg.stream_id);
                    }
                    _ => {}
                }
            }
        });

        Ok(())
    }

    async fn bidirectional_stream<T>(&self, topic: &str) -> Result<(StreamSender<T>, StreamReceiver<T>)>
    where
        T: Message + Serialize + DeserializeOwned + Clone + Send + 'static,
    {
        let stream_id = Uuid::new_v4().to_string();
        let client_topic = format!("{}/bidir/client", topic);
        let server_topic = format!("{}/bidir/server", topic);

        let receiver = self.subscribe_raw::<StreamMessage<T>>(&server_topic).await?;

        let sender = StreamSender::new(stream_id.clone(), client_topic, Arc::new(self.clone()));

        info!("Bidirectional stream created: {} (stream_id: {})", topic, stream_id);

        Ok((sender, StreamReceiver::new(stream_id, receiver)))
    }
}