cargo fix

src/auth.rs

···
 use axum::{extract::FromRequestParts, http::StatusCode};
 use base64::Engine;
 
-use crate::{AppState, Error, auth, error::ErrorMessage};
+use crate::{AppState, Error, error::ErrorMessage};
 
 /// This is an axum request extractor that represents an authenticated user.
 ///
 /// If specified in an API endpoint, this will guarantee that the API can only be called
 /// by an authenticated user.
-pub struct AuthenticatedUser {
+pub(crate) struct AuthenticatedUser {
     did: String,
 }
 
 impl AuthenticatedUser {
-    pub fn did(&self) -> String {
+    pub(crate) fn did(&self) -> String {
         self.did.clone()
     }
 }
 
 impl FromRequestParts<AppState> for AuthenticatedUser {
-    type Rejection = crate::Error;
+    type Rejection = Error;
 
     async fn from_request_parts(
         parts: &mut axum::http::request::Parts,
         state: &AppState,
-    ) -> std::result::Result<Self, Self::Rejection> {
+    ) -> Result<Self, Self::Rejection> {
         let token = parts
             .headers
             .get(axum::http::header::AUTHORIZATION)
···
 
         // N.B: We ignore all fields inside of the token up until this point because they can be
         // attacker-controlled.
-        let (typ, claims) = auth::verify(&state.signing_key.did(), token).map_err(|e| {
+        let (typ, claims) = verify(&state.signing_key.did(), token).map_err(|e| {
             Error::with_status(
                 StatusCode::UNAUTHORIZED,
                 e.context("failed to verify auth token"),
···
 }
 
 /// Cryptographically sign a JSON web token with the specified key.
-pub fn sign(
+pub(crate) fn sign(
     key: &Secp256k1Keypair,
     typ: &str,
     claims: serde_json::Value,
···
 }
 
 /// Cryptographically verify a JSON web token's validity using the specified public key.
-pub fn verify(key: &str, token: &str) -> anyhow::Result<(String, serde_json::Value)> {
+pub(crate) fn verify(key: &str, token: &str) -> anyhow::Result<(String, serde_json::Value)> {
     let mut parts = token.splitn(3, '.');
     let hdr = parts.next().context("no header")?;
     let claims = parts.next().context("no claims")?;

src/config.rs

···
 use serde::Deserialize;
 use url::Url;
 
-pub mod metrics {
+pub(crate) mod metrics {
     use super::*;
 
     #[derive(Deserialize, Debug, Clone)]
-    pub struct PrometheusConfig {
+    pub(crate) struct PrometheusConfig {
         /// The URL of the Prometheus server's exporter endpoint.
         pub url: Url,
     }
···
 
 #[derive(Deserialize, Debug, Clone)]
 #[serde(tag = "type")]
-pub enum MetricConfig {
+pub(crate) enum MetricConfig {
     PrometheusPush(metrics::PrometheusConfig),
 }
 
 #[derive(Deserialize, Debug, Clone)]
-pub struct FirehoseConfig {
+pub(crate) struct FirehoseConfig {
     /// A list of upstream relays that this PDS will try to reach out to.
     pub relays: Vec<Url>,
 }
 
 #[derive(Deserialize, Debug, Clone)]
-pub struct RepoConfig {
+pub(crate) struct RepoConfig {
     /// The path to the repository storage.
     pub path: PathBuf,
 }
 
 #[derive(Deserialize, Debug, Clone)]
-pub struct PlcConfig {
+pub(crate) struct PlcConfig {
     /// The path to the local PLC cache.
     pub path: PathBuf,
 }
 
 #[derive(Deserialize, Debug, Clone)]
-pub struct BlobConfig {
+pub(crate) struct BlobConfig {
     /// The path to store blobs into.
     pub path: PathBuf,
     /// The maximum size limit of blobs.
···
 }
 
 #[derive(Deserialize, Debug, Clone)]
-pub struct AppConfig {
+pub(crate) struct AppConfig {
     /// The primary signing keys for all PLC/DID operations.
     pub key: PathBuf,
     /// The hostname of the PDS. Typically a domain name.

src/did.rs

···
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
-pub struct DidVerificationMethod {
+pub(crate) struct DidVerificationMethod {
     pub id: String,
     #[serde(rename = "type")]
     pub ty: String,
···
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
-pub struct DidService {
+pub(crate) struct DidService {
     pub id: String,
     #[serde(rename = "type")]
     pub ty: String,
···
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
-pub struct DidDocument {
+pub(crate) struct DidDocument {
     #[serde(rename = "@context", skip_serializing_if = "Vec::is_empty")]
     pub context: Vec<Url>,
     pub id: Did,
···
 }
 
 /// Resolve a DID document using the specified reqwest client.
-pub async fn resolve(client: &Client, did: Did) -> Result<DidDocument> {
+pub(crate) async fn resolve(client: &Client, did: Did) -> Result<DidDocument> {
     let url = match did.method() {
         "did:web" => {
             // N.B: This is a potentially hostile operation, so we are only going to allow

src/endpoints/identity.rs

···
 }
 
 #[rustfmt::skip]
-pub fn routes() -> Router<AppState> {
+pub(super) fn routes() -> Router<AppState> {
     // AP /xrpc/com.atproto.identity.updateHandle
     // AP /xrpc/com.atproto.identity.requestPlcOperationSignature
     // AP /xrpc/com.atproto.identity.signPlcOperation

src/endpoints/mod.rs

···
 mod server;
 mod sync;
 
-pub async fn health() -> Result<Json<serde_json::Value>> {
+pub(crate) async fn health() -> Result<Json<serde_json::Value>> {
     Ok(Json(json!({
         "version": "bluepds"
     })))
 }
 
-pub fn routes() -> Router<AppState> {
+pub(crate) fn routes() -> Router<AppState> {
     Router::new()
         .route("/_health", get(health))
         .merge(identity::routes()) // com.atproto.identity

src/endpoints/repo.rs

···
 
         // The swap failed. Return the old commit and do not update the repository.
         return Ok(Json(
-            repo::apply_writes::OutputData {
+            apply_writes::OutputData {
                 results: None,
                 commit: Some(
                     CommitMetaData {
···
     .await;
 
     Ok(Json(
-        repo::apply_writes::OutputData {
+        apply_writes::OutputData {
             results: Some(res),
             commit: Some(
                 CommitMetaData {
···
 
 #[derive(serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
-pub struct ListRecordsParameters {
+pub(super) struct ListRecordsParameters {
     ///The NSID of the record type.
-    pub collection: atrium_api::types::string::Nsid,
+    pub collection: Nsid,
     #[serde(skip_serializing_if = "core::option::Option::is_none")]
-    pub cursor: core::option::Option<String>,
+    pub cursor: Option<String>,
     ///The number of records to return.
     #[serde(skip_serializing_if = "core::option::Option::is_none")]
-    pub limit: core::option::Option<String>,
+    pub limit: Option<String>,
     ///The handle or DID of the repo.
-    pub repo: atrium_api::types::string::AtIdentifier,
+    pub repo: AtIdentifier,
     ///Flag to reverse the order of the returned records.
     #[serde(skip_serializing_if = "core::option::Option::is_none")]
-    pub reverse: core::option::Option<bool>,
+    pub reverse: Option<bool>,
     ///DEPRECATED: The highest sort-ordered rkey to stop at (exclusive)
     #[serde(skip_serializing_if = "core::option::Option::is_none")]
-    pub rkey_end: core::option::Option<String>,
+    pub rkey_end: Option<String>,
     ///DEPRECATED: The lowest sort-ordered rkey to start from (exclusive)
     #[serde(skip_serializing_if = "core::option::Option::is_none")]
-    pub rkey_start: core::option::Option<String>,
+    pub rkey_start: Option<String>,
 }
 
 async fn list_records(
···
     drop(file);
     let hash = sha.finalize();
 
-    let cid = atrium_repo::Cid::new_v1(
+    let cid = Cid::new_v1(
         IPLD_RAW,
         atrium_repo::Multihash::wrap(IPLD_MH_SHA2_256, hash.as_slice()).unwrap(),
     );
···
 }
 
 #[rustfmt::skip]
-pub fn routes() -> Router<AppState> {
+pub(super) fn routes() -> Router<AppState> {
     // AP /xrpc/com.atproto.repo.applyWrites
     // AP /xrpc/com.atproto.repo.createRecord
     // AP /xrpc/com.atproto.repo.putRecord

src/endpoints/server.rs

···
         // SEC: Call argon2's `verify_password` to simulate password verification and discard the result.
         // We do this to avoid exposing a timing attack where attackers can measure the response time to
         // determine whether or not an account exists.
-        let _ = argon2::Argon2::default().verify_password(
+        let _ = Argon2::default().verify_password(
             password.as_bytes(),
             &PasswordHash::new(DUMMY_PASSWORD).unwrap(),
         );
···
         ));
     };
 
-    match argon2::Argon2::default().verify_password(
+    match Argon2::default().verify_password(
         password.as_bytes(),
         &PasswordHash::new(account.password.as_str()).context("invalid password hash in db")?,
     ) {
···
             refresh_jwt: refresh_token,
 
             active: Some(active), // TODO?
-            did: atrium_api::types::string::Did::new(did.to_string()).unwrap(),
+            did: Did::new(did.to_string()).unwrap(),
             did_doc: None,
-            handle: atrium_api::types::string::Handle::new(user.handle).unwrap(),
+            handle: Handle::new(user.handle).unwrap(),
             status,
         }
         .into(),
···
 }
 
 #[rustfmt::skip]
-pub fn routes() -> Router<AppState> {
+pub(super) fn routes() -> Router<AppState> {
     // UG /xrpc/com.atproto.server.describeServer
     // UP /xrpc/com.atproto.server.createAccount
     // UP /xrpc/com.atproto.server.createSession

src/endpoints/sync.rs

···
 // HACK: `limit` may be passed as a string, so we must treat it as one.
 #[derive(serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
-pub struct ListBlobsParameters {
+pub(super) struct ListBlobsParameters {
     #[serde(skip_serializing_if = "core::option::Option::is_none")]
-    pub cursor: core::option::Option<String>,
+    pub cursor: Option<String>,
     ///The DID of the repo.
-    pub did: atrium_api::types::string::Did,
+    pub did: Did,
     #[serde(skip_serializing_if = "core::option::Option::is_none")]
-    pub limit: core::option::Option<String>,
+    pub limit: Option<String>,
     ///Optional revision of the repo to list blobs since.
     #[serde(skip_serializing_if = "core::option::Option::is_none")]
-    pub since: core::option::Option<String>,
+    pub since: Option<String>,
 }
 
 async fn list_blobs(
···
     let cids = cids
         .into_iter()
         .map(|c| {
-            atrium_repo::Cid::from_str(&c)
+            Cid::from_str(&c)
                 .map(atrium_api::types::string::Cid::new)
                 .map_err(anyhow::Error::new)
         })
···
 // HACK: `limit` may be passed as a string, so we must treat it as one.
 #[derive(serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
-pub struct ListReposParameters {
+pub(super) struct ListReposParameters {
     #[serde(skip_serializing_if = "core::option::Option::is_none")]
-    pub cursor: core::option::Option<String>,
+    pub cursor: Option<String>,
     #[serde(skip_serializing_if = "core::option::Option::is_none")]
-    pub limit: core::option::Option<String>,
+    pub limit: Option<String>,
 }
 
 async fn list_repos(
···
 // HACK: `cursor` may be passed as a string, so we must treat it as one.
 #[derive(serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
-pub struct SubscribeReposParametersData {
+pub(super) struct SubscribeReposParametersData {
     ///The last known event seq number to backfill from.
     #[serde(skip_serializing_if = "core::option::Option::is_none")]
-    pub cursor: core::option::Option<String>,
+    pub cursor: Option<String>,
 }
 
 async fn subscribe_repos(
···
 }
 
 #[rustfmt::skip]
-pub fn routes() -> axum::Router<AppState> {
+pub(super) fn routes() -> Router<AppState> {
     // UG /xrpc/com.atproto.sync.getBlob
     // UG /xrpc/com.atproto.sync.getBlocks
     // UG /xrpc/com.atproto.sync.getLatestCommit

src/error.rs

···
 }
 
 impl IntoResponse for Error {
-    fn into_response(self) -> axum::response::Response {
+    fn into_response(self) -> Response {
         error!("{:?}", self.err);
 
         // N.B: Forward out the error message to the requester if this is a debug build.

src/firehose.rs

···
 
 enum FirehoseMessage {
     Broadcast(sync::subscribe_repos::Message),
-    Connect(Box<(axum::extract::ws::WebSocket, Option<i64>)>),
+    Connect(Box<(WebSocket, Option<i64>)>),
 }
 
 enum FrameHeader {
···
 
 /// A firehose producer. This is used to transmit messages to the firehose for broadcast.
 #[derive(Clone, Debug)]
-pub struct FirehoseProducer {
+pub(crate) struct FirehoseProducer {
     tx: tokio::sync::mpsc::Sender<FirehoseMessage>,
 }
 
 impl FirehoseProducer {
     /// Broadcast an `#account` event.
-    pub async fn account(&self, account: impl Into<sync::subscribe_repos::Account>) {
+    pub(crate) async fn account(&self, account: impl Into<sync::subscribe_repos::Account>) {
         let _ = self
             .tx
             .send(FirehoseMessage::Broadcast(
···
     }
 
     /// Broadcast an `#identity` event.
-    pub async fn identity(&self, identity: impl Into<sync::subscribe_repos::Identity>) {
+    pub(crate) async fn identity(&self, identity: impl Into<sync::subscribe_repos::Identity>) {
         let _ = self
             .tx
             .send(FirehoseMessage::Broadcast(
···
     }
 
     /// Broadcast a `#commit` event.
-    pub async fn commit(&self, commit: impl Into<sync::subscribe_repos::Commit>) {
+    pub(crate) async fn commit(&self, commit: impl Into<sync::subscribe_repos::Commit>) {
         let _ = self
             .tx
             .send(FirehoseMessage::Broadcast(
···
             .await;
     }
 
-    pub async fn client_connection(&self, ws: WebSocket, cursor: Option<i64>) {
+    pub(crate) async fn client_connection(&self, ws: WebSocket, cursor: Option<i64>) {
         let _ = self
             .tx
             .send(FirehoseMessage::Connect(Box::new((ws, cursor))))
···
     seq: u64,
     history: &VecDeque<(u64, &str, sync::subscribe_repos::Message)>,
     cursor: Option<i64>,
-) -> anyhow::Result<WebSocket> {
+) -> Result<WebSocket> {
     if let Some(cursor) = cursor {
         let mut frame = Vec::new();
         let cursor = cursor as u64;
···
     Ok(ws)
 }
 
-pub async fn reconnect_relays(client: &Client, config: &AppConfig) {
+pub(crate) async fn reconnect_relays(client: &Client, config: &AppConfig) {
     // Avoid connecting to upstream relays in test mode.
     if config.test {
         return;
···
 /// This will broadcast all updates in this PDS out to anyone who is listening.
 ///
 /// Reference: https://atproto.com/specs/sync
-pub async fn spawn(
+pub(crate) async fn spawn(
     client: Client,
     config: AppConfig,
 ) -> (tokio::task::JoinHandle<()>, FirehoseProducer) {

src/main.rs

···
 mod plc;
 mod storage;
 
-pub type Result<T> = std::result::Result<T, error::Error>;
+pub type Result<T> = std::result::Result<T, Error>;
 pub use error::Error;
 use uuid::Uuid;
 
 pub type Client = reqwest_middleware::ClientWithMiddleware;
-pub type Db = sqlx::SqlitePool;
+pub type Db = SqlitePool;
 pub type Cred = Arc<dyn TokenCredential>;
 
 pub const APP_USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"),);
···
     }
 
     #[rustfmt::skip]
-    pub fn routes() -> Router<AppState> {
+    pub(crate) fn routes() -> Router<AppState> {
         // AP /xrpc/app.bsky.actor.putPreferences
         // AG /xrpc/app.bsky.actor.getPreferences
         Router::new()
···
     let r = client
         .request(request.method().clone(), url)
         .headers(h)
-        .header(axum::http::header::AUTHORIZATION, format!("Bearer {token}"))
+        .header(http::header::AUTHORIZATION, format!("Bearer {token}"))
         .body(reqwest::Body::wrap_stream(
             request.into_body().into_data_stream(),
         ))

src/metrics.rs

···
 
 use crate::config;
 
-pub const AUTH_FAILED: &str = "bluepds.auth.failed"; // Counter.
+pub(crate) const AUTH_FAILED: &str = "bluepds.auth.failed"; // Counter.
 
-pub const FIREHOSE_HISTORY: &str = "bluepds.firehose.history"; // Gauge.
-pub const FIREHOSE_LISTENERS: &str = "bluepds.firehose.listeners"; // Gauge.
-pub const FIREHOSE_MESSAGES: &str = "bluepds.firehose.messages"; // Counter.
-pub const FIREHOSE_SEQUENCE: &str = "bluepds.firehose.sequence"; // Counter.
+pub(crate) const FIREHOSE_HISTORY: &str = "bluepds.firehose.history"; // Gauge.
+pub(crate) const FIREHOSE_LISTENERS: &str = "bluepds.firehose.listeners"; // Gauge.
+pub(crate) const FIREHOSE_MESSAGES: &str = "bluepds.firehose.messages"; // Counter.
+pub(crate) const FIREHOSE_SEQUENCE: &str = "bluepds.firehose.sequence"; // Counter.
 
-pub const REPO_COMMITS: &str = "bluepds.repo.commits"; // Counter.
-pub const REPO_OP_CREATE: &str = "bluepds.repo.op.create"; // Counter.
-pub const REPO_OP_UPDATE: &str = "bluepds.repo.op.update"; // Counter.
-pub const REPO_OP_DELETE: &str = "bluepds.repo.op.delete"; // Counter.
+pub(crate) const REPO_COMMITS: &str = "bluepds.repo.commits"; // Counter.
+pub(crate) const REPO_OP_CREATE: &str = "bluepds.repo.op.create"; // Counter.
+pub(crate) const REPO_OP_UPDATE: &str = "bluepds.repo.op.update"; // Counter.
+pub(crate) const REPO_OP_DELETE: &str = "bluepds.repo.op.delete"; // Counter.
 
 /// Must be ran exactly once on startup. This will declare all of the instruments for `metrics`.
-pub fn setup(config: &Option<config::MetricConfig>) -> anyhow::Result<()> {
+pub(crate) fn setup(config: &Option<config::MetricConfig>) -> anyhow::Result<()> {
     describe_counter!(AUTH_FAILED, "The number of failed authentication attempts.");
 
     describe_gauge!(FIREHOSE_HISTORY, "The size of the firehose history buffer.");

src/plc.rs

···
 
 #[derive(Debug, Deserialize, Serialize, Clone)]
 #[serde(rename_all = "camelCase", tag = "type")]
-pub enum PlcService {
+pub(crate) enum PlcService {
     #[serde(rename = "AtprotoPersonalDataServer")]
     Pds { endpoint: String },
 }
 
 #[derive(Debug, Deserialize, Serialize, Clone)]
 #[serde(rename_all = "camelCase")]
-pub struct PlcOperation {
+pub(crate) struct PlcOperation {
     #[serde(rename = "type")]
     pub typ: String,
     pub rotation_keys: Vec<String>,
···
 }
 
 impl PlcOperation {
-    pub fn sign(self, sig: Vec<u8>) -> SignedPlcOperation {
+    pub(crate) fn sign(self, sig: Vec<u8>) -> SignedPlcOperation {
         SignedPlcOperation {
             typ: self.typ,
             rotation_keys: self.rotation_keys,
···
 
 #[derive(Debug, Deserialize, Serialize, Clone)]
 #[serde(rename_all = "camelCase")]
-pub struct SignedPlcOperation {
+pub(crate) struct SignedPlcOperation {
     #[serde(rename = "type")]
     pub typ: String,
     pub rotation_keys: Vec<String>,
···
     pub sig: String,
 }
 
-pub async fn sign_op(rkey: &RotationKey, op: PlcOperation) -> anyhow::Result<SignedPlcOperation> {
+pub(crate) async fn sign_op(rkey: &RotationKey, op: PlcOperation) -> anyhow::Result<SignedPlcOperation> {
     let bytes = serde_ipld_dagcbor::to_vec(&op).context("failed to encode op")?;
     let bytes = rkey.sign(&bytes).context("failed to sign op")?;
 
···
 }
 
 /// Submit a PLC operation to the public directory.
-pub async fn submit(client: &Client, did: &str, op: &SignedPlcOperation) -> anyhow::Result<()> {
+pub(crate) async fn submit(client: &Client, did: &str, op: &SignedPlcOperation) -> anyhow::Result<()> {
     debug!("submitting {} {}", did, serde_json::to_string(&op).unwrap());
 
     let res = client

src/storage.rs

···
 
 use crate::{Db, config::RepoConfig};
 
-pub async fn open_store(
+pub(crate) async fn open_store(
     config: &RepoConfig,
     did: impl Into<String>,
 ) -> Result<impl AsyncBlockStoreRead + AsyncBlockStoreWrite> {
···
     CarStore::open(f).await.context("failed to open car store")
 }
 
-pub async fn open_repo_db(
+pub(crate) async fn open_repo_db(
     config: &RepoConfig,
     db: &Db,
     did: impl Into<String>,
···
     open_repo(config, did, Cid::from_str(&cid).unwrap()).await
 }
 
-pub async fn open_repo(
+pub(crate) async fn open_repo(
     config: &RepoConfig,
     did: impl Into<String>,
     cid: Cid,