livekit 0.7.42

// Copyright 2026 LiveKit, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

mod client;
mod server;

#[cfg(test)]
mod tests;

pub use client::RpcClientManager;
pub use server::{HandleRequestOptions, RpcServerManager};

use crate::data_stream::{StreamResult, StreamTextOptions, TextStreamInfo};
use crate::room::id::ParticipantIdentity;
use livekit_protocol::RpcError as RpcError_Proto;
use std::{error::Error, fmt::Display, future::Future, time::Duration};

// RPC protocol version constants (distinct from client_protocol; this is the
// version field on RpcRequest / v2 stream attributes).
pub(crate) const RPC_VERSION_V1: u32 = 1;
pub(crate) const RPC_VERSION_V2: u32 = 2;

// Data stream topic constants for RPC v2
pub(crate) const RPC_REQUEST_TOPIC: &str = "lk.rpc_request";
pub(crate) const RPC_RESPONSE_TOPIC: &str = "lk.rpc_response";

// Stream attribute keys for RPC v2
pub(crate) const ATTR_REQUEST_ID: &str = "lk.rpc_request_id";
pub(crate) const ATTR_METHOD: &str = "lk.rpc_request_method";
pub(crate) const ATTR_RESPONSE_TIMEOUT_MS: &str = "lk.rpc_request_response_timeout_ms";
pub(crate) const ATTR_VERSION: &str = "lk.rpc_request_version";

/// Transport abstraction for RPC operations.
///
/// Decouples the RPC managers from concrete engine/session types,
/// enabling in-memory unit testing with a mock transport.
pub(crate) trait RpcTransport: Send + Sync {
    /// Send a data packet (used for v1 RPC packets and ACKs).
    fn publish_data(
        &self,
        data: livekit_protocol::DataPacket,
    ) -> impl Future<Output = Result<(), crate::room::RoomError>> + Send;

    /// Send text as a data stream (used for v2 RPC requests and responses).
    fn send_text(
        &self,
        text: &str,
        options: StreamTextOptions,
    ) -> impl Future<Output = StreamResult<TextStreamInfo>> + Send;

    /// Look up a remote participant's client_protocol value.
    fn remote_client_protocol(&self, identity: &ParticipantIdentity) -> i32;

    /// Get the server version string, if available.
    fn server_version(&self) -> Option<String>;
}

/// Production implementation of `RpcTransport` backed by a `RoomSession`.
pub(crate) struct SessionTransport(pub(crate) std::sync::Arc<super::RoomSession>);

impl RpcTransport for SessionTransport {
    async fn publish_data(
        &self,
        data: livekit_protocol::DataPacket,
    ) -> Result<(), crate::room::RoomError> {
        self.0
            .rtc_engine
            .publish_data(data, crate::DataPacketKind::Reliable, false)
            .await
            .map_err(Into::into)
    }

    async fn send_text(
        &self,
        text: &str,
        options: StreamTextOptions,
    ) -> StreamResult<TextStreamInfo> {
        self.0.outgoing_stream_manager.send_text(text, options).await
    }

    fn remote_client_protocol(&self, identity: &ParticipantIdentity) -> i32 {
        self.0.get_remote_client_protocol(identity)
    }

    fn server_version(&self) -> Option<String> {
        self.0
            .rtc_engine
            .session()
            .signal_client()
            .join_response()
            .server_info
            .and_then(|info| info.version.is_empty().then(|| info.version))
    }
}

/// Parameters for performing an RPC call
#[derive(Debug, Clone)]
pub struct PerformRpcData {
    pub destination_identity: String,
    pub method: String,
    pub payload: String,
    pub response_timeout: Duration,
}

impl Default for PerformRpcData {
    fn default() -> Self {
        Self {
            destination_identity: Default::default(),
            method: Default::default(),
            payload: Default::default(),
            response_timeout: Duration::from_secs(15),
        }
    }
}

