diff --git a/migrations/postgres/20260511000000_add_space_did.sql b/migrations/postgres/20260511000000_add_space_did.sql new file mode 100644 index 0000000..e33ff12 --- /dev/null +++ b/migrations/postgres/20260511000000_add_space_did.sql @@ -0,0 +1,4 @@ +ALTER TABLE spaces ADD COLUMN did TEXT; +UPDATE spaces SET did = owner_did; +ALTER TABLE spaces ALTER COLUMN did SET NOT NULL; +CREATE UNIQUE INDEX idx_spaces_did_type_skey ON spaces(did, type_nsid, skey); diff --git a/migrations/postgres/20260511100000_add_space_revision.sql b/migrations/postgres/20260511100000_add_space_revision.sql new file mode 100644 index 0000000..800df84 --- /dev/null +++ b/migrations/postgres/20260511100000_add_space_revision.sql @@ -0,0 +1 @@ +ALTER TABLE spaces ADD COLUMN revision TEXT; diff --git a/migrations/sqlite/20260511000000_add_space_did.sql b/migrations/sqlite/20260511000000_add_space_did.sql new file mode 100644 index 0000000..07747e4 --- /dev/null +++ b/migrations/sqlite/20260511000000_add_space_did.sql @@ -0,0 +1,3 @@ +ALTER TABLE spaces ADD COLUMN did TEXT; +UPDATE spaces SET did = owner_did; +CREATE UNIQUE INDEX idx_spaces_did_type_skey ON spaces(did, type_nsid, skey); diff --git a/migrations/sqlite/20260511100000_add_space_revision.sql b/migrations/sqlite/20260511100000_add_space_revision.sql new file mode 100644 index 0000000..800df84 --- /dev/null +++ b/migrations/sqlite/20260511100000_add_space_revision.sql @@ -0,0 +1 @@ +ALTER TABLE spaces ADD COLUMN revision TEXT; diff --git a/src/lua/atproto_api.rs b/src/lua/atproto_api.rs index bde9321..e1669a6 100644 --- a/src/lua/atproto_api.rs +++ b/src/lua/atproto_api.rs @@ -280,7 +280,7 @@ pub fn register_atproto_api( let space = crate::spaces::db::get_space_by_address( &state.db, state.db_backend, - &uri.owner_did, + &uri.did, &uri.type_nsid, &uri.skey, ) @@ -321,7 +321,7 @@ pub fn register_atproto_api( let space = crate::spaces::db::get_space_by_address( &state.db, state.db_backend, - &uri.owner_did, + &uri.did, &uri.type_nsid, &uri.skey, ) @@ -361,7 +361,7 @@ pub fn register_atproto_api( let space = crate::spaces::db::get_space_by_address( &state.db, state.db_backend, - &uri.owner_did, + &uri.did, &uri.type_nsid, &uri.skey, ) @@ -416,7 +416,7 @@ pub fn register_atproto_api( let space = crate::spaces::db::get_space_by_address( &state.db, state.db_backend, - &uri.owner_did, + &uri.did, &uri.type_nsid, &uri.skey, ) @@ -433,9 +433,11 @@ pub fn register_atproto_api( &state.db, state.db_backend, &space.id, + None, collection.as_deref(), limit.min(100), cursor.as_deref(), + false, ) .await .map_err(|e| mlua::Error::runtime(format!("record query failed: {e}")))?; diff --git a/src/lua/context.rs b/src/lua/context.rs index eab578e..66e25be 100644 --- a/src/lua/context.rs +++ b/src/lua/context.rs @@ -5,8 +5,9 @@ use std::collections::HashMap; /// Optional space context passed to Lua scripts when the request is space-scoped. #[derive(Debug, Clone)] pub struct SpaceContext { - pub space_uri: String, + pub space: String, pub space_id: String, + pub did: String, pub owner_did: String, pub type_nsid: String, pub skey: String, @@ -17,8 +18,9 @@ fn set_space_context(lua: &Lua, space: Option<&SpaceContext>) -> LuaResult<()> { match space { Some(ctx) => { let table = lua.create_table()?; - table.set("space_uri", ctx.space_uri.as_str())?; + table.set("space", ctx.space.as_str())?; table.set("space_id", ctx.space_id.as_str())?; + table.set("did", ctx.did.as_str())?; table.set("owner_did", ctx.owner_did.as_str())?; table.set("type_nsid", ctx.type_nsid.as_str())?; table.set("skey", ctx.skey.as_str())?; @@ -269,8 +271,9 @@ mod tests { let lua = create_sandbox().unwrap(); let params = HashMap::new(); let space = SpaceContext { - space_uri: "ats://did:plc:owner/com.example.forum/main".into(), + space: "ats://did:plc:owner/com.example.forum/main".into(), space_id: "space-123".into(), + did: "did:plc:owner".into(), owner_did: "did:plc:owner".into(), type_nsid: "com.example.forum".into(), skey: "main".into(), @@ -288,14 +291,11 @@ mod tests { let globals = lua.globals(); let space_table: mlua::Table = globals.get("space").unwrap(); assert_eq!( - space_table.get::("space_uri").unwrap(), + space_table.get::("space").unwrap(), "ats://did:plc:owner/com.example.forum/main" ); assert_eq!(space_table.get::("space_id").unwrap(), "space-123"); - assert_eq!( - space_table.get::("owner_did").unwrap(), - "did:plc:owner" - ); + assert_eq!(space_table.get::("did").unwrap(), "did:plc:owner"); } #[test] diff --git a/src/lua/mod.rs b/src/lua/mod.rs index 69cb3cc..58733eb 100644 --- a/src/lua/mod.rs +++ b/src/lua/mod.rs @@ -5,7 +5,7 @@ mod execute; mod http_api; mod record; pub(crate) mod sandbox; -mod tid; +pub(crate) mod tid; mod xrpc_api; #[allow(unused_imports)] diff --git a/src/spaces/auth.rs b/src/spaces/auth.rs index 2b70089..9996c1b 100644 --- a/src/spaces/auth.rs +++ b/src/spaces/auth.rs @@ -9,7 +9,7 @@ use crate::db::{DatabaseBackend, adapt_sql, now_rfc3339}; use crate::error::AppError; use crate::plugin::encryption::{decrypt, encrypt}; use crate::spaces::credential::{ - DEFAULT_CREDENTIAL_TTL_SECS, SpaceCredentialClaims, sign_credential, verify_credential, + DEFAULT_CREDENTIAL_TTL_SECS, SpaceCredentialClaims, sign_credential, }; use crate::spaces::types::{AccessMode, Space}; @@ -37,9 +37,9 @@ pub async fn issue_credential( let exp = now + DEFAULT_CREDENTIAL_TTL_SECS; let claims = SpaceCredentialClaims { - iss: space.owner_did.clone(), + iss: space.did.clone(), sub: subject_did.to_string(), - space: format!("{}/{}/{}", space.owner_did, space.type_nsid, space.skey), + space: format!("ats://{}/{}/{}", space.did, space.type_nsid, space.skey), scope: "read".into(), iat: now, exp, @@ -57,19 +57,6 @@ pub async fn issue_credential( Ok(IssuedCredential { token, expires_at }) } -pub async fn refresh_credential( - pool: &sqlx::AnyPool, - backend: DatabaseBackend, - encryption_key: &[u8; 32], - space: &Space, - current_token: &str, -) -> Result { - let public_jwk = get_public_key(pool, backend, encryption_key, space).await?; - let claims = verify_credential(current_token, &public_jwk)?; - - issue_credential(pool, backend, encryption_key, space, &claims.sub, None).await -} - pub fn check_app_access(space: &Space, client_id: Option<&str>) -> Result<(), AppError> { let Some(client_id) = client_id else { return Ok(()); @@ -148,7 +135,7 @@ async fn get_or_create_signing_key( sqlx::query(&insert_sql) .bind(Uuid::new_v4().to_string()) - .bind(&space.owner_did) + .bind(&space.did) .bind(&space.id) .bind(&encrypted_signing) .bind(&encrypted_rotation) @@ -161,21 +148,6 @@ async fn get_or_create_signing_key( Ok(keypair.private_jwk) } -async fn get_public_key( - pool: &sqlx::AnyPool, - backend: DatabaseBackend, - encryption_key: &[u8; 32], - space: &Space, -) -> Result { - let private_jwk = get_or_create_signing_key(pool, backend, encryption_key, space).await?; - Ok(serde_json::json!({ - "kty": "EC", - "crv": "P-256", - "x": private_jwk["x"], - "y": private_jwk["y"], - })) -} - struct SpaceKeypair { private_jwk: serde_json::Value, } @@ -252,6 +224,7 @@ mod tests { fn test_space(access_mode: AccessMode) -> Space { Space { id: "test-space".into(), + did: "did:plc:owner".into(), owner_did: "did:plc:owner".into(), type_nsid: "com.example.forum".into(), skey: "main".into(), @@ -262,6 +235,7 @@ mod tests { app_denylist: None, managing_app_did: None, config: SpaceConfig::default(), + revision: None, created_at: String::new(), updated_at: String::new(), } diff --git a/src/spaces/credential.rs b/src/spaces/credential.rs index a2bbbc3..d20054f 100644 --- a/src/spaces/credential.rs +++ b/src/spaces/credential.rs @@ -7,6 +7,43 @@ use crate::error::AppError; use crate::profile; pub const DEFAULT_CREDENTIAL_TTL_SECS: u64 = 4 * 60 * 60; // 4 hours +pub const GRANT_TTL_SECS: u64 = 5 * 60; // 5 minutes + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MemberGrantClaims { + pub sub: String, + pub space: String, + pub scope: String, + pub iat: u64, + pub exp: u64, +} + +pub fn sign_grant(claims: &MemberGrantClaims, secret: &[u8; 32]) -> Result { + let header = jsonwebtoken::Header::new(jsonwebtoken::Algorithm::HS256); + let key = jsonwebtoken::EncodingKey::from_secret(secret); + jsonwebtoken::encode(&header, claims, &key) + .map_err(|e| AppError::Internal(format!("failed to sign member grant: {e}"))) +} + +pub fn verify_grant(token: &str, secret: &[u8; 32]) -> Result { + let key = jsonwebtoken::DecodingKey::from_secret(secret); + let mut validation = jsonwebtoken::Validation::new(jsonwebtoken::Algorithm::HS256); + validation.required_spec_claims.clear(); + validation.validate_exp = false; + let data = jsonwebtoken::decode::(token, &key, &validation) + .map_err(|e| AppError::Auth(format!("invalid member grant: {e}")))?; + + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + + if now > data.claims.exp { + return Err(AppError::Auth("member grant has expired".into())); + } + + Ok(data.claims) +} #[derive(Debug, Clone, Serialize, Deserialize)] pub struct SpaceCredentialClaims { @@ -281,4 +318,73 @@ mod tests { let result = verify_credential("not-a-jwt", &keypair.public_jwk); assert!(result.is_err()); } + + fn test_secret() -> [u8; 32] { + [0xAB; 32] + } + + #[test] + fn grant_sign_and_verify_roundtrip() { + let secret = test_secret(); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + let claims = MemberGrantClaims { + sub: "did:plc:member".into(), + space: "ats://did:plc:space/com.example.forum/main".into(), + scope: "read".into(), + iat: now, + exp: now + GRANT_TTL_SECS, + }; + + let token = sign_grant(&claims, &secret).unwrap(); + let verified = verify_grant(&token, &secret).unwrap(); + + assert_eq!(verified.sub, claims.sub); + assert_eq!(verified.space, claims.space); + assert_eq!(verified.scope, claims.scope); + } + + #[test] + fn grant_rejects_wrong_secret() { + let secret1 = [0xAB; 32]; + let secret2 = [0xCD; 32]; + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + let claims = MemberGrantClaims { + sub: "did:plc:member".into(), + space: "ats://did:plc:space/com.example.forum/main".into(), + scope: "read".into(), + iat: now, + exp: now + GRANT_TTL_SECS, + }; + + let token = sign_grant(&claims, &secret1).unwrap(); + let result = verify_grant(&token, &secret2); + assert!(result.is_err()); + } + + #[test] + fn grant_rejects_expired() { + let secret = test_secret(); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + let claims = MemberGrantClaims { + sub: "did:plc:member".into(), + space: "ats://did:plc:space/com.example.forum/main".into(), + scope: "read".into(), + iat: now - 600, + exp: now - 300, + }; + + let token = sign_grant(&claims, &secret).unwrap(); + let result = verify_grant(&token, &secret); + assert!(result.is_err()); + assert!(result.unwrap_err().to_string().contains("expired")); + } } diff --git a/src/spaces/db.rs b/src/spaces/db.rs index 5577a50..88d19b3 100644 --- a/src/spaces/db.rs +++ b/src/spaces/db.rs @@ -24,12 +24,13 @@ pub async fn create_space( .map(|v| serde_json::to_string(v).unwrap_or_default()); let sql = adapt_sql( - "INSERT INTO spaces (id, owner_did, type_nsid, skey, display_name, description, access_mode, app_allowlist, app_denylist, managing_app_did, config, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + "INSERT INTO spaces (id, did, owner_did, type_nsid, skey, display_name, description, access_mode, app_allowlist, app_denylist, managing_app_did, config, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", backend, ); sqlx::query(&sql) .bind(&space.id) + .bind(&space.did) .bind(&space.owner_did) .bind(&space.type_nsid) .bind(&space.skey) @@ -55,7 +56,7 @@ pub async fn get_space( id: &str, ) -> Result, AppError> { let sql = adapt_sql( - "SELECT id, owner_did, type_nsid, skey, display_name, description, access_mode, app_allowlist, app_denylist, managing_app_did, config, created_at, updated_at FROM spaces WHERE id = ?", + "SELECT id, did, owner_did, type_nsid, skey, display_name, description, access_mode, app_allowlist, app_denylist, managing_app_did, config, revision, created_at, updated_at FROM spaces WHERE id = ?", backend, ); @@ -71,17 +72,17 @@ pub async fn get_space( pub async fn get_space_by_address( pool: &sqlx::AnyPool, backend: DatabaseBackend, - owner_did: &str, + did: &str, type_nsid: &str, skey: &str, ) -> Result, AppError> { let sql = adapt_sql( - "SELECT id, owner_did, type_nsid, skey, display_name, description, access_mode, app_allowlist, app_denylist, managing_app_did, config, created_at, updated_at FROM spaces WHERE owner_did = ? AND type_nsid = ? AND skey = ?", + "SELECT id, did, owner_did, type_nsid, skey, display_name, description, access_mode, app_allowlist, app_denylist, managing_app_did, config, revision, created_at, updated_at FROM spaces WHERE did = ? AND type_nsid = ? AND skey = ?", backend, ); let row: Option = sqlx::query_as(&sql) - .bind(owner_did) + .bind(did) .bind(type_nsid) .bind(skey) .fetch_optional(pool) @@ -94,15 +95,15 @@ pub async fn get_space_by_address( pub async fn list_spaces_by_owner( pool: &sqlx::AnyPool, backend: DatabaseBackend, - owner_did: &str, + did: &str, ) -> Result, AppError> { let sql = adapt_sql( - "SELECT id, owner_did, type_nsid, skey, display_name, description, access_mode, app_allowlist, app_denylist, managing_app_did, config, created_at, updated_at FROM spaces WHERE owner_did = ? ORDER BY created_at DESC", + "SELECT id, did, owner_did, type_nsid, skey, display_name, description, access_mode, app_allowlist, app_denylist, managing_app_did, config, revision, created_at, updated_at FROM spaces WHERE owner_did = ? ORDER BY created_at DESC", backend, ); let rows: Vec = sqlx::query_as(&sql) - .bind(owner_did) + .bind(did) .fetch_all(pool) .await .map_err(|e| AppError::Internal(format!("failed to list spaces: {e}")))?; @@ -110,6 +111,58 @@ pub async fn list_spaces_by_owner( rows.into_iter().map(parse_space_row).collect() } +pub struct SpaceView { + pub uri: String, + pub is_owner: bool, +} + +pub async fn list_spaces_for_user( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + did: &str, + limit: i64, + cursor: Option<&str>, +) -> Result, AppError> { + let (sql, has_cursor) = if cursor.is_some() { + ( + adapt_sql( + "SELECT s.did, s.owner_did, s.type_nsid, s.skey, sm.created_at FROM space_members sm JOIN spaces s ON s.id = sm.space_id WHERE sm.member_did = ? AND sm.created_at > ? ORDER BY sm.created_at ASC LIMIT ?", + backend, + ), + true, + ) + } else { + ( + adapt_sql( + "SELECT s.did, s.owner_did, s.type_nsid, s.skey, sm.created_at FROM space_members sm JOIN spaces s ON s.id = sm.space_id WHERE sm.member_did = ? ORDER BY sm.created_at ASC LIMIT ?", + backend, + ), + false, + ) + }; + + let mut query = sqlx::query_as::<_, (String, String, String, String, String)>(&sql).bind(did); + if has_cursor { + query = query.bind(cursor.unwrap()); + } + query = query.bind(limit); + + let rows = query + .fetch_all(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to list spaces for user: {e}")))?; + + Ok(rows + .into_iter() + .map( + |(space_did, owner_did, type_nsid, skey, _created_at)| SpaceView { + uri: format!("ats://{}/{}/{}", space_did, type_nsid, skey), + is_owner: owner_did == did, + }, + ) + .collect()) +} + pub async fn update_space( pool: &sqlx::AnyPool, backend: DatabaseBackend, @@ -170,6 +223,7 @@ type SpaceRow = ( String, String, String, + String, Option, Option, String, @@ -177,40 +231,43 @@ type SpaceRow = ( Option, Option, String, + Option, String, String, ); fn parse_space_row(r: SpaceRow) -> Result { - let access_mode = AccessMode::parse(&r.6) - .ok_or_else(|| AppError::Internal(format!("invalid access_mode: {}", r.6)))?; + let access_mode = AccessMode::parse(&r.7) + .ok_or_else(|| AppError::Internal(format!("invalid access_mode: {}", r.7)))?; let app_allowlist: Option> = - r.7.as_deref() + r.8.as_deref() .map(serde_json::from_str) .transpose() .map_err(|e| AppError::Internal(format!("invalid app_allowlist: {e}")))?; let app_denylist: Option> = - r.8.as_deref() + r.9.as_deref() .map(serde_json::from_str) .transpose() .map_err(|e| AppError::Internal(format!("invalid app_denylist: {e}")))?; - let config: SpaceConfig = serde_json::from_str(&r.10) + let config: SpaceConfig = serde_json::from_str(&r.11) .map_err(|e| AppError::Internal(format!("invalid space config: {e}")))?; Ok(Space { id: r.0, - owner_did: r.1, - type_nsid: r.2, - skey: r.3, - display_name: r.4, - description: r.5, + did: r.1, + owner_did: r.2, + type_nsid: r.3, + skey: r.4, + display_name: r.5, + description: r.6, access_mode, app_allowlist, app_denylist, - managing_app_did: r.9, + managing_app_did: r.10, config, - created_at: r.11, - updated_at: r.12, + revision: r.12, + created_at: r.13, + updated_at: r.14, }) } @@ -232,7 +289,7 @@ pub async fn add_member( sqlx::query(&sql) .bind(&member.id) .bind(&member.space_id) - .bind(&member.member_did) + .bind(&member.did) .bind(member.access.as_str()) .bind(member.is_delegation as i32) .bind(&member.granted_by) @@ -248,7 +305,7 @@ pub async fn remove_member( pool: &sqlx::AnyPool, backend: DatabaseBackend, space_id: &str, - member_did: &str, + did: &str, ) -> Result { let sql = adapt_sql( "DELETE FROM space_members WHERE space_id = ? AND member_did = ?", @@ -257,7 +314,7 @@ pub async fn remove_member( let result = sqlx::query(&sql) .bind(space_id) - .bind(member_did) + .bind(did) .execute(pool) .await .map_err(|e| AppError::Internal(format!("failed to remove member: {e}")))?; @@ -269,7 +326,7 @@ pub async fn get_member( pool: &sqlx::AnyPool, backend: DatabaseBackend, space_id: &str, - member_did: &str, + did: &str, ) -> Result, AppError> { let sql = adapt_sql( "SELECT id, space_id, member_did, access, is_delegation, granted_by, created_at FROM space_members WHERE space_id = ? AND member_did = ?", @@ -278,7 +335,7 @@ pub async fn get_member( let row: Option = sqlx::query_as(&sql) .bind(space_id) - .bind(member_did) + .bind(did) .fetch_optional(pool) .await .map_err(|e| AppError::Internal(format!("failed to get member: {e}")))?; @@ -308,7 +365,7 @@ pub async fn list_direct_members( pub async fn list_spaces_for_member( pool: &sqlx::AnyPool, backend: DatabaseBackend, - member_did: &str, + did: &str, ) -> Result, AppError> { let sql = adapt_sql( "SELECT id, space_id, member_did, access, is_delegation, granted_by, created_at FROM space_members WHERE member_did = ? ORDER BY created_at ASC", @@ -316,7 +373,7 @@ pub async fn list_spaces_for_member( ); let rows: Vec = sqlx::query_as(&sql) - .bind(member_did) + .bind(did) .fetch_all(pool) .await .map_err(|e| AppError::Internal(format!("failed to list spaces for member: {e}")))?; @@ -333,7 +390,7 @@ fn parse_member_row(r: MemberRow) -> Result { Ok(SpaceMember { id: r.0, space_id: r.1, - member_did: r.2, + did: r.2, access, is_delegation: r.4 != 0, granted_by: r.5, @@ -422,56 +479,49 @@ pub async fn get_space_record_by_parts( row.map(parse_record_row).transpose() } +#[allow(clippy::too_many_arguments)] pub async fn list_space_records( pool: &sqlx::AnyPool, backend: DatabaseBackend, space_id: &str, + repo: Option<&str>, collection: Option<&str>, limit: i64, cursor: Option<&str>, + reverse: bool, ) -> Result, AppError> { - let (sql, has_collection, has_cursor) = match (collection, cursor) { - (Some(_), Some(_)) => ( - adapt_sql( - "SELECT uri, space_id, author_did, collection, rkey, record, cid, indexed_at FROM space_records WHERE space_id = ? AND collection = ? AND indexed_at > ? ORDER BY indexed_at ASC LIMIT ?", - backend, - ), - true, - true, - ), - (Some(_), None) => ( - adapt_sql( - "SELECT uri, space_id, author_did, collection, rkey, record, cid, indexed_at FROM space_records WHERE space_id = ? AND collection = ? ORDER BY indexed_at ASC LIMIT ?", - backend, - ), - true, - false, - ), - (None, Some(_)) => ( - adapt_sql( - "SELECT uri, space_id, author_did, collection, rkey, record, cid, indexed_at FROM space_records WHERE space_id = ? AND indexed_at > ? ORDER BY indexed_at ASC LIMIT ?", - backend, - ), - false, - true, - ), - (None, None) => ( - adapt_sql( - "SELECT uri, space_id, author_did, collection, rkey, record, cid, indexed_at FROM space_records WHERE space_id = ? ORDER BY indexed_at ASC LIMIT ?", - backend, - ), - false, - false, - ), + let mut conditions = vec!["space_id = ?".to_string()]; + if repo.is_some() { + conditions.push("author_did = ?".to_string()); + } + if collection.is_some() { + conditions.push("collection = ?".to_string()); + } + let (cursor_op, order) = if reverse { + ("indexed_at < ?", "DESC") + } else { + ("indexed_at > ?", "ASC") }; + if cursor.is_some() { + conditions.push(cursor_op.to_string()); + } - let mut query = sqlx::query_as::<_, RecordRow>(&sql).bind(space_id); + let where_clause = conditions.join(" AND "); + let raw = format!( + "SELECT uri, space_id, author_did, collection, rkey, record, cid, indexed_at FROM space_records WHERE {} ORDER BY indexed_at {} LIMIT ?", + where_clause, order + ); + let sql = adapt_sql(&raw, backend); - if has_collection { - query = query.bind(collection.unwrap()); + let mut query = sqlx::query_as::<_, RecordRow>(&sql).bind(space_id); + if let Some(r) = repo { + query = query.bind(r); } - if has_cursor { - query = query.bind(cursor.unwrap()); + if let Some(c) = collection { + query = query.bind(c); + } + if let Some(cur) = cursor { + query = query.bind(cur); } query = query.bind(limit); @@ -483,6 +533,79 @@ pub async fn list_space_records( rows.into_iter().map(parse_record_row).collect() } +pub async fn insert_space_record( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + record: &SpaceRecord, +) -> Result<(), AppError> { + let now = now_rfc3339(); + let record_json = serde_json::to_string(&record.record) + .map_err(|e| AppError::Internal(format!("failed to serialize record: {e}")))?; + + let sql = adapt_sql( + "INSERT INTO space_records (uri, space_id, author_did, collection, rkey, record, cid, indexed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + backend, + ); + + sqlx::query(&sql) + .bind(&record.uri) + .bind(&record.space_id) + .bind(&record.author_did) + .bind(&record.collection) + .bind(&record.rkey) + .bind(&record_json) + .bind(&record.cid) + .bind(&now) + .execute(pool) + .await + .map_err(|e| { + let msg = e.to_string(); + if msg.contains("UNIQUE") || msg.contains("duplicate") || msg.contains("unique") { + AppError::Conflict("Record already exists".into()) + } else { + AppError::Internal(format!("failed to create space record: {e}")) + } + })?; + + Ok(()) +} + +pub async fn upsert_space_record_with_swap( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + record: &SpaceRecord, + swap_cid: &str, +) -> Result<(), AppError> { + let now = now_rfc3339(); + let record_json = serde_json::to_string(&record.record) + .map_err(|e| AppError::Internal(format!("failed to serialize record: {e}")))?; + + let sql = adapt_sql( + "UPDATE space_records SET record = ?, cid = ?, indexed_at = ? WHERE uri = ? AND cid = ?", + backend, + ); + + let result = sqlx::query(&sql) + .bind(&record_json) + .bind(&record.cid) + .bind(&now) + .bind(&record.uri) + .bind(swap_cid) + .execute(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to update space record: {e}")))?; + + if result.rows_affected() == 0 { + let existing = get_space_record(pool, backend, &record.uri).await?; + if existing.is_some() { + return Err(AppError::Conflict("Record CID mismatch".into())); + } + return Err(AppError::NotFound("Record not found".into())); + } + + Ok(()) +} + pub async fn delete_space_record( pool: &sqlx::AnyPool, backend: DatabaseBackend, @@ -499,6 +622,58 @@ pub async fn delete_space_record( Ok(result.rows_affected() > 0) } +pub async fn delete_space_record_with_swap( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + uri: &str, + swap_cid: &str, +) -> Result { + let sql = adapt_sql( + "DELETE FROM space_records WHERE uri = ? AND cid = ?", + backend, + ); + + let result = sqlx::query(&sql) + .bind(uri) + .bind(swap_cid) + .execute(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to delete space record: {e}")))?; + + if result.rows_affected() == 0 { + let existing = get_space_record(pool, backend, uri).await?; + if existing.is_some() { + return Err(AppError::Conflict("Record CID mismatch".into())); + } + return Err(AppError::NotFound("Record not found".into())); + } + + Ok(true) +} + +pub async fn update_space_revision( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + space_id: &str, + revision: &str, +) -> Result<(), AppError> { + let now = now_rfc3339(); + let sql = adapt_sql( + "UPDATE spaces SET revision = ?, updated_at = ? WHERE id = ?", + backend, + ); + + sqlx::query(&sql) + .bind(revision) + .bind(&now) + .bind(space_id) + .execute(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to update space revision: {e}")))?; + + Ok(()) +} + type RecordRow = ( String, String, diff --git a/src/spaces/members.rs b/src/spaces/members.rs index a7212d7..477df80 100644 --- a/src/spaces/members.rs +++ b/src/spaces/members.rs @@ -77,7 +77,7 @@ fn resolve_members_recursive<'a>( .await?; } } else { - merge_access(resolved, &member.member_did, member.access); + merge_access(resolved, &member.did, member.access); } } @@ -93,14 +93,13 @@ async fn resolve_delegation_target( backend: DatabaseBackend, member: &SpaceMember, ) -> Result, AppError> { - if member.member_did.starts_with("ats://") { - let uri = SpaceUri::parse(&member.member_did)?; + if member.did.starts_with("ats://") { + let uri = SpaceUri::parse(&member.did)?; let space = - db::get_space_by_address(pool, backend, &uri.owner_did, &uri.type_nsid, &uri.skey) - .await?; + db::get_space_by_address(pool, backend, &uri.did, &uri.type_nsid, &uri.skey).await?; Ok(space.map(|s| s.id)) } else { - let space = db::get_space(pool, backend, &member.member_did).await?; + let space = db::get_space(pool, backend, &member.did).await?; Ok(space.map(|s| s.id)) } } diff --git a/src/spaces/mod.rs b/src/spaces/mod.rs index 385f183..8dc3c7d 100644 --- a/src/spaces/mod.rs +++ b/src/spaces/mod.rs @@ -10,11 +10,11 @@ use std::fmt; /// A parsed `ats://` URI for addressing permissioned data. /// -/// Full form: `ats:///////` -/// Space-only form: `ats:////` +/// Full form: `ats:///////` +/// Space-only form: `ats:////` #[derive(Debug, Clone, PartialEq, Eq, Hash)] pub struct SpaceUri { - pub owner_did: String, + pub did: String, pub type_nsid: String, pub skey: String, pub user_did: Option, @@ -32,7 +32,7 @@ impl SpaceUri { if parts.len() < 3 { return Err(AppError::BadRequest( - "SpaceUri requires at least owner_did/type_nsid/skey".into(), + "SpaceUri requires at least did/type_nsid/skey".into(), )); } @@ -42,7 +42,7 @@ impl SpaceUri { )); } - let owner_did = parts[0].to_string(); + let did = parts[0].to_string(); let type_nsid = parts[1].to_string(); let skey = parts[2].to_string(); @@ -61,7 +61,7 @@ impl SpaceUri { }; Ok(SpaceUri { - owner_did, + did, type_nsid, skey, user_did, @@ -71,7 +71,7 @@ impl SpaceUri { } pub fn space_uri(&self) -> String { - format!("ats://{}/{}/{}", self.owner_did, self.type_nsid, self.skey) + format!("ats://{}/{}/{}", self.did, self.type_nsid, self.skey) } pub fn is_record_uri(&self) -> bool { @@ -85,11 +85,7 @@ impl SpaceUri { impl fmt::Display for SpaceUri { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!( - f, - "ats://{}/{}/{}", - self.owner_did, self.type_nsid, self.skey - )?; + write!(f, "ats://{}/{}/{}", self.did, self.type_nsid, self.skey)?; if let (Some(user), Some(col), Some(rkey)) = (&self.user_did, &self.collection, &self.rkey) { write!(f, "/{}/{}/{}", user, col, rkey)?; @@ -105,7 +101,7 @@ mod tests { #[test] fn parse_space_uri() { let uri = SpaceUri::parse("ats://did:plc:abc123/com.example.forum/main").unwrap(); - assert_eq!(uri.owner_did, "did:plc:abc123"); + assert_eq!(uri.did, "did:plc:abc123"); assert_eq!(uri.type_nsid, "com.example.forum"); assert_eq!(uri.skey, "main"); assert!(uri.is_space_uri()); @@ -119,7 +115,7 @@ mod tests { "ats://did:plc:abc123/com.example.forum/main/did:plc:user1/com.example.forum.post/3k2abc", ) .unwrap(); - assert_eq!(uri.owner_did, "did:plc:abc123"); + assert_eq!(uri.did, "did:plc:abc123"); assert_eq!(uri.type_nsid, "com.example.forum"); assert_eq!(uri.skey, "main"); assert_eq!(uri.user_did.as_deref(), Some("did:plc:user1")); @@ -132,7 +128,7 @@ mod tests { #[test] fn display_space_uri() { let uri = SpaceUri { - owner_did: "did:plc:abc123".into(), + did: "did:plc:abc123".into(), type_nsid: "com.example.forum".into(), skey: "main".into(), user_did: None, @@ -148,7 +144,7 @@ mod tests { #[test] fn display_record_uri() { let uri = SpaceUri { - owner_did: "did:plc:abc123".into(), + did: "did:plc:abc123".into(), type_nsid: "com.example.forum".into(), skey: "main".into(), user_did: Some("did:plc:user1".into()), diff --git a/src/spaces/routes.rs b/src/spaces/routes.rs index 543707c..29abc02 100644 --- a/src/spaces/routes.rs +++ b/src/spaces/routes.rs @@ -11,6 +11,7 @@ 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::types::*; use crate::spaces::{SpaceUri, db, members}; @@ -21,6 +22,7 @@ use crate::spaces::{SpaceUri, db, members}; #[derive(Deserialize)] #[serde(rename_all = "camelCase")] struct CreateSpaceInput { + #[serde(rename = "type")] type_nsid: String, skey: String, display_name: Option, @@ -33,25 +35,27 @@ struct CreateSpaceInput { #[derive(Deserialize)] #[serde(rename_all = "camelCase")] struct SpaceUriQuery { - space_uri: String, + space: String, } #[derive(Deserialize)] #[serde(rename_all = "camelCase")] struct ListSpacesQuery { - owner_did: Option, + did: Option, + limit: Option, + cursor: Option, } #[derive(Deserialize)] #[serde(rename_all = "camelCase")] struct DeleteSpaceInput { - space_uri: String, + space: String, } #[derive(Deserialize)] #[serde(rename_all = "camelCase")] struct UpdateSpaceInput { - space_uri: String, + space: String, display_name: Option>, description: Option>, access_mode: Option, @@ -64,24 +68,26 @@ struct UpdateSpaceInput { #[derive(Deserialize)] #[serde(rename_all = "camelCase")] struct PutRecordInput { - space_uri: String, + space: String, collection: String, rkey: String, record: serde_json::Value, + swap_record: Option, } #[derive(Deserialize)] #[serde(rename_all = "camelCase")] struct DeleteRecordInput { - space_uri: String, + space: String, collection: String, rkey: String, + swap_record: Option, } #[derive(Deserialize)] #[serde(rename_all = "camelCase")] struct GetRecordQuery { - space_uri: String, + space: String, collection: String, rkey: String, } @@ -89,17 +95,19 @@ struct GetRecordQuery { #[derive(Deserialize)] #[serde(rename_all = "camelCase")] struct ListRecordsQuery { - space_uri: String, + space: String, + repo: Option, collection: Option, limit: Option, cursor: Option, + reverse: Option, } #[derive(Deserialize)] #[serde(rename_all = "camelCase")] struct AddMemberInput { - space_uri: String, - member_did: String, + space: String, + did: String, access: Option, is_delegation: Option, } @@ -107,14 +115,14 @@ struct AddMemberInput { #[derive(Deserialize)] #[serde(rename_all = "camelCase")] struct RemoveMemberInput { - space_uri: String, - member_did: String, + space: String, + did: String, } #[derive(Deserialize)] #[serde(rename_all = "camelCase")] struct CreateInviteInput { - space_uri: String, + space: String, access: Option, max_uses: Option, expires_at: Option, @@ -129,21 +137,59 @@ struct RedeemInviteInput { #[derive(Deserialize)] #[serde(rename_all = "camelCase")] struct RevokeInviteInput { - space_uri: String, + space: String, invite_id: String, } #[derive(Deserialize)] #[serde(rename_all = "camelCase")] -struct GetCredentialInput { - space_uri: String, +struct GetMemberGrantInput { + space: 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 RefreshCredentialInput { - space_uri: String, - credential: String, +struct ApplyWritesInput { + space: String, + swap_commit: Option, + writes: Vec, +} + +#[derive(Deserialize)] +#[serde(tag = "action", rename_all = "camelCase")] +enum WriteOp { + Create { + collection: String, + rkey: Option, + value: serde_json::Value, + }, + Update { + collection: String, + rkey: String, + value: serde_json::Value, + #[serde(rename = "swapRecord")] + swap_record: Option, + }, + Delete { + collection: String, + rkey: String, + #[serde(rename = "swapRecord")] + swap_record: Option, + }, } // --------------------------------------------------------------------------- @@ -161,11 +207,16 @@ pub fn space_routes() -> Router { .route(&format!("/xrpc/{NS}.space.delete"), post(delete_space)) .route(&format!("/xrpc/{NS}.space.update"), post(update_space)) // Records + .route( + &format!("/xrpc/{NS}.space.createRecord"), + post(create_record), + ) .route(&format!("/xrpc/{NS}.space.putRecord"), post(put_record)) .route( &format!("/xrpc/{NS}.space.deleteRecord"), post(delete_record), ) + .route(&format!("/xrpc/{NS}.space.applyWrites"), post(apply_writes)) .route(&format!("/xrpc/{NS}.space.getRecord"), get(get_record)) .route(&format!("/xrpc/{NS}.space.listRecords"), get(list_records)) // Members @@ -191,12 +242,12 @@ pub fn space_routes() -> Router { .route(&format!("/xrpc/{NS}.space.invite.list"), get(list_invites)) // Credentials .route( - &format!("/xrpc/{NS}.space.getCredential"), - post(get_credential), + &format!("/xrpc/{NS}.space.getMemberGrant"), + post(get_member_grant), ) .route( - &format!("/xrpc/{NS}.space.refreshCredential"), - post(refresh_credential), + &format!("/xrpc/{NS}.space.getSpaceCredential"), + post(get_space_credential), ) } @@ -216,7 +267,7 @@ async fn resolve_space(state: &AppState, space_uri: &str) -> Result, ) -> Result { if let Some(token) = space_credential { - let space_uri = format!( - "ats://{}/{}/{}", - space.owner_did, space.type_nsid, space.skey - ); + let space_uri = format!("ats://{}/{}/{}", space.did, space.type_nsid, space.skey); match crate::spaces::credential::verify_external_credential( token, &state.http, @@ -319,9 +367,7 @@ async fn create_space( let did = claims.did().to_string(); if input.type_nsid.is_empty() || input.skey.is_empty() { - return Err(AppError::BadRequest( - "type_nsid and skey are required".into(), - )); + return Err(AppError::BadRequest("type and skey are required".into())); } let existing = db::get_space_by_address( @@ -340,6 +386,7 @@ async fn create_space( let space = Space { id: Uuid::new_v4().to_string(), + did: did.clone(), owner_did: did.clone(), type_nsid: input.type_nsid, skey: input.skey, @@ -350,6 +397,7 @@ async fn create_space( app_denylist: None, managing_app_did: input.managing_app_did, config: input.config.unwrap_or_default(), + revision: None, created_at: now_rfc3339(), updated_at: now_rfc3339(), }; @@ -360,7 +408,7 @@ async fn create_space( let member = SpaceMember { id: Uuid::new_v4().to_string(), space_id: space.id.clone(), - member_did: did.clone(), + did: did.clone(), access: SpaceAccess::Write, is_delegation: false, granted_by: Some(did), @@ -368,13 +416,9 @@ async fn create_space( }; db::add_member(&state.db, state.db_backend, &member).await?; - let space_uri = format!( - "ats://{}/{}/{}", - space.owner_did, space.type_nsid, space.skey - ); + let space_uri = format!("ats://{}/{}/{}", space.did, space.type_nsid, space.skey); let body = serde_json::json!({ - "spaceUri": space_uri, - "space": space, + "uri": space_uri, }); let mut response = Json(body).into_response(); @@ -387,7 +431,7 @@ async fn get_space( xrpc_claims: XrpcClaims, Query(query): Query, ) -> Result, AppError> { - let space = resolve_space(&state, &query.space_uri).await?; + let space = resolve_space(&state, &query.space).await?; // If the space's membership is not public, require auth + membership if !space.config.membership_public { @@ -400,12 +444,9 @@ async fn get_space( } } - let space_uri = format!( - "ats://{}/{}/{}", - space.owner_did, space.type_nsid, space.skey - ); + let space_uri = format!("ats://{}/{}/{}", space.did, space.type_nsid, space.skey); Ok(Json(serde_json::json!({ - "spaceUri": space_uri, + "uri": space_uri, "space": space, }))) } @@ -416,20 +457,38 @@ async fn list_spaces( Query(query): Query, ) -> Result, AppError> { let claims = require_auth(&xrpc_claims)?; - let did = claims.did().to_string(); + let did = query.did.unwrap_or_else(|| claims.did().to_string()); + let limit = query.limit.unwrap_or(50).min(100); - let owner = query.owner_did.as_deref().unwrap_or(&did); - let spaces = db::list_spaces_by_owner(&state.db, state.db_backend, owner).await?; + let views = db::list_spaces_for_user( + &state.db, + state.db_backend, + &did, + limit, + query.cursor.as_deref(), + ) + .await?; + + let cursor = if views.len() as i64 == limit { + views.last().map(|v| v.uri.clone()) + } else { + None + }; - let spaces_with_uris: Vec = spaces + let spaces_json: Vec = views .into_iter() - .map(|s| { - let uri = format!("ats://{}/{}/{}", s.owner_did, s.type_nsid, s.skey); - serde_json::json!({ "spaceUri": uri, "space": s }) + .map(|v| { + serde_json::json!({ + "uri": v.uri, + "isOwner": v.is_owner, + }) }) .collect(); - Ok(Json(serde_json::json!({ "spaces": spaces_with_uris }))) + Ok(Json(serde_json::json!({ + "spaces": spaces_json, + "cursor": cursor, + }))) } async fn delete_space( @@ -438,7 +497,7 @@ async fn delete_space( Json(input): Json, ) -> Result, AppError> { let claims = require_auth(&xrpc_claims)?; - let space = resolve_space(&state, &input.space_uri).await?; + let space = resolve_space(&state, &input.space).await?; require_space_admin(&state, &space, claims.did()).await?; db::delete_space(&state.db, state.db_backend, &space.id).await?; @@ -452,7 +511,7 @@ async fn update_space( Json(input): Json, ) -> Result, AppError> { let claims = require_auth(&xrpc_claims)?; - let mut space = resolve_space(&state, &input.space_uri).await?; + let mut space = resolve_space(&state, &input.space).await?; require_space_admin(&state, &space, claims.did()).await?; if let Some(name) = input.display_name { @@ -479,12 +538,9 @@ async fn update_space( db::update_space(&state.db, state.db_backend, &space).await?; - let space_uri = format!( - "ats://{}/{}/{}", - space.owner_did, space.type_nsid, space.skey - ); + let space_uri = format!("ats://{}/{}/{}", space.did, space.type_nsid, space.skey); Ok(Json(serde_json::json!({ - "spaceUri": space_uri, + "uri": space_uri, "space": space, }))) } @@ -493,6 +549,51 @@ async fn update_space( // Record handlers // --------------------------------------------------------------------------- +async fn create_record( + State(state): State, + xrpc_claims: XrpcClaims, + headers: HeaderMap, + Json(input): Json, +) -> Result { + let claims = require_auth(&xrpc_claims)?; + let did = claims.did().to_string(); + let space = resolve_space(&state, &input.space).await?; + let cred = extract_space_credential(&headers); + require_membership(&state, &space, &did, true, cred.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, xrpc_claims: XrpcClaims, @@ -501,19 +602,19 @@ async fn put_record( ) -> Result { let claims = require_auth(&xrpc_claims)?; let did = claims.did().to_string(); - let space = resolve_space(&state, &input.space_uri).await?; + let space = resolve_space(&state, &input.space).await?; let cred = extract_space_credential(&headers); require_membership(&state, &space, &did, true, cred.as_deref()).await?; let cid = content_cid(&input.record); let record_uri = format!( "ats://{}/{}/{}/{}/{}/{}", - space.owner_did, space.type_nsid, space.skey, did, input.collection, input.rkey + space.did, space.type_nsid, space.skey, did, input.collection, input.rkey ); let record = SpaceRecord { uri: record_uri.clone(), - space_id: space.id, + space_id: space.id.clone(), author_did: did, collection: input.collection, rkey: input.rkey, @@ -522,7 +623,14 @@ async fn put_record( indexed_at: now_rfc3339(), }; - db::upsert_space_record(&state.db, state.db_backend, &record).await?; + 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, @@ -541,31 +649,163 @@ async fn delete_record( ) -> Result, AppError> { let claims = require_auth(&xrpc_claims)?; let did = claims.did().to_string(); - let space = resolve_space(&state, &input.space_uri).await?; + let space = resolve_space(&state, &input.space).await?; let record_uri = format!( "ats://{}/{}/{}/{}/{}/{}", - space.owner_did, space.type_nsid, space.skey, did, input.collection, input.rkey + space.did, space.type_nsid, space.skey, did, input.collection, input.rkey ); - 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())); + 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?; } - 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, + xrpc_claims: XrpcClaims, + headers: HeaderMap, + Json(input): Json, +) -> Result, AppError> { + let claims = require_auth(&xrpc_claims)?; + let did = claims.did().to_string(); + let space = resolve_space(&state, &input.space).await?; + let cred = extract_space_credential(&headers); + require_membership(&state, &space, &did, true, cred.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, xrpc_claims: XrpcClaims, @@ -573,7 +813,7 @@ async fn get_record( Query(query): Query, ) -> Result, AppError> { let claims = require_auth(&xrpc_claims)?; - let space = resolve_space(&state, &query.space_uri).await?; + let space = resolve_space(&state, &query.space).await?; let cred = extract_space_credential(&headers); require_membership(&state, &space, claims.did(), false, cred.as_deref()).await?; @@ -589,10 +829,8 @@ async fn get_record( Ok(Json(serde_json::json!({ "uri": record.uri, - "space": query.space_uri, - "collection": record.collection, - "record": record.record, "cid": record.cid, + "value": record.record, }))) } @@ -603,18 +841,29 @@ async fn list_records( Query(query): Query, ) -> Result, AppError> { let claims = require_auth(&xrpc_claims)?; - let space = resolve_space(&state, &query.space_uri).await?; + let space = resolve_space(&state, &query.space).await?; let cred = extract_space_credential(&headers); require_membership(&state, &space, claims.did(), false, cred.as_deref()).await?; + let repo = query.repo.as_deref().or_else(|| { + if cred.is_some() { + None + } else { + Some(claims.did()) + } + }); + let limit = query.limit.unwrap_or(50).min(100); + let reverse = query.reverse.unwrap_or(false); let records = db::list_space_records( &state.db, state.db_backend, &space.id, + repo, query.collection.as_deref(), limit, query.cursor.as_deref(), + reverse, ) .await?; @@ -624,10 +873,8 @@ async fn list_records( .into_iter() .map(|r| { serde_json::json!({ - "uri": r.uri, - "space": query.space_uri, "collection": r.collection, - "record": r.record, + "rkey": r.rkey, "cid": r.cid, }) }) @@ -649,7 +896,7 @@ async fn list_members( headers: HeaderMap, Query(query): Query, ) -> Result, AppError> { - let space = resolve_space(&state, &query.space_uri).await?; + let space = resolve_space(&state, &query.space).await?; if !space.config.membership_public { let claims = require_auth(&xrpc_claims)?; @@ -668,11 +915,10 @@ async fn add_member( Json(input): Json, ) -> Result { let claims = require_auth(&xrpc_claims)?; - let space = resolve_space(&state, &input.space_uri).await?; + let space = resolve_space(&state, &input.space).await?; require_space_admin(&state, &space, claims.did()).await?; - let existing = - db::get_member(&state.db, state.db_backend, &space.id, &input.member_did).await?; + let existing = db::get_member(&state.db, state.db_backend, &space.id, &input.did).await?; if existing.is_some() { return Err(AppError::Conflict( "Member already exists in this space".into(), @@ -682,7 +928,7 @@ async fn add_member( let member = SpaceMember { id: Uuid::new_v4().to_string(), space_id: space.id, - member_did: input.member_did, + did: input.did, access: input.access.unwrap_or(SpaceAccess::Read), is_delegation: input.is_delegation.unwrap_or(false), granted_by: Some(claims.did().to_string()), @@ -702,11 +948,10 @@ async fn remove_member( Json(input): Json, ) -> Result, AppError> { let claims = require_auth(&xrpc_claims)?; - let space = resolve_space(&state, &input.space_uri).await?; + let space = resolve_space(&state, &input.space).await?; require_space_admin(&state, &space, claims.did()).await?; - let removed = - db::remove_member(&state.db, state.db_backend, &space.id, &input.member_did).await?; + let removed = db::remove_member(&state.db, state.db_backend, &space.id, &input.did).await?; if !removed { return Err(AppError::NotFound("Member not found in this space".into())); @@ -725,7 +970,7 @@ async fn create_invite( Json(input): Json, ) -> Result { let claims = require_auth(&xrpc_claims)?; - let space = resolve_space(&state, &input.space_uri).await?; + let space = resolve_space(&state, &input.space).await?; require_space_admin(&state, &space, claims.did()).await?; let mut token_bytes = [0u8; 24]; @@ -802,7 +1047,7 @@ async fn redeem_invite( let member = SpaceMember { id: Uuid::new_v4().to_string(), space_id: invite.space_id.clone(), - member_did: did, + did, access: invite.access, is_delegation: false, granted_by: Some(invite.created_by.clone()), @@ -813,10 +1058,10 @@ async fn redeem_invite( 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.owner_did, s.type_nsid, s.skey)); + let space_uri = space.map(|s| format!("ats://{}/{}/{}", s.did, s.type_nsid, s.skey)); let mut response = Json(serde_json::json!({ - "spaceUri": space_uri, + "uri": space_uri, "access": member.access, })) .into_response(); @@ -830,7 +1075,7 @@ async fn revoke_invite( Json(input): Json, ) -> Result, AppError> { let claims = require_auth(&xrpc_claims)?; - let space = resolve_space(&state, &input.space_uri).await?; + 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?; @@ -847,7 +1092,7 @@ async fn list_invites( Query(query): Query, ) -> Result, AppError> { let claims = require_auth(&xrpc_claims)?; - let space = resolve_space(&state, &query.space_uri).await?; + 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?; @@ -875,14 +1120,14 @@ async fn list_invites( // Credential handlers // --------------------------------------------------------------------------- -async fn get_credential( +async fn get_member_grant( State(state): State, xrpc_claims: XrpcClaims, - Json(input): Json, + Json(input): Json, ) -> Result, AppError> { let claims = require_auth(&xrpc_claims)?; let did = claims.did().to_string(); - let space = resolve_space(&state, &input.space_uri).await?; + let space = resolve_space(&state, &input.space).await?; require_membership(&state, &space, &did, false, None).await?; @@ -890,41 +1135,56 @@ async fn get_credential( AppError::Internal("TOKEN_ENCRYPTION_KEY is required for space credentials".into()) })?; - let client_id = claims.client_key().map(|k| k.to_string()); - let issued = crate::spaces::auth::issue_credential( - &state.db, - state.db_backend, - encryption_key, - &space, - &did, - client_id.as_deref(), - ) - .await?; + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + let exp = now + crate::spaces::credential::GRANT_TTL_SECS; + + let space_uri = format!("ats://{}/{}/{}", space.did, space.type_nsid, space.skey); + let grant_claims = crate::spaces::credential::MemberGrantClaims { + sub: did, + space: space_uri, + scope: "read".into(), + iat: now, + exp, + }; + + let grant = crate::spaces::credential::sign_grant(&grant_claims, encryption_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!({ - "credential": issued.token, - "expiresAt": issued.expires_at, + "grant": grant, + "expiresAt": expires_at, }))) } -async fn refresh_credential( +async fn get_space_credential( State(state): State, xrpc_claims: XrpcClaims, - Json(input): Json, + Json(input): Json, ) -> Result, AppError> { - let _claims = require_auth(&xrpc_claims)?; - let space = resolve_space(&state, &input.space_uri).await?; + 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 issued = crate::spaces::auth::refresh_credential( + let grant_claims = crate::spaces::credential::verify_grant(&input.grant, encryption_key)?; + + let space = resolve_space(&state, &grant_claims.space).await?; + + let client_id = claims.client_key().map(|k| k.to_string()); + let issued = crate::spaces::auth::issue_credential( &state.db, state.db_backend, encryption_key, &space, - &input.credential, + &grant_claims.sub, + client_id.as_deref(), ) .await?; @@ -933,3 +1193,211 @@ async fn refresh_credential( "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::(json!({ + "action": "unknown", + "collection": "test", + "rkey": "key" + })); + assert!(result.is_err()); + } +} diff --git a/src/spaces/types.rs b/src/spaces/types.rs index c62c116..db5f902 100644 --- a/src/spaces/types.rs +++ b/src/spaces/types.rs @@ -72,7 +72,9 @@ impl fmt::Display for AccessMode { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Space { pub id: String, + pub did: String, pub owner_did: String, + #[serde(rename = "type")] pub type_nsid: String, pub skey: String, pub display_name: Option, @@ -82,6 +84,7 @@ pub struct Space { pub app_denylist: Option>, pub managing_app_did: Option, pub config: SpaceConfig, + pub revision: Option, pub created_at: String, pub updated_at: String, } @@ -100,7 +103,7 @@ pub struct SpaceConfig { pub struct SpaceMember { pub id: String, pub space_id: String, - pub member_did: String, + pub did: String, pub access: SpaceAccess, pub is_delegation: bool, pub granted_by: Option,