livekit-data-stream 0.1.1

// 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.

use from_variants::FromVariants;
use livekit_common::EncryptionType;
use livekit_protocol::data_stream as proto;
use std::collections::HashMap;

use super::StreamId;

/// Operation type for text streams.
#[derive(Clone, Copy, Default, Debug, Hash, Eq, PartialEq)]
pub enum OperationType {
    #[default]
    Create,
    Update,
    Delete,
    Reaction,
}

impl From<proto::OperationType> for OperationType {
    fn from(value: proto::OperationType) -> Self {
        match value {
            proto::OperationType::Create => Self::Create,
            proto::OperationType::Update => Self::Update,
            proto::OperationType::Delete => Self::Delete,
            proto::OperationType::Reaction => Self::Reaction,
        }
    }
}

impl From<OperationType> for proto::OperationType {
    fn from(value: OperationType) -> Self {
        match value {
            OperationType::Create => Self::Create,
            OperationType::Update => Self::Update,
            OperationType::Delete => Self::Delete,
            OperationType::Reaction => Self::Reaction,
        }
    }
}

/// Header information included exclusively in text data streams
#[derive(Clone, Debug, Default, PartialEq)]
pub struct TextHeader {
    pub(crate) operation_type: OperationType,
    /// Optional: Version for updates/edits
    pub(crate) version: i32,
    /// Optional: Reply to specific message
    pub(crate) reply_to_stream_id: Option<StreamId>,
    /// file attachments for text streams
    pub(crate) attached_stream_ids: Vec<StreamId>,
    /// true if the text has been generated by an agent from a participant's audio transcription
    pub(crate) generated: bool,
}

impl From<proto::TextHeader> for TextHeader {
    fn from(value: proto::TextHeader) -> Self {
        Self {
            operation_type: value.operation_type().into(),
            version: value.version,
            reply_to_stream_id: if !value.reply_to_stream_id.is_empty() {
                Some(value.reply_to_stream_id.into())
            } else {
                None
            },
            attached_stream_ids: value.attached_stream_ids.into_iter().map(Into::into).collect(),
            generated: value.generated,
        }
    }
}

impl From<TextHeader> for proto::TextHeader {
    fn from(value: TextHeader) -> Self {
        Self {
            operation_type: proto::OperationType::from(value.operation_type) as i32,
            version: value.version,
            reply_to_stream_id: value.reply_to_stream_id.map(Into::into).unwrap_or_default(),
            attached_stream_ids: value.attached_stream_ids.into_iter().map(Into::into).collect(),
            generated: value.generated,
        }
    }
}

/// Header information included exclusively in byte data streams
#[derive(Clone, Debug, Default, PartialEq)]
pub struct ByteHeader {
    pub(crate) name: String,
}

impl From<proto::ByteHeader> for ByteHeader {
    fn from(value: proto::ByteHeader) -> Self {
        Self { name: value.name }
    }
}

impl From<ByteHeader> for proto::ByteHeader {
    fn from(value: ByteHeader) -> Self {
        Self { name: value.name }
    }
}

#[derive(Clone, Debug, PartialEq, FromVariants)]
pub enum ContentHeader {
    TextHeader(TextHeader),
    ByteHeader(ByteHeader),
}

impl From<proto::header::ContentHeader> for ContentHeader {
    fn from(value: proto::header::ContentHeader) -> Self {
        match value {
            proto::header::ContentHeader::TextHeader(text_header) => {
                Self::TextHeader(text_header.into())
            }
            proto::header::ContentHeader::ByteHeader(text_header) => {
                Self::ByteHeader(text_header.into())
            }
        }
    }
}

impl From<ContentHeader> for proto::header::ContentHeader {
    fn from(value: ContentHeader) -> Self {
        match value {
            ContentHeader::TextHeader(text_header) => Self::TextHeader(text_header.into()),
            ContentHeader::ByteHeader(byte_header) => Self::ByteHeader(byte_header.into()),
        }
    }
}

/// Type of compression used when sending a data stream chunk
#[derive(Clone, Debug, Default, PartialEq)]
pub enum CompressionType {
    #[default]
    None,
    /// DEFLATE_RAW = DEFLATE without header+checksum/trailer
    DeflateRaw,
    /// A compression type this SDK version doesn't recognize (i.e. from a future protocol
    /// version). Streams carrying it cannot be decoded and are dropped on receive; the send
    /// path never constructs this variant.
    Unrecognized,
}

impl From<proto::CompressionType> for CompressionType {
    fn from(value: proto::CompressionType) -> Self {
        match value {
            proto::CompressionType::DeflateRaw => Self::DeflateRaw,
            proto::CompressionType::None => Self::None,
        }
    }
}

impl From<CompressionType> for proto::CompressionType {
    fn from(value: CompressionType) -> Self {
        match value {
            CompressionType::DeflateRaw => Self::DeflateRaw,
            // `Unrecognized` only arises from decoding a foreign header and is never sent.
            CompressionType::None | CompressionType::Unrecognized => Self::None,
        }
    }
}

