tenant.openagents/omega

1pub mod proto;
2pub mod token;
3
4use anyhow::Result;
5use async_trait::async_trait;
6use prost::Message;
7use reqwest::header::CONTENT_TYPE;
8use std::{future::Future, sync::Arc, time::Duration};
9
10#[async_trait]
11pub trait Client: Send + Sync {
12    fn url(&self) -> &str;
13    async fn create_room(&self, name: String) -> Result<()>;
14    async fn delete_room(&self, name: String) -> Result<()>;
15    async fn remove_participant(&self, room: String, identity: String) -> Result<()>;
16    async fn update_participant(
17        &self,
18        room: String,
19        identity: String,
20        permission: proto::ParticipantPermission,
21    ) -> Result<()>;
22    fn room_token(&self, room: &str, identity: &str) -> Result<String>;
23    fn guest_token(&self, room: &str, identity: &str) -> Result<String>;
24}
25
26pub struct LiveKitParticipantUpdate {}
27
28#[derive(Clone)]
29pub struct LiveKitClient {
30    http: reqwest::Client,
31    url: Arc<str>,
32    key: Arc<str>,
33    secret: Arc<str>,
34    timestamp_source: Arc<dyn token::UnixTimestampSource>,
35}
36
37impl LiveKitClient {
38    pub fn new(url: String, key: String, secret: String) -> Self {
39        Self::new_with_timestamp_source(
40            url,
41            key,
42            secret,
43            Arc::new(token::SystemUnixTimestampSource),
44        )
45    }
46
47    pub(crate) fn new_with_timestamp_source(
48        mut url: String,
49        key: String,
50        secret: String,
51        timestamp_source: Arc<dyn token::UnixTimestampSource>,
52    ) -> Self {
53        if url.ends_with('/') {
54            url.pop();
55        }
56
57        Self {
58            http: reqwest::ClientBuilder::new()
59                .timeout(Duration::from_secs(5))
60                .build()
61                .unwrap(),
62            url: url.into(),
63            key: key.into(),
64            secret: secret.into(),
65            timestamp_source,
66        }
67    }
68
69    fn request<Req, Res>(
70        &self,
71        path: &str,
72        grant: token::VideoGrant,
73        body: Req,
74    ) -> impl Future<Output = Result<Res>>
75    where
76        Req: Message,
77        Res: Default + Message,
78    {
79        let client = self.http.clone();
80        let token = token::create_with_timestamp_source(
81            &self.key,
82            &self.secret,
83            None,
84            grant,
85            self.timestamp_source.as_ref(),
86        );
87        let url = format!("{}/{}", self.url, path);
88        log::info!("Request {}: {:?}", url, body);
89        async move {
90            let token = token?;
91            let response = client
92                .post(&url)
93                .header(CONTENT_TYPE, "application/protobuf")
94                .bearer_auth(token)
95                .body(body.encode_to_vec())
96                .send()
97                .await?;
98
99            if response.status().is_success() {
100                log::info!("Response {}: {:?}", url, response.status());
101                Ok(Res::decode(response.bytes().await?)?)
102            } else {
103                log::error!("Response {}: {:?}", url, response.status());
104                anyhow::bail!(
105                    "POST {} failed with status code {:?}, {:?}",
106                    url,
107                    response.status(),
108                    response.text().await
109                );
110            }
111        }
112    }
113}
114
115#[async_trait]
116impl Client for LiveKitClient {
117    fn url(&self) -> &str {
118        &self.url
119    }
120
121    async fn create_room(&self, name: String) -> Result<()> {
122        let _: proto::Room = self
123            .request(
124                "twirp/livekit.RoomService/CreateRoom",
125                token::VideoGrant {
126                    room_create: Some(true),
127                    ..Default::default()
128                },
129                proto::CreateRoomRequest {
130                    name,
131                    ..Default::default()
132                },
133            )
134            .await?;
135        Ok(())
136    }
137
138    async fn delete_room(&self, name: String) -> Result<()> {
139        let _: proto::DeleteRoomResponse = self
140            .request(
141                "twirp/livekit.RoomService/DeleteRoom",
142                token::VideoGrant {
143                    room_create: Some(true),
144                    ..Default::default()
145                },
146                proto::DeleteRoomRequest { room: name },
147            )
148            .await?;
149        Ok(())
150    }
151
152    async fn remove_participant(&self, room: String, identity: String) -> Result<()> {
153        let _: proto::RemoveParticipantResponse = self
154            .request(
155                "twirp/livekit.RoomService/RemoveParticipant",
156                token::VideoGrant::to_admin(&room),
157                proto::RoomParticipantIdentity {
158                    room: room.clone(),
159                    identity,
160                },
161            )
162            .await?;
163        Ok(())
164    }
165
166    async fn update_participant(
167        &self,
168        room: String,
169        identity: String,
170        permission: proto::ParticipantPermission,
171    ) -> Result<()> {
172        let _: proto::ParticipantInfo = self
173            .request(
174                "twirp/livekit.RoomService/UpdateParticipant",
175                token::VideoGrant::to_admin(&room),
176                proto::UpdateParticipantRequest {
177                    room: room.clone(),
178                    identity,
179                    metadata: "".to_string(),
180                    permission: Some(permission),
181                },
182            )
183            .await?;
184        Ok(())
185    }
186
187    fn room_token(&self, room: &str, identity: &str) -> Result<String> {
188        token::create_with_timestamp_source(
189            &self.key,
190            &self.secret,
191            Some(identity),
192            token::VideoGrant::to_join(room),
193            self.timestamp_source.as_ref(),
194        )
195    }
196
197    fn guest_token(&self, room: &str, identity: &str) -> Result<String> {
198        token::create_with_timestamp_source(
199            &self.key,
200            &self.secret,
201            Some(identity),
202            token::VideoGrant::for_guest(room),
203            self.timestamp_source.as_ref(),
204        )
205    }
206}
207