livekit_api/public_api/
mod.rs1#[allow(clippy::all, dead_code, unused_mut, clippy::needless_lifetimes)]
21mod generated;
22
23use std::time::Duration;
24
25pub(crate) use generated::{types, Client};
26use livekit_token::{AccessToken, AgentGrants, AgentThreadGrants};
27use progenitor_client::{ClientHooks, OperationInfo};
28use reqwest::header::{HeaderValue, AUTHORIZATION};
29use serde::Deserialize;
30
31use crate::services::{ServerError, ServerErrorCode, ServiceBase, ServiceError, ServiceResult};
32
33pub(crate) const PROJECT: &str = "p_";
35
36const TOKEN_TTL: Duration = Duration::from_secs(60);
37const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
38const STAGING_URL: &str = "https://beta-api.staging.livekit.cloud";
39const PRODUCTION_URL: &str = "https://beta-api.livekit.cloud";
40
41pub fn public_api_url(project_url: &str) -> String {
44 let http = if let Some(rest) = project_url.strip_prefix("wss://") {
45 format!("https://{rest}")
46 } else if let Some(rest) = project_url.strip_prefix("ws://") {
47 format!("http://{rest}")
48 } else {
49 project_url.to_owned()
50 };
51 let hostname = url::Url::parse(&http).ok().and_then(|u| u.host_str().map(str::to_owned));
52 match hostname.as_deref() {
53 Some(h) if h.ends_with(".staging.livekit.cloud") => STAGING_URL.to_owned(),
54 Some(h) if h.ends_with(".livekit.cloud") => PRODUCTION_URL.to_owned(),
55 _ => http.trim_end_matches('/').to_owned(),
56 }
57}
58
59#[derive(Clone, Debug)]
61pub struct Credentials {
62 base: ServiceBase,
63 pub(crate) timeout: Duration,
64}
65
66impl Credentials {
67 fn bearer(&self) -> Result<String, livekit_token::AccessTokenError> {
68 let token = match &self.base {
69 ServiceBase::Token(token) => token.clone(),
70 ServiceBase::ApiKeySecret { api_key, api_secret } => {
71 AccessToken::with_api_key(api_key, api_secret)
72 .with_ttl(TOKEN_TTL)
73 .with_agent_grants(AgentGrants { database_admin: true, ..Default::default() })
74 .with_agent_thread_grants(AgentThreadGrants {
75 all: true,
76 list: true,
77 create: true,
78 write: true,
79 read: true,
80 delete: true,
81 ..Default::default()
82 })
83 .to_jwt()?
84 }
85 };
86 Ok(format!("Bearer {token}"))
87 }
88}
89
90impl ClientHooks<Credentials> for Client {
91 async fn pre<E>(
92 &self,
93 request: &mut reqwest::Request,
94 _: &OperationInfo,
95 ) -> Result<(), progenitor_client::Error<E>> {
96 let bearer =
97 self.inner.bearer().map_err(|e| progenitor_client::Error::Custom(e.to_string()))?;
98 let value = HeaderValue::from_str(&bearer)
99 .map_err(|e| progenitor_client::Error::Custom(e.to_string()))?;
100 request.headers_mut().insert(AUTHORIZATION, value);
101 *request.timeout_mut() = Some(self.inner.timeout);
102 Ok(())
103 }
104}
105
106pub(crate) fn client(host: &str, base: ServiceBase, http: crate::http_client::Client) -> Client {
107 Client::new_with_client(
108 &public_api_url(host),
109 http,
110 Credentials { base, timeout: DEFAULT_TIMEOUT },
111 )
112}
113
114pub(crate) async fn error(err: progenitor_client::Error<()>) -> ServiceError {
116 use progenitor_client::Error as E;
117 let code = |code: &str, msg: String| ServerErrorCode {
118 code: code.to_owned(),
119 msg,
120 meta: Default::default(),
121 };
122 let err = match err {
123 E::CommunicationError(e) | E::ResponseBodyError(e) | E::InvalidUpgrade(e) => {
124 ServerError::Request(e)
125 }
126 E::UnexpectedResponse(resp) => {
127 #[derive(Deserialize)]
128 struct Body {
129 #[serde(default)]
130 code: serde_json::Value,
131 #[serde(default)]
132 message: String,
133 }
134 let status = resp.status();
135 let text = resp.text().await.unwrap_or_default();
136 let body = serde_json::from_str::<Body>(&text).ok();
137 let named = body.as_ref().and_then(|b| b.code.as_str()).map(str::to_owned);
138 let msg = body.map(|b| b.message).filter(|m| !m.is_empty()).unwrap_or(text);
139 let fallback = match status.as_u16() {
140 400 => ServerErrorCode::INVALID_ARGUMENT,
141 401 => ServerErrorCode::UNAUTHENTICATED,
142 403 => ServerErrorCode::PERMISSION_DENIED,
143 404 => ServerErrorCode::NOT_FOUND,
144 409 => ServerErrorCode::ALREADY_EXISTS,
145 429 => ServerErrorCode::RESOURCE_EXHAUSTED,
146 501 => ServerErrorCode::UNIMPLEMENTED,
147 503 => ServerErrorCode::UNAVAILABLE,
148 504 => ServerErrorCode::DEADLINE_EXCEEDED,
149 s if s >= 500 => ServerErrorCode::INTERNAL,
150 _ => ServerErrorCode::UNKNOWN,
151 };
152 ServerError::Twirp(code(named.as_deref().unwrap_or(fallback), msg))
153 }
154 other => ServerError::Twirp(code(ServerErrorCode::INTERNAL, other.to_string())),
155 };
156 ServiceError::Twirp(err)
157}
158
159pub(crate) async fn call<T>(
161 fut: impl std::future::Future<
162 Output = Result<progenitor_client::ResponseValue<T>, progenitor_client::Error<()>>,
163 >,
164) -> ServiceResult<T> {
165 match fut.await {
166 Ok(v) => Ok(v.into_inner()),
167 Err(e) => Err(error(e).await),
168 }
169}
170
171pub(crate) fn opt(s: &str) -> Option<&str> {
173 (!s.is_empty()).then_some(s)
174}
175
176pub(crate) fn page_size(limit: i32) -> Option<i32> {
177 (limit > 0).then_some(limit)
178}
179
180pub(crate) fn next_cursor(page: Option<types::LivekitPublicapiCommonV1PageInfo>) -> String {
181 page.and_then(|p| p.next_cursor).unwrap_or_default()
182}
183
184#[cfg(test)]
185mod tests {
186 use super::public_api_url;
187
188 #[test]
189 fn public_api_url_from_project_url() {
190 let staging = "https://beta-api.staging.livekit.cloud";
191 assert_eq!(public_api_url("wss://my-app-abc.staging.livekit.cloud"), staging);
192 assert_eq!(public_api_url("https://my-app-abc.staging.livekit.cloud/"), staging);
193 assert_eq!(
194 public_api_url("wss://my-app-abc.livekit.cloud"),
195 "https://beta-api.livekit.cloud"
196 );
197 assert_eq!(public_api_url("ws://localhost:7880"), "http://localhost:7880");
198 assert_eq!(public_api_url("http://127.0.0.1:8080/"), "http://127.0.0.1:8080");
199 assert_eq!(public_api_url("https://livekit.example.com"), "https://livekit.example.com");
200 }
201}