#[derive(Clone, Debug, Default, PartialEq)]
pub struct Header {
    /// Unique identifier for this data stream
    pub(crate) stream_id: StreamId,
    /// using int64 for Unix timestamp
    pub(crate) timestamp: i64,
    pub(crate) topic: String,
    pub(crate) mime_type: String,
    /// only populated for finite streams, if it's a stream of unknown size this stays empty
    pub(crate) total_length: Option<u64>,
    /// user defined attributes map that can carry additional info
    pub(crate) attributes: HashMap<String, String>,
    /// Optional inline content so that a data stream can be sent as a single packet for short payloads.
    ///
    /// content as binary (bytes)
    pub(crate) inline_content: Option<Vec<u8>>,
    pub(crate) compression: CompressionType,
    /// oneof to choose between specific header types
    pub(crate) content_header: Option<ContentHeader>,
}

impl From<proto::Header> for Header {
    fn from(value: proto::Header) -> Self {
        // Don't use the prost `compression()` accessor here: it silently maps out-of-range
        // values (a compression type from a future protocol version) to the default `None`,
        // which would make the receiver deliver still-compressed bytes as content. Preserve
        // unknown values as `Unrecognized` so the incoming manager can drop the stream.
        let compression = proto::CompressionType::try_from(value.compression)
            .map(CompressionType::from)
            .unwrap_or(CompressionType::Unrecognized);
        let content_header: Option<ContentHeader> =
            value.content_header.map(|content_header| content_header.into());
        Self {
            stream_id: value.stream_id.into(),
            timestamp: value.timestamp,
            topic: value.topic,
            mime_type: value.mime_type,
            total_length: value.total_length,
            attributes: value.attributes,
            inline_content: value.inline_content,
            compression,
            content_header,
        }
    }
}

impl From<Header> for proto::Header {
    fn from(value: Header) -> Self {
        // `encryption_type` is deprecated on the proto (it's carried on the DataPacket instead);
        // `..Default::default()` fills it without naming the deprecated field.
        Self {
            stream_id: value.stream_id.into(),
            timestamp: value.timestamp,
            topic: value.topic,
            mime_type: value.mime_type,
            total_length: value.total_length,
            attributes: value.attributes,
            inline_content: value.inline_content,
            compression: proto::CompressionType::from(value.compression) as i32,
            content_header: value.content_header.map(Into::into),
            ..Default::default()
        }
    }
}

#[derive(Clone, Debug, Default, PartialEq)]
pub struct Chunk {
    /// Unique identifier for this data stream to map it to the correct header
    pub(crate) stream_id: StreamId,
    pub(crate) chunk_index: u64,
    /// Content as binary (bytes)
    pub(crate) content: Vec<u8>,
    /// A version indicating that this chunk_index has been retroactively modified and the original one needs to be replaced
    pub(crate) version: i32,
    pub(crate) encryption_type: EncryptionType,
}

impl From<proto::Chunk> for Chunk {
    fn from(value: proto::Chunk) -> Self {
        // The proto carries encryption on the enclosing `DataPacket`, not the chunk, so the
        // chunk's own `encryption_type` defaults here; the authoritative value rides on `Packet`.
        Self {
            stream_id: value.stream_id.into(),
            chunk_index: value.chunk_index,
            content: value.content,
            version: value.version,
            encryption_type: EncryptionType::default(),
        }
    }
}

impl From<Chunk> for proto::Chunk {
    fn from(value: Chunk) -> Self {
        // `iv` is deprecated on the proto (encryption rides on the DataPacket);
        // `..Default::default()` fills it without naming the deprecated field.
        Self {
            stream_id: value.stream_id.into(),
            chunk_index: value.chunk_index,
            content: value.content,
            version: value.version,
            ..Default::default()
        }
    }
}

#[derive(Clone, Debug, Default, PartialEq)]
pub struct Trailer {
    /// Unique identifier for this data stream
    pub(crate) stream_id: StreamId,
    /// Reason why the stream was closed (could contain "error" / "interrupted" / empty for expected end)
    pub(crate) reason: String,
    /// Any final attribute updates for the stream
    pub(crate) attributes: HashMap<String, String>,
}

impl From<proto::Trailer> for Trailer {
    fn from(value: proto::Trailer) -> Self {
        Self {
            stream_id: value.stream_id.into(),
            reason: value.reason,
            attributes: value.attributes,
        }
    }
}

impl From<Trailer> for proto::Trailer {
    fn from(value: Trailer) -> Self {
        Self {
            stream_id: value.stream_id.into(),
            reason: value.reason,
            attributes: value.attributes,
        }
    }
}

#[derive(Clone, Debug, PartialEq)]
pub enum Packet {
    Header { header: Header, encryption_type: EncryptionType },
    Chunk { chunk: Chunk, encryption_type: EncryptionType },
    Trailer(Trailer),
}