feat: Add supporting atproto headers

atrium-api/src/agent/bluesky.rs

@@ -0,0 +1,19 @@
+//! Bluesky specific constants.
+
+/// DID of the bluesky labeler service.
+pub const CHAT_BSKY_DID: &str = "did:web:api.bsky.chat";
+
+/// Supported proxy targets, which includes the bluesky specific services.
+pub enum AtprotoServiceType {
+    AtprotoLabeler,
+    BskyChat,
+}
+
+impl AsRef<str> for AtprotoServiceType {
+    fn as_ref(&self) -> &str {
+        match self {
+            Self::AtprotoLabeler => "atproto_labeler",
+            Self::BskyChat => "bsky_chat",
+        }
+    }
+}

atrium-api/src/agent/inner.rs

@@ -1,5 +1,6 @@
 use super::{Session, SessionStore};
 use crate::did_doc::DidDocument;
+use crate::types::string::Did;
 use async_trait::async_trait;
 use atrium_xrpc::error::{Error, Result, XrpcErrorKind};
 use atrium_xrpc::{HttpClient, OutputDataOrBytes, XrpcClient, XrpcRequest};
@@ -8,16 +9,48 @@ use serde::{de::DeserializeOwned, Serialize};
 use std::sync::{Arc, RwLock};
 use tokio::sync::{Mutex, Notify};
 
-const REFRESH_SESSION: &str = "com.atproto.server.refreshSession";
-
-struct SessionStoreClient<S, T> {
+struct WrapperClient<S, T> {
     store: Arc<Store<S>>,
-    inner: T,
+    proxy_header: RwLock<Option<String>>,
+    labelers_header: Arc<RwLock<Option<Vec<String>>>>,
+    inner: Arc<T>,
+}
+
+impl<S, T> WrapperClient<S, T> {
+    fn configure_proxy_header(&self, value: String) {
+        self.proxy_header
+            .write()
+            .expect("failed to write proxy header")
+            .replace(value);
+    }
+    fn configure_labelers_header(&self, labelers_dids: Option<Vec<Did>>) {
+        *self
+            .labelers_header
+            .write()
+            .expect("failed to write labelers header") =
+            labelers_dids.map(|dids| dids.iter().map(|did| did.as_ref().into()).collect())
+    }
+}
+
+impl<S, T> Clone for WrapperClient<S, T> {
+    fn clone(&self) -> Self {
+        Self {
+            store: self.store.clone(),
+            labelers_header: self.labelers_header.clone(),
+            proxy_header: RwLock::new(
+                self.proxy_header
+                    .read()
+                    .expect("failed to read proxy header")
+                    .clone(),
+            ),
+            inner: self.inner.clone(),
+        }
+    }
 }
 
 #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
 #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
-impl<S, T> HttpClient for SessionStoreClient<S, T>
+impl<S, T> HttpClient for WrapperClient<S, T>
 where
     S: Send + Sync,
     T: HttpClient + Send + Sync,
@@ -33,15 +66,15 @@ where
 
 #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
 #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
-impl<S, T> XrpcClient for SessionStoreClient<S, T>
+impl<S, T> XrpcClient for WrapperClient<S, T>
 where
     S: SessionStore + Send + Sync,
     T: XrpcClient + Send + Sync,
 {
     fn base_uri(&self) -> String {
         self.store.get_endpoint()
     }
-    async fn auth(&self, is_refresh: bool) -> Option<String> {
+    async fn authentication_token(&self, is_refresh: bool) -> Option<String> {
         self.store.get_session().await.map(|session| {
             if is_refresh {
                 session.refresh_jwt
@@ -50,13 +83,25 @@ where
             }
         })
     }
+    async fn atproto_proxy_header(&self) -> Option<String> {
+        self.proxy_header
+            .read()
+            .expect("failed to read proxy header")
+            .clone()
+    }
+    async fn atproto_accept_labelers_header(&self) -> Option<Vec<String>> {
+        self.labelers_header
+            .read()
+            .expect("failed to read labelers header")
+            .clone()
+    }
 }
 
 pub struct Client<S, T> {
     store: Arc<Store<S>>,
-    inner: SessionStoreClient<S, T>,
-    is_refreshing: Mutex<bool>,
-    notify: Notify,
+    inner: WrapperClient<S, T>,
+    is_refreshing: Arc<Mutex<bool>>,
+    notify: Arc<Notify>,
 }
 
 impl<S, T> Client<S, T>
@@ -65,17 +110,33 @@ where
     T: XrpcClient + Send + Sync,
 {
     pub(crate) fn new(store: Arc<Store<S>>, xrpc: T) -> Self {
-        let inner = SessionStoreClient {
+        let inner = WrapperClient {
             store: Arc::clone(&store),
-            inner: xrpc,
+            labelers_header: Arc::new(RwLock::new(None)),
+            proxy_header: RwLock::new(None),
+            inner: Arc::new(xrpc),
         };
         Self {
             store,
             inner,
-            is_refreshing: Mutex::new(false),
-            notify: Notify::new(),
+            is_refreshing: Arc::new(Mutex::new(false)),
+            notify: Arc::new(Notify::new()),
         }
     }
+    pub(crate) fn configure_proxy_header(&self, did: Did, service_type: impl AsRef<str>) {
+        self.inner
+            .configure_proxy_header(format!("{}#{}", did.as_ref(), service_type.as_ref()));
+    }
+    pub(crate) fn clone_with_proxy(&self, did: Did, service_type: impl AsRef<str>) -> Self {
+        let cloned = self.clone();
+        cloned
+            .inner
+            .configure_proxy_header(format!("{}#{}", did.as_ref(), service_type.as_ref()));
+        cloned
+    }
+    pub(crate) fn configure_labelers_header(&self, labeler_dids: Option<Vec<Did>>) {
+        self.inner.configure_labelers_header(labeler_dids);
+    }
     // Internal helper to refresh sessions
     // - Wraps the actual implementation to ensure only one refresh is attempted at a time.
     async fn refresh_session(&self) {
@@ -120,7 +181,7 @@ where
             .inner
             .send_xrpc::<(), (), _, _>(&XrpcRequest {
                 method: Method::POST,
-                path: REFRESH_SESSION.into(),
+                nsid: crate::com::atproto::server::refresh_session::NSID.into(),
                 parameters: None,
                 input: None,
                 encoding: None,
@@ -147,6 +208,21 @@ where
     }
 }
 
+impl<S, T> Clone for Client<S, T>
+where
+    S: SessionStore + Send + Sync,
+    T: XrpcClient + Send + Sync,
+{
+    fn clone(&self) -> Self {
+        Self {
+            store: self.store.clone(),
+            inner: self.inner.clone(),
+            is_refreshing: self.is_refreshing.clone(),
+            notify: self.notify.clone(),
+        }
+    }
+}
+
 #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
 #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
 impl<S, T> HttpClient for Client<S, T>