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

add account localization and switchable rooms

Browse Source 
Centralize control and OBS copy in a shared TOML catalog, persist the selected locale per account, and broadcast language changes to component streams. Allow account owners to atomically switch their Bilibili room and restart the shared listener without changing component URLs.
This commit is contained in:
felis
2026-07-18 23:28:05 -07:00
parent 53ae90ec13
commit f79852d8e6
35 changed files with 1946 additions and 472 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/server-rust/README.md
+1
View File
@@ -15,6 +15,7 @@ crate 导出。
| `db` | PostgreSQL pool、迁移和 RLS tenant context |
| `domain` | provider-independent event 与 WebSocket envelope |
| `http_api` | REST/WS、会话、权限、same-origin、静态资源与安全响应头 |
| `i18n` | 嵌入共享 TOML 语言目录并验证 locale 与键完整性 |
| `live` | provider trait、Bilibili adapter 与 source supervisor |
| `overlay` | 弹幕姬设置及礼物/表情目录 |
| `rate_limit` | 匿名登录和 enrollment 滥用限制 |
apps/server-rust/migrations/007_switchable_account_room.sql
+31
View File
@@ -0,0 +1,31 @@
-- Treat an invitation room as the account's initial room rather than an
-- immutable lifetime binding. Room IDs remain unique across accounts so two
-- listeners cannot accidentally claim the same upstream room.
DROP TRIGGER IF EXISTS users_room_id_immutable ON users;
DROP FUNCTION IF EXISTS prevent_user_room_id_change();
CREATE OR REPLACE FUNCTION enforce_live_source_account_room()
RETURNS trigger
LANGUAGE plpgsql
SET search_path = pg_catalog, public
AS $$
BEGIN
IF NOT EXISTS (
SELECT 1 FROM public.users
WHERE id = NEW.owner_user_id AND room_id = NEW.room_id
) THEN
RAISE EXCEPTION 'live source room_id must equal its owning account room_id';
END IF;
IF TG_OP = 'UPDATE'
AND NEW.owner_user_id IS DISTINCT FROM OLD.owner_user_id THEN
RAISE EXCEPTION 'live source ownership is immutable';
END IF;
RETURN NEW;
END;
$$;
COMMENT ON COLUMN users.room_id IS
'Current account-level Bilibili room; initialized by invitation and user-switchable.';
COMMENT ON COLUMN live_sources.room_id IS
'Current room of the owning account; updated atomically with users.room_id.';
apps/server-rust/migrations/008_account_language.sql
+23
View File
@@ -0,0 +1,23 @@
-- Account language preference. The browser keeps a local pre-login choice;
-- authenticated saves become authoritative for every control session and OBS
-- component owned by the account.
CREATE TABLE IF NOT EXISTS account_preferences (
user_id UUID PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
-- Supported locale codes are validated against resources/i18n.toml by the
-- repository. Keeping the column open avoids a schema migration per locale.
language TEXT NOT NULL DEFAULT 'zh-CN',
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
INSERT INTO account_preferences(user_id)
SELECT id FROM users
ON CONFLICT(user_id) DO NOTHING;
ALTER TABLE account_preferences ENABLE ROW LEVEL SECURITY;
ALTER TABLE account_preferences FORCE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS account_preferences_owner ON account_preferences;
CREATE POLICY account_preferences_owner ON account_preferences
USING (user_id = NULLIF(current_setting('app.user_id', true), '')::UUID)
WITH CHECK (user_id = NULLIF(current_setting('app.user_id', true), '')::UUID);
apps/server-rust/src/app.rs
+8
View File
@@ -283,6 +283,14 @@ async fn migrate(db: &Db) -> Result<(), String> {
6_i32,
include_str!("../migrations/006_account_live_source.sql"),
),
(
7_i32,
include_str!("../migrations/007_switchable_account_room.sql"),
),
(
8_i32,
include_str!("../migrations/008_account_language.sql"),
),
] {
let applied = transaction
.query_one(
apps/server-rust/src/auth.rs
+134 -26
View File
@@ -19,13 +19,14 @@ use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use subtle::ConstantTimeEq;
use tokio_postgres::Transaction;
use tokio_postgres::{Transaction, error::SqlState};
use totp_rs::{Algorithm, Secret, TOTP};
use uuid::Uuid;
use crate::{
credentials::{CookieCloudCredentials, CookieCloudSecrets, normalize_cookiecloud_host},
db::{Db, DbError},
i18n,
overlay::OverlaySettings,
song_request::{SONG_REQUEST_KIND, SONG_REQUEST_NAME, SongRequestSettings},
};
@@ -479,7 +480,11 @@ impl AuthService {
&self,
enrollment_token: &str,
totp_code: &str,
language: &str,
) -> Result<RegistrationComplete, AuthError> {
if !i18n::is_supported(language) {
return Err(AuthError::InvalidInput("language is not supported"));
}
let digest = token_digest(enrollment_token.trim());
let mut client = self.db.get().await?;
let transaction = client.transaction().await?;
@@ -598,6 +603,12 @@ impl AuthService {
let recovery_codes = replace_recovery_codes(&transaction, user_id).await?;
let session = insert_session(&transaction, user_id, self.session_ttl).await?;
Db::set_tenant(&transaction, user_id).await?;
transaction
.execute(
"INSERT INTO account_preferences(user_id,language) VALUES($1,$2)",
&[&user_id, &language],
)
.await?;
let default_source_id = Uuid::new_v4();
transaction
.execute(
@@ -909,6 +920,100 @@ impl AuthService {
user_id: Uuid,
credentials: &CookieCloudCredentials,
) -> Result<(), AuthError> {
let (host, encrypted) = self.encrypt_cookiecloud_credentials(user_id, credentials)?;
let mut client = self.db.get().await?;
let transaction = client.transaction().await?;
ensure_active_user(&transaction, user_id).await?;
Db::set_tenant(&transaction, user_id).await?;
upsert_cookiecloud_credentials(&transaction, user_id, &host, &encrypted).await?;
insert_audit(
&transaction,
Some(user_id),
"cookiecloud.credentials.updated",
"user",
Some(user_id.to_string()),
json!({"host":host}),
)
.await?;
transaction.commit().await?;
Ok(())
}
/// Atomically changes the account's current room and encrypted CookieCloud
/// source. The live source keeps its stable ID, allowing the supervisor to
/// cancel the old room connection and start the replacement without
/// changing any component or OBS token.
pub async fn set_account_live_source(
&self,
user_id: Uuid,
room_id: &str,
credentials: &CookieCloudCredentials,
) -> Result<(), AuthError> {
let room_id = room_id.trim();
validate_room_id(room_id)?;
let (host, encrypted) = self.encrypt_cookiecloud_credentials(user_id, credentials)?;
let mut client = self.db.get().await?;
let transaction = client.transaction().await?;
ensure_active_user(&transaction, user_id).await?;
let old_room_id: String = transaction
.query_one("SELECT room_id FROM users WHERE id=$1", &[&user_id])
.await?
.get(0);
let room_in_use: bool = transaction
.query_one(
"SELECT EXISTS(SELECT 1 FROM users WHERE room_id=$1 AND id<>$2)",
&[&room_id, &user_id],
)
.await?
.get(0);
if room_in_use {
return Err(AuthError::RoomUnavailable);
}
Db::set_tenant(&transaction, user_id).await?;
transaction
.execute(
"UPDATE users SET room_id=$1,updated_at=now() WHERE id=$2 AND status='active'",
&[&room_id, &user_id],
)
.await
.map_err(map_room_write_error)?;
let changed = transaction
.execute(
"UPDATE live_sources SET room_id=$1,updated_at=now() \
WHERE owner_user_id=$2",
&[&room_id, &user_id],
)
.await
.map_err(map_room_write_error)?;
if changed != 1 {
return Err(AuthError::DatabaseInvariant(
"account must own exactly one live source",
));
}
upsert_cookiecloud_credentials(&transaction, user_id, &host, &encrypted).await?;
insert_audit(
&transaction,
Some(user_id),
"account.live_source.updated",
"user",
Some(user_id.to_string()),
json!({
"host": host,
"oldRoomId": old_room_id,
"newRoomId": room_id,
"roomChanged": old_room_id != room_id,
}),
)
.await?;
transaction.commit().await?;
Ok(())
}
fn encrypt_cookiecloud_credentials(
&self,
user_id: Uuid,
credentials: &CookieCloudCredentials,
) -> Result<(String, EncryptedSecret), AuthError> {
let host = normalize_cookiecloud_host(&credentials.host)
.map_err(|_| AuthError::InvalidInput("CookieCloud host is invalid"))?;
if credentials.secrets.key.trim().is_empty() || credentials.secrets.password.is_empty() {
@@ -924,31 +1029,7 @@ impl AuthService {
let encrypted = self
.cipher
.encrypt(&plaintext, format!("cookiecloud:{user_id}").as_bytes())?;
let mut client = self.db.get().await?;
let transaction = client.transaction().await?;
ensure_active_user(&transaction, user_id).await?;
Db::set_tenant(&transaction, user_id).await?;
transaction
.execute(
"INSERT INTO cookiecloud_credentials \
(user_id,host,secrets_ciphertext,secrets_nonce) VALUES($1,$2,$3,$4) \
ON CONFLICT(user_id) DO UPDATE SET host=EXCLUDED.host,\
secrets_ciphertext=EXCLUDED.secrets_ciphertext,\
secrets_nonce=EXCLUDED.secrets_nonce,updated_at=now()",
&[&user_id, &host, &encrypted.ciphertext, &encrypted.nonce],
)
.await?;
insert_audit(
&transaction,
Some(user_id),
"cookiecloud.credentials.updated",
"user",
Some(user_id.to_string()),
json!({"host":host}),
)
.await?;
transaction.commit().await?;
Ok(())
Ok((host, encrypted))
}
pub async fn get_cookiecloud_credentials(
@@ -1274,6 +1355,33 @@ impl AuthService {
}
}
async fn upsert_cookiecloud_credentials(
transaction: &Transaction<'_>,
user_id: Uuid,
host: &str,
encrypted: &EncryptedSecret,
) -> Result<(), AuthError> {
transaction
.execute(
"INSERT INTO cookiecloud_credentials \
(user_id,host,secrets_ciphertext,secrets_nonce) VALUES($1,$2,$3,$4) \
ON CONFLICT(user_id) DO UPDATE SET host=EXCLUDED.host,\
secrets_ciphertext=EXCLUDED.secrets_ciphertext,\
secrets_nonce=EXCLUDED.secrets_nonce,updated_at=now()",
&[&user_id, &host, &encrypted.ciphertext, &encrypted.nonce],
)
.await?;
Ok(())
}
fn map_room_write_error(error: tokio_postgres::Error) -> AuthError {
if error.code() == Some(&SqlState::UNIQUE_VIOLATION) {
AuthError::RoomUnavailable
} else {
error.into()
}
}
#[derive(Clone)]
struct SecretCipher {
cipher: Arc<XChaCha20Poly1305>,
apps/server-rust/src/http_api.rs
+82 -11
View File
@@ -68,6 +68,10 @@ pub fn router(state: AppState) -> Router {
"/api/v1/account/live-source/reconnect",
post(reconnect_source),
)
.route(
"/api/v1/account/preferences",
get(get_account_preferences).put(put_account_preferences),
)
.route(
"/api/v1/components",
get(list_components).post(create_component),
@@ -275,6 +279,7 @@ async fn register_start(
struct EnrollmentConfirmRequest {
enrollment_token: String,
code: String,
language: Option<String>,
}
async fn enrollment_confirm(
@@ -284,9 +289,20 @@ async fn enrollment_confirm(
) -> Result<Response, ApiError> {
same_origin(&state, &headers)?;
consume_enrollment_budget(&state, &headers).await?;
let requested_language = body
.language
.unwrap_or_else(|| crate::i18n::default_language().to_owned());
let language = requested_language.trim();
if !crate::i18n::is_supported(language) {
return Err(ApiError::new(
StatusCode::BAD_REQUEST,
"invalid_input",
"Language is not supported",
));
}
let result = state
.auth
.registration_confirm(&body.enrollment_token, &body.code)
.registration_confirm(&body.enrollment_token, &body.code, language)
.await?;
if result.user.role == UserRole::SystemAdmin {
if let Err(error) = state
@@ -520,13 +536,7 @@ async fn put_source(
) -> Result<Json<Value>, ApiError> {
same_origin(&state, &headers)?;
let session = require_session(&state, &headers).await?;
if body.room_id.trim() != session.user.room_id {
return Err(ApiError::new(
StatusCode::BAD_REQUEST,
"room_is_immutable",
"The room bound by the invitation cannot be changed",
));
}
let room_id = body.room_id.trim().to_owned();
let existing = state
.auth
.get_cookiecloud_credentials(session.user.id)
@@ -577,7 +587,7 @@ async fn put_source(
})?;
state
.auth
.set_cookiecloud_credentials(session.user.id, &credentials)
.set_account_live_source(session.user.id, &room_id, &credentials)
.await?;
state
.restart_user_source(session.user.id)
@@ -599,6 +609,50 @@ async fn reconnect_source(
Ok(Json(json!({"ok":true})))
}
async fn get_account_preferences(
State(state): State<AppState>,
headers: HeaderMap,
) -> Result<Json<Value>, ApiError> {
let session = require_session(&state, &headers).await?;
let language = state.repository.account_language(session.user.id).await?;
Ok(Json(json!({"language":language})))
}
#[derive(Deserialize)]
struct PutAccountPreferences {
language: String,
}
async fn put_account_preferences(
State(state): State<AppState>,
headers: HeaderMap,
Json(body): Json<PutAccountPreferences>,
) -> Result<Json<Value>, ApiError> {
same_origin(&state, &headers)?;
let session = require_session(&state, &headers).await?;
let language = state
.repository
.set_account_language(session.user.id, body.language.trim())
.await?;
let room_id = state.repository.room_id(session.user.id).await?;
for view in state.repository.list_components(session.user.id).await? {
let component = state
.repository
.get_component(session.user.id, view.id)
.await?;
state.hub.publish(
component.id,
Arc::new(ComponentMessage::new(
&component,
&room_id,
"component.language.updated",
json!({"language":language}),
)),
);
}
Ok(Json(json!({"language":language})))
}
async fn list_components(
State(state): State<AppState>,
headers: HeaderMap,
@@ -1016,9 +1070,20 @@ async fn component_ws(mut socket: WebSocket, state: AppState, component_id: Uuid
}
};
let mut receiver = state.hub.subscribe(component_id);
let language = match state
.repository
.account_language(identity.owner_user_id)
.await
{
Ok(language) => language,
Err(_) => {
close_ws(&mut socket, "Component is unavailable").await;
return;
}
};
if socket
.send(Message::Text(
json!({"version":COMPONENT_PROTOCOL_VERSION,"type":"authenticated","componentId":component_id,"componentKind":component.kind})
json!({"version":COMPONENT_PROTOCOL_VERSION,"type":"authenticated","componentId":component_id,"componentKind":component.kind,"language":language})
.to_string()
.into(),
))
@@ -1296,9 +1361,11 @@ fn constant_time_secret_eq(provided: &str, expected: &str) -> bool {
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct ErrorBody {
error: &'static str,
code: &'static str,
message_key: String,
message: String,
}
@@ -1345,6 +1412,7 @@ impl IntoResponse for ApiError {
Json(ErrorBody {
error: self.code,
code: self.code,
message_key: format!("api.error.{}", self.code),
message: self.message,
}),
)
@@ -1380,11 +1448,14 @@ impl From<AuthError> for ApiError {
"enrollment_unavailable",
error.to_string(),
),
AuthError::AccountUnavailable | AuthError::RoomUnavailable => Self::new(
AuthError::AccountUnavailable => Self::new(
StatusCode::CONFLICT,
"account_unavailable",
error.to_string(),
),
AuthError::RoomUnavailable => {
Self::new(StatusCode::CONFLICT, "room_unavailable", error.to_string())
}
AuthError::InvalidTotp
| AuthError::TotpReplay
| AuthError::InvalidCredentials
apps/server-rust/src/i18n.rs
+65
View File
@@ -0,0 +1,65 @@
//! Shared language-resource validation.
//!
//! Rust embeds the same TOML file that Vite compiles into the browser bundle.
//! Backend persistence accepts only locale codes declared by that resource, so
//! the database and deployed frontend cannot drift into unsupported values.
use std::{collections::HashMap, sync::OnceLock};
use serde::Deserialize;
const RESOURCE: &str = include_str!("../../../resources/i18n.toml");
#[derive(Deserialize)]
struct Catalog {
default_language: String,
locales: HashMap<String, Locale>,
}
#[derive(Deserialize)]
struct Locale {
#[allow(dead_code)]
name: String,
messages: HashMap<String, String>,
}
fn catalog() -> &'static Catalog {
static CATALOG: OnceLock<Catalog> = OnceLock::new();
CATALOG.get_or_init(|| {
let catalog: Catalog = toml::from_str(RESOURCE).expect("resources/i18n.toml must be valid");
let default = catalog
.locales
.get(&catalog.default_language)
.expect("default language must be declared");
for (code, locale) in &catalog.locales {
for key in default.messages.keys() {
assert!(
locale.messages.contains_key(key),
"locale {code} is missing translation key {key}"
);
}
}
catalog
})
}
pub fn default_language() -> &'static str {
&catalog().default_language
}
pub fn is_supported(language: &str) -> bool {
catalog().locales.contains_key(language)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn shared_catalog_is_complete_and_declares_supported_languages() {
assert_eq!(default_language(), "zh-CN");
assert!(is_supported("zh-CN"));
assert!(is_supported("en-US"));
assert!(!is_supported("unknown"));
}
}
apps/server-rust/src/lib.rs
+1
View File
@@ -13,6 +13,7 @@ pub mod credentials;
pub mod db;
pub mod domain;
pub mod http_api;
pub mod i18n;
pub mod live;
pub mod overlay;
pub mod rate_limit;
apps/server-rust/src/repository.rs
+39
View File
@@ -15,6 +15,7 @@ use uuid::Uuid;
use crate::{
components::{ComponentInstance, ComponentRegistry},
db::{ComponentRecord, Db, DbError},
i18n,
realtime::InMemoryComponentStore,
song_request::{SONG_REQUEST_KIND, SONG_REQUEST_NAME, SongRequestSettings},
};
@@ -300,6 +301,44 @@ impl TenantRepository {
Ok(row.get(0))
}
pub async fn account_language(&self, owner_id: Uuid) -> Result<String, RepositoryError> {
let mut client = self.db.get().await?;
let transaction = client.transaction().await?;
Db::set_tenant(&transaction, owner_id).await?;
let language = transaction
.query_opt(
"SELECT language FROM account_preferences WHERE user_id=$1",
&[&owner_id],
)
.await?
.map(|row| row.get(0))
.unwrap_or_else(|| i18n::default_language().to_owned());
transaction.commit().await?;
Ok(language)
}
pub async fn set_account_language(
&self,
owner_id: Uuid,
language: &str,
) -> Result<String, RepositoryError> {
if !i18n::is_supported(language) {
return Err(RepositoryError::Invalid("language is not supported".into()));
}
let mut client = self.db.get().await?;
let transaction = client.transaction().await?;
Db::set_tenant(&transaction, owner_id).await?;
transaction
.execute(
"INSERT INTO account_preferences(user_id,language) VALUES($1,$2) \
ON CONFLICT(user_id) DO UPDATE SET language=EXCLUDED.language,updated_at=now()",
&[&owner_id, &language],
)
.await?;
transaction.commit().await?;
Ok(language.to_owned())
}
pub async fn token_summary(
&self,
owner_id: Uuid,