objectstore_client/
client.rs

1use std::io;
2use std::sync::Arc;
3use std::time::Duration;
4
5use bytes::Bytes;
6use futures_util::stream::BoxStream;
7use objectstore_types::metadata::{Compression, ExpirationPolicy};
8use objectstore_types::scope;
9use reqwest::RequestBuilder;
10use url::Url;
11
12use crate::IntoTokenProvider;
13use crate::auth::{TokenProvider, TokenRequest};
14
15const USER_AGENT: &str = concat!("objectstore-client/", env!("CARGO_PKG_VERSION"));
16
17#[derive(Debug)]
18struct ClientBuilderInner {
19    service_url: Url,
20    propagate_traces: bool,
21    reqwest_builder: reqwest::ClientBuilder,
22    token: Option<TokenProvider>,
23}
24
25impl ClientBuilderInner {
26    /// Applies defaults that cannot be overridden by the caller.
27    fn apply_defaults(mut self) -> Self {
28        self.reqwest_builder = self
29            .reqwest_builder
30            // hickory-dns: Controlled by the `reqwest/hickory-dns` feature flag
31            // we are dealing with de/compression ourselves:
32            .no_brotli()
33            .no_deflate()
34            .no_gzip()
35            .no_zstd();
36        self
37    }
38}
39
40/// Builder to create a [`Client`].
41#[must_use = "call .build() on this ClientBuilder to create a Client"]
42#[derive(Debug)]
43pub struct ClientBuilder(crate::Result<ClientBuilderInner>);
44
45impl ClientBuilder {
46    /// Creates a new [`ClientBuilder`], configured with the given `service_url`.
47    ///
48    /// To perform CRUD operations, one has to create a [`Client`], and then scope it to a [`Usecase`]
49    /// and Scope in order to create a [`Session`].
50    pub fn new(service_url: impl reqwest::IntoUrl) -> Self {
51        let service_url = match service_url.into_url() {
52            Ok(url) => url,
53            Err(err) => return Self(Err(err.into())),
54        };
55        if service_url.cannot_be_a_base() {
56            return ClientBuilder(Err(crate::Error::InvalidUrl {
57                message: "service_url cannot be a base".to_owned(),
58            }));
59        }
60
61        let reqwest_builder = reqwest::Client::builder()
62            // We define just a connection timeout by default but do not limit reads. A connect
63            // timeout of 100ms is still very conservative, but should provide a sensible upper
64            // bound to expected request latencies.
65            .connect_timeout(Duration::from_millis(100))
66            .user_agent(USER_AGENT);
67
68        Self(Ok(ClientBuilderInner {
69            service_url,
70            propagate_traces: false,
71            reqwest_builder,
72            token: None,
73        }))
74    }
75
76    /// Changes whether the `sentry-trace` header will be sent to Objectstore
77    /// to take advantage of Sentry's distributed tracing.
78    ///
79    /// By default, tracing headers will not be propagated.
80    pub fn propagate_traces(mut self, propagate_traces: bool) -> Self {
81        if let Ok(ref mut inner) = self.0 {
82            inner.propagate_traces = propagate_traces;
83        }
84        self
85    }
86
87    /// Defines a read timeout for the [`reqwest::Client`].
88    ///
89    /// The read timeout is defined to be "between consecutive read operations", for example between
90    /// chunks of a streaming response. For more fine-grained configuration of this and other
91    /// timeouts, use [`Self::configure_reqwest`].
92    ///
93    /// By default, no read timeout and a connect timeout of 100ms is set.
94    pub fn timeout(self, timeout: Duration) -> Self {
95        let Ok(mut inner) = self.0 else { return self };
96        inner.reqwest_builder = inner.reqwest_builder.read_timeout(timeout);
97        Self(Ok(inner))
98    }
99
100    /// Calls the closure with the underlying [`reqwest::ClientBuilder`].
101    ///
102    /// By default, the ClientBuilder is configured to create a reqwest Client with a connect and read timeout of 500ms and a user agent identifying this library.
103    pub fn configure_reqwest<F>(self, closure: F) -> Self
104    where
105        F: FnOnce(reqwest::ClientBuilder) -> reqwest::ClientBuilder,
106    {
107        let Ok(mut inner) = self.0 else { return self };
108        inner.reqwest_builder = closure(inner.reqwest_builder);
109        Self(Ok(inner))
110    }
111
112    /// Sets the authentication token to use for requests to Objectstore.
113    ///
114    /// Accepts anything that implements [`IntoTokenProvider`]:
115    /// - A [`TokenGenerator`](crate::TokenGenerator) — for internal services that have access to
116    ///   an EdDSA keypair. The generator signs a fresh JWT for each request.
117    /// - A `String` or `&str` — a pre-signed JWT, used as-is for every request.
118    /// - An `Option` of any of the above — a `None` leaves the client unauthenticated, which is
119    ///   convenient when authentication is configured conditionally.
120    pub fn token(self, token: impl IntoTokenProvider) -> Self {
121        let Ok(mut inner) = self.0 else { return self };
122        inner.token = token.into_token_provider();
123        Self(Ok(inner))
124    }
125
126    /// Returns a [`Client`] that uses this [`ClientBuilder`] configuration.
127    ///
128    /// # Errors
129    ///
130    /// This method fails if:
131    /// - the given `service_url` is invalid or cannot be used as a base URL
132    /// - the [`reqwest::Client`] fails to build. Refer to [`reqwest::ClientBuilder::build`] for
133    ///   more information on when this can happen.
134    pub fn build(self) -> crate::Result<Client> {
135        let inner = self.0?.apply_defaults();
136
137        Ok(Client {
138            inner: Arc::new(ClientInner {
139                reqwest: inner.reqwest_builder.build()?,
140                service_url: inner.service_url,
141                propagate_traces: inner.propagate_traces,
142                token: inner.token,
143            }),
144        })
145    }
146}
147
148/// An identifier for a workload in Objectstore, along with defaults to use for all
149/// operations within that Usecase.
150///
151/// Usecases need to be statically defined in Objectstore's configuration server-side.
152/// Objectstore can make decisions based on the Usecase. For example, choosing the most
153/// suitable storage backend.
154#[derive(Debug, Clone)]
155pub struct Usecase {
156    name: Arc<str>,
157    compression: Option<Compression>,
158    expiration_policy: ExpirationPolicy,
159}
160
161impl Usecase {
162    /// Creates a new Usecase.
163    pub fn new(name: &str) -> Self {
164        Self {
165            name: name.into(),
166            compression: Some(Compression::Zstd),
167            expiration_policy: ExpirationPolicy::default(),
168        }
169    }
170
171    /// Returns the name of this usecase.
172    #[inline]
173    pub fn name(&self) -> &str {
174        &self.name
175    }
176
177    /// Returns the compression algorithm to use for operations within this usecase.
178    #[inline]
179    pub fn compression(&self) -> Option<Compression> {
180        self.compression
181    }
182
183    /// Sets the compression algorithm to use for operations within this usecase.
184    ///
185    /// It's still possible to override this default on each operation's builder.
186    ///
187    /// By default, [`Compression::Zstd`] is used. Pass [`None`] to disable compression.
188    pub fn with_compression(self, compression: impl Into<Option<Compression>>) -> Self {
189        Self {
190            compression: compression.into(),
191            ..self
192        }
193    }
194
195    /// Returns the expiration policy to use by default for operations within this usecase.
196    #[inline]
197    pub fn expiration_policy(&self) -> ExpirationPolicy {
198        self.expiration_policy
199    }
200
201    /// Sets the expiration policy to use for operations within this usecase.
202    ///
203    /// It's still possible to override this default on each operation's builder.
204    ///
205    /// By default, [`ExpirationPolicy::Manual`] is used, meaning that objects won't automatically
206    /// expire.
207    pub fn with_expiration_policy(self, expiration_policy: ExpirationPolicy) -> Self {
208        Self {
209            expiration_policy,
210            ..self
211        }
212    }
213
214    /// Creates a new custom [`Scope`].
215    ///
216    /// Add parts to it using [`Scope::push`].
217    ///
218    /// Generally, [`Usecase::for_organization`] and [`Usecase::for_project`] should fit most usecases,
219    /// so prefer using those methods rather than creating your own custom [`Scope`].
220    pub fn scope(&self) -> Scope {
221        Scope::new(self.clone())
222    }
223
224    /// Creates a new [`Scope`] tied to the given organization.
225    pub fn for_organization(&self, organization: u64) -> Scope {
226        Scope::for_organization(self.clone(), organization)
227    }
228
229    /// Creates a new [`Scope`] tied to the given organization and project.
230    pub fn for_project(&self, organization: u64, project: u64) -> Scope {
231        Scope::for_project(self.clone(), organization, project)
232    }
233}
234
235#[derive(Debug)]
236pub(crate) struct ScopeInner {
237    usecase: Usecase,
238    scopes: scope::Scopes,
239}
240
241impl ScopeInner {
242    #[inline]
243    pub(crate) fn usecase(&self) -> &Usecase {
244        &self.usecase
245    }
246
247    #[inline]
248    pub(crate) fn scopes(&self) -> &scope::Scopes {
249        &self.scopes
250    }
251}
252
253/// A [`Scope`] is a sequence of key-value pairs that defines a (possibly nested) namespace within a
254/// [`Usecase`].
255///
256/// To construct a [`Scope`], use [`Usecase::for_organization`], [`Usecase::for_project`], or
257/// [`Usecase::scope`] for custom scopes.
258#[derive(Debug)]
259pub struct Scope(pub(crate) crate::Result<ScopeInner>);
260
261impl Scope {
262    /// Creates a new root-level Scope for the given usecase.
263    ///
264    /// Using a custom Scope is discouraged, prefer using [`Usecase::for_organization`] or [`Usecase::for_project`] instead.
265    pub fn new(usecase: Usecase) -> Self {
266        Self(Ok(ScopeInner {
267            usecase,
268            scopes: scope::Scopes::empty(),
269        }))
270    }
271
272    fn for_organization(usecase: Usecase, organization: u64) -> Self {
273        Self::new(usecase).push("org", organization)
274    }
275
276    fn for_project(usecase: Usecase, organization: u64, project: u64) -> Self {
277        Self::for_organization(usecase, organization).push("project", project)
278    }
279
280    /// Extends this Scope by creating a new sub-scope nested within it.
281    pub fn push<V>(self, key: &str, value: V) -> Self
282    where
283        V: std::fmt::Display,
284    {
285        let result = self.0.and_then(|mut inner| {
286            inner.scopes.push(key, value)?;
287            Ok(inner)
288        });
289
290        Self(result)
291    }
292
293    /// Creates a session for this scope using the given client.
294    ///
295    /// # Errors
296    ///
297    /// Returns an error if the scope is invalid (e.g. it contains invalid characters).
298    pub fn session(self, client: &Client) -> crate::Result<Session> {
299        client.session(self)
300    }
301}
302
303#[derive(Debug)]
304pub(crate) struct ClientInner {
305    reqwest: reqwest::Client,
306    service_url: Url,
307    propagate_traces: bool,
308    token: Option<TokenProvider>,
309}
310
311/// A client for Objectstore. Use [`Client::builder`] to configure and construct a Client.
312///
313/// To perform CRUD operations, one has to create a Client, and then scope it to a [`Usecase`]
314/// and Scope in order to create a [`Session`].
315///
316/// If your Objectstore instance enforces authorization checks, you must provide
317/// authentication via [`ClientBuilder::token`]. It accepts anything that converts
318/// into a [`TokenProvider`]:
319///
320/// - **[`TokenGenerator`](crate::TokenGenerator)** — for internal services that have access to
321///   an EdDSA keypair. The generator signs a fresh JWT for each request, scoped to the
322///   specific usecase and scope being accessed.
323/// - **`String` / `&str`** — a pre-signed JWT, used as-is for every request.
324///   Use this for external services that receive a token from another source.
325///
326/// # Examples
327///
328/// Internal service with a keypair:
329///
330/// ```no_run
331/// use std::time::Duration;
332/// use objectstore_client::{Client, SecretKey, TokenGenerator, Usecase};
333/// use objectstore_types::auth::Permission;
334///
335/// # async fn example() -> objectstore_client::Result<()> {
336/// let token_generator = TokenGenerator::new(SecretKey {
337///         secret_key: "<safely inject secret key>".into(),
338///         kid: "my-service".into(),
339///     })?
340///     .expiry_seconds(30)
341///     .permissions(&[Permission::ObjectRead]);
342///
343/// let client = Client::builder("http://localhost:8888/")
344///     .timeout(Duration::from_secs(1))
345///     .propagate_traces(true)
346///     .token(token_generator)
347///     .build()?;
348/// # Ok(())
349/// # }
350/// ```
351///
352/// External service with a pre-signed JWT (obtained via
353/// [`TokenGenerator::create_token`](crate::TokenGenerator::create_token)):
354///
355/// ```no_run
356/// use objectstore_client::{Client, SecretKey, TokenGenerator, Usecase};
357///
358/// # fn example() -> objectstore_client::Result<()> {
359/// let scope = Usecase::new("my_app").for_project(42, 1337);
360/// let token = TokenGenerator::new(SecretKey {
361///     secret_key: "<private key>".into(),
362///     kid: "my-service".into(),
363/// })?.create_token(&scope).sign()?;
364///
365/// let client = Client::builder("http://localhost:8888/")
366///     .token(token)
367///     .build()?;
368/// # Ok(())
369/// # }
370/// ```
371///
372/// Optional authentication — pass an `Option` straight through, leaving the client
373/// unauthenticated when it is `None`:
374///
375/// ```no_run
376/// use objectstore_client::Client;
377///
378/// # fn example() -> objectstore_client::Result<()> {
379/// // Authenticate only if a token is present in the environment.
380/// let token_opt = std::env::var("OBJECTSTORE_TOKEN").ok();
381///
382/// let client = Client::builder("http://localhost:8888/")
383///     .token(token_opt)
384///     .build()?;
385/// # Ok(())
386/// # }
387/// ```
388#[derive(Debug, Clone)]
389pub struct Client {
390    inner: Arc<ClientInner>,
391}
392
393impl Client {
394    /// Creates a new [`Client`], configured with the given `service_url` and default
395    /// configuration.
396    ///
397    /// Use [`Client::builder`] for more fine-grained configuration.
398    ///
399    /// # Errors
400    ///
401    /// This method fails if [`ClientBuilder::build`] fails.
402    pub fn new(service_url: impl reqwest::IntoUrl) -> crate::Result<Client> {
403        ClientBuilder::new(service_url).build()
404    }
405
406    /// Convenience function to create a [`ClientBuilder`].
407    pub fn builder(service_url: impl reqwest::IntoUrl) -> ClientBuilder {
408        ClientBuilder::new(service_url)
409    }
410
411    /// Creates a session for the given scope using this client.
412    ///
413    /// # Errors
414    ///
415    /// Returns an error if the scope is invalid (e.g. it contains invalid characters).
416    pub fn session(&self, scope: Scope) -> crate::Result<Session> {
417        scope.0.map(|inner| Session {
418            scope: inner.into(),
419            client: self.inner.clone(),
420        })
421    }
422}
423
424/// Represents a session with Objectstore, tied to a specific Usecase and Scope within it.
425///
426/// Create a Session using [`Client::session`] or [`Scope::session`].
427#[derive(Debug, Clone)]
428pub struct Session {
429    pub(crate) scope: Arc<ScopeInner>,
430    pub(crate) client: Arc<ClientInner>,
431}
432
433/// The type of [`Stream`](futures_util::Stream) to be used for a PUT request.
434pub type ClientStream = BoxStream<'static, io::Result<Bytes>>;
435
436impl Session {
437    /// Generates a GET url to the object with the given `key`.
438    ///
439    /// This can then be used by downstream services to fetch the given object.
440    /// NOTE however that the service does not strictly follow HTTP semantics,
441    /// in particular in relation to `Accept-Encoding`.
442    pub fn object_url(&self, object_key: &str) -> Url {
443        let mut url = self.client.service_url.clone();
444
445        // `path_segments_mut` can only error if the url is cannot-be-a-base,
446        // and we check that in `ClientBuilder::new`, therefore this will never panic.
447        let mut segments = url.path_segments_mut().unwrap();
448        segments
449            .push("v1")
450            .push("objects")
451            .push(&self.scope.usecase.name)
452            .push(&self.scope.scopes.as_api_path().to_string())
453            .extend(object_key.split('/'));
454        drop(segments);
455
456        url
457    }
458
459    /// Returns a token authorizing access to this session's scope.
460    ///
461    /// With a [`TokenGenerator`](crate::TokenGenerator), each call signs a fresh token whose expiry
462    /// starts now, using the configured permissions; use [`create_token`](Self::create_token) to
463    /// narrow either.
464    ///
465    /// If a pre-signed token was used to construct the client, it is returned verbatim, including
466    /// its original expiry.
467    ///
468    /// Returns `None` if the client was constructed without authentication.
469    ///
470    /// # Errors
471    ///
472    /// Returns an error if a new token cannot be signed with the configured key. Retrieving a
473    /// pre-signed token never fails.
474    pub fn get_token(&self) -> crate::Result<Option<String>> {
475        match self.client.token {
476            Some(TokenProvider::Generator(ref generator)) => {
477                generator.request_inner(&self.scope).sign().map(Some)
478            }
479            Some(TokenProvider::Static(ref token)) => Ok(Some(token.clone())),
480            None => Ok(None),
481        }
482    }
483
484    /// Deprecated. Use [`get_token`](Self::get_token) instead.
485    #[deprecated(note = "Use `get_token` instead.")]
486    pub fn mint_token(&self) -> crate::Result<Option<String>> {
487        self.get_token()
488    }
489
490    /// Creates a new token with configurable permissions and expiry.
491    ///
492    /// Use this to produce a static token that can be handed to an external service
493    /// with lower permissions than the current session.
494    ///
495    /// # Errors
496    ///
497    /// This requires a [`TokenGenerator`](crate::TokenGenerator) to sign a new token.
498    /// If not available, this will result in an error. To retrieve a token with the current
499    /// session's default permissions and expiry, use [`get_token`](Self::get_token) instead.
500    pub fn create_token(&self) -> crate::Result<Option<TokenRequest<'_>>> {
501        match self.client.token {
502            Some(TokenProvider::Generator(ref generator)) => {
503                Ok(Some(generator.request_inner(&self.scope)))
504            }
505            Some(TokenProvider::Static(_)) => Err(crate::Error::StaticTokenOverride),
506            None => Ok(None),
507        }
508    }
509
510    fn batch_url(&self) -> Url {
511        let mut url = self.client.service_url.clone();
512
513        // `path_segments_mut` can only error if the url is cannot-be-a-base,
514        // and we check that in `ClientBuilder::new`, therefore this will never panic.
515        let mut segments = url.path_segments_mut().unwrap();
516        segments
517            .push("v1")
518            .push("objects:batch")
519            .push(self.scope.usecase().name())
520            .push(&self.scope.scopes.as_api_path().to_string())
521            .push(""); // trailing slash
522        drop(segments);
523
524        url
525    }
526
527    #[cfg(feature = "multipart")]
528    fn multipart_url(
529        &self,
530        suffix: Option<&'static str>,
531        object_key: Option<&str>,
532        query_pairs: Option<Vec<(&str, String)>>,
533    ) -> Url {
534        let mut url = self.client.service_url.clone();
535
536        // `path_segments_mut` can only error if the url is cannot-be-a-base,
537        // and we check that in `ClientBuilder::new`, therefore this will never panic.
538        let mut segments = url.path_segments_mut().unwrap();
539        segments
540            .push("v1")
541            .push(match suffix {
542                Some("parts") => "objects:multipart:parts",
543                Some("complete") => "objects:multipart:complete",
544                _ => "objects:multipart",
545            })
546            .push(&self.scope.usecase.name)
547            .push(&self.scope.scopes.as_api_path().to_string());
548        if let Some(object_key) = object_key.filter(|key| !key.is_empty()) {
549            segments.extend(object_key.split('/'));
550        }
551        drop(segments);
552        if let Some(query_pairs) = query_pairs {
553            let mut pairs = url.query_pairs_mut();
554            for (key, value) in query_pairs {
555                pairs.append_pair(key, &value);
556            }
557        }
558
559        url
560    }
561
562    fn prepare_builder(&self, mut builder: RequestBuilder) -> crate::Result<RequestBuilder> {
563        if let Some(token) = self.get_token()? {
564            builder = builder.header("x-os-auth", format!("Bearer {token}"));
565        }
566        if self.client.propagate_traces {
567            let trace_headers =
568                sentry_core::configure_scope(|scope| Some(scope.iter_trace_propagation_headers()));
569            for (header_name, value) in trace_headers.into_iter().flatten() {
570                builder = builder.header(header_name, value);
571            }
572        }
573        Ok(builder)
574    }
575
576    pub(crate) fn request(
577        &self,
578        method: reqwest::Method,
579        object_key: &str,
580    ) -> crate::Result<RequestBuilder> {
581        let url = self.object_url(object_key);
582        let builder = self.client.reqwest.request(method, url);
583        self.prepare_builder(builder)
584    }
585
586    pub(crate) fn batch_request(&self) -> crate::Result<RequestBuilder> {
587        let url = self.batch_url();
588        let builder = self.client.reqwest.post(url);
589        self.prepare_builder(builder)
590    }
591
592    #[cfg(feature = "multipart")]
593    pub(crate) fn multipart_request(
594        &self,
595        method: reqwest::Method,
596        action: Option<&'static str>,
597        object_key: Option<&str>,
598        query_pairs: Option<Vec<(&str, String)>>,
599    ) -> crate::Result<RequestBuilder> {
600        let url = self.multipart_url(action, object_key, query_pairs);
601        let builder = self.client.reqwest.request(method, url);
602        self.prepare_builder(builder)
603    }
604}
605
606#[cfg(test)]
607mod tests {
608    use super::*;
609
610    #[test]
611    fn test_object_url() {
612        let client = Client::new("http://127.0.0.1:8888/").unwrap();
613        let usecase = Usecase::new("testing");
614        let scope = usecase
615            .for_project(12345, 1337)
616            .push("app_slug", "email_app");
617        let session = client.session(scope).unwrap();
618
619        assert_eq!(
620            session.object_url("foo/bar").to_string(),
621            "http://127.0.0.1:8888/v1/objects/testing/org=12345;project=1337;app_slug=email_app/foo/bar"
622        )
623    }
624
625    #[test]
626    fn test_object_url_with_base_path() {
627        let client = Client::new("http://127.0.0.1:8888/api/prefix").unwrap();
628        let usecase = Usecase::new("testing");
629        let scope = usecase.for_project(12345, 1337);
630        let session = client.session(scope).unwrap();
631
632        assert_eq!(
633            session.object_url("foo/bar").to_string(),
634            "http://127.0.0.1:8888/api/prefix/v1/objects/testing/org=12345;project=1337/foo/bar"
635        )
636    }
637}
objectstore_client/client.rs

objectstore_client/
client.rs
