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