Something went wrong. Try again.
A downstream fork of Tranquil PDS for use in margin.cafe
Something went wrong. Try again.
16 kB · 384 lines
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385use axum::{ Json, extract::{Query, State},};use chrono::Utc;use serde::{Deserialize, Serialize};use serde_json::{Value, json};use tracing::{error, warn};use tranquil_pds::api::error::ApiError;use tranquil_pds::auth::{Admin, Auth};use tranquil_pds::state::AppState;use tranquil_pds::types::{CidLink, Did};
#[derive(Deserialize)]pub struct GetSubjectStatusParams { pub did: Option<String>, pub uri: Option<String>, pub blob: Option<String>,}
#[derive(Serialize)]pub struct SubjectStatus { pub subject: serde_json::Value, pub takedown: Option<StatusAttr>, pub deactivated: Option<StatusAttr>,}
#[derive(Serialize)]#[serde(rename_all = "camelCase")]pub struct StatusAttr { pub applied: bool, pub r#ref: Option<String>,}
pub async fn get_subject_status( State(state): State<AppState>, _auth: Auth<Admin>, Query(params): Query<GetSubjectStatusParams>,) -> Result<Json<SubjectStatus>, ApiError> { if params.did.is_none() && params.uri.is_none() && params.blob.is_none() { return Err(ApiError::InvalidRequest( "Must provide did, uri, or blob".into(), )); } if let Some(did_str) = ¶ms.did { let did: Did = did_str .parse() .map_err(|_| ApiError::InvalidDid("Invalid DID format".into()))?; match state.repos.user.get_status_by_did(&did).await { Ok(Some(status)) => { let deactivated = status.deactivated_at.map(|_| StatusAttr { applied: true, r#ref: None, }); let takedown = status.takedown_ref.as_ref().map(|r| StatusAttr { applied: true, r#ref: Some(r.clone()), }); return Ok(Json(SubjectStatus { subject: json!({ "$type": "com.atproto.admin.defs#repoRef", "did": did_str }), takedown, deactivated, })); } Ok(None) => { return Err(ApiError::SubjectNotFound); } Err(e) => { error!("DB error in get_subject_status: {:?}", e); return Err(ApiError::InternalError(None)); } } } if let Some(uri_str) = ¶ms.uri { let cid: CidLink = uri_str .parse() .map_err(|_| ApiError::InvalidRequest("Invalid CID format".into()))?; match state.repos.repo.get_record_by_cid(&cid).await { Ok(Some(record)) => { let takedown = record.takedown_ref.as_ref().map(|r| StatusAttr { applied: true, r#ref: Some(r.clone()), }); return Ok(Json(SubjectStatus { subject: json!({ "$type": "com.atproto.repo.strongRef", "uri": uri_str, "cid": uri_str }), takedown, deactivated: None, })); } Ok(None) => { return Err(ApiError::RecordNotFound); } Err(e) => { error!("DB error in get_subject_status: {:?}", e); return Err(ApiError::InternalError(None)); } } } if let Some(blob_cid_str) = ¶ms.blob { let blob_cid: CidLink = blob_cid_str .parse() .map_err(|_| ApiError::InvalidRequest("Invalid CID format".into()))?; let did = params.did.as_ref().ok_or_else(|| { ApiError::InvalidRequest("Must provide a did to request blob state".into()) })?; match state.repos.blob.get_blob_with_takedown(&blob_cid).await { Ok(Some(blob)) => { let takedown = blob.takedown_ref.as_ref().map(|r| StatusAttr { applied: true, r#ref: Some(r.clone()), }); return Ok(Json(SubjectStatus { subject: json!({ "$type": "com.atproto.admin.defs#repoBlobRef", "did": did, "cid": blob.cid }), takedown, deactivated: None, })); } Ok(None) => { return Err(ApiError::BlobNotFound(None)); } Err(e) => { error!("DB error in get_subject_status: {:?}", e); return Err(ApiError::InternalError(None)); } } } Err(ApiError::InvalidRequest("Invalid subject type".into()))}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]pub struct UpdateSubjectStatusInput { pub subject: serde_json::Value, pub takedown: Option<StatusAttrInput>, pub deactivated: Option<StatusAttrInput>,}
#[derive(Deserialize)]pub struct StatusAttrInput { pub applied: bool, pub r#ref: Option<String>,}
pub async fn update_subject_status( State(state): State<AppState>, _auth: Auth<Admin>, Json(input): Json<UpdateSubjectStatusInput>,) -> Result<Json<serde_json::Value>, ApiError> { let subject_type = input.subject.get("$type").and_then(Value::as_str); match subject_type { Some("com.atproto.admin.defs#repoRef") => { let did_str = input.subject.get("did").and_then(Value::as_str); if let Some(did_str) = did_str { if input.takedown.as_ref().is_some_and(|status| status.applied) && input .deactivated .as_ref() .is_some_and(|status| !status.applied) { return Err(ApiError::InvalidRequest( "Cannot activate and takedown an account at the same time".into(), )); } let did: Did = match did_str.parse() { Ok(d) => d, Err(_) => return Err(ApiError::InvalidDid("Invalid DID format".into())), }; let takedown_ref = input.takedown.as_ref().and_then(|takedown| { takedown.applied.then(|| { takedown .r#ref .clone() .unwrap_or_else(|| Utc::now().to_rfc3339()) }) }); if input.takedown.is_some() { state .repos .user .set_user_takedown(&did, takedown_ref.as_deref()) .await .map_err(|e| { error!("Failed to update user takedown status for {}: {:?}", did, e); ApiError::InternalError(Some("Failed to update takedown status".into())) })?; } if let Some(deactivated) = &input.deactivated { let result = if deactivated.applied { state.repos.user.deactivate_account(&did, None).await } else { let status = state .repos .user .get_status_by_did(&did) .await .map_err(|e| { error!("Failed to read account status for {}: {:?}", did, e); ApiError::InternalError(None) })? .ok_or(ApiError::AccountNotFound)?; if status.takedown_ref.is_some() { return Err(ApiError::AccountNotFound); } state.repos.user.activate_account(&did).await }; result.map_err(|e| { error!( "Failed to update user deactivation status for {}: {:?}", did, e ); ApiError::InternalError(Some("Failed to update deactivation status".into())) })?; } let takedown_update = input.takedown.as_ref().map(|t| t.applied); let deactivated_update = input.deactivated.as_ref().map(|d| d.applied); if takedown_update.is_some() || deactivated_update.is_some() { state .repos .repo .update_repo_status( &did, takedown_update, takedown_ref.as_deref(), deactivated_update, ) .await .map_err(|e| { error!("Failed to sync account status to repo backend: {e:?}"); ApiError::InternalError(Some("Failed to update account status".into())) })?; } if input.takedown.as_ref().is_some_and(|status| status.applied) { state .repos .session .delete_sessions_by_did(&did) .await .map_err(|e| { error!("Failed to revoke sessions for taken-down account: {e:?}"); ApiError::InternalError(Some( "Failed to revoke account sessions".into(), )) })?; state .repos .oauth .delete_sessions_by_did(&did) .await .map_err(|e| { error!("Failed to revoke OAuth sessions for taken-down account: {e:?}"); ApiError::InternalError(Some("Failed to revoke OAuth sessions".into())) })?; } tranquil_pds::auth::invalidate_auth_cache(state.cache.as_ref(), &did).await; if let Ok(Some(handle)) = state.repos.user.get_handle_by_did(&did).await { let _ = state .cache .delete(&tranquil_pds::cache_keys::handle_key(&handle)) .await; } if takedown_update.is_some() || deactivated_update.is_some() { let status = tranquil_pds::sync::get_account_with_status( state.repos.repo.as_ref(), &did, ) .await .map_err(|e| { error!("Failed to read updated account status: {e:?}"); ApiError::InternalError(Some("Failed to read account status".into())) })? .ok_or(ApiError::SubjectNotFound)? .status; if let Err(e) = tranquil_pds::repo_ops::sequence_account_event(&state, &did, status).await { warn!("Failed to sequence account status event: {}", e); } } return Ok(Json(json!({ "subject": input.subject, "takedown": input.takedown.as_ref().map(|t| json!({ "applied": t.applied, "ref": t.r#ref })), "deactivated": input.deactivated.as_ref().map(|d| json!({ "applied": d.applied })) }))); } } Some("com.atproto.repo.strongRef") => { let uri_str = input.subject.get("uri").and_then(Value::as_str); if let Some(uri_str) = uri_str { let cid_str = input .subject .get("cid") .and_then(Value::as_str) .ok_or_else(|| { ApiError::InvalidRequest("Record subject must include a CID".into()) })?; let cid: CidLink = cid_str .parse() .map_err(|_| ApiError::InvalidRequest("Invalid CID format".into()))?; let mut applied_ref = None; if let Some(takedown) = &input.takedown { applied_ref = takedown.applied.then(|| { takedown .r#ref .clone() .unwrap_or_else(|| Utc::now().to_rfc3339()) }); state .repos .repo .set_record_takedown(&cid, applied_ref.as_deref()) .await .map_err(|e| { error!( "Failed to update record takedown status for {}: {:?}", uri_str, e ); ApiError::InternalError(Some("Failed to update takedown status".into())) })?; } return Ok(Json(json!({ "subject": input.subject, "takedown": input.takedown.as_ref().map(|t| json!({ "applied": t.applied, "ref": applied_ref })) }))); } } Some("com.atproto.admin.defs#repoBlobRef") => { let cid_str = input.subject.get("cid").and_then(Value::as_str); if let Some(cid_str) = cid_str { let cid: CidLink = cid_str .parse() .map_err(|_| ApiError::InvalidRequest("Invalid CID format".into()))?; let mut applied_ref = None; if let Some(takedown) = &input.takedown { applied_ref = takedown.applied.then(|| { takedown .r#ref .clone() .unwrap_or_else(|| Utc::now().to_rfc3339()) }); state .repos .blob .update_blob_takedown(&cid, applied_ref.as_deref()) .await .map_err(|e| { error!( "Failed to update blob takedown status for {}: {:?}", cid_str, e ); ApiError::InternalError(Some("Failed to update takedown status".into())) })?; } return Ok(Json(json!({ "subject": input.subject, "takedown": input.takedown.as_ref().map(|t| json!({ "applied": t.applied, "ref": applied_ref })) }))); } } _ => {} } Err(ApiError::InvalidRequest("Invalid subject type".into()))}