lxc-streamutils/apps/server-rust/src/components.rs

//! Extensible component registry and event-processing contracts.
//!
//! A component kind supplies a versioned settings definition, a pure browser
//! projection and optional durable handlers. The live provider and router know
//! only these traits, so adding a gift wall or song-request component does not
//! require branching on component kinds in the ingestion pipeline.

use std::{
    collections::{BTreeSet, HashMap},
    error::Error,
    fmt,
    future::Future,
    pin::Pin,
    sync::{Arc, RwLock},
};

use serde::{Deserialize, Serialize};
use serde_json::Value;
use uuid::Uuid;

use crate::{
    domain::{ComponentMessage, LiveEvent, LiveEventKind},
    gift_effect::GiftEffectDefinition,
    gift_menu::{GiftMenuDefinition, GiftMenuProjection},
    guard_effect::GuardEffectDefinition,
    overlay::OverlaySettings,
    song_request::{SongRequestDefinition, SongRequestProjection},
};

pub const DANMAKU_OVERLAY_KIND: &str = "danmaku_overlay";

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ComponentError {
    EmptyKind,
    AlreadyRegistered(String),
    NotRegistered(String),
    InvalidSettings {
        kind: String,
        detail: String,
    },
    UnsupportedSettingsVersion {
        kind: String,
        found: u32,
        expected: u32,
    },
    Projection(String),
    Handler(String),
}

impl fmt::Display for ComponentError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::EmptyKind => formatter.write_str("component kind cannot be empty"),
            Self::AlreadyRegistered(kind) => {
                write!(formatter, "component kind `{kind}` is already registered")
            }
            Self::NotRegistered(kind) => {
                write!(formatter, "component kind `{kind}` is not registered")
            }
            Self::InvalidSettings { kind, detail } => {
                write!(
                    formatter,
                    "invalid settings for component `{kind}`: {detail}"
                )
            }
            Self::UnsupportedSettingsVersion {
                kind,
                found,
                expected,
            } => write!(
                formatter,
                "component `{kind}` settings version {found} is unsupported; expected {expected}"
            ),
            Self::Projection(detail) => write!(formatter, "component projection failed: {detail}"),
            Self::Handler(detail) => write!(formatter, "component event handler failed: {detail}"),
        }
    }
}

impl Error for ComponentError {}

/// Persisted component instance. `owner_id` is trusted tenancy context and is
/// omitted from public serialization.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ComponentInstance {
    pub id: Uuid,
    #[serde(skip_serializing)]
    pub owner_id: Uuid,
    /// Runtime identity of the owner's account-level source. This value is
    /// derived while loading the component and is not stored on its row.
    #[serde(rename = "sourceId")]
    pub account_source_id: Uuid,
    pub kind: String,
    pub name: String,
    pub enabled: bool,
    pub settings_version: u32,
    pub settings: Value,
}

impl ComponentInstance {
    pub fn new(
        owner_id: Uuid,
        account_source_id: Uuid,
        kind: impl Into<String>,
        name: impl Into<String>,
        settings_version: u32,
        settings: Value,
    ) -> Self {
        Self {
            id: Uuid::new_v4(),
            owner_id,
            account_source_id,
            kind: kind.into(),
            name: name.into(),
            enabled: true,
            settings_version,
            settings,
        }
    }
}

#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct EventSubscription {
    kinds: BTreeSet<LiveEventKind>,
}

impl EventSubscription {
    pub fn new(kinds: impl IntoIterator<Item = LiveEventKind>) -> Self {
        Self {
            kinds: kinds.into_iter().collect(),
        }
    }

    pub fn matches(&self, event: &LiveEvent) -> bool {
        self.kinds.contains(&event.kind())
    }

    pub fn contains(&self, kind: LiveEventKind) -> bool {
        self.kinds.contains(&kind)
    }

    pub fn kinds(&self) -> impl Iterator<Item = LiveEventKind> + '_ {
        self.kinds.iter().copied()
    }
}

