reorganzie

.nix/flake.nix

···
           git
           nixd
           direnv
+          libpq
         ];
       overlays = [ (import rust-overlay) ];
       pkgs = import nixpkgs {
···
       nativeBuildInputs = with pkgs; [ rust pkg-config ];
     in
     with pkgs;
-    {      
+    {
       devShells.default = mkShell {
         inherit buildInputs nativeBuildInputs;
         LD_LIBRARY_PATH = nixpkgs.legacyPackages.x86_64-linux.lib.makeLibraryPath buildInputs;
···
         DATABASE_URL = "sqlite://data/sqlite.db";
       };
     });
-}
+}

deployment.bicep

···
-param webAppName string
-param location string = resourceGroup().location // Location for all resources
-
-param sku string = 'B1' // The SKU of App Service Plan
-param dockerContainerName string = '${webAppName}:latest'
-param repositoryUrl string = 'https://github.com/DrChat/bluepds'
-param branch string = 'main'
-param customDomain string
-
-@description('Redeploy hostnames without SSL binding. Just specify `true` if this is the first time you\'re deploying the app.')
-param redeployHostnamesHack bool = false
-
-var acrName = toLower('${webAppName}${uniqueString(resourceGroup().id)}')
-var aspName = toLower('${webAppName}-asp')
-var webName = toLower('${webAppName}${uniqueString(resourceGroup().id)}')
-var sanName = toLower('${webAppName}${uniqueString(resourceGroup().id)}')
-
-// resource appInsights 'Microsoft.OperationalInsights/workspaces@2023-09-01' = {
-//   name: '${webAppName}-ai'
-//   location: location
-//   properties: {
-//     publicNetworkAccessForIngestion: 'Enabled'
-//     workspaceCapping: {
-//       dailyQuotaGb: 1
-//     }
-//     sku: {
-//       name: 'Standalone'
-//     }
-//   }
-// }
-
-// resource appServicePlanDiagnostics 'Microsoft.Insights/diagnosticSettings@2021-05-01-preview' = {
-//   name: appServicePlan.name
-//   scope: appServicePlan
-//   properties: {
-//     workspaceId: appInsights.id
-//     metrics: [
-//       {
-//         category: 'AllMetrics'
-//         enabled: true
-//       }
-//     ]
-//   }
-// }
-
-resource appServicePlan 'Microsoft.Web/serverfarms@2020-06-01' = {
-  name: aspName
-  location: location
-  properties: {
-    reserved: true
-  }
-  sku: {
-    name: sku
-  }
-  kind: 'linux'
-}
-
-resource acrResource 'Microsoft.ContainerRegistry/registries@2023-01-01-preview' = {
-  name: acrName
-  location: location
-  sku: {
-    name: 'Basic'
-  }
-  properties: {
-    adminUserEnabled: false
-  }
-}
-
-resource appStorage 'Microsoft.Storage/storageAccounts@2023-05-01' = {
-  name: sanName
-  location: location
-  kind: 'StorageV2'
-  sku: {
-    name: 'Standard_LRS'
-  }
-}
-
-resource fileShare 'Microsoft.Storage/storageAccounts/fileServices/shares@2023-05-01' = {
-  name: '${appStorage.name}/default/data'
-  properties: {}
-}
-
-resource appService 'Microsoft.Web/sites@2020-06-01' = {
-  name: webName
-  location: location
-  identity: {
-    type: 'SystemAssigned'
-  }
-  properties: {
-    httpsOnly: true
-    serverFarmId: appServicePlan.id
-    siteConfig: {
-      // Sigh. This took _far_ too long to figure out.
-      // We must authenticate to ACR, as no credentials are set up by default
-      // (the Az CLI will implicitly set them up in the background)
-      acrUseManagedIdentityCreds: true
-      appSettings: [
-        {
-          name: 'BLUEPDS_HOST_NAME'
-          value: empty(customDomain) ? '${webName}.azurewebsites.net' : customDomain
-        }
-        {
-          name: 'BLUEPDS_TEST'
-          value: 'false'
-        }
-        {
-          name: 'WEBSITES_PORT'
-          value: '8000'
-        }
-      ]
-      linuxFxVersion: 'DOCKER|${acrName}.azurecr.io/${dockerContainerName}'
-    }
-  }
-}
-
-resource hostNameBinding 'Microsoft.Web/sites/hostNameBindings@2024-04-01' = if (redeployHostnamesHack) {
-  name: customDomain
-  parent: appService
-  properties: {
-    siteName: appService.name
-    hostNameType: 'Verified'
-    sslState: 'Disabled'
-  }
-}
-
-// This stupidity is required because Azure requires a circular dependency in order to define a custom hostname with SSL.
-// https://stackoverflow.com/questions/73077972/how-to-deploy-app-service-with-managed-ssl-certificate-using-arm
-module certificateBindings './deploymentBindingHack.bicep' = {
-  name: '${deployment().name}-ssl'
-  params: {
-    appServicePlanResourceId: appServicePlan.id
-    customHostnames: [customDomain]
-    location: location
-    webAppName: appService.name
-  }
-  dependsOn: [hostNameBinding]
-}
-
-resource appServiceStorageConfig 'Microsoft.Web/sites/config@2024-04-01' = {
-  name: 'azurestorageaccounts'
-  parent: appService
-  properties: {
-    data: {
-      type: 'AzureFiles'
-      shareName: 'data'
-      mountPath: '/app/data'
-      accountName: appStorage.name
-      // WTF? Where's the ability to mount storage via managed identity?
-      accessKey: appStorage.listKeys().keys[0].value
-    }
-  }
-}
-
-@description('This is the built-in AcrPull role. See https://docs.microsoft.com/azure/role-based-access-control/built-in-roles#acrpull')
-resource acrPullRoleDefinition 'Microsoft.Authorization/roleDefinitions@2018-01-01-preview' existing = {
-  scope: subscription()
-  name: '7f951dda-4ed3-4680-a7ca-43fe172d538d'
-}
-
-resource appServiceAcrPull 'Microsoft.Authorization/roleAssignments@2020-04-01-preview' = {
-  name: guid(resourceGroup().id, acrResource.id, appService.id, 'AssignAcrPullToAS')
-  scope: acrResource
-  properties: {
-    description: 'Assign AcrPull role to AS'
-    principalId: appService.identity.principalId
-    principalType: 'ServicePrincipal'
-    roleDefinitionId: acrPullRoleDefinition.id
-  }
-}
-
-resource srcControls 'Microsoft.Web/sites/sourcecontrols@2021-01-01' = {
-  name: 'web'
-  parent: appService
-  properties: {
-    repoUrl: repositoryUrl
-    branch: branch
-    isManualIntegration: true
-  }
-}
-
-output acr string = acrResource.name
-output domain string = appService.properties.hostNames[0]

deploymentBindingHack.bicep

···
-// https://stackoverflow.com/questions/73077972/how-to-deploy-app-service-with-managed-ssl-certificate-using-arm
-//
-// TLDR: Azure requires a circular dependency in order to define an app service with a custom domain with SSL enabled.
-// Terrific user experience. Really makes me love using Azure in my free time.
-param webAppName string
-param location string
-param appServicePlanResourceId string
-param customHostnames array
-
-// Managed certificates can only be created once the hostname is added to the web app.
-resource certificates 'Microsoft.Web/certificates@2022-03-01' = [for (fqdn, i) in customHostnames: {
-  name: '${fqdn}-${webAppName}'
-  location: location
-  properties: {
-    serverFarmId: appServicePlanResourceId
-    canonicalName: fqdn
-  }
-}]
-
-// sslState and thumbprint can only be set once the managed certificate is created
-@batchSize(1)
-resource customHostname 'Microsoft.web/sites/hostnameBindings@2019-08-01' = [for (fqdn, i) in customHostnames: {
-  name: '${webAppName}/${fqdn}'
-  properties: {
-    siteName: webAppName
-    hostNameType: 'Verified'
-    sslState: 'SniEnabled'
-    thumbprint: certificates[i].properties.thumbprint
-  }
-}]

flake.nix

···
             "rust-analyzer"
           ];
         }));
-        
+
         inherit (pkgs) lib;
         unfilteredRoot = ./.; # The original, unfiltered source
         src = lib.fileset.toSource {
···
             git
             nixd
             direnv
+            libpq
           ];
         };
       })
···
             };
         };
       });
-}
+}

src/account_manager/helpers/account.rs

···
 //! blacksky-algorithms/rsky is licensed under the Apache License 2.0
 //!
 //! Modified for SQLite backend
+use crate::schema::pds::account::dsl as AccountSchema;
+use crate::schema::pds::actor::dsl as ActorSchema;
 use anyhow::Result;
 use chrono::DateTime;
 use chrono::offset::Utc as UtcOffset;
···
     AccountStatus, ActorAccount, ActorJoinAccount, AvailabilityFlags, FormattedAccountStatus,
     GetAccountAdminStatusOutput, format_account_status,
 };
-use rsky_pds::schema::pds::account::dsl as AccountSchema;
-use rsky_pds::schema::pds::actor::dsl as ActorSchema;
 use std::ops::Add;
 use std::time::SystemTime;
 use thiserror::Error;
