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

move live source ownership to accounts

Browse Source
This commit is contained in:
felis
2026-07-18 22:24:00 -07:00
parent 1598d3c403
commit 53ae90ec13
20 changed files with 219 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/server-rust/src/app.rs
+4
View File
@@ -279,6 +279,10 @@ 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")),
(
6_i32,
include_str!("../migrations/006_account_live_source.sql"),
),
] {
let applied = transaction
.query_one(
apps/server-rust/src/auth.rs
+8 -21
View File
@@ -611,14 +611,9 @@ impl AuthService {
.expect("OverlaySettings is always JSON serializable");
transaction
.execute(
"INSERT INTO component_instances(id,owner_user_id,source_id,kind,name,settings) \
VALUES($1,$2,$3,'danmaku_overlay','弹幕姬',$4)",
&[
&default_component_id,
&user_id,
&default_source_id,
&default_settings,
],
"INSERT INTO component_instances(id,owner_user_id,kind,name,settings) \
VALUES($1,$2,'danmaku_overlay','弹幕姬',$3)",
&[&default_component_id, &user_id, &default_settings],
)
.await?;
let song_component_id = Uuid::new_v4();
@@ -627,12 +622,11 @@ impl AuthService {
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)",
(id,owner_user_id,kind,name,settings,settings_version,enabled) \
VALUES($1,$2,$3,$4,$5,1,true)",
&[
&song_component_id,
&user_id,
&default_source_id,
&SONG_REQUEST_KIND,
&SONG_REQUEST_NAME,
&song_settings,
@@ -1240,13 +1234,6 @@ impl AuthService {
.ok_or(AuthError::InvalidCredentials)?
.get(0);
Db::set_tenant(&transaction, owner_user_id).await?;
let source_id: Uuid = transaction
.query_one(
"SELECT id FROM live_sources WHERE owner_user_id=$1",
&[&owner_user_id],
)
.await?
.get(0);
let component_id = if let Some(row) = transaction
.query_opt(
"SELECT id FROM component_instances \
@@ -1268,9 +1255,9 @@ impl AuthService {
let id = Uuid::new_v4();
transaction
.execute(
"INSERT INTO component_instances(id,owner_user_id,source_id,kind,name,settings) \
VALUES($1,$2,$3,'danmaku_overlay','弹幕姬',$4)",
&[&id, &owner_user_id, &source_id, &settings],
"INSERT INTO component_instances(id,owner_user_id,kind,name,settings) \
VALUES($1,$2,'danmaku_overlay','弹幕姬',$3)",
&[&id, &owner_user_id, &settings],
)
.await?;
id
apps/server-rust/src/components.rs
+6 -3
View File
@@ -84,7 +84,10 @@ pub struct ComponentInstance {
pub id: Uuid,
#[serde(skip_serializing)]
pub owner_id: Uuid,
pub source_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,
@@ -95,7 +98,7 @@ pub struct ComponentInstance {
impl ComponentInstance {
pub fn new(
owner_id: Uuid,
source_id: Uuid,
account_source_id: Uuid,
kind: impl Into<String>,
name: impl Into<String>,
settings_version: u32,
@@ -104,7 +107,7 @@ impl ComponentInstance {
Self {
id: Uuid::new_v4(),
owner_id,
source_id,
account_source_id,
kind: kind.into(),
name: name.into(),
enabled: true,
apps/server-rust/src/db.rs
+9 -4
View File
@@ -108,8 +108,11 @@ impl Db {
Self::set_tenant(&transaction, user_id).await?;
let rows = transaction
.query(
"SELECT id,source_id,kind,name,settings,settings_version,enabled \
FROM component_instances WHERE owner_user_id=$1 ORDER BY created_at",
"SELECT component.id,source.id,component.kind,component.name,component.settings,\
component.settings_version,component.enabled \
FROM component_instances AS component \
JOIN live_sources AS source ON source.owner_user_id=component.owner_user_id \
WHERE component.owner_user_id=$1 ORDER BY component.created_at",
&[&user_id],
)
.await?;
@@ -119,7 +122,7 @@ impl Db {
.map(|row| ComponentRecord {
id: row.get(0),
owner_user_id: user_id,
source_id: row.get(1),
account_source_id: row.get(1),
kind: row.get(2),
name: row.get(3),
settings: row.get(4),
@@ -141,7 +144,9 @@ pub struct ActiveTenant {
pub struct ComponentRecord {
pub id: Uuid,
pub owner_user_id: Uuid,
pub source_id: Uuid,
/// Derived from the account's singleton live source; it is not component
/// configuration and is never persisted on `component_instances`.
pub account_source_id: Uuid,
pub kind: String,
pub name: String,
pub settings: Value,
apps/server-rust/src/domain.rs
+1 -1
View File
@@ -288,7 +288,7 @@ impl ComponentMessage {
Self {
owner_id: component.owner_id,
component_id: component.id,
source_id: component.source_id,
source_id: component.account_source_id,
version: COMPONENT_PROTOCOL_VERSION,
id: Uuid::new_v4(),
occurred_at: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
apps/server-rust/src/http_api.rs
+10 -1
View File
@@ -60,6 +60,14 @@ pub fn router(state: AppState) -> Router {
.route("/api/v1/invitations/{id}", delete(revoke_invitation))
.route("/api/v1/source", get(get_source).put(put_source))
.route("/api/v1/source/reconnect", post(reconnect_source))
.route(
"/api/v1/account/live-source",
get(get_source).put(put_source),
)
.route(
"/api/v1/account/live-source/reconnect",
post(reconnect_source),
)
.route(
"/api/v1/components",
get(list_components).post(create_component),
@@ -108,6 +116,7 @@ pub fn router(state: AppState) -> Router {
.route("/control/register", get(spa_page))
.route("/control/setup", get(spa_page))
.route("/control/invitations", get(spa_page))
.route("/control/account", get(spa_page))
.route("/obs/{public_id}", get(spa_page))
.fallback_service(
ServeDir::new("/app/web").not_found_service(ServeFile::new("/app/web/index.html")),
@@ -899,7 +908,7 @@ async fn component_test_event(
};
let mut event = LiveEvent::new(
component.owner_id,
component.source_id,
component.account_source_id,
"test",
session.user.room_id,
payload,
apps/server-rust/src/live/mod.rs
+3 -3
View File
@@ -42,9 +42,9 @@ impl SourceStatus {
#[derive(Clone)]
pub struct SourceContext {
pub owner_id: Uuid,
/// Stable database identity for the account's single fixed live source.
/// It remains distinct from the owner id so future provider/source models
/// do not leak the current one-room product rule into event contracts.
/// Stable identity for the account's single fixed listener. Components do
/// not persist or select this ID; it remains in event envelopes so clients
/// can identify the shared upstream connection.
pub source_id: Uuid,
pub room_id: String,
}
apps/server-rust/src/live/supervisor.rs
+2 -2
View File
@@ -1,4 +1,4 @@
//! Per-source task ownership, restart and cancellation.
//! Per-account 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
@@ -27,7 +27,7 @@ struct RunningSource {
status: watch::Receiver<SourceStatus>,
}
/// Owns exactly one provider task per account/source. Restart always cancels
/// Owns exactly one provider task per account. Restart always cancels
/// the prior generation before starting another, which prevents the duplicate
/// listeners produced by the legacy `/reconnect` handler.
#[derive(Clone)]
apps/server-rust/src/realtime.rs
+39 -32
View File
@@ -1,6 +1,6 @@
//! Tenant-aware event routing and component-scoped realtime fanout.
//!
//! The router resolves enabled instances by owner and source, validates their
//! The router resolves every enabled instance owned by the account, validates its
//! 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.
@@ -126,15 +126,11 @@ impl Error for ComponentStoreError {}
pub type ComponentStoreFuture<'a> =
Pin<Box<dyn Future<Output = Result<Vec<ComponentInstance>, ComponentStoreError>> + Send + 'a>>;
/// Persistence port used by source routing. A PostgreSQL implementation should
/// always scope its query by both owner and source; the router repeats that
/// check as defense in depth.
/// Persistence port used by account-stream routing. A provider event is offered
/// to every enabled component owned by that account; each component definition
/// then filters it through its event subscription.
pub trait ComponentInstanceStore: Send + Sync {
fn list_enabled_for_source<'a>(
&'a self,
owner_id: Uuid,
source_id: Uuid,
) -> ComponentStoreFuture<'a>;
fn list_enabled_for_owner<'a>(&'a self, owner_id: Uuid) -> ComponentStoreFuture<'a>;
}
#[derive(Clone, Default)]
@@ -173,22 +169,14 @@ impl InMemoryComponentStore {
}
impl ComponentInstanceStore for InMemoryComponentStore {
fn list_enabled_for_source<'a>(
&'a self,
owner_id: Uuid,
source_id: Uuid,
) -> ComponentStoreFuture<'a> {
fn list_enabled_for_owner<'a>(&'a self, owner_id: Uuid) -> ComponentStoreFuture<'a> {
Box::pin(async move {
Ok(self
.instances
.read()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.iter()
.filter(|instance| {
instance.enabled
&& instance.owner_id == owner_id
&& instance.source_id == source_id
})
.filter(|instance| instance.enabled && instance.owner_id == owner_id)
.cloned()
.collect())
})
@@ -229,7 +217,7 @@ pub enum RouteError {
impl fmt::Display for RouteError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Store(error) => write!(formatter, "cannot resolve source components: {error}"),
Self::Store(error) => write!(formatter, "cannot resolve account components: {error}"),
}
}
}
@@ -282,7 +270,7 @@ impl SourceEventRouter {
pub async fn route(&self, event: Arc<LiveEvent>) -> Result<RouteReport, RouteError> {
let components = self
.components
.list_enabled_for_source(event.owner_id, event.source_id)
.list_enabled_for_owner(event.owner_id)
.await?;
let mut report = RouteReport {
considered: components.len(),
@@ -290,14 +278,11 @@ impl SourceEventRouter {
};
for component in components {
if !component.enabled
|| component.owner_id != event.owner_id
|| component.source_id != event.source_id
{
if !component.enabled || component.owner_id != event.owner_id {
report.failures.push(RouteFailure {
component_id: component.id,
stage: RouteStage::Scope,
detail: "component owner/source does not match the source event".into(),
detail: "component owner does not match the account event stream".into(),
});
continue;
}
@@ -349,7 +334,7 @@ impl SourceEventRouter {
match runtime.project(&component, &event) {
Ok(Some(message)) => {
if message.owner_id != component.owner_id
|| message.source_id != component.source_id
|| message.source_id != event.source_id
|| message.component_id != component.id
{
report.failures.push(RouteFailure {
@@ -517,16 +502,38 @@ mod tests {
assert!(report.failures.is_empty());
}
#[tokio::test]
async fn one_account_event_is_offered_to_all_owned_components() {
let owner_id = Uuid::new_v4();
let account_source_id = Uuid::new_v4();
let first = overlay(owner_id, account_source_id);
let second = overlay(owner_id, account_source_id);
let first_id = first.id;
let second_id = second.id;
let hub = EventHub::new(8);
let mut first_rx = hub.subscribe(first_id);
let mut second_rx = hub.subscribe(second_id);
let store = Arc::new(InMemoryComponentStore::new(vec![first, second]));
let router = SourceEventRouter::new(ComponentRegistry::default(), store, hub);
let report = router
.route(danmaku(owner_id, account_source_id))
.await
.unwrap();
assert_eq!(report.considered, 2);
assert_eq!(report.projected, 2);
assert_eq!(first_rx.try_recv().unwrap().component_id, first_id);
assert_eq!(second_rx.try_recv().unwrap().component_id, second_id);
assert!(report.failures.is_empty());
}
struct LeakyStore {
instances: Vec<ComponentInstance>,
}
impl ComponentInstanceStore for LeakyStore {
fn list_enabled_for_source<'a>(
&'a self,
_owner_id: Uuid,
_source_id: Uuid,
) -> ComponentStoreFuture<'a> {
fn list_enabled_for_owner<'a>(&'a self, _owner_id: Uuid) -> ComponentStoreFuture<'a> {
Box::pin(async move { Ok(self.instances.clone()) })
}
}
apps/server-rust/src/repository.rs
+12 -11
View File
@@ -2,7 +2,7 @@
//!
//! 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
//! account-source restarts hydrate that cache from tenant-scoped rows before events are
//! routed.
use std::{fmt, sync::Arc};
@@ -69,12 +69,11 @@ impl TenantRepository {
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,owner_user_id,kind,name,settings,settings_version,enabled) \
VALUES($1,$2,$3,$4,$5,1,true) ON CONFLICT DO NOTHING",
&[
&id,
&tenant.user_id,
&tenant.source_id,
&SONG_REQUEST_KIND,
&SONG_REQUEST_NAME,
&settings,
@@ -159,12 +158,11 @@ impl TenantRepository {
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,$7,true)",
(id,owner_user_id,kind,name,settings,settings_version,enabled) \
VALUES($1,$2,$3,$4,$5,$6,true)",
&[
&component.id,
&owner_id,
&source_id,
&component.kind,
&component.name,
&component.settings,
@@ -220,8 +218,11 @@ impl TenantRepository {
Db::set_tenant(&transaction, owner_id).await?;
let row = transaction
.query_opt(
"SELECT id,source_id,kind,name,settings,settings_version,enabled \
FROM component_instances WHERE owner_user_id=$1 AND id=$2",
"SELECT component.id,source.id,component.kind,component.name,component.settings,\
component.settings_version,component.enabled \
FROM component_instances AS component \
JOIN live_sources AS source ON source.owner_user_id=component.owner_user_id \
WHERE component.owner_user_id=$1 AND component.id=$2",
&[&owner_id, &component_id],
)
.await?
@@ -230,7 +231,7 @@ impl TenantRepository {
self.validate_loaded_component(component_from_record(ComponentRecord {
id: row.get(0),
owner_user_id: owner_id,
source_id: row.get(1),
account_source_id: row.get(1),
kind: row.get(2),
name: row.get(3),
settings: row.get(4),
@@ -423,7 +424,7 @@ fn component_from_record(record: ComponentRecord) -> Result<ComponentInstance, R
Ok(ComponentInstance {
id: record.id,
owner_id: record.owner_user_id,
source_id: record.source_id,
account_source_id: record.account_source_id,
kind: record.kind,
name: record.name,
enabled: record.enabled,
apps/server-rust/src/song_request.rs
+2 -2
View File
@@ -678,8 +678,8 @@ async fn ensure_song_component(
let valid: bool = transaction
.query_one(
"SELECT EXISTS(SELECT 1 FROM component_instances WHERE owner_user_id=$1 \
AND id=$2 AND source_id=$3 AND kind='song_request' AND enabled)",
&[&component.owner_id, &component.id, &component.source_id],
AND id=$2 AND kind='song_request' AND enabled)",
&[&component.owner_id, &component.id],
)
.await?
.get(0);