/// Static behavior and settings contract for one component kind.
pub trait ComponentDefinition: Send + Sync {
    fn kind(&self) -> &'static str;
    fn settings_version(&self) -> u32;
    fn default_settings(&self) -> Value;
    fn validate_settings(&self, settings: Value) -> Result<Value, ComponentError>;
    fn subscriptions(&self, settings: &Value) -> Result<EventSubscription, ComponentError>;

    /// Override when a component changes its settings schema. Keeping migration
    /// here allows old instances to be upgraded without teaching the router
    /// about component-specific fields.
    fn migrate_settings(
        &self,
        from_version: u32,
        settings: Value,
    ) -> Result<Value, ComponentError> {
        if from_version == self.settings_version() {
            Ok(settings)
        } else {
            Err(ComponentError::UnsupportedSettingsVersion {
                kind: self.kind().to_owned(),
                found: from_version,
                expected: self.settings_version(),
            })
        }
    }
}

/// A passive, side-effect-free transformation for a browser-facing component.
/// It may filter or reshape an event, but must not write business data.
pub trait EventProjection: Send + Sync {
    fn project(
        &self,
        component: &ComponentInstance,
        event: &LiveEvent,
    ) -> Result<Option<ComponentMessage>, ComponentError>;
}

/// Future returned by a durable business handler without requiring an
/// `async-trait` dependency.
pub type HandlerFuture<'a> = Pin<Box<dyn Future<Output = Result<(), ComponentError>> + Send + 'a>>;

/// Future used by component-specific durable state snapshots during WebSocket
/// authentication. Snapshot messages share the normal component envelope.
pub type SnapshotFuture<'a> =
    Pin<Box<dyn Future<Output = Result<Vec<ComponentMessage>, ComponentError>> + Send + 'a>>;

pub trait ComponentSnapshotProvider: Send + Sync {
    fn snapshot<'a>(
        &'a self,
        component: &'a ComponentInstance,
        room_id: &'a str,
    ) -> SnapshotFuture<'a>;
}

/// An active handler may perform durable side effects (for example recording a
/// song request). It runs independently of WebSocket receiver count and should
/// implement idempotency in its persistence layer.
pub trait EventHandler: Send + Sync {
    fn name(&self) -> &'static str;

    fn accepts(&self, _component: &ComponentInstance, _event: &LiveEvent) -> bool {
        true
    }

    fn handle<'a>(
        &'a self,
        component: &'a ComponentInstance,
        event: Arc<LiveEvent>,
    ) -> HandlerFuture<'a>;
}

#[derive(Default)]
pub struct PassthroughProjection;

impl EventProjection for PassthroughProjection {
    fn project(
        &self,
        component: &ComponentInstance,
        event: &LiveEvent,
    ) -> Result<Option<ComponentMessage>, ComponentError> {
        ComponentMessage::from_live_event(component.id, event)
            .map(Some)
            .map_err(|error| ComponentError::Projection(error.to_string()))
    }
}

pub struct DanmakuOverlayDefinition;

impl DanmakuOverlayDefinition {
    fn parse(&self, settings: Value) -> Result<OverlaySettings, ComponentError> {
        serde_json::from_value(settings).map_err(|error| ComponentError::InvalidSettings {
            kind: DANMAKU_OVERLAY_KIND.to_owned(),
            detail: error.to_string(),
        })
    }
}