···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<()> {
-    use rsky_pds::schema::pds::email_token::dsl as EmailTokenSchema;
-    use rsky_pds::schema::pds::refresh_token::dsl as RefreshTokenSchema;
-    use rsky_pds::schema::pds::repo_root::dsl as RepoRootSchema;
+    use crate::schema::pds::email_token::dsl as EmailTokenSchema;
+    use crate::schema::pds::refresh_token::dsl as RefreshTokenSchema;
+    use crate::schema::pds::repo_root::dsl as RepoRootSchema;
 
     let did = did.to_owned();
     _ = db
···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<()> {
-    use rsky_pds::schema::pds::actor;
+    use crate::schema::pds::actor;
 
     let actor2 = diesel::alias!(actor as actor2);
 

src/account_manager/helpers/auth.rs

···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<()> {
-    use rsky_pds::schema::pds::refresh_token::dsl as RefreshTokenSchema;
+    use crate::schema::pds::refresh_token::dsl as RefreshTokenSchema;
 
     let exp = from_micros_to_utc((payload.exp.as_millis() / 1000) as i64);
 
···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<bool> {
-    use rsky_pds::schema::pds::refresh_token::dsl as RefreshTokenSchema;
+    use crate::schema::pds::refresh_token::dsl as RefreshTokenSchema;
     db.get()
         .await?
         .interact(move |conn| {
···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<bool> {
-    use rsky_pds::schema::pds::refresh_token::dsl as RefreshTokenSchema;
+    use crate::schema::pds::refresh_token::dsl as RefreshTokenSchema;
     let did = did.to_owned();
     db.get()
         .await?
···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<bool> {
-    use rsky_pds::schema::pds::refresh_token::dsl as RefreshTokenSchema;
+    use crate::schema::pds::refresh_token::dsl as RefreshTokenSchema;
 
     let did = did.to_owned();
     let app_pass_name = app_pass_name.to_owned();
···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<Option<models::RefreshToken>> {
-    use rsky_pds::schema::pds::refresh_token::dsl as RefreshTokenSchema;
+    use crate::schema::pds::refresh_token::dsl as RefreshTokenSchema;
     let id = id.to_owned();
     db.get()
         .await?
···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<()> {
-    use rsky_pds::schema::pds::refresh_token::dsl as RefreshTokenSchema;
+    use crate::schema::pds::refresh_token::dsl as RefreshTokenSchema;
     let did = did.to_owned();
 
     db.get()
···
                 expires_at,
                 next_id,
             } = opts;
-            use rsky_pds::schema::pds::refresh_token::dsl as RefreshTokenSchema;
+            use crate::schema::pds::refresh_token::dsl as RefreshTokenSchema;
 
             drop(
                 update(RefreshTokenSchema::refresh_token)

src/account_manager/helpers/email_token.rs

···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<String> {
-    use rsky_pds::schema::pds::email_token::dsl as EmailTokenSchema;
+    use crate::schema::pds::email_token::dsl as EmailTokenSchema;
     let token = get_random_token().to_uppercase();
     let now = rsky_common::now();
 
···
     >,
 ) -> Result<()> {
     let expiration_len = expiration_len.unwrap_or(MINUTE * 15);
-    use rsky_pds::schema::pds::email_token::dsl as EmailTokenSchema;
+    use crate::schema::pds::email_token::dsl as EmailTokenSchema;
 
     let did = did.to_owned();
     let token = token.to_owned();
···
     >,
 ) -> Result<String> {
     let expiration_len = expiration_len.unwrap_or(MINUTE * 15);
-    use rsky_pds::schema::pds::email_token::dsl as EmailTokenSchema;
+    use crate::schema::pds::email_token::dsl as EmailTokenSchema;
 
     let token = token.to_owned();
     let res = db
···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<()> {
-    use rsky_pds::schema::pds::email_token::dsl as EmailTokenSchema;
+    use crate::schema::pds::email_token::dsl as EmailTokenSchema;
     let did = did.to_owned();
     _ = db
         .get()
···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<()> {
-    use rsky_pds::schema::pds::email_token::dsl as EmailTokenSchema;
+    use crate::schema::pds::email_token::dsl as EmailTokenSchema;
 
     let did = did.to_owned();
     _ = db

src/account_manager/helpers/invite.rs

···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<()> {
-    use rsky_pds::schema::pds::actor::dsl as ActorSchema;
-    use rsky_pds::schema::pds::invite_code::dsl as InviteCodeSchema;
-    use rsky_pds::schema::pds::invite_code_use::dsl as InviteCodeUseSchema;
+    use crate::schema::pds::actor::dsl as ActorSchema;
+    use crate::schema::pds::invite_code::dsl as InviteCodeSchema;
+    use crate::schema::pds::invite_code_use::dsl as InviteCodeUseSchema;
 
     db.get().await?.interact(move |conn| {
         let invite: Option<models::InviteCode> = InviteCodeSchema::invite_code
···
     >,
 ) -> Result<()> {
     if let Some(invite_code) = invite_code {
-        use rsky_pds::schema::pds::invite_code_use::dsl as InviteCodeUseSchema;
+        use crate::schema::pds::invite_code_use::dsl as InviteCodeUseSchema;
 
         _ = db
             .get()
···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<()> {
-    use rsky_pds::schema::pds::invite_code::dsl as InviteCodeSchema;
+    use crate::schema::pds::invite_code::dsl as InviteCodeSchema;
     let created_at = rsky_common::now();
 
     _ = db
···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<Vec<CodeDetail>> {
-    use rsky_pds::schema::pds::invite_code::dsl as InviteCodeSchema;
+    use crate::schema::pds::invite_code::dsl as InviteCodeSchema;
 
     let for_account = for_account.to_owned();
     let rows = db
···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<Vec<CodeDetail>> {
-    use rsky_pds::schema::pds::invite_code::dsl as InviteCodeSchema;
+    use crate::schema::pds::invite_code::dsl as InviteCodeSchema;
 
     let did = did.to_owned();
     let res: Vec<models::InviteCode> = db
···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<BTreeMap<String, Vec<CodeUse>>> {
-    use rsky_pds::schema::pds::invite_code_use::dsl as InviteCodeUseSchema;
+    use crate::schema::pds::invite_code_use::dsl as InviteCodeUseSchema;
 
     let mut uses: BTreeMap<String, Vec<CodeUse>> = BTreeMap::new();
     if !codes.is_empty() {
···
     if dids.is_empty() {
         return Ok(BTreeMap::new());
     }
-    use rsky_pds::schema::pds::invite_code::dsl as InviteCodeSchema;
-    use rsky_pds::schema::pds::invite_code_use::dsl as InviteCodeUseSchema;
+    use crate::schema::pds::invite_code::dsl as InviteCodeSchema;
+    use crate::schema::pds::invite_code_use::dsl as InviteCodeUseSchema;
 
     let dids = dids.clone();
     let res: Vec<models::InviteCode> = db
···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<()> {
-    use rsky_pds::schema::pds::account::dsl as AccountSchema;
+    use crate::schema::pds::account::dsl as AccountSchema;
 
     let disabled: i16 = if disabled { 1 } else { 0 };
     let did = did.to_owned();
···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<()> {
-    use rsky_pds::schema::pds::invite_code::dsl as InviteCodeSchema;
+    use crate::schema::pds::invite_code::dsl as InviteCodeSchema;
 
     let DisableInviteCodesOpts { codes, accounts } = opts;
     if !codes.is_empty() {

src/account_manager/helpers/password.rs

···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<bool> {
-    use rsky_pds::schema::pds::account::dsl as AccountSchema;
+    use crate::schema::pds::account::dsl as AccountSchema;
 
     let did = did.to_owned();
     let found = db
···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<Option<String>> {
-    use rsky_pds::schema::pds::app_password::dsl as AppPasswordSchema;
+    use crate::schema::pds::app_password::dsl as AppPasswordSchema;
 
     let did = did.to_owned();
     let password = password.to_owned();
···
     let password = chunks.join("-");
     let password_encrypted = hash_app_password(&did, &password).await?;
 
-    use rsky_pds::schema::pds::app_password::dsl as AppPasswordSchema;
+    use crate::schema::pds::app_password::dsl as AppPasswordSchema;
 
     let created_at = now();
 
···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<Vec<(String, String)>> {
-    use rsky_pds::schema::pds::app_password::dsl as AppPasswordSchema;
+    use crate::schema::pds::app_password::dsl as AppPasswordSchema;
 
     let did = did.to_owned();
     db.get()
···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<()> {
-    use rsky_pds::schema::pds::account::dsl as AccountSchema;
+    use crate::schema::pds::account::dsl as AccountSchema;
 
     db.get()
         .await?
···
         deadpool_diesel::sqlite::Object,
     >,
 ) -> Result<()> {
-    use rsky_pds::schema::pds::app_password::dsl as AppPasswordSchema;
+    use crate::schema::pds::app_password::dsl as AppPasswordSchema;
 
     let did = did.to_owned();
     let name = name.to_owned();

src/account_manager/helpers/repo.rs

···
     >,
 ) -> Result<()> {
     // @TODO balance risk of a race in the case of a long retry
-    use rsky_pds::schema::pds::repo_root::dsl as RepoRootSchema;
+    use crate::schema::pds::repo_root::dsl as RepoRootSchema;
 
     let now = rsky_common::now();
 

src/actor_store/blob.rs

···
 
     /// Get metadata for a blob by CID
     pub async fn get_blob_metadata(&self, cid: Cid) -> Result<GetBlobMetadataOutput> {
-        use rsky_pds::schema::pds::blob::dsl as BlobSchema;
+        use crate::schema::pds::blob::dsl as BlobSchema;
 
         let did = self.did.clone();
         let found = self
···
 
     /// Get all records that reference a specific blob
     pub async fn get_records_for_blob(&self, cid: Cid) -> Result<Vec<String>> {
-        use rsky_pds::schema::pds::record_blob::dsl as RecordBlobSchema;
+        use crate::schema::pds::record_blob::dsl as RecordBlobSchema;
 
         let did = self.did.clone();
         let res = self
···
 
     /// Track a blob that hasn't been associated with any records yet
     pub async fn track_untethered_blob(&self, metadata: BlobMetadata) -> Result<BlobRef> {
-        use rsky_pds::schema::pds::blob::dsl as BlobSchema;
+        use crate::schema::pds::blob::dsl as BlobSchema;
 
         let did = self.did.clone();
         self.db.get().await?.interact(move |conn| {
···
 
     /// Delete blobs that are no longer referenced by any records
     pub async fn delete_dereferenced_blobs(&self, writes: Vec<PreparedWrite>) -> Result<()> {
-        use rsky_pds::schema::pds::blob::dsl as BlobSchema;
-        use rsky_pds::schema::pds::record_blob::dsl as RecordBlobSchema;
+        use crate::schema::pds::blob::dsl as BlobSchema;
+        use crate::schema::pds::record_blob::dsl as RecordBlobSchema;
 
         // Extract URIs
         let uris: Vec<String> = writes
···
 
     /// Verify a blob and make it permanent
     pub async fn verify_blob_and_make_permanent(&self, blob: PreparedBlobRef) -> Result<()> {
-        use rsky_pds::schema::pds::blob::dsl as BlobSchema;
+        use crate::schema::pds::blob::dsl as BlobSchema;
 
         let found = self
             .db
···
 
     /// Associate a blob with a record
     pub async fn associate_blob(&self, blob: PreparedBlobRef, record_uri: String) -> Result<()> {
-        use rsky_pds::schema::pds::record_blob::dsl as RecordBlobSchema;
+        use crate::schema::pds::record_blob::dsl as RecordBlobSchema;
 
         let cid = blob.cid.to_string();
         let did = self.did.clone();
···
 
     /// Count all blobs for this actor
     pub async fn blob_count(&self) -> Result<i64> {
-        use rsky_pds::schema::pds::blob::dsl as BlobSchema;
+        use crate::schema::pds::blob::dsl as BlobSchema;
 
         let did = self.did.clone();
         self.db
···
 
     /// Count blobs associated with records
     pub async fn record_blob_count(&self) -> Result<i64> {
-        use rsky_pds::schema::pds::record_blob::dsl as RecordBlobSchema;
+        use crate::schema::pds::record_blob::dsl as RecordBlobSchema;
 
         let did = self.did.clone();
         self.db
···
         &self,
         opts: ListMissingBlobsOpts,
     ) -> Result<Vec<ListMissingBlobsRefRecordBlob>> {
-        use rsky_pds::schema::pds::blob::dsl as BlobSchema;
-        use rsky_pds::schema::pds::record_blob::dsl as RecordBlobSchema;
+        use crate::schema::pds::blob::dsl as BlobSchema;
+        use crate::schema::pds::record_blob::dsl as RecordBlobSchema;
 
         let did = self.did.clone();
         self.db
···
 
     /// List all blobs with optional filtering
     pub async fn list_blobs(&self, opts: ListBlobsOpts) -> Result<Vec<String>> {
-        use rsky_pds::schema::pds::record::dsl as RecordSchema;
-        use rsky_pds::schema::pds::record_blob::dsl as RecordBlobSchema;
+        use crate::schema::pds::record::dsl as RecordSchema;
+        use crate::schema::pds::record_blob::dsl as RecordBlobSchema;
 
         let ListBlobsOpts {
             since,
···
 
     /// Get the takedown status of a blob
     pub async fn get_blob_takedown_status(&self, cid: Cid) -> Result<Option<StatusAttr>> {
-        use rsky_pds::schema::pds::blob::dsl as BlobSchema;
+        use crate::schema::pds::blob::dsl as BlobSchema;
 
         self.db
             .get()
···
 
     /// Update the takedown status of a blob
     pub async fn update_blob_takedown_status(&self, blob: Cid, takedown: StatusAttr) -> Result<()> {
-        use rsky_pds::schema::pds::blob::dsl as BlobSchema;
+        use crate::schema::pds::blob::dsl as BlobSchema;
 
         let takedown_ref: Option<String> = match takedown.applied {
             true => takedown.r#ref.map_or_else(|| Some(now()), Some),

src/actor_store/mod.rs

···
 
     pub async fn destroy(&mut self) -> Result<()> {
         let did: String = self.did.clone();
-        use rsky_pds::schema::pds::blob::dsl as BlobSchema;
+        use crate::schema::pds::blob::dsl as BlobSchema;
 
         let blob_rows: Vec<String> = self
             .storage
···
             return Ok(vec![]);
         }
         let did: String = self.did.clone();
-        use rsky_pds::schema::pds::record::dsl as RecordSchema;
+        use crate::schema::pds::record::dsl as RecordSchema;
 
         let cid_strs: Vec<String> = cids.into_iter().map(|c| c.to_string()).collect();
         let touched_uri_strs: Vec<String> = touched_uris.iter().map(|t| t.to_string()).collect();

src/actor_store/preference.rs

···
         namespace: Option<String>,
         scope: AuthScope,
     ) -> Result<Vec<RefPreferences>> {
-        use rsky_pds::schema::pds::account_pref::dsl as AccountPrefSchema;
+        use crate::schema::pds::account_pref::dsl as AccountPrefSchema;
 
         let did = self.did.clone();
         self.db
···
                             bail!("Do not have authorization to set preferences.");
                         }
                         // get all current prefs for user and prep new pref rows
-                        use rsky_pds::schema::pds::account_pref::dsl as AccountPrefSchema;
+                        use crate::schema::pds::account_pref::dsl as AccountPrefSchema;
                         let all_prefs = AccountPrefSchema::account_pref
                             .filter(AccountPrefSchema::did.eq(&did))
                             .select(AccountPref::as_select())

src/actor_store/record.rs

···
 
     /// Count the total number of records.
     pub(crate) async fn record_count(&mut self) -> Result<i64> {
-        use rsky_pds::schema::pds::record::dsl::*;
+        use crate::schema::pds::record::dsl::*;
 
         let other_did = self.did.clone();
         self.db
···
 
     /// List all collections in the repository.
     pub(crate) async fn list_collections(&self) -> Result<Vec<String>> {
-        use rsky_pds::schema::pds::record::dsl::*;
+        use crate::schema::pds::record::dsl::*;
 
         let other_did = self.did.clone();
         self.db
···
         rkey_end: Option<String>,
         include_soft_deleted: Option<bool>,
     ) -> Result<Vec<RecordsForCollection>> {
-        use rsky_pds::schema::pds::record::dsl as RecordSchema;
-        use rsky_pds::schema::pds::repo_block::dsl as RepoBlockSchema;
+        use crate::schema::pds::record::dsl as RecordSchema;
+        use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
 
         let include_soft_deleted: bool = include_soft_deleted.unwrap_or(false);
         let mut builder = RecordSchema::record
···
         cid: Option<String>,
         include_soft_deleted: Option<bool>,
     ) -> Result<Option<GetRecord>> {
-        use rsky_pds::schema::pds::record::dsl as RecordSchema;
-        use rsky_pds::schema::pds::repo_block::dsl as RepoBlockSchema;
+        use crate::schema::pds::record::dsl as RecordSchema;
+        use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
 
         let include_soft_deleted: bool = include_soft_deleted.unwrap_or(false);
         let mut builder = RecordSchema::record
···
         cid: Option<String>,
         include_soft_deleted: Option<bool>,
     ) -> Result<bool> {
-        use rsky_pds::schema::pds::record::dsl as RecordSchema;
+        use crate::schema::pds::record::dsl as RecordSchema;
 
         let include_soft_deleted: bool = include_soft_deleted.unwrap_or(false);
         let mut builder = RecordSchema::record
···
         &self,
         uri: String,
     ) -> Result<Option<StatusAttr>> {
-        use rsky_pds::schema::pds::record::dsl as RecordSchema;
+        use crate::schema::pds::record::dsl as RecordSchema;
 
         let res = self
             .db
···
 
     /// Get the current CID for a record URI.
     pub(crate) async fn get_current_record_cid(&self, uri: String) -> Result<Option<Cid>> {
-        use rsky_pds::schema::pds::record::dsl as RecordSchema;
+        use crate::schema::pds::record::dsl as RecordSchema;
 
         let res = self
             .db
···
         path: String,
         link_to: String,
     ) -> Result<Vec<Record>> {
-        use rsky_pds::schema::pds::backlink::dsl as BacklinkSchema;
-        use rsky_pds::schema::pds::record::dsl as RecordSchema;
+        use crate::schema::pds::backlink::dsl as BacklinkSchema;
+        use crate::schema::pds::record::dsl as RecordSchema;
 
         let res = self
             .db
···
             bail!("Expected indexed URI to contain a record key")
         }
 
-        use rsky_pds::schema::pds::record::dsl as RecordSchema;
+        use crate::schema::pds::record::dsl as RecordSchema;
 
         // Track current version of record
         let (record, uri) = self
···
     #[tracing::instrument(skip_all)]
     pub(crate) async fn delete_record(&self, uri: &AtUri) -> Result<()> {
         tracing::debug!("@LOG DEBUG RecordReader::delete_record, deleting indexed record {uri}");
-        use rsky_pds::schema::pds::backlink::dsl as BacklinkSchema;
-        use rsky_pds::schema::pds::record::dsl as RecordSchema;
+        use crate::schema::pds::backlink::dsl as BacklinkSchema;
+        use crate::schema::pds::record::dsl as RecordSchema;
         let uri = uri.to_string();
         self.db
             .get()
···
 
     /// Remove backlinks for a URI.
     pub(crate) async fn remove_backlinks_by_uri(&self, uri: &AtUri) -> Result<()> {
-        use rsky_pds::schema::pds::backlink::dsl as BacklinkSchema;
+        use crate::schema::pds::backlink::dsl as BacklinkSchema;
         let uri = uri.to_string();
         self.db
             .get()
···
         if backlinks.is_empty() {
             Ok(())
         } else {
-            use rsky_pds::schema::pds::backlink::dsl as BacklinkSchema;
+            use crate::schema::pds::backlink::dsl as BacklinkSchema;
             self.db
                 .get()
                 .await?
···
         uri: &AtUri,
         takedown: StatusAttr,
     ) -> Result<()> {
-        use rsky_pds::schema::pds::record::dsl as RecordSchema;
+        use crate::schema::pds::record::dsl as RecordSchema;
 
         let takedown_ref: Option<String> = match takedown.applied {
             true => takedown

src/actor_store/sql_repo.rs

···
         let cid = *cid;
 
         Box::pin(async move {
-            use rsky_pds::schema::pds::repo_block::dsl as RepoBlockSchema;
+            use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
             let cached = {
                 let cache_guard = self.cache.read().await;
                 cache_guard.get(cid).cloned()
···
         let did: String = self.did.clone();
 
         Box::pin(async move {
-            use rsky_pds::schema::pds::repo_block::dsl as RepoBlockSchema;
+            use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
             let cached = {
                 let mut cache_guard = self.cache.write().await;
                 cache_guard.get_many(cids)?
···
         let did: String = self.did.clone();
         let bytes_cloned = bytes.clone();
         Box::pin(async move {
-            use rsky_pds::schema::pds::repo_block::dsl as RepoBlockSchema;
+            use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
 
             _ = self
                 .db
···
         let did: String = self.did.clone();
 
         Box::pin(async move {
-            use rsky_pds::schema::pds::repo_block::dsl as RepoBlockSchema;
+            use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
 
             let blocks: Vec<RepoBlock> = to_put
                 .map
···
         let now: String = self.now.clone();
 
         Box::pin(async move {
-            use rsky_pds::schema::pds::repo_root::dsl as RepoRootSchema;
+            use crate::schema::pds::repo_root::dsl as RepoRootSchema;
 
             let is_create = is_create.unwrap_or(false);
             if is_create {
···
         let did: String = self.did.clone();
         let since = since.clone();
         let cursor = cursor.clone();
-        use rsky_pds::schema::pds::repo_block::dsl as RepoBlockSchema;
+        use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
 
         Ok(self
             .db
···
 
     pub async fn count_blocks(&self) -> Result<i64> {
         let did: String = self.did.clone();
-        use rsky_pds::schema::pds::repo_block::dsl as RepoBlockSchema;
+        use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
 
         let res = self
             .db
···
     /// Proactively cache all blocks from a particular commit (to prevent multiple roundtrips)
     pub async fn cache_rev(&mut self, rev: String) -> Result<()> {
         let did: String = self.did.clone();
-        use rsky_pds::schema::pds::repo_block::dsl as RepoBlockSchema;
+        use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
 
         let result: Vec<(String, Vec<u8>)> = self
             .db
···
             return Ok(());
         }
         let did: String = self.did.clone();
-        use rsky_pds::schema::pds::repo_block::dsl as RepoBlockSchema;
+        use crate::schema::pds::repo_block::dsl as RepoBlockSchema;
 
         let cid_strings: Vec<String> = cids.into_iter().map(|c| c.to_string()).collect();
         _ = self
···
 
     pub async fn get_root_detailed(&self) -> Result<CidAndRev> {
         let did: String = self.did.clone();
-        use rsky_pds::schema::pds::repo_root::dsl as RepoRootSchema;
+        use crate::schema::pds::repo_root::dsl as RepoRootSchema;
 
         let res = self
             .db

src/auth.rs

···
 
     // Extract subject (DID)
     if let Some(did) = claims.get("sub").and_then(serde_json::Value::as_str) {
-        use rsky_pds::schema::pds::account::dsl as AccountSchema;
+        use crate::schema::pds::account::dsl as AccountSchema;
         let did_clone = did.to_owned();
 
         let _did = state
···
 
     // Extract subject (DID) from access token
     if let Some(did) = claims.get("sub").and_then(|v| v.as_str()) {
-        use rsky_pds::schema::pds::account::dsl as AccountSchema;
+        use crate::schema::pds::account::dsl as AccountSchema;
 
         let did_clone = did.to_owned();
 

src/lib.rs

···
+//! PDS implementation.
+mod account_manager;
+mod actor_endpoints;
+mod actor_store;
+mod auth;
+mod config;
+mod db;
+mod did;
+mod endpoints;
+pub mod error;
+mod firehose;
+mod metrics;
+mod mmap;
+mod oauth;
+mod plc;
+mod schema;
+mod service_proxy;
+#[cfg(test)]
+mod tests;
+
+use anyhow::{Context as _, anyhow};
+use atrium_api::types::string::Did;
+use atrium_crypto::keypair::{Export as _, Secp256k1Keypair};
+use auth::AuthenticatedUser;
+use axum::{
+    Router,
+    body::Body,
+    extract::{FromRef, Request, State},
+    http::{self, HeaderMap, Response, StatusCode, Uri},
+    response::IntoResponse,
+    routing::get,
+};
+use azure_core::credentials::TokenCredential;
+use clap::Parser;
+use clap_verbosity_flag::{InfoLevel, Verbosity, log::LevelFilter};
+use config::AppConfig;
+use db::establish_pool;
+use deadpool_diesel::sqlite::Pool;
+use diesel::prelude::*;
+use diesel_migrations::{EmbeddedMigrations, embed_migrations};
+pub use error::Error;
+use figment::{Figment, providers::Format as _};
+use firehose::FirehoseProducer;
+use http_cache_reqwest::{CacheMode, HttpCacheOptions, MokaManager};
+use rand::Rng as _;
+use serde::{Deserialize, Serialize};
+use service_proxy::service_proxy;
+use std::{
+    net::{IpAddr, Ipv4Addr, SocketAddr},
+    path::PathBuf,
+    str::FromStr as _,
+    sync::Arc,
+};
+use tokio::net::TcpListener;
+use tower_http::{cors::CorsLayer, trace::TraceLayer};
+use tracing::{info, warn};
+use uuid::Uuid;
+
+/// The application user agent. Concatenates the package name and version. e.g. `bluepds/0.0.0`.
+pub const APP_USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"),);
+
+/// Embedded migrations
+pub const MIGRATIONS: EmbeddedMigrations = embed_migrations!("./migrations");
+
+/// The application-wide result type.
+pub type Result<T> = std::result::Result<T, Error>;
+/// The reqwest client type with middleware.
+pub type Client = reqwest_middleware::ClientWithMiddleware;
+/// The Azure credential type.
+pub type Cred = Arc<dyn TokenCredential>;
+
+#[expect(
+    clippy::arbitrary_source_item_ordering,
+    reason = "serialized data might be structured"
+)]
+#[derive(Serialize, Deserialize, Debug, Clone)]
+/// The key data structure.
+struct KeyData {
+    /// Primary signing key for all repo operations.
+    skey: Vec<u8>,
+    /// Primary signing (rotation) key for all PLC operations.
+    rkey: Vec<u8>,
+}
+
+// FIXME: We should use P256Keypair instead. SecP256K1 is primarily used for cryptocurrencies,
+// and the implementations of this algorithm are much more limited as compared to P256.
+//
+// Reference: https://soatok.blog/2022/05/19/guidance-for-choosing-an-elliptic-curve-signature-algorithm-in-2022/
+#[derive(Clone)]
+/// The signing key for PLC/DID operations.
+pub struct SigningKey(Arc<Secp256k1Keypair>);
+#[derive(Clone)]
+/// The rotation key for PLC operations.
+pub struct RotationKey(Arc<Secp256k1Keypair>);
+
+impl std::ops::Deref for SigningKey {
+    type Target = Secp256k1Keypair;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl SigningKey {
+    /// Import from a private key.
+    pub fn import(key: &[u8]) -> Result<Self> {
+        let key = Secp256k1Keypair::import(key).context("failed to import signing key")?;
+        Ok(Self(Arc::new(key)))
+    }
+}
+
+impl std::ops::Deref for RotationKey {
+    type Target = Secp256k1Keypair;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+#[derive(Parser, Debug, Clone)]
+/// Command line arguments.
+pub struct Args {
+    /// Path to the configuration file
+    #[arg(short, long, default_value = "default.toml")]
+    pub config: PathBuf,
+    /// The verbosity level.
+    #[command(flatten)]
+    pub verbosity: Verbosity<InfoLevel>,
+}
+
+pub struct ActorPools {
+    pub repo: Pool,
+    pub blob: Pool,
+}
+
+impl Clone for ActorPools {
+    fn clone(&self) -> Self {
+        Self {
+            repo: self.repo.clone(),
+            blob: self.blob.clone(),
+        }
+    }
+}
+
+#[expect(clippy::arbitrary_source_item_ordering, reason = "arbitrary")]
+#[derive(Clone, FromRef)]
+pub struct AppState {
+    /// The application configuration.
+    pub config: AppConfig,
+    /// The Azure credential.
+    pub cred: Cred,
+    /// The main database connection pool. Used for common PDS data, like invite codes.
+    pub db: Pool,
+    /// Actor-specific database connection pools. Hashed by DID.
+    pub db_actors: std::collections::HashMap<String, ActorPools>,
+
+    /// The HTTP client with middleware.
+    pub client: Client,
+    /// The simple HTTP client.
+    pub simple_client: reqwest::Client,
+    /// The firehose producer.
+    pub firehose: FirehoseProducer,
+
+    /// The signing key.
+    pub signing_key: SigningKey,
+    /// The rotation key.
+    pub rotation_key: RotationKey,
+}
+
+/// The index (/) route.
+async fn index() -> impl IntoResponse {
+    r"
+         __                         __
+        /\ \__                     /\ \__
+    __  \ \ ,_\  _____   _ __   ___\ \ ,_\   ___
+  /'__'\ \ \ \/ /\ '__'\/\''__\/ __'\ \ \/  / __'\
+ /\ \L\.\_\ \ \_\ \ \L\ \ \ \//\ \L\ \ \ \_/\ \L\ \
+ \ \__/.\_\\ \__\\ \ ,__/\ \_\\ \____/\ \__\ \____/
+  \/__/\/_/ \/__/ \ \ \/  \/_/ \/___/  \/__/\/___/
+                   \ \_\
+                    \/_/
+
+
+This is an AT Protocol Personal Data Server (aka, an atproto PDS)
+
+Most API routes are under /xrpc/
+
+      Code: https://github.com/DrChat/bluepds
+  Protocol: https://atproto.com
+    "
+}
+
+/// The main application entry point.
+#[expect(
+    clippy::cognitive_complexity,
+    clippy::too_many_lines,
+    unused_qualifications,
+    reason = "main function has high complexity"
+)]
+pub async fn run() -> anyhow::Result<()> {
+    let args = Args::parse();
+
+    // Set up trace logging to console and account for the user-provided verbosity flag.
+    if args.verbosity.log_level_filter() != LevelFilter::Off {
+        let lvl = match args.verbosity.log_level_filter() {
+            LevelFilter::Error => tracing::Level::ERROR,
+            LevelFilter::Warn => tracing::Level::WARN,
+            LevelFilter::Info | LevelFilter::Off => tracing::Level::INFO,
+            LevelFilter::Debug => tracing::Level::DEBUG,
+            LevelFilter::Trace => tracing::Level::TRACE,
+        };
+        tracing_subscriber::fmt().with_max_level(lvl).init();
+    }
+
+    if !args.config.exists() {
+        // Throw up a warning if the config file does not exist.
+        //
+        // This is not fatal because users can specify all configuration settings via
+        // the environment, but the most likely scenario here is that a user accidentally
+        // omitted the config file for some reason (e.g. forgot to mount it into Docker).
+        warn!(
+            "configuration file {} does not exist",
+            args.config.display()
+        );
+    }
+
+    // Read and parse the user-provided configuration.
+    let config: AppConfig = Figment::new()
+        .admerge(figment::providers::Toml::file(args.config))
+        .admerge(figment::providers::Env::prefixed("BLUEPDS_"))
+        .extract()
+        .context("failed to load configuration")?;
+
+    if config.test {
+        warn!("BluePDS starting up in TEST mode.");
+        warn!("This means the application will not federate with the rest of the network.");
+        warn!(
+            "If you want to turn this off, either set `test` to false in the config or define `BLUEPDS_TEST = false`"
+        );
+    }
+
+    // Initialize metrics reporting.
+    metrics::setup(config.metrics.as_ref()).context("failed to set up metrics exporter")?;
+
+    // Create a reqwest client that will be used for all outbound requests.
+    let simple_client = reqwest::Client::builder()
+        .user_agent(APP_USER_AGENT)
+        .build()
+        .context("failed to build requester client")?;
+    let client = reqwest_middleware::ClientBuilder::new(simple_client.clone())
+        .with(http_cache_reqwest::Cache(http_cache_reqwest::HttpCache {
+            mode: CacheMode::Default,
+            manager: MokaManager::default(),
+            options: HttpCacheOptions::default(),
+        }))
+        .build();
+
+    tokio::fs::create_dir_all(&config.key.parent().context("should have parent")?)
+        .await
+        .context("failed to create key directory")?;
+
+    // Check if crypto keys exist. If not, create new ones.
+    let (skey, rkey) = if let Ok(f) = std::fs::File::open(&config.key) {
+        let keys: KeyData = serde_ipld_dagcbor::from_reader(std::io::BufReader::new(f))
+            .context("failed to deserialize crypto keys")?;
+
+        let skey = Secp256k1Keypair::import(&keys.skey).context("failed to import signing key")?;
+        let rkey = Secp256k1Keypair::import(&keys.rkey).context("failed to import rotation key")?;
+
+        (SigningKey(Arc::new(skey)), RotationKey(Arc::new(rkey)))
+    } else {
+        info!("signing keys not found, generating new ones");
+
+        let skey = Secp256k1Keypair::create(&mut rand::thread_rng());
+        let rkey = Secp256k1Keypair::create(&mut rand::thread_rng());
+
+        let keys = KeyData {
+            skey: skey.export(),
+            rkey: rkey.export(),
+        };
+
+        let mut f = std::fs::File::create(&config.key).context("failed to create key file")?;
+        serde_ipld_dagcbor::to_writer(&mut f, &keys).context("failed to serialize crypto keys")?;
+
+        (SigningKey(Arc::new(skey)), RotationKey(Arc::new(rkey)))
+    };
+
+    tokio::fs::create_dir_all(&config.repo.path).await?;
+    tokio::fs::create_dir_all(&config.plc.path).await?;
+    tokio::fs::create_dir_all(&config.blob.path).await?;
+
+    let cred = azure_identity::DefaultAzureCredential::new()
+        .context("failed to create Azure credential")?;
+
+    // Create a database connection manager and pool for the main database.
+    let pool =
+        establish_pool(&config.db).context("failed to establish database connection pool")?;
+    // Create a dictionary of database connection pools for each actor.
+    let mut actor_pools = std::collections::HashMap::new();
+    // let mut actor_blob_pools = std::collections::HashMap::new();
+    // We'll determine actors by looking in the data/repo dir for .db files.
+    let mut actor_dbs = tokio::fs::read_dir(&config.repo.path)
+        .await
+        .context("failed to read repo directory")?;
+    while let Some(entry) = actor_dbs
+        .next_entry()
+        .await
+        .context("failed to read repo dir")?
+    {
+        let path = entry.path();
+        if path.extension().and_then(|s| s.to_str()) == Some("db") {
+            let did_path = path
+                .file_stem()
+                .and_then(|s| s.to_str())
+                .context("failed to get actor DID")?;
+            let did = Did::from_str(&format!("did:plc:{}", did_path))
+                .expect("should be able to parse actor DID");
+
+            // Create a new database connection manager and pool for the actor.
+            // The path for the SQLite connection needs to look like "sqlite://data/repo/<actor>.db"
+            let path_repo = format!("sqlite://{}", did_path);
+            let actor_repo_pool =
+                establish_pool(&path_repo).context("failed to create database connection pool")?;
+            // Create a new database connection manager and pool for the actor blobs.
+            // The path for the SQLite connection needs to look like "sqlite://data/blob/<actor>.db"
+            let path_blob = path_repo.replace("repo", "blob");
+            let actor_blob_pool =
+                establish_pool(&path_blob).context("failed to create database connection pool")?;
+            drop(actor_pools.insert(
+                did.to_string(),
+                ActorPools {
+                    repo: actor_repo_pool,
+                    blob: actor_blob_pool,
+                },
+            ));
+        }
+    }
+    // Apply pending migrations
+    // let conn = pool.get().await?;
+    // conn.run_pending_migrations(MIGRATIONS)
+    //     .expect("should be able to run migrations");
+
+    let (_fh, fhp) = firehose::spawn(client.clone(), config.clone());
+
+    let addr = config
+        .listen_address
+        .unwrap_or(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8000));
+
+    let app = Router::new()
+        .route("/", get(index))
+        .merge(oauth::routes())
+        .nest(
+            "/xrpc",
+            endpoints::routes()
+                .merge(actor_endpoints::routes())
+                .fallback(service_proxy),
+        )
+        // .layer(RateLimitLayer::new(30, Duration::from_secs(30)))
+        .layer(CorsLayer::permissive())
+        .layer(TraceLayer::new_for_http())
+        .with_state(AppState {
+            cred,
+            config: config.clone(),
+            db: pool.clone(),
+            db_actors: actor_pools.clone(),
+            client: client.clone(),
+            simple_client,
+            firehose: fhp,
+            signing_key: skey,
+            rotation_key: rkey,
+        });
+
+    info!("listening on {addr}");
+    info!("connect to: http://127.0.0.1:{}", addr.port());
+
+    // Determine whether or not this was the first startup (i.e. no accounts exist and no invite codes were created).
+    // If so, create an invite code and share it via the console.
+    let conn = pool.get().await.context("failed to get db connection")?;
+
+    #[derive(QueryableByName)]
+    struct TotalCount {
+        #[diesel(sql_type = diesel::sql_types::Integer)]
+        total_count: i32,
+    }
+
+    let result = conn.interact(move |conn| {
+        diesel::sql_query(
+            "SELECT (SELECT COUNT(*) FROM accounts) + (SELECT COUNT(*) FROM invites) AS total_count",
+        )
+        .get_result::<TotalCount>(conn)
+    })
+    .await
+    .expect("should be able to query database")?;
+
+    let c = result.total_count;
+
+    #[expect(clippy::print_stdout)]
+    if c == 0 {
+        let uuid = Uuid::new_v4().to_string();
+
+        let uuid_clone = uuid.clone();
+        _ = conn
+            .interact(move |conn| {
+                diesel::sql_query(
+            "INSERT INTO invites (id, did, count, created_at) VALUES (?, NULL, 1, datetime('now'))",
+        )
+        .bind::<diesel::sql_types::Text, _>(uuid_clone)
+        .execute(conn)
+        .context("failed to create new invite code")
+        .expect("should be able to create invite code")
+            })
+            .await
+            .expect("should be able to create invite code");
+
+        // N.B: This is a sensitive message, so we're bypassing `tracing` here and
+        // logging it directly to console.
+        println!("=====================================");
+        println!("            FIRST STARTUP            ");
+        println!("=====================================");
+        println!("Use this code to create an account:");
+        println!("{uuid}");
+        println!("=====================================");
+    }
+
+    let listener = TcpListener::bind(&addr)
+        .await
+        .context("failed to bind address")?;
+
+    // Serve the app, and request crawling from upstream relays.
+    let serve = tokio::spawn(async move {
+        axum::serve(listener, app.into_make_service())
+            .await
+            .context("failed to serve app")
+    });
+
+    // Now that the app is live, request a crawl from upstream relays.
+    firehose::reconnect_relays(&client, &config).await;
+
+    serve
+        .await
+        .map_err(Into::into)
+        .and_then(|r| r)
+        .context("failed to serve app")
+}
+
+/// Creates an app router with the provided AppState.
+pub fn create_app(state: AppState) -> Router {
+    Router::new()
+        .route("/", get(index))
+        .merge(oauth::routes())
+        .nest(
+            "/xrpc",
+            endpoints::routes()
+                .merge(actor_endpoints::routes())
+                .fallback(service_proxy),
+        )
+        .layer(CorsLayer::permissive())
+        .layer(TraceLayer::new_for_http())
+        .with_state(state)
+}

src/main.rs

···
-//! PDS implementation.
-mod account_manager;
-mod actor_endpoints;
-mod actor_store;
-mod auth;
-mod config;
-mod db;
-mod did;
-mod endpoints;
-mod error;
-mod firehose;
-mod metrics;
-mod mmap;
-mod oauth;
-mod plc;
-mod schema;
-mod service_proxy;
-#[cfg(test)]
-mod tests;
+//! BluePDS binary entry point.
 
-use anyhow::{Context as _, anyhow};
-use atrium_api::types::string::Did;
-use atrium_crypto::keypair::{Export as _, Secp256k1Keypair};
-use auth::AuthenticatedUser;
-use axum::{
-    Router,
-    body::Body,
-    extract::{FromRef, Request, State},
-    http::{self, HeaderMap, Response, StatusCode, Uri},
-    response::IntoResponse,
-    routing::get,
-};
-use azure_core::credentials::TokenCredential;
+use anyhow::Context as _;
 use clap::Parser;
-use clap_verbosity_flag::{InfoLevel, Verbosity, log::LevelFilter};
-use config::AppConfig;
-use db::establish_pool;
-use deadpool_diesel::sqlite::Pool;
-use diesel::prelude::*;
-use diesel_migrations::{EmbeddedMigrations, embed_migrations};
-#[expect(clippy::pub_use, clippy::useless_attribute)]
-pub use error::Error;
-use figment::{Figment, providers::Format as _};
-use firehose::FirehoseProducer;
-use http_cache_reqwest::{CacheMode, HttpCacheOptions, MokaManager};
-use rand::Rng as _;
-use serde::{Deserialize, Serialize};
-use service_proxy::service_proxy;
-use std::{
-    net::{IpAddr, Ipv4Addr, SocketAddr},
-    path::PathBuf,
-    str::FromStr as _,
-    sync::Arc,
-};
-use tokio::net::TcpListener;
-use tower_http::{cors::CorsLayer, trace::TraceLayer};
-use tracing::{info, warn};
-use uuid::Uuid;
-
-/// The application user agent. Concatenates the package name and version. e.g. `bluepds/0.0.0`.
-pub const APP_USER_AGENT: &str = concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"),);
-
-/// Embedded migrations
-pub const MIGRATIONS: EmbeddedMigrations = embed_migrations!("./migrations");
-
-/// The application-wide result type.
-pub type Result<T> = std::result::Result<T, Error>;
-/// The reqwest client type with middleware.
-pub type Client = reqwest_middleware::ClientWithMiddleware;
-/// The Azure credential type.
-pub type Cred = Arc<dyn TokenCredential>;
-
-#[expect(
-    clippy::arbitrary_source_item_ordering,
-    reason = "serialized data might be structured"
-)]
-#[derive(Serialize, Deserialize, Debug, Clone)]
-/// The key data structure.
-struct KeyData {
-    /// Primary signing key for all repo operations.
-    skey: Vec<u8>,
-    /// Primary signing (rotation) key for all PLC operations.
-    rkey: Vec<u8>,
-}
-
-// FIXME: We should use P256Keypair instead. SecP256K1 is primarily used for cryptocurrencies,
-// and the implementations of this algorithm are much more limited as compared to P256.
-//
-// Reference: https://soatok.blog/2022/05/19/guidance-for-choosing-an-elliptic-curve-signature-algorithm-in-2022/
-#[derive(Clone)]
-/// The signing key for PLC/DID operations.
-pub struct SigningKey(Arc<Secp256k1Keypair>);
-#[derive(Clone)]
-/// The rotation key for PLC operations.
-pub struct RotationKey(Arc<Secp256k1Keypair>);
-
-impl std::ops::Deref for SigningKey {
-    type Target = Secp256k1Keypair;
-
-    fn deref(&self) -> &Self::Target {
-        &self.0
-    }
-}
-
-impl SigningKey {
-    /// Import from a private key.
-    pub fn import(key: &[u8]) -> Result<Self> {
-        let key = Secp256k1Keypair::import(key).context("failed to import signing key")?;
-        Ok(Self(Arc::new(key)))
-    }
-}
-
-impl std::ops::Deref for RotationKey {
-    type Target = Secp256k1Keypair;
-
-    fn deref(&self) -> &Self::Target {
-        &self.0
-    }
-}
-
-#[derive(Parser, Debug, Clone)]
-/// Command line arguments.
-struct Args {
-    /// Path to the configuration file
-    #[arg(short, long, default_value = "default.toml")]
-    config: PathBuf,
-    /// The verbosity level.
-    #[command(flatten)]
-    verbosity: Verbosity<InfoLevel>,
-}
-
-struct ActorPools {
-    repo: Pool,
-    blob: Pool,
-}
-impl Clone for ActorPools {
-    fn clone(&self) -> Self {
-        Self {
-            repo: self.repo.clone(),
-            blob: self.blob.clone(),
-        }
-    }
-}
-
-#[expect(clippy::arbitrary_source_item_ordering, reason = "arbitrary")]
-#[derive(Clone, FromRef)]
-struct AppState {
-    /// The application configuration.
-    config: AppConfig,
-    /// The Azure credential.
-    cred: Cred,
-    /// The main database connection pool. Used for common PDS data, like invite codes.
-    db: Pool,
-    /// Actor-specific database connection pools. Hashed by DID.
-    db_actors: std::collections::HashMap<String, ActorPools>,
-
-    /// The HTTP client with middleware.
-    client: Client,
-    /// The simple HTTP client.
-    simple_client: reqwest::Client,
-    /// The firehose producer.
-    firehose: FirehoseProducer,
-
-    /// The signing key.
-    signing_key: SigningKey,
-    /// The rotation key.
-    rotation_key: RotationKey,
-}
-
-/// The index (/) route.
-async fn index() -> impl IntoResponse {
-    r"
-         __                         __
-        /\ \__                     /\ \__
-    __  \ \ ,_\  _____   _ __   ___\ \ ,_\   ___
-  /'__'\ \ \ \/ /\ '__'\/\''__\/ __'\ \ \/  / __'\
- /\ \L\.\_\ \ \_\ \ \L\ \ \ \//\ \L\ \ \ \_/\ \L\ \
- \ \__/.\_\\ \__\\ \ ,__/\ \_\\ \____/\ \__\ \____/
-  \/__/\/_/ \/__/ \ \ \/  \/_/ \/___/  \/__/\/___/
-                   \ \_\
-                    \/_/
-
-
-This is an AT Protocol Personal Data Server (aka, an atproto PDS)
-
-Most API routes are under /xrpc/
-
-      Code: https://github.com/DrChat/bluepds
-  Protocol: https://atproto.com
-    "
-}
-
-/// The main application entry point.
-#[expect(
-    clippy::cognitive_complexity,
-    clippy::too_many_lines,
-    unused_qualifications,
-    reason = "main function has high complexity"
-)]
-async fn run() -> anyhow::Result<()> {
-    let args = Args::parse();
-
-    // Set up trace logging to console and account for the user-provided verbosity flag.
-    if args.verbosity.log_level_filter() != LevelFilter::Off {
-        let lvl = match args.verbosity.log_level_filter() {
-            LevelFilter::Error => tracing::Level::ERROR,
-            LevelFilter::Warn => tracing::Level::WARN,
-            LevelFilter::Info | LevelFilter::Off => tracing::Level::INFO,
-            LevelFilter::Debug => tracing::Level::DEBUG,
-            LevelFilter::Trace => tracing::Level::TRACE,
-        };
-        tracing_subscriber::fmt().with_max_level(lvl).init();
-    }
-
-    if !args.config.exists() {
-        // Throw up a warning if the config file does not exist.
-        //
-        // This is not fatal because users can specify all configuration settings via
-        // the environment, but the most likely scenario here is that a user accidentally
-        // omitted the config file for some reason (e.g. forgot to mount it into Docker).
-        warn!(
-            "configuration file {} does not exist",
-            args.config.display()
-        );
-    }
-
-    // Read and parse the user-provided configuration.
-    let config: AppConfig = Figment::new()
-        .admerge(figment::providers::Toml::file(args.config))
-        .admerge(figment::providers::Env::prefixed("BLUEPDS_"))
-        .extract()
-        .context("failed to load configuration")?;
-
-    if config.test {
-        warn!("BluePDS starting up in TEST mode.");
-        warn!("This means the application will not federate with the rest of the network.");
-        warn!(
-            "If you want to turn this off, either set `test` to false in the config or define `BLUEPDS_TEST = false`"
-        );
-    }
-
-    // Initialize metrics reporting.
-    metrics::setup(config.metrics.as_ref()).context("failed to set up metrics exporter")?;
-
-    // Create a reqwest client that will be used for all outbound requests.
-    let simple_client = reqwest::Client::builder()
-        .user_agent(APP_USER_AGENT)
-        .build()
-        .context("failed to build requester client")?;
-    let client = reqwest_middleware::ClientBuilder::new(simple_client.clone())
-        .with(http_cache_reqwest::Cache(http_cache_reqwest::HttpCache {
-            mode: CacheMode::Default,
-            manager: MokaManager::default(),
-            options: HttpCacheOptions::default(),
-        }))
-        .build();
-
-    tokio::fs::create_dir_all(&config.key.parent().context("should have parent")?)
-        .await
-        .context("failed to create key directory")?;
-
-    // Check if crypto keys exist. If not, create new ones.
-    let (skey, rkey) = if let Ok(f) = std::fs::File::open(&config.key) {
-        let keys: KeyData = serde_ipld_dagcbor::from_reader(std::io::BufReader::new(f))
-            .context("failed to deserialize crypto keys")?;
-
-        let skey = Secp256k1Keypair::import(&keys.skey).context("failed to import signing key")?;
-        let rkey = Secp256k1Keypair::import(&keys.rkey).context("failed to import rotation key")?;
-
-        (SigningKey(Arc::new(skey)), RotationKey(Arc::new(rkey)))
-    } else {
-        info!("signing keys not found, generating new ones");
-
-        let skey = Secp256k1Keypair::create(&mut rand::thread_rng());
-        let rkey = Secp256k1Keypair::create(&mut rand::thread_rng());
-
-        let keys = KeyData {
-            skey: skey.export(),
-            rkey: rkey.export(),
-        };
-
-        let mut f = std::fs::File::create(&config.key).context("failed to create key file")?;
-        serde_ipld_dagcbor::to_writer(&mut f, &keys).context("failed to serialize crypto keys")?;
-
-        (SigningKey(Arc::new(skey)), RotationKey(Arc::new(rkey)))
-    };
-
-    tokio::fs::create_dir_all(&config.repo.path).await?;
-    tokio::fs::create_dir_all(&config.plc.path).await?;
-    tokio::fs::create_dir_all(&config.blob.path).await?;
-
-    let cred = azure_identity::DefaultAzureCredential::new()
-        .context("failed to create Azure credential")?;
-
-    // Create a database connection manager and pool for the main database.
-    let pool =
-        establish_pool(&config.db).context("failed to establish database connection pool")?;
-    // Create a dictionary of database connection pools for each actor.
-    let mut actor_pools = std::collections::HashMap::new();
-    // let mut actor_blob_pools = std::collections::HashMap::new();
-    // We'll determine actors by looking in the data/repo dir for .db files.
-    let mut actor_dbs = tokio::fs::read_dir(&config.repo.path)
-        .await
-        .context("failed to read repo directory")?;
-    while let Some(entry) = actor_dbs
-        .next_entry()
-        .await
-        .context("failed to read repo dir")?
-    {
-        let path = entry.path();
-        if path.extension().and_then(|s| s.to_str()) == Some("db") {
-            let did = path
-                .file_stem()
-                .and_then(|s| s.to_str())
-                .context("failed to get actor DID")?;
-            let did = Did::from_str(did).expect("should be able to parse actor DID");
-
-            // Create a new database connection manager and pool for the actor.
-            // The path for the SQLite connection needs to look like "sqlite://data/repo/<actor>.db"
-            let path_repo = format!("sqlite://{}", path.display());
-            let actor_repo_pool =
-                establish_pool(&path_repo).context("failed to create database connection pool")?;
-            // Create a new database connection manager and pool for the actor blobs.
-            // The path for the SQLite connection needs to look like "sqlite://data/blob/<actor>.db"
-            let path_blob = path_repo.replace("repo", "blob");
-            let actor_blob_pool =
-                establish_pool(&path_blob).context("failed to create database connection pool")?;
-            drop(actor_pools.insert(
-                did.to_string(),
-                ActorPools {
-                    repo: actor_repo_pool,
-                    blob: actor_blob_pool,
-                },
-            ));
-        }
-    }
-    // Apply pending migrations
-    // let conn = pool.get().await?;
-    // conn.run_pending_migrations(MIGRATIONS)
-    //     .expect("should be able to run migrations");
-
-    let (_fh, fhp) = firehose::spawn(client.clone(), config.clone());
-
-    let addr = config
-        .listen_address
-        .unwrap_or(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8000));
-
-    let app = Router::new()
-        .route("/", get(index))
-        .merge(oauth::routes())
-        .nest(
-            "/xrpc",
-            endpoints::routes()
-                .merge(actor_endpoints::routes())
-                .fallback(service_proxy),
-        )
-        // .layer(RateLimitLayer::new(30, Duration::from_secs(30)))
-        .layer(CorsLayer::permissive())
-        .layer(TraceLayer::new_for_http())
-        .with_state(AppState {
-            cred,
-            config: config.clone(),
-            db: pool.clone(),
-            db_actors: actor_pools.clone(),
-            client: client.clone(),
-            simple_client,
-            firehose: fhp,
-            signing_key: skey,
-            rotation_key: rkey,
-        });
-
-    info!("listening on {addr}");
-    info!("connect to: http://127.0.0.1:{}", addr.port());
-
-    // Determine whether or not this was the first startup (i.e. no accounts exist and no invite codes were created).
-    // If so, create an invite code and share it via the console.
-    let conn = pool.get().await.context("failed to get db connection")?;
-
-    #[derive(QueryableByName)]
-    struct TotalCount {
-        #[diesel(sql_type = diesel::sql_types::Integer)]
-        total_count: i32,
-    }
-
-    let result = conn.interact(move |conn| {
-        diesel::sql_query(
-            "SELECT (SELECT COUNT(*) FROM accounts) + (SELECT COUNT(*) FROM invites) AS total_count",
-        )
-        .get_result::<TotalCount>(conn)
-    })
-    .await
-    .expect("should be able to query database")?;
-
-    let c = result.total_count;
-
-    #[expect(clippy::print_stdout)]
-    if c == 0 {
-        let uuid = Uuid::new_v4().to_string();
-
-        let uuid_clone = uuid.clone();
-        _ = conn
-            .interact(move |conn| {
-                diesel::sql_query(
-            "INSERT INTO invites (id, did, count, created_at) VALUES (?, NULL, 1, datetime('now'))",
-        )
-        .bind::<diesel::sql_types::Text, _>(uuid_clone)
-        .execute(conn)
-        .context("failed to create new invite code")
-        .expect("should be able to create invite code")
-            })
-            .await
-            .expect("should be able to create invite code");
-
-        // N.B: This is a sensitive message, so we're bypassing `tracing` here and
-        // logging it directly to console.
-        println!("=====================================");
-        println!("            FIRST STARTUP            ");
-        println!("=====================================");
-        println!("Use this code to create an account:");
-        println!("{uuid}");
-        println!("=====================================");
-    }
-
-    let listener = TcpListener::bind(&addr)
-        .await
-        .context("failed to bind address")?;
-
-    // Serve the app, and request crawling from upstream relays.
-    let serve = tokio::spawn(async move {
-        axum::serve(listener, app.into_make_service())
-            .await
-            .context("failed to serve app")
-    });
-
-    // Now that the app is live, request a crawl from upstream relays.
-    firehose::reconnect_relays(&client, &config).await;
-
-    serve
-        .await
-        .map_err(Into::into)
-        .and_then(|r| r)
-        .context("failed to serve app")
-}
 
 #[tokio::main(flavor = "multi_thread")]
 async fn main() -> anyhow::Result<()> {
-    // Dispatch out to a separate function without a derive macro to help rust-analyzer along.
-    run().await
-}
+    // Parse command line arguments and call into the library's run function
+    bluepds::run().await.context("failed to run application")
+}

src/oauth.rs

···
         .expect("Failed to query PAR request");
 
     // Authenticate the user
-    use rsky_pds::schema::pds::account::dsl as AccountSchema;
-    use rsky_pds::schema::pds::actor::dsl as ActorSchema;
+    use crate::schema::pds::account::dsl as AccountSchema;
+    use crate::schema::pds::actor::dsl as ActorSchema;
     let username_clone = username.to_owned();
     let account = db
         .get()

src/schema.rs

···
     // Legacy tables
 
     diesel::table! {
-        pds.oauth_par_requests (request_uri) {
+        oauth_par_requests (request_uri) {
             request_uri -> Varchar,
             client_id -> Varchar,
             response_type -> Varchar,
···
         }
     }
     diesel::table! {
-        pds.oauth_authorization_codes (code) {
+        oauth_authorization_codes (code) {
             code -> Varchar,
             client_id -> Varchar,
             subject -> Varchar,
···
         }
     }
     diesel::table! {
-        pds.oauth_refresh_tokens (token) {
+        oauth_refresh_tokens (token) {
             token -> Varchar,
             client_id -> Varchar,
             subject -> Varchar,
···
         }
     }
     diesel::table! {
-        pds.oauth_used_jtis (jti) {
+        oauth_used_jtis (jti) {
             jti -> Varchar,
             issuer -> Varchar,
             created_at -> Int8,
···
     // Upcoming tables
 
     diesel::table! {
-        pds.authorization_request (id) {
+        account (did) {
+            did -> Varchar,
+            email -> Varchar,
+            recoveryKey -> Nullable<Varchar>,
+            password -> Varchar,
+            createdAt -> Varchar,
+            invitesDisabled -> Int2,
+            emailConfirmedAt -> Nullable<Varchar>,
+        }
+    }
+
+    diesel::table! {
+        account_pref (id) {
+            id -> Int4,
+            did -> Varchar,
+            name -> Varchar,
+            valueJson -> Nullable<Text>,
+        }
+    }
+
+    diesel::table! {
+        actor (did) {
+            did -> Varchar,
+            handle -> Nullable<Varchar>,
+            createdAt -> Varchar,
+            takedownRef -> Nullable<Varchar>,
+            deactivatedAt -> Nullable<Varchar>,
+            deleteAfter -> Nullable<Varchar>,
+        }
+    }
+
+    diesel::table! {
+        app_password (did, name) {
+            did -> Varchar,
+            name -> Varchar,
+            password -> Varchar,
+            createdAt -> Varchar,
+        }
+    }
+
+    diesel::table! {
+        authorization_request (id) {
             id -> Varchar,
             did -> Nullable<Varchar>,
             deviceId -> Nullable<Varchar>,
···
     }
 
     diesel::table! {
-        pds.device (id) {
+        backlink (uri, path) {
+            uri -> Varchar,
+            path -> Varchar,
+            linkTo -> Varchar,
+        }
+    }
+
+    diesel::table! {
+        blob (cid, did) {
+            cid -> Varchar,
+            did -> Varchar,
+            mimeType -> Varchar,
+            size -> Int4,
+            tempKey -> Nullable<Varchar>,
+            width -> Nullable<Int4>,
+            height -> Nullable<Int4>,
+            createdAt -> Varchar,
+            takedownRef -> Nullable<Varchar>,
+        }
+    }
+
+    diesel::table! {
+        device (id) {
             id -> Varchar,
             sessionId -> Nullable<Varchar>,
             userAgent -> Nullable<Varchar>,
···
     }
 
     diesel::table! {
-        pds.device_account (deviceId, did) {
+        device_account (deviceId, did) {
             did -> Varchar,
             deviceId -> Varchar,
             authenticatedAt -> Timestamptz,
···
     }
 
     diesel::table! {
-        pds.token (id) {
+        did_doc (did) {
+            did -> Varchar,
+            doc -> Text,
+            updatedAt -> Int8,
+        }
+    }
+
+    diesel::table! {
+        email_token (purpose, did) {
+            purpose -> Varchar,
+            did -> Varchar,
+            token -> Varchar,
+            requestedAt -> Varchar,
+        }
+    }
+
+    diesel::table! {
+        invite_code (code) {
+            code -> Varchar,
+            availableUses -> Int4,
+            disabled -> Int2,
+            forAccount -> Varchar,
+            createdBy -> Varchar,
+            createdAt -> Varchar,
+        }
+    }
+
+    diesel::table! {
+        invite_code_use (code, usedBy) {
+            code -> Varchar,
+            usedBy -> Varchar,
+            usedAt -> Varchar,
+        }
+    }
+
+    diesel::table! {
+        record (uri) {
+            uri -> Varchar,
+            cid -> Varchar,
+            did -> Varchar,
+            collection -> Varchar,
+            rkey -> Varchar,
+            repoRev -> Nullable<Varchar>,
+            indexedAt -> Varchar,
+            takedownRef -> Nullable<Varchar>,
+        }
+    }
+
+    diesel::table! {
+        record_blob (blobCid, recordUri) {
+            blobCid -> Varchar,
+            recordUri -> Varchar,
+            did -> Varchar,
+        }
+    }
+
+    diesel::table! {
+        refresh_token (id) {
+            id -> Varchar,
+            did -> Varchar,
+            expiresAt -> Varchar,
+            nextId -> Nullable<Varchar>,
+            appPasswordName -> Nullable<Varchar>,
+        }
+    }
+
+    diesel::table! {
+        repo_block (cid, did) {
+            cid -> Varchar,
+            did -> Varchar,
+            repoRev -> Varchar,
+            size -> Int4,
+            content -> Bytea,
+        }
+    }
+
+    diesel::table! {
+        repo_root (did) {
+            did -> Varchar,
+            cid -> Varchar,
+            rev -> Varchar,
+            indexedAt -> Varchar,
+        }
+    }
+
+    diesel::table! {
+        repo_seq (seq) {
+            seq -> Int8,
+            did -> Varchar,
+            eventType -> Varchar,
+            event -> Bytea,
+            invalidated -> Int2,
+            sequencedAt -> Varchar,
+        }
+    }
+
+    diesel::table! {
+        token (id) {
             id -> Varchar,
             did -> Varchar,
             tokenId -> Varchar,
···
     }
 
     diesel::table! {
-        pds.used_refresh_token (refreshToken) {
+        used_refresh_token (refreshToken) {
             refreshToken -> Varchar,
             tokenId -> Varchar,
         }
     }
+
+    diesel::allow_tables_to_appear_in_same_query!(
+        account,
+        account_pref,
+        actor,
+        app_password,
+        authorization_request,
+        backlink,
+        blob,
+        device,
+        device_account,
+        did_doc,
+        email_token,
+        invite_code,
+        invite_code_use,
+        record,
+        record_blob,
+        refresh_token,
+        repo_block,
+        repo_root,
+        repo_seq,
+        token,
+        used_refresh_token,
+    );
 }