proper productionize project

This commit is contained in:
2026-07-15 23:54:53 -07:00
parent b8850c87a4
commit edb6d2b5b4
46 changed files with 11346 additions and 1560 deletions
+211 -4
View File
@@ -8,6 +8,16 @@ version = "2.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa"
[[package]]
name = "aead"
version = "0.5.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d122413f284cf2d62fb1b7db97e02edb8cda96d769b16e443a4f6195e35662b0"
dependencies = [
"crypto-common 0.1.7",
"generic-array",
]
[[package]]
name = "aho-corasick"
version = "1.1.4"
@@ -209,6 +219,12 @@ dependencies = [
"tracing",
]
[[package]]
name = "base32"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "022dfe9eb35f19ebbcb51e0b40a5ab759f46ad60cadf7297e0bd085afb50e076"
[[package]]
name = "base64"
version = "0.21.7"
@@ -401,6 +417,17 @@ version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724"
[[package]]
name = "chacha20"
version = "0.9.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c3613f74bd2eac03dad61bd53dbe620703d4371614fe0bc3b9f04dd36fe4e818"
dependencies = [
"cfg-if",
"cipher",
"cpufeatures 0.2.17",
]
[[package]]
name = "chacha20"
version = "0.10.1"
@@ -412,6 +439,19 @@ dependencies = [
"rand_core 0.10.1",
]
[[package]]
name = "chacha20poly1305"
version = "0.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "10cd79432192d1c0f4e1a0fef9527696cc039165d729fb41b3f4f4f354c2dc35"
dependencies = [
"aead",
"chacha20 0.9.1",
"cipher",
"poly1305",
"zeroize",
]
[[package]]
name = "chrono"
version = "0.4.45"
@@ -420,9 +460,21 @@ checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327"
dependencies = [
"iana-time-zone",
"num-traits",
"serde",
"windows-link",
]
[[package]]
name = "cipher"
version = "0.4.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad"
dependencies = [
"crypto-common 0.1.7",
"inout",
"zeroize",
]
[[package]]
name = "clang-sys"
version = "1.8.1"
@@ -540,6 +592,12 @@ version = "0.10.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c"
[[package]]
name = "constant_time_eq"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7c74b8349d32d297c9134b8c88677813a227df8f779daa29bfc29c183fe3dca6"
[[package]]
name = "cookie"
version = "0.17.0"
@@ -702,6 +760,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a"
dependencies = [
"generic-array",
"rand_core 0.6.4",
"typenum",
]
@@ -769,6 +828,41 @@ version = "2.11.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8"
[[package]]
name = "deadpool"
version = "0.12.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0be2b1d1d6ec8d846f05e137292d0b89133caf95ef33695424c09568bdd39b1b"
dependencies = [
"deadpool-runtime",
"lazy_static",
"num_cpus",
"tokio",
]
[[package]]
name = "deadpool-postgres"
version = "0.14.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3d697d376cbfa018c23eb4caab1fd1883dd9c906a8c034e8d9a3cb06a7e0bef9"
dependencies = [
"async-trait",
"deadpool",
"getrandom 0.2.17",
"tokio",
"tokio-postgres",
"tracing",
]
[[package]]
name = "deadpool-runtime"
version = "0.1.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "092966b41edc516079bdf31ec78a2e0588d1d0c08f78b91d8307215928642b2b"
dependencies = [
"tokio",
]
[[package]]
name = "defmt"
version = "1.1.1"
@@ -1228,6 +1322,12 @@ version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea"
[[package]]
name = "hermit-abi"
version = "0.5.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c"
[[package]]
name = "hmac"
version = "0.12.1"
@@ -1608,6 +1708,15 @@ dependencies = [
"rustversion",
]
[[package]]
name = "inout"
version = "0.1.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01"
dependencies = [
"generic-array",
]
[[package]]
name = "instability"
version = "0.3.12"
@@ -1832,21 +1941,25 @@ checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154"
name = "lxc-stream-server"
version = "0.1.0"
dependencies = [
"async-trait",
"axum",
"base64 0.22.1",
"blivedm",
"chacha20poly1305",
"chrono",
"futures",
"deadpool-postgres",
"futures-channel",
"hmac 0.12.1",
"http 1.4.2",
"rand 0.9.5",
"reqwest 0.12.28",
"serde",
"serde_json",
"sha2 0.10.9",
"subtle",
"tokio",
"tokio-postgres",
"tokio-util",
"toml",
"totp-rs",
"tower-http",
"tracing",
"tracing-subscriber",
@@ -2053,6 +2166,16 @@ dependencies = [
"autocfg",
]
[[package]]
name = "num_cpus"
version = "1.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "91df4bbde75afed763b708b7eee1e8e7651e02d97f6d5dd763e89367e957b23b"
dependencies = [
"hermit-abi",
"libc",
]
[[package]]
name = "num_enum"
version = "0.7.6"
@@ -2201,6 +2324,12 @@ version = "1.70.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe"
[[package]]
name = "opaque-debug"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381"
[[package]]
name = "openssl"
version = "0.10.81"
@@ -2350,6 +2479,17 @@ dependencies = [
"miniz_oxide",
]
[[package]]
name = "poly1305"
version = "0.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8159bd90725d2df49889a078b54f4f79e87f1f8a8444194cdca81d38f5393abf"
dependencies = [
"cpufeatures 0.2.17",
"opaque-debug",
"universal-hash",
]
[[package]]
name = "portable-atomic"
version = "1.13.1"
@@ -2390,10 +2530,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "851ca9db4932932d69f3ea811b1abe63087a0f740a47692619dd40d4899b68be"
dependencies = [
"bytes",
"chrono",
"fallible-iterator",
"postgres-protocol",
"serde_core",
"serde_json",
"uuid",
]
[[package]]
@@ -2460,6 +2602,23 @@ version = "0.1.30"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d55d956fa96f5ec02be2e13af0e20391a5aa83d6a074e3ad368959d0fab299ea"
[[package]]
name = "qrcodegen"
version = "1.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4339fc7a1021c9c1621d87f5e3505f2805c8c105420ba2f2a4df86814590c142"
[[package]]
name = "qrcodegen-image"
version = "1.5.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2e3dd60f5b603f72c307455fc52deec52ada1ba53c7580918bb2a8e3247d4fe7"
dependencies = [
"base64 0.22.1",
"image",
"qrcodegen",
]
[[package]]
name = "quick-error"
version = "2.0.1"
@@ -2579,7 +2738,7 @@ version = "0.10.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80"
dependencies = [
"chacha20",
"chacha20 0.10.1",
"getrandom 0.4.3",
"rand_core 0.10.1",
]
@@ -3653,6 +3812,24 @@ version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801"
[[package]]
name = "totp-rs"
version = "5.7.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "50e69a15e21b2ff22c415446983978bded3244195f17d59cb113551c1e806f91"
dependencies = [
"base32",
"constant_time_eq",
"hmac 0.12.1",
"qrcodegen-image",
"rand 0.9.5",
"sha1",
"sha2 0.10.9",
"url",
"urlencoding",
"zeroize",
]
[[package]]
name = "tower"
version = "0.5.3"
@@ -3903,6 +4080,16 @@ version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fc81956842c57dac11422a97c3b8195a1ff727f06e85c84ed2e8aa277c9a0fd"
[[package]]
name = "universal-hash"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fc1de2c688dc15305988b563c3854064043356019f97a4b46276fe734c4f07ea"
dependencies = [
"crypto-common 0.1.7",
"subtle",
]
[[package]]
name = "untrusted"
version = "0.9.0"
@@ -3921,6 +4108,12 @@ dependencies = [
"serde",
]
[[package]]
name = "urlencoding"
version = "2.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da"
[[package]]
name = "utf-8"
version = "0.7.6"
@@ -4773,6 +4966,20 @@ name = "zeroize"
version = "1.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e"
dependencies = [
"zeroize_derive",
]
[[package]]
name = "zeroize_derive"
version = "1.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3c50655cbb0fe3fc43170059e702f1ce5e19b84cec58dc87b037a09935c2f328"
dependencies = [
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "zerotrie"
+9 -5
View File
@@ -5,23 +5,27 @@ edition = "2024"
[dependencies]
axum = { version = "0.8", features = ["ws", "json"] }
async-trait = "0.1"
base64 = "0.22"
# Patched local copy of the published blivedm_rs crate. The patch preserves
# the upstream raw payload so the application can retain UID, price and event
# identifiers for atomic accounting and gift de-duplication.
blivedm = { path = "../../vendor/blivedm", default-features = false }
chrono = { version = "0.4", default-features = false, features = ["clock"] }
futures = "0.3"
chrono = { version = "0.4", default-features = false, features = ["clock", "serde"] }
chacha20poly1305 = "0.10"
deadpool-postgres = "0.14"
futures-channel = "0.3"
hmac = "0.12"
http = "1"
rand = "0.9"
reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
sha2 = "0.10"
subtle = "2"
tokio = { version = "1", features = ["full"] }
tokio-postgres = { version = "0.7", features = ["with-serde_json-1"] }
tokio-postgres = { version = "0.7", features = ["with-serde_json-1", "with-uuid-1", "with-chrono-0_4"] }
tokio-util = "0.7"
toml = "0.8"
totp-rs = { version = "5.7", features = ["gen_secret", "qr", "zeroize"] }
tower-http = { version = "0.6", features = ["fs"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
@@ -0,0 +1,339 @@
-- Identity, tenant ownership and component foundations.
--
-- Raw invitation, enrollment, session, recovery and component access tokens
-- must never be stored in PostgreSQL. Their SHA-256 digests are the only
-- persisted representation. TOTP and CookieCloud secrets are encrypted by the
-- application with XChaCha20-Poly1305 before they reach this schema.
CREATE TABLE IF NOT EXISTS users (
id UUID PRIMARY KEY,
username TEXT NOT NULL,
username_normalized TEXT NOT NULL UNIQUE,
room_id TEXT NOT NULL UNIQUE CHECK (room_id ~ '^[1-9][0-9]*$'),
role TEXT NOT NULL CHECK (role IN ('system_admin', 'user')),
status TEXT NOT NULL CHECK (status IN ('active', 'disabled')),
totp_secret_ciphertext BYTEA NOT NULL CHECK (octet_length(totp_secret_ciphertext) >= 16),
totp_secret_nonce BYTEA NOT NULL CHECK (octet_length(totp_secret_nonce) = 24),
last_totp_step BIGINT,
totp_enrolled_at TIMESTAMPTZ NOT NULL,
auth_version BIGINT NOT NULL DEFAULT 1,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
disabled_at TIMESTAMPTZ
);
-- A room is an account invariant rather than editable profile data.
CREATE OR REPLACE FUNCTION prevent_user_room_id_change()
RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
IF NEW.room_id IS DISTINCT FROM OLD.room_id THEN
RAISE EXCEPTION 'a user room_id is immutable';
END IF;
RETURN NEW;
END;
$$;
DROP TRIGGER IF EXISTS users_room_id_immutable ON users;
CREATE TRIGGER users_room_id_immutable
BEFORE UPDATE OF room_id ON users
FOR EACH ROW EXECUTE FUNCTION prevent_user_room_id_change();
CREATE TABLE IF NOT EXISTS invitations (
id UUID PRIMARY KEY,
code_digest BYTEA NOT NULL UNIQUE CHECK (octet_length(code_digest) = 32),
code_prefix TEXT NOT NULL,
room_id TEXT NOT NULL CHECK (room_id ~ '^[1-9][0-9]*$'),
grant_role TEXT NOT NULL DEFAULT 'user' CHECK (grant_role IN ('system_admin', 'user')),
created_by UUID REFERENCES users(id) ON DELETE RESTRICT,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
expires_at TIMESTAMPTZ NOT NULL,
consumed_by UUID UNIQUE REFERENCES users(id) ON DELETE RESTRICT,
consumed_at TIMESTAMPTZ,
revoked_at TIMESTAMPTZ,
CHECK ((consumed_by IS NULL) = (consumed_at IS NULL)),
-- Only the one-time bootstrap path may mint the first system administrator.
CHECK (grant_role <> 'system_admin' OR created_by IS NULL)
);
CREATE INDEX IF NOT EXISTS invitations_created_by_idx
ON invitations(created_by, created_at DESC);
CREATE INDEX IF NOT EXISTS invitations_room_idx
ON invitations(room_id, expires_at DESC);
-- Pending enrollment is deliberately separate from users. An account does not
-- exist until a valid TOTP has been confirmed. Rows are short lived and are
-- pruned opportunistically by registration calls.
CREATE TABLE IF NOT EXISTS pending_registrations (
id UUID PRIMARY KEY,
enrollment_token_digest BYTEA NOT NULL UNIQUE
CHECK (octet_length(enrollment_token_digest) = 32),
invitation_id UUID NOT NULL UNIQUE REFERENCES invitations(id) ON DELETE CASCADE,
username TEXT NOT NULL,
username_normalized TEXT NOT NULL UNIQUE,
room_id TEXT NOT NULL UNIQUE CHECK (room_id ~ '^[1-9][0-9]*$'),
totp_secret_ciphertext BYTEA NOT NULL CHECK (octet_length(totp_secret_ciphertext) >= 16),
totp_secret_nonce BYTEA NOT NULL CHECK (octet_length(totp_secret_nonce) = 24),
failed_attempts INTEGER NOT NULL DEFAULT 0 CHECK (failed_attempts >= 0),
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
expires_at TIMESTAMPTZ NOT NULL
);
CREATE INDEX IF NOT EXISTS pending_registrations_expiry_idx
ON pending_registrations(expires_at);
CREATE TABLE IF NOT EXISTS user_sessions (
id UUID PRIMARY KEY,
user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
token_digest BYTEA NOT NULL UNIQUE CHECK (octet_length(token_digest) = 32),
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
expires_at TIMESTAMPTZ NOT NULL,
last_seen_at TIMESTAMPTZ NOT NULL DEFAULT now(),
revoked_at TIMESTAMPTZ,
user_agent_hash BYTEA,
ip_prefix TEXT
);
CREATE INDEX IF NOT EXISTS user_sessions_active_user_idx
ON user_sessions(user_id, expires_at DESC)
WHERE revoked_at IS NULL;
CREATE INDEX IF NOT EXISTS user_sessions_expiry_idx
ON user_sessions(expires_at)
WHERE revoked_at IS NULL;
CREATE TABLE IF NOT EXISTS recovery_codes (
id UUID PRIMARY KEY,
user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
code_digest BYTEA NOT NULL CHECK (octet_length(code_digest) = 32),
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
consumed_at TIMESTAMPTZ,
UNIQUE (user_id, code_digest)
);
CREATE INDEX IF NOT EXISTS recovery_codes_available_idx
ON recovery_codes(user_id)
WHERE consumed_at IS NULL;
CREATE TABLE IF NOT EXISTS cookiecloud_credentials (
user_id UUID PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
host TEXT NOT NULL,
secrets_ciphertext BYTEA NOT NULL CHECK (octet_length(secrets_ciphertext) >= 16),
secrets_nonce BYTEA NOT NULL CHECK (octet_length(secrets_nonce) = 24),
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- The current product assigns exactly one immutable Bilibili source to an
-- account. Keeping it as an explicit entity gives component routing a stable
-- source_id while preserving the one-account/one-room product rule.
CREATE TABLE IF NOT EXISTS live_sources (
id UUID PRIMARY KEY,
owner_user_id UUID NOT NULL UNIQUE REFERENCES users(id) ON DELETE CASCADE,
provider TEXT NOT NULL DEFAULT 'bilibili' CHECK (provider = 'bilibili'),
room_id TEXT NOT NULL UNIQUE CHECK (room_id ~ '^[1-9][0-9]*$'),
enabled BOOLEAN NOT NULL DEFAULT true,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
UNIQUE (owner_user_id, id)
);
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
OR NEW.room_id IS DISTINCT FROM OLD.room_id) THEN
RAISE EXCEPTION 'live source ownership and room_id are immutable';
END IF;
RETURN NEW;
END;
$$;
DROP TRIGGER IF EXISTS live_source_account_room ON live_sources;
CREATE TRIGGER live_source_account_room
BEFORE INSERT OR UPDATE OF owner_user_id,room_id ON live_sources
FOR EACH ROW EXECUTE FUNCTION enforce_live_source_account_room();
-- Every future OBS feature is a component instance. Component-specific state
-- belongs in dedicated tables when it becomes relational; settings remain JSON
-- so a new renderer does not require a core schema rewrite.
CREATE TABLE IF NOT EXISTS component_instances (
id UUID PRIMARY KEY,
owner_user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
source_id UUID NOT NULL,
kind TEXT NOT NULL CHECK (kind ~ '^[a-z][a-z0-9_.-]{1,63}$'),
name TEXT NOT NULL,
settings JSONB NOT NULL DEFAULT '{}'::jsonb CHECK (jsonb_typeof(settings) = 'object'),
settings_version INTEGER NOT NULL DEFAULT 1 CHECK (settings_version > 0),
enabled BOOLEAN NOT NULL DEFAULT true,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
UNIQUE (owner_user_id, id),
FOREIGN KEY (owner_user_id, source_id)
REFERENCES live_sources(owner_user_id, id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS component_instances_owner_kind_idx
ON component_instances(owner_user_id, kind, created_at);
CREATE INDEX IF NOT EXISTS component_instances_source_idx
ON component_instances(owner_user_id, source_id, enabled);
CREATE TABLE IF NOT EXISTS component_access_tokens (
id UUID PRIMARY KEY,
owner_user_id UUID NOT NULL,
component_instance_id UUID NOT NULL,
label TEXT NOT NULL,
token_prefix TEXT NOT NULL,
token_digest BYTEA NOT NULL UNIQUE CHECK (octet_length(token_digest) = 32),
scopes TEXT[] NOT NULL DEFAULT ARRAY['events:subscribe']::TEXT[],
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
expires_at TIMESTAMPTZ,
last_used_at TIMESTAMPTZ,
revoked_at TIMESTAMPTZ,
FOREIGN KEY (owner_user_id, component_instance_id)
REFERENCES component_instances(owner_user_id, id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS component_access_tokens_component_idx
ON component_access_tokens(component_instance_id, created_at DESC);
CREATE TABLE IF NOT EXISTS audit_log (
id BIGSERIAL PRIMARY KEY,
actor_user_id UUID REFERENCES users(id) ON DELETE SET NULL,
action TEXT NOT NULL,
target_type TEXT NOT NULL,
target_id TEXT,
metadata JSONB NOT NULL DEFAULT '{}'::jsonb CHECK (jsonb_typeof(metadata) = 'object'),
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS audit_log_actor_time_idx
ON audit_log(actor_user_id, created_at DESC);
-- Legacy settings remain readable during the staged migration. main.rs can
-- associate and copy each row into the owner's initial danmaku component, then
-- stop writing this table without a destructive migration.
ALTER TABLE overlay_settings
ADD COLUMN IF NOT EXISTS owner_user_id UUID REFERENCES users(id) ON DELETE SET NULL;
ALTER TABLE overlay_settings
ADD COLUMN IF NOT EXISTS component_instance_id UUID REFERENCES component_instances(id)
ON DELETE SET NULL;
CREATE INDEX IF NOT EXISTS overlay_settings_owner_idx
ON overlay_settings(owner_user_id);
-- The old outbox is currently unused. Adding a nullable owner makes old rows
-- valid while ensuring any newly adopted outbox workflow can be tenant-aware.
ALTER TABLE live_session_outbox
ADD COLUMN IF NOT EXISTS owner_user_id UUID REFERENCES users(id) ON DELETE CASCADE;
-- Database-enforced tenant isolation for tables that are always accessed in a
-- known user's context. Call `set_config('app.user_id', <uuid>, true)` inside a
-- transaction before touching them. FORCE also protects against accidental
-- table-owner bypass by the runtime role.
ALTER TABLE cookiecloud_credentials ENABLE ROW LEVEL SECURITY;
ALTER TABLE cookiecloud_credentials FORCE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS cookiecloud_credentials_owner ON cookiecloud_credentials;
CREATE POLICY cookiecloud_credentials_owner ON cookiecloud_credentials
USING (user_id = NULLIF(current_setting('app.user_id', true), '')::UUID)
WITH CHECK (user_id = NULLIF(current_setting('app.user_id', true), '')::UUID);
ALTER TABLE component_instances ENABLE ROW LEVEL SECURITY;
ALTER TABLE component_instances FORCE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS component_instances_owner ON component_instances;
CREATE POLICY component_instances_owner ON component_instances
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 live_sources ENABLE ROW LEVEL SECURITY;
ALTER TABLE live_sources FORCE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS live_sources_owner ON live_sources;
CREATE POLICY live_sources_owner ON live_sources
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 component_access_tokens ENABLE ROW LEVEL SECURITY;
ALTER TABLE component_access_tokens FORCE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS component_access_tokens_owner ON component_access_tokens;
CREATE POLICY component_access_tokens_owner ON component_access_tokens
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);
-- A presented component token is the one case where the owner is not known
-- before lookup. This narrowly scoped function crosses RLS using only a
-- full-entropy SHA-256 digest, returns no secret material, and pins search_path
-- to prevent object-shadowing attacks. Once the owner is known, all component
-- reads/writes continue in a normal tenant transaction.
CREATE OR REPLACE FUNCTION lookup_component_access_token(p_token_digest BYTEA)
RETURNS TABLE (
token_id UUID,
owner_user_id UUID,
component_instance_id UUID,
scopes TEXT[]
)
LANGUAGE plpgsql
SECURITY DEFINER
VOLATILE
SET search_path = pg_catalog, public
AS $$
DECLARE
account_id UUID;
BEGIN
-- FORCE RLS intentionally remains enabled. Enter each active account's
-- context before checking the digest instead of granting a broad bypass.
FOR account_id IN
SELECT account.id FROM public.users AS account WHERE account.status='active'
LOOP
PERFORM pg_catalog.set_config('app.user_id', account_id::TEXT, true);
RETURN QUERY
SELECT token.id, token.owner_user_id, token.component_instance_id, token.scopes
FROM public.component_access_tokens AS token
JOIN public.component_instances AS component
ON component.id = token.component_instance_id
AND component.owner_user_id = token.owner_user_id
WHERE token.owner_user_id = account_id
AND token.token_digest = p_token_digest
AND token.revoked_at IS NULL
AND (token.expires_at IS NULL OR token.expires_at > pg_catalog.now())
AND component.enabled
LIMIT 1;
IF FOUND THEN
RETURN;
END IF;
END LOOP;
END
$$;
-- Startup source enumeration is another service operation whose tenant is not
-- known in advance. This function returns only routing identifiers (never
-- CookieCloud or TOTP material) and enters each account's RLS context in turn.
CREATE OR REPLACE FUNCTION list_active_live_sources()
RETURNS TABLE (
owner_user_id UUID,
source_id UUID,
room_id TEXT
)
LANGUAGE plpgsql
SECURITY DEFINER
VOLATILE
SET search_path = pg_catalog, public
AS $$
DECLARE
account RECORD;
BEGIN
FOR account IN
SELECT users.id,users.room_id
FROM public.users
WHERE users.status='active'
ORDER BY users.created_at
LOOP
PERFORM pg_catalog.set_config('app.user_id', account.id::TEXT, true);
RETURN QUERY
SELECT account.id,source.id,account.room_id
FROM public.live_sources AS source
WHERE source.owner_user_id=account.id AND source.enabled;
END LOOP;
END
$$;
@@ -0,0 +1,12 @@
-- The bootstrap flow is the sole way to create a system administrator.
-- This database constraint closes the last concurrent-request race even if
-- two bootstrap enrollments reach their final transaction simultaneously.
CREATE UNIQUE INDEX IF NOT EXISTS users_single_system_admin
ON users ((role))
WHERE role = 'system_admin';
-- PostgreSQL grants EXECUTE on new functions to PUBLIC by default. These two
-- SECURITY DEFINER helpers deliberately cross tenant discovery boundaries and
-- must only be callable by their owner (the current runtime/migration role).
REVOKE EXECUTE ON FUNCTION public.lookup_component_access_token(BYTEA) FROM PUBLIC;
REVOKE EXECUTE ON FUNCTION public.list_active_live_sources() FROM PUBLIC;
+280
View File
@@ -0,0 +1,280 @@
use std::{sync::Arc, time::Duration};
use async_trait::async_trait;
use tokio::sync::{Semaphore, mpsc};
use tracing::{error, info, warn};
use uuid::Uuid;
use crate::{
auth::AuthService,
components::ComponentRegistry,
config::Config,
credentials::{CookieCloudCredentials, CookieCloudSecrets, fetch_bilibili_cookie},
db::{ActiveTenant, Db},
domain::LiveEvent,
live::{
LiveProvider, SourceContext,
bilibili::BilibiliProvider,
supervisor::{ProviderFactory, SourceSupervisor},
},
rate_limit::AuthRateLimiter,
realtime::{EventHub, InMemoryComponentStore, SourceEventRouter},
repository::TenantRepository,
};
#[derive(Clone)]
pub struct AppState {
pub config: Arc<Config>,
pub db: Db,
pub auth: AuthService,
pub repository: TenantRepository,
pub registry: ComponentRegistry,
pub hub: EventHub,
pub supervisor: SourceSupervisor,
pub login_limiter: AuthRateLimiter,
pub enrollment_limiter: AuthRateLimiter,
pub component_socket_slots: Arc<Semaphore>,
pub http: reqwest::Client,
}
impl AppState {
pub async fn build(config: Config) -> Result<Self, String> {
let config = Arc::new(config);
let db = Db::connect(&config.database_url, 16).map_err(|error| error.to_string())?;
migrate(&db).await?;
let auth = AuthService::new(
db.clone(),
config.data_encryption_key,
config.totp_issuer.clone(),
Duration::from_secs((config.session_ttl_hours as u64) * 3_600),
Duration::from_secs((config.registration_ttl_minutes as u64) * 60),
)
.map_err(|error| error.to_string())?;
let registry = ComponentRegistry::with_builtin_components();
let component_cache = Arc::new(InMemoryComponentStore::default());
let hub = EventHub::new(512);
let router = SourceEventRouter::new(registry.clone(), component_cache.clone(), hub.clone());
let repository =
TenantRepository::new(db.clone(), registry.clone(), component_cache.clone());
repository
.hydrate_all()
.await
.map_err(|error| error.to_string())?;
let http = reqwest::Client::builder()
.timeout(Duration::from_secs(20))
.redirect(reqwest::redirect::Policy::none())
.build()
.map_err(|error| error.to_string())?;
let provider_factory = Arc::new(BilibiliProviderFactory {
auth: auth.clone(),
config: config.clone(),
http: http.clone(),
});
let (source_events, mut source_event_rx) = mpsc::channel::<Arc<LiveEvent>>(512);
let supervisor = SourceSupervisor::new(provider_factory, source_events);
let event_router = router.clone();
tokio::spawn(async move {
while let Some(event) = source_event_rx.recv().await {
match event_router.route(event).await {
Ok(report) => {
if !report.failures.is_empty() {
warn!(
failures = report.failures.len(),
"component routing completed with failures"
);
}
}
Err(error) => error!(%error, "source event routing failed"),
}
}
});
let state = Self {
config,
db,
auth,
repository,
registry,
hub,
supervisor,
login_limiter: AuthRateLimiter::default(),
enrollment_limiter: AuthRateLimiter::new(
12,
Duration::from_secs(5 * 60),
Duration::from_secs(10 * 60),
),
component_socket_slots: Arc::new(Semaphore::new(128)),
http,
};
state.start_all_sources().await?;
Ok(state)
}
pub async fn start_all_sources(&self) -> Result<(), String> {
for tenant in self
.db
.list_active_tenants()
.await
.map_err(|error| error.to_string())?
{
if let Err(error) = self.start_source(tenant.clone()).await {
warn!(user_id = %tenant.user_id, room_id = %tenant.room_id, %error, "live source is not started");
}
}
Ok(())
}
pub async fn start_source(&self, tenant: ActiveTenant) -> Result<(), String> {
self.repository
.hydrate_tenant(tenant.user_id)
.await
.map_err(|error| error.to_string())?;
self.supervisor
.start(SourceContext {
owner_id: tenant.user_id,
source_id: tenant.source_id,
room_id: tenant.room_id,
})
.await
}
pub async fn restart_user_source(&self, owner_id: Uuid) -> Result<(), String> {
let tenant = self
.db
.list_active_tenants()
.await
.map_err(|error| error.to_string())?
.into_iter()
.find(|tenant| tenant.user_id == owner_id)
.ok_or_else(|| "live source was not found".to_string())?;
self.start_source(tenant).await
}
pub async fn import_legacy_owner(&self, owner_id: Uuid, room_id: &str) -> Result<Uuid, String> {
let credentials = CookieCloudCredentials {
host: self.config.legacy_cookiecloud_host.clone(),
secrets: CookieCloudSecrets {
key: self.config.legacy_cookiecloud_key.clone(),
password: self.config.legacy_cookiecloud_password.clone(),
},
};
self.auth
.set_cookiecloud_credentials(owner_id, &credentials)
.await
.map_err(|error| error.to_string())?;
let fallback = serde_json::to_value(&self.config.legacy_overlay_defaults)
.map_err(|error| error.to_string())?;
let settings = self
.repository
.legacy_overlay_settings(room_id, fallback)
.await
.map_err(|error| error.to_string())?;
let component_id = self
.auth
.import_legacy_overlay_settings(owner_id, settings)
.await
.map_err(|error| error.to_string())?;
if !self.config.legacy_obs_access_token.trim().is_empty() {
self.auth
.import_component_access_token(
owner_id,
component_id,
"Legacy OBS browser source",
&["events:subscribe".into()],
self.config.legacy_obs_access_token.clone(),
)
.await
.map_err(|error| error.to_string())?;
}
self.repository
.hydrate_tenant(owner_id)
.await
.map_err(|error| error.to_string())?;
self.restart_user_source(owner_id).await?;
Ok(component_id)
}
}
struct BilibiliProviderFactory {
auth: AuthService,
config: Arc<Config>,
http: reqwest::Client,
}
#[async_trait]
impl ProviderFactory for BilibiliProviderFactory {
async fn build(&self, source: &SourceContext) -> Result<Arc<dyn LiveProvider>, String> {
let stored = self
.auth
.get_cookiecloud_credentials(source.owner_id)
.await
.map_err(|error| error.to_string())?
.ok_or_else(|| "CookieCloud credentials have not been configured".to_string())?;
self.config.allowed_cookiecloud_host(&stored.host)?;
let cookie = fetch_bilibili_cookie(&self.http, &stored).await?;
Ok(Arc::new(BilibiliProvider::new(
cookie,
self.config.gift_refresh_seconds,
self.config.gift_request_timeout_seconds,
self.config.emoticon_refresh_seconds,
self.config.emoticon_request_timeout_seconds,
)))
}
}
async fn migrate(db: &Db) -> Result<(), String> {
let mut client = db.get().await.map_err(|error| error.to_string())?;
let transaction = client
.transaction()
.await
.map_err(|error| error.to_string())?;
transaction
.batch_execute(
"CREATE TABLE IF NOT EXISTS schema_migrations (\
version INTEGER PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT now()\
)",
)
.await
.map_err(|error| error.to_string())?;
transaction
.query_one("SELECT pg_advisory_xact_lock(1280529235)", &[])
.await
.map_err(|error| error.to_string())?;
for (version, sql) in [
(1_i32, include_str!("../migrations/001_initial.sql")),
(
2_i32,
include_str!("../migrations/002_overlay_settings.sql"),
),
(3_i32, include_str!("../migrations/003_multitenancy.sql")),
(4_i32, include_str!("../migrations/004_auth_hardening.sql")),
] {
let applied = transaction
.query_one(
"SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version=$1)",
&[&version],
)
.await
.map_err(|error| error.to_string())?
.get::<_, bool>(0);
if !applied {
transaction
.batch_execute(sql)
.await
.map_err(|error| format!("migration {version} failed: {error}"))?;
transaction
.execute(
"INSERT INTO schema_migrations(version) VALUES($1)",
&[&version],
)
.await
.map_err(|error| error.to_string())?;
info!(version, "database migration applied");
}
}
transaction
.commit()
.await
.map_err(|error| error.to_string())
}
File diff suppressed because it is too large Load Diff
+488
View File
@@ -0,0 +1,488 @@
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},
overlay::OverlaySettings,
};
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,
pub 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,
source_id: Uuid,
kind: impl Into<String>,
name: impl Into<String>,
settings_version: u32,
settings: Value,
) -> Self {
Self {
id: Uuid::new_v4(),
owner_id,
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>>;
/// 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>>,
}
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()
}
}
#[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
}
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(),
},
);
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 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 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 { .. })
));
}
}
+360
View File
@@ -0,0 +1,360 @@
use std::{env, fs, net::IpAddr, path::PathBuf};
use base64::{Engine, engine::general_purpose::STANDARD};
use serde::Deserialize;
use sha2::{Digest, Sha256};
use crate::{credentials::normalize_cookiecloud_host, overlay::OverlaySettings};
/// Process-level configuration. Tenant-owned room, CookieCloud and component
/// settings are imported from the legacy sections once and then live in
/// PostgreSQL; these values are not used as global runtime state afterwards.
#[derive(Clone)]
pub struct Config {
pub port: u16,
pub bind_address: IpAddr,
pub database_url: String,
pub bootstrap_password: String,
pub legacy_room_id: String,
pub legacy_cookiecloud_host: String,
pub legacy_cookiecloud_key: String,
pub legacy_cookiecloud_password: String,
pub cookiecloud_allowed_hosts: Vec<String>,
pub legacy_obs_access_token: String,
pub legacy_overlay_defaults: OverlaySettings,
pub log_filter: String,
pub gift_refresh_seconds: u64,
pub gift_request_timeout_seconds: u64,
pub emoticon_refresh_seconds: u64,
pub emoticon_request_timeout_seconds: u64,
pub data_encryption_key: [u8; 32],
pub session_ttl_hours: i64,
pub registration_ttl_minutes: i64,
pub invitation_ttl_hours: i64,
pub totp_issuer: String,
pub secure_cookies: bool,
pub derived_encryption_key: bool,
}
#[derive(Deserialize)]
struct FileConfig {
connection: ConnectionConfig,
#[serde(default)]
server: ServerConfig,
database: DatabaseConfig,
cookiecloud: CookieCloudConfig,
admin: AdminConfig,
obs: ObsConfig,
#[serde(default)]
security: SecurityConfig,
#[serde(default)]
gifts: GiftsConfig,
#[serde(default)]
emoticons: EmoticonsConfig,
#[serde(default)]
overlay: OverlayFileConfig,
#[serde(default)]
logging: LoggingConfig,
}
#[derive(Deserialize)]
struct ConnectionConfig {
room_id: String,
}
#[derive(Deserialize, Default)]
struct ServerConfig {
port: Option<u16>,
bind_address: Option<String>,
}
#[derive(Deserialize)]
struct DatabaseConfig {
url: String,
}
#[derive(Deserialize)]
struct CookieCloudConfig {
host: String,
key: String,
password: String,
}
#[derive(Deserialize)]
struct AdminConfig {
password: String,
session_secret: String,
}
#[derive(Deserialize)]
struct ObsConfig {
access_token: String,
}
#[derive(Deserialize, Default)]
struct SecurityConfig {
data_encryption_key: Option<String>,
session_ttl_hours: Option<i64>,
registration_ttl_minutes: Option<i64>,
invitation_ttl_hours: Option<i64>,
totp_issuer: Option<String>,
secure_cookies: Option<bool>,
cookiecloud_allowed_hosts: Option<Vec<String>>,
}
#[derive(Deserialize, Default)]
struct GiftsConfig {
refresh_interval_seconds: Option<u64>,
request_timeout_seconds: Option<u64>,
}
#[derive(Deserialize, Default)]
struct EmoticonsConfig {
refresh_interval_seconds: Option<u64>,
request_timeout_seconds: Option<u64>,
}
#[derive(Deserialize, Default)]
struct OverlayFileConfig {
font_scale: Option<u16>,
max_visible: Option<u8>,
collapse_after_seconds: Option<u16>,
unfold_duration_ms: Option<u16>,
motion_intensity: Option<u8>,
particle_count: Option<u8>,
particle_speed: Option<u16>,
low_performance_mode: Option<bool>,
high_value_threshold: Option<i64>,
featured_value_threshold: Option<i64>,
#[serde(default)]
events: OverlayEventsConfig,
}
#[derive(Deserialize, Default)]
struct OverlayEventsConfig {
danmaku: Option<bool>,
enter: Option<bool>,
gift: Option<bool>,
superchat: Option<bool>,
guard: Option<bool>,
like: Option<bool>,
share: Option<bool>,
}
#[derive(Deserialize, Default)]
struct LoggingConfig {
filter: Option<String>,
}
impl Config {
pub fn load() -> Result<Self, String> {
let path = config_path()?;
let source = fs::read_to_string(&path)
.map_err(|error| format!("Cannot read configuration {}: {error}", path.display()))?;
let file: FileConfig = toml::from_str(&source)
.map_err(|error| format!("Invalid TOML in {}: {error}", path.display()))?;
validate_non_empty("connection.room_id", &file.connection.room_id)?;
validate_non_empty("database.url", &file.database.url)?;
validate_non_empty("admin.password", &file.admin.password)?;
validate_non_empty("admin.session_secret", &file.admin.session_secret)?;
let legacy_cookiecloud_host = normalize_cookiecloud_host(&file.cookiecloud.host)?;
let cookiecloud_allowed_hosts = file
.security
.cookiecloud_allowed_hosts
.unwrap_or_else(|| vec![legacy_cookiecloud_host.clone()])
.into_iter()
.map(|host| normalize_cookiecloud_host(&host))
.collect::<Result<Vec<_>, _>>()?;
if cookiecloud_allowed_hosts.is_empty() {
return Err("security.cookiecloud_allowed_hosts must not be empty".into());
}
if !cookiecloud_allowed_hosts.contains(&legacy_cookiecloud_host) {
return Err(
"security.cookiecloud_allowed_hosts must include cookiecloud.host for legacy import"
.into(),
);
}
let bind_address = file
.server
.bind_address
.as_deref()
.unwrap_or("127.0.0.1")
.parse::<IpAddr>()
.map_err(|_| "server.bind_address must be an IPv4 or IPv6 address".to_string())?;
let (data_encryption_key, derived_encryption_key) =
match file.security.data_encryption_key.as_deref() {
Some(value) if !value.trim().is_empty() => (decode_key(value)?, false),
_ => (
derive_key(
&file.admin.session_secret,
b"lxc-streamutils/data-encryption/v1",
),
true,
),
};
Ok(Self {
port: file.server.port.unwrap_or(9719),
bind_address,
database_url: file.database.url,
bootstrap_password: file.admin.password,
legacy_room_id: file.connection.room_id,
legacy_cookiecloud_host,
legacy_cookiecloud_key: file.cookiecloud.key,
legacy_cookiecloud_password: file.cookiecloud.password,
cookiecloud_allowed_hosts,
legacy_obs_access_token: file.obs.access_token,
legacy_overlay_defaults: overlay_defaults(file.overlay),
log_filter: file.logging.filter.unwrap_or_else(|| {
"lxc_stream_server=info,blivedm=warn,tokio_postgres=warn".into()
}),
gift_refresh_seconds: file
.gifts
.refresh_interval_seconds
.unwrap_or(600)
.clamp(60, 86_400),
gift_request_timeout_seconds: file
.gifts
.request_timeout_seconds
.unwrap_or(10)
.clamp(2, 120),
emoticon_refresh_seconds: file
.emoticons
.refresh_interval_seconds
.unwrap_or(600)
.clamp(60, 86_400),
emoticon_request_timeout_seconds: file
.emoticons
.request_timeout_seconds
.unwrap_or(10)
.clamp(2, 120),
data_encryption_key,
session_ttl_hours: file.security.session_ttl_hours.unwrap_or(12).clamp(1, 720),
registration_ttl_minutes: file
.security
.registration_ttl_minutes
.unwrap_or(15)
.clamp(5, 120),
invitation_ttl_hours: file
.security
.invitation_ttl_hours
.unwrap_or(72)
.clamp(1, 8_760),
totp_issuer: file
.security
.totp_issuer
.unwrap_or_else(|| "danmaku.luoxingci.com".into()),
secure_cookies: file.security.secure_cookies.unwrap_or(true),
derived_encryption_key,
})
}
pub fn allowed_cookiecloud_host(&self, value: &str) -> Result<String, String> {
let normalized = normalize_cookiecloud_host(value)?;
if self.cookiecloud_allowed_hosts.contains(&normalized) {
Ok(normalized)
} else {
Err("CookieCloud host is not approved by this deployment".into())
}
}
pub fn default_cookiecloud_host(&self) -> &str {
self.cookiecloud_allowed_hosts
.first()
.expect("configuration requires at least one CookieCloud host")
}
}
fn config_path() -> Result<PathBuf, String> {
let mut args = env::args_os().skip(1);
let mut path = PathBuf::from("config.toml");
while let Some(argument) = args.next() {
if argument == "--config" {
path = PathBuf::from(args.next().ok_or("--config requires a TOML path")?);
} else {
return Err(format!(
"Unknown argument: {argument:?}; use --config <path>"
));
}
}
Ok(path)
}
fn validate_non_empty(name: &str, value: &str) -> Result<(), String> {
if value.trim().is_empty() {
Err(format!("{name} must not be empty"))
} else {
Ok(())
}
}
fn decode_key(value: &str) -> Result<[u8; 32], String> {
let value = value.trim();
let bytes = STANDARD
.decode(value)
.or_else(|_| base64::engine::general_purpose::URL_SAFE_NO_PAD.decode(value))
.map_err(|_| "security.data_encryption_key must be base64-encoded".to_string())?;
bytes
.try_into()
.map_err(|_| "security.data_encryption_key must decode to exactly 32 bytes".to_string())
}
fn derive_key(secret: &str, domain: &[u8]) -> [u8; 32] {
let mut digest = Sha256::new();
digest.update(domain);
digest.update([0]);
digest.update(secret.as_bytes());
digest.finalize().into()
}
fn overlay_defaults(file: OverlayFileConfig) -> OverlaySettings {
let default = OverlaySettings::default();
OverlaySettings {
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),
show_gift: file.events.gift.unwrap_or(default.show_gift),
show_superchat: file.events.superchat.unwrap_or(default.show_superchat),
show_guard: file.events.guard.unwrap_or(default.show_guard),
show_like: file.events.like.unwrap_or(default.show_like),
show_share: file.events.share.unwrap_or(default.show_share),
max_visible: file.max_visible.unwrap_or(default.max_visible),
collapse_after_seconds: file
.collapse_after_seconds
.unwrap_or(default.collapse_after_seconds),
unfold_duration_ms: file
.unfold_duration_ms
.unwrap_or(default.unfold_duration_ms),
motion_intensity: file.motion_intensity.unwrap_or(default.motion_intensity),
particle_count: file.particle_count.unwrap_or(default.particle_count),
particle_speed: file.particle_speed.unwrap_or(default.particle_speed),
low_performance_mode: file
.low_performance_mode
.unwrap_or(default.low_performance_mode),
high_value_threshold: file
.high_value_threshold
.unwrap_or(default.high_value_threshold),
featured_value_threshold: file
.featured_value_threshold
.unwrap_or(default.featured_value_threshold),
}
.sanitize()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn accepts_exactly_32_byte_base64_key() {
let value = STANDARD.encode([7_u8; 32]);
assert_eq!(decode_key(&value).unwrap(), [7_u8; 32]);
assert!(decode_key("too-short").is_err());
}
#[test]
fn domain_separates_derived_keys() {
assert_ne!(derive_key("secret", b"a"), derive_key("secret", b"b"));
}
}
+193
View File
@@ -0,0 +1,193 @@
use reqwest::{
Url,
header::{REFERER, USER_AGENT},
};
use serde::{Deserialize, Serialize};
use serde_json::Value;
/// The secret portion is encrypted as one JSON document in PostgreSQL. The
/// host remains queryable so status pages can show where a tenant connects
/// without ever returning its key or password.
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CookieCloudSecrets {
pub key: String,
pub password: String,
}
#[derive(Clone, Debug)]
pub struct CookieCloudCredentials {
pub host: String,
pub secrets: CookieCloudSecrets,
}
impl CookieCloudCredentials {
pub fn validate(&self) -> Result<(), String> {
normalize_cookiecloud_host(&self.host)?;
if self.secrets.key.trim().is_empty() || self.secrets.password.is_empty() {
return Err("CookieCloud key and password are required".into());
}
if self.secrets.key.len() > 256 || self.secrets.password.len() > 4_096 {
return Err("CookieCloud credentials are too long".into());
}
Ok(())
}
}
/// Canonicalize a deployment-approved CookieCloud base URL. Tenant input is
/// matched against these exact canonical values before any network request.
pub fn normalize_cookiecloud_host(value: &str) -> Result<String, String> {
if value.len() > 2_048 {
return Err("CookieCloud host is too long".into());
}
let mut url = Url::parse(value.trim()).map_err(|_| "CookieCloud host is not a valid URL")?;
if !matches!(url.scheme(), "http" | "https") || url.host_str().is_none() {
return Err("CookieCloud host must be an absolute http:// or https:// URL".into());
}
if !url.username().is_empty() || url.password().is_some() {
return Err("CookieCloud host must not contain URL credentials".into());
}
if url.query().is_some() || url.fragment().is_some() {
return Err("CookieCloud host must not contain a query or fragment".into());
}
let normalized_path = url.path().trim_end_matches('/').to_owned();
url.set_path(&normalized_path);
Ok(url.as_str().trim_end_matches('/').to_owned())
}
pub fn cookiecloud_endpoint(credentials: &CookieCloudCredentials) -> Result<Url, String> {
credentials.validate()?;
let host = normalize_cookiecloud_host(&credentials.host)?;
let mut endpoint = Url::parse(&host).map_err(|_| "CookieCloud host is not a valid URL")?;
endpoint
.path_segments_mut()
.map_err(|_| "CookieCloud host cannot be used as a base URL")?
.pop_if_empty()
.push("get")
.push(credentials.secrets.key.trim());
Ok(endpoint)
}
/// Resolve only Bilibili cookies from a CookieCloud sync bucket. The returned
/// value is kept in the listener task and is never included in status/errors.
pub async fn fetch_bilibili_cookie(
client: &reqwest::Client,
credentials: &CookieCloudCredentials,
) -> Result<String, String> {
let endpoint = cookiecloud_endpoint(credentials)?;
let response = client
.post(endpoint)
.form(&[("password", credentials.secrets.password.as_str())])
.header(REFERER, "https://live.bilibili.com/")
.header(USER_AGENT, "Mozilla/5.0 lxc-streamutils/2.0")
.send()
.await
// reqwest errors can include the full URL, whose path contains the
// CookieCloud key. Keep that bearer-like value out of logs/responses.
.map_err(|_| "CookieCloud network request failed".to_string())?;
if !response.status().is_success() {
return Err(format!("CookieCloud returned HTTP {}", response.status()));
}
let value: Value = response
.json()
.await
.map_err(|_| "CookieCloud response was not valid JSON".to_string())?;
cookie_header(&value)
}
fn cookie_header(value: &Value) -> Result<String, String> {
let mut cookies = Vec::new();
if let Some(domains) = value.get("cookie_data").and_then(Value::as_object) {
for (domain, stored) in domains {
if !domain.contains("bilibili.com") {
continue;
}
let entries: Vec<&Value> = if let Some(array) = stored.as_array() {
array.iter().collect()
} else {
stored
.as_object()
.map(|values| values.values().collect())
.unwrap_or_default()
};
for item in entries {
let Some(name) = item.get("name").and_then(Value::as_str) else {
continue;
};
let Some(value) = item.get("value").and_then(Value::as_str) else {
continue;
};
// Cookie names cannot contain these delimiters. Ignore malformed
// upstream entries instead of allowing header injection.
if name.is_empty()
|| name.contains([';', '=', '\r', '\n'])
|| value.contains([';', '\r', '\n'])
{
continue;
}
cookies.push(format!("{name}={value}"));
}
}
}
if cookies.iter().any(|cookie| cookie.starts_with("SESSDATA=")) {
Ok(cookies.join("; "))
} else {
Err("CookieCloud does not contain a Bilibili SESSDATA cookie".into())
}
}
#[cfg(test)]
mod tests {
use serde_json::json;
use super::*;
#[test]
fn extracts_only_bilibili_cookie_data() {
let value = json!({"cookie_data": {
".bilibili.com": [
{"name":"SESSDATA","value":"session"},
{"name":"bili_jct","value":"csrf"}
],
"example.com": [{"name":"secret","value":"must-not-leak"}]
}});
let result = cookie_header(&value).unwrap();
assert!(result.contains("SESSDATA=session"));
assert!(result.contains("bili_jct=csrf"));
assert!(!result.contains("must-not-leak"));
}
#[test]
fn requires_authenticated_cookie() {
let value = json!({"cookie_data": {"bilibili.com": [
{"name":"buvid3","value":"anonymous"}
]}});
assert!(cookie_header(&value).is_err());
}
#[test]
fn cookiecloud_key_is_encoded_as_one_path_segment() {
let credentials = CookieCloudCredentials {
host: "https://cookies.example.test/base/".into(),
secrets: CookieCloudSecrets {
key: "bucket/../../admin?x=1".into(),
password: "secret".into(),
},
};
let endpoint = cookiecloud_endpoint(&credentials).unwrap();
assert_eq!(
endpoint.as_str(),
"https://cookies.example.test/base/get/bucket%2F..%2F..%2Fadmin%3Fx=1"
);
}
#[test]
fn canonical_host_rejects_embedded_credentials_and_queries() {
assert!(normalize_cookiecloud_host("https://user:pass@example.test").is_err());
assert!(normalize_cookiecloud_host("https://example.test/?next=internal").is_err());
assert_eq!(
normalize_cookiecloud_host("https://example.test/base/").unwrap(),
"https://example.test/base"
);
}
}
+175
View File
@@ -0,0 +1,175 @@
use std::{fmt, str::FromStr};
use deadpool_postgres::{Manager, ManagerConfig, Object, Pool, RecyclingMethod, Runtime};
use serde_json::Value;
use tokio_postgres::{Config as PostgresConfig, NoTls, Transaction};
use uuid::Uuid;
/// Cloneable database handle intended to live directly in Axum's `AppState`.
///
/// HTTP handlers should acquire a pooled connection through [`Db::get`]. Any
/// query against an RLS-protected tenant table must run in a transaction after
/// calling [`Db::set_tenant`]. `SET LOCAL` is important: a session-level setting
/// could otherwise leak an identity when the connection returns to the pool.
#[derive(Clone)]
pub struct Db {
pool: Pool,
}
impl Db {
pub fn from_pool(pool: Pool) -> Self {
Self { pool }
}
pub fn pool(&self) -> &Pool {
&self.pool
}
pub fn connect(database_url: &str, max_size: usize) -> Result<Self, DbError> {
let config = PostgresConfig::from_str(database_url)
.map_err(|error| DbError::Configuration(error.to_string()))?;
let manager = Manager::from_config(
config,
NoTls,
ManagerConfig {
recycling_method: RecyclingMethod::Fast,
},
);
let pool = Pool::builder(manager)
.runtime(Runtime::Tokio1)
.max_size(max_size.max(1))
.build()
.map_err(|error| DbError::Configuration(error.to_string()))?;
Ok(Self { pool })
}
pub async fn get(&self) -> Result<Object, DbError> {
self.pool
.get()
.await
.map_err(|error| DbError::Pool(error.to_string()))
}
/// Applies the additive multi-tenant migration. The legacy migrations must
/// already have run because this migration deliberately links their tables.
pub async fn migrate_multitenancy(&self) -> Result<(), DbError> {
let client = self.get().await?;
client
.batch_execute(include_str!("../migrations/003_multitenancy.sql"))
.await
.map_err(DbError::Postgres)
}
pub async fn set_tenant(transaction: &Transaction<'_>, user_id: Uuid) -> Result<(), DbError> {
transaction
.query_one(
"SELECT set_config('app.user_id', $1, true)",
&[&user_id.to_string()],
)
.await
.map_err(DbError::Postgres)?;
Ok(())
}
/// Minimal, non-secret account inventory for the source supervisor at
/// startup. Component rows are deliberately loaded in a second,
/// tenant-scoped query so RLS remains effective.
pub async fn list_active_tenants(&self) -> Result<Vec<ActiveTenant>, DbError> {
let client = self.get().await?;
let rows = client
.query(
"SELECT owner_user_id,room_id,source_id FROM list_active_live_sources()",
&[],
)
.await?;
Ok(rows
.into_iter()
.map(|row| ActiveTenant {
user_id: row.get(0),
room_id: row.get(1),
source_id: row.get(2),
})
.collect())
}
pub async fn list_tenant_components(
&self,
user_id: Uuid,
) -> Result<Vec<ComponentRecord>, DbError> {
let mut client = self.get().await?;
let transaction = client.transaction().await?;
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",
&[&user_id],
)
.await?;
transaction.commit().await?;
Ok(rows
.into_iter()
.map(|row| ComponentRecord {
id: row.get(0),
owner_user_id: user_id,
source_id: row.get(1),
kind: row.get(2),
name: row.get(3),
settings: row.get(4),
settings_version: row.get(5),
enabled: row.get(6),
})
.collect())
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ActiveTenant {
pub user_id: Uuid,
pub room_id: String,
pub source_id: Uuid,
}
#[derive(Clone, Debug, PartialEq)]
pub struct ComponentRecord {
pub id: Uuid,
pub owner_user_id: Uuid,
pub source_id: Uuid,
pub kind: String,
pub name: String,
pub settings: Value,
pub settings_version: i32,
pub enabled: bool,
}
#[derive(Debug)]
pub enum DbError {
Configuration(String),
Pool(String),
Postgres(tokio_postgres::Error),
}
impl fmt::Display for DbError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Configuration(message) => write!(formatter, "database configuration: {message}"),
Self::Pool(message) => write!(formatter, "database pool: {message}"),
Self::Postgres(error) => write!(formatter, "database query: {error}"),
}
}
}
impl std::error::Error for DbError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Postgres(error) => Some(error),
_ => None,
}
}
}
impl From<tokio_postgres::Error> for DbError {
fn from(error: tokio_postgres::Error) -> Self {
Self::Postgres(error)
}
}
+341
View File
@@ -0,0 +1,341 @@
use serde::{Deserialize, Serialize};
use serde_json::Value;
use uuid::Uuid;
/// Version of the component-facing WebSocket envelope.
pub const COMPONENT_PROTOCOL_VERSION: u16 = 1;
/// Stable event categories used by component subscriptions.
///
/// Keep these independent from Bilibili command names: another live provider can
/// normalize its payloads into the same domain events later.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum LiveEventKind {
Danmaku,
Enter,
Gift,
GiftCombo,
SuperChat,
GuardPurchase,
Like,
Share,
Unknown,
}
impl LiveEventKind {
pub const fn wire_name(self) -> &'static str {
match self {
Self::Danmaku => "live.danmaku",
Self::Enter => "live.enter",
Self::Gift => "live.gift",
Self::GiftCombo => "live.gift.combo",
Self::SuperChat => "live.superchat",
Self::GuardPurchase => "live.guard.buy",
Self::Like => "live.like",
Self::Share => "live.share",
Self::Unknown => "live.unknown",
}
}
}
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct PlatformViewer {
pub uid: String,
pub name: String,
}
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(
tag = "type",
rename_all = "lowercase",
rename_all_fields = "camelCase"
)]
pub enum DanmakuSegment {
Text {
text: String,
},
Emoticon {
text: String,
unique: Option<String>,
url: String,
width: Option<u32>,
height: Option<u32>,
is_dynamic: bool,
standalone: bool,
},
}
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct DanmakuEvent {
pub viewer: PlatformViewer,
pub text: String,
#[serde(default)]
pub segments: Vec<DanmakuSegment>,
}
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EnterEvent {
pub viewer: PlatformViewer,
}
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct GiftDetails {
pub id: Option<i64>,
pub name: String,
pub coin_type: String,
pub unit_price: i64,
pub total_price: i64,
pub price_cny: f64,
pub image_url: Option<String>,
pub animation_url: Option<String>,
pub effect_type: Option<String>,
pub stay_time: Option<i64>,
}
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct GiftEvent {
pub viewer: PlatformViewer,
pub gift: GiftDetails,
pub quantity: i32,
pub source_event_id: String,
}
/// A visual combo update. Durable business handlers should normally consume
/// `Gift`, not both `Gift` and `GiftCombo`, to avoid counting the same gift twice.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct GiftComboEvent {
pub viewer: PlatformViewer,
pub gift: GiftDetails,
pub quantity: i32,
pub combo_id: String,
}
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct SuperChatEvent {
pub viewer: PlatformViewer,
pub message: String,
pub price: i64,
pub source_event_id: String,
}
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct GuardPurchaseEvent {
pub viewer: PlatformViewer,
pub guard_name: String,
pub quantity: i32,
pub price: i64,
}
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ViewerInteractionEvent {
pub viewer: PlatformViewer,
}
/// Escape hatch for provider commands that have not been normalized yet.
/// Providers should put only bounded, sanitized metadata here, never credentials
/// or an unbounded raw packet.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct UnknownLiveEvent {
pub command: String,
#[serde(default)]
pub metadata: Value,
}
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(tag = "kind", content = "data", rename_all = "snake_case")]
pub enum LiveEventPayload {
Danmaku(DanmakuEvent),
Enter(EnterEvent),
Gift(GiftEvent),
GiftCombo(GiftComboEvent),
SuperChat(SuperChatEvent),
GuardPurchase(GuardPurchaseEvent),
Like(ViewerInteractionEvent),
Share(ViewerInteractionEvent),
Unknown(UnknownLiveEvent),
}
impl LiveEventPayload {
pub const fn kind(&self) -> LiveEventKind {
match self {
Self::Danmaku(_) => LiveEventKind::Danmaku,
Self::Enter(_) => LiveEventKind::Enter,
Self::Gift(_) => LiveEventKind::Gift,
Self::GiftCombo(_) => LiveEventKind::GiftCombo,
Self::SuperChat(_) => LiveEventKind::SuperChat,
Self::GuardPurchase(_) => LiveEventKind::GuardPurchase,
Self::Like(_) => LiveEventKind::Like,
Self::Share(_) => LiveEventKind::Share,
Self::Unknown(_) => LiveEventKind::Unknown,
}
}
/// Serialize only the variant body. The discriminant is carried by the
/// envelope's `type` field for compatibility with the browser SDK.
pub fn to_wire_payload(&self) -> Result<Value, serde_json::Error> {
match self {
Self::Danmaku(value) => serde_json::to_value(value),
Self::Enter(value) => serde_json::to_value(value),
Self::Gift(value) => serde_json::to_value(value),
Self::GiftCombo(value) => serde_json::to_value(value),
Self::SuperChat(value) => serde_json::to_value(value),
Self::GuardPurchase(value) => serde_json::to_value(value),
Self::Like(value) => serde_json::to_value(value),
Self::Share(value) => serde_json::to_value(value),
Self::Unknown(value) => serde_json::to_value(value),
}
}
}
/// Canonical provider-independent event.
///
/// `owner_id` is an internal routing boundary and is deliberately omitted when
/// serializing. It must always be supplied by trusted source configuration, not
/// by provider input or an HTTP request body.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct LiveEvent {
#[serde(skip_serializing)]
pub owner_id: Uuid,
pub source_id: Uuid,
pub provider: String,
pub room_id: String,
pub id: Uuid,
pub occurred_at_ms: i64,
pub received_at_ms: i64,
#[serde(default)]
pub simulated: bool,
pub payload: LiveEventPayload,
}
impl LiveEvent {
pub fn new(
owner_id: Uuid,
source_id: Uuid,
provider: impl Into<String>,
room_id: impl Into<String>,
payload: LiveEventPayload,
) -> Self {
let now = chrono::Utc::now().timestamp_millis();
Self {
owner_id,
source_id,
provider: provider.into(),
room_id: room_id.into(),
id: Uuid::new_v4(),
occurred_at_ms: now,
received_at_ms: now,
simulated: false,
payload,
}
}
pub const fn kind(&self) -> LiveEventKind {
self.payload.kind()
}
pub const fn wire_type(&self) -> &'static str {
self.kind().wire_name()
}
}
/// Component-scoped wire event produced by a passive projection.
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ComponentMessage {
#[serde(skip_serializing)]
pub owner_id: Uuid,
pub component_id: Uuid,
pub source_id: Uuid,
pub version: u16,
pub id: Uuid,
/// RFC3339 timestamp retained for compatibility with the existing browser
/// protocol; canonical events use milliseconds internally.
pub occurred_at: String,
pub room_id: String,
#[serde(rename = "type")]
pub event_type: String,
pub payload: Value,
}
impl ComponentMessage {
pub fn from_live_event(
component_id: Uuid,
event: &LiveEvent,
) -> Result<Self, serde_json::Error> {
Ok(Self {
owner_id: event.owner_id,
component_id,
source_id: event.source_id,
version: COMPONENT_PROTOCOL_VERSION,
id: event.id,
occurred_at: chrono::DateTime::<chrono::Utc>::from_timestamp_millis(
event.occurred_at_ms,
)
.map(|timestamp| timestamp.to_rfc3339_opts(chrono::SecondsFormat::Millis, true))
.unwrap_or_else(|| "1970-01-01T00:00:00.000Z".to_owned()),
room_id: event.room_id.clone(),
event_type: event.wire_type().to_owned(),
payload: event.payload.to_wire_payload()?,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
fn viewer() -> PlatformViewer {
PlatformViewer {
uid: "42".into(),
name: "观众".into(),
}
}
#[test]
fn typed_event_projects_to_legacy_compatible_wire_shape() {
let owner_id = Uuid::new_v4();
let source_id = Uuid::new_v4();
let component_id = Uuid::new_v4();
let event = LiveEvent::new(
owner_id,
source_id,
"bilibili",
"123",
LiveEventPayload::Danmaku(DanmakuEvent {
viewer: viewer(),
text: "晚上好".into(),
segments: vec![DanmakuSegment::Text {
text: "晚上好".into(),
}],
}),
);
let projected = ComponentMessage::from_live_event(component_id, &event).unwrap();
assert_eq!(projected.event_type, "live.danmaku");
assert_eq!(projected.payload["viewer"]["uid"], "42");
assert_eq!(projected.payload["text"], "晚上好");
assert_eq!(projected.component_id, component_id);
let serialized = serde_json::to_value(projected).unwrap();
assert!(serialized.get("ownerId").is_none());
}
#[test]
fn gift_and_combo_have_distinct_subscription_kinds() {
assert_eq!(LiveEventKind::Gift.wire_name(), "live.gift");
assert_eq!(LiveEventKind::GiftCombo.wire_name(), "live.gift.combo");
assert_ne!(LiveEventKind::Gift, LiveEventKind::GiftCombo);
}
}
File diff suppressed because it is too large Load Diff
+20
View File
@@ -0,0 +1,20 @@
//! Reusable application core for the multi-tenant livestream component host.
//!
//! The binary is intentionally only a configuration/bootstrap shell. Providers,
//! authentication, tenant repositories, typed events and component runtimes are
//! exported here so future gift, song-request and overlay components can be
//! developed and tested without growing `main.rs` back into a monolith.
pub mod app;
pub mod auth;
pub mod components;
pub mod config;
pub mod credentials;
pub mod db;
pub mod domain;
pub mod http_api;
pub mod live;
pub mod overlay;
pub mod rate_limit;
pub mod realtime;
pub mod repository;
+756
View File
@@ -0,0 +1,756 @@
use std::{sync::Arc, time::Duration};
use async_trait::async_trait;
use blivedm::client::{models::BiliMessage, websocket::BiliLiveClient};
use futures_channel::mpsc as futures_mpsc;
use serde_json::{Value, json};
use tokio::sync::{mpsc, watch};
use tokio_util::sync::CancellationToken;
use tracing::{info, warn};
use uuid::Uuid;
use crate::{
domain::{
DanmakuEvent, DanmakuSegment, EnterEvent, GiftComboEvent, GiftDetails, GiftEvent,
GuardPurchaseEvent, LiveEvent, LiveEventPayload, PlatformViewer, SuperChatEvent,
UnknownLiveEvent, ViewerInteractionEvent,
},
live::{LiveProvider, SourceContext, SourceStatus},
overlay::{EmoticonCatalog, EmoticonMeta, GiftCatalog, normalize_image_url},
};
#[derive(Clone)]
pub struct BilibiliProvider {
cookie: Arc<str>,
gift_catalog: GiftCatalog,
emoticon_catalog: EmoticonCatalog,
gift_refresh_seconds: u64,
gift_timeout_seconds: u64,
emoticon_refresh_seconds: u64,
emoticon_timeout_seconds: u64,
}
impl BilibiliProvider {
pub fn new(
cookie: String,
gift_refresh_seconds: u64,
gift_timeout_seconds: u64,
emoticon_refresh_seconds: u64,
emoticon_timeout_seconds: u64,
) -> Self {
Self {
cookie: cookie.into(),
gift_catalog: GiftCatalog::default(),
emoticon_catalog: EmoticonCatalog::default(),
gift_refresh_seconds,
gift_timeout_seconds,
emoticon_refresh_seconds,
emoticon_timeout_seconds,
}
}
pub async fn gift_catalog_size(&self) -> usize {
self.gift_catalog.len().await
}
pub async fn emoticon_catalog_size(&self) -> usize {
self.emoticon_catalog.len().await
}
async fn initial_catalogs(&self, room_id: &str) {
if let Err(error) = self
.gift_catalog
.refresh(room_id, self.gift_timeout_seconds)
.await
{
warn!(%error, %room_id, "initial gift catalog refresh failed");
}
if let Err(error) = self
.emoticon_catalog
.refresh(room_id, &self.cookie, self.emoticon_timeout_seconds)
.await
{
warn!(%error, %room_id, "initial emoticon catalog refresh failed");
}
}
fn spawn_catalog_refreshes(&self, room_id: String, cancel: CancellationToken) {
let gift = self.gift_catalog.clone();
let gift_interval = self.gift_refresh_seconds;
let gift_timeout = self.gift_timeout_seconds;
let gift_room = room_id.clone();
let gift_cancel = cancel.clone();
tokio::spawn(async move {
loop {
tokio::select! {
_ = gift_cancel.cancelled() => break,
_ = tokio::time::sleep(Duration::from_secs(gift_interval)) => {
match gift.refresh(&gift_room, gift_timeout).await {
Ok(count) => info!(count, room_id = %gift_room, "gift catalog refreshed"),
Err(error) => warn!(%error, room_id = %gift_room, "gift catalog refresh failed; retaining cache"),
}
}
}
}
});
let emoticons = self.emoticon_catalog.clone();
let cookie = self.cookie.clone();
let interval = self.emoticon_refresh_seconds;
let timeout = self.emoticon_timeout_seconds;
tokio::spawn(async move {
loop {
tokio::select! {
_ = cancel.cancelled() => break,
_ = tokio::time::sleep(Duration::from_secs(interval)) => {
match emoticons.refresh(&room_id, &cookie, timeout).await {
Ok(count) => info!(count, room_id = %room_id, "emoticon catalog refreshed"),
Err(error) => warn!(%error, room_id = %room_id, "emoticon catalog refresh failed; retaining cache"),
}
}
}
}
});
}
async fn enrich(&self, context: &SourceContext, raw: ProviderEvent) -> LiveEvent {
let payload = match raw {
ProviderEvent::Enter { viewer } => LiveEventPayload::Enter(EnterEvent { viewer }),
ProviderEvent::Danmaku {
viewer,
text,
emoticons,
} => {
let segments = danmaku_segments(&self.emoticon_catalog, &text, emoticons).await;
LiveEventPayload::Danmaku(DanmakuEvent {
viewer,
text,
segments,
})
}
ProviderEvent::Gift {
viewer,
name,
gift_id,
battery,
quantity,
event_id,
} => LiveEventPayload::Gift(GiftEvent {
viewer,
gift: gift_details(&self.gift_catalog, name, gift_id, battery, quantity).await,
quantity: quantity.max(1),
source_event_id: event_id,
}),
ProviderEvent::GiftCombo {
viewer,
name,
gift_id,
battery,
quantity,
combo_id,
} => LiveEventPayload::GiftCombo(GiftComboEvent {
viewer,
gift: gift_details(&self.gift_catalog, name, gift_id, battery, quantity).await,
quantity: quantity.max(1),
combo_id,
}),
ProviderEvent::SuperChat {
viewer,
message,
price,
event_id,
} => LiveEventPayload::SuperChat(SuperChatEvent {
viewer,
message,
price,
source_event_id: event_id,
}),
ProviderEvent::Guard {
viewer,
name,
quantity,
price,
} => LiveEventPayload::GuardPurchase(GuardPurchaseEvent {
viewer,
guard_name: name,
quantity,
price,
}),
ProviderEvent::Like { viewer } => {
LiveEventPayload::Like(ViewerInteractionEvent { viewer })
}
ProviderEvent::Share { viewer } => {
LiveEventPayload::Share(ViewerInteractionEvent { viewer })
}
ProviderEvent::Unknown { command } => LiveEventPayload::Unknown(UnknownLiveEvent {
command,
metadata: json!({}),
}),
};
LiveEvent::new(
context.owner_id,
context.source_id,
self.provider_name(),
context.room_id.clone(),
payload,
)
}
}
#[async_trait]
impl LiveProvider for BilibiliProvider {
fn provider_name(&self) -> &'static str {
"bilibili"
}
async fn run(
self: Arc<Self>,
context: SourceContext,
events: mpsc::Sender<Arc<LiveEvent>>,
status: watch::Sender<SourceStatus>,
cancel: CancellationToken,
) -> Result<(), String> {
self.initial_catalogs(&context.room_id).await;
self.spawn_catalog_refreshes(context.room_id.clone(), cancel.clone());
let (raw_sender, mut raw_receiver) = mpsc::channel::<ProviderEvent>(256);
let cookie = self.cookie.to_string();
let room_id = context.room_id.clone();
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(),
});
while !listener_cancel.is_cancelled() {
if let Err(error) = client.receive() {
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}"),
});
}
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(());
}
}
}
}
client.close();
Ok(())
});
loop {
tokio::select! {
_ = cancel.cancelled() => break,
value = raw_receiver.recv() => {
let Some(value) = value else { break };
let event = Arc::new(self.enrich(&context, value).await);
if events.send(event).await.is_err() { break; }
}
}
}
cancel.cancel();
listener
.await
.map_err(|error| format!("listener task failed: {error}"))??;
let _ = status.send(SourceStatus {
source_id: context.source_id,
room_id: context.room_id,
connected: false,
cookie_cloud: true,
detail: "Listener stopped".into(),
});
Ok(())
}
}
#[derive(Clone, Debug)]
struct EmoticonHint {
text: String,
unique: Option<String>,
url: Option<String>,
width: Option<u32>,
height: Option<u32>,
is_dynamic: bool,
bulge_display: bool,
standalone: bool,
}
enum ProviderEvent {
Enter {
viewer: PlatformViewer,
},
Danmaku {
viewer: PlatformViewer,
text: String,
emoticons: Vec<EmoticonHint>,
},
Gift {
viewer: PlatformViewer,
name: String,
gift_id: Option<i64>,
battery: i32,
quantity: i32,
event_id: String,
},
GiftCombo {
viewer: PlatformViewer,
name: String,
gift_id: Option<i64>,
battery: i32,
quantity: i32,
combo_id: String,
},
SuperChat {
viewer: PlatformViewer,
message: String,
price: i64,
event_id: String,
},
Guard {
viewer: PlatformViewer,
name: String,
quantity: i32,
price: i64,
},
Like {
viewer: PlatformViewer,
},
Share {
viewer: PlatformViewer,
},
Unknown {
command: String,
},
}
fn normalize(message: BiliMessage) -> Option<ProviderEvent> {
let raw = match message {
BiliMessage::Raw(value) => value,
_ => return None,
};
let command = raw.get("cmd")?.as_str()?.split(':').next()?.to_owned();
let data = raw.get("data").unwrap_or(&raw);
let viewer = |uid: &Value, name: &Value| {
Some(PlatformViewer {
uid: uid
.as_i64()
.map(|id| id.to_string())
.or_else(|| uid.as_str().map(str::to_owned))?,
name: name.as_str()?.to_owned(),
})
};
let data_viewer = |value: &Value| {
viewer(
value.get("uid")?,
value
.get("uname")
.or_else(|| value.pointer("/sender_uinfo/base/name"))
.or_else(|| value.pointer("/user_info/uname"))?,
)
};
match command.as_str() {
"DANMU_MSG" => {
let info = raw.get("info")?.as_array()?;
let text = info.get(1)?.as_str()?.to_owned();
Some(ProviderEvent::Danmaku {
viewer: viewer(info.get(2)?.get(0)?, info.get(2)?.get(1)?)?,
emoticons: parse_danmaku_emoticons(info, &text),
text,
})
}
"SEND_GIFT" => Some(ProviderEvent::Gift {
viewer: data_viewer(data)?,
name: data
.get("giftName")
.or_else(|| data.get("gift_name"))?
.as_str()?
.to_owned(),
gift_id: data
.get("giftId")
.or_else(|| data.get("gift_id"))
.and_then(Value::as_i64),
battery: data.get("price").and_then(Value::as_i64).unwrap_or(0) as i32,
quantity: data.get("num").and_then(Value::as_i64).unwrap_or(1) as i32,
event_id: data
.get("tid")
.and_then(Value::as_str)
.map(str::to_owned)
.unwrap_or_else(|| {
format!(
"{}-{}",
data.get("uid").unwrap_or(&Value::Null),
data.get("timestamp").unwrap_or(&Value::Null)
)
}),
}),
"COMBO_SEND" => Some(ProviderEvent::GiftCombo {
viewer: data_viewer(data)?,
name: data
.get("gift_name")
.or_else(|| data.get("giftName"))
.and_then(Value::as_str)
.unwrap_or("礼物")
.to_owned(),
gift_id: data
.get("gift_id")
.or_else(|| data.get("giftId"))
.and_then(Value::as_i64),
battery: data.get("price").and_then(Value::as_i64).unwrap_or(0) as i32,
quantity: data
.get("combo_num")
.or_else(|| data.get("total_num"))
.and_then(Value::as_i64)
.unwrap_or(1) as i32,
combo_id: data
.get("combo_id")
.map(Value::to_string)
.unwrap_or_default(),
}),
"INTERACT_WORD" => Some(ProviderEvent::Enter {
viewer: data_viewer(data)?,
}),
"GUARD_BUY" => Some(ProviderEvent::Guard {
viewer: data_viewer(data)?,
name: data
.get("gift_name")
.or_else(|| data.get("giftName"))
.and_then(Value::as_str)
.unwrap_or("舰长")
.to_owned(),
quantity: data.get("num").and_then(Value::as_i64).unwrap_or(1) as i32,
price: data.get("price").and_then(Value::as_i64).unwrap_or(0),
}),
"SUPER_CHAT_MESSAGE" | "SUPER_CHAT_MESSAGE_JPN" => Some(ProviderEvent::SuperChat {
viewer: data_viewer(data)?,
message: data
.get("message")
.and_then(Value::as_str)
.unwrap_or("")
.to_owned(),
price: data.get("price").and_then(Value::as_i64).unwrap_or(0),
event_id: data
.get("id")
.map(Value::to_string)
.unwrap_or_else(|| Uuid::new_v4().to_string()),
}),
"LIKE_INFO_V3_CLICK" => Some(ProviderEvent::Like {
viewer: data_viewer(data)?,
}),
"SHARE" => Some(ProviderEvent::Share {
viewer: data_viewer(data)?,
}),
_ => Some(ProviderEvent::Unknown { command }),
}
}
fn json_object(value: &Value) -> Option<Value> {
match value {
Value::Object(_) => Some(value.clone()),
Value::String(value) => serde_json::from_str::<Value>(value)
.ok()
.filter(Value::is_object),
_ => None,
}
}
fn value_as_u32(value: Option<&Value>) -> Option<u32> {
value.and_then(|value| {
value
.as_u64()
.and_then(|number| u32::try_from(number).ok())
.or_else(|| value.as_str()?.parse().ok())
})
}
fn value_is_truthy(value: Option<&Value>) -> bool {
value.is_some_and(|value| {
value.as_bool().unwrap_or(false)
|| value.as_i64().is_some_and(|number| number != 0)
|| value.as_str().is_some_and(|text| text == "1")
})
}
fn emoticon_hint(value: &Value, text: &str, standalone: bool) -> Option<EmoticonHint> {
let value = json_object(value)?;
let unique = value
.get("emoticon_unique")
.and_then(Value::as_str)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned);
let url = value
.get("url")
.and_then(Value::as_str)
.and_then(normalize_image_url);
if unique.is_none() && url.is_none() {
return None;
}
Some(EmoticonHint {
text: text.to_owned(),
unique,
url,
width: value_as_u32(value.get("width")),
height: value_as_u32(value.get("height")),
is_dynamic: value_is_truthy(value.get("is_dynamic")),
bulge_display: value_is_truthy(value.get("bulge_display")),
standalone,
})
}
fn parse_danmaku_emoticons(info: &[Value], text: &str) -> Vec<EmoticonHint> {
let Some(header) = info.first().and_then(Value::as_array) else {
return Vec::new();
};
let mut hints = Vec::new();
if let Some(direct) = header
.get(13)
.and_then(|value| emoticon_hint(value, text, true))
.or_else(|| {
header.iter().find_map(|value| {
let object = json_object(value)?;
object.get("url")?;
emoticon_hint(&object, text, true)
})
})
{
hints.push(direct);
}
let extra = header.iter().find_map(|value| {
let object = json_object(value)?;
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(unique) = extra
.get("emoticon_unique")
.and_then(Value::as_str)
.filter(|value| !value.is_empty())
{
if !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
}
#[derive(Clone)]
struct ResolvedEmoticon {
text: String,
unique: Option<String>,
url: String,
width: Option<u32>,
height: Option<u32>,
is_dynamic: bool,
standalone: bool,
}
fn emoticon_segment(emoticon: &ResolvedEmoticon) -> DanmakuSegment {
DanmakuSegment::Emoticon {
text: emoticon.text.clone(),
unique: emoticon.unique.clone(),
url: emoticon.url.clone(),
width: emoticon.width,
height: emoticon.height,
is_dynamic: emoticon.is_dynamic,
standalone: emoticon.standalone,
}
}
async fn danmaku_segments(
catalog: &EmoticonCatalog,
text: &str,
hints: Vec<EmoticonHint>,
) -> Vec<DanmakuSegment> {
let mut resolved = Vec::<ResolvedEmoticon>::new();
for hint in hints {
let fallback: Option<EmoticonMeta> = catalog.get(hint.unique.as_deref(), &hint.text).await;
let Some(url) = hint
.url
.clone()
.or_else(|| fallback.as_ref().map(|value| value.url.clone()))
else {
continue;
};
let standalone = hint.standalone
|| (hint.bulge_display && hint.text == text)
|| fallback
.as_ref()
.is_some_and(|value| value.bulge_display && hint.text == text);
let value = ResolvedEmoticon {
text: if hint.text.is_empty() {
fallback
.as_ref()
.map(|value| value.emoji.clone())
.unwrap_or_default()
} else {
hint.text
},
unique: hint
.unique
.or_else(|| fallback.as_ref().and_then(|value| value.unique.clone())),
url,
width: hint
.width
.or_else(|| fallback.as_ref().and_then(|value| value.width)),
height: hint
.height
.or_else(|| fallback.as_ref().and_then(|value| value.height)),
is_dynamic: hint.is_dynamic || fallback.as_ref().is_some_and(|value| value.is_dynamic),
standalone,
};
if !resolved
.iter()
.any(|existing| existing.text == value.text && existing.standalone == value.standalone)
{
resolved.push(value);
}
}
if let Some(emoticon) = resolved.iter().find(|value| value.standalone) {
return vec![emoticon_segment(emoticon)];
}
let mut segments = Vec::new();
let mut cursor = 0;
while cursor < text.len() {
let remaining = &text[cursor..];
let mut next: Option<(usize, usize)> = None;
for (index, emoticon) in resolved.iter().enumerate() {
if emoticon.text.is_empty() {
continue;
}
let Some(position) = remaining.find(&emoticon.text) else {
continue;
};
if next.is_none_or(|(old_index, old_position)| {
position < old_position
|| (position == old_position
&& emoticon.text.len() > resolved[old_index].text.len())
}) {
next = Some((index, position));
}
}
let Some((index, position)) = next else {
segments.push(DanmakuSegment::Text {
text: remaining.to_owned(),
});
break;
};
let absolute = cursor + position;
if absolute > cursor {
segments.push(DanmakuSegment::Text {
text: text[cursor..absolute].to_owned(),
});
}
segments.push(emoticon_segment(&resolved[index]));
cursor = absolute + resolved[index].text.len();
}
if segments.is_empty() {
segments.push(DanmakuSegment::Text {
text: text.to_owned(),
});
}
segments
}
async fn gift_details(
catalog: &GiftCatalog,
name: String,
gift_id: Option<i64>,
battery: i32,
quantity: i32,
) -> GiftDetails {
let metadata = catalog.get(gift_id, &name).await;
let unit_price = metadata
.as_ref()
.map(|gift| gift.unit_price)
.unwrap_or_else(|| i64::from(battery.max(0)));
let total_price = unit_price.saturating_mul(i64::from(quantity.max(1)));
GiftDetails {
id: metadata.as_ref().and_then(|gift| gift.id).or(gift_id),
name: metadata
.as_ref()
.map(|gift| gift.name.clone())
.unwrap_or(name),
coin_type: metadata
.as_ref()
.map(|gift| gift.coin_type.clone())
.unwrap_or_else(|| "gold".into()),
unit_price,
total_price,
price_cny: total_price as f64 / 1000.0,
image_url: metadata.as_ref().and_then(|gift| gift.image_url.clone()),
animation_url: metadata
.as_ref()
.and_then(|gift| gift.animation_url.clone()),
effect_type: metadata.as_ref().and_then(|gift| gift.effect_type.clone()),
stay_time: metadata.as_ref().and_then(|gift| gift.stay_time),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn normalizes_raw_danmaku_to_provider_event() {
let message = normalize(BiliMessage::Raw(
json!({"cmd":"DANMU_MSG:4:0:2","info":[[],"晚上好",[12345,"观众"]]}),
))
.unwrap();
match message {
ProviderEvent::Danmaku { viewer, text, .. } => {
assert_eq!(viewer.uid, "12345");
assert_eq!(viewer.name, "观众");
assert_eq!(text, "晚上好");
}
_ => panic!("expected danmaku"),
}
}
#[test]
fn unknown_events_do_not_forward_raw_payload() {
let event = normalize(BiliMessage::Raw(json!({
"cmd":"FUTURE_SECRET_EVENT",
"data":{"cookie":"must-not-cross-provider-boundary"}
})))
.unwrap();
match event {
ProviderEvent::Unknown { command } => assert_eq!(command, "FUTURE_SECRET_EVENT"),
_ => panic!("expected unknown event"),
}
}
}
+56
View File
@@ -0,0 +1,56 @@
pub mod bilibili;
pub mod supervisor;
use std::sync::Arc;
use async_trait::async_trait;
use tokio::sync::{mpsc, watch};
use tokio_util::sync::CancellationToken;
use uuid::Uuid;
use crate::domain::LiveEvent;
#[derive(Clone, Debug, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct SourceStatus {
pub source_id: Uuid,
pub room_id: String,
pub connected: bool,
pub cookie_cloud: bool,
pub detail: String,
}
impl SourceStatus {
pub fn starting(source_id: Uuid, room_id: impl Into<String>) -> Self {
Self {
source_id,
room_id: room_id.into(),
connected: false,
cookie_cloud: false,
detail: "Waiting for CookieCloud credentials".into(),
}
}
}
#[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.
pub source_id: Uuid,
pub room_id: String,
}
#[async_trait]
pub trait LiveProvider: Send + Sync {
fn provider_name(&self) -> &'static str;
async fn run(
self: Arc<Self>,
context: SourceContext,
events: mpsc::Sender<Arc<LiveEvent>>,
status: watch::Sender<SourceStatus>,
cancel: CancellationToken,
) -> Result<(), String>;
}
+128
View File
@@ -0,0 +1,128 @@
use std::{collections::HashMap, sync::Arc};
use async_trait::async_trait;
use tokio::sync::{RwLock, mpsc, watch};
use tokio_util::sync::CancellationToken;
use tracing::{error, info};
use uuid::Uuid;
use crate::{
domain::LiveEvent,
live::{LiveProvider, SourceContext, SourceStatus},
};
#[async_trait]
pub trait ProviderFactory: Send + Sync {
async fn build(&self, source: &SourceContext) -> Result<Arc<dyn LiveProvider>, String>;
}
struct RunningSource {
cancel: CancellationToken,
status: watch::Receiver<SourceStatus>,
}
/// Owns exactly one provider task per account/source. Restart always cancels
/// the prior generation before starting another, which prevents the duplicate
/// listeners produced by the legacy `/reconnect` handler.
#[derive(Clone)]
pub struct SourceSupervisor {
factory: Arc<dyn ProviderFactory>,
events: mpsc::Sender<Arc<LiveEvent>>,
running: Arc<RwLock<HashMap<Uuid, RunningSource>>>,
}
impl SourceSupervisor {
pub fn new(factory: Arc<dyn ProviderFactory>, events: mpsc::Sender<Arc<LiveEvent>>) -> Self {
Self {
factory,
events,
running: Arc::new(RwLock::new(HashMap::new())),
}
}
pub async fn start(&self, source: SourceContext) -> Result<(), String> {
self.stop(source.source_id).await;
let cancel = CancellationToken::new();
let provider = match self.factory.build(&source).await {
Ok(provider) => provider,
Err(error) => {
let (_, status) = watch::channel(SourceStatus {
source_id: source.source_id,
room_id: source.room_id.clone(),
connected: false,
cookie_cloud: false,
detail: error.clone(),
});
self.running
.write()
.await
.insert(source.source_id, RunningSource { cancel, status });
return Err(error);
}
};
let initial = SourceStatus::starting(source.source_id, source.room_id.clone());
let (status_tx, status_rx) = watch::channel(initial);
self.running.write().await.insert(
source.source_id,
RunningSource {
cancel: cancel.clone(),
status: status_rx,
},
);
let events = self.events.clone();
let source_id = source.source_id;
let terminal_room_id = source.room_id.clone();
let terminal_status = status_tx.clone();
tokio::spawn(async move {
info!(%source_id, room_id = %source.room_id, provider = provider.provider_name(), "starting live source");
if let Err(error) = provider.run(source, events, status_tx, cancel).await {
error!(%source_id, %error, "live source stopped with error");
let _ = terminal_status.send(SourceStatus {
source_id,
room_id: terminal_room_id,
connected: false,
cookie_cloud: true,
detail: error,
});
}
// Keep the terminal status visible. A deliberate stop or restart
// removes the receiver first, so sends from an older task cannot
// affect a newer generation.
});
Ok(())
}
pub async fn restart(&self, source: SourceContext) -> Result<(), String> {
self.start(source).await
}
pub async fn stop(&self, source_id: Uuid) {
if let Some(entry) = self.running.write().await.remove(&source_id) {
entry.cancel.cancel();
}
}
pub async fn stop_all(&self) {
let mut running = self.running.write().await;
for (_, entry) in running.drain() {
entry.cancel.cancel();
}
}
pub async fn status(&self, source_id: Uuid) -> Option<SourceStatus> {
self.running
.read()
.await
.get(&source_id)
.map(|entry| entry.status.borrow().clone())
}
pub async fn statuses(&self) -> Vec<SourceStatus> {
self.running
.read()
.await
.values()
.map(|entry| entry.status.borrow().clone())
.collect()
}
}
File diff suppressed because it is too large Load Diff
+3 -61
View File
@@ -1,15 +1,12 @@
use std::{collections::HashMap, sync::Arc};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use serde_json::Value;
use tokio::sync::RwLock;
use tokio_postgres::NoTls;
use tracing::warn;
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct OverlaySettings {
pub title: String,
#[serde(default = "default_font_scale")]
pub font_scale: u16,
pub show_danmaku: bool,
@@ -37,7 +34,6 @@ pub struct OverlaySettings {
impl Default for OverlaySettings {
fn default() -> Self {
Self {
title: "洛星瓷专用弹幕猪!".into(),
font_scale: default_font_scale(),
show_danmaku: true,
show_enter: true,
@@ -61,10 +57,6 @@ impl Default for OverlaySettings {
impl OverlaySettings {
pub fn sanitize(mut self) -> Self {
self.title = self.title.trim().chars().take(48).collect();
if self.title.is_empty() {
self.title = Self::default().title;
}
self.max_visible = self.max_visible.clamp(1, 12);
self.font_scale = self.font_scale.clamp(50, 300);
self.collapse_after_seconds = self.collapse_after_seconds.clamp(2, 120);
@@ -95,45 +87,6 @@ fn default_particle_speed() -> u16 {
100
}
pub async fn load_settings(
database_url: &str,
room_id: &str,
defaults: OverlaySettings,
) -> Result<OverlaySettings, String> {
let (client, connection) = tokio_postgres::connect(database_url, NoTls)
.await
.map_err(|e| e.to_string())?;
tokio::spawn(async move {
let _ = connection.await;
});
let row = client
.query_opt(
"SELECT settings FROM overlay_settings WHERE room_id=$1",
&[&room_id],
)
.await
.map_err(|e| e.to_string())?;
Ok(row
.and_then(|row| serde_json::from_value::<OverlaySettings>(row.get::<_, Value>(0)).ok())
.unwrap_or(defaults)
.sanitize())
}
pub async fn save_settings(
database_url: &str,
room_id: &str,
settings: &OverlaySettings,
) -> Result<(), String> {
let (client, connection) = tokio_postgres::connect(database_url, NoTls)
.await
.map_err(|e| e.to_string())?;
tokio::spawn(async move {
let _ = connection.await;
});
client.execute("INSERT INTO overlay_settings(room_id,settings,updated_at) VALUES($1,$2,now()) ON CONFLICT(room_id) DO UPDATE SET settings=EXCLUDED.settings,updated_at=now()", &[&room_id, &json!(settings)]).await.map_err(|e| e.to_string())?;
Ok(())
}
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct GiftMeta {
@@ -276,17 +229,6 @@ impl GiftCatalog {
*self.by_name.write().await = names;
Ok(count)
}
pub fn spawn_refresh(self, room_id: String, interval_seconds: u64, timeout_seconds: u64) {
tokio::spawn(async move {
loop {
tokio::time::sleep(std::time::Duration::from_secs(interval_seconds)).await;
if let Err(error) = self.refresh(&room_id, timeout_seconds).await {
warn!(%error, "gift catalog refresh failed; retaining the last successful cache");
}
}
});
}
}
fn parse_catalog(
@@ -441,6 +383,7 @@ fn normalize_name(name: &str) -> String {
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn parses_current_gift_panel_fields_and_indexes_by_id_and_name() {
@@ -497,7 +440,6 @@ mod tests {
#[test]
fn settings_are_bounded_before_persistence() {
let settings = OverlaySettings {
title: " ".into(),
font_scale: 999,
max_visible: 99,
collapse_after_seconds: 1,
@@ -510,7 +452,6 @@ mod tests {
..OverlaySettings::default()
}
.sanitize();
assert_eq!(settings.title, "洛星瓷专用弹幕猪!");
assert_eq!(settings.font_scale, 300);
assert_eq!(settings.max_visible, 12);
assert_eq!(settings.collapse_after_seconds, 2);
@@ -532,6 +473,7 @@ mod tests {
fn old_saved_settings_receive_new_field_defaults() {
let mut value = json!(OverlaySettings::default());
let object = value.as_object_mut().expect("settings object");
object.insert("title".into(), json!("旧版弹幕栏标题"));
object.remove("fontScale");
object.remove("unfoldDurationMs");
object.remove("particleCount");
+154
View File
@@ -0,0 +1,154 @@
use std::{
collections::{HashMap, VecDeque},
sync::Arc,
time::{Duration, Instant},
};
use tokio::sync::Mutex;
#[derive(Clone)]
pub struct AuthRateLimiter {
attempts: Arc<Mutex<HashMap<String, AttemptBucket>>>,
window: Duration,
block_for: Duration,
max_failures: usize,
}
#[derive(Default)]
struct AttemptBucket {
failures: VecDeque<Instant>,
blocked_until: Option<Instant>,
}
#[derive(Clone, Copy, Debug)]
pub struct RateLimited {
pub retry_after: Duration,
}
impl Default for AuthRateLimiter {
fn default() -> Self {
Self::new(5, Duration::from_secs(5 * 60), Duration::from_secs(10 * 60))
}
}
impl AuthRateLimiter {
pub fn new(max_failures: usize, window: Duration, block_for: Duration) -> Self {
Self {
attempts: Arc::new(Mutex::new(HashMap::new())),
window,
block_for,
max_failures: max_failures.max(1),
}
}
/// Check both account and network dimensions. Callers intentionally receive
/// one generic result so this cannot be used to enumerate usernames.
pub async fn check(&self, username: &str, ip: &str) -> Result<(), RateLimited> {
let now = Instant::now();
let mut attempts = self.attempts.lock().await;
for key in keys(username, ip) {
let bucket = attempts.entry(key).or_default();
prune(bucket, now, self.window);
if let Some(until) = bucket.blocked_until.filter(|until| *until > now) {
return Err(RateLimited {
retry_after: until.duration_since(now),
});
}
}
Ok(())
}
pub async fn failure(&self, username: &str, ip: &str) {
let now = Instant::now();
let mut attempts = self.attempts.lock().await;
for key in keys(username, ip) {
let bucket = attempts.entry(key).or_default();
prune(bucket, now, self.window);
bucket.failures.push_back(now);
if bucket.failures.len() >= self.max_failures {
bucket.blocked_until = Some(now + self.block_for);
bucket.failures.clear();
}
}
// Opportunistic pruning bounds memory for a public login endpoint.
if attempts.len() > 8_192 {
attempts.retain(|_, bucket| {
prune(bucket, now, self.window);
!bucket.failures.is_empty() || bucket.blocked_until.is_some_and(|until| until > now)
});
}
}
pub async fn success(&self, username: &str, ip: &str) {
let mut attempts = self.attempts.lock().await;
// A successful account verification clears the account bucket. Keep
// the IP bucket so one valid account cannot reset an attack on others.
attempts.remove(&format!("account:{}", normalize_username(username)));
let _ = ip;
}
/// Consume one request from an IP-scoped budget. This is used for costly
/// anonymous enrollment work even when a request would otherwise succeed.
pub async fn consume_ip(&self, namespace: &str, ip: &str) -> Result<(), RateLimited> {
let identity = format!("{namespace}:{}", normalize_ip(ip));
self.check(&identity, ip).await?;
self.failure(&identity, ip).await;
Ok(())
}
}
fn keys(username: &str, ip: &str) -> [String; 2] {
[
format!("account:{}", normalize_username(username)),
format!("network:{}", normalize_ip(ip)),
]
}
fn normalize_username(value: &str) -> String {
value.trim().to_lowercase()
}
fn normalize_ip(value: &str) -> String {
let value = value.trim();
if value.is_empty() {
"unknown".into()
} else {
value.chars().take(96).collect()
}
}
fn prune(bucket: &mut AttemptBucket, now: Instant, window: Duration) {
while bucket
.failures
.front()
.is_some_and(|timestamp| now.duration_since(*timestamp) >= window)
{
bucket.failures.pop_front();
}
if bucket.blocked_until.is_some_and(|until| until <= now) {
bucket.blocked_until = None;
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn blocks_account_and_network_after_threshold() {
let limiter = AuthRateLimiter::new(2, Duration::from_secs(60), Duration::from_secs(60));
assert!(limiter.check("Streamer", "127.0.0.1").await.is_ok());
limiter.failure("Streamer", "127.0.0.1").await;
limiter.failure("Streamer", "127.0.0.1").await;
assert!(limiter.check("streamer", "127.0.0.1").await.is_err());
assert!(limiter.check("another", "127.0.0.1").await.is_err());
}
#[tokio::test]
async fn request_budget_counts_successful_anonymous_work() {
let limiter = AuthRateLimiter::new(2, Duration::from_secs(60), Duration::from_secs(60));
assert!(limiter.consume_ip("enroll", "127.0.0.1").await.is_ok());
assert!(limiter.consume_ip("enroll", "127.0.0.1").await.is_ok());
assert!(limiter.consume_ip("enroll", "127.0.0.1").await.is_err());
}
}
+559
View File
@@ -0,0 +1,559 @@
use std::{
collections::HashMap,
error::Error,
fmt,
future::Future,
pin::Pin,
sync::{Arc, RwLock},
};
use tokio::sync::broadcast;
use uuid::Uuid;
use crate::{
components::{ComponentError, ComponentInstance, ComponentRegistry},
domain::{ComponentMessage, LiveEvent},
};
/// Component-scoped in-process fanout. There is deliberately no global
/// receiver: possession of a receiver for component A cannot observe component
/// B, even when both consume the same source.
#[derive(Clone)]
pub struct EventHub {
capacity: usize,
channels: Arc<RwLock<HashMap<Uuid, broadcast::Sender<Arc<ComponentMessage>>>>>,
}
impl EventHub {
pub fn new(capacity: usize) -> Self {
Self {
capacity: capacity.max(1),
channels: Arc::new(RwLock::new(HashMap::new())),
}
}
fn sender(&self, component_id: Uuid) -> broadcast::Sender<Arc<ComponentMessage>> {
if let Some(sender) = self
.channels
.read()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.get(&component_id)
.cloned()
{
return sender;
}
let mut channels = self
.channels
.write()
.unwrap_or_else(|poisoned| poisoned.into_inner());
channels
.entry(component_id)
.or_insert_with(|| broadcast::channel(self.capacity).0)
.clone()
}
pub fn subscribe(&self, component_id: Uuid) -> broadcast::Receiver<Arc<ComponentMessage>> {
self.sender(component_id).subscribe()
}
/// Publish to exactly one component. A missing receiver is not an error:
/// side-effect handlers are run by the router independently of this hub.
pub fn publish(&self, component_id: Uuid, message: Arc<ComponentMessage>) -> usize {
self.sender(component_id).send(message).unwrap_or(0)
}
pub fn receiver_count(&self, component_id: Uuid) -> usize {
self.channels
.read()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.get(&component_id)
.map_or(0, broadcast::Sender::receiver_count)
}
/// Remove the sender after a component is deleted or its token is revoked.
/// Existing receivers observe channel closure once outstanding sender clones
/// are dropped.
pub fn remove(&self, component_id: Uuid) -> bool {
self.channels
.write()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.remove(&component_id)
.is_some()
}
pub fn active_component_count(&self) -> usize {
self.channels
.read()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.len()
}
}
impl Default for EventHub {
fn default() -> Self {
Self::new(256)
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ComponentStoreError {
pub detail: String,
}
impl ComponentStoreError {
pub fn new(detail: impl Into<String>) -> Self {
Self {
detail: detail.into(),
}
}
}
impl fmt::Display for ComponentStoreError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(&self.detail)
}
}
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.
pub trait ComponentInstanceStore: Send + Sync {
fn list_enabled_for_source<'a>(
&'a self,
owner_id: Uuid,
source_id: Uuid,
) -> ComponentStoreFuture<'a>;
}
#[derive(Clone, Default)]
pub struct InMemoryComponentStore {
instances: Arc<RwLock<Vec<ComponentInstance>>>,
}
impl InMemoryComponentStore {
pub fn new(instances: Vec<ComponentInstance>) -> Self {
Self {
instances: Arc::new(RwLock::new(instances)),
}
}
pub fn upsert(&self, instance: ComponentInstance) {
let mut instances = self
.instances
.write()
.unwrap_or_else(|poisoned| poisoned.into_inner());
if let Some(current) = instances.iter_mut().find(|item| item.id == instance.id) {
*current = instance;
} else {
instances.push(instance);
}
}
pub fn remove(&self, component_id: Uuid) -> bool {
let mut instances = self
.instances
.write()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let old_len = instances.len();
instances.retain(|instance| instance.id != component_id);
old_len != instances.len()
}
}
impl ComponentInstanceStore for InMemoryComponentStore {
fn list_enabled_for_source<'a>(
&'a self,
owner_id: Uuid,
source_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
})
.cloned()
.collect())
})
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum RouteStage {
Scope,
Registry,
Settings,
Handler,
Projection,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RouteFailure {
pub component_id: Uuid,
pub stage: RouteStage,
pub detail: String,
}
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct RouteReport {
pub considered: usize,
pub matched: usize,
pub handler_runs: usize,
pub projected: usize,
pub receiver_deliveries: usize,
pub failures: Vec<RouteFailure>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum RouteError {
Store(ComponentStoreError),
}
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}"),
}
}
}
impl Error for RouteError {}
impl From<ComponentStoreError> for RouteError {
fn from(value: ComponentStoreError) -> Self {
Self::Store(value)
}
}
pub type DispatchFuture<'a> =
Pin<Box<dyn Future<Output = Result<RouteReport, RouteError>> + Send + 'a>>;
/// Provider/source-facing abstraction. A Bilibili adapter only needs this port;
/// it does not need to know about WebSockets or any concrete component kind.
pub trait SourceEventSink: Send + Sync {
fn dispatch<'a>(&'a self, event: Arc<LiveEvent>) -> DispatchFuture<'a>;
}
#[derive(Clone)]
pub struct SourceEventRouter {
registry: ComponentRegistry,
components: Arc<dyn ComponentInstanceStore>,
hub: EventHub,
}
impl SourceEventRouter {
pub fn new(
registry: ComponentRegistry,
components: Arc<dyn ComponentInstanceStore>,
hub: EventHub,
) -> Self {
Self {
registry,
components,
hub,
}
}
pub fn registry(&self) -> &ComponentRegistry {
&self.registry
}
pub fn hub(&self) -> &EventHub {
&self.hub
}
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)
.await?;
let mut report = RouteReport {
considered: components.len(),
..RouteReport::default()
};
for component in components {
if !component.enabled
|| component.owner_id != event.owner_id
|| component.source_id != event.source_id
{
report.failures.push(RouteFailure {
component_id: component.id,
stage: RouteStage::Scope,
detail: "component owner/source does not match the source event".into(),
});
continue;
}
let runtime = match self.registry.runtime(&component.kind) {
Ok(runtime) => runtime,
Err(error) => {
report.failures.push(component_failure(
&component,
RouteStage::Registry,
error,
));
continue;
}
};
let subscription = match runtime.subscriptions(&component) {
Ok(subscription) => subscription,
Err(error) => {
report.failures.push(component_failure(
&component,
RouteStage::Settings,
error,
));
continue;
}
};
if !subscription.matches(&event) {
continue;
}
report.matched += 1;
// Active handlers are independent from projection and fanout. A
// handler failure is reported but does not make an OBS projection
// disappear.
for handler in runtime.handlers() {
if !handler.accepts(&component, &event) {
continue;
}
report.handler_runs += 1;
if let Err(error) = handler.handle(&component, event.clone()).await {
report.failures.push(RouteFailure {
component_id: component.id,
stage: RouteStage::Handler,
detail: format!("{}: {error}", handler.name()),
});
}
}
match runtime.project(&component, &event) {
Ok(Some(message)) => {
if message.owner_id != component.owner_id
|| message.source_id != component.source_id
|| message.component_id != component.id
{
report.failures.push(RouteFailure {
component_id: component.id,
stage: RouteStage::Projection,
detail: "projection changed component tenancy scope".into(),
});
continue;
}
report.receiver_deliveries += self.hub.publish(component.id, Arc::new(message));
report.projected += 1;
}
Ok(None) => {}
Err(error) => report.failures.push(component_failure(
&component,
RouteStage::Projection,
error,
)),
}
}
Ok(report)
}
}
impl SourceEventSink for SourceEventRouter {
fn dispatch<'a>(&'a self, event: Arc<LiveEvent>) -> DispatchFuture<'a> {
Box::pin(async move { self.route(event).await })
}
}
fn component_failure(
component: &ComponentInstance,
stage: RouteStage,
error: ComponentError,
) -> RouteFailure {
RouteFailure {
component_id: component.id,
stage,
detail: error.to_string(),
}
}
#[cfg(test)]
mod tests {
use std::sync::atomic::{AtomicUsize, Ordering};
use serde_json::json;
use tokio::sync::broadcast::error::TryRecvError;
use super::*;
use crate::{
components::{DANMAKU_OVERLAY_KIND, EventHandler, HandlerFuture},
domain::{DanmakuEvent, LiveEventPayload, PlatformViewer},
overlay::OverlaySettings,
};
fn danmaku(owner_id: Uuid, source_id: Uuid) -> Arc<LiveEvent> {
Arc::new(LiveEvent::new(
owner_id,
source_id,
"bilibili",
"123",
LiveEventPayload::Danmaku(DanmakuEvent {
viewer: PlatformViewer {
uid: "42".into(),
name: "观众".into(),
},
text: "晚上好".into(),
segments: Vec::new(),
}),
))
}
fn overlay(owner_id: Uuid, source_id: Uuid) -> ComponentInstance {
ComponentInstance::new(
owner_id,
source_id,
DANMAKU_OVERLAY_KIND,
"弹幕姬",
1,
serde_json::to_value(OverlaySettings::default()).unwrap(),
)
}
#[test]
fn event_hub_does_not_leak_between_component_channels() {
let hub = EventHub::new(8);
let owner_id = Uuid::new_v4();
let source_id = Uuid::new_v4();
let first = Uuid::new_v4();
let second = Uuid::new_v4();
let mut first_rx = hub.subscribe(first);
let mut second_rx = hub.subscribe(second);
let event = danmaku(owner_id, source_id);
let message = ComponentMessage::from_live_event(first, &event).unwrap();
assert_eq!(hub.publish(first, Arc::new(message)), 1);
assert_eq!(first_rx.try_recv().unwrap().component_id, first);
assert!(matches!(second_rx.try_recv(), Err(TryRecvError::Empty)));
}
#[tokio::test]
async fn removing_a_channel_disconnects_existing_token_subscribers() {
let hub = EventHub::new(8);
let component_id = Uuid::new_v4();
let mut old_receiver = hub.subscribe(component_id);
assert!(hub.remove(component_id));
assert!(matches!(
old_receiver.recv().await,
Err(tokio::sync::broadcast::error::RecvError::Closed)
));
let mut new_receiver = hub.subscribe(component_id);
let event = danmaku(Uuid::new_v4(), Uuid::new_v4());
let message = ComponentMessage::from_live_event(component_id, &event).unwrap();
assert_eq!(hub.publish(component_id, Arc::new(message)), 1);
assert_eq!(
new_receiver.recv().await.unwrap().component_id,
component_id
);
}
struct CountingHandler(Arc<AtomicUsize>);
impl EventHandler for CountingHandler {
fn name(&self) -> &'static str {
"counting-handler"
}
fn handle<'a>(
&'a self,
_component: &'a ComponentInstance,
_event: Arc<LiveEvent>,
) -> HandlerFuture<'a> {
Box::pin(async move {
self.0.fetch_add(1, Ordering::SeqCst);
Ok(())
})
}
}
#[tokio::test]
async fn router_runs_handlers_without_receivers_and_projects_the_event() {
let owner_id = Uuid::new_v4();
let source_id = Uuid::new_v4();
let component = overlay(owner_id, source_id);
let count = Arc::new(AtomicUsize::new(0));
let registry = ComponentRegistry::default();
registry
.register_handler(
DANMAKU_OVERLAY_KIND,
Arc::new(CountingHandler(count.clone())),
)
.unwrap();
let store = Arc::new(InMemoryComponentStore::new(vec![component]));
let router = SourceEventRouter::new(registry, store, EventHub::new(8));
let report = router.route(danmaku(owner_id, source_id)).await.unwrap();
assert_eq!(count.load(Ordering::SeqCst), 1);
assert_eq!(report.handler_runs, 1);
assert_eq!(report.projected, 1);
assert_eq!(report.receiver_deliveries, 0);
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> {
Box::pin(async move { Ok(self.instances.clone()) })
}
}
#[tokio::test]
async fn router_rejects_cross_owner_rows_even_if_store_is_buggy() {
let owner_id = Uuid::new_v4();
let source_id = Uuid::new_v4();
let other = overlay(Uuid::new_v4(), source_id);
let component_id = other.id;
let store = Arc::new(LeakyStore {
instances: vec![other],
});
let router = SourceEventRouter::new(ComponentRegistry::default(), store, EventHub::new(8));
let report = router.route(danmaku(owner_id, source_id)).await.unwrap();
assert_eq!(report.projected, 0);
assert_eq!(report.failures.len(), 1);
assert_eq!(report.failures[0].component_id, component_id);
assert_eq!(report.failures[0].stage, RouteStage::Scope);
}
#[tokio::test]
async fn disabled_event_categories_are_not_projected() {
let owner_id = Uuid::new_v4();
let source_id = Uuid::new_v4();
let mut component = overlay(owner_id, source_id);
component.settings["showDanmaku"] = json!(false);
let store = Arc::new(InMemoryComponentStore::new(vec![component]));
let router = SourceEventRouter::new(ComponentRegistry::default(), store, EventHub::new(8));
let report = router.route(danmaku(owner_id, source_id)).await.unwrap();
assert_eq!(report.matched, 0);
assert_eq!(report.projected, 0);
assert!(report.failures.is_empty());
}
}
+452
View File
@@ -0,0 +1,452 @@
use std::{fmt, sync::Arc};
use chrono::{DateTime, Utc};
use serde::Serialize;
use serde_json::Value;
use uuid::Uuid;
use crate::{
components::{ComponentInstance, ComponentRegistry},
db::{ComponentRecord, Db, DbError},
realtime::InMemoryComponentStore,
};
#[derive(Clone)]
pub struct TenantRepository {
db: Db,
registry: ComponentRegistry,
cache: Arc<InMemoryComponentStore>,
}
impl TenantRepository {
pub fn new(db: Db, registry: ComponentRegistry, cache: Arc<InMemoryComponentStore>) -> Self {
Self {
db,
registry,
cache,
}
}
pub fn cache(&self) -> Arc<InMemoryComponentStore> {
self.cache.clone()
}
pub async fn hydrate_all(&self) -> Result<(), RepositoryError> {
for tenant in self.db.list_active_tenants().await? {
self.hydrate_tenant(tenant.user_id).await?;
}
Ok(())
}
pub async fn hydrate_tenant(
&self,
owner_id: Uuid,
) -> Result<Vec<ComponentView>, RepositoryError> {
let rows = self.db.list_tenant_components(owner_id).await?;
let mut views = Vec::with_capacity(rows.len());
for row in rows {
let component = self.validate_loaded_component(component_from_record(row)?)?;
self.cache.upsert(component.clone());
views.push(ComponentView::from(&component));
}
Ok(views)
}
pub async fn list_components(
&self,
owner_id: Uuid,
) -> Result<Vec<ComponentView>, RepositoryError> {
self.hydrate_tenant(owner_id).await
}
pub async fn create_component(
&self,
owner_id: Uuid,
kind: &str,
name: &str,
) -> Result<ComponentInstance, RepositoryError> {
let runtime = self
.registry
.runtime(kind)
.map_err(|error| RepositoryError::Invalid(error.to_string()))?;
let name = name.trim();
if name.is_empty() || name.chars().count() > 80 {
return Err(RepositoryError::Invalid(
"component name must contain 1-80 characters".into(),
));
}
let source_id = self.source_id(owner_id).await?;
let component = ComponentInstance::new(
owner_id,
source_id,
runtime.kind(),
name,
runtime.definition().settings_version(),
runtime.definition().default_settings(),
);
let mut client = self.db.get().await?;
let transaction = client.transaction().await?;
Db::set_tenant(&transaction, owner_id).await?;
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)",
&[
&component.id,
&owner_id,
&source_id,
&component.kind,
&component.name,
&component.settings,
&(component.settings_version as i32),
],
)
.await?;
transaction.commit().await?;
self.cache.upsert(component.clone());
Ok(component)
}
pub async fn delete_component(
&self,
owner_id: Uuid,
component_id: Uuid,
) -> Result<(), RepositoryError> {
let mut client = self.db.get().await?;
let transaction = client.transaction().await?;
Db::set_tenant(&transaction, owner_id).await?;
let changed = transaction
.execute(
"DELETE FROM component_instances WHERE owner_user_id=$1 AND id=$2",
&[&owner_id, &component_id],
)
.await?;
if changed != 1 {
return Err(RepositoryError::NotFound);
}
transaction.commit().await?;
self.cache.remove(component_id);
Ok(())
}
pub async fn get_component(
&self,
owner_id: Uuid,
component_id: Uuid,
) -> Result<ComponentInstance, RepositoryError> {
let mut client = self.db.get().await?;
let transaction = client.transaction().await?;
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",
&[&owner_id, &component_id],
)
.await?
.ok_or(RepositoryError::NotFound)?;
transaction.commit().await?;
self.validate_loaded_component(component_from_record(ComponentRecord {
id: row.get(0),
owner_user_id: owner_id,
source_id: row.get(1),
kind: row.get(2),
name: row.get(3),
settings: row.get(4),
settings_version: row.get(5),
enabled: row.get(6),
})?)
}
pub async fn update_component_settings(
&self,
owner_id: Uuid,
component_id: Uuid,
settings: Value,
) -> Result<ComponentInstance, RepositoryError> {
let current = self.get_component(owner_id, component_id).await?;
let validated = self
.registry
.validate_settings(&current.kind, current.settings_version, settings)
.map_err(|error| RepositoryError::Invalid(error.to_string()))?;
let mut client = self.db.get().await?;
let transaction = client.transaction().await?;
Db::set_tenant(&transaction, owner_id).await?;
let changed = transaction
.execute(
"UPDATE component_instances SET settings=$1,updated_at=now() \
WHERE id=$2 AND owner_user_id=$3",
&[&validated, &component_id, &owner_id],
)
.await?;
if changed != 1 {
return Err(RepositoryError::NotFound);
}
transaction.commit().await?;
let updated = ComponentInstance {
settings: validated,
..current
};
self.cache.upsert(updated.clone());
Ok(updated)
}
pub async fn source_id(&self, owner_id: Uuid) -> Result<Uuid, RepositoryError> {
let mut client = self.db.get().await?;
let transaction = client.transaction().await?;
Db::set_tenant(&transaction, owner_id).await?;
let row = transaction
.query_opt(
"SELECT id FROM live_sources WHERE owner_user_id=$1 AND enabled",
&[&owner_id],
)
.await?
.ok_or(RepositoryError::NotFound)?;
transaction.commit().await?;
Ok(row.get(0))
}
pub async fn room_id(&self, owner_id: Uuid) -> Result<String, RepositoryError> {
let client = self.db.get().await?;
let row = client
.query_opt(
"SELECT room_id FROM users WHERE id=$1 AND status='active'",
&[&owner_id],
)
.await?
.ok_or(RepositoryError::NotFound)?;
Ok(row.get(0))
}
pub async fn token_summary(
&self,
owner_id: Uuid,
component_id: Uuid,
) -> Result<ComponentTokenSummary, RepositoryError> {
let _ = self.get_component(owner_id, component_id).await?;
let mut client = self.db.get().await?;
let transaction = client.transaction().await?;
Db::set_tenant(&transaction, owner_id).await?;
let row = transaction
.query_opt(
"SELECT created_at,last_used_at FROM component_access_tokens \
WHERE owner_user_id=$1 AND component_instance_id=$2 AND revoked_at IS NULL \
AND (expires_at IS NULL OR expires_at>now()) ORDER BY created_at DESC LIMIT 1",
&[&owner_id, &component_id],
)
.await?;
transaction.commit().await?;
Ok(ComponentTokenSummary {
configured: row.is_some(),
updated_at: row.as_ref().map(|row| row.get(0)),
last_used_at: row.as_ref().and_then(|row| row.get(1)),
})
}
pub async fn revoke_component_tokens(
&self,
owner_id: Uuid,
component_id: Uuid,
) -> Result<u64, RepositoryError> {
let _ = self.get_component(owner_id, component_id).await?;
let mut client = self.db.get().await?;
let transaction = client.transaction().await?;
Db::set_tenant(&transaction, owner_id).await?;
let changed = transaction
.execute(
"UPDATE component_access_tokens SET revoked_at=now() \
WHERE owner_user_id=$1 AND component_instance_id=$2 AND revoked_at IS NULL",
&[&owner_id, &component_id],
)
.await?;
transaction.commit().await?;
Ok(changed)
}
pub async fn setup_required(&self) -> Result<bool, RepositoryError> {
let client = self.db.get().await?;
Ok(!client
.query_one("SELECT EXISTS(SELECT 1 FROM users)", &[])
.await?
.get::<_, bool>(0))
}
pub async fn list_invitations(
&self,
actor_id: Uuid,
) -> Result<Vec<InvitationView>, RepositoryError> {
let client = self.db.get().await?;
require_system_admin(&client, actor_id).await?;
let rows = client
.query(
"SELECT id,code_prefix,room_id,created_at,expires_at,consumed_at,revoked_at \
FROM invitations WHERE grant_role='user' ORDER BY created_at DESC LIMIT 250",
&[],
)
.await?;
Ok(rows
.into_iter()
.map(|row| InvitationView {
id: row.get(0),
code_prefix: row.get(1),
room_id: row.get(2),
created_at: row.get(3),
expires_at: row.get(4),
consumed_at: row.get(5),
revoked_at: row.get(6),
max_uses: 1,
used_count: if row.get::<_, Option<DateTime<Utc>>>(5).is_some() {
1
} else {
0
},
})
.collect())
}
pub async fn legacy_overlay_settings(
&self,
room_id: &str,
fallback: Value,
) -> Result<Value, RepositoryError> {
let client = self.db.get().await?;
Ok(client
.query_opt(
"SELECT settings FROM overlay_settings WHERE room_id=$1",
&[&room_id],
)
.await?
.map(|row| row.get(0))
.unwrap_or(fallback))
}
fn validate_loaded_component(
&self,
mut component: ComponentInstance,
) -> Result<ComponentInstance, RepositoryError> {
component.settings = self
.registry
.validate_settings(
&component.kind,
component.settings_version,
component.settings,
)
.map_err(|error| RepositoryError::Invalid(error.to_string()))?;
Ok(component)
}
}
fn component_from_record(record: ComponentRecord) -> Result<ComponentInstance, RepositoryError> {
let settings_version = u32::try_from(record.settings_version)
.map_err(|_| RepositoryError::Invalid("negative settings version".into()))?;
Ok(ComponentInstance {
id: record.id,
owner_id: record.owner_user_id,
source_id: record.source_id,
kind: record.kind,
name: record.name,
enabled: record.enabled,
settings_version,
settings: record.settings,
})
}
async fn require_system_admin(
client: &deadpool_postgres::Object,
actor_id: Uuid,
) -> Result<(), RepositoryError> {
let allowed = client
.query_one(
"SELECT EXISTS(SELECT 1 FROM users WHERE id=$1 AND role='system_admin' AND status='active')",
&[&actor_id],
)
.await?
.get::<_, bool>(0);
if allowed {
Ok(())
} else {
Err(RepositoryError::Forbidden)
}
}
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ComponentView {
pub id: Uuid,
pub public_id: Uuid,
pub kind: String,
pub name: String,
pub enabled: bool,
pub settings: Value,
}
impl From<&ComponentInstance> for ComponentView {
fn from(component: &ComponentInstance) -> Self {
Self {
id: component.id,
public_id: component.id,
kind: component.kind.clone(),
name: component.name.clone(),
enabled: component.enabled,
settings: component.settings.clone(),
}
}
}
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ComponentTokenSummary {
pub configured: bool,
pub updated_at: Option<DateTime<Utc>>,
pub last_used_at: Option<DateTime<Utc>>,
}
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct InvitationView {
pub id: Uuid,
pub code_prefix: String,
pub room_id: String,
pub created_at: DateTime<Utc>,
pub expires_at: DateTime<Utc>,
pub consumed_at: Option<DateTime<Utc>>,
pub revoked_at: Option<DateTime<Utc>>,
pub max_uses: i32,
pub used_count: i32,
}
#[derive(Debug)]
pub enum RepositoryError {
NotFound,
Forbidden,
Invalid(String),
Database(DbError),
Postgres(tokio_postgres::Error),
}
impl fmt::Display for RepositoryError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::NotFound => formatter.write_str("resource was not found"),
Self::Forbidden => formatter.write_str("operation is not permitted"),
Self::Invalid(message) => write!(formatter, "invalid value: {message}"),
Self::Database(error) => error.fmt(formatter),
Self::Postgres(error) => error.fmt(formatter),
}
}
}
impl std::error::Error for RepositoryError {}
impl From<DbError> for RepositoryError {
fn from(value: DbError) -> Self {
Self::Database(value)
}
}
impl From<tokio_postgres::Error> for RepositoryError {
fn from(value: tokio_postgres::Error) -> Self {
Self::Postgres(value)
}
}