prototype apis com atproto repo

Cargo.lock

···
  "reqwest 0.12.15",
  "reqwest-middleware",
  "rsky-common",
+ "rsky-identity",
  "rsky-lexicon",
  "rsky-pds",
  "rsky-repo",

Cargo.toml

···
 rsky-pds = { git = "https://github.com/blacksky-algorithms/rsky.git" }
 rsky-common = { git = "https://github.com/blacksky-algorithms/rsky.git" }
 rsky-lexicon = { git = "https://github.com/blacksky-algorithms/rsky.git" }
+rsky-identity = { git = "https://github.com/blacksky-algorithms/rsky.git" }
 
 # async in streams
 # async-stream = "0.3"

src/apis/com/atproto/repo/apply_writes.rs

···
 };
 use rsky_pds::sequencer::Sequencer;
 use rsky_repo::types::PreparedWrite;
+use std::collections::HashMap;
+use std::hash::RandomState;
 use std::str::FromStr;
+use std::sync::Arc;
 use tokio::sync::RwLock;
 
 async fn inner_apply_writes(
     body: ApplyWritesInput,
     user: AuthenticatedUser,
-    sequencer: &RwLock<Sequencer>,
-    actor_pools: std::collections::HashMap<String, ActorStorage>,
-    account_manager: &RwLock<AccountManager>,
+    sequencer: Arc<RwLock<Sequencer>>,
+    actor_pools: HashMap<String, ActorStorage>,
+    account_manager: Arc<RwLock<AccountManager>>,
 ) -> Result<()> {
     let tx: ApplyWritesInput = body;
     let ApplyWritesInput {
···
 #[axum::debug_handler(state = AppState)]
 pub(crate) async fn apply_writes(
     user: AuthenticatedUser,
-    State(state): State<AppState>,
+    State(db_actors): State<HashMap<String, ActorStorage, RandomState>>,
+    State(account_manager): State<Arc<RwLock<AccountManager>>>,
+    State(sequencer): State<Arc<RwLock<Sequencer>>>,
     Json(body): Json<ApplyWritesInput>,
 ) -> Result<(), ApiError> {
     tracing::debug!("@LOG: debug apply_writes {body:#?}");
-    let db_actors = state.db_actors;
-    let sequencer = &state.sequencer.sequencer;
-    let account_manager = &state.account_manager.account_manager;
     match inner_apply_writes(body, user, sequencer, db_actors, account_manager).await {
         Ok(()) => Ok(()),
         Err(error) => {

src/apis/com/atproto/repo/create_record.rs

···
+//! Create a single new repository record. Requires auth, implemented by PDS.
+use crate::account_manager::AccountManager;
+use crate::account_manager::helpers::account::AvailabilityFlags;
+use crate::{
+    actor_store::ActorStore,
+    auth::AuthenticatedUser,
+    error::ApiError,
+    serve::{ActorStorage, AppState},
+};
+use anyhow::{Result, bail};
+use axum::{Json, extract::State};
+use cidv10::Cid;
+use rsky_lexicon::com::atproto::repo::{CreateRecordInput, CreateRecordOutput};
+use rsky_pds::SharedIdResolver;
+use rsky_pds::repo::prepare::{
+    PrepareCreateOpts, PrepareDeleteOpts, prepare_create, prepare_delete,
+};
+use rsky_pds::sequencer::Sequencer;
+use rsky_repo::types::{PreparedDelete, PreparedWrite};
+use rsky_syntax::aturi::AtUri;
+use std::collections::HashMap;
+use std::hash::RandomState;
+use std::str::FromStr;
+use std::sync::Arc;
+use tokio::sync::RwLock;
+
+async fn inner_create_record(
+    body: CreateRecordInput,
+    user: AuthenticatedUser,
+    sequencer: Arc<RwLock<Sequencer>>,
+    actor_pools: std::collections::HashMap<String, ActorStorage>,
+    account_manager: Arc<RwLock<AccountManager>>,
+) -> Result<CreateRecordOutput> {
+    let CreateRecordInput {
+        repo,
+        collection,
+        record,
+        rkey,
+        validate,
+        swap_commit,
+    } = body;
+    let account = account_manager
+        .read()
+        .await
+        .get_account(
+            &repo,
+            Some(AvailabilityFlags {
+                include_deactivated: Some(true),
+                include_taken_down: None,
+            }),
+        )
+        .await?;
+    if let Some(account) = account {
+        if account.deactivated_at.is_some() {
+            bail!("Account is deactivated")
+        }
+        let did = account.did;
+        // if did != auth.access.credentials.unwrap().did.unwrap() {
+        if did != user.did() {
+            bail!("AuthRequiredError")
+        }
+        let swap_commit_cid = match swap_commit {
+            Some(swap_commit) => Some(Cid::from_str(&swap_commit)?),
+            None => None,
+        };
+        let write = prepare_create(PrepareCreateOpts {
+            did: did.clone(),
+            collection: collection.clone(),
+            record: serde_json::from_value(record)?,
+            rkey,
+            validate,
+            swap_cid: None,
+        })
+        .await?;
+
+        let did: &String = &did;
+        let mut actor_store = ActorStore::from_actor_pools(did, &actor_pools).await;
+        let backlink_conflicts: Vec<AtUri> = match validate {
+            Some(true) => {
+                let write_at_uri: AtUri = write.uri.clone().try_into()?;
+                actor_store
+                    .record
+                    .get_backlink_conflicts(&write_at_uri, &write.record)
+                    .await?
+            }
+            _ => Vec::new(),
+        };
+
+        let backlink_deletions: Vec<PreparedDelete> = backlink_conflicts
+            .iter()
+            .map(|at_uri| {
+                prepare_delete(PrepareDeleteOpts {
+                    did: at_uri.get_hostname().to_string(),
+                    collection: at_uri.get_collection(),
+                    rkey: at_uri.get_rkey(),
+                    swap_cid: None,
+                })
+            })
+            .collect::<Result<Vec<PreparedDelete>>>()?;
+        let mut writes: Vec<PreparedWrite> = vec![PreparedWrite::Create(write.clone())];
+        for delete in backlink_deletions {
+            writes.push(PreparedWrite::Delete(delete));
+        }
+        let commit = actor_store
+            .process_writes(writes.clone(), swap_commit_cid)
+            .await?;
+
+        _ = sequencer
+            .write()
+            .await
+            .sequence_commit(did.clone(), commit.clone())
+            .await?;
+        account_manager
+            .write()
+            .await
+            .update_repo_root(
+                did.to_string(),
+                commit.commit_data.cid,
+                commit.commit_data.rev,
+                &actor_pools,
+            )
+            .await?;
+
+        Ok(CreateRecordOutput {
+            uri: write.uri.clone(),
+            cid: write.cid.to_string(),
+        })
+    } else {
+        bail!("Could not find repo: `{repo}`")
+    }
+}
+
+/// Create a single new repository record. Requires auth, implemented by PDS.
+/// - POST /xrpc/com.atproto.repo.createRecord
+/// ### Request Body
+/// - `repo`: `at-identifier` // The handle or DID of the repo (aka, current account).
+/// - `collection`: `nsid` // The NSID of the record collection.
+/// - `rkey`: `string` // The record key. <= 512 characters.
+/// - `validate`: `boolean` // Can be set to 'false' to skip Lexicon schema validation of record data, 'true' to require it, or leave unset to validate only for known Lexicons.
+/// - `record`
+/// - `swap_commit`: `cid` // Compare and swap with the previous commit by CID.
+/// ### Responses
+/// - 200 OK: {`cid`: `cid`, `uri`: `at-uri`, `commit`: {`cid`: `cid`, `rev`: `tid`}, `validation_status`: [`valid`, `unknown`]}
+/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`, `InvalidSwap`]}
+/// - 401 Unauthorized
+#[axum::debug_handler(state = AppState)]
+pub async fn create_record(
+    user: AuthenticatedUser,
+    State(db_actors): State<HashMap<String, ActorStorage, RandomState>>,
+    State(account_manager): State<Arc<RwLock<AccountManager>>>,
+    State(sequencer): State<Arc<RwLock<Sequencer>>>,
+    Json(body): Json<CreateRecordInput>,
+) -> Result<Json<CreateRecordOutput>, ApiError> {
+    tracing::debug!("@LOG: debug create_record {body:#?}");
+    match inner_create_record(body, user, sequencer, db_actors, account_manager).await {
+        Ok(res) => Ok(Json(res)),
+        Err(error) => {
+            tracing::error!("@LOG: ERROR: {error}");
+            Err(ApiError::RuntimeError)
+        }
+    }
+}

src/apis/com/atproto/repo/delete_record.rs

···
+/// Delete a repository record, or ensure it doesn't exist. Requires auth, implemented by PDS.
+use crate::account_manager::AccountManager;
+use crate::account_manager::helpers::account::AvailabilityFlags;
+use crate::{
+    actor_store::ActorStore,
+    auth::AuthenticatedUser,
+    error::ApiError,
+    serve::{ActorStorage, AppState},
+};
+use anyhow::{Result, bail};
+use axum::{Json, extract::State};
+use cidv10::Cid;
+use rsky_lexicon::com::atproto::repo::DeleteRecordInput;
+use rsky_pds::repo::prepare::{PrepareDeleteOpts, prepare_delete};
+use rsky_pds::sequencer::Sequencer;
+use rsky_repo::types::PreparedWrite;
+use rsky_syntax::aturi::AtUri;
+use std::collections::HashMap;
+use std::hash::RandomState;
+use std::str::FromStr;
+use std::sync::Arc;
+use tokio::sync::RwLock;
+
+async fn inner_delete_record(
+    body: DeleteRecordInput,
+    user: AuthenticatedUser,
+    sequencer: Arc<RwLock<Sequencer>>,
+    actor_pools: HashMap<String, ActorStorage>,
+    account_manager: Arc<RwLock<AccountManager>>,
+) -> Result<()> {
+    let DeleteRecordInput {
+        repo,
+        collection,
+        rkey,
+        swap_record,
+        swap_commit,
+    } = body;
+    let account = account_manager
+        .read()
+        .await
+        .get_account(
+            &repo,
+            Some(AvailabilityFlags {
+                include_deactivated: Some(true),
+                include_taken_down: None,
+            }),
+        )
+        .await?;
+    match account {
+        None => bail!("Could not find repo: `{repo}`"),
+        Some(account) if account.deactivated_at.is_some() => bail!("Account is deactivated"),
+        Some(account) => {
+            let did = account.did;
+            // if did != auth.access.credentials.unwrap().did.unwrap() {
+            if did != user.did() {
+                bail!("AuthRequiredError")
+            }
+
+            let swap_commit_cid = match swap_commit {
+                Some(swap_commit) => Some(Cid::from_str(&swap_commit)?),
+                None => None,
+            };
+            let swap_record_cid = match swap_record {
+                Some(swap_record) => Some(Cid::from_str(&swap_record)?),
+                None => None,
+            };
+
+            let write = prepare_delete(PrepareDeleteOpts {
+                did: did.clone(),
+                collection,
+                rkey,
+                swap_cid: swap_record_cid,
+            })?;
+            let mut actor_store = ActorStore::from_actor_pools(&did, &actor_pools).await;
+            let write_at_uri: AtUri = write.uri.clone().try_into()?;
+            let record = actor_store
+                .record
+                .get_record(&write_at_uri, None, Some(true))
+                .await?;
+            let commit = match record {
+                None => return Ok(()), // No-op if record already doesn't exist
+                Some(_) => {
+                    actor_store
+                        .process_writes(vec![PreparedWrite::Delete(write.clone())], swap_commit_cid)
+                        .await?
+                }
+            };
+
+            _ = sequencer
+                .write()
+                .await
+                .sequence_commit(did.clone(), commit.clone())
+                .await?;
+            account_manager
+                .write()
+                .await
+                .update_repo_root(
+                    did,
+                    commit.commit_data.cid,
+                    commit.commit_data.rev,
+                    &actor_pools,
+                )
+                .await?;
+
+            Ok(())
+        }
+    }
+}
+
+/// Delete a repository record, or ensure it doesn't exist. Requires auth, implemented by PDS.
+/// - POST /xrpc/com.atproto.repo.deleteRecord
+/// ### Request Body
+/// - `repo`: `at-identifier` // The handle or DID of the repo (aka, current account).
+/// - `collection`: `nsid` // The NSID of the record collection.
+/// - `rkey`: `string` // The record key. <= 512 characters.
+/// - `swap_record`: `boolean` // Compare and swap with the previous record by CID.
+/// - `swap_commit`: `cid` // Compare and swap with the previous commit by CID.
+/// ### Responses
+/// - 200 OK: {"commit": {"cid": "string","rev": "string"}}
+/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`, `InvalidSwap`]}
+/// - 401 Unauthorized
+#[axum::debug_handler(state = AppState)]
+pub async fn delete_record(
+    user: AuthenticatedUser,
+    State(db_actors): State<HashMap<String, ActorStorage, RandomState>>,
+    State(account_manager): State<Arc<RwLock<AccountManager>>>,
+    State(sequencer): State<Arc<RwLock<Sequencer>>>,
+    Json(body): Json<DeleteRecordInput>,
+) -> Result<(), ApiError> {
+    match inner_delete_record(body, user, sequencer, db_actors, account_manager).await {
+        Ok(()) => Ok(()),
+        Err(error) => {
+            tracing::error!("@LOG: ERROR: {error}");
+            Err(ApiError::RuntimeError)
+        }
+    }
+}

src/apis/com/atproto/repo/describe_repo.rs

···
+//! Get information about an account and repository, including the list of collections. Does not require auth.
+use crate::account_manager::AccountManager;
+use crate::serve::ActorStorage;
+use crate::{actor_store::ActorStore, error::ApiError, serve::AppState};
+use anyhow::{Result, bail};
+use axum::extract::Query;
+use axum::{Json, extract::State};
+use rsky_identity::IdResolver;
+use rsky_identity::types::DidDocument;
+use rsky_lexicon::com::atproto::repo::DescribeRepoOutput;
+use rsky_syntax::handle::INVALID_HANDLE;
+use std::collections::HashMap;
+use std::hash::RandomState;
+use std::sync::Arc;
+use tokio::sync::RwLock;
+
+async fn inner_describe_repo(
+    repo: String,
+    id_resolver: Arc<RwLock<IdResolver>>,
+    actor_pools: HashMap<String, ActorStorage>,
+    account_manager: Arc<RwLock<AccountManager>>,
+) -> Result<DescribeRepoOutput> {
+    let account = account_manager
+        .read()
+        .await
+        .get_account(&repo, None)
+        .await?;
+    match account {
+        None => bail!("Cound not find user: `{repo}`"),
+        Some(account) => {
+            let mut lock = id_resolver.write().await;
+            let did_doc: DidDocument = match lock.did.ensure_resolve(&account.did, None).await {
+                Err(err) => bail!("Could not resolve DID: `{err}`"),
+                Ok(res) => res,
+            };
+            let handle = rsky_common::get_handle(&did_doc);
+            let handle_is_correct = handle == account.handle;
+
+            let actor_store =
+                ActorStore::from_actor_pools(&account.did.clone(), &actor_pools).await;
+            let collections = actor_store.record.list_collections().await?;
+
+            Ok(DescribeRepoOutput {
+                handle: account.handle.unwrap_or(INVALID_HANDLE.to_string()),
+                did: account.did,
+                did_doc: serde_json::to_value(did_doc)?,
+                collections,
+                handle_is_correct,
+            })
+        }
+    }
+}
+
+/// Get information about an account and repository, including the list of collections. Does not require auth.
+/// - GET /xrpc/com.atproto.repo.describeRepo
+/// ### Query Parameters
+/// - `repo`: `at-identifier` // The handle or DID of the repo.
+/// ### Responses
+/// - 200 OK: {"handle": "string","did": "string","didDoc": {},"collections": [string],"handleIsCorrect": true} \
+///   handeIsCorrect - boolean - Indicates if handle is currently valid (resolves bi-directionally)
+/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`]}
+/// - 401 Unauthorized
+#[tracing::instrument(skip_all)]
+#[axum::debug_handler(state = AppState)]
+pub async fn describe_repo(
+    Query(input): Query<atrium_api::com::atproto::repo::describe_repo::ParametersData>,
+    State(db_actors): State<HashMap<String, ActorStorage, RandomState>>,
+    State(account_manager): State<Arc<RwLock<AccountManager>>>,
+    State(id_resolver): State<Arc<RwLock<IdResolver>>>,
+) -> Result<Json<DescribeRepoOutput>, ApiError> {
+    match inner_describe_repo(input.repo.into(), id_resolver, db_actors, account_manager).await {
+        Ok(res) => Ok(Json(res)),
+        Err(error) => {
+            tracing::error!("{error:?}");
+            Err(ApiError::RuntimeError)
+        }
+    }
+}

src/apis/com/atproto/repo/ex.rs

···
+//!
+use crate::account_manager::AccountManager;
+use crate::serve::ActorStorage;
+use crate::{actor_store::ActorStore, error::ApiError, serve::AppState};
+use anyhow::{Result, bail};
+use axum::extract::Query;
+use axum::{Json, extract::State};
+use rsky_identity::IdResolver;
+use rsky_pds::sequencer::Sequencer;
+use std::collections::HashMap;
+use std::hash::RandomState;
+use std::sync::Arc;
+use tokio::sync::RwLock;
+
+async fn fun(
+    actor_pools: HashMap<String, ActorStorage>,
+    account_manager: Arc<RwLock<AccountManager>>,
+    id_resolver: Arc<RwLock<IdResolver>>,
+    sequencer: Arc<RwLock<Sequencer>>,
+) -> Result<_> {
+    todo!();
+}
+
+///
+#[tracing::instrument(skip_all)]
+#[axum::debug_handler(state = AppState)]
+pub async fn fun(
+    Query(input): Query<atrium_api::com::atproto::repo::describe_repo::ParametersData>,
+    State(db_actors): State<HashMap<String, ActorStorage, RandomState>>,
+    State(account_manager): State<Arc<RwLock<AccountManager>>>,
+    State(id_resolver): State<Arc<RwLock<IdResolver>>>,
+    State(sequencer): State<Arc<RwLock<Sequencer>>>,
+) -> Result<Json<_>, ApiError> {
+    todo!();
+}

src/apis/com/atproto/repo/get_record.rs

···
+//! Get a single record from a repository. Does not require auth.
+use crate::account_manager::AccountManager;
+use crate::serve::ActorStorage;
+use crate::{actor_store::ActorStore, error::ApiError, serve::AppState};
+use anyhow::{Result, bail};
+use axum::extract::Query;
+use axum::{Json, extract::State};
+use rsky_identity::IdResolver;
+use rsky_lexicon::com::atproto::repo::GetRecordOutput;
+use rsky_pds::pipethrough::{OverrideOpts, ProxyRequest, pipethrough};
+use rsky_pds::sequencer::Sequencer;
+use rsky_syntax::aturi::AtUri;
+use std::collections::HashMap;
+use std::hash::RandomState;
+use std::sync::Arc;
+use tokio::sync::RwLock;
+
+async fn inner_get_record(
+    repo: String,
+    collection: String,
+    rkey: String,
+    cid: Option<String>,
+    // req: ProxyRequest<'_>,
+    actor_pools: HashMap<String, ActorStorage>,
+    account_manager: Arc<RwLock<AccountManager>>,
+) -> Result<GetRecordOutput> {
+    let did = account_manager
+        .read()
+        .await
+        .get_did_for_actor(&repo, None)
+        .await?;
+
+    // fetch from pds if available, if not then fetch from appview
+    if let Some(did) = did {
+        let uri = AtUri::make(did.clone(), Some(collection), Some(rkey))?;
+
+        let mut actor_store = ActorStore::from_actor_pools(&did, &actor_pools).await;
+
+        match actor_store.record.get_record(&uri, cid, None).await {
+            Ok(Some(record)) if record.takedown_ref.is_none() => Ok(GetRecordOutput {
+                uri: uri.to_string(),
+                cid: Some(record.cid),
+                value: serde_json::to_value(record.value)?,
+            }),
+            _ => bail!("Could not locate record: `{uri}`"),
+        }
+    } else {
+        // match req.cfg.bsky_app_view {
+        //     None => bail!("Could not locate record"),
+        //     Some(_) => match pipethrough(
+        //         &req,
+        //         None,
+        //         OverrideOpts {
+        //             aud: None,
+        //             lxm: None,
+        //         },
+        //     )
+        //     .await
+        //     {
+        //         Err(error) => {
+        //             tracing::error!("@LOG: ERROR: {error}");
+        bail!("Could not locate record")
+        //         }
+        //         Ok(res) => {
+        //             let output: GetRecordOutput = serde_json::from_slice(res.buffer.as_slice())?;
+        //             Ok(output)
+        //         }
+        //     },
+        // }
+    }
+}
+
+/// Get a single record from a repository. Does not require auth.
+/// - GET /xrpc/com.atproto.repo.getRecord
+/// ### Query Parameters
+/// - `repo`: `at-identifier` // The handle or DID of the repo.
+/// - `collection`: `nsid` // The NSID of the record collection.
+/// - `rkey`: `string` // The record key. <= 512 characters.
+/// - `cid`: `cid` // The CID of the version of the record. If not specified, then return the most recent version.
+/// ### Responses
+/// - 200 OK: {"uri": "string","cid": "string","value": {}}
+/// - 400 Bad Request: {error:[`InvalidRequest`, `ExpiredToken`, `InvalidToken`, `RecordNotFound`]}
+/// - 401 Unauthorized
+#[tracing::instrument(skip_all)]
+#[axum::debug_handler(state = AppState)]
+pub async fn get_record(
+    Query(input): Query<ParametersData>,
+    State(db_actors): State<HashMap<String, ActorStorage, RandomState>>,
+    State(account_manager): State<Arc<RwLock<AccountManager>>>,
+) -> Result<Json<GetRecordOutput>, ApiError> {
+    let repo = input.repo;
+    let collection = input.collection;
+    let rkey = input.rkey;
+    let cid = input.cid;
+    // let req: ProxyRequest = todo!(); // TODO: Implement service proxy
+    match inner_get_record(
+        repo,
+        collection,
+        rkey,
+        cid,
+        // req,
+        db_actors,
+        account_manager,
+    )
+    .await
+    {
+        Ok(res) => Ok(Json(res)),
+        Err(error) => {
+            tracing::error!("@LOG: ERROR: {error}");
+            Err(ApiError::RecordNotFound)
+        }
+    }
+}
+
+#[derive(serde::Deserialize, Debug)]
+pub struct ParametersData {
+    pub cid: Option<String>,
+    pub collection: String,
+    pub repo: String,
+    pub rkey: String,
+}

src/apis/com/atproto/repo/import_repo.rs

···
+use crate::account_manager::AccountManager;
+use crate::auth::AuthenticatedUser;
+use crate::serve::ActorStorage;
+use crate::{actor_store::ActorStore, error::ApiError, serve::AppState};
+use anyhow::{Result, bail};
+use axum::extract::Query;
+use axum::{Json, extract::State};
+use cidv10::Cid;
+use futures::{StreamExt, stream};
+use reqwest::header;
+use rsky_common::env::env_int;
+use rsky_identity::IdResolver;
+use rsky_pds::repo::prepare::{
+    PrepareCreateOpts, PrepareDeleteOpts, PrepareUpdateOpts, prepare_create, prepare_delete,
+    prepare_update,
+};
+use rsky_pds::sequencer::Sequencer;
+use rsky_repo::block_map::BlockMap;
+use rsky_repo::car::{CarWithRoot, read_stream_car_with_root};
+use rsky_repo::parse::get_and_parse_record;
+use rsky_repo::repo::Repo;
+use rsky_repo::sync::consumer::{VerifyRepoInput, verify_diff};
+use rsky_repo::types::{PreparedWrite, RecordWriteDescript, VerifiedDiff};
+use serde::Deserialize;
+use std::collections::HashMap;
+use std::hash::RandomState;
+use std::num::NonZeroU64;
+use std::sync::Arc;
+use tokio::sync::RwLock;
+
+struct ImportRepoInput {
+    car_with_root: CarWithRoot,
+}
+
+impl<'de> Deserialize<'de> for ImportRepoInput {
+    fn deserialize<D>(deserializer: D) -> core::result::Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        async fn from_data(req: &Request<'_>, data: Data<'_>) -> Result<CarWithRoot, ApiError> {
+            let max_import_size = env_int("IMPORT_REPO_LIMIT").unwrap_or(100).megabytes();
+            match req.headers().get_one(header::CONTENT_LENGTH.as_ref()) {
+                None => {
+                    let error =
+                        ApiError::InvalidRequest("Missing content-length header".to_string());
+                    req.local_cache(|| Some(error.clone()));
+                    return Err(error);
+                }
+                Some(res) => match res.parse::<NonZeroU64>() {
+                    Ok(content_length) => {
+                        if content_length.get().bytes() > max_import_size {
+                            let error = ApiError::InvalidRequest(format!(
+                                "Content-Length is greater than maximum of {max_import_size}"
+                            ));
+                            req.local_cache(|| Some(error.clone()));
+                            return Err(error);
+                        }
+
+                        let import_datastream = data.open(content_length.get().bytes());
+                        read_stream_car_with_root(import_datastream).await
+                    }
+                    Err(_error) => {
+                        tracing::error!("{}", format!("Error parsing content-length\n{_error}"));
+                        let error =
+                            ApiError::InvalidRequest("Error parsing content-length".to_string());
+                        req.local_cache(|| Some(error.clone()));
+                        Err(error)
+                    }
+                },
+            }
+        }
+    }
+}
+
+// #[rocket::async_trait]
+// impl<'r> FromData<'r> for ImportRepoInput {
+//     type Error = ApiError;
+
+//     #[tracing::instrument(skip_all)]
+//     async fn from_data(req: &'r Request<'_>, data: Data<'r>) -> Outcome<'r, Self, Self::Error> {
+//         let max_import_size = env_int("IMPORT_REPO_LIMIT").unwrap_or(100).megabytes();
+//         match req.headers().get_one(header::CONTENT_LENGTH.as_ref()) {
+//             None => {
+//                 let error = ApiError::InvalidRequest("Missing content-length header".to_string());
+//                 req.local_cache(|| Some(error.clone()));
+//                 Outcome::Error((Status::BadRequest, error))
+//             }
+//             Some(res) => match res.parse::<NonZeroU64>() {
+//                 Ok(content_length) => {
+//                     if content_length.get().bytes() > max_import_size {
+//                         let error = ApiError::InvalidRequest(format!(
+//                             "Content-Length is greater than maximum of {max_import_size}"
+//                         ));
+//                         req.local_cache(|| Some(error.clone()));
+//                         return Outcome::Error((Status::BadRequest, error));
+//                     }
+
+//                     let import_datastream = data.open(content_length.get().bytes());
+//                     match read_stream_car_with_root(import_datastream).await {
+//                         Ok(car_with_root) => Outcome::Success(ImportRepoInput { car_with_root }),
+//                         Err(error) => {
+//                             let error = ApiError::InvalidRequest(error.to_string());
+//                             req.local_cache(|| Some(error.clone()));
+//                             Outcome::Error((Status::BadRequest, error))
+//                         }
+//                     }
+//                 }
+//                 Err(_error) => {
+//                     tracing::error!("{}", format!("Error parsing content-length\n{_error}"));
+//                     let error =
+//                         ApiError::InvalidRequest("Error parsing content-length".to_string());
+//                     req.local_cache(|| Some(error.clone()));
+//                     Outcome::Error((Status::BadRequest, error))
+//                 }
+//             },
+//         }
+//     }
+// }
+
+#[tracing::instrument(skip_all)]
+#[axum::debug_handler(state = AppState)]
+pub async fn import_repo(
+    // auth: AccessFullImport,
+    user: AuthenticatedUser,
+    Query(import_repo_input): Query<ImportRepoInput>,
+    State(actor_pools): State<HashMap<String, ActorStorage, RandomState>>,
+) -> Result<(), ApiError> {
+    // let requester = auth.access.credentials.unwrap().did.unwrap();
+    let requester = user.did();
+    let mut actor_store = ActorStore::from_actor_pools(&requester, &actor_pools).await;
+
+    // Get current repo if it exists
+    let curr_root: Option<Cid> = actor_store.get_repo_root().await;
+    let curr_repo: Option<Repo> = match curr_root {
+        None => None,
+        Some(_root) => Some(Repo::load(actor_store.storage.clone(), curr_root).await?),
+    };
+
+    // Process imported car
+    let car_with_root = import_repo_input.car_with_root;
+
+    // Get verified difference from current repo and imported repo
+    let mut imported_blocks: BlockMap = car_with_root.blocks;
+    let imported_root: Cid = car_with_root.root;
+    let opts = VerifyRepoInput {
+        ensure_leaves: Some(false),
+    };
+
+    let diff: VerifiedDiff = match verify_diff(
+        curr_repo,
+        &mut imported_blocks,
+        imported_root,
+        None,
+        None,
+        Some(opts),
+    )
+    .await
+    {
+        Ok(res) => res,
+        Err(error) => {
+            tracing::error!("{:?}", error);
+            return Err(ApiError::RuntimeError);
+        }
+    };
+
+    let commit_data = diff.commit;
+    let prepared_writes: Vec<PreparedWrite> =
+        prepare_import_repo_writes(requester, diff.writes, &imported_blocks).await?;
+    match actor_store
+        .process_import_repo(commit_data, prepared_writes)
+        .await
+    {
+        Ok(_res) => {}
+        Err(error) => {
+            tracing::error!("Error importing repo\n{error}");
+            return Err(ApiError::RuntimeError);
+        }
+    }
+
+    Ok(())
+}
+
+/// Converts list of RecordWriteDescripts into a list of PreparedWrites
+async fn prepare_import_repo_writes(
+    _did: String,
+    writes: Vec<RecordWriteDescript>,
+    blocks: &BlockMap,
+) -> Result<Vec<PreparedWrite>, ApiError> {
+    match stream::iter(writes)
+        .then(|write| {
+            let did = _did.clone();
+            async move {
+                Ok::<PreparedWrite, anyhow::Error>(match write {
+                    RecordWriteDescript::Create(write) => {
+                        let parsed_record = get_and_parse_record(blocks, write.cid)?;
+                        PreparedWrite::Create(
+                            prepare_create(PrepareCreateOpts {
+                                did: did.clone(),
+                                collection: write.collection,
+                                rkey: Some(write.rkey),
+                                swap_cid: None,
+                                record: parsed_record.record,
+                                validate: Some(true),
+                            })
+                            .await?,
+                        )
+                    }
+                    RecordWriteDescript::Update(write) => {
+                        let parsed_record = get_and_parse_record(blocks, write.cid)?;
+                        PreparedWrite::Update(
+                            prepare_update(PrepareUpdateOpts {
+                                did: did.clone(),
+                                collection: write.collection,
+                                rkey: write.rkey,
+                                swap_cid: None,
+                                record: parsed_record.record,
+                                validate: Some(true),
+                            })
+                            .await?,
+                        )
+                    }
+                    RecordWriteDescript::Delete(write) => {
+                        PreparedWrite::Delete(prepare_delete(PrepareDeleteOpts {
+                            did: did.clone(),
+                            collection: write.collection,
+                            rkey: write.rkey,
+                            swap_cid: None,
+                        })?)
+                    }
+                })
+            }
+        })
+        .collect::<Vec<_>>()
+        .await
+        .into_iter()
+        .collect::<anyhow::Result<Vec<PreparedWrite>, _>>()
+    {
+        Ok(res) => Ok(res),
+        Err(error) => {
+            tracing::error!("Error preparing import repo writes\n{error}");
+            Err(ApiError::RuntimeError)
+        }
+    }
+}

src/apis/com/atproto/repo/mod.rs

···
 use atrium_api::com::atproto::repo;
-use axum::{Router, routing::post};
+use axum::{
+    Router,
+    routing::{get, post},
+};
 use constcat::concat;
 
 use crate::serve::AppState;
 
 pub mod apply_writes;
-// pub mod create_record;
-// pub mod delete_record;
-// pub mod describe_repo;
-// pub mod get_record;
+pub mod create_record;
+pub mod delete_record;
+pub mod describe_repo;
+pub mod get_record;
 // pub mod import_repo;
 // pub mod list_missing_blobs;
 // pub mod list_records;
···
 ///     - [ ] xx /xrpc/com.atproto.repo.importRepo
 // - [ ] xx /xrpc/com.atproto.repo.listMissingBlobs
 pub(crate) fn routes() -> Router<AppState> {
-    Router::new().route(
-        concat!("/", repo::apply_writes::NSID),
-        post(apply_writes::apply_writes),
-    )
-    // .route(concat!("/", repo::create_record::NSID), post(create_record))
-    // .route(concat!("/", repo::put_record::NSID), post(put_record))
-    // .route(concat!("/", repo::delete_record::NSID), post(delete_record))
-    // .route(concat!("/", repo::upload_blob::NSID), post(upload_blob))
-    // .route(concat!("/", repo::describe_repo::NSID), get(describe_repo))
-    // .route(concat!("/", repo::get_record::NSID), get(get_record))
+    Router::new()
+        .route(
+            concat!("/", repo::apply_writes::NSID),
+            post(apply_writes::apply_writes),
+        )
+        .route(
+            concat!("/", repo::create_record::NSID),
+            post(create_record::create_record),
+        )
+        // .route(concat!("/", repo::put_record::NSID), post(put_record))
+        .route(
+            concat!("/", repo::delete_record::NSID),
+            post(delete_record::delete_record),
+        )
+        // .route(concat!("/", repo::upload_blob::NSID), post(upload_blob))
+        .route(
+            concat!("/", repo::describe_repo::NSID),
+            get(describe_repo::describe_repo),
+        )
+        .route(
+            concat!("/", repo::get_record::NSID),
+            get(get_record::get_record),
+        )
     // .route(concat!("/", repo::import_repo::NSID), post(todo))
     // .route(concat!("/", repo::list_missing_blobs::NSID), get(todo))
     // .route(concat!("/", repo::list_records::NSID), get(list_records))

src/error.rs

···
     }
 }
 
+impl From<anyhow::Error> for ApiError {
+    fn from(_value: anyhow::Error) -> Self {
+        Self::RuntimeError
+    }
+}
+
 impl From<handle::errors::Error> for ApiError {
     fn from(value: handle::errors::Error) -> Self {
         match value.kind {

src/serve.rs

···
 use diesel_migrations::{EmbeddedMigrations, embed_migrations};
 use figment::{Figment, providers::Format as _};
 use http_cache_reqwest::{CacheMode, HttpCacheOptions, MokaManager};
+use rsky_common::env::env_list;
+use rsky_identity::IdResolver;
+use rsky_identity::types::{DidCache, IdentityResolverOpts};
+use rsky_pds::SharedIdResolver;
 use rsky_pds::{crawlers::Crawlers, sequencer::Sequencer};
 use serde::{Deserialize, Serialize};
+use std::env;
 use std::{
     net::{IpAddr, Ipv4Addr, SocketAddr},
     path::PathBuf,
···
 
 /// Embedded migrations
 pub const MIGRATIONS: EmbeddedMigrations = embed_migrations!("./migrations");
+pub const MIGRATIONS_ACTOR: EmbeddedMigrations = embed_migrations!("./migrations_actor");
 
 /// 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 Shared Sequencer which requests crawls from upstream relays and emits events to the firehose.
-pub struct SharedSequencer {
-    /// The sequencer instance.
-    pub sequencer: RwLock<Sequencer>,
-}
 
 #[expect(
     clippy::arbitrary_source_item_ordering,
···
     /// The simple HTTP client.
     pub simple_client: reqwest::Client,
     /// The firehose producer.
-    pub sequencer: Arc<SharedSequencer>,
+    pub sequencer: Arc<RwLock<Sequencer>>,
     /// The account manager.
-    pub account_manager: Arc<SharedAccountManager>,
+    pub account_manager: Arc<RwLock<AccountManager>>,
+    /// The ID resolver.
+    pub id_resolver: Arc<RwLock<IdResolver>>,
 
     /// The signing key.
     pub signing_key: SigningKey,
···
         .iter()
         .map(|s| s.to_string())
         .collect();
-    let sequencer = Arc::new(SharedSequencer {
-        sequencer: RwLock::new(Sequencer::new(
-            Crawlers::new(hostname, crawlers.clone()),
-            None,
-        )),
-    });
-    let account_manager = SharedAccountManager {
-        account_manager: RwLock::new(AccountManager::new(pool.clone())),
+    let sequencer = Arc::new(RwLock::new(Sequencer::new(
+        Crawlers::new(hostname, crawlers.clone()),
+        None,
+    )));
+    let account_manager = Arc::new(RwLock::new(AccountManager::new(pool.clone())));
+    let plc_url = if cfg!(debug_assertions) {
+        "http://localhost:8000".to_owned() // dummy for debug
+    } else {
+        env::var("PDS_DID_PLC_URL").unwrap_or("https://plc.directory".to_owned()) // TODO: toml config
     };
+    let id_resolver = Arc::new(RwLock::new(IdResolver::new(IdentityResolverOpts {
+        timeout: None,
+        plc_url: Some(plc_url),
+        did_cache: Some(DidCache::new(None, None)),
+        backup_nameservers: Some(env_list("PDS_HANDLE_BACKUP_NAMESERVERS")),
+    })));
 
     let addr = config
         .listen_address
···
             client: client.clone(),
             simple_client,
             sequencer: sequencer.clone(),
-            account_manager: Arc::new(account_manager),
+            account_manager,
+            id_resolver,
             signing_key: skey,
             rotation_key: rkey,
         });
···
         info!("debug mode: not requesting crawl");
     } else {
         info!("requesting crawl from upstream relays");
-        let mut background_sequencer = sequencer.sequencer.write().await.clone();
+        let mut background_sequencer = sequencer.write().await.clone();
         drop(tokio::spawn(
             async move { background_sequencer.start().await },
         ));