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

formatting and comments

Browse Source
This commit is contained in:
felis
2026-07-16 00:12:26 -07:00
parent edb6d2b5b4
commit 994854d104
45 changed files with 2514 additions and 628 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/server-rust/README.md
+50
View File
@@ -0,0 +1,50 @@
# Rust 后端
这是多租户直播组件服务的可复用核心和唯一运行时进程。`src/main.rs` 只处理进程启动；主要行为由 library
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 和热路径缓存同步 |
## 重要不变量
- handler 不能信任请求体中的 owner；owner 必须来自 session 或 source context。
- tenant table 查询必须在 `Db::set_tenant` 后的事务中执行。
- provider 只能输出 canonical、bounded、sanitized `LiveEvent`。
- projection 无副作用；可靠业务动作必须使用幂等 handler。
- token、邀请码和恢复码只存摘要，TOTP/CookieCloud Secret 只存认证加密密文。
- EventHub 的 channel key 是 component ID，不允许增加无权限的全局 receiver。
## 本地质量检查
在仓库根目录执行：
```bash
cargo fmt --manifest-path apps/server-rust/Cargo.toml --check
cargo clippy --manifest-path apps/server-rust/Cargo.toml --all-targets --no-deps -- -D warnings
cargo test --manifest-path apps/server-rust/Cargo.toml --all-targets
```
第三方 `vendor/blivedm` 不作为本项目风格重写目标；项目只维护保留原始 JSON 所需的小补丁。
更多设计说明：
- [总体架构](../../docs/architecture.md)
- [组件开发](../../docs/components/README.md)
- [实时协议](../../docs/protocol.md)
- [安全模型](../../docs/security.md)
apps/server-rust/rustfmt.toml
+4
View File
@@ -0,0 +1,4 @@
edition = "2024"
max_width = 100
newline_style = "Unix"
use_small_heuristics = "Default"
apps/server-rust/src/app.rs
+7
View File
@@ -1,3 +1,10 @@
//! Application composition root and process-wide dependencies.
//!
//! [`AppState::build`] wires the database, authentication service, component
//! registry, live-source supervisor and realtime router in dependency order.
//! Tenant-owned runtime state stays in PostgreSQL or the source supervisor;
//! this module only owns cloneable handles shared by Axum handlers.
use std::{sync::Arc, time::Duration};
use async_trait::async_trait;
apps/server-rust/src/auth.rs
+8
View File
@@ -1,3 +1,11 @@
//! Passwordless authentication, enrollment and secret-storage domain service.
//!
//! This module owns invite redemption, TOTP replay protection, recovery codes,
//! session issuance and tenant credential encryption. Raw sessions, recovery
//! codes and component tokens are returned only at creation time; persistent
//! rows contain hashes or authenticated ciphertext. HTTP-specific cookie and
//! origin policy deliberately live in `http_api`, not here.
use std::{fmt, sync::Arc, time::Duration};
use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
apps/server-rust/src/components.rs
+7
View File
@@ -1,3 +1,10 @@
//! 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,
apps/server-rust/src/config.rs
+7
View File
@@ -1,3 +1,10 @@
//! TOML configuration loading, validation and legacy bootstrap compatibility.
//!
//! Deployment-wide policy (bind address, encryption key, allowed CookieCloud
//! hosts and timeouts) remains in [`Config`]. Room IDs, credentials and component
//! settings become tenant-owned database records after enrollment; the legacy
//! TOML fields are import inputs and must not be treated as global live state.
use std::{env, fs, net::IpAddr, path::PathBuf};
use base64::{Engine, engine::general_purpose::STANDARD};
apps/server-rust/src/credentials.rs
+7
View File
@@ -1,3 +1,10 @@
//! CookieCloud boundary and Bilibili cookie extraction.
//!
//! Tenant input reaches an outbound HTTP client only after canonical URL
//! validation and exact allow-list matching in the caller. Redirects are
//! disabled by the shared client, the synchronization key is encoded as one
//! path segment, and only the minimum Bilibili cookie fields leave this module.
use reqwest::{
Url,
header::{REFERER, USER_AGENT},
apps/server-rust/src/db.rs
+7
View File
@@ -1,3 +1,10 @@
//! PostgreSQL pool, migrations and low-level tenant-scoped queries.
//!
//! Multi-tenant tables use PostgreSQL row-level security in addition to owner
//! columns and composite foreign keys. Every tenant query must execute inside a
//! transaction after [`Db::set_tenant`], which uses `SET LOCAL` so pooled
//! connections cannot retain the previous request's identity.
use std::{fmt, str::FromStr};
use deadpool_postgres::{Manager, ManagerConfig, Object, Pool, RecyclingMethod, Runtime};
apps/server-rust/src/domain.rs
+7
View File
@@ -1,3 +1,10 @@
//! Provider-independent live events and the versioned component wire envelope.
//!
//! Provider adapters normalize platform packets into [`LiveEvent`]. Components
//! subscribe to stable [`LiveEventKind`] values and project them to
//! [`ComponentMessage`], keeping Bilibili command names and raw payloads out of
//! browser contracts and future provider implementations.
use serde::{Deserialize, Serialize};
use serde_json::Value;
use uuid::Uuid;
apps/server-rust/src/http_api.rs
+14 -6
View File
@@ -1,3 +1,11 @@
//! Axum HTTP/WebSocket adapter and public trust boundary.
//!
//! Handlers derive the tenant from a server-side session cookie; owner IDs from
//! request bodies are never trusted. This module also enforces same-origin
//! mutation checks, response cache policy, component-token WebSocket
//! authentication, bounded request bodies and security headers for the static
//! control console and OBS entry point.
use std::{sync::Arc, time::Duration};
use axum::{
@@ -326,7 +334,7 @@ async fn login(
Ok(result) => result,
Err(error) => {
state.login_limiter.failure(&body.username, &ip).await;
return Err(ApiError::from(error).as_generic_login());
return Err(ApiError::from(error).into_generic_login());
}
};
state.login_limiter.success(&body.username, &ip).await;
@@ -1146,7 +1154,7 @@ impl ApiError {
self
}
fn as_generic_login(mut self) -> Self {
fn into_generic_login(mut self) -> Self {
if self.status != StatusCode::TOO_MANY_REQUESTS {
self.status = StatusCode::UNAUTHORIZED;
self.code = "invalid_credentials";
@@ -1170,10 +1178,10 @@ impl IntoResponse for ApiError {
}),
)
.into_response();
if let Some(retry_after) = self.retry_after {
if let Ok(value) = HeaderValue::from_str(&retry_after.as_secs().max(1).to_string()) {
response.headers_mut().insert(header::RETRY_AFTER, value);
}
if let Some(retry_after) = self.retry_after
&& let Ok(value) = HeaderValue::from_str(&retry_after.as_secs().max(1).to_string())
{
response.headers_mut().insert(header::RETRY_AFTER, value);
}
response
}
apps/server-rust/src/live/bilibili.rs
+30 -25
View File
@@ -1,3 +1,10 @@
//! Bilibili `blivedm_rs` adapter and raw-command normalization.
//!
//! The adapter authenticates with a CookieCloud-derived cookie, refreshes gift
//! and emoticon catalogs, and converts supported Bilibili commands into bounded
//! domain payloads. Unknown commands expose only sanitized metadata—never the
//! original unbounded packet or authentication material.
use std::{sync::Arc, time::Duration};
use async_trait::async_trait;
@@ -241,11 +248,11 @@ impl LiveProvider for BilibiliProvider {
});
}
while let Ok(message) = upstream_receiver.try_recv() {
if let Some(message) = normalize(message) {
if raw_sender.blocking_send(message).is_err() {
client.close();
return Ok(());
}
if let Some(message) = normalize(message)
&& raw_sender.blocking_send(message).is_err()
{
client.close();
return Ok(());
}
}
}
@@ -534,12 +541,12 @@ fn parse_danmaku_emoticons(info: &[Value], text: &str) -> Vec<EmoticonHint> {
json_object(object.get("extra")?)
});
if let Some(extra) = extra {
if let Some(emoticons) = extra.get("emots").and_then(json_object) {
if let Some(emoticons) = emoticons.as_object() {
for (token, metadata) in emoticons {
if let Some(hint) = emoticon_hint(metadata, token, false) {
hints.push(hint);
}
if let Some(emoticons) = extra.get("emots").and_then(json_object)
&& let Some(emoticons) = emoticons.as_object()
{
for (token, metadata) in emoticons {
if let Some(hint) = emoticon_hint(metadata, token, false) {
hints.push(hint);
}
}
}
@@ -547,22 +554,20 @@ fn parse_danmaku_emoticons(info: &[Value], text: &str) -> Vec<EmoticonHint> {
.get("emoticon_unique")
.and_then(Value::as_str)
.filter(|value| !value.is_empty())
{
if !hints
&& !hints
.iter()
.any(|hint| hint.unique.as_deref() == Some(unique))
{
hints.push(EmoticonHint {
text: text.to_owned(),
unique: Some(unique.to_owned()),
url: None,
width: None,
height: None,
is_dynamic: false,
bulge_display: value_is_truthy(extra.get("bulge_display")),
standalone: extra.get("dm_type").and_then(Value::as_i64) == Some(1),
});
}
{
hints.push(EmoticonHint {
text: text.to_owned(),
unique: Some(unique.to_owned()),
url: None,
width: None,
height: None,
is_dynamic: false,
bulge_display: value_is_truthy(extra.get("bulge_display")),
standalone: extra.get("dm_type").and_then(Value::as_i64) == Some(1),
});
}
}
hints
apps/server-rust/src/live/mod.rs
+7
View File
@@ -1,3 +1,10 @@
//! Live-provider abstraction and source lifecycle types.
//!
//! A provider receives trusted tenant/source context and emits canonical
//! [`LiveEvent`] values plus observable connection status. Cancellation and
//! restart ownership live in [`supervisor::SourceSupervisor`], keeping provider
//! implementations focused on one upstream connection.
pub mod bilibili;
pub mod supervisor;
apps/server-rust/src/live/supervisor.rs
+6
View File
@@ -1,3 +1,9 @@
//! Per-source task ownership, restart and cancellation.
//!
//! The supervisor guarantees at most one provider generation for a source ID.
//! Reconfiguration cancels the old task before a replacement starts, preventing
//! duplicate Bilibili listeners and duplicate downstream events.
use std::{collections::HashMap, sync::Arc};
use async_trait::async_trait;
apps/server-rust/src/main.rs
+6
View File
@@ -1,3 +1,9 @@
//! Executable entry point for the livestream component host.
//!
//! Keep this file limited to process concerns: configuration, logging, socket
//! binding and graceful shutdown. All application behavior belongs in the
//! library crate so it can be exercised without starting a real HTTP server.
use lxc_stream_server::{app::AppState, config::Config, http_api};
use tracing::{info, warn};
apps/server-rust/src/overlay.rs
+35 -21
View File
@@ -1,3 +1,10 @@
//! Danmaku-overlay settings plus Bilibili gift and emoticon metadata caches.
//!
//! Settings are sanitized before persistence or projection. Catalog refreshes
//! replace the in-memory snapshot only after a complete valid response, so a
//! transient upstream failure preserves the last known gift images, prices and
//! emoticon URLs instead of breaking live rendering.
use std::{collections::HashMap, sync::Arc};
use serde::{Deserialize, Serialize};
@@ -123,16 +130,30 @@ pub struct EmoticonCatalog {
by_emoji: Arc<RwLock<HashMap<String, EmoticonMeta>>>,
}
// Parser return aliases keep the atomic cache-replacement contract visible:
// both lookup maps and their source count are produced before either lock is
// updated, so readers never observe a half-refreshed catalog.
type GiftCatalogSnapshot = (HashMap<i64, GiftMeta>, HashMap<String, GiftMeta>, usize);
type EmoticonCatalogSnapshot = (
HashMap<String, EmoticonMeta>,
HashMap<String, EmoticonMeta>,
usize,
);
impl EmoticonCatalog {
pub async fn len(&self) -> usize {
self.by_unique.read().await.len()
}
pub async fn is_empty(&self) -> bool {
self.len().await == 0
}
pub async fn get(&self, unique: Option<&str>, emoji: &str) -> Option<EmoticonMeta> {
if let Some(unique) = unique.filter(|value| !value.is_empty()) {
if let Some(emoticon) = self.by_unique.read().await.get(unique).cloned() {
return Some(emoticon);
}
if let Some(unique) = unique.filter(|value| !value.is_empty())
&& let Some(emoticon) = self.by_unique.read().await.get(unique).cloned()
{
return Some(emoticon);
}
self.by_emoji.read().await.get(emoji).cloned()
}
@@ -184,11 +205,15 @@ impl GiftCatalog {
self.by_id.read().await.len()
}
pub async fn is_empty(&self) -> bool {
self.len().await == 0
}
pub async fn get(&self, id: Option<i64>, name: &str) -> Option<GiftMeta> {
if let Some(id) = id {
if let Some(gift) = self.by_id.read().await.get(&id).cloned() {
return Some(gift);
}
if let Some(id) = id
&& let Some(gift) = self.by_id.read().await.get(&id).cloned()
{
return Some(gift);
}
self.by_name
.read()
@@ -231,9 +256,7 @@ impl GiftCatalog {
}
}
fn parse_catalog(
payload: &Value,
) -> Result<(HashMap<i64, GiftMeta>, HashMap<String, GiftMeta>, usize), String> {
fn parse_catalog(payload: &Value) -> Result<GiftCatalogSnapshot, String> {
let list = payload
.pointer("/data/gift_config/base_config/list")
.and_then(Value::as_array)
@@ -282,16 +305,7 @@ fn string_field(value: &Value, name: &str) -> Option<String> {
.map(ToOwned::to_owned)
}
fn parse_emoticon_catalog(
payload: &Value,
) -> Result<
(
HashMap<String, EmoticonMeta>,
HashMap<String, EmoticonMeta>,
usize,
),
String,
> {
fn parse_emoticon_catalog(payload: &Value) -> Result<EmoticonCatalogSnapshot, String> {
let packages = payload
.pointer("/data/data")
.and_then(Value::as_array)
apps/server-rust/src/rate_limit.rs
+6
View File
@@ -1,3 +1,9 @@
//! In-memory abuse limits for anonymous authentication and enrollment routes.
//!
//! Limits are evaluated across both account and network-derived keys while
//! returning one generic result to callers. This reduces brute-force attempts
//! without turning timing or error messages into a username-enumeration API.
use std::{
collections::{HashMap, VecDeque},
sync::Arc,
apps/server-rust/src/realtime.rs
+7
View File
@@ -1,3 +1,10 @@
//! Tenant-aware event routing and component-scoped realtime fanout.
//!
//! The router resolves enabled instances by owner and source, validates their
//! settings/subscriptions, runs durable handlers, then publishes passive
//! projections. Each component owns a separate broadcast channel; there is no
//! global receiver that could accidentally observe another tenant's events.
use std::{
collections::HashMap,
error::Error,
apps/server-rust/src/repository.rs
+7
View File
@@ -1,3 +1,10 @@
//! Repository facade for tenant components, settings and the routing cache.
//!
//! PostgreSQL is authoritative. Successful writes are reflected into the
//! in-process [`InMemoryComponentStore`] used by the hot event path; startup and
//! source restarts hydrate that cache from tenant-scoped rows before events are
//! routed.
use std::{fmt, sync::Arc};
use chrono::{DateTime, Utc};