pispas_modules/

service.rs

use async_trait::async_trait;
use serde_json::Value;
use std::any::Any;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
use tokio_tungstenite::tungstenite::protocol::Message;
use tokio_tungstenite::WebSocketStream;
use futures_util::stream::SplitSink;
use tokio::sync::RwLock;

/// Supertrait that bundles the async IO bounds we need for an erased stream.
///
/// Rust trait objects only allow **one** non-auto trait, so we cannot write
/// `Box<dyn AsyncRead + AsyncWrite + …>`. This supertrait collapses the
/// bounds into a single name that is trait-object-compatible.
///
/// The blanket impl covers every concrete stream type we care about
/// (`TcpStream`, `TlsStream<TcpStream>`), so wrapping is a zero-cost
/// `Box::new(stream)`.
pub trait AsyncStream: AsyncRead + AsyncWrite + Unpin + Send {}
impl<T: AsyncRead + AsyncWrite + Unpin + Send> AsyncStream for T {}

/// Type-erased async stream used throughout the WebSocket layer.
///
/// The local WebSocket listener in `pispas-modules` operates in dual mode:
/// it accepts plain TCP (`ws://`) **and** TLS (`wss://`) on the same port.
/// The two concrete stream types are different (`TcpStream` vs
/// `TlsStream<TcpStream>`), but every service in the workspace needs to
/// store a write-half of the socket in a single concrete type.
///
/// `BoxedStream` wraps the stream in `Box<dyn AsyncStream>` and forwards
/// `poll_read`, `poll_write`, `poll_flush`, `poll_shutdown` to the inner
/// stream, so downstream code is transport-agnostic.
///
/// See `docs/DESIGN.md §2` for the reasoning and `docs/ARCHITECTURE.md §2`
/// for the WebSocket topology.
pub struct BoxedStream(pub Box<dyn AsyncStream>);

impl AsyncRead for BoxedStream {
    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
        Pin::new(&mut *self.0).poll_read(cx, buf)
    }
}

impl AsyncWrite for BoxedStream {
    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<std::io::Result<usize>> {
        Pin::new(&mut *self.0).poll_write(cx, buf)
    }
    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        Pin::new(&mut *self.0).poll_flush(cx)
    }
    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        Pin::new(&mut *self.0).poll_shutdown(cx)
    }
}

impl Unpin for BoxedStream {}

/// Handle used by services to push WebSocket messages back to a client.
///
/// * `Some(lock)` — the service was invoked from the **local** WebSocket,
///   which supports async back-channel notifications. Services can acquire
///   the write lock and push arbitrary `Message::Text(_)` frames.
/// * `None` — the service was invoked from the **remote** WebSocket
///   (outbound connection to `wss.unpispas.es`). There is no streaming
///   back-channel; the service must respond only through the return value
///   of `run()`.
///
/// Use `send_message` from the `pispas_modules::utils` module to write
/// through the lock safely.
pub type WebSocketWrite = Option<Arc<RwLock<SplitSink<WebSocketStream<BoxedStream>, Message>>>>;

/// Common interface every business module in `pispas-modules` implements.
///
/// Incoming WebSocket messages are dispatched to an implementation by the
/// `TARGET` field of the JSON payload. The trait stays deliberately small
/// so adding a new module is cheap: implement [`Service::run`], return the
/// static version via [`Service::get_version`], hook it up in
/// `load_services`, and document it in `docs/MODULES.md`.
///
/// ## Example skeleton
///
/// ```ignore
/// use async_trait::async_trait;
/// use serde_json::Value;
/// use pispas_modules::prelude::{Service, WebSocketWrite};
///
/// pub struct MyService;
///
/// #[async_trait]
/// impl Service for MyService {
///     async fn run(&self, action: Value, _write: WebSocketWrite) -> (i32, String) {
///         match action.get("ACTION").and_then(|v| v.as_str()) {
///             Some("PING") => (0, "pong".into()),
///             _            => (1, "unknown action".into()),
///         }
///     }
///
///     fn as_any(&self) -> &dyn std::any::Any { self }
///     fn stop_service(&self) {}
///     fn get_version(&self) -> String { env!("CARGO_PKG_VERSION").into() }
/// }
/// ```
#[async_trait]
pub trait Service: Send + Sync {
    /// Executes an action for the service.
    ///
    /// # Arguments
    ///
    /// * `action` - A `Value` object representing the action to be performed.
    ///   The structure of the action depends on the service implementation.
    /// * `write` - A `WebSocketWrite` object for sending responses or notifications
    ///   via a WebSocket connection.
    ///
    /// # Returns
    ///
    /// A tuple containing:
    /// * `i32` - Status code (e.g., 0 for success, 1 for error).
    /// * `String` - A descriptive message or response related to the action.
    ///
    /// # Notes
    ///
    /// This method must be implemented asynchronously by the service.
    async fn run(
        &self,
        action: Value,
        write: WebSocketWrite,
    ) -> (i32, String);

    /// Allows dynamic casting of the service to a specific type.
    ///
    /// # Returns
    ///
    /// A reference to `dyn Any`, which can be used for type-safe downcasting.
    #[allow(dead_code)]
    fn as_any(&self) -> &dyn Any;

    /// Stops the service and cleans up resources.
    ///
    /// # Notes
    ///
    /// This method should ensure that all tasks and processes related to the
    /// service are properly terminated. It can also reset any internal states.
    #[allow(dead_code)]
    fn stop_service(&self);

    /// Retrieves the version of the service.
    ///
    /// # Returns
    ///
    /// A `String` containing the version information of the service.
    fn get_version(&self) -> String;
}