/// Data passed to method handler for incoming RPC invocations
///
/// Attributes:
///     request_id (String): The unique request ID. Will match at both sides of the call, useful for debugging or logging.
///     caller_identity (ParticipantIdentity): The unique participant identity of the caller.
///     payload (String): The payload of the request. User-definable format, typically JSON.
///     response_timeout (Duration): The maximum time the caller will wait for a response.
#[derive(Debug, Clone)]
pub struct RpcInvocationData {
    pub request_id: String,
    pub caller_identity: ParticipantIdentity,
    pub payload: String,
    pub response_timeout: Duration,
}

/// Specialized error handling for RPC methods.
///
/// Instances of this type, when thrown in a method handler, will have their `message`
/// serialized and sent across the wire. The caller will receive an equivalent error on the other side.
///
/// Built-in types are included but developers may use any string, with a max length of 256 bytes.
#[derive(Debug, Clone)]
pub struct RpcError {
    pub code: u32,
    pub message: String,
    pub data: Option<String>,
}

impl RpcError {
    pub const MAX_MESSAGE_BYTES: usize = 256;
    pub const MAX_DATA_BYTES: usize = MAX_V1_PAYLOAD_BYTES;

    /// Creates an error object with the given code and message, plus an optional data payload.
    ///
    /// If thrown in an RPC method handler, the error will be sent back to the caller.
    ///
    /// Error codes 1001-1999 are reserved for built-in errors (see RpcErrorCode for their meanings).
    pub fn new(code: u32, message: String, data: Option<String>) -> Self {
        Self {
            code,
            message: truncate_bytes(&message, Self::MAX_MESSAGE_BYTES),
            data: data.map(|d| truncate_bytes(&d, Self::MAX_DATA_BYTES)),
        }
    }

    pub fn from_proto(proto: RpcError_Proto) -> Self {
        Self::new(proto.code, proto.message, Some(proto.data))
    }

    pub fn to_proto(&self) -> RpcError_Proto {
        RpcError_Proto {
            code: self.code,
            message: self.message.clone(),
            data: self.data.clone().unwrap_or_default(),
        }
    }
}

impl Display for RpcError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "RPC Error: {} ({})", self.message, self.code)
    }
}
impl Error for RpcError {}

#[derive(Debug, Clone, Copy)]
pub enum RpcErrorCode {
    ApplicationError = 1500,
    ConnectionTimeout = 1501,
    ResponseTimeout = 1502,
    RecipientDisconnected = 1503,
    ResponsePayloadTooLarge = 1504,
    SendFailed = 1505,

    UnsupportedMethod = 1400,
    RecipientNotFound = 1401,
    RequestPayloadTooLarge = 1402,
    UnsupportedServer = 1403,
    UnsupportedVersion = 1404,
}

impl RpcErrorCode {
    pub(crate) fn message(&self) -> &'static str {
        match self {
            Self::ApplicationError => "Application error in method handler",
            Self::ConnectionTimeout => "Connection timeout",
            Self::ResponseTimeout => "Response timeout",
            Self::RecipientDisconnected => "Recipient disconnected",
            Self::ResponsePayloadTooLarge => "Response payload too large",
            Self::SendFailed => "Failed to send",

            Self::UnsupportedMethod => "Method not supported at destination",
            Self::RecipientNotFound => "Recipient not found",
            Self::RequestPayloadTooLarge => "Request payload too large",
            Self::UnsupportedServer => "RPC not supported by server",
            Self::UnsupportedVersion => "Unsupported RPC version",
        }
    }
}

impl RpcError {
    /// Creates an error object from the code, with an auto-populated message.
    pub(crate) fn built_in(code: RpcErrorCode, data: Option<String>) -> Self {
        Self::new(code as u32, code.message().to_string(), data)
    }
}

/// Maximum payload size in bytes for RPC v1
pub const MAX_V1_PAYLOAD_BYTES: usize = 15360; // 15 KB

/// Calculate the byte length of a string
pub(crate) fn byte_length(s: &str) -> usize {
    s.as_bytes().len()
}

/// Truncate a string to a maximum number of bytes
pub(crate) fn truncate_bytes(s: &str, max_bytes: usize) -> String {
    if byte_length(s) <= max_bytes {
        return s.to_string();
    }

    let mut result = String::new();
    for c in s.chars() {
        if byte_length(&(result.clone() + &c.to_string())) > max_bytes {
            break;
        }
        result.push(c);
    }
    result
}