impl ComponentDefinition for DanmakuOverlayDefinition {
    fn kind(&self) -> &'static str {
        DANMAKU_OVERLAY_KIND
    }

    fn settings_version(&self) -> u32 {
        1
    }

    fn default_settings(&self) -> Value {
        serde_json::to_value(OverlaySettings::default())
            .expect("OverlaySettings is always JSON serializable")
    }

    fn validate_settings(&self, settings: Value) -> Result<Value, ComponentError> {
        let settings = self.parse(settings)?.sanitize();
        serde_json::to_value(settings).map_err(|error| ComponentError::InvalidSettings {
            kind: DANMAKU_OVERLAY_KIND.to_owned(),
            detail: error.to_string(),
        })
    }

    fn subscriptions(&self, settings: &Value) -> Result<EventSubscription, ComponentError> {
        let settings = self.parse(settings.clone())?;
        let mut kinds = Vec::with_capacity(9);
        if settings.show_danmaku {
            kinds.push(LiveEventKind::Danmaku);
        }
        if settings.show_enter {
            kinds.push(LiveEventKind::Enter);
        }
        if settings.show_gift {
            kinds.extend([LiveEventKind::Gift, LiveEventKind::GiftCombo]);
        }
        if settings.show_superchat {
            kinds.push(LiveEventKind::SuperChat);
        }
        if settings.show_guard {
            kinds.push(LiveEventKind::GuardPurchase);
        }
        if settings.show_like {
            kinds.push(LiveEventKind::Like);
        }
        if settings.show_share {
            kinds.push(LiveEventKind::Share);
        }
        Ok(EventSubscription::new(kinds))
    }
}

/// Immutable routing snapshot returned by the registry. All contained trait
/// objects are `Arc`, so routing never holds the registry lock across awaits.
#[derive(Clone)]
pub struct ComponentRuntime {
    definition: Arc<dyn ComponentDefinition>,
    projection: Arc<dyn EventProjection>,
    handlers: Vec<Arc<dyn EventHandler>>,
    snapshot_provider: Option<Arc<dyn ComponentSnapshotProvider>>,
}

impl ComponentRuntime {
    pub fn kind(&self) -> &'static str {
        self.definition.kind()
    }

    pub fn definition(&self) -> Arc<dyn ComponentDefinition> {
        self.definition.clone()
    }

    pub fn validated_settings(
        &self,
        instance: &ComponentInstance,
    ) -> Result<Value, ComponentError> {
        let settings = self
            .definition
            .migrate_settings(instance.settings_version, instance.settings.clone())?;
        self.definition.validate_settings(settings)
    }

    pub fn subscriptions(
        &self,
        instance: &ComponentInstance,
    ) -> Result<EventSubscription, ComponentError> {
        let settings = self.validated_settings(instance)?;
        self.definition.subscriptions(&settings)
    }

    pub fn project(
        &self,
        instance: &ComponentInstance,
        event: &LiveEvent,
    ) -> Result<Option<ComponentMessage>, ComponentError> {
        self.projection.project(instance, event)
    }

    pub fn handlers(&self) -> Vec<Arc<dyn EventHandler>> {
        self.handlers.clone()
    }

    pub async fn snapshot(
        &self,
        component: &ComponentInstance,
        room_id: &str,
    ) -> Result<Vec<ComponentMessage>, ComponentError> {
        match &self.snapshot_provider {
            Some(provider) => provider.snapshot(component, room_id).await,
            None => Ok(Vec::new()),
        }
    }
}

#[derive(Clone)]
pub struct ComponentRegistry {
    entries: Arc<RwLock<HashMap<String, ComponentRuntime>>>,
}

