felis/lxc-streamutils
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

old backend core lib

Browse Source
This commit is contained in:
felis
2026-07-18 11:44:58 -07:00
parent 994854d104
commit f599672a30
36 changed files with 3776 additions and 276 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/server-rust/README.md
+17 -15
View File
@@ -5,27 +5,29 @@ crate 导出。
## 模块职责
| 模块 | 职责 |
| ------------- | -------------------------------------------------------- |
| `app` | 依赖组装、迁移、事件队列、源启动与旧配置导入 |
| `auth` | 邀请码、TOTP、恢复码、会话、secret 加密和组件 token |
| `components` | 组件定义、设置版本、订阅、projection 与 handler registry |
| `config` | TOML 解析、部署策略校验和旧单用户兼容字段 |
| `credentials` | CookieCloud URL 边界与 Bilibili Cookie 提取 |
| `db` | PostgreSQL pool、迁移和 RLS tenant context |
| `domain` | provider-independent event 与 WebSocket envelope |
| `http_api` | REST/WS、会话、权限、same-origin、静态资源与安全响应头 |
| `live` | provider trait、Bilibili adapter 与 source supervisor |
| `overlay` | 弹幕姬设置及礼物/表情目录 |
| `rate_limit` | 匿名登录和 enrollment 滥用限制 |
| `realtime` | source event routing 与 component-scoped fanout |
| `repository` | PostgreSQL component facade 和热路径缓存同步 |
| 模块 | 职责 |
| -------------- | -------------------------------------------------------- |
| `app` | 依赖组装、迁移、事件队列、源启动与旧配置导入 |
| `auth` | 邀请码、TOTP、恢复码、会话、secret 加密和组件 token |
| `components` | 组件定义、设置版本、订阅、projection 与 handler registry |
| `config` | TOML 解析、部署策略校验和旧单用户兼容字段 |
| `credentials` | CookieCloud URL 边界与 Bilibili Cookie 提取 |
| `db` | PostgreSQL pool、迁移和 RLS tenant context |
| `domain` | provider-independent event 与 WebSocket envelope |
| `http_api` | REST/WS、会话、权限、same-origin、静态资源与安全响应头 |
| `live` | provider trait、Bilibili adapter 与 source supervisor |
| `overlay` | 弹幕姬设置及礼物/表情目录 |
| `rate_limit` | 匿名登录和 enrollment 滥用限制 |
| `realtime` | source event routing 与 component-scoped fanout |
| `repository` | PostgreSQL component facade 和热路径缓存同步 |
| `song_request` | 点歌命令、事务队列、评分、快照和管理服务 |
## 重要不变量
- handler 不能信任请求体中的 owner；owner 必须来自 session 或 source context。
- tenant table 查询必须在 `Db::set_tenant` 后的事务中执行。
- provider 只能输出 canonical、bounded、sanitized `LiveEvent`。
- Bilibili listener 必须保持 20 秒心跳；socket 内部重连失败后由 adapter 重建完整客户端。
- projection 无副作用；可靠业务动作必须使用幂等 handler。
- token、邀请码和恢复码只存摘要，TOTP/CookieCloud Secret 只存认证加密密文。
- EventHub 的 channel key 是 component ID，不允许增加无权限的全局 receiver。
apps/server-rust/migrations/005_song_request.sql
+97
View File
@@ -0,0 +1,97 @@
-- Persistent, tenant-isolated state for the built-in song request component.
-- Component settings remain JSONB; relational queue and rating data live here
-- because they are durable business facts rather than renderer preferences.
CREATE UNIQUE INDEX IF NOT EXISTS component_instances_single_song_request
ON component_instances(owner_user_id, kind)
WHERE kind = 'song_request';
CREATE TABLE IF NOT EXISTS song_request_state (
owner_user_id UUID NOT NULL,
component_instance_id UUID PRIMARY KEY,
revision BIGINT NOT NULL DEFAULT 0 CHECK (revision >= 0),
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
UNIQUE (owner_user_id, component_instance_id),
FOREIGN KEY (owner_user_id, component_instance_id)
REFERENCES component_instances(owner_user_id, id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS song_requests (
id UUID PRIMARY KEY,
owner_user_id UUID NOT NULL,
component_instance_id UUID NOT NULL,
requester_uid TEXT NOT NULL CHECK (char_length(requester_uid) BETWEEN 1 AND 64),
requester_name TEXT NOT NULL CHECK (char_length(requester_name) BETWEEN 1 AND 80),
song_title TEXT NOT NULL CHECK (char_length(song_title) BETWEEN 1 AND 80),
normalized_song_title TEXT NOT NULL CHECK (char_length(normalized_song_title) BETWEEN 1 AND 80),
status TEXT NOT NULL CHECK (status IN ('current', 'queued', 'completed', 'cancelled')),
queue_position BIGINT NOT NULL DEFAULT 0 CHECK (queue_position >= 0),
source_event_id UUID NOT NULL,
requested_at TIMESTAMPTZ NOT NULL DEFAULT now(),
started_at TIMESTAMPTZ,
finished_at TIMESTAMPTZ,
UNIQUE (owner_user_id, id),
UNIQUE (owner_user_id, component_instance_id, id),
UNIQUE (component_instance_id, source_event_id),
FOREIGN KEY (owner_user_id, component_instance_id)
REFERENCES song_request_state(owner_user_id, component_instance_id) ON DELETE CASCADE,
CHECK ((status = 'queued' AND queue_position > 0 AND started_at IS NULL AND finished_at IS NULL)
OR (status = 'current' AND queue_position = 0 AND started_at IS NOT NULL AND finished_at IS NULL)
OR (status IN ('completed', 'cancelled') AND queue_position = 0 AND finished_at IS NOT NULL))
);
CREATE UNIQUE INDEX IF NOT EXISTS song_requests_one_current
ON song_requests(component_instance_id)
WHERE status = 'current';
CREATE UNIQUE INDEX IF NOT EXISTS song_requests_unique_active_title
ON song_requests(component_instance_id, normalized_song_title)
WHERE status IN ('current', 'queued');
CREATE INDEX IF NOT EXISTS song_requests_active_order
ON song_requests(owner_user_id, component_instance_id, status, queue_position);
CREATE INDEX IF NOT EXISTS song_requests_history
ON song_requests(owner_user_id, component_instance_id, finished_at DESC)
WHERE status IN ('completed', 'cancelled');
CREATE INDEX IF NOT EXISTS song_requests_requester_active
ON song_requests(owner_user_id, component_instance_id, requester_uid, requested_at DESC)
WHERE status IN ('current', 'queued');
CREATE TABLE IF NOT EXISTS song_ratings (
id UUID PRIMARY KEY,
owner_user_id UUID NOT NULL,
component_instance_id UUID NOT NULL,
song_request_id UUID NOT NULL,
viewer_uid TEXT NOT NULL CHECK (char_length(viewer_uid) BETWEEN 1 AND 64),
viewer_name TEXT NOT NULL CHECK (char_length(viewer_name) BETWEEN 1 AND 80),
score SMALLINT NOT NULL CHECK (score BETWEEN 1 AND 5),
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
UNIQUE (song_request_id, viewer_uid),
FOREIGN KEY (owner_user_id, component_instance_id)
REFERENCES song_request_state(owner_user_id, component_instance_id) ON DELETE CASCADE,
FOREIGN KEY (owner_user_id, component_instance_id, song_request_id)
REFERENCES song_requests(owner_user_id, component_instance_id, id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS song_ratings_request
ON song_ratings(owner_user_id, component_instance_id, song_request_id);
ALTER TABLE song_request_state ENABLE ROW LEVEL SECURITY;
ALTER TABLE song_request_state FORCE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS song_request_state_owner ON song_request_state;
CREATE POLICY song_request_state_owner ON song_request_state
USING (owner_user_id = NULLIF(current_setting('app.user_id', true), '')::UUID)
WITH CHECK (owner_user_id = NULLIF(current_setting('app.user_id', true), '')::UUID);
ALTER TABLE song_requests ENABLE ROW LEVEL SECURITY;
ALTER TABLE song_requests FORCE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS song_requests_owner ON song_requests;
CREATE POLICY song_requests_owner ON song_requests
USING (owner_user_id = NULLIF(current_setting('app.user_id', true), '')::UUID)
WITH CHECK (owner_user_id = NULLIF(current_setting('app.user_id', true), '')::UUID);
ALTER TABLE song_ratings ENABLE ROW LEVEL SECURITY;
ALTER TABLE song_ratings FORCE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS song_ratings_owner ON song_ratings;
CREATE POLICY song_ratings_owner ON song_ratings
USING (owner_user_id = NULLIF(current_setting('app.user_id', true), '')::UUID)
WITH CHECK (owner_user_id = NULLIF(current_setting('app.user_id', true), '')::UUID);
apps/server-rust/src/app.rs
+23
View File
@@ -27,6 +27,9 @@ use crate::{
rate_limit::AuthRateLimiter,
realtime::{EventHub, InMemoryComponentStore, SourceEventRouter},
repository::TenantRepository,
song_request::{
SONG_REQUEST_KIND, SongRequestHandler, SongRequestService, SongRequestSnapshotProvider,
},
};
#[derive(Clone)]
@@ -42,6 +45,7 @@ pub struct AppState {
pub enrollment_limiter: AuthRateLimiter,
pub component_socket_slots: Arc<Semaphore>,
pub http: reqwest::Client,
pub song_requests: SongRequestService,
}
impl AppState {
@@ -60,9 +64,26 @@ impl AppState {
let registry = ComponentRegistry::with_builtin_components();
let component_cache = Arc::new(InMemoryComponentStore::default());
let hub = EventHub::new(512);
let song_requests = SongRequestService::new(db.clone(), hub.clone());
registry
.register_handler(
SONG_REQUEST_KIND,
Arc::new(SongRequestHandler::new(song_requests.clone())),
)
.map_err(|error| error.to_string())?;
registry
.register_snapshot_provider(
SONG_REQUEST_KIND,
Arc::new(SongRequestSnapshotProvider::new(song_requests.clone())),
)
.map_err(|error| error.to_string())?;
let router = SourceEventRouter::new(registry.clone(), component_cache.clone(), hub.clone());
let repository =
TenantRepository::new(db.clone(), registry.clone(), component_cache.clone());
repository
.ensure_song_request_components()
.await
.map_err(|error| error.to_string())?;
repository
.hydrate_all()
.await
@@ -113,6 +134,7 @@ impl AppState {
),
component_socket_slots: Arc::new(Semaphore::new(128)),
http,
song_requests,
};
state.start_all_sources().await?;
Ok(state)
@@ -256,6 +278,7 @@ async fn migrate(db: &Db) -> Result<(), String> {
),
(3_i32, include_str!("../migrations/003_multitenancy.sql")),
(4_i32, include_str!("../migrations/004_auth_hardening.sql")),
(5_i32, include_str!("../migrations/005_song_request.sql")),
] {
let applied = transaction
.query_one(
apps/server-rust/src/auth.rs
+25
View File
@@ -27,6 +27,7 @@ use crate::{
credentials::{CookieCloudCredentials, CookieCloudSecrets, normalize_cookiecloud_host},
db::{Db, DbError},
overlay::OverlaySettings,
song_request::{SONG_REQUEST_KIND, SONG_REQUEST_NAME, SongRequestSettings},
};
const TOTP_DIGITS: usize = 6;
@@ -620,6 +621,30 @@ impl AuthService {
],
)
.await?;
let song_component_id = Uuid::new_v4();
let song_settings = serde_json::to_value(SongRequestSettings::default())
.expect("SongRequestSettings is always JSON serializable");
transaction
.execute(
"INSERT INTO component_instances \
(id,owner_user_id,source_id,kind,name,settings,settings_version,enabled) \
VALUES($1,$2,$3,$4,$5,$6,1,true)",
&[
&song_component_id,
&user_id,
&default_source_id,
&SONG_REQUEST_KIND,
&SONG_REQUEST_NAME,
&song_settings,
],
)
.await?;
transaction
.execute(
"INSERT INTO song_request_state(owner_user_id,component_instance_id) VALUES($1,$2)",
&[&user_id, &song_component_id],
)
.await?;
transaction
.execute(
"DELETE FROM pending_registrations WHERE id=$1",
apps/server-rust/src/components.rs
+76
View File
@@ -21,6 +21,7 @@ use uuid::Uuid;
use crate::{
domain::{ComponentMessage, LiveEvent, LiveEventKind},
overlay::OverlaySettings,
song_request::{SongRequestDefinition, SongRequestProjection},
};
pub const DANMAKU_OVERLAY_KIND: &str = "danmaku_overlay";
@@ -180,6 +181,19 @@ pub trait EventProjection: Send + Sync {
/// `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.
@@ -280,6 +294,7 @@ pub struct ComponentRuntime {
definition: Arc<dyn ComponentDefinition>,
projection: Arc<dyn EventProjection>,
handlers: Vec<Arc<dyn EventHandler>>,
snapshot_provider: Option<Arc<dyn ComponentSnapshotProvider>>,
}
impl ComponentRuntime {
@@ -320,6 +335,17 @@ impl ComponentRuntime {
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)]
@@ -347,6 +373,12 @@ impl ComponentRegistry {
)
.expect("built-in component kinds are unique");
registry
.register(
Arc::new(SongRequestDefinition),
Arc::new(SongRequestProjection),
)
.expect("built-in component kinds are unique");
registry
}
pub fn register(
@@ -371,6 +403,7 @@ impl ComponentRegistry {
definition,
projection,
handlers: Vec::new(),
snapshot_provider: None,
},
);
Ok(())
@@ -392,6 +425,22 @@ impl ComponentRegistry {
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()
@@ -469,6 +518,33 @@ mod tests {
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 duplicate_component_kinds_are_rejected() {
let registry = ComponentRegistry::default();
apps/server-rust/src/config.rs
+6 -1
View File
@@ -11,7 +11,10 @@ use base64::{Engine, engine::general_purpose::STANDARD};
use serde::Deserialize;
use sha2::{Digest, Sha256};
use crate::{credentials::normalize_cookiecloud_host, overlay::OverlaySettings};
use crate::{
credentials::normalize_cookiecloud_host,
overlay::{OverlaySettings, OverlayThemeId},
};
/// Process-level configuration. Tenant-owned room, CookieCloud and component
/// settings are imported from the legacy sections once and then live in
@@ -123,6 +126,7 @@ struct EmoticonsConfig {
#[derive(Deserialize, Default)]
struct OverlayFileConfig {
theme_id: Option<OverlayThemeId>,
font_scale: Option<u16>,
max_visible: Option<u8>,
collapse_after_seconds: Option<u16>,
@@ -318,6 +322,7 @@ fn derive_key(secret: &str, domain: &[u8]) -> [u8; 32] {
fn overlay_defaults(file: OverlayFileConfig) -> OverlaySettings {
let default = OverlaySettings::default();
OverlaySettings {
theme_id: file.theme_id.unwrap_or(default.theme_id),
font_scale: file.font_scale.unwrap_or(default.font_scale),
show_danmaku: file.events.danmaku.unwrap_or(default.show_danmaku),
show_enter: file.events.enter.unwrap_or(default.show_enter),
apps/server-rust/src/domain.rs
+21
View File
@@ -277,6 +277,27 @@ pub struct ComponentMessage {
}
impl ComponentMessage {
/// Build a component-owned message that does not originate directly from a
/// provider packet, such as a durable queue snapshot or settings update.
pub fn new(
component: &crate::components::ComponentInstance,
room_id: impl Into<String>,
event_type: impl Into<String>,
payload: Value,
) -> Self {
Self {
owner_id: component.owner_id,
component_id: component.id,
source_id: component.source_id,
version: COMPONENT_PROTOCOL_VERSION,
id: Uuid::new_v4(),
occurred_at: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
room_id: room_id.into(),
event_type: event_type.into(),
payload,
}
}
pub fn from_live_event(
component_id: Uuid,
event: &LiveEvent,
apps/server-rust/src/http_api.rs
+196 -15
View File
@@ -11,7 +11,7 @@ use std::{sync::Arc, time::Duration};
use axum::{
Json, Router,
extract::{
DefaultBodyLimit, Path, Request, State,
DefaultBodyLimit, Path, Query, Request, State,
ws::{CloseFrame, Message, WebSocket, WebSocketUpgrade, close_code},
},
http::{HeaderMap, HeaderValue, StatusCode, header},
@@ -37,6 +37,7 @@ use crate::{
GiftDetails, GiftEvent, LiveEvent, LiveEventPayload, PlatformViewer,
},
repository::{ComponentView, RepositoryError},
song_request::{SONG_REQUEST_KIND, SongListScope, SongRequestError},
};
const SECURE_SESSION_COOKIE: &str = "__Host-lxc_session";
@@ -80,6 +81,22 @@ pub fn router(state: AppState) -> Router {
"/api/v1/components/{public_id}/stream",
get(component_stream),
)
.route(
"/api/v1/components/{id}/song-requests",
get(list_song_requests),
)
.route(
"/api/v1/components/{id}/song-requests/{request_id}/promote",
post(promote_song_request),
)
.route(
"/api/v1/components/{id}/song-requests/{request_id}/complete",
post(complete_song_request),
)
.route(
"/api/v1/components/{id}/song-requests/{request_id}/cancel",
post(cancel_song_request),
)
.route("/api/{*path}", any(api_not_found))
.route("/", get(spa_page))
.route("/login", get(spa_page))
@@ -645,12 +662,127 @@ async fn put_component_settings(
let message = settings_message(
&component,
&session.user.room_id,
"overlay.settings.updated",
"component.settings.updated",
);
state.hub.publish(component.id, Arc::new(message));
if component.kind == "danmaku_overlay" {
state.hub.publish(
component.id,
Arc::new(settings_message(
&component,
&session.user.room_id,
"overlay.settings.updated",
)),
);
}
Ok(Json(json!({"settings":component.settings})))
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct SongRequestsQuery {
scope: Option<String>,
cursor: Option<i64>,
limit: Option<i64>,
}
async fn list_song_requests(
State(state): State<AppState>,
headers: HeaderMap,
Path(id): Path<Uuid>,
Query(query): Query<SongRequestsQuery>,
) -> Result<Json<Value>, ApiError> {
let session = require_session(&state, &headers).await?;
let component = state.repository.get_component(session.user.id, id).await?;
require_song_component(&component)?;
let scope = match query.scope.as_deref().unwrap_or("active") {
"active" => SongListScope::Active,
"history" => SongListScope::History,
_ => {
return Err(ApiError::new(
StatusCode::BAD_REQUEST,
"invalid_scope",
"scope must be active or history",
));
}
};
let page = state
.song_requests
.list_page(
&component,
scope,
query.cursor.unwrap_or(0),
query.limit.unwrap_or(50),
)
.await?;
Ok(Json(serde_json::to_value(page).map_err(internal)?))
}
async fn promote_song_request(
State(state): State<AppState>,
headers: HeaderMap,
Path((id, request_id)): Path<(Uuid, Uuid)>,
) -> Result<Json<Value>, ApiError> {
same_origin(&state, &headers)?;
let session = require_session(&state, &headers).await?;
let component = state.repository.get_component(session.user.id, id).await?;
require_song_component(&component)?;
state
.song_requests
.promote(&component, request_id, &session.user.room_id)
.await?;
Ok(Json(json!({"ok":true})))
}
async fn complete_song_request(
State(state): State<AppState>,
headers: HeaderMap,
Path((id, request_id)): Path<(Uuid, Uuid)>,
) -> Result<Json<Value>, ApiError> {
change_song_request(state, headers, id, request_id, false).await
}
async fn cancel_song_request(
State(state): State<AppState>,
headers: HeaderMap,
Path((id, request_id)): Path<(Uuid, Uuid)>,
) -> Result<Json<Value>, ApiError> {
change_song_request(state, headers, id, request_id, true).await
}
async fn change_song_request(
state: AppState,
headers: HeaderMap,
component_id: Uuid,
request_id: Uuid,
cancelled: bool,
) -> Result<Json<Value>, ApiError> {
same_origin(&state, &headers)?;
let session = require_session(&state, &headers).await?;
let component = state
.repository
.get_component(session.user.id, component_id)
.await?;
require_song_component(&component)?;
state
.song_requests
.finish(&component, request_id, cancelled, &session.user.room_id)
.await?;
Ok(Json(json!({"ok":true})))
}
fn require_song_component(component: &ComponentInstance) -> Result<(), ApiError> {
if component.kind == SONG_REQUEST_KIND {
Ok(())
} else {
Err(ApiError::new(
StatusCode::NOT_FOUND,
"song_component_not_found",
"Song request component not found",
))
}
}
async fn get_component_token(
State(state): State<AppState>,
headers: HeaderMap,
@@ -877,7 +1009,7 @@ async fn component_ws(mut socket: WebSocket, state: AppState, component_id: Uuid
let mut receiver = state.hub.subscribe(component_id);
if socket
.send(Message::Text(
json!({"version":COMPONENT_PROTOCOL_VERSION,"type":"authenticated","componentId":component_id})
json!({"version":COMPONENT_PROTOCOL_VERSION,"type":"authenticated","componentId":component_id,"componentKind":component.kind})
.to_string()
.into(),
))
@@ -893,19 +1025,54 @@ async fn component_ws(mut socket: WebSocket, state: AppState, component_id: Uuid
return;
}
};
let snapshot = settings_message(&component, &room_id, "overlay.settings.snapshot");
let snapshot = settings_message(&component, &room_id, "component.settings.snapshot");
if send_component_message(&mut socket, &snapshot)
.await
.is_err()
{
return;
}
if component.kind == "danmaku_overlay"
&& send_component_message(
&mut socket,
&settings_message(&component, &room_id, "overlay.settings.snapshot"),
)
.await
.is_err()
{
return;
}
let runtime = match state.registry.runtime(&component.kind) {
Ok(runtime) => runtime,
Err(_) => {
close_ws(&mut socket, "Component is unavailable").await;
return;
}
};
let component_snapshots = match runtime.snapshot(&component, &room_id).await {
Ok(messages) => messages,
Err(error) => {
warn!(component_id = %component.id, %error, "component snapshot failed");
close_ws(&mut socket, "Component snapshot is unavailable").await;
return;
}
};
for message in component_snapshots {
if send_component_message(&mut socket, &message).await.is_err() {
return;
}
}
loop {
tokio::select! {
event = receiver.recv() => match event {
Ok(message) => {
if send_component_message(&mut socket, &message).await.is_err() { break; }
}
// A durable component cannot safely guess dropped deltas. A
// disconnect makes the browser obtain a fresh paged snapshot;
// passive visual components can keep their legacy best-effort
// behavior.
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) if component.kind == SONG_REQUEST_KIND => break,
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
},
@@ -945,17 +1112,12 @@ fn settings_message(
room_id: &str,
event_type: &str,
) -> ComponentMessage {
ComponentMessage {
owner_id: component.owner_id,
component_id: component.id,
source_id: component.source_id,
version: COMPONENT_PROTOCOL_VERSION,
id: Uuid::new_v4(),
occurred_at: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
room_id: room_id.to_owned(),
event_type: event_type.into(),
payload: json!({"settings":component.settings}),
}
ComponentMessage::new(
component,
room_id,
event_type,
json!({"settings":component.settings}),
)
}
async fn require_session(
@@ -1256,6 +1418,25 @@ impl From<RepositoryError> for ApiError {
}
}
impl From<SongRequestError> for ApiError {
fn from(error: SongRequestError) -> Self {
match error {
SongRequestError::NotFound => Self::new(
StatusCode::NOT_FOUND,
"song_request_not_found",
"Song request was not found",
),
SongRequestError::Conflict(message) => {
Self::new(StatusCode::CONFLICT, "song_request_conflict", message)
}
SongRequestError::Invalid(message) => {
Self::new(StatusCode::BAD_REQUEST, "invalid_song_request", message)
}
other => internal(other),
}
}
}
fn internal(error: impl std::fmt::Display) -> ApiError {
error!(%error, "internal operation failed");
ApiError::new(
apps/server-rust/src/lib.rs
+1
View File
@@ -18,3 +18,4 @@ pub mod overlay;
pub mod rate_limit;
pub mod realtime;
pub mod repository;
pub mod song_request;
apps/server-rust/src/live/bilibili.rs
+107 -20
View File
@@ -5,7 +5,11 @@
//! domain payloads. Unknown commands expose only sanitized metadata—never the
//! original unbounded packet or authentication material.
use std::{sync::Arc, time::Duration};
use std::{
sync::Arc,
thread,
time::{Duration, Instant},
};
use async_trait::async_trait;
use blivedm::client::{models::BiliMessage, websocket::BiliLiveClient};
@@ -26,6 +30,9 @@ use crate::{
overlay::{EmoticonCatalog, EmoticonMeta, GiftCatalog, normalize_image_url},
};
const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(20);
const RECONNECT_BACKOFF: Duration = Duration::from_secs(3);
#[derive(Clone)]
pub struct BilibiliProvider {
cookie: Arc<str>,
@@ -226,37 +233,104 @@ impl LiveProvider for BilibiliProvider {
let listener_cancel = cancel.clone();
let listener_status = status.clone();
let listener = tokio::task::spawn_blocking(move || -> Result<(), String> {
let (upstream_sender, mut upstream_receiver) = futures_mpsc::channel(256);
let mut client = BiliLiveClient::new_auto(Some(&cookie), &room_id, upstream_sender)?;
client.set_read_timeout(Some(Duration::from_secs(1)))?;
client.send_auth();
let _ = listener_status.send(SourceStatus {
source_id: context.source_id,
room_id: room_id.clone(),
connected: true,
cookie_cloud: true,
detail: "Connected with authenticated blivedm_rs listener".into(),
});
// A full-client retry loop complements blivedm_rs' short socket
// reconnect loop. If all in-place attempts fail, recreating the
// client refreshes the danmaku host and authentication token.
while !listener_cancel.is_cancelled() {
if let Err(error) = client.receive() {
let (upstream_sender, mut upstream_receiver) = futures_mpsc::channel(256);
let mut client =
match BiliLiveClient::new_auto(Some(&cookie), &room_id, upstream_sender) {
Ok(client) => client,
Err(error) => {
warn!(%error, %room_id, "Bilibili listener creation failed; retrying");
let _ = listener_status.send(SourceStatus {
source_id: context.source_id,
room_id: room_id.clone(),
connected: false,
cookie_cloud: true,
detail: format!("Bilibili connection failed; retrying: {error}"),
});
wait_for_retry(&listener_cancel, RECONNECT_BACKOFF);
continue;
}
};
if let Err(error) = client.set_read_timeout(Some(Duration::from_secs(1))) {
client.close();
return Err(format!("cannot configure Bilibili socket timeout: {error}"));
}
// Bilibili closes otherwise healthy danmaku sockets after
// roughly one minute without a heartbeat. The upstream CLI
// sends one every 20 seconds; embedded users must do the same.
if !client.send_auth() || !client.send_heart_beat() {
client.close();
let _ = listener_status.send(SourceStatus {
source_id: context.source_id,
room_id: room_id.clone(),
connected: false,
cookie_cloud: true,
detail: format!("Bilibili connection error: {error}"),
detail: "Bilibili authentication failed; retrying".into(),
});
wait_for_retry(&listener_cancel, RECONNECT_BACKOFF);
continue;
}
while let Ok(message) = upstream_receiver.try_recv() {
if let Some(message) = normalize(message)
&& raw_sender.blocking_send(message).is_err()
{
let _ = listener_status.send(SourceStatus {
source_id: context.source_id,
room_id: room_id.clone(),
connected: true,
cookie_cloud: true,
detail: "Bilibili socket connected; waiting for live events".into(),
});
info!(%room_id, "Bilibili live socket connected and heartbeat started");
let mut last_heartbeat = Instant::now();
let mut received_event = false;
let disconnect_error = loop {
if listener_cancel.is_cancelled() {
client.close();
return Ok(());
}
}
if last_heartbeat.elapsed() >= HEARTBEAT_INTERVAL {
if !client.send_heart_beat() {
break "heartbeat recovery failed".to_owned();
}
last_heartbeat = Instant::now();
}
if let Err(error) = client.receive() {
break error;
}
while let Ok(message) = upstream_receiver.try_recv() {
let Some(message) = normalize(message) else {
continue;
};
if !received_event {
received_event = true;
info!(%room_id, "Bilibili listener received its first live event");
let _ = listener_status.send(SourceStatus {
source_id: context.source_id,
room_id: room_id.clone(),
connected: true,
cookie_cloud: true,
detail: "Connected and receiving live events".into(),
});
}
if raw_sender.blocking_send(message).is_err() {
client.close();
return Ok(());
}
}
};
client.close();
warn!(error = %disconnect_error, %room_id, "Bilibili listener disconnected; rebuilding client");
let _ = listener_status.send(SourceStatus {
source_id: context.source_id,
room_id: room_id.clone(),
connected: false,
cookie_cloud: true,
detail: format!("Bilibili disconnected; retrying: {disconnect_error}"),
});
wait_for_retry(&listener_cancel, RECONNECT_BACKOFF);
}
client.close();
Ok(())
});
@@ -285,6 +359,19 @@ impl LiveProvider for BilibiliProvider {
}
}
/// Blocking listener tasks cannot await a cancellation token. Sleeping in
/// short slices keeps shutdown/reconfiguration latency bounded.
fn wait_for_retry(cancel: &CancellationToken, duration: Duration) {
let deadline = Instant::now() + duration;
while !cancel.is_cancelled() {
let remaining = deadline.saturating_duration_since(Instant::now());
if remaining.is_zero() {
break;
}
thread::sleep(remaining.min(Duration::from_millis(200)));
}
}
#[derive(Clone, Debug)]
struct EmoticonHint {
text: String,
apps/server-rust/src/overlay.rs
+27
View File
@@ -11,9 +11,25 @@ use serde::{Deserialize, Serialize};
use serde_json::Value;
use tokio::sync::RwLock;
/// Stable identifier for a renderer theme.
///
/// This enum is deliberately shared by defaults, validation and serialization:
/// persisted component settings can therefore never reference a theme that the
/// deployed frontend does not know how to render. Add a variant only together
/// with its frontend theme definition and stylesheet.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum OverlayThemeId {
#[default]
JadeScroll,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct OverlaySettings {
/// Visual theme selected for this tenant-owned component instance.
#[serde(default)]
pub theme_id: OverlayThemeId,
#[serde(default = "default_font_scale")]
pub font_scale: u16,
pub show_danmaku: bool,
@@ -41,6 +57,7 @@ pub struct OverlaySettings {
impl Default for OverlaySettings {
fn default() -> Self {
Self {
theme_id: OverlayThemeId::default(),
font_scale: default_font_scale(),
show_danmaku: true,
show_enter: true,
@@ -492,10 +509,20 @@ mod tests {
object.remove("unfoldDurationMs");
object.remove("particleCount");
object.remove("particleSpeed");
object.remove("themeId");
let settings: OverlaySettings = serde_json::from_value(value).expect("legacy settings");
assert_eq!(settings.theme_id, OverlayThemeId::JadeScroll);
assert_eq!(settings.font_scale, 140);
assert_eq!(settings.unfold_duration_ms, 1_000);
assert_eq!(settings.particle_count, 8);
assert_eq!(settings.particle_speed, 100);
}
#[test]
fn unknown_theme_ids_are_rejected() {
let mut value = json!(OverlaySettings::default());
value["themeId"] = json!("theme-that-is-not-installed");
let error = serde_json::from_value::<OverlaySettings>(value).expect_err("unknown theme");
assert!(error.to_string().contains("unknown variant"));
}
}
apps/server-rust/src/repository.rs
+73
View File
@@ -16,6 +16,7 @@ use crate::{
components::{ComponentInstance, ComponentRegistry},
db::{ComponentRecord, Db, DbError},
realtime::InMemoryComponentStore,
song_request::{SONG_REQUEST_KIND, SONG_REQUEST_NAME, SongRequestSettings},
};
#[derive(Clone)]
@@ -45,6 +46,61 @@ impl TenantRepository {
Ok(())
}
/// Ensure every active tenant has the built-in singleton song component.
/// The partial unique index makes this safe across concurrent application
/// starts; the state row is repaired independently for existing instances.
pub async fn ensure_song_request_components(&self) -> Result<(), RepositoryError> {
for tenant in self.db.list_active_tenants().await? {
let mut client = self.db.get().await?;
let transaction = client.transaction().await?;
Db::set_tenant(&transaction, tenant.user_id).await?;
let existing = transaction
.query_opt(
"SELECT id FROM component_instances WHERE owner_user_id=$1 AND kind=$2",
&[&tenant.user_id, &SONG_REQUEST_KIND],
)
.await?;
let component_id: Uuid = if let Some(row) = existing {
row.get(0)
} else {
let id = Uuid::new_v4();
let settings = serde_json::to_value(SongRequestSettings::default())
.map_err(|error| RepositoryError::Invalid(error.to_string()))?;
transaction
.execute(
"INSERT INTO component_instances \
(id,owner_user_id,source_id,kind,name,settings,settings_version,enabled) \
VALUES($1,$2,$3,$4,$5,$6,1,true) ON CONFLICT DO NOTHING",
&[
&id,
&tenant.user_id,
&tenant.source_id,
&SONG_REQUEST_KIND,
&SONG_REQUEST_NAME,
&settings,
],
)
.await?;
transaction
.query_one(
"SELECT id FROM component_instances WHERE owner_user_id=$1 AND kind=$2",
&[&tenant.user_id, &SONG_REQUEST_KIND],
)
.await?
.get(0)
};
transaction
.execute(
"INSERT INTO song_request_state(owner_user_id,component_instance_id) \
VALUES($1,$2) ON CONFLICT(component_instance_id) DO NOTHING",
&[&tenant.user_id, &component_id],
)
.await?;
transaction.commit().await?;
}
Ok(())
}
pub async fn hydrate_tenant(
&self,
owner_id: Uuid,
@@ -72,6 +128,12 @@ impl TenantRepository {
kind: &str,
name: &str,
) -> Result<ComponentInstance, RepositoryError> {
// Every tenant receives this singleton during registration/startup.
// Keeping creation internal prevents a second instance from racing the
// partial unique index and turning a domain conflict into a DB error.
if kind == SONG_REQUEST_KIND {
return Err(RepositoryError::Forbidden);
}
let runtime = self
.registry
.runtime(kind)
@@ -123,6 +185,17 @@ impl TenantRepository {
let mut client = self.db.get().await?;
let transaction = client.transaction().await?;
Db::set_tenant(&transaction, owner_id).await?;
let kind: String = transaction
.query_opt(
"SELECT kind FROM component_instances WHERE owner_user_id=$1 AND id=$2",
&[&owner_id, &component_id],
)
.await?
.ok_or(RepositoryError::NotFound)?
.get(0);
if kind == SONG_REQUEST_KIND {
return Err(RepositoryError::Forbidden);
}
let changed = transaction
.execute(
"DELETE FROM component_instances WHERE owner_user_id=$1 AND id=$2",
apps/server-rust/src/song_request.rs
+1038
View File
File diff suppressed because it is too large Load Diff