Something went wrong. Try again.
A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602use axum::extract::{Query, State};use axum::http::{HeaderMap, StatusCode};use axum::response::{IntoResponse, Response};use axum::routing::{get, post};use axum::{Json, Router};use base64::Engine as _;use k256;use serde::Deserialize;use sha2::{Digest, Sha256};use uuid::Uuid;
use crate::AppState;use crate::auth::XrpcClaims;use crate::db::{adapt_sql, now_rfc3339};use crate::error::AppError;use crate::lua::tid::generate_tid;use crate::spaces::scope::{SpaceReadAccess, check_delegation_token_access, check_read_access};use crate::spaces::types::*;use crate::spaces::{SpaceUri, db, members, notifications, oplog};
// ---------------------------------------------------------------------------// Request / response types// ---------------------------------------------------------------------------
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]struct RepoStateQuery { space: String, did: String,}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]struct ListRepoOpsQuery { space: String, did: String, limit: Option<i64>, cursor: Option<String>,}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]struct RegisterNotifyInput { space: String, service_did: String, endpoint: String,}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]struct NotifyWriteInput { space: String, did: String, collection: String, rkey: String, cid: Option<String>,}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]struct NotifySpaceDeletedInput { space: String,}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]struct GetDelegationTokenQuery { space: String,}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]struct SpaceUriQuery { space: String,}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]struct ListSpacesQuery { did: Option<String>, limit: Option<i64>, cursor: Option<String>,}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]struct PutRecordInput { space: String, collection: String, rkey: String, record: serde_json::Value, swap_record: Option<String>,}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]struct DeleteRecordInput { space: String, collection: String, rkey: String, swap_record: Option<String>,}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]struct GetRecordQuery { space: String, collection: String, rkey: String,}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]struct ListRecordsQuery { space: String, repo: Option<String>, collection: Option<String>, limit: Option<i64>, cursor: Option<String>, reverse: Option<bool>,}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]struct CreateInviteInput { space: String, access: Option<SpaceAccess>, max_uses: Option<i64>, expires_at: Option<String>,}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]struct RedeemInviteInput { token: String,}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]struct RevokeInviteInput { space: String, invite_id: String,}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]struct GetSpaceCredentialInput { grant: String,}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]struct CreateRecordInput { space: String, collection: String, record: serde_json::Value,}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]struct GetSpaceBlobQuery { space: String, cid: String,}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]struct ApplyWritesInput { space: String, swap_commit: Option<String>, writes: Vec<WriteOp>,}
#[derive(Deserialize)]#[serde(tag = "action", rename_all = "camelCase")]enum WriteOp { Create { collection: String, rkey: Option<String>, value: serde_json::Value, }, Update { collection: String, rkey: String, value: serde_json::Value, #[serde(rename = "swapRecord")] swap_record: Option<String>, }, Delete { collection: String, rkey: String, #[serde(rename = "swapRecord")] swap_record: Option<String>, },}
// ---------------------------------------------------------------------------// Route registration// ---------------------------------------------------------------------------
const PROTO_NS: &str = "com.atproto";const LEGACY_NS: &str = "dev.happyview";
pub fn space_routes() -> Router<AppState> { Router::new() // Protocol-level routes (com.atproto.space.*) .route(&format!("/xrpc/{PROTO_NS}.space.getSpace"), get(get_space)) .route( &format!("/xrpc/{PROTO_NS}.space.listSpaces"), get(list_spaces), ) .route( &format!("/xrpc/{PROTO_NS}.space.getRecord"), get(get_record), ) .route( &format!("/xrpc/{PROTO_NS}.space.listRecords"), get(list_records), ) .route( &format!("/xrpc/{PROTO_NS}.space.getRepoState"), get(get_repo_state), ) .route( &format!("/xrpc/{PROTO_NS}.space.listRepoOps"), get(list_repo_ops), ) .route( &format!("/xrpc/{PROTO_NS}.space.listRepos"), get(list_repos), ) .route( &format!("/xrpc/{PROTO_NS}.space.getDelegationToken"), get(get_delegation_token), ) .route( &format!("/xrpc/{PROTO_NS}.space.getSpaceCredential"), post(get_space_credential), ) .route( &format!("/xrpc/{PROTO_NS}.space.createRecord"), post(create_record), ) .route( &format!("/xrpc/{PROTO_NS}.space.putRecord"), post(put_record), ) .route( &format!("/xrpc/{PROTO_NS}.space.deleteRecord"), post(delete_record), ) .route( &format!("/xrpc/{PROTO_NS}.space.applyWrites"), post(apply_writes), ) .route( &format!("/xrpc/{PROTO_NS}.space.registerNotify"), post(register_notify), ) .route( &format!("/xrpc/{PROTO_NS}.space.notifyWrite"), post(notify_write), ) .route( &format!("/xrpc/{PROTO_NS}.space.notifySpaceDeleted"), post(notify_space_deleted), ) .route( &format!("/xrpc/{PROTO_NS}.space.getBlob"), get(get_space_blob), ) // Invites (HappyView extension, no com.atproto equivalent) .route( &format!("/xrpc/{LEGACY_NS}.space.createInvite"), post(create_invite), ) .route( &format!("/xrpc/{LEGACY_NS}.space.acceptInvite"), post(accept_invite), ) .route( &format!("/xrpc/{LEGACY_NS}.space.revokeInvite"), post(revoke_invite), ) .route( &format!("/xrpc/{LEGACY_NS}.space.listInvites"), get(list_invites), ) // Backward-compatible aliases (dev.happyview.space.*) — kept until v3 .route(&format!("/xrpc/{LEGACY_NS}.space.getSpace"), get(get_space)) .route( &format!("/xrpc/{LEGACY_NS}.space.listSpaces"), get(list_spaces), ) .route( &format!("/xrpc/{LEGACY_NS}.space.getRecord"), get(get_record), ) .route( &format!("/xrpc/{LEGACY_NS}.space.listRecords"), get(list_records), ) .route( &format!("/xrpc/{LEGACY_NS}.space.getMemberGrant"), get(get_delegation_token), ) .route( &format!("/xrpc/{LEGACY_NS}.space.getSpaceCredential"), post(get_space_credential), ) .route( &format!("/xrpc/{LEGACY_NS}.space.createRecord"), post(create_record), ) .route( &format!("/xrpc/{LEGACY_NS}.space.putRecord"), post(put_record), ) .route( &format!("/xrpc/{LEGACY_NS}.space.deleteRecord"), post(delete_record), ) .route( &format!("/xrpc/{LEGACY_NS}.space.applyWrites"), post(apply_writes), ) .route( &format!("/xrpc/{LEGACY_NS}.space.getBlob"), get(get_space_blob), )}
// ---------------------------------------------------------------------------// Helpers// ---------------------------------------------------------------------------
fn require_auth(claims: &XrpcClaims) -> Result<&crate::auth::Claims, AppError> { claims .identity .as_ref() .ok_or_else(|| AppError::Auth("This endpoint requires authentication".into()))}
/// Like `require_auth`, but also accepts a verified space credential as an/// identity source. Use this in space endpoints that support `Bearer/// <space_credential>` in addition to DPoP auth.async fn require_auth_or_credential( state: &AppState, claims: &XrpcClaims,) -> Result<String, AppError> { if let Some(identity) = &claims.identity { return Ok(identity.did().to_string()); }
if let Some(token) = &claims.space_credential { let verified = crate::spaces::credential::verify_external_credential( token, &state.http, &state.config.plc_url, ) .await?; return Ok(verified.sub); }
Err(AppError::Auth( "This endpoint requires authentication".into(), ))}
async fn resolve_space(state: &AppState, space_uri: &str) -> Result<Space, AppError> { let uri = SpaceUri::parse(space_uri)?; db::get_space_by_address( &state.db, state.db_backend, &uri.did, &uri.type_nsid, &uri.skey, ) .await? .ok_or_else(|| AppError::NotFound("Space not found".into()))}
async fn require_space_admin(state: &AppState, space: &Space, did: &str) -> Result<(), AppError> { if space.authority_did == did { return Ok(()); } let sql = adapt_sql( "SELECT is_super FROM happyview_users WHERE did = ?", state.db_backend, ); let row: Option<(i32,)> = sqlx::query_as(&sql) .bind(did) .fetch_optional(&state.db) .await .map_err(|e| AppError::Internal(format!("failed to check admin status: {e}")))?; if row.is_some_and(|(is_super,)| is_super != 0) { return Ok(()); } Err(AppError::Forbidden( "Only the space authority can perform this action".into(), ))}
async fn require_membership( state: &AppState, space: &Space, did: &str, require_write: bool, space_credential: Option<&str>,) -> Result<SpaceAccess, AppError> { if let Some(token) = space_credential { let space_uri = format!("ats://{}/{}/{}", space.did, space.type_nsid, space.skey); match crate::spaces::credential::verify_external_credential( token, &state.http, &state.config.plc_url, ) .await { Ok(claims) if claims.sub == space_uri => { // External credential grants read access; write is not supported via space credential if require_write { return Err(AppError::Forbidden( "Write access is required for this action".into(), )); } return Ok(SpaceAccess::Read); } Ok(_) => { // Credential is valid but for a different space — fall through } Err(_) => { // External verification failed — fall through to local check } } }
let access = members::is_member(&state.db, state.db_backend, &space.id, did) .await? .ok_or_else(|| AppError::Forbidden("You are not a member of this space".into()))?; if require_write && !access.can_write() { return Err(AppError::Forbidden( "Write access is required for this action".into(), )); } Ok(access)}
fn content_cid(record: &serde_json::Value) -> String { let bytes = serde_json::to_vec(record).unwrap_or_default(); let hash = Sha256::digest(&bytes); format!("bafyrei{}", hex::encode(&hash[..20]))}
async fn resolve_client_id_url( state: &AppState, client_key: &str,) -> Result<Option<String>, AppError> { let sql = adapt_sql( "SELECT client_id_url FROM happyview_api_clients WHERE client_key = ?", state.db_backend, ); let row: Option<(String,)> = sqlx::query_as(&sql) .bind(client_key) .fetch_optional(&state.db) .await .map_err(|e| AppError::Internal(format!("failed to look up API client: {e}")))?; Ok(row.map(|(url,)| url))}
// ---------------------------------------------------------------------------// Space read handlers// ---------------------------------------------------------------------------
async fn get_space( State(state): State<AppState>, xrpc_claims: XrpcClaims, Query(query): Query<SpaceUriQuery>,) -> Result<Json<serde_json::Value>, AppError> { let space = resolve_space(&state, &query.space).await?;
// If the space's membership is not public, require auth + membership if !space.config.membership_public { let claims = require_auth(&xrpc_claims)?; let did = claims.did(); if space.authority_did != did { members::is_member(&state.db, state.db_backend, &space.id, did) .await? .ok_or_else(|| AppError::NotFound("Space not found".into()))?; } }
let space_uri = format!("ats://{}/{}/{}", space.did, space.type_nsid, space.skey); let simplespace_config = serde_json::json!({ "$type": "com.atproto.simplespace.defs#spaceConfig", "mintPolicy": space.mint_policy, "appAccess": space.app_access, "managingApp": space.managing_app_did, }); Ok(Json(serde_json::json!({ "uri": space_uri, "space": space, "config": simplespace_config, })))}
async fn list_spaces( State(state): State<AppState>, xrpc_claims: XrpcClaims, Query(query): Query<ListSpacesQuery>,) -> Result<Json<serde_json::Value>, AppError> { let claims = require_auth(&xrpc_claims)?; let did = query.did.unwrap_or_else(|| claims.did().to_string()); let limit = query.limit.unwrap_or(50).min(100);
let (views, cursor) = db::list_spaces_for_user( &state.db, state.db_backend, &did, limit, query.cursor.as_deref(), ) .await?;
let spaces_json: Vec<serde_json::Value> = views .into_iter() .map(|v| { serde_json::json!({ "uri": v.uri, "isOwner": v.is_owner, }) }) .collect();
Ok(Json(serde_json::json!({ "spaces": spaces_json, "cursor": cursor, })))}
// ---------------------------------------------------------------------------// Record handlers// ---------------------------------------------------------------------------
async fn create_record( State(state): State<AppState>, xrpc_claims: XrpcClaims, Json(input): Json<CreateRecordInput>,) -> Result<Response, AppError> { let did = require_auth_or_credential(&state, &xrpc_claims).await?; let space = resolve_space(&state, &input.space).await?; require_membership( &state, &space, &did, true, xrpc_claims.space_credential.as_deref(), ) .await?;
let rkey = generate_tid(); let cid = content_cid(&input.record); let record_uri = format!( "ats://{}/{}/{}/{}/{}/{}", space.did, space.type_nsid, space.skey, did, input.collection, rkey );
let record = SpaceRecord { uri: record_uri.clone(), space_id: space.id.clone(), author_did: did, collection: input.collection, rkey, record: input.record, cid: cid.clone(), indexed_at: now_rfc3339(), };
db::insert_space_record(&state.db, state.db_backend, &record).await?;
let rev = generate_tid(); db::update_space_revision(&state.db, state.db_backend, &space.id, &rev).await?;
let body = serde_json::json!({ "uri": record_uri, "cid": cid, });
let mut response = Json(body).into_response(); *response.status_mut() = StatusCode::CREATED; Ok(response)}
async fn put_record( State(state): State<AppState>, xrpc_claims: XrpcClaims, Json(input): Json<PutRecordInput>,) -> Result<Response, AppError> { let did = require_auth_or_credential(&state, &xrpc_claims).await?; let space = resolve_space(&state, &input.space).await?; require_membership( &state, &space, &did, true, xrpc_claims.space_credential.as_deref(), ) .await?;
let cid = content_cid(&input.record); let record_uri = format!( "ats://{}/{}/{}/{}/{}/{}", space.did, space.type_nsid, space.skey, did, input.collection, input.rkey );
let record = SpaceRecord { uri: record_uri.clone(), space_id: space.id.clone(), author_did: did, collection: input.collection, rkey: input.rkey, record: input.record, cid: cid.clone(), indexed_at: now_rfc3339(), };
if let Some(swap_cid) = input.swap_record { db::upsert_space_record_with_swap(&state.db, state.db_backend, &record, &swap_cid).await?; } else { db::upsert_space_record(&state.db, state.db_backend, &record).await?; }
let rev = generate_tid(); db::update_space_revision(&state.db, state.db_backend, &space.id, &rev).await?;
let body = serde_json::json!({ "uri": record_uri, "cid": cid, });
let mut response = Json(body).into_response(); *response.status_mut() = StatusCode::CREATED; Ok(response)}
async fn delete_record( State(state): State<AppState>, xrpc_claims: XrpcClaims, Json(input): Json<DeleteRecordInput>,) -> Result<Json<serde_json::Value>, AppError> { let claims = require_auth(&xrpc_claims)?; let did = claims.did().to_string(); let space = resolve_space(&state, &input.space).await?;
let record_uri = format!( "ats://{}/{}/{}/{}/{}/{}", space.did, space.type_nsid, space.skey, did, input.collection, input.rkey );
if let Some(swap_cid) = input.swap_record { db::delete_space_record_with_swap(&state.db, state.db_backend, &record_uri, &swap_cid) .await?; } else { let record = db::get_space_record(&state.db, state.db_backend, &record_uri).await?; match record { Some(r) if r.author_did != did => { return Err(AppError::Forbidden( "You can only delete your own records".into(), )); } None => { return Err(AppError::NotFound("Record not found".into())); } _ => {} } db::delete_space_record(&state.db, state.db_backend, &record_uri).await?; }
let rev = generate_tid(); db::update_space_revision(&state.db, state.db_backend, &space.id, &rev).await?;
Ok(Json(serde_json::json!({ "success": true })))}
async fn apply_writes( State(state): State<AppState>, xrpc_claims: XrpcClaims, Json(input): Json<ApplyWritesInput>,) -> Result<Json<serde_json::Value>, AppError> { let did = require_auth_or_credential(&state, &xrpc_claims).await?; let space = resolve_space(&state, &input.space).await?; require_membership( &state, &space, &did, true, xrpc_claims.space_credential.as_deref(), ) .await?;
if let Some(ref expected_rev) = input.swap_commit { match &space.revision { Some(current_rev) if current_rev != expected_rev => { return Err(AppError::Conflict("swapCommit mismatch".into())); } None if !expected_rev.is_empty() => { return Err(AppError::Conflict("swapCommit mismatch".into())); } _ => {} } }
let mut results = Vec::with_capacity(input.writes.len());
for op in input.writes { match op { WriteOp::Create { collection, rkey, value, } => { let rkey = rkey.unwrap_or_else(generate_tid); let cid = content_cid(&value); let record_uri = format!( "ats://{}/{}/{}/{}/{}/{}", space.did, space.type_nsid, space.skey, did, collection, rkey ); let record = SpaceRecord { uri: record_uri.clone(), space_id: space.id.clone(), author_did: did.clone(), collection, rkey, record: value, cid: cid.clone(), indexed_at: now_rfc3339(), }; db::insert_space_record(&state.db, state.db_backend, &record).await?; results.push(serde_json::json!({ "uri": record_uri, "cid": cid, })); } WriteOp::Update { collection, rkey, value, swap_record, } => { let cid = content_cid(&value); let record_uri = format!( "ats://{}/{}/{}/{}/{}/{}", space.did, space.type_nsid, space.skey, did, collection, rkey ); let record = SpaceRecord { uri: record_uri.clone(), space_id: space.id.clone(), author_did: did.clone(), collection, rkey, record: value, cid: cid.clone(), indexed_at: now_rfc3339(), }; if let Some(swap_cid) = swap_record { db::upsert_space_record_with_swap( &state.db, state.db_backend, &record, &swap_cid, ) .await?; } else { db::upsert_space_record(&state.db, state.db_backend, &record).await?; } results.push(serde_json::json!({ "uri": record_uri, "cid": cid, })); } WriteOp::Delete { collection, rkey, swap_record, } => { let record_uri = format!( "ats://{}/{}/{}/{}/{}/{}", space.did, space.type_nsid, space.skey, did, collection, rkey ); if let Some(swap_cid) = swap_record { db::delete_space_record_with_swap( &state.db, state.db_backend, &record_uri, &swap_cid, ) .await?; } else { db::delete_space_record(&state.db, state.db_backend, &record_uri).await?; } results.push(serde_json::json!({})); } } }
let rev = generate_tid(); db::update_space_revision(&state.db, state.db_backend, &space.id, &rev).await?;
Ok(Json(serde_json::json!({ "results": results, })))}
async fn get_record( State(state): State<AppState>, xrpc_claims: XrpcClaims, Query(query): Query<GetRecordQuery>,) -> Result<Json<serde_json::Value>, AppError> { let did = require_auth_or_credential(&state, &xrpc_claims).await?; let space = resolve_space(&state, &query.space).await?; let has_credential = xrpc_claims.space_credential.is_some(); let membership = require_membership( &state, &space, &did, false, xrpc_claims.space_credential.as_deref(), ) .await?;
let record = db::get_space_record_by_parts( &state.db, state.db_backend, &space.id, &query.collection, &query.rkey, ) .await? .ok_or_else(|| AppError::NotFound("Record not found".into()))?;
let read_access = SpaceReadAccess::from_space_access(membership); check_read_access(&did, &record.author_did, read_access, has_credential)?;
Ok(Json(serde_json::json!({ "uri": record.uri, "cid": record.cid, "value": record.record, })))}
async fn list_records( State(state): State<AppState>, xrpc_claims: XrpcClaims, Query(query): Query<ListRecordsQuery>,) -> Result<Json<serde_json::Value>, AppError> { let did = require_auth_or_credential(&state, &xrpc_claims).await?; let space = resolve_space(&state, &query.space).await?; let has_credential = xrpc_claims.space_credential.is_some(); let membership = require_membership( &state, &space, &did, false, xrpc_claims.space_credential.as_deref(), ) .await?;
let read_access = SpaceReadAccess::from_space_access(membership);
// read_self members may only list their own records regardless of what the caller requests let repo = if !has_credential && read_access == SpaceReadAccess::ReadSelf { Some(did.as_str()) } else { query.repo.as_deref().or(if has_credential { None } else { Some(did.as_str()) }) };
let limit = query.limit.unwrap_or(50).min(100); let reverse = query.reverse.unwrap_or(false); let (records, cursor) = db::list_space_records( &state.db, state.db_backend, &space.id, repo, query.collection.as_deref(), limit, query.cursor.as_deref(), reverse, ) .await?;
let records_json: Vec<serde_json::Value> = records .into_iter() .map(|r| { serde_json::json!({ "collection": r.collection, "rkey": r.rkey, "cid": r.cid, }) }) .collect();
Ok(Json(serde_json::json!({ "records": records_json, "cursor": cursor, })))}
// ---------------------------------------------------------------------------// Invite handlers// ---------------------------------------------------------------------------
async fn create_invite( State(state): State<AppState>, xrpc_claims: XrpcClaims, Json(input): Json<CreateInviteInput>,) -> Result<Response, AppError> { let claims = require_auth(&xrpc_claims)?; let space = resolve_space(&state, &input.space).await?; require_space_admin(&state, &space, claims.did()).await?;
let mut token_bytes = [0u8; 24]; rand::Fill::fill(&mut token_bytes, &mut rand::rng()); let token = hex::encode(token_bytes); let token_hash = hex::encode(Sha256::digest(token.as_bytes()));
let invite = SpaceInvite { id: Uuid::new_v4().to_string(), space_id: space.id, token_hash, created_by: claims.did().to_string(), access: input.access.unwrap_or(SpaceAccess::Read), max_uses: input.max_uses, uses: 0, expires_at: input.expires_at, revoked: false, created_at: now_rfc3339(), };
db::create_invite(&state.db, state.db_backend, &invite).await?;
let mut response = Json(serde_json::json!({ "inviteId": invite.id, "token": token, "access": invite.access, "maxUses": invite.max_uses, "expiresAt": invite.expires_at, })) .into_response(); *response.status_mut() = StatusCode::CREATED; Ok(response)}
async fn accept_invite( State(state): State<AppState>, xrpc_claims: XrpcClaims, Json(input): Json<RedeemInviteInput>,) -> Result<Response, AppError> { let claims = require_auth(&xrpc_claims)?; let did = claims.did().to_string();
let token_hash = hex::encode(Sha256::digest(input.token.as_bytes())); let invite = db::get_invite_by_token_hash(&state.db, state.db_backend, &token_hash) .await? .ok_or_else(|| AppError::NotFound("Invalid invite token".into()))?;
if invite.revoked { return Err(AppError::BadRequest("This invite has been revoked".into())); }
if let Some(max) = invite.max_uses && invite.uses >= max { return Err(AppError::BadRequest( "This invite has reached its maximum uses".into(), )); }
if let Some(ref expires) = invite.expires_at { let now = now_rfc3339(); if now > *expires { return Err(AppError::BadRequest("This invite has expired".into())); } }
let existing = db::get_member(&state.db, state.db_backend, &invite.space_id, &did).await?; if existing.is_some() { return Err(AppError::Conflict( "You are already a member of this space".into(), )); }
let member = SpaceMember { id: Uuid::new_v4().to_string(), space_id: invite.space_id.clone(), did, access: invite.access, is_delegation: false, granted_by: Some(invite.created_by.clone()), created_at: now_rfc3339(), };
db::add_member(&state.db, state.db_backend, &member).await?; db::increment_invite_uses(&state.db, state.db_backend, &invite.id).await?;
let space = db::get_space(&state.db, state.db_backend, &invite.space_id).await?; let space_uri = space.map(|s| format!("ats://{}/{}/{}", s.did, s.type_nsid, s.skey));
let mut response = Json(serde_json::json!({ "uri": space_uri, "access": member.access, })) .into_response(); *response.status_mut() = StatusCode::CREATED; Ok(response)}
async fn revoke_invite( State(state): State<AppState>, xrpc_claims: XrpcClaims, Json(input): Json<RevokeInviteInput>,) -> Result<Json<serde_json::Value>, AppError> { let claims = require_auth(&xrpc_claims)?; let space = resolve_space(&state, &input.space).await?; require_space_admin(&state, &space, claims.did()).await?;
let revoked = db::revoke_invite(&state.db, state.db_backend, &input.invite_id).await?; if !revoked { return Err(AppError::NotFound("Invite not found".into())); }
Ok(Json(serde_json::json!({ "success": true })))}
async fn list_invites( State(state): State<AppState>, xrpc_claims: XrpcClaims, Query(query): Query<SpaceUriQuery>,) -> Result<Json<serde_json::Value>, AppError> { let claims = require_auth(&xrpc_claims)?; let space = resolve_space(&state, &query.space).await?; require_space_admin(&state, &space, claims.did()).await?;
let invites = db::list_invites(&state.db, state.db_backend, &space.id).await?;
let invites_json: Vec<serde_json::Value> = invites .into_iter() .map(|i| { serde_json::json!({ "id": i.id, "access": i.access, "maxUses": i.max_uses, "uses": i.uses, "expiresAt": i.expires_at, "revoked": i.revoked, "createdBy": i.created_by, "createdAt": i.created_at, }) }) .collect();
Ok(Json(serde_json::json!({ "invites": invites_json })))}
// ---------------------------------------------------------------------------// Credential handlers// ---------------------------------------------------------------------------
async fn get_delegation_token( State(state): State<AppState>, xrpc_claims: XrpcClaims, Query(params): Query<GetDelegationTokenQuery>,) -> Result<Json<serde_json::Value>, AppError> { let claims = require_auth(&xrpc_claims)?; let did = claims.did().to_string(); let space = resolve_space(&state, ¶ms.space).await?;
let membership = require_membership(&state, &space, &did, false, None).await?; let read_access = SpaceReadAccess::from_space_access(membership); check_delegation_token_access(read_access, false)?;
let encryption_key = state.config.token_encryption_key.as_ref().ok_or_else(|| { AppError::Internal("TOKEN_ENCRYPTION_KEY is required for space credentials".into()) })?;
let signing_key = k256::ecdsa::SigningKey::from_bytes(encryption_key.into()) .map_err(|e| AppError::Internal(format!("failed to derive delegation signing key: {e}")))?;
let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap() .as_secs(); let exp = now + crate::spaces::credential::DELEGATION_TOKEN_TTL_SECS;
let space_uri = format!("ats://{}/{}/{}", space.did, space.type_nsid, space.skey); let space_host = format!("{}#atproto_space_host", space.did); let delegation_claims = crate::spaces::credential::DelegationTokenClaims { iss: did, sub: space_uri, aud: space_host, iat: now, exp, jti: crate::spaces::credential::make_jti(), };
let grant = crate::spaces::credential::sign_delegation_token(&delegation_claims, &signing_key)?;
let expires_at = chrono::DateTime::from_timestamp(exp as i64, 0) .map(|dt| dt.to_rfc3339()) .unwrap_or_default();
Ok(Json(serde_json::json!({ "delegationToken": grant, "expiresAt": expires_at, })))}
// ---------------------------------------------------------------------------// Protocol endpoint implementations// ---------------------------------------------------------------------------
async fn get_repo_state( State(state): State<AppState>, claims: XrpcClaims, Query(params): Query<RepoStateQuery>,) -> Result<impl IntoResponse, AppError> { let did = require_auth_or_credential(&state, &claims).await?; let space = resolve_space(&state, ¶ms.space).await?; let has_credential = claims.space_credential.is_some(); let membership = require_membership( &state, &space, &did, false, claims.space_credential.as_deref(), ) .await?;
let read_access = SpaceReadAccess::from_space_access(membership); check_read_access(&did, ¶ms.did, read_access, has_credential)?;
let repo_state = db::get_or_create_repo_state(&state.db, state.db_backend, &space.id, ¶ms.did).await?;
Ok(Json(serde_json::json!({ "rev": repo_state.rev, "commit": repo_state.hash.as_ref().map(|h| { serde_json::json!({ "hash": base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(h), "ikm": base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(repo_state.ikm.as_deref().unwrap_or_default()), "sig": base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(repo_state.sig.as_deref().unwrap_or_default()), "mac": base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(repo_state.mac.as_deref().unwrap_or_default()), "rev": repo_state.rev, }) }), })))}
async fn list_repo_ops( State(state): State<AppState>, claims: XrpcClaims, Query(params): Query<ListRepoOpsQuery>,) -> Result<impl IntoResponse, AppError> { let did = require_auth_or_credential(&state, &claims).await?; let space = resolve_space(&state, ¶ms.space).await?; let has_credential = claims.space_credential.is_some(); let membership = require_membership( &state, &space, &did, false, claims.space_credential.as_deref(), ) .await?;
let read_access = SpaceReadAccess::from_space_access(membership); check_read_access(&did, ¶ms.did, read_access, has_credential)?;
let limit = params.limit.unwrap_or(100).min(1000); let ops = oplog::list_ops( &state.db, state.db_backend, &space.id, ¶ms.did, params.cursor.as_deref(), limit, ) .await?;
Ok(Json(serde_json::json!({ "ops": ops })))}
async fn list_repos( State(state): State<AppState>, claims: XrpcClaims, Query(params): Query<SpaceUriQuery>,) -> Result<impl IntoResponse, AppError> { let _did = require_auth_or_credential(&state, &claims).await?; let space = resolve_space(&state, ¶ms.space).await?;
let repos = db::list_space_repos(&state.db, state.db_backend, &space.id).await?; Ok(Json(serde_json::json!({ "repos": repos })))}
async fn get_space_blob( State(state): State<AppState>, claims: XrpcClaims, Query(params): Query<GetSpaceBlobQuery>,) -> Result<impl IntoResponse, AppError> { let did = require_auth_or_credential(&state, &claims).await?; let space = resolve_space(&state, ¶ms.space).await?; let has_credential = claims.space_credential.is_some(); let membership = require_membership( &state, &space, &did, false, claims.space_credential.as_deref(), ) .await?;
let author_did = db::find_blob_author_did(&state.db, state.db_backend, &space.id, ¶ms.cid) .await? .ok_or_else(|| AppError::NotFound("Blob not found in this space".into()))?;
let read_access = SpaceReadAccess::from_space_access(membership); check_read_access(&did, &author_did, read_access, has_credential)?;
let pds_endpoint = crate::profile::resolve_pds_endpoint(&state.http, &state.config.plc_url, &author_did) .await?;
let url = format!( "{}/xrpc/com.atproto.sync.getBlob?did={}&cid={}", pds_endpoint, urlencoding::encode(&author_did), urlencoding::encode(¶ms.cid), );
let resp = state .http .get(&url) .send() .await .map_err(|e| AppError::BadGateway(format!("blob fetch failed: {e}")))?;
let status = resp.status(); if !status.is_success() { return Err(AppError::BadGateway(format!( "PDS returned {status} for blob cid={}", params.cid ))); }
let content_type = resp .headers() .get("content-type") .and_then(|v| v.to_str().ok()) .unwrap_or("application/octet-stream") .to_string();
let bytes = resp .bytes() .await .map_err(|e| AppError::BadGateway(format!("failed to read blob body: {e}")))?;
let mut headers = HeaderMap::new(); headers.insert( axum::http::header::CONTENT_TYPE, content_type .parse() .unwrap_or_else(|_| "application/octet-stream".parse().unwrap()), );
Ok((status, headers, bytes))}
async fn register_notify( State(state): State<AppState>, claims: XrpcClaims, Json(input): Json<RegisterNotifyInput>,) -> Result<impl IntoResponse, AppError> { let did = require_auth_or_credential(&state, &claims).await?; let space = resolve_space(&state, &input.space).await?;
let id = notifications::register( &state.db, state.db_backend, &space.id, &input.service_did, &input.endpoint, &did, ) .await?;
Ok(Json(serde_json::json!({ "id": id })))}
async fn notify_write( State(state): State<AppState>, _claims: XrpcClaims, Json(input): Json<NotifyWriteInput>,) -> Result<impl IntoResponse, AppError> { let space = resolve_space(&state, &input.space).await?;
notifications::dispatch_write_notification( &state.db, state.db_backend, &state.http, &space.id, &input.did, &input.collection, &input.rkey, input.cid.as_deref(), ) .await?;
Ok(Json(serde_json::json!({ "success": true })))}
async fn notify_space_deleted( State(state): State<AppState>, _claims: XrpcClaims, Json(input): Json<NotifySpaceDeletedInput>,) -> Result<impl IntoResponse, AppError> { let space = resolve_space(&state, &input.space).await?;
notifications::dispatch_space_deleted(&state.db, state.db_backend, &state.http, &space.id) .await?;
Ok(Json(serde_json::json!({ "success": true })))}
async fn get_space_credential( State(state): State<AppState>, xrpc_claims: XrpcClaims, Json(input): Json<GetSpaceCredentialInput>,) -> Result<Json<serde_json::Value>, AppError> { let claims = require_auth(&xrpc_claims)?;
let encryption_key = state.config.token_encryption_key.as_ref().ok_or_else(|| { AppError::Internal("TOKEN_ENCRYPTION_KEY is required for space credentials".into()) })?;
let verifying_key = { let signing_key = k256::ecdsa::SigningKey::from_bytes(encryption_key.into()).map_err(|e| { AppError::Internal(format!("failed to derive delegation signing key: {e}")) })?; k256::ecdsa::VerifyingKey::from(&signing_key) };
let delegation_claims = { let unverified_sub = crate::spaces::credential::peek_delegation_sub(&input.grant) .ok_or_else(|| AppError::Auth("invalid delegation token".into()))?; let space_did = crate::spaces::SpaceUri::parse(&unverified_sub) .map(|u| u.did.clone()) .unwrap_or_default(); let expected_aud = format!("{space_did}#atproto_space_host"); crate::spaces::credential::verify_delegation_token( &input.grant, &verifying_key, &expected_aud, )? };
let space = resolve_space(&state, &delegation_claims.sub).await?;
let client_id = if let Some(key) = claims.client_key() { resolve_client_id_url(&state, key).await? } else { None }; let issued = crate::spaces::auth::issue_credential( &state.db, state.db_backend, &state.http, encryption_key, &space, &delegation_claims.iss, client_id.as_deref(), &space.authority_did, ) .await?;
Ok(Json(serde_json::json!({ "credential": issued.token, "expiresAt": issued.expires_at, })))}
#[cfg(test)]mod tests { use super::*; use serde_json::json;
#[test] fn content_cid_deterministic() { let record = json!({"text": "hello"}); let cid1 = content_cid(&record); let cid2 = content_cid(&record); assert_eq!(cid1, cid2); assert!(cid1.starts_with("bafyrei")); }
#[test] fn content_cid_changes_for_different_records() { let a = content_cid(&json!({"text": "hello"})); let b = content_cid(&json!({"text": "world"})); assert_ne!(a, b); }
#[test] fn deserialize_create_record_input() { let input: CreateRecordInput = serde_json::from_value(json!({ "space": "ats://did:plc:abc/com.example.forum/main", "collection": "com.example.forum.post", "record": { "text": "hello" } })) .unwrap(); assert_eq!(input.space, "ats://did:plc:abc/com.example.forum/main"); assert_eq!(input.collection, "com.example.forum.post"); assert_eq!(input.record["text"], "hello"); }
#[test] fn deserialize_put_record_with_swap() { let input: PutRecordInput = serde_json::from_value(json!({ "space": "ats://did:plc:abc/com.example.forum/main", "collection": "com.example.forum.post", "rkey": "3k2abc", "record": { "text": "updated" }, "swapRecord": "bafyrei123" })) .unwrap(); assert_eq!(input.swap_record.as_deref(), Some("bafyrei123")); }
#[test] fn deserialize_put_record_without_swap() { let input: PutRecordInput = serde_json::from_value(json!({ "space": "ats://did:plc:abc/com.example.forum/main", "collection": "com.example.forum.post", "rkey": "3k2abc", "record": { "text": "hello" } })) .unwrap(); assert_eq!(input.swap_record, None); }
#[test] fn deserialize_delete_record_with_swap() { let input: DeleteRecordInput = serde_json::from_value(json!({ "space": "ats://did:plc:abc/com.example.forum/main", "collection": "com.example.forum.post", "rkey": "3k2abc", "swapRecord": "bafyrei456" })) .unwrap(); assert_eq!(input.swap_record.as_deref(), Some("bafyrei456")); }
#[test] fn deserialize_write_op_create() { let op: WriteOp = serde_json::from_value(json!({ "action": "create", "collection": "com.example.forum.post", "value": { "text": "new post" } })) .unwrap(); match op { WriteOp::Create { collection, rkey, value, } => { assert_eq!(collection, "com.example.forum.post"); assert_eq!(rkey, None); assert_eq!(value["text"], "new post"); } _ => panic!("expected Create"), } }
#[test] fn deserialize_write_op_create_with_rkey() { let op: WriteOp = serde_json::from_value(json!({ "action": "create", "collection": "com.example.forum.post", "rkey": "custom-key", "value": { "text": "new post" } })) .unwrap(); match op { WriteOp::Create { rkey, .. } => { assert_eq!(rkey.as_deref(), Some("custom-key")); } _ => panic!("expected Create"), } }
#[test] fn deserialize_write_op_update() { let op: WriteOp = serde_json::from_value(json!({ "action": "update", "collection": "com.example.forum.post", "rkey": "3k2abc", "value": { "text": "updated" }, "swapRecord": "bafyrei789" })) .unwrap(); match op { WriteOp::Update { collection, rkey, swap_record, .. } => { assert_eq!(collection, "com.example.forum.post"); assert_eq!(rkey, "3k2abc"); assert_eq!(swap_record.as_deref(), Some("bafyrei789")); } _ => panic!("expected Update"), } }
#[test] fn deserialize_write_op_delete() { let op: WriteOp = serde_json::from_value(json!({ "action": "delete", "collection": "com.example.forum.post", "rkey": "3k2abc" })) .unwrap(); match op { WriteOp::Delete { collection, rkey, swap_record, } => { assert_eq!(collection, "com.example.forum.post"); assert_eq!(rkey, "3k2abc"); assert_eq!(swap_record, None); } _ => panic!("expected Delete"), } }
#[test] fn deserialize_apply_writes_input() { let input: ApplyWritesInput = serde_json::from_value(json!({ "space": "ats://did:plc:abc/com.example.forum/main", "swapCommit": "tid123", "writes": [ { "action": "create", "collection": "com.example.forum.post", "value": { "text": "post 1" } }, { "action": "delete", "collection": "com.example.forum.post", "rkey": "old-key" } ] })) .unwrap(); assert_eq!(input.space, "ats://did:plc:abc/com.example.forum/main"); assert_eq!(input.swap_commit.as_deref(), Some("tid123")); assert_eq!(input.writes.len(), 2); }
#[test] fn deserialize_apply_writes_without_swap_commit() { let input: ApplyWritesInput = serde_json::from_value(json!({ "space": "ats://did:plc:abc/com.example.forum/main", "writes": [ { "action": "create", "collection": "com.example.forum.post", "value": { "text": "post" } } ] })) .unwrap(); assert_eq!(input.swap_commit, None); }
#[test] fn deserialize_write_op_rejects_unknown_action() { let result = serde_json::from_value::<WriteOp>(json!({ "action": "unknown", "collection": "test", "rkey": "key" })); assert!(result.is_err()); }}