impl ComponentRegistry {
    /// Create an empty registry for tests or applications that select their own
    /// component modules.
    pub fn new() -> Self {
        Self {
            entries: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// Registry used by the current application. Future modules can be added by
    /// calling `register` during bootstrap.
    pub fn with_builtin_components() -> Self {
        let registry = Self::new();
        registry
            .register(
                Arc::new(DanmakuOverlayDefinition),
                Arc::new(PassthroughProjection),
            )
            .expect("built-in component kinds are unique");
        registry
            .register(
                Arc::new(SongRequestDefinition),
                Arc::new(SongRequestProjection),
            )
            .expect("built-in component kinds are unique");
        registry
            .register(
                Arc::new(GiftEffectDefinition),
                Arc::new(PassthroughProjection),
            )
            .expect("built-in component kinds are unique");
        registry
            .register(
                Arc::new(GuardEffectDefinition),
                Arc::new(PassthroughProjection),
            )
            .expect("built-in component kinds are unique");
        registry
            .register(Arc::new(GiftMenuDefinition), Arc::new(GiftMenuProjection))
            .expect("built-in component kinds are unique");
        registry
    }

    pub fn register(
        &self,
        definition: Arc<dyn ComponentDefinition>,
        projection: Arc<dyn EventProjection>,
    ) -> Result<(), ComponentError> {
        let kind = definition.kind().trim();
        if kind.is_empty() {
            return Err(ComponentError::EmptyKind);
        }
        let mut entries = self
            .entries
            .write()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if entries.contains_key(kind) {
            return Err(ComponentError::AlreadyRegistered(kind.to_owned()));
        }
        entries.insert(
            kind.to_owned(),
            ComponentRuntime {
                definition,
                projection,
                handlers: Vec::new(),
                snapshot_provider: None,
            },
        );
        Ok(())
    }

    pub fn register_handler(
        &self,
        kind: &str,
        handler: Arc<dyn EventHandler>,
    ) -> Result<(), ComponentError> {
        let mut entries = self
            .entries
            .write()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let runtime = entries
            .get_mut(kind)
            .ok_or_else(|| ComponentError::NotRegistered(kind.to_owned()))?;
        runtime.handlers.push(handler);
        Ok(())
    }

    pub fn register_snapshot_provider(
        &self,
        kind: &str,
        provider: Arc<dyn ComponentSnapshotProvider>,
    ) -> Result<(), ComponentError> {
        let mut entries = self
            .entries
            .write()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let runtime = entries
            .get_mut(kind)
            .ok_or_else(|| ComponentError::NotRegistered(kind.to_owned()))?;
        runtime.snapshot_provider = Some(provider);
        Ok(())
    }

    pub fn runtime(&self, kind: &str) -> Result<ComponentRuntime, ComponentError> {
        self.entries
            .read()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .get(kind)
            .cloned()
            .ok_or_else(|| ComponentError::NotRegistered(kind.to_owned()))
    }

    pub fn validate_settings(
        &self,
        kind: &str,
        from_version: u32,
        settings: Value,
    ) -> Result<Value, ComponentError> {
        let runtime = self.runtime(kind)?;
        let settings = runtime
            .definition
            .migrate_settings(from_version, settings)?;
        runtime.definition.validate_settings(settings)
    }

    pub fn kinds(&self) -> Vec<String> {
        let mut kinds: Vec<_> = self
            .entries
            .read()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .keys()
            .cloned()
            .collect();
        kinds.sort();
        kinds
    }
}

impl Default for ComponentRegistry {
    fn default() -> Self {
        Self::with_builtin_components()
    }
}

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

    #[test]
    fn builtin_overlay_settings_are_sanitized_and_define_subscriptions() {
        let registry = ComponentRegistry::default();
        let mut settings = serde_json::to_value(OverlaySettings::default()).unwrap();
        settings["maxVisible"] = Value::from(250);
        settings["showGift"] = Value::Bool(false);
        settings["showLike"] = Value::Bool(true);

        let validated = registry
            .validate_settings(DANMAKU_OVERLAY_KIND, 1, settings)
            .unwrap();
        assert_eq!(validated["maxVisible"], 12);

        let instance = ComponentInstance::new(
            Uuid::new_v4(),
            Uuid::new_v4(),
            DANMAKU_OVERLAY_KIND,
            "弹幕姬",
            1,
            validated,
        );
        let subscriptions = registry
            .runtime(DANMAKU_OVERLAY_KIND)
            .unwrap()
            .subscriptions(&instance)
            .unwrap();
        assert!(subscriptions.contains(LiveEventKind::Danmaku));
        assert!(subscriptions.contains(LiveEventKind::Like));
        assert!(!subscriptions.contains(LiveEventKind::Gift));
        assert!(!subscriptions.contains(LiveEventKind::GiftCombo));
    }

    #[test]
    fn builtin_song_request_is_registered_with_bounded_unlimited_defaults() {
        let registry = ComponentRegistry::default();
        let runtime = registry.runtime("song_request").unwrap();
        let mut settings = runtime.definition().default_settings();
        settings["scrollSpeedPixelsPerSecond"] = Value::from(999);
        let validated = registry
            .validate_settings("song_request", 1, settings)
            .unwrap();
        assert_eq!(validated["scrollSpeedPixelsPerSecond"], 200);
        assert_eq!(validated["maxQueueSize"], 0);
        assert_eq!(validated["maxRequestsPerViewer"], 0);
        assert_eq!(validated["requestCooldownSeconds"], 0);

        let instance = ComponentInstance::new(
            Uuid::new_v4(),
            Uuid::new_v4(),
            "song_request",
            "点歌姬",
            1,
            validated,
        );
        let subscriptions = runtime.subscriptions(&instance).unwrap();
        assert!(subscriptions.contains(LiveEventKind::Danmaku));
        assert!(!subscriptions.contains(LiveEventKind::Gift));
    }

    #[test]
    fn builtin_gift_effect_only_subscribes_to_gifts() {
        let registry = ComponentRegistry::default();
        assert!(registry.kinds().contains(&"gift_effect".to_owned()));
        let runtime = registry.runtime("gift_effect").unwrap();
        let instance = ComponentInstance::new(
            Uuid::new_v4(),
            Uuid::new_v4(),
            "gift_effect",
            "礼物星雨",
            1,
            runtime.definition().default_settings(),
        );
        let subscriptions = runtime.subscriptions(&instance).unwrap();
        assert!(subscriptions.contains(LiveEventKind::Gift));
        assert!(!subscriptions.contains(LiveEventKind::GuardPurchase));
        assert!(!subscriptions.contains(LiveEventKind::GiftCombo));
    }

    #[test]
    fn builtin_guard_effect_only_subscribes_to_guard_purchases() {
        let registry = ComponentRegistry::default();
        assert!(registry.kinds().contains(&"guard_effect".to_owned()));
        let runtime = registry.runtime("guard_effect").unwrap();
        let instance = ComponentInstance::new(
            Uuid::new_v4(),
            Uuid::new_v4(),
            "guard_effect",
            "大航海特效",
            1,
            runtime.definition().default_settings(),
        );
        let subscriptions = runtime.subscriptions(&instance).unwrap();
        assert!(subscriptions.contains(LiveEventKind::GuardPurchase));
        assert!(!subscriptions.contains(LiveEventKind::Gift));
        assert!(!subscriptions.contains(LiveEventKind::GiftCombo));
    }

    #[test]
    fn builtin_gift_menu_is_registered_as_a_gift_and_guard_projection() {
        let registry = ComponentRegistry::default();
        let runtime = registry.runtime("gift_menu").unwrap();
        let instance = ComponentInstance::new(
            Uuid::new_v4(),
            Uuid::new_v4(),
            "gift_menu",
            "礼物菜单",
            1,
            runtime.definition().default_settings(),
        );
        let subscriptions = runtime.subscriptions(&instance).unwrap();
        assert!(subscriptions.contains(LiveEventKind::Gift));
        assert!(subscriptions.contains(LiveEventKind::GuardPurchase));
        assert!(!subscriptions.contains(LiveEventKind::GiftCombo));
    }

    #[test]
    fn duplicate_component_kinds_are_rejected() {
        let registry = ComponentRegistry::default();
        let result = registry.register(
            Arc::new(DanmakuOverlayDefinition),
            Arc::new(PassthroughProjection),
        );
        assert!(matches!(result, Err(ComponentError::AlreadyRegistered(_))));
    }

    #[test]
    fn unsupported_settings_versions_fail_closed() {
        let registry = ComponentRegistry::default();
        let result = registry.validate_settings(
            DANMAKU_OVERLAY_KIND,
            99,
            serde_json::to_value(OverlaySettings::default()).unwrap(),
        );
        assert!(matches!(
            result,
            Err(ComponentError::UnsupportedSettingsVersion { .. })
        ));
    }
}