feat(chat): add live crate (presence, typing, service, webhook, token)
zhenyi committed
Added chat/live/Cargo.toml chat/live/Cargo.toml +33 −0 View chat/live/Cargo.toml
@@ -0,0 +1,33 @@
1[package]
2name = "live"
3version.workspace = true
4edition.workspace = true
5authors.workspace = true
6categories.workspace = true
7keywords.workspace = true
8repository.workspace = true
9homepage.workspace = true
10
11[lib]
12path = "lib.rs"
13name = "live"
14
15[dependencies]
16chatlayer.workspace = true
17storagelayer.workspace = true
18config.workspace = true
19
20thiserror.workspace = true
21serde.workspace = true
22serde_json.workspace = true
23time.workspace = true
24uuid.workspace = true
25metrics.workspace = true
26axum.workspace = true
27reqwest.workspace = true
28jsonwebtoken.workspace = true
29sha2.workspace = true
30hex.workspace = true
31
32[dev-dependencies]
33storagelayer = { workspace = true, features = ["mock"] }
No newline at end of file
Added chat/live/client.rs chat/live/client.rs +126 −0 View chat/live/client.rs
@@ -0,0 +1,126 @@
1use serde::{Deserialize, Serialize};
2
3use crate::error::{LiveError, LiveResult};
4use crate::token::mint_admin_token;
5
6#[derive(Debug, Deserialize, Clone)]
7pub struct Participant {
8 pub sid: String,
9 pub identity: String,
10}
11
12#[derive(Debug, Clone)]
13pub struct LiveKitClient {
14 base_url: String,
15 api_key: String,
16 api_secret: String,
17 http: reqwest::Client,
18}
19
20impl LiveKitClient {
21 pub fn new(
22 base_url: impl Into<String>,
23 api_key: impl Into<String>,
24 api_secret: impl Into<String>,
25 ) -> Self {
26 Self {
27 base_url: base_url.into(),
28 api_key: api_key.into(),
29 api_secret: api_secret.into(),
30 http: reqwest::Client::new(),
31 }
32 }
33
34 pub fn configured(&self) -> bool {
35 !self.base_url.is_empty() && !self.api_key.is_empty() && !self.api_secret.is_empty()
36 }
37
38 fn endpoint(&self, path: &str) -> String {
39 format!("{}/twirp/livekit.RoomService/{}", self.base_url, path)
40 }
41
42 async fn admin_token(&self) -> LiveResult<String> {
43 mint_admin_token(&self.api_key, &self.api_secret, 60)
44 }
45
46 async fn post<T: Serialize, R: for<'de> Deserialize<'de>>(
47 &self,
48 path: &str,
49 body: &T,
50 ) -> LiveResult<R> {
51 let token = self.admin_token().await?;
52 let resp = self
53 .http
54 .post(self.endpoint(path))
55 .bearer_auth(token)
56 .json(body)
57 .send()
58 .await?;
59 if !resp.status().is_success() {
60 let status = resp.status();
61 let text = resp.text().await.unwrap_or_default();
62 return Err(LiveError::LiveKitApi(format!(
63 "POST {path} -> {status}: {text}"
64 )));
65 }
66 resp.json().await.map_err(LiveError::from)
67 }
68
69 pub async fn create_room(&self, room_name: &str, empty_timeout_seconds: u64) -> LiveResult<()> {
70 #[derive(Serialize)]
71 struct CreateRoomBody<'a> {
72 name: &'a str,
73 empty_timeout: u64,
74 }
75 #[derive(Deserialize)]
76 struct CreateRoomResp {}
77 let body = CreateRoomBody {
78 name: room_name,
79 empty_timeout: empty_timeout_seconds,
80 };
81 let _: CreateRoomResp = self.post("CreateRoom", &body).await?;
82 Ok(())
83 }
84
85 pub async fn delete_room(&self, room_name: &str) -> LiveResult<()> {
86 #[derive(Serialize)]
87 struct DeleteRoomBody<'a> {
88 room: &'a str,
89 }
90 #[derive(Deserialize)]
91 struct DeleteRoomResp {}
92 let body = DeleteRoomBody { room: room_name };
93 let _: DeleteRoomResp = self.post("DeleteRoom", &body).await?;
94 Ok(())
95 }
96
97 pub async fn list_participants(&self, room_name: &str) -> LiveResult<Vec<Participant>> {
98 #[derive(Serialize)]
99 struct ListParticipantsBody<'a> {
100 room: &'a str,
101 }
102 #[derive(Deserialize)]
103 struct ListParticipantsResp {
104 participants: Vec<Participant>,
105 }
106 let body = ListParticipantsBody { room: room_name };
107 let resp: ListParticipantsResp = self.post("ListParticipants", &body).await?;
108 Ok(resp.participants)
109 }
110
111 pub async fn remove_participant(&self, room_name: &str, identity: &str) -> LiveResult<()> {
112 #[derive(Serialize)]
113 struct RemoveBody<'a> {
114 room: &'a str,
115 identity: &'a str,
116 }
117 #[derive(Deserialize)]
118 struct RemoveResp {}
119 let body = RemoveBody {
120 room: room_name,
121 identity,
122 };
123 let _: RemoveResp = self.post("RemoveParticipant", &body).await?;
124 Ok(())
125 }
126}
Added chat/live/error.rs chat/live/error.rs +84 −0 View chat/live/error.rs
@@ -0,0 +1,84 @@
1use axum::http::StatusCode;
2use axum::response::{IntoResponse, Response};
3use thiserror::Error;
4
5pub type LiveResult<T> = Result<T, LiveError>;
6
7#[derive(Debug, Error)]
8pub enum LiveError {
9 #[error("bad request: {0}")]
10 BadRequest(String),
11 #[error("not found: {0}")]
12 NotFound(String),
13 #[error("conflict: {0}")]
14 Conflict(String),
15 #[error("livekit not configured: {0}")]
16 NotConfigured(String),
17 #[error("livekit api error: {0}")]
18 LiveKitApi(String),
19 #[error("webhook verification failed: {0}")]
20 WebhookVerification(String),
21 #[error("chat layer error: {0}")]
22 ChatLayer(#[from] chatlayer::ChatLayerError),
23 #[error("storage error: {0}")]
24 Storage(#[from] storagelayer::ChatStorageError),
25 #[error("serde error: {0}")]
26 Serde(#[from] serde_json::Error),
27 #[error("reqwest error: {0}")]
28 Reqwest(#[from] reqwest::Error),
29 #[error("jwt error: {0}")]
30 Jwt(#[from] jsonwebtoken::errors::Error),
31 #[error("internal: {0}")]
32 Internal(String),
33}
34
35impl LiveError {
36 fn status(&self) -> StatusCode {
37 match self {
38 Self::BadRequest(_) => StatusCode::BAD_REQUEST,
39 Self::NotFound(_) => StatusCode::NOT_FOUND,
40 Self::Conflict(_) => StatusCode::CONFLICT,
41 Self::NotConfigured(_) => StatusCode::SERVICE_UNAVAILABLE,
42 Self::LiveKitApi(_) | Self::WebhookVerification(_) => StatusCode::BAD_GATEWAY,
43 Self::ChatLayer(_) | Self::Storage(_) => StatusCode::INTERNAL_SERVER_ERROR,
44 Self::Serde(_) | Self::Reqwest(_) | Self::Jwt(_) | Self::Internal(_) => {
45 StatusCode::INTERNAL_SERVER_ERROR
46 }
47 }
48 }
49
50 pub fn error_kind(&self) -> &'static str {
51 match self {
52 Self::BadRequest(_) => "bad_request",
53 Self::NotFound(_) => "not_found",
54 Self::Conflict(_) => "conflict",
55 Self::NotConfigured(_) => "not_configured",
56 Self::LiveKitApi(_) => "livekit_api",
57 Self::WebhookVerification(_) => "webhook_verification",
58 Self::ChatLayer(_) => "chat_layer",
59 Self::Storage(_) => "storage",
60 Self::Serde(_) => "serde",
61 Self::Reqwest(_) => "reqwest",
62 Self::Jwt(_) => "jwt",
63 Self::Internal(_) => "internal",
64 }
65 }
66}
67
68impl IntoResponse for LiveError {
69 fn into_response(self) -> Response {
70 let status = self.status();
71 let (status, body) = match &self {
72 Self::WebhookVerification(_) | Self::NotConfigured(_) => {
73 (StatusCode::OK, serde_json::json!({ "ok": false }))
74 }
75 _ => {
76 let body = serde_json::json!({
77 "error": { "kind": self.error_kind(), "message": self.to_string() }
78 });
79 (status, body)
80 }
81 };
82 (status, axum::Json(body)).into_response()
83 }
84}
Added chat/live/lib.rs chat/live/lib.rs +20 −0 View chat/live/lib.rs
@@ -0,0 +1,20 @@
1#![warn(missing_debug_implementations)]
2#![warn(unreachable_pub)]
3
4pub mod client;
5pub mod error;
6pub mod metrics;
7pub mod service;
8pub mod token;
9pub mod types;
10pub mod webhook;
11
12pub use client::LiveKitClient;
13pub use error::{LiveError, LiveResult};
14pub use service::LiveService;
15pub use token::{LiveTokenClaims, VideoGrants, mint_access_token};
16pub use types::{JoinInfo, MeetInfo, MeetPresence, ParticipantRecord};
17pub use webhook::{
18 LiveKitWebhook, WebhookParticipant, WebhookRoom, build_room_name, event_from_webhook,
19 verify_webhook,
20};
Added chat/live/metrics.rs chat/live/metrics.rs +21 −0 View chat/live/metrics.rs
@@ -0,0 +1,21 @@
1use metrics::counter;
2
3pub fn webhook_received(event: &str) {
4 counter!("chat.live_webhooks_total", "event" => event.to_string()).increment(1);
5}
6
7pub fn webhook_ignored(event: &str) {
8 counter!("chat.live_webhooks_ignored_total", "event" => event.to_string()).increment(1);
9}
10
11pub fn token_minted() {
12 counter!("chat.live_tokens_minted_total").increment(1);
13}
14
15pub fn room_created(result: &'static str) {
16 counter!("chat.live_rooms_created_total", "result" => result).increment(1);
17}
18
19pub fn room_ended(result: &'static str) {
20 counter!("chat.live_rooms_ended_total", "result" => result).increment(1);
21}
Added chat/live/service.rs chat/live/service.rs +291 −0 View chat/live/service.rs
@@ -0,0 +1,291 @@
1use std::sync::Arc;
2
3use axum::http::HeaderMap;
4use storagelayer::{ChatStorageCtx, DynChatStorage};
5use uuid::Uuid;
6
7use config::live::LiveSettings;
8
9use crate::client::LiveKitClient;
10use crate::error::{LiveError, LiveResult};
11use crate::types::{JoinInfo, MeetInfo, MeetPresence, ParticipantRecord};
12use crate::webhook::{LiveKitWebhook, event_from_webhook, verify_webhook};
13
14#[derive(Clone)]
15pub struct LiveService {
16 pub(crate) store: DynChatStorage,
17 pub(crate) producer: Arc<chatlayer::ChatProducer>,
18 pub(crate) client: LiveKitClient,
19 pub(crate) settings: LiveSettings,
20}
21
22impl std::fmt::Debug for LiveService {
23 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
24 f.debug_struct("LiveService")
25 .field("enabled", &self.settings.enabled)
26 .field("base_url", &self.settings.base_url)
27 .finish_non_exhaustive()
28 }
29}
30
31impl LiveService {
32 pub fn new(
33 store: DynChatStorage,
34 producer: Arc<chatlayer::ChatProducer>,
35 settings: LiveSettings,
36 ) -> Self {
37 let client = LiveKitClient::new(
38 settings.base_url.clone(),
39 settings.api_key.clone(),
40 settings.api_secret.clone(),
41 );
42 Self {
43 store,
44 producer,
45 client,
46 settings,
47 }
48 }
49
50 pub fn enabled(&self) -> bool {
51 self.settings.enabled && self.client.configured()
52 }
53
54 fn require_enabled(&self) -> LiveResult<()> {
55 if self.enabled() {
56 Ok(())
57 } else {
58 Err(LiveError::NotConfigured(
59 "live module is disabled or livekit credentials missing".into(),
60 ))
61 }
62 }
63
64 fn room_url(&self) -> String {
65 self.settings
66 .base_url
67 .trim_end_matches('/')
68 .replace("http://", "wss://")
69 .replace("https://", "wss://")
70 + "/room"
71 }
72
73 pub async fn join_room(&self, user_id: Uuid, channel_id: Uuid) -> LiveResult<JoinInfo> {
74 self.require_enabled()?;
75 let ctx = ChatStorageCtx::new(user_id);
76
77 let (meet, is_first) = match self
78 .store
79 .meets()
80 .find_active_by_channel(&ctx, channel_id)
81 .await?
82 {
83 Some(meet) => (meet, false),
84 None => {
85 let room_name = crate::webhook::build_room_name(channel_id);
86 let created = self
87 .store
88 .meets()
89 .create(
90 &ctx,
91 storagelayer::NewMeet {
92 channel_id,
93 room_name: room_name.clone(),
94 created_by: user_id,
95 },
96 )
97 .await?;
98 self.client
99 .create_room(&room_name, self.settings.empty_timeout_seconds)
100 .await?;
101 let activated = self.store.meets().mark_active(&ctx, created.id).await?;
102 (activated, true)
103 }
104 };
105
106 let token = crate::token::mint_access_token(
107 &self.settings.api_key,
108 &self.settings.api_secret,
109 user_id,
110 &meet.room_name,
111 self.settings.room_token_ttl_seconds,
112 )?;
113
114 self.store
115 .meet_participants()
116 .join(
117 &ctx,
118 storagelayer::NewMeetParticipant {
119 meet_id: meet.id,
120 user_id,
121 joined_at: time::OffsetDateTime::now_utc(),
122 connection_id: None,
123 },
124 )
125 .await?;
126
127 let ts = time::OffsetDateTime::now_utc();
128 let envelope = if is_first {
129 chatlayer::EventEnvelope::new(
130 user_id,
131 chatlayer::ChatLiveEvent::MeetStarted {
132 channel_id,
133 room_name: meet.room_name.clone(),
134 by: user_id,
135 ts,
136 },
137 )
138 } else {
139 chatlayer::EventEnvelope::new(
140 user_id,
141 chatlayer::ChatLiveEvent::ParticipantJoined {
142 channel_id,
143 room_name: meet.room_name.clone(),
144 user_id,
145 ts,
146 },
147 )
148 };
149 self.producer.publish_live(&envelope).await?;
150
151 let participants = self
152 .store
153 .meet_participants()
154 .list_active(&ctx, meet.id)
155 .await?
156 .into_iter()
157 .map(ParticipantRecord::from)
158 .collect();
159
160 Ok(JoinInfo {
161 meet: MeetInfo::from(meet),
162 access_token: token,
163 room_url: self.room_url(),
164 participants,
165 })
166 }
167
168 pub async fn leave_room(&self, user_id: Uuid, meet_id: Uuid) -> LiveResult<()> {
169 self.require_enabled()?;
170 let ctx = ChatStorageCtx::new(user_id);
171 let active = self
172 .store
173 .meet_participants()
174 .list_active(&ctx, meet_id)
175 .await?;
176 let Some(row) = active.iter().find(|p| p.user_id == user_id).map(|p| p.id) else {
177 return Err(LiveError::Conflict(format!(
178 "user {user_id} is not in meet {meet_id}"
179 )));
180 };
181 let meet = self.store.meets().find_by_id(&ctx, meet_id).await?;
182 self.store
183 .meet_participants()
184 .leave(&ctx, row, storagelayer::LeaveReason::Left)
185 .await?;
186
187 if let Some(meet) = meet {
188 let envelope = chatlayer::EventEnvelope::new(
189 user_id,
190 chatlayer::ChatLiveEvent::ParticipantLeft {
191 channel_id: meet.channel_id,
192 room_name: meet.room_name.clone(),
193 user_id,
194 reason: storagelayer::LeaveReason::Left,
195 ts: time::OffsetDateTime::now_utc(),
196 },
197 );
198 self.producer.publish_live(&envelope).await?;
199 }
200 Ok(())
201 }
202
203 pub async fn end_room(&self, user_id: Uuid, channel_id: Uuid) -> LiveResult<()> {
204 self.require_enabled()?;
205 let ctx = ChatStorageCtx::new(user_id);
206 let Some(meet) = self
207 .store
208 .meets()
209 .find_active_by_channel(&ctx, channel_id)
210 .await?
211 else {
212 return Err(LiveError::NotFound(format!(
213 "no active meet for channel {channel_id}"
214 )));
215 };
216 self.client.delete_room(&meet.room_name).await?;
217 self.store.meets().mark_ended(&ctx, meet.id).await?;
218 self.store
219 .meet_participants()
220 .mark_all_ended(&ctx, meet.id)
221 .await?;
222
223 let envelope = chatlayer::EventEnvelope::new(
224 user_id,
225 chatlayer::ChatLiveEvent::MeetEnded {
226 channel_id: meet.channel_id,
227 room_name: meet.room_name.clone(),
228 by: user_id,
229 ts: time::OffsetDateTime::now_utc(),
230 },
231 );
232 self.producer.publish_live(&envelope).await?;
233 Ok(())
234 }
235
236 pub async fn presence(&self, user_id: Uuid, channel_id: Uuid) -> LiveResult<MeetPresence> {
237 let ctx = ChatStorageCtx::new(user_id);
238 let Some(meet) = self
239 .store
240 .meets()
241 .find_active_by_channel(&ctx, channel_id)
242 .await?
243 else {
244 return Err(LiveError::NotFound(format!(
245 "no active meet for channel {channel_id}"
246 )));
247 };
248 let participants = self
249 .store
250 .meet_participants()
251 .list_active(&ctx, meet.id)
252 .await?
253 .into_iter()
254 .map(ParticipantRecord::from)
255 .collect();
256 Ok(MeetPresence {
257 meet: MeetInfo::from(meet),
258 participants,
259 })
260 }
261
262 pub async fn list_history(
263 &self,
264 user_id: Uuid,
265 limit: u32,
266 ) -> LiveResult<Vec<ParticipantRecord>> {
267 let ctx = ChatStorageCtx::new(user_id);
268 let records = self
269 .store
270 .meet_participants()
271 .list_history(&ctx, user_id, limit)
272 .await?
273 .into_iter()
274 .map(ParticipantRecord::from)
275 .collect();
276 Ok(records)
277 }
278
279 pub async fn handle_webhook(&self, headers: &HeaderMap, body: &[u8]) -> LiveResult<()> {
280 verify_webhook(headers, body, &self.settings.webhook_secret)?;
281 let webhook: LiveKitWebhook = serde_json::from_slice(body)?;
282 crate::metrics::webhook_received(&webhook.event);
283 let Some(envelope_payload) = event_from_webhook(&webhook)? else {
284 crate::metrics::webhook_ignored(&webhook.event);
285 return Ok(());
286 };
287 let sub = chatlayer::EventEnvelope::new(Uuid::nil(), envelope_payload);
288 self.producer.publish_live(&sub).await?;
289 Ok(())
290 }
291}
Added chat/live/token.rs chat/live/token.rs +193 −0 View chat/live/token.rs
@@ -0,0 +1,193 @@
1use jsonwebtoken::{Algorithm, EncodingKey, Header, encode};
2use serde::{Deserialize, Serialize};
3use time::OffsetDateTime;
4use uuid::Uuid;
5
6use crate::error::{LiveError, LiveResult};
7
8/// LiveKit `VideoGrants` — the authorization claims embedded in an access
9#[derive(Debug, Clone, Serialize, Deserialize)]
10pub struct VideoGrants {
11 #[serde(rename = "room", skip_serializing_if = "Option::is_none")]
12 pub room: Option<String>,
13 #[serde(rename = "roomJoin", skip_serializing_if = "Option::is_none")]
14 pub room_join: Option<bool>,
15 #[serde(rename = "roomPublish", skip_serializing_if = "Option::is_none")]
16 pub room_publish: Option<bool>,
17 #[serde(rename = "roomSubscribe", skip_serializing_if = "Option::is_none")]
18 pub room_subscribe: Option<bool>,
19 #[serde(rename = "roomAdmin", skip_serializing_if = "Option::is_none")]
20 pub room_admin: Option<bool>,
21 #[serde(rename = "roomRecord", skip_serializing_if = "Option::is_none")]
22 pub room_record: Option<bool>,
23 #[serde(rename = "canPublish", skip_serializing_if = "Option::is_none")]
24 pub can_publish: Option<bool>,
25 #[serde(rename = "canSubscribe", skip_serializing_if = "Option::is_none")]
26 pub can_subscribe: Option<bool>,
27 #[serde(rename = "canPublishData", skip_serializing_if = "Option::is_none")]
28 pub can_publish_data: Option<bool>,
29 #[serde(
30 rename = "canUpdateOwnMetadata",
31 skip_serializing_if = "Option::is_none"
32 )]
33 pub can_update_own_metadata: Option<bool>,
34 #[serde(rename = "hidden", skip_serializing_if = "Option::is_none")]
35 pub hidden: Option<bool>,
36 #[serde(rename = "recorder", skip_serializing_if = "Option::is_none")]
37 pub recorder: Option<bool>,
38}
39
40impl VideoGrants {
41 pub fn participant(room: impl Into<String>) -> Self {
42 Self {
43 room: Some(room.into()),
44 room_join: Some(true),
45 room_publish: Some(true),
46 room_subscribe: Some(true),
47 room_admin: None,
48 room_record: Some(false),
49 can_publish: Some(true),
50 can_subscribe: Some(true),
51 can_publish_data: Some(true),
52 can_update_own_metadata: Some(true),
53 hidden: Some(false),
54 recorder: Some(false),
55 }
56 }
57
58 pub fn admin() -> Self {
59 Self {
60 room: None,
61 room_join: Some(false),
62 room_publish: Some(false),
63 room_subscribe: Some(false),
64 room_admin: Some(true),
65 room_record: None,
66 can_publish: Some(false),
67 can_subscribe: Some(false),
68 can_publish_data: Some(false),
69 can_update_own_metadata: Some(false),
70 hidden: Some(true),
71 recorder: None,
72 }
73 }
74}
75
76#[derive(Debug, Clone, Serialize, Deserialize)]
77pub struct LiveTokenClaims {
78 pub iss: String,
79 #[serde(rename = "sub")]
80 pub sub: String,
81 #[serde(rename = "name", skip_serializing_if = "Option::is_none")]
82 pub name: Option<String>,
83 #[serde(rename = "video")]
84 pub video: VideoGrants,
85 #[serde(rename = "nbf")]
86 pub nbf: i64,
87 #[serde(rename = "exp")]
88 pub exp: i64,
89 #[serde(rename = "iat")]
90 pub iat: i64,
91 #[serde(rename = "jti", skip_serializing_if = "Option::is_none")]
92 pub jti: Option<String>,
93}
94
95pub fn mint_access_token(
96 api_key: &str,
97 api_secret: &str,
98 user_id: Uuid,
99 room_name: &str,
100 ttl_seconds: u64,
101) -> LiveResult<String> {
102 mint_with_grants(
103 api_key,
104 api_secret,
105 user_id,
106 VideoGrants::participant(room_name),
107 ttl_seconds,
108 )
109}
110
111pub fn mint_admin_token(api_key: &str, api_secret: &str, ttl_seconds: u64) -> LiveResult<String> {
112 mint_with_grants(
113 api_key,
114 api_secret,
115 Uuid::nil(),
116 VideoGrants::admin(),
117 ttl_seconds,
118 )
119}
120
121fn mint_with_grants(
122 api_key: &str,
123 api_secret: &str,
124 user_id: Uuid,
125 video: VideoGrants,
126 ttl_seconds: u64,
127) -> LiveResult<String> {
128 if api_key.is_empty() || api_secret.is_empty() {
129 return Err(LiveError::NotConfigured(
130 "livekit api_key/api_secret must be set".into(),
131 ));
132 }
133 let now = OffsetDateTime::now_utc().unix_timestamp();
134 let claims = LiveTokenClaims {
135 iss: api_key.to_string(),
136 sub: user_id.to_string(),
137 name: None,
138 video,
139 nbf: now - 10,
140 exp: now + ttl_seconds as i64,
141 iat: now,
142 jti: Some(Uuid::now_v7().to_string()),
143 };
144 let mut header = Header::new(Algorithm::HS256);
145 header.kid = Some(api_key.to_string());
146 let key = EncodingKey::from_secret(api_secret.as_bytes());
147 encode(&header, &claims, &key).map_err(LiveError::from)
148}
149
150#[cfg(test)]
151mod tests {
152 use super::*;
153 use jsonwebtoken::{DecodingKey, Validation, decode};
154
155 #[test]
156 fn participant_token_decodes_with_matching_grants() {
157 let api_key = "devkey";
158 let api_secret = "devsecret";
159 let user = Uuid::now_v7();
160 let room = format!("room-{}", Uuid::now_v7());
161
162 let token = mint_access_token(api_key, api_secret, user, &room, 300).unwrap();
163 let mut validation = Validation::new(Algorithm::HS256);
164 validation.validate_exp = true;
165 validation.validate_nbf = true;
166 let key = DecodingKey::from_secret(api_secret.as_bytes());
167 let data = decode::<LiveTokenClaims>(&token, &key, &validation).unwrap();
168 let claims = data.claims;
169
170 assert_eq!(claims.iss, api_key);
171 assert_eq!(claims.sub, user.to_string());
172 assert_eq!(claims.video.room.as_deref(), Some(room.as_str()));
173 assert_eq!(claims.video.room_join, Some(true));
174 assert!(claims.exp > claims.iat);
175 }
176
177 #[test]
178 fn token_rejects_wrong_secret() {
179 let api_key = "devkey";
180 let user = Uuid::now_v7();
181 let token = mint_access_token(api_key, "secret-a", user, "room-x", 60).unwrap();
182 let key = DecodingKey::from_secret(b"secret-b");
183 let mut validation = Validation::new(Algorithm::HS256);
184 let res = decode::<LiveTokenClaims>(&token, &key, &validation);
185 assert!(res.is_err(), "wrong secret must not verify");
186 }
187
188 #[test]
189 fn empty_secret_is_not_configured() {
190 let err = mint_access_token("", "", Uuid::now_v7(), "room", 60).unwrap_err();
191 assert!(matches!(err, LiveError::NotConfigured(_)));
192 }
193}
Added chat/live/types.rs chat/live/types.rs +68 −0 View chat/live/types.rs
@@ -0,0 +1,68 @@
1use serde::{Deserialize, Serialize};
2use storagelayer::MeetStatus;
3use time::OffsetDateTime;
4use uuid::Uuid;
5
6#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
7pub struct MeetInfo {
8 pub id: Uuid,
9 pub channel_id: Uuid,
10 pub room_name: String,
11 pub status: MeetStatus,
12 pub created_by: Uuid,
13 pub started_at: Option<OffsetDateTime>,
14 pub ended_at: Option<OffsetDateTime>,
15 pub created_at: OffsetDateTime,
16}
17
18impl From<storagelayer::Meet> for MeetInfo {
19 fn from(m: storagelayer::Meet) -> Self {
20 Self {
21 id: m.id,
22 channel_id: m.channel_id,
23 room_name: m.room_name,
24 status: m.status,
25 created_by: m.created_by,
26 started_at: m.started_at,
27 ended_at: m.ended_at,
28 created_at: m.created_at,
29 }
30 }
31}
32
33#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
34pub struct ParticipantRecord {
35 pub id: Uuid,
36 pub meet_id: Uuid,
37 pub user_id: Uuid,
38 pub joined_at: OffsetDateTime,
39 pub left_at: Option<OffsetDateTime>,
40 pub leave_reason: Option<storagelayer::LeaveReason>,
41}
42
43impl From<storagelayer::MeetParticipant> for ParticipantRecord {
44 fn from(p: storagelayer::MeetParticipant) -> Self {
45 Self {
46 id: p.id,
47 meet_id: p.meet_id,
48 user_id: p.user_id,
49 joined_at: p.joined_at,
50 left_at: p.left_at,
51 leave_reason: p.leave_reason,
52 }
53 }
54}
55
56#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
57pub struct JoinInfo {
58 pub meet: MeetInfo,
59 pub access_token: String,
60 pub room_url: String,
61 pub participants: Vec<ParticipantRecord>,
62}
63
64#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
65pub struct MeetPresence {
66 pub meet: MeetInfo,
67 pub participants: Vec<ParticipantRecord>,
68}
Added chat/live/webhook.rs chat/live/webhook.rs +279 −0 View chat/live/webhook.rs
@@ -0,0 +1,279 @@
1use axum::Json;
2use axum::body::Bytes;
3use axum::extract::State;
4use axum::http::HeaderMap;
5use jsonwebtoken::{Algorithm, DecodingKey, Validation, decode};
6use serde::Deserialize;
7use sha2::{Digest, Sha256};
8use time::OffsetDateTime;
9use uuid::Uuid;
10
11use crate::error::{LiveError, LiveResult};
12use crate::service::LiveService;
13
14const AUTH_HEADER: &str = "authorization";
15
16#[derive(Debug, Clone, Deserialize)]
17pub struct LiveKitWebhook {
18 pub event: String,
19 #[serde(default)]
20 pub room: Option<WebhookRoom>,
21 #[serde(default)]
22 pub participant: Option<WebhookParticipant>,
23}
24
25#[derive(Debug, Clone, Deserialize)]
26pub struct WebhookRoom {
27 pub sid: String,
28 pub name: String,
29 #[serde(default)]
30 pub empty_timeout: u64,
31}
32
33#[derive(Debug, Clone, Deserialize)]
34pub struct WebhookParticipant {
35 pub sid: String,
36 pub identity: String,
37 #[serde(default)]
38 pub disconnect_reason: Option<String>,
39}
40
41#[derive(Debug, Clone, Deserialize)]
42struct WebhookClaims {
43 #[allow(dead_code)]
44 #[serde(rename = "exp")]
45 exp: i64,
46 #[serde(rename = "nh")]
47 nh: String,
48}
49
50pub fn verify_webhook(headers: &HeaderMap, body: &[u8], webhook_secret: &str) -> LiveResult<()> {
51 if webhook_secret.is_empty() {
52 return Err(LiveError::WebhookVerification(
53 "webhook secret is empty".into(),
54 ));
55 }
56 let auth = headers
57 .get(AUTH_HEADER)
58 .and_then(|v| v.to_str().ok())
59 .ok_or_else(|| LiveError::WebhookVerification("missing authorization header".into()))?;
60
61 let digest = hex::encode(Sha256::digest(body));
62 let mut validation = Validation::new(Algorithm::HS256);
63 validation.validate_exp = true;
64 let key = DecodingKey::from_secret(webhook_secret.as_bytes());
65 let claims: WebhookClaims = decode(auth, &key, &validation)
66 .map_err(|e| LiveError::WebhookVerification(format!("bad jwt: {e}")))?
67 .claims;
68 if claims.nh != digest {
69 return Err(LiveError::WebhookVerification(
70 "body hash mismatch (nh claim)".into(),
71 ));
72 }
73 Ok(())
74}
75
76pub async fn handle_webhook(
77 State(service): State<LiveService>,
78 headers: HeaderMap,
79 body: Bytes,
80) -> Result<Json<serde_json::Value>, LiveError> {
81 service.handle_webhook(&headers, &body).await?;
82 Ok(Json(serde_json::json!({ "ok": true })))
83}
84
85pub fn channel_id_from_room_name(room_name: &str) -> LiveResult<Uuid> {
86 let channel = room_name
87 .split('_')
88 .next()
89 .ok_or_else(|| LiveError::BadRequest("room name has no channel prefix".into()))?;
90 Uuid::parse_str(channel)
91 .map_err(|e| LiveError::BadRequest(format!("invalid room name {room_name}: {e}")))
92}
93
94pub fn build_room_name(channel_id: Uuid) -> String {
95 let nonce = Uuid::now_v7().simple().to_string();
96 let nonce = &nonce[..8];
97 format!("{channel_id}_{nonce}")
98}
99
100pub fn parse_identity(identity: &str) -> LiveResult<Uuid> {
101 Uuid::parse_str(identity)
102 .map_err(|e| LiveError::BadRequest(format!("invalid participant identity: {e}")))
103}
104
105pub fn event_from_webhook(
106 webhook: &LiveKitWebhook,
107) -> LiveResult<Option<chatlayer::ChatLiveEvent>> {
108 use chatlayer::ChatLiveEvent;
109 let Some(room) = &webhook.room else {
110 return Ok(None);
111 };
112 let channel_id = channel_id_from_room_name(&room.name)?;
113 let ts = OffsetDateTime::now_utc();
114 match webhook.event.as_str() {
115 "room.started" => Ok(Some(ChatLiveEvent::MeetStarted {
116 channel_id,
117 room_name: room.name.clone(),
118 by: Uuid::nil(),
119 ts,
120 })),
121 "room.finished" => Ok(Some(ChatLiveEvent::MeetEnded {
122 channel_id,
123 room_name: room.name.clone(),
124 by: Uuid::nil(),
125 ts,
126 })),
127 "participant.joined" => {
128 let user_id = parse_identity(&participant_identity(webhook)?)?;
129 Ok(Some(ChatLiveEvent::ParticipantJoined {
130 channel_id,
131 room_name: room.name.clone(),
132 user_id,
133 ts,
134 }))
135 }
136 "participant.left" => {
137 let user_id = parse_identity(&participant_identity(webhook)?)?;
138 Ok(Some(ChatLiveEvent::ParticipantLeft {
139 channel_id,
140 room_name: room.name.clone(),
141 user_id,
142 reason: reason_from_disconnect(disconnect_reason(webhook)),
143 ts,
144 }))
145 }
146 _ => Ok(None),
147 }
148}
149
150fn participant_identity(webhook: &LiveKitWebhook) -> LiveResult<String> {
151 webhook
152 .participant
153 .as_ref()
154 .map(|p| p.identity.clone())
155 .ok_or_else(|| LiveError::BadRequest("webhook missing participant".into()))
156}
157
158fn disconnect_reason(webhook: &LiveKitWebhook) -> Option<String> {
159 webhook
160 .participant
161 .as_ref()
162 .and_then(|p| p.disconnect_reason.clone())
163}
164
165pub fn reason_from_disconnect(reason: Option<String>) -> storagelayer::LeaveReason {
166 use storagelayer::LeaveReason;
167 match reason.as_deref() {
168 Some("client_initiated") | Some("unknown") | None => LeaveReason::Left,
169 Some("duplicate_identity") | Some("server_shutdown") => LeaveReason::Kicked,
170 Some("participant_removed") => LeaveReason::Removed,
171 Some("room_closed") => LeaveReason::RoomEnded,
172 Some("network_error") | Some("disconnected") => LeaveReason::NetworkDrop,
173 Some(_) => LeaveReason::Left,
174 }
175}
176
177#[cfg(test)]
178mod tests {
179 use super::*;
180
181 #[test]
182 fn reason_mapping_covers_livekit_values() {
183 assert_eq!(
184 reason_from_disconnect(None),
185 storagelayer::LeaveReason::Left
186 );
187 assert_eq!(
188 reason_from_disconnect(Some("client_initiated".into())),
189 storagelayer::LeaveReason::Left
190 );
191 assert_eq!(
192 reason_from_disconnect(Some("participant_removed".into())),
193 storagelayer::LeaveReason::Removed
194 );
195 assert_eq!(
196 reason_from_disconnect(Some("room_closed".into())),
197 storagelayer::LeaveReason::RoomEnded
198 );
199 assert_eq!(
200 reason_from_disconnect(Some("network_error".into())),
201 storagelayer::LeaveReason::NetworkDrop
202 );
203 }
204
205 #[test]
206 fn webhook_signature_verifies_with_matching_secret() {
207 let secret = "wh-secret";
208 let body = br#"{"event":"room.started","room":{"name":"room-1"}}"#;
209 let digest = hex::encode(Sha256::digest(body));
210
211 let claims = serde_json::json!({
212 "iss": "wh-key",
213 "sub": "room-1",
214 "nh": digest,
215 "exp": OffsetDateTime::now_utc().unix_timestamp() + 300,
216 });
217 let token = jsonwebtoken::encode(
218 &jsonwebtoken::Header::new(Algorithm::HS256),
219 &claims,
220 &jsonwebtoken::EncodingKey::from_secret(secret.as_bytes()),
221 )
222 .unwrap();
223
224 let mut headers = HeaderMap::new();
225 headers.insert(AUTH_HEADER, token.parse().unwrap());
226 assert!(verify_webhook(&headers, body, secret).is_ok());
227 }
228
229 #[test]
230 fn webhook_signature_rejects_tampered_body() {
231 let secret = "wh-secret";
232 let body = br#"{"event":"room.started"}"#;
233 let tampered = br#"{"event":"room.finished"}"#;
234 let digest = hex::encode(Sha256::digest(body));
235 let claims = serde_json::json!({
236 "iss": "wh-key", "sub": "room-1", "nh": digest,
237 "exp": OffsetDateTime::now_utc().unix_timestamp() + 300,
238 });
239 let token = jsonwebtoken::encode(
240 &jsonwebtoken::Header::new(Algorithm::HS256),
241 &claims,
242 &jsonwebtoken::EncodingKey::from_secret(secret.as_bytes()),
243 )
244 .unwrap();
245 let mut headers = HeaderMap::new();
246 headers.insert(AUTH_HEADER, token.parse().unwrap());
247 assert!(verify_webhook(&headers, tampered, secret).is_err());
248 }
249
250 #[test]
251 fn event_mapping_for_join_and_left() {
252 let channel = Uuid::now_v7();
253 let joined = LiveKitWebhook {
254 event: "participant.joined".into(),
255 room: Some(WebhookRoom {
256 sid: "rm_1".into(),
257 name: build_room_name(channel),
258 empty_timeout: 0,
259 }),
260 participant: Some(WebhookParticipant {
261 sid: "PA_1".into(),
262 identity: Uuid::now_v7().to_string(),
263 disconnect_reason: None,
264 }),
265 };
266 let evt = event_from_webhook(&joined).unwrap().unwrap();
267 assert!(matches!(
268 evt,
269 chatlayer::ChatLiveEvent::ParticipantJoined { .. }
270 ));
271 }
272
273 #[test]
274 fn room_name_roundtrip_recovers_channel() {
275 let channel = Uuid::now_v7();
276 let name = build_room_name(channel);
277 assert_eq!(channel_id_from_room_name(&name).unwrap(), channel);
278 }
279}