diff --git a/migrations/postgres/20260429000000_create_spaces.sql b/migrations/postgres/20260429000000_create_spaces.sql new file mode 100644 --- /dev/null +++ b/migrations/postgres/20260429000000_create_spaces.sql @@ -0,0 +1,19 @@ +CREATE TABLE IF NOT EXISTS spaces ( + id TEXT PRIMARY KEY, + owner_did TEXT NOT NULL, + type_nsid TEXT NOT NULL, + skey TEXT NOT NULL, + display_name TEXT, + description TEXT, + access_mode TEXT NOT NULL DEFAULT 'default_allow', + app_allowlist TEXT, + app_denylist TEXT, + managing_app_did TEXT, + config TEXT NOT NULL DEFAULT '{}', + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + UNIQUE(owner_did, type_nsid, skey) +); + +CREATE INDEX idx_spaces_owner_did ON spaces(owner_did); +CREATE INDEX idx_spaces_type_nsid ON spaces(type_nsid); diff --git a/migrations/postgres/20260429000001_create_space_members.sql b/migrations/postgres/20260429000001_create_space_members.sql new file mode 100644 --- /dev/null +++ b/migrations/postgres/20260429000001_create_space_members.sql @@ -0,0 +1,13 @@ +CREATE TABLE IF NOT EXISTS space_members ( + id TEXT PRIMARY KEY, + space_id TEXT NOT NULL REFERENCES spaces(id) ON DELETE CASCADE, + member_did TEXT NOT NULL, + access TEXT NOT NULL DEFAULT 'read', + is_delegation INTEGER NOT NULL DEFAULT 0, + granted_by TEXT, + created_at TEXT NOT NULL, + UNIQUE(space_id, member_did) +); + +CREATE INDEX idx_space_members_did ON space_members(member_did); +CREATE INDEX idx_space_members_space_id ON space_members(space_id); diff --git a/migrations/postgres/20260429000002_create_space_records.sql b/migrations/postgres/20260429000002_create_space_records.sql new file mode 100644 --- /dev/null +++ b/migrations/postgres/20260429000002_create_space_records.sql @@ -0,0 +1,14 @@ +CREATE TABLE IF NOT EXISTS space_records ( + uri TEXT PRIMARY KEY, + space_id TEXT NOT NULL REFERENCES spaces(id) ON DELETE CASCADE, + author_did TEXT NOT NULL, + collection TEXT NOT NULL, + rkey TEXT NOT NULL, + record TEXT NOT NULL, + cid TEXT NOT NULL, + indexed_at TEXT NOT NULL +); + +CREATE INDEX idx_space_records_space_id ON space_records(space_id); +CREATE INDEX idx_space_records_author ON space_records(author_did); +CREATE INDEX idx_space_records_collection ON space_records(space_id, collection); diff --git a/migrations/postgres/20260429000003_create_space_credentials.sql b/migrations/postgres/20260429000003_create_space_credentials.sql new file mode 100644 --- /dev/null +++ b/migrations/postgres/20260429000003_create_space_credentials.sql @@ -0,0 +1,10 @@ +CREATE TABLE IF NOT EXISTS space_credentials ( + id TEXT PRIMARY KEY, + space_id TEXT NOT NULL REFERENCES spaces(id) ON DELETE CASCADE, + issued_to TEXT NOT NULL, + token_hash TEXT NOT NULL, + expires_at TEXT NOT NULL, + created_at TEXT NOT NULL +); + +CREATE INDEX idx_space_credentials_space_id ON space_credentials(space_id); diff --git a/migrations/postgres/20260429000004_create_space_invites.sql b/migrations/postgres/20260429000004_create_space_invites.sql new file mode 100644 --- /dev/null +++ b/migrations/postgres/20260429000004_create_space_invites.sql @@ -0,0 +1,14 @@ +CREATE TABLE IF NOT EXISTS space_invites ( + id TEXT PRIMARY KEY, + space_id TEXT NOT NULL REFERENCES spaces(id) ON DELETE CASCADE, + token_hash TEXT NOT NULL UNIQUE, + created_by TEXT NOT NULL, + access TEXT NOT NULL DEFAULT 'read', + max_uses INTEGER, + uses INTEGER NOT NULL DEFAULT 0, + expires_at TEXT, + revoked INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL +); + +CREATE INDEX idx_space_invites_space_id ON space_invites(space_id); diff --git a/migrations/postgres/20260429000005_create_space_dids.sql b/migrations/postgres/20260429000005_create_space_dids.sql new file mode 100644 --- /dev/null +++ b/migrations/postgres/20260429000005_create_space_dids.sql @@ -0,0 +1,12 @@ +CREATE TABLE IF NOT EXISTS space_dids ( + id TEXT PRIMARY KEY, + did TEXT NOT NULL UNIQUE, + space_id TEXT REFERENCES spaces(id) ON DELETE SET NULL, + signing_key_enc BYTEA NOT NULL, + rotation_key_enc BYTEA NOT NULL, + created_by TEXT NOT NULL, + created_at TEXT NOT NULL +); + +CREATE INDEX idx_space_dids_did ON space_dids(did); +CREATE INDEX idx_space_dids_space_id ON space_dids(space_id); diff --git a/migrations/postgres/20260429000006_create_space_sync_state.sql b/migrations/postgres/20260429000006_create_space_sync_state.sql new file mode 100644 --- /dev/null +++ b/migrations/postgres/20260429000006_create_space_sync_state.sql @@ -0,0 +1,12 @@ +CREATE TABLE IF NOT EXISTS space_sync_state ( + id TEXT PRIMARY KEY, + space_id TEXT NOT NULL REFERENCES spaces(id) ON DELETE CASCADE, + member_did TEXT NOT NULL, + cursor TEXT, + last_synced_at TEXT, + status TEXT NOT NULL DEFAULT 'pending', + error TEXT, + UNIQUE(space_id, member_did) +); + +CREATE INDEX idx_space_sync_state_space_id ON space_sync_state(space_id); diff --git a/migrations/sqlite/20260429000000_create_spaces.sql b/migrations/sqlite/20260429000000_create_spaces.sql new file mode 100644 --- /dev/null +++ b/migrations/sqlite/20260429000000_create_spaces.sql @@ -0,0 +1,19 @@ +CREATE TABLE IF NOT EXISTS spaces ( + id TEXT PRIMARY KEY, + owner_did TEXT NOT NULL, + type_nsid TEXT NOT NULL, + skey TEXT NOT NULL, + display_name TEXT, + description TEXT, + access_mode TEXT NOT NULL DEFAULT 'default_allow', + app_allowlist TEXT, + app_denylist TEXT, + managing_app_did TEXT, + config TEXT NOT NULL DEFAULT '{}', + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + UNIQUE(owner_did, type_nsid, skey) +); + +CREATE INDEX idx_spaces_owner_did ON spaces(owner_did); +CREATE INDEX idx_spaces_type_nsid ON spaces(type_nsid); diff --git a/migrations/sqlite/20260429000001_create_space_members.sql b/migrations/sqlite/20260429000001_create_space_members.sql new file mode 100644 --- /dev/null +++ b/migrations/sqlite/20260429000001_create_space_members.sql @@ -0,0 +1,13 @@ +CREATE TABLE IF NOT EXISTS space_members ( + id TEXT PRIMARY KEY, + space_id TEXT NOT NULL REFERENCES spaces(id) ON DELETE CASCADE, + member_did TEXT NOT NULL, + access TEXT NOT NULL DEFAULT 'read', + is_delegation INTEGER NOT NULL DEFAULT 0, + granted_by TEXT, + created_at TEXT NOT NULL, + UNIQUE(space_id, member_did) +); + +CREATE INDEX idx_space_members_did ON space_members(member_did); +CREATE INDEX idx_space_members_space_id ON space_members(space_id); diff --git a/migrations/sqlite/20260429000002_create_space_records.sql b/migrations/sqlite/20260429000002_create_space_records.sql new file mode 100644 --- /dev/null +++ b/migrations/sqlite/20260429000002_create_space_records.sql @@ -0,0 +1,14 @@ +CREATE TABLE IF NOT EXISTS space_records ( + uri TEXT PRIMARY KEY, + space_id TEXT NOT NULL REFERENCES spaces(id) ON DELETE CASCADE, + author_did TEXT NOT NULL, + collection TEXT NOT NULL, + rkey TEXT NOT NULL, + record TEXT NOT NULL, + cid TEXT NOT NULL, + indexed_at TEXT NOT NULL +); + +CREATE INDEX idx_space_records_space_id ON space_records(space_id); +CREATE INDEX idx_space_records_author ON space_records(author_did); +CREATE INDEX idx_space_records_collection ON space_records(space_id, collection); diff --git a/migrations/sqlite/20260429000003_create_space_credentials.sql b/migrations/sqlite/20260429000003_create_space_credentials.sql new file mode 100644 --- /dev/null +++ b/migrations/sqlite/20260429000003_create_space_credentials.sql @@ -0,0 +1,10 @@ +CREATE TABLE IF NOT EXISTS space_credentials ( + id TEXT PRIMARY KEY, + space_id TEXT NOT NULL REFERENCES spaces(id) ON DELETE CASCADE, + issued_to TEXT NOT NULL, + token_hash TEXT NOT NULL, + expires_at TEXT NOT NULL, + created_at TEXT NOT NULL +); + +CREATE INDEX idx_space_credentials_space_id ON space_credentials(space_id); diff --git a/migrations/sqlite/20260429000004_create_space_invites.sql b/migrations/sqlite/20260429000004_create_space_invites.sql new file mode 100644 --- /dev/null +++ b/migrations/sqlite/20260429000004_create_space_invites.sql @@ -0,0 +1,14 @@ +CREATE TABLE IF NOT EXISTS space_invites ( + id TEXT PRIMARY KEY, + space_id TEXT NOT NULL REFERENCES spaces(id) ON DELETE CASCADE, + token_hash TEXT NOT NULL UNIQUE, + created_by TEXT NOT NULL, + access TEXT NOT NULL DEFAULT 'read', + max_uses INTEGER, + uses INTEGER NOT NULL DEFAULT 0, + expires_at TEXT, + revoked INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL +); + +CREATE INDEX idx_space_invites_space_id ON space_invites(space_id); diff --git a/migrations/sqlite/20260429000005_create_space_dids.sql b/migrations/sqlite/20260429000005_create_space_dids.sql new file mode 100644 --- /dev/null +++ b/migrations/sqlite/20260429000005_create_space_dids.sql @@ -0,0 +1,12 @@ +CREATE TABLE IF NOT EXISTS space_dids ( + id TEXT PRIMARY KEY, + did TEXT NOT NULL UNIQUE, + space_id TEXT REFERENCES spaces(id) ON DELETE SET NULL, + signing_key_enc BLOB NOT NULL, + rotation_key_enc BLOB NOT NULL, + created_by TEXT NOT NULL, + created_at TEXT NOT NULL +); + +CREATE INDEX idx_space_dids_did ON space_dids(did); +CREATE INDEX idx_space_dids_space_id ON space_dids(space_id); diff --git a/migrations/sqlite/20260429000006_create_space_sync_state.sql b/migrations/sqlite/20260429000006_create_space_sync_state.sql new file mode 100644 --- /dev/null +++ b/migrations/sqlite/20260429000006_create_space_sync_state.sql @@ -0,0 +1,12 @@ +CREATE TABLE IF NOT EXISTS space_sync_state ( + id TEXT PRIMARY KEY, + space_id TEXT NOT NULL REFERENCES spaces(id) ON DELETE CASCADE, + member_did TEXT NOT NULL, + cursor TEXT, + last_synced_at TEXT, + status TEXT NOT NULL DEFAULT 'pending', + error TEXT, + UNIQUE(space_id, member_did) +); + +CREATE INDEX idx_space_sync_state_space_id ON space_sync_state(space_id); diff --git a/src/admin/permissions.rs b/src/admin/permissions.rs --- a/src/admin/permissions.rs +++ b/src/admin/permissions.rs @@ -2,7 +2,7 @@ use std::collections::HashSet; use serde::{Deserialize, Serialize}; -/// All 29 permissions in the system. +/// All 37 permissions in the system. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] pub enum Permission { #[serde(rename = "lexicons:create")] @@ -83,6 +83,23 @@ #[serde(rename = "dead-letters:read")] DeadLettersRead, #[serde(rename = "dead-letters:manage")] DeadLettersManage, + + #[serde(rename = "spaces:create")] + SpacesCreate, + #[serde(rename = "spaces:read")] + SpacesRead, + #[serde(rename = "spaces:update")] + SpacesUpdate, + #[serde(rename = "spaces:delete")] + SpacesDelete, + #[serde(rename = "spaces:manage-members")] + SpacesManageMembers, + #[serde(rename = "spaces:manage-invites")] + SpacesManageInvites, + #[serde(rename = "spaces:manage-records")] + SpacesManageRecords, + #[serde(rename = "spaces:manage-credentials")] + SpacesManageCredentials, } impl Permission { @@ -122,6 +139,14 @@ Self::ApiClientsEdit => "api-clients:edit", Self::ApiClientsDelete => "api-clients:delete", Self::DeadLettersRead => "dead-letters:read", Self::DeadLettersManage => "dead-letters:manage", + Self::SpacesCreate => "spaces:create", + Self::SpacesRead => "spaces:read", + Self::SpacesUpdate => "spaces:update", + Self::SpacesDelete => "spaces:delete", + Self::SpacesManageMembers => "spaces:manage-members", + Self::SpacesManageInvites => "spaces:manage-invites", + Self::SpacesManageRecords => "spaces:manage-records", + Self::SpacesManageCredentials => "spaces:manage-credentials", } } @@ -161,6 +186,14 @@ Self::ApiClientsEdit, Self::ApiClientsDelete, Self::DeadLettersRead, Self::DeadLettersManage, + Self::SpacesCreate, + Self::SpacesRead, + Self::SpacesUpdate, + Self::SpacesDelete, + Self::SpacesManageMembers, + Self::SpacesManageInvites, + Self::SpacesManageRecords, + Self::SpacesManageCredentials, ]) } } @@ -215,6 +248,14 @@ perms.insert(Permission::ApiClientsView); perms.insert(Permission::ApiClientsCreate); perms.insert(Permission::ApiClientsEdit); perms.insert(Permission::ApiClientsDelete); + perms.insert(Permission::SpacesCreate); + perms.insert(Permission::SpacesRead); + perms.insert(Permission::SpacesUpdate); + perms.insert(Permission::SpacesDelete); + perms.insert(Permission::SpacesManageMembers); + perms.insert(Permission::SpacesManageInvites); + perms.insert(Permission::SpacesManageRecords); + perms.insert(Permission::SpacesManageCredentials); perms } Self::FullAccess => Permission::all(), diff --git a/src/lexicon.rs b/src/lexicon.rs --- a/src/lexicon.rs +++ b/src/lexicon.rs @@ -85,6 +85,8 @@ /// Optional Lua script that runs when a record in this collection is indexed. pub index_hook: Option, /// Optional per-NSID token cost for rate limiting. pub token_cost: Option, + /// Optional space type NSID indicating this lexicon is designed for use within spaces of that type. + pub space_type: Option, } impl ParsedLexicon { @@ -127,6 +129,11 @@ let input = main_def.and_then(|m| m.get("input")).cloned(); let output = main_def.and_then(|m| m.get("output")).cloned(); let record_schema = main_def.and_then(|m| m.get("record")).cloned(); + let space_type = raw + .get("spaceType") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + Ok(Self { id, lexicon_type, @@ -142,6 +149,7 @@ action, script, index_hook, token_cost, + space_type, }) } } @@ -792,5 +800,44 @@ async fn registry_get_index_hook_returns_none_for_unknown() { let reg = LexiconRegistry::new(); let script = reg.get_index_hook("nonexistent").await; assert!(script.is_none()); + } + + #[test] + fn parse_space_type_from_lexicon() { + let raw = json!({ + "lexicon": 1, + "id": "com.example.forum.post", + "spaceType": "com.example.forum", + "defs": { + "main": { + "type": "record", + "key": "tid", + "record": { + "type": "object", + "properties": { + "text": { "type": "string" } + } + } + } + } + }); + let parsed = + ParsedLexicon::parse(raw, 1, None, ProcedureAction::Upsert, None, None, None).unwrap(); + assert_eq!(parsed.space_type.as_deref(), Some("com.example.forum")); + } + + #[test] + fn parse_space_type_none_by_default() { + let parsed = ParsedLexicon::parse( + record_lexicon_json(), + 1, + None, + ProcedureAction::Upsert, + None, + None, + None, + ) + .unwrap(); + assert!(parsed.space_type.is_none()); } } diff --git a/src/lib.rs b/src/lib.rs --- a/src/lib.rs +++ b/src/lib.rs @@ -23,6 +23,7 @@ pub mod record_refs; pub mod repo; pub mod resolve; pub mod server; +pub mod spaces; pub mod xrpc; use auth::oauth_store::{DbSessionStore, DbStateStore}; diff --git a/src/lua/atproto_api.rs b/src/lua/atproto_api.rs --- a/src/lua/atproto_api.rs +++ b/src/lua/atproto_api.rs @@ -257,6 +257,181 @@ )?; atproto_table.set("verify_signature", verify_fn)?; } + // atproto.spaces sub-table + let spaces_table = lua.create_table()?; + + // atproto.spaces.is_member(space_uri, did) -> boolean + let state_clone = state.clone(); + let is_member_fn = + lua.create_async_function(move |_lua, (space_uri, did): (String, String)| { + let state = state_clone.clone(); + async move { + let uri = crate::spaces::SpaceUri::parse(&space_uri) + .map_err(|e| mlua::Error::runtime(format!("invalid space URI: {e}")))?; + let space = crate::spaces::db::get_space_by_address( + &state.db, + state.db_backend, + &uri.owner_did, + &uri.type_nsid, + &uri.skey, + ) + .await + .map_err(|e| mlua::Error::runtime(format!("space lookup failed: {e}")))?; + let space = match space { + Some(s) => s, + None => return Ok(false), + }; + let access = + crate::spaces::members::is_member(&state.db, state.db_backend, &space.id, &did) + .await + .map_err(|e| { + mlua::Error::runtime(format!("membership check failed: {e}")) + })?; + Ok(access.is_some()) + } + })?; + spaces_table.set("is_member", is_member_fn)?; + + // atproto.spaces.get_access(space_uri, did) -> 'read' | 'write' | nil + let state_clone = state.clone(); + let get_access_fn = + lua.create_async_function(move |_lua, (space_uri, did): (String, String)| { + let state = state_clone.clone(); + async move { + let uri = crate::spaces::SpaceUri::parse(&space_uri) + .map_err(|e| mlua::Error::runtime(format!("invalid space URI: {e}")))?; + let space = crate::spaces::db::get_space_by_address( + &state.db, + state.db_backend, + &uri.owner_did, + &uri.type_nsid, + &uri.skey, + ) + .await + .map_err(|e| mlua::Error::runtime(format!("space lookup failed: {e}")))?; + let space = match space { + Some(s) => s, + None => return Ok(None), + }; + let access = + crate::spaces::members::is_member(&state.db, state.db_backend, &space.id, &did) + .await + .map_err(|e| { + mlua::Error::runtime(format!("membership check failed: {e}")) + })?; + Ok(access.map(|a| a.as_str().to_string())) + } + })?; + spaces_table.set("get_access", get_access_fn)?; + + // atproto.spaces.list_members(space_uri) -> array of { did, access } + let state_clone = state.clone(); + let list_members_fn = lua.create_async_function(move |lua, space_uri: String| { + let state = state_clone.clone(); + async move { + let uri = crate::spaces::SpaceUri::parse(&space_uri) + .map_err(|e| mlua::Error::runtime(format!("invalid space URI: {e}")))?; + let space = crate::spaces::db::get_space_by_address( + &state.db, + state.db_backend, + &uri.owner_did, + &uri.type_nsid, + &uri.skey, + ) + .await + .map_err(|e| mlua::Error::runtime(format!("space lookup failed: {e}")))?; + let space = match space { + Some(s) => s, + None => { + return Err(mlua::Error::runtime("space not found")); + } + }; + let members = + crate::spaces::members::resolve_members(&state.db, state.db_backend, &space.id) + .await + .map_err(|e| mlua::Error::runtime(format!("member resolution failed: {e}")))?; + + let result = lua.create_table()?; + for (i, member) in members.iter().enumerate() { + let entry = lua.create_table()?; + entry.set("did", member.did.as_str())?; + entry.set("access", member.access.as_str())?; + result.set(i + 1, entry)?; + } + Ok(mlua::Value::Table(result)) + } + })?; + spaces_table.set("list_members", list_members_fn)?; + + // atproto.spaces.query({ space_uri, collection, limit, cursor }) -> { records, cursor } + let state_clone = state.clone(); + let query_fn = lua.create_async_function(move |lua, opts: mlua::Table| { + let state = state_clone.clone(); + async move { + let space_uri: String = opts + .get("space_uri") + .map_err(|_| mlua::Error::runtime("space_uri is required"))?; + let collection: Option = opts.get("collection").ok(); + let limit: i64 = opts.get("limit").unwrap_or(50); + let cursor: Option = opts.get("cursor").ok(); + + let uri = crate::spaces::SpaceUri::parse(&space_uri) + .map_err(|e| mlua::Error::runtime(format!("invalid space URI: {e}")))?; + let space = crate::spaces::db::get_space_by_address( + &state.db, + state.db_backend, + &uri.owner_did, + &uri.type_nsid, + &uri.skey, + ) + .await + .map_err(|e| mlua::Error::runtime(format!("space lookup failed: {e}")))?; + let space = match space { + Some(s) => s, + None => { + return Err(mlua::Error::runtime("space not found")); + } + }; + + let records = crate::spaces::db::list_space_records( + &state.db, + state.db_backend, + &space.id, + collection.as_deref(), + limit.min(100), + cursor.as_deref(), + ) + .await + .map_err(|e| mlua::Error::runtime(format!("record query failed: {e}")))?; + + let next_cursor = records.last().map(|r| r.indexed_at.clone()); + + let result = lua.create_table()?; + let records_table = lua.create_table()?; + for (i, record) in records.iter().enumerate() { + let entry = lua.to_value(&serde_json::json!({ + "uri": record.uri, + "collection": record.collection, + "rkey": record.rkey, + "record": record.record, + "cid": record.cid, + "authorDid": record.author_did, + }))?; + records_table.set(i + 1, entry)?; + } + result.set("records", records_table)?; + match next_cursor { + Some(c) => result.set("cursor", c)?, + None => result.set("cursor", mlua::Value::Nil)?, + } + + Ok(mlua::Value::Table(result)) + } + })?; + spaces_table.set("query", query_fn)?; + + atproto_table.set("spaces", spaces_table)?; + lua.globals().set("atproto", atproto_table)?; Ok(()) } @@ -516,5 +691,22 @@ return atproto.verify_signature(record, sig, "did:plc:caller") "#; let result: bool = lua.load(chunk).eval_async().await.unwrap(); assert!(!result); + } + + #[tokio::test] + async fn spaces_api_is_registered() { + let state = test_state_with_plc(""); + let lua = mlua::Lua::new(); + register_atproto_api(&lua, Arc::new(state), None).unwrap(); + + let chunk = r#" + return type(atproto.spaces) == "table" + and type(atproto.spaces.is_member) == "function" + and type(atproto.spaces.get_access) == "function" + and type(atproto.spaces.list_members) == "function" + and type(atproto.spaces.query) == "function" + "#; + let result: bool = lua.load(chunk).eval_async().await.unwrap(); + assert!(result); } } diff --git a/src/lua/context.rs b/src/lua/context.rs --- a/src/lua/context.rs +++ b/src/lua/context.rs @@ -2,6 +2,35 @@ use mlua::{Lua, LuaSerdeExt, Result as LuaResult}; use serde_json::Value; 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_id: String, + pub owner_did: String, + pub type_nsid: String, + pub skey: String, +} + +fn set_space_context(lua: &Lua, space: Option<&SpaceContext>) -> LuaResult<()> { + let globals = lua.globals(); + match space { + Some(ctx) => { + let table = lua.create_table()?; + table.set("space_uri", ctx.space_uri.as_str())?; + table.set("space_id", ctx.space_id.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())?; + globals.set("space", table)?; + } + None => { + globals.set("space", mlua::Value::Nil)?; + } + } + Ok(()) +} + /// Set global context variables for a procedure script. pub fn set_procedure_context( lua: &Lua, @@ -10,6 +39,7 @@ input: &Value, params: &HashMap, caller_did: &str, collection: &str, + space: Option<&SpaceContext>, ) -> LuaResult<()> { let globals = lua.globals(); globals.set("method", method.to_string())?; @@ -17,6 +47,7 @@ globals.set("input", lua.to_value(input)?)?; globals.set("params", lua.to_value(params)?)?; globals.set("caller_did", caller_did.to_string())?; globals.set("collection", collection.to_string())?; + set_space_context(lua, space)?; Ok(()) } @@ -27,6 +58,7 @@ method: &str, params: &HashMap, collection: &str, caller_did: Option<&str>, + space: Option<&SpaceContext>, ) -> LuaResult<()> { let globals = lua.globals(); globals.set("method", method.to_string())?; @@ -36,6 +68,7 @@ match caller_did { Some(did) => globals.set("caller_did", did.to_string())?, None => globals.set("caller_did", mlua::Value::Nil)?, } + set_space_context(lua, space)?; Ok(()) } @@ -117,6 +150,7 @@ &input, ¶ms, "did:plc:test", "com.example.thing", + None, ) .unwrap(); @@ -150,6 +184,7 @@ "com.example.listThings", ¶ms, "com.example.thing", Some("did:plc:test"), + None, ) .unwrap(); @@ -191,6 +226,58 @@ let globals = lua.globals(); let env: mlua::Table = globals.get("env").unwrap(); assert!(env.get::("anything").unwrap().is_nil()); + } + + #[test] + fn query_context_with_space() { + let lua = create_sandbox().unwrap(); + let params = HashMap::new(); + let space = SpaceContext { + space_uri: "ats://did:plc:owner/com.example.forum/main".into(), + space_id: "space-123".into(), + owner_did: "did:plc:owner".into(), + type_nsid: "com.example.forum".into(), + skey: "main".into(), + }; + set_query_context( + &lua, + "com.example.listPosts", + ¶ms, + "com.example.forum.post", + Some("did:plc:test"), + Some(&space), + ) + .unwrap(); + + let globals = lua.globals(); + let space_table: mlua::Table = globals.get("space").unwrap(); + assert_eq!( + space_table.get::("space_uri").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" + ); + } + + #[test] + fn query_context_without_space() { + let lua = create_sandbox().unwrap(); + let params = HashMap::new(); + set_query_context( + &lua, + "com.example.listThings", + ¶ms, + "com.example.thing", + None, + None, + ) + .unwrap(); + + let globals = lua.globals(); + assert!(globals.get::("space").unwrap().is_nil()); } #[test] diff --git a/src/lua/execute.rs b/src/lua/execute.rs --- a/src/lua/execute.rs +++ b/src/lua/execute.rs @@ -33,6 +33,7 @@ .collect() } /// Execute a Lua script for a procedure endpoint. +#[allow(clippy::too_many_arguments)] pub async fn execute_procedure_script( state: &AppState, method: &str, @@ -41,6 +42,7 @@ input: &Value, params: &std::collections::HashMap, lexicon: &ParsedLexicon, script: &str, + space_ctx: Option<&context::SpaceContext>, ) -> Result { let start = Instant::now(); let backend = state.db_backend; @@ -275,9 +277,15 @@ .await; return Err(AppError::Internal(error_message)); } - if let Err(e) = - context::set_procedure_context(&lua, method, input, params, claims.did(), collection) - { + if let Err(e) = context::set_procedure_context( + &lua, + method, + input, + params, + claims.did(), + collection, + space_ctx, + ) { let error_message = format!("failed to set context: {e}"); log_event( &state.db, @@ -503,6 +511,7 @@ params: &HashMap, lexicon: &ParsedLexicon, script: &str, claims: Option<&Claims>, + space_ctx: Option<&context::SpaceContext>, ) -> Result { let start = Instant::now(); let backend = state.db_backend; @@ -632,9 +641,14 @@ .await; return Err(AppError::Internal(error_message)); } - if let Err(e) = - context::set_query_context(&lua, method, params, collection, claims.map(|c| c.did())) - { + if let Err(e) = context::set_query_context( + &lua, + method, + params, + collection, + claims.map(|c| c.did()), + space_ctx, + ) { let error_message = format!("failed to set context: {e}"); log_event( &state.db, diff --git a/src/lua/mod.rs b/src/lua/mod.rs --- a/src/lua/mod.rs +++ b/src/lua/mod.rs @@ -8,6 +8,8 @@ pub(crate) mod sandbox; mod tid; mod xrpc_api; +#[allow(unused_imports)] +pub(crate) use context::SpaceContext; pub(crate) use execute::{ HookEvent, execute_hook_script, execute_procedure_script, execute_query_script, run_hook_once, }; diff --git a/src/lua/xrpc_api.rs b/src/lua/xrpc_api.rs --- a/src/lua/xrpc_api.rs +++ b/src/lua/xrpc_api.rs @@ -310,6 +310,7 @@ action: ProcedureAction::Create, script: script.map(|s| s.to_string()), index_hook: None, token_cost: None, + space_type: None, } } @@ -329,6 +330,7 @@ action: ProcedureAction::Create, script: script.map(|s| s.to_string()), index_hook: None, token_cost: None, + space_type: None, } } diff --git a/src/profile.rs b/src/profile.rs --- a/src/profile.rs +++ b/src/profile.rs @@ -27,18 +27,30 @@ } #[derive(Deserialize)] #[serde(rename_all = "camelCase")] -struct DidDocument { +pub struct DidDocument { #[serde(default)] - also_known_as: Vec, + pub also_known_as: Vec, #[serde(default)] - service: Vec, + pub verification_method: Vec, + #[serde(default)] + pub service: Vec, } #[derive(Deserialize)] #[serde(rename_all = "camelCase")] -struct DidService { - id: String, - service_endpoint: String, +pub struct DidVerificationMethod { + pub id: String, + #[serde(rename = "type")] + pub method_type: String, + #[serde(default)] + pub public_key_multibase: Option, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct DidService { + pub id: String, + pub service_endpoint: String, } #[derive(Deserialize)] @@ -117,7 +129,7 @@ .ok_or_else(|| AppError::NotFound("no labeler or PDS endpoint in DID document".into())) } /// Fetch a DID document from the PLC directory or via `did:web` resolution. -async fn resolve_did_document( +pub async fn resolve_did_document( http: &reqwest::Client, plc_url: &str, did: &str, diff --git a/src/server.rs b/src/server.rs --- a/src/server.rs +++ b/src/server.rs @@ -62,6 +62,7 @@ let serve_dir = ServeDir::new(&static_dir).not_found_service(spa_fallback); let domain_routes = Router::new() + .merge(crate::spaces::routes::space_routes()) .nest("/auth", crate::auth::routes::routes()) .nest("/external-auth", crate::external_auth::routes()) .nest("/oauth", crate::oauth::routes::routes()) diff --git a/src/spaces/auth.rs b/src/spaces/auth.rs new file mode 100644 --- /dev/null +++ b/src/spaces/auth.rs @@ -0,0 +1,315 @@ +use base64::Engine; +use base64::engine::general_purpose::URL_SAFE_NO_PAD; +use p256::ecdsa::SigningKey; +use rand::RngCore; +use sha2::{Digest, Sha256}; +use uuid::Uuid; + +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, +}; +use crate::spaces::types::{AccessMode, Space}; + +pub struct IssuedCredential { + pub token: String, + pub expires_at: String, +} + +pub async fn issue_credential( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + encryption_key: &[u8; 32], + space: &Space, + subject_did: &str, + client_id: Option<&str>, +) -> Result { + check_app_access(space, client_id)?; + + let private_jwk = get_or_create_signing_key(pool, backend, encryption_key, space).await?; + + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + let exp = now + DEFAULT_CREDENTIAL_TTL_SECS; + + let claims = SpaceCredentialClaims { + iss: space.owner_did.clone(), + sub: subject_did.to_string(), + space: format!("{}/{}/{}", space.owner_did, space.type_nsid, space.skey), + scope: "read".into(), + iat: now, + exp, + }; + + let token = sign_credential(&claims, &private_jwk)?; + + let token_hash = hex::encode(Sha256::digest(token.as_bytes())); + store_credential_record(pool, backend, &space.id, subject_did, &token_hash, exp).await?; + + let expires_at = chrono::DateTime::from_timestamp(exp as i64, 0) + .map(|dt| dt.to_rfc3339()) + .unwrap_or_default(); + + 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(()); + }; + + match space.access_mode { + AccessMode::DefaultDeny => { + if let Some(ref allowlist) = space.app_allowlist { + if !allowlist.iter().any(|id| id == client_id) { + return Err(AppError::Forbidden( + "This app is not authorized to access this space".into(), + )); + } + } else { + return Err(AppError::Forbidden( + "Space is in default_deny mode with no allowlist".into(), + )); + } + } + AccessMode::DefaultAllow => { + if let Some(ref denylist) = space.app_denylist + && denylist.iter().any(|id| id == client_id) + { + return Err(AppError::Forbidden( + "This app has been denied access to this space".into(), + )); + } + } + } + + Ok(()) +} + +async fn get_or_create_signing_key( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + encryption_key: &[u8; 32], + space: &Space, +) -> Result { + let sql = adapt_sql( + "SELECT signing_key_enc FROM space_dids WHERE space_id = ?", + backend, + ); + let row: Option<(Vec,)> = sqlx::query_as(&sql) + .bind(&space.id) + .fetch_optional(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to look up space signing key: {e}")))?; + + if let Some((encrypted,)) = row { + let decrypted = decrypt(encryption_key, &encrypted) + .map_err(|e| AppError::Internal(format!("failed to decrypt signing key: {e}")))?; + let jwk: serde_json::Value = serde_json::from_slice(&decrypted) + .map_err(|e| AppError::Internal(format!("failed to parse signing key: {e}")))?; + return Ok(jwk); + } + + let keypair = generate_space_keypair()?; + let key_bytes = serde_json::to_vec(&keypair.private_jwk) + .map_err(|e| AppError::Internal(format!("failed to serialize signing key: {e}")))?; + let encrypted_signing = encrypt(encryption_key, &key_bytes) + .map_err(|e| AppError::Internal(format!("failed to encrypt signing key: {e}")))?; + + // Rotation key is a separate keypair for recovery + let rotation_keypair = generate_space_keypair()?; + let rotation_bytes = serde_json::to_vec(&rotation_keypair.private_jwk) + .map_err(|e| AppError::Internal(format!("failed to serialize rotation key: {e}")))?; + let encrypted_rotation = encrypt(encryption_key, &rotation_bytes) + .map_err(|e| AppError::Internal(format!("failed to encrypt rotation key: {e}")))?; + + let now = now_rfc3339(); + let insert_sql = adapt_sql( + "INSERT INTO space_dids (id, did, space_id, signing_key_enc, rotation_key_enc, created_by, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)", + backend, + ); + + sqlx::query(&insert_sql) + .bind(Uuid::new_v4().to_string()) + .bind(&space.owner_did) + .bind(&space.id) + .bind(&encrypted_signing) + .bind(&encrypted_rotation) + .bind(&space.owner_did) + .bind(&now) + .execute(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to store space signing key: {e}")))?; + + 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, +} + +fn generate_space_keypair() -> Result { + let mut rng_bytes = [0u8; 32]; + rand::rng().fill_bytes(&mut rng_bytes); + + let signing_key = SigningKey::from_bytes((&rng_bytes[..]).into()) + .map_err(|e| AppError::Internal(format!("failed to generate signing key: {e}")))?; + + let verifying_key = signing_key.verifying_key(); + let public_point = verifying_key.to_encoded_point(false); + + let x_bytes = public_point + .x() + .ok_or_else(|| AppError::Internal("missing x coordinate".into()))?; + let y_bytes = public_point + .y() + .ok_or_else(|| AppError::Internal("missing y coordinate".into()))?; + + let x_b64 = URL_SAFE_NO_PAD.encode(x_bytes); + let y_b64 = URL_SAFE_NO_PAD.encode(y_bytes); + let d_b64 = URL_SAFE_NO_PAD.encode(rng_bytes); + + let private_jwk = serde_json::json!({ + "kty": "EC", + "crv": "P-256", + "x": x_b64, + "y": y_b64, + "d": d_b64, + }); + + Ok(SpaceKeypair { private_jwk }) +} + +async fn store_credential_record( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + space_id: &str, + issued_to: &str, + token_hash: &str, + expires_at_epoch: u64, +) -> Result<(), AppError> { + let now = now_rfc3339(); + let expires_at = chrono::DateTime::from_timestamp(expires_at_epoch as i64, 0) + .map(|dt| dt.to_rfc3339()) + .unwrap_or_default(); + + let sql = adapt_sql( + "INSERT INTO space_credentials (id, space_id, issued_to, token_hash, expires_at, created_at) VALUES (?, ?, ?, ?, ?, ?)", + backend, + ); + + sqlx::query(&sql) + .bind(Uuid::new_v4().to_string()) + .bind(space_id) + .bind(issued_to) + .bind(token_hash) + .bind(&expires_at) + .bind(&now) + .execute(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to store credential record: {e}")))?; + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::spaces::types::{AccessMode, Space, SpaceConfig}; + + fn test_space(access_mode: AccessMode) -> Space { + Space { + id: "test-space".into(), + owner_did: "did:plc:owner".into(), + type_nsid: "com.example.forum".into(), + skey: "main".into(), + display_name: None, + description: None, + access_mode, + app_allowlist: None, + app_denylist: None, + managing_app_did: None, + config: SpaceConfig::default(), + created_at: String::new(), + updated_at: String::new(), + } + } + + #[test] + fn app_access_default_allow_no_lists() { + let space = test_space(AccessMode::DefaultAllow); + assert!(check_app_access(&space, Some("any-app")).is_ok()); + } + + #[test] + fn app_access_default_allow_denied() { + let mut space = test_space(AccessMode::DefaultAllow); + space.app_denylist = Some(vec!["bad-app".into()]); + + assert!(check_app_access(&space, Some("good-app")).is_ok()); + assert!(check_app_access(&space, Some("bad-app")).is_err()); + } + + #[test] + fn app_access_default_deny_no_allowlist() { + let space = test_space(AccessMode::DefaultDeny); + assert!(check_app_access(&space, Some("any-app")).is_err()); + } + + #[test] + fn app_access_default_deny_allowed() { + let mut space = test_space(AccessMode::DefaultDeny); + space.app_allowlist = Some(vec!["good-app".into()]); + + assert!(check_app_access(&space, Some("good-app")).is_ok()); + assert!(check_app_access(&space, Some("other-app")).is_err()); + } + + #[test] + fn app_access_no_client_id_always_passes() { + let space = test_space(AccessMode::DefaultDeny); + assert!(check_app_access(&space, None).is_ok()); + } + + #[test] + fn generate_keypair_produces_valid_jwk() { + let kp = generate_space_keypair().unwrap(); + assert_eq!(kp.private_jwk["kty"], "EC"); + assert_eq!(kp.private_jwk["crv"], "P-256"); + assert!(kp.private_jwk["d"].is_string()); + assert!(kp.private_jwk["x"].is_string()); + assert!(kp.private_jwk["y"].is_string()); + } +} diff --git a/src/spaces/credential.rs b/src/spaces/credential.rs new file mode 100644 --- /dev/null +++ b/src/spaces/credential.rs @@ -0,0 +1,284 @@ +use base64::Engine; +use base64::engine::general_purpose::URL_SAFE_NO_PAD; +use p256::ecdsa::{Signature, SigningKey, VerifyingKey, signature::Signer, signature::Verifier}; +use serde::{Deserialize, Serialize}; + +use crate::error::AppError; +use crate::profile; + +pub const DEFAULT_CREDENTIAL_TTL_SECS: u64 = 4 * 60 * 60; // 4 hours + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SpaceCredentialClaims { + pub iss: String, + pub sub: String, + pub space: String, + pub scope: String, + pub iat: u64, + pub exp: u64, +} + +pub fn sign_credential( + claims: &SpaceCredentialClaims, + private_jwk: &serde_json::Value, +) -> Result { + let d_b64 = private_jwk["d"] + .as_str() + .ok_or_else(|| AppError::Internal("signing key missing d parameter".into()))?; + + let d_bytes = URL_SAFE_NO_PAD + .decode(d_b64) + .map_err(|_| AppError::Internal("invalid signing key d parameter".into()))?; + + let signing_key = SigningKey::from_bytes((&d_bytes[..]).into()) + .map_err(|e| AppError::Internal(format!("invalid signing key: {e}")))?; + + let header = serde_json::json!({ + "alg": "ES256", + "typ": "JWT", + }); + + let header_b64 = URL_SAFE_NO_PAD.encode(serde_json::to_vec(&header).unwrap()); + let payload_b64 = URL_SAFE_NO_PAD.encode(serde_json::to_vec(claims).unwrap()); + + let message = format!("{}.{}", header_b64, payload_b64); + let signature: Signature = signing_key.sign(message.as_bytes()); + let sig_b64 = URL_SAFE_NO_PAD.encode(signature.to_bytes()); + + Ok(format!("{}.{}.{}", header_b64, payload_b64, sig_b64)) +} + +pub fn verify_credential( + token: &str, + public_jwk: &serde_json::Value, +) -> Result { + let parts: Vec<&str> = token.split('.').collect(); + if parts.len() != 3 { + return Err(AppError::Auth("invalid credential format".into())); + } + + let header_bytes = URL_SAFE_NO_PAD + .decode(parts[0]) + .map_err(|_| AppError::Auth("invalid credential header encoding".into()))?; + let header: serde_json::Value = serde_json::from_slice(&header_bytes) + .map_err(|_| AppError::Auth("invalid credential header".into()))?; + + if header["alg"].as_str() != Some("ES256") { + return Err(AppError::Auth("credential alg must be ES256".into())); + } + + let x_b64 = public_jwk["x"] + .as_str() + .ok_or_else(|| AppError::Auth("public key missing x".into()))?; + let y_b64 = public_jwk["y"] + .as_str() + .ok_or_else(|| AppError::Auth("public key missing y".into()))?; + + let x_bytes = URL_SAFE_NO_PAD + .decode(x_b64) + .map_err(|_| AppError::Auth("invalid public key x".into()))?; + let y_bytes = URL_SAFE_NO_PAD + .decode(y_b64) + .map_err(|_| AppError::Auth("invalid public key y".into()))?; + + let mut sec1 = Vec::with_capacity(1 + 32 + 32); + sec1.push(0x04); + sec1.extend_from_slice(&x_bytes); + sec1.extend_from_slice(&y_bytes); + + let verifying_key = VerifyingKey::from_sec1_bytes(&sec1) + .map_err(|_| AppError::Auth("invalid space credential public key".into()))?; + + let message = format!("{}.{}", parts[0], parts[1]); + let sig_bytes = URL_SAFE_NO_PAD + .decode(parts[2]) + .map_err(|_| AppError::Auth("invalid credential signature encoding".into()))?; + let signature = Signature::from_bytes(sig_bytes.as_slice().into()) + .map_err(|_| AppError::Auth("invalid credential signature format".into()))?; + + verifying_key + .verify(message.as_bytes(), &signature) + .map_err(|_| AppError::Auth("credential signature verification failed".into()))?; + + let payload_bytes = URL_SAFE_NO_PAD + .decode(parts[1]) + .map_err(|_| AppError::Auth("invalid credential payload encoding".into()))?; + let claims: SpaceCredentialClaims = serde_json::from_slice(&payload_bytes) + .map_err(|_| AppError::Auth("invalid credential payload".into()))?; + + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + + if now > claims.exp { + return Err(AppError::Auth("credential has expired".into())); + } + + Ok(claims) +} + +/// Convert a multibase-encoded P-256 public key (from a DID doc `publicKeyMultibase`) +/// into a JWK suitable for `verify_credential`. +pub fn multikey_to_p256_jwk(public_key_multibase: &str) -> Result { + let (_base, key_bytes) = multibase::decode(public_key_multibase) + .map_err(|e| AppError::Auth(format!("invalid multibase encoding: {e}")))?; + + // P-256 multicodec prefix: varint 0x1200 → bytes [0x80, 0x24] + if key_bytes.len() < 2 || key_bytes[0] != 0x80 || key_bytes[1] != 0x24 { + return Err(AppError::Auth( + "public key is not a P-256 multicodec key".into(), + )); + } + + let compressed = &key_bytes[2..]; + let verifying_key = VerifyingKey::from_sec1_bytes(compressed) + .map_err(|_| AppError::Auth("invalid P-256 public key bytes".into()))?; + + let point = verifying_key.to_encoded_point(false); + let x = point + .x() + .ok_or_else(|| AppError::Auth("failed to extract x coordinate".into()))?; + let y = point + .y() + .ok_or_else(|| AppError::Auth("failed to extract y coordinate".into()))?; + + Ok(serde_json::json!({ + "kty": "EC", + "crv": "P-256", + "x": URL_SAFE_NO_PAD.encode(x), + "y": URL_SAFE_NO_PAD.encode(y), + })) +} + +/// Verify a space credential JWT issued by an external space host. +/// +/// Resolves the issuer's DID document, extracts the `#atproto` signing key, +/// and verifies the JWT signature and expiry. +pub async fn verify_external_credential( + token: &str, + http: &reqwest::Client, + plc_url: &str, +) -> Result { + // Peek at the payload to extract the issuer DID without verifying yet + let parts: Vec<&str> = token.split('.').collect(); + if parts.len() != 3 { + return Err(AppError::Auth("invalid credential format".into())); + } + + let payload_bytes = URL_SAFE_NO_PAD + .decode(parts[1]) + .map_err(|_| AppError::Auth("invalid credential payload encoding".into()))?; + let peek: SpaceCredentialClaims = serde_json::from_slice(&payload_bytes) + .map_err(|_| AppError::Auth("invalid credential payload".into()))?; + + let did_doc = profile::resolve_did_document(http, plc_url, &peek.iss).await?; + + let vm = did_doc + .verification_method + .iter() + .find(|v| v.id.ends_with("#atproto")) + .ok_or_else(|| AppError::Auth("issuer DID has no #atproto verification method".into()))?; + + let multibase = vm + .public_key_multibase + .as_deref() + .ok_or_else(|| AppError::Auth("verification method missing publicKeyMultibase".into()))?; + + let jwk = multikey_to_p256_jwk(multibase)?; + verify_credential(token, &jwk) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::oauth::keys::generate_dpop_keypair; + + fn make_claims() -> SpaceCredentialClaims { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + SpaceCredentialClaims { + iss: "did:plc:spaceowner".into(), + sub: "did:plc:requester".into(), + space: "did:plc:spaceowner/com.example.forum/main".into(), + scope: "read".into(), + iat: now, + exp: now + DEFAULT_CREDENTIAL_TTL_SECS, + } + } + + #[test] + fn sign_and_verify_roundtrip() { + let keypair = generate_dpop_keypair().unwrap(); + let claims = make_claims(); + + let token = sign_credential(&claims, &keypair.private_jwk).unwrap(); + let verified = verify_credential(&token, &keypair.public_jwk).unwrap(); + + assert_eq!(verified.iss, claims.iss); + assert_eq!(verified.sub, claims.sub); + assert_eq!(verified.space, claims.space); + assert_eq!(verified.scope, claims.scope); + assert_eq!(verified.iat, claims.iat); + assert_eq!(verified.exp, claims.exp); + } + + #[test] + fn verify_rejects_tampered_payload() { + let keypair = generate_dpop_keypair().unwrap(); + let claims = make_claims(); + let token = sign_credential(&claims, &keypair.private_jwk).unwrap(); + + // Tamper with the payload + let parts: Vec<&str> = token.split('.').collect(); + let mut payload_bytes = URL_SAFE_NO_PAD.decode(parts[1]).unwrap(); + payload_bytes[0] ^= 0xFF; + let tampered_payload = URL_SAFE_NO_PAD.encode(&payload_bytes); + let tampered = format!("{}.{}.{}", parts[0], tampered_payload, parts[2]); + + let result = verify_credential(&tampered, &keypair.public_jwk); + assert!(result.is_err()); + } + + #[test] + fn verify_rejects_wrong_key() { + let keypair1 = generate_dpop_keypair().unwrap(); + let keypair2 = generate_dpop_keypair().unwrap(); + let claims = make_claims(); + let token = sign_credential(&claims, &keypair1.private_jwk).unwrap(); + + let result = verify_credential(&token, &keypair2.public_jwk); + assert!(result.is_err()); + } + + #[test] + fn verify_rejects_expired() { + let keypair = generate_dpop_keypair().unwrap(); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + let claims = SpaceCredentialClaims { + iss: "did:plc:owner".into(), + sub: "did:plc:user".into(), + space: "did:plc:owner/test/main".into(), + scope: "read".into(), + iat: now - 7200, + exp: now - 3600, // expired 1 hour ago + }; + + let token = sign_credential(&claims, &keypair.private_jwk).unwrap(); + let result = verify_credential(&token, &keypair.public_jwk); + assert!(result.is_err()); + assert!(result.unwrap_err().to_string().contains("expired")); + } + + #[test] + fn verify_rejects_invalid_format() { + let keypair = generate_dpop_keypair().unwrap(); + let result = verify_credential("not-a-jwt", &keypair.public_jwk); + assert!(result.is_err()); + } +} diff --git a/src/spaces/db.rs b/src/spaces/db.rs new file mode 100644 --- /dev/null +++ b/src/spaces/db.rs @@ -0,0 +1,780 @@ +use crate::db::{DatabaseBackend, adapt_sql, now_rfc3339}; +use crate::error::AppError; +use crate::spaces::types::*; + +// --------------------------------------------------------------------------- +// Spaces +// --------------------------------------------------------------------------- + +pub async fn create_space( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + space: &Space, +) -> Result<(), AppError> { + let now = now_rfc3339(); + let config_json = serde_json::to_string(&space.config) + .map_err(|e| AppError::Internal(format!("failed to serialize space config: {e}")))?; + let allowlist_json = space + .app_allowlist + .as_ref() + .map(|v| serde_json::to_string(v).unwrap_or_default()); + let denylist_json = space + .app_denylist + .as_ref() + .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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + backend, + ); + + sqlx::query(&sql) + .bind(&space.id) + .bind(&space.owner_did) + .bind(&space.type_nsid) + .bind(&space.skey) + .bind(&space.display_name) + .bind(&space.description) + .bind(space.access_mode.as_str()) + .bind(&allowlist_json) + .bind(&denylist_json) + .bind(&space.managing_app_did) + .bind(&config_json) + .bind(&now) + .bind(&now) + .execute(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to create space: {e}")))?; + + Ok(()) +} + +pub async fn get_space( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + 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 = ?", + backend, + ); + + let row: Option = sqlx::query_as(&sql) + .bind(id) + .fetch_optional(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to get space: {e}")))?; + + row.map(parse_space_row).transpose() +} + +pub async fn get_space_by_address( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + owner_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 = ?", + backend, + ); + + let row: Option = sqlx::query_as(&sql) + .bind(owner_did) + .bind(type_nsid) + .bind(skey) + .fetch_optional(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to get space: {e}")))?; + + row.map(parse_space_row).transpose() +} + +pub async fn list_spaces_by_owner( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + owner_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", + backend, + ); + + let rows: Vec = sqlx::query_as(&sql) + .bind(owner_did) + .fetch_all(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to list spaces: {e}")))?; + + rows.into_iter().map(parse_space_row).collect() +} + +pub async fn update_space( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + space: &Space, +) -> Result { + let now = now_rfc3339(); + let config_json = serde_json::to_string(&space.config) + .map_err(|e| AppError::Internal(format!("failed to serialize space config: {e}")))?; + let allowlist_json = space + .app_allowlist + .as_ref() + .map(|v| serde_json::to_string(v).unwrap_or_default()); + let denylist_json = space + .app_denylist + .as_ref() + .map(|v| serde_json::to_string(v).unwrap_or_default()); + + let sql = adapt_sql( + "UPDATE spaces SET display_name = ?, description = ?, access_mode = ?, app_allowlist = ?, app_denylist = ?, managing_app_did = ?, config = ?, updated_at = ? WHERE id = ?", + backend, + ); + + let result = sqlx::query(&sql) + .bind(&space.display_name) + .bind(&space.description) + .bind(space.access_mode.as_str()) + .bind(&allowlist_json) + .bind(&denylist_json) + .bind(&space.managing_app_did) + .bind(&config_json) + .bind(&now) + .bind(&space.id) + .execute(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to update space: {e}")))?; + + Ok(result.rows_affected() > 0) +} + +pub async fn delete_space( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + id: &str, +) -> Result { + let sql = adapt_sql("DELETE FROM spaces WHERE id = ?", backend); + + let result = sqlx::query(&sql) + .bind(id) + .execute(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to delete space: {e}")))?; + + Ok(result.rows_affected() > 0) +} + +type SpaceRow = ( + String, + String, + String, + String, + Option, + Option, + String, + Option, + Option, + Option, + String, + 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 app_allowlist: Option> = + r.7.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() + .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) + .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, + access_mode, + app_allowlist, + app_denylist, + managing_app_did: r.9, + config, + created_at: r.11, + updated_at: r.12, + }) +} + +// --------------------------------------------------------------------------- +// Space Members +// --------------------------------------------------------------------------- + +pub async fn add_member( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + member: &SpaceMember, +) -> Result<(), AppError> { + let now = now_rfc3339(); + let sql = adapt_sql( + "INSERT INTO space_members (id, space_id, member_did, access, is_delegation, granted_by, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)", + backend, + ); + + sqlx::query(&sql) + .bind(&member.id) + .bind(&member.space_id) + .bind(&member.member_did) + .bind(member.access.as_str()) + .bind(member.is_delegation as i32) + .bind(&member.granted_by) + .bind(&now) + .execute(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to add member: {e}")))?; + + Ok(()) +} + +pub async fn remove_member( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + space_id: &str, + member_did: &str, +) -> Result { + let sql = adapt_sql( + "DELETE FROM space_members WHERE space_id = ? AND member_did = ?", + backend, + ); + + let result = sqlx::query(&sql) + .bind(space_id) + .bind(member_did) + .execute(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to remove member: {e}")))?; + + Ok(result.rows_affected() > 0) +} + +pub async fn get_member( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + space_id: &str, + member_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 = ?", + backend, + ); + + let row: Option = sqlx::query_as(&sql) + .bind(space_id) + .bind(member_did) + .fetch_optional(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to get member: {e}")))?; + + row.map(parse_member_row).transpose() +} + +pub async fn list_direct_members( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + space_id: &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 = ? ORDER BY created_at ASC", + backend, + ); + + let rows: Vec = sqlx::query_as(&sql) + .bind(space_id) + .fetch_all(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to list members: {e}")))?; + + rows.into_iter().map(parse_member_row).collect() +} + +pub async fn list_spaces_for_member( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + member_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", + backend, + ); + + let rows: Vec = sqlx::query_as(&sql) + .bind(member_did) + .fetch_all(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to list spaces for member: {e}")))?; + + rows.into_iter().map(parse_member_row).collect() +} + +type MemberRow = (String, String, String, String, i32, Option, String); + +fn parse_member_row(r: MemberRow) -> Result { + let access = SpaceAccess::parse(&r.3) + .ok_or_else(|| AppError::Internal(format!("invalid access: {}", r.3)))?; + + Ok(SpaceMember { + id: r.0, + space_id: r.1, + member_did: r.2, + access, + is_delegation: r.4 != 0, + granted_by: r.5, + created_at: r.6, + }) +} + +// --------------------------------------------------------------------------- +// Space Records +// --------------------------------------------------------------------------- + +pub async fn upsert_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 = match backend { + DatabaseBackend::Sqlite => { + "INSERT OR REPLACE INTO space_records (uri, space_id, author_did, collection, rkey, record, cid, indexed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)".to_string() + } + DatabaseBackend::Postgres => adapt_sql( + "INSERT INTO space_records (uri, space_id, author_did, collection, rkey, record, cid, indexed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (uri) DO UPDATE SET record = EXCLUDED.record, cid = EXCLUDED.cid, indexed_at = EXCLUDED.indexed_at", + 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| AppError::Internal(format!("failed to upsert space record: {e}")))?; + + Ok(()) +} + +pub async fn get_space_record( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + uri: &str, +) -> Result, AppError> { + let sql = adapt_sql( + "SELECT uri, space_id, author_did, collection, rkey, record, cid, indexed_at FROM space_records WHERE uri = ?", + backend, + ); + + let row: Option = sqlx::query_as(&sql) + .bind(uri) + .fetch_optional(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to get space record: {e}")))?; + + row.map(parse_record_row).transpose() +} + +pub async fn get_space_record_by_parts( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + space_id: &str, + collection: &str, + rkey: &str, +) -> Result, AppError> { + let sql = adapt_sql( + "SELECT uri, space_id, author_did, collection, rkey, record, cid, indexed_at FROM space_records WHERE space_id = ? AND collection = ? AND rkey = ? LIMIT 1", + backend, + ); + + let row: Option = sqlx::query_as(&sql) + .bind(space_id) + .bind(collection) + .bind(rkey) + .fetch_optional(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to get space record: {e}")))?; + + row.map(parse_record_row).transpose() +} + +pub async fn list_space_records( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + space_id: &str, + collection: Option<&str>, + limit: i64, + cursor: Option<&str>, +) -> 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 query = sqlx::query_as::<_, RecordRow>(&sql).bind(space_id); + + if has_collection { + query = query.bind(collection.unwrap()); + } + 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 space records: {e}")))?; + + rows.into_iter().map(parse_record_row).collect() +} + +pub async fn delete_space_record( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + uri: &str, +) -> Result { + let sql = adapt_sql("DELETE FROM space_records WHERE uri = ?", backend); + + let result = sqlx::query(&sql) + .bind(uri) + .execute(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to delete space record: {e}")))?; + + Ok(result.rows_affected() > 0) +} + +type RecordRow = ( + String, + String, + String, + String, + String, + String, + String, + String, +); + +fn parse_record_row(r: RecordRow) -> Result { + let record: serde_json::Value = serde_json::from_str(&r.5) + .map_err(|e| AppError::Internal(format!("invalid record JSON: {e}")))?; + + Ok(SpaceRecord { + uri: r.0, + space_id: r.1, + author_did: r.2, + collection: r.3, + rkey: r.4, + record, + cid: r.6, + indexed_at: r.7, + }) +} + +// --------------------------------------------------------------------------- +// Space Invites +// --------------------------------------------------------------------------- + +pub async fn create_invite( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + invite: &SpaceInvite, +) -> Result<(), AppError> { + let now = now_rfc3339(); + let sql = adapt_sql( + "INSERT INTO space_invites (id, space_id, token_hash, created_by, access, max_uses, uses, expires_at, revoked, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + backend, + ); + + sqlx::query(&sql) + .bind(&invite.id) + .bind(&invite.space_id) + .bind(&invite.token_hash) + .bind(&invite.created_by) + .bind(invite.access.as_str()) + .bind(invite.max_uses) + .bind(invite.uses) + .bind(&invite.expires_at) + .bind(invite.revoked as i32) + .bind(&now) + .execute(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to create invite: {e}")))?; + + Ok(()) +} + +pub async fn get_invite_by_token_hash( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + token_hash: &str, +) -> Result, AppError> { + let sql = adapt_sql( + "SELECT id, space_id, token_hash, created_by, access, max_uses, uses, expires_at, revoked, created_at FROM space_invites WHERE token_hash = ?", + backend, + ); + + let row: Option = sqlx::query_as(&sql) + .bind(token_hash) + .fetch_optional(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to get invite: {e}")))?; + + row.map(parse_invite_row).transpose() +} + +pub async fn increment_invite_uses( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + invite_id: &str, +) -> Result<(), AppError> { + let sql = adapt_sql( + "UPDATE space_invites SET uses = uses + 1 WHERE id = ?", + backend, + ); + + sqlx::query(&sql) + .bind(invite_id) + .execute(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to increment invite uses: {e}")))?; + + Ok(()) +} + +pub async fn revoke_invite( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + invite_id: &str, +) -> Result { + let sql = adapt_sql("UPDATE space_invites SET revoked = 1 WHERE id = ?", backend); + + let result = sqlx::query(&sql) + .bind(invite_id) + .execute(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to revoke invite: {e}")))?; + + Ok(result.rows_affected() > 0) +} + +pub async fn list_invites( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + space_id: &str, +) -> Result, AppError> { + let sql = adapt_sql( + "SELECT id, space_id, token_hash, created_by, access, max_uses, uses, expires_at, revoked, created_at FROM space_invites WHERE space_id = ? ORDER BY created_at DESC", + backend, + ); + + let rows: Vec = sqlx::query_as(&sql) + .bind(space_id) + .fetch_all(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to list invites: {e}")))?; + + rows.into_iter().map(parse_invite_row).collect() +} + +type InviteRow = ( + String, + String, + String, + String, + String, + Option, + i64, + Option, + i32, + String, +); + +fn parse_invite_row(r: InviteRow) -> Result { + let access = SpaceAccess::parse(&r.4) + .ok_or_else(|| AppError::Internal(format!("invalid invite access: {}", r.4)))?; + + Ok(SpaceInvite { + id: r.0, + space_id: r.1, + token_hash: r.2, + created_by: r.3, + access, + max_uses: r.5, + uses: r.6, + expires_at: r.7, + revoked: r.8 != 0, + created_at: r.9, + }) +} + +// --------------------------------------------------------------------------- +// Space Sync State +// --------------------------------------------------------------------------- + +pub async fn get_sync_state( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + space_id: &str, + member_did: &str, +) -> Result, AppError> { + let sql = adapt_sql( + "SELECT id, space_id, member_did, cursor, last_synced_at, status, error FROM space_sync_state WHERE space_id = ? AND member_did = ?", + backend, + ); + + let row: Option = sqlx::query_as(&sql) + .bind(space_id) + .bind(member_did) + .fetch_optional(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to get sync state: {e}")))?; + + row.map(parse_sync_state_row).transpose() +} + +pub async fn upsert_sync_state( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + state: &SpaceSyncState, +) -> Result<(), AppError> { + let sql = match backend { + DatabaseBackend::Sqlite => { + "INSERT OR REPLACE INTO space_sync_state (id, space_id, member_did, cursor, last_synced_at, status, error) VALUES (?, ?, ?, ?, ?, ?, ?)".to_string() + } + DatabaseBackend::Postgres => adapt_sql( + "INSERT INTO space_sync_state (id, space_id, member_did, cursor, last_synced_at, status, error) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT (space_id, member_did) DO UPDATE SET cursor = EXCLUDED.cursor, last_synced_at = EXCLUDED.last_synced_at, status = EXCLUDED.status, error = EXCLUDED.error", + backend, + ), + }; + + sqlx::query(&sql) + .bind(&state.id) + .bind(&state.space_id) + .bind(&state.member_did) + .bind(&state.cursor) + .bind(&state.last_synced_at) + .bind(state.status.as_str()) + .bind(&state.error) + .execute(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to upsert sync state: {e}")))?; + + Ok(()) +} + +pub async fn list_sync_states_for_space( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + space_id: &str, +) -> Result, AppError> { + let sql = adapt_sql( + "SELECT id, space_id, member_did, cursor, last_synced_at, status, error FROM space_sync_state WHERE space_id = ? ORDER BY member_did ASC", + backend, + ); + + let rows: Vec = sqlx::query_as(&sql) + .bind(space_id) + .fetch_all(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to list sync states: {e}")))?; + + rows.into_iter().map(parse_sync_state_row).collect() +} + +pub async fn list_pending_syncs( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, +) -> Result, AppError> { + let sql = adapt_sql( + "SELECT id, space_id, member_did, cursor, last_synced_at, status, error FROM space_sync_state WHERE status = 'pending' OR status = 'error' ORDER BY last_synced_at ASC NULLS FIRST LIMIT 50", + backend, + ); + + let rows: Vec = sqlx::query_as(&sql) + .fetch_all(pool) + .await + .map_err(|e| AppError::Internal(format!("failed to list pending syncs: {e}")))?; + + rows.into_iter().map(parse_sync_state_row).collect() +} + +type SyncStateRow = ( + String, + String, + String, + Option, + Option, + String, + Option, +); + +fn parse_sync_state_row(r: SyncStateRow) -> Result { + let status = SyncStatus::parse(&r.5) + .ok_or_else(|| AppError::Internal(format!("invalid sync status: {}", r.5)))?; + + Ok(SpaceSyncState { + id: r.0, + space_id: r.1, + member_did: r.2, + cursor: r.3, + last_synced_at: r.4, + status, + error: r.6, + }) +} diff --git a/src/spaces/members.rs b/src/spaces/members.rs new file mode 100644 --- /dev/null +++ b/src/spaces/members.rs @@ -0,0 +1,142 @@ +use std::collections::{HashMap, HashSet}; + +use crate::db::DatabaseBackend; +use crate::error::AppError; +use crate::spaces::SpaceUri; +use crate::spaces::db; +use crate::spaces::types::{ResolvedMember, SpaceAccess, SpaceMember}; + +const MAX_DELEGATION_DEPTH: usize = 10; + +/// Resolve the full member list for a space, traversing delegation references. +/// +/// When a space delegates to another space (is_delegation=true), the delegated +/// space's members are included in the result. If both a direct membership and +/// a delegated membership exist for the same DID, the higher access level wins +/// (write > read). +pub async fn resolve_members( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + space_id: &str, +) -> Result, AppError> { + let mut resolved: HashMap = HashMap::new(); + let mut visited: HashSet = HashSet::new(); + + resolve_members_recursive(pool, backend, space_id, &mut resolved, &mut visited, 0).await?; + + let mut members: Vec = resolved + .into_iter() + .map(|(did, access)| ResolvedMember { did, access }) + .collect(); + members.sort_by(|a, b| a.did.cmp(&b.did)); + Ok(members) +} + +/// Check if a DID is a member of a space (resolving delegations). +pub async fn is_member( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + space_id: &str, + did: &str, +) -> Result, AppError> { + let members = resolve_members(pool, backend, space_id).await?; + Ok(members.into_iter().find(|m| m.did == did).map(|m| m.access)) +} + +fn resolve_members_recursive<'a>( + pool: &'a sqlx::AnyPool, + backend: DatabaseBackend, + space_id: &'a str, + resolved: &'a mut HashMap, + visited: &'a mut HashSet, + depth: usize, +) -> std::pin::Pin> + Send + 'a>> { + Box::pin(async move { + if depth >= MAX_DELEGATION_DEPTH { + return Ok(()); + } + + if !visited.insert(space_id.to_string()) { + return Ok(()); + } + + let direct_members = db::list_direct_members(pool, backend, space_id).await?; + + for member in direct_members { + if member.is_delegation { + let delegated_space_id = resolve_delegation_target(pool, backend, &member).await?; + if let Some(target_id) = delegated_space_id { + resolve_members_recursive( + pool, + backend, + &target_id, + resolved, + visited, + depth + 1, + ) + .await?; + } + } else { + merge_access(resolved, &member.member_did, member.access); + } + } + + Ok(()) + }) +} + +/// Resolve a delegation member entry to the target space ID. +/// +/// Delegation entries store either an ats:// URI or a space ID directly. +async fn resolve_delegation_target( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + member: &SpaceMember, +) -> Result, AppError> { + if member.member_did.starts_with("ats://") { + let uri = SpaceUri::parse(&member.member_did)?; + let space = + db::get_space_by_address(pool, backend, &uri.owner_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?; + Ok(space.map(|s| s.id)) + } +} + +fn merge_access(resolved: &mut HashMap, did: &str, access: SpaceAccess) { + let entry = resolved.entry(did.to_string()).or_insert(SpaceAccess::Read); + if access.can_write() { + *entry = SpaceAccess::Write; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn merge_access_write_wins() { + let mut map = HashMap::new(); + merge_access(&mut map, "did:plc:user1", SpaceAccess::Read); + assert_eq!(map["did:plc:user1"], SpaceAccess::Read); + + merge_access(&mut map, "did:plc:user1", SpaceAccess::Write); + assert_eq!(map["did:plc:user1"], SpaceAccess::Write); + + // Write should not be downgraded to Read + merge_access(&mut map, "did:plc:user1", SpaceAccess::Read); + assert_eq!(map["did:plc:user1"], SpaceAccess::Write); + } + + #[test] + fn merge_access_multiple_users() { + let mut map = HashMap::new(); + merge_access(&mut map, "did:plc:alice", SpaceAccess::Write); + merge_access(&mut map, "did:plc:bob", SpaceAccess::Read); + assert_eq!(map.len(), 2); + assert_eq!(map["did:plc:alice"], SpaceAccess::Write); + assert_eq!(map["did:plc:bob"], SpaceAccess::Read); + } +} diff --git a/src/spaces/mod.rs b/src/spaces/mod.rs new file mode 100644 --- /dev/null +++ b/src/spaces/mod.rs @@ -0,0 +1,212 @@ +pub mod auth; +pub mod credential; +pub mod db; +pub mod members; +pub mod notifications; +pub mod routes; +pub mod sync; +pub mod types; + +use crate::error::AppError; +use std::fmt; + +/// A parsed `ats://` URI for addressing permissioned data. +/// +/// Full form: `ats:///////` +/// Space-only form: `ats:////` +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct SpaceUri { + pub owner_did: String, + pub type_nsid: String, + pub skey: String, + pub user_did: Option, + pub collection: Option, + pub rkey: Option, +} + +impl SpaceUri { + pub fn parse(uri: &str) -> Result { + let stripped = uri + .strip_prefix("ats://") + .ok_or_else(|| AppError::BadRequest("SpaceUri must start with ats://".into()))?; + + let parts: Vec<&str> = stripped.split('/').collect(); + + if parts.len() < 3 { + return Err(AppError::BadRequest( + "SpaceUri requires at least owner_did/type_nsid/skey".into(), + )); + } + + if parts[0].is_empty() || parts[1].is_empty() || parts[2].is_empty() { + return Err(AppError::BadRequest( + "SpaceUri components must not be empty".into(), + )); + } + + let owner_did = parts[0].to_string(); + let type_nsid = parts[1].to_string(); + let skey = parts[2].to_string(); + + let (user_did, collection, rkey) = if parts.len() >= 6 { + ( + Some(parts[3].to_string()), + Some(parts[4].to_string()), + Some(parts[5].to_string()), + ) + } else if parts.len() == 3 { + (None, None, None) + } else { + return Err(AppError::BadRequest( + "SpaceUri must have 3 components (space) or 6 components (record)".into(), + )); + }; + + Ok(SpaceUri { + owner_did, + type_nsid, + skey, + user_did, + collection, + rkey, + }) + } + + pub fn space_uri(&self) -> String { + format!("ats://{}/{}/{}", self.owner_did, self.type_nsid, self.skey) + } + + pub fn is_record_uri(&self) -> bool { + self.user_did.is_some() + } + + pub fn is_space_uri(&self) -> bool { + self.user_did.is_none() + } +} + +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 + )?; + if let (Some(user), Some(col), Some(rkey)) = (&self.user_did, &self.collection, &self.rkey) + { + write!(f, "/{}/{}/{}", user, col, rkey)?; + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[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.type_nsid, "com.example.forum"); + assert_eq!(uri.skey, "main"); + assert!(uri.is_space_uri()); + assert!(!uri.is_record_uri()); + assert_eq!(uri.user_did, None); + } + + #[test] + fn parse_record_uri() { + let uri = SpaceUri::parse( + "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.type_nsid, "com.example.forum"); + assert_eq!(uri.skey, "main"); + assert_eq!(uri.user_did.as_deref(), Some("did:plc:user1")); + assert_eq!(uri.collection.as_deref(), Some("com.example.forum.post")); + assert_eq!(uri.rkey.as_deref(), Some("3k2abc")); + assert!(uri.is_record_uri()); + assert!(!uri.is_space_uri()); + } + + #[test] + fn display_space_uri() { + let uri = SpaceUri { + owner_did: "did:plc:abc123".into(), + type_nsid: "com.example.forum".into(), + skey: "main".into(), + user_did: None, + collection: None, + rkey: None, + }; + assert_eq!( + uri.to_string(), + "ats://did:plc:abc123/com.example.forum/main" + ); + } + + #[test] + fn display_record_uri() { + let uri = SpaceUri { + owner_did: "did:plc:abc123".into(), + type_nsid: "com.example.forum".into(), + skey: "main".into(), + user_did: Some("did:plc:user1".into()), + collection: Some("com.example.forum.post".into()), + rkey: Some("3k2abc".into()), + }; + assert_eq!( + uri.to_string(), + "ats://did:plc:abc123/com.example.forum/main/did:plc:user1/com.example.forum.post/3k2abc" + ); + } + + #[test] + fn space_uri_extracts_space_part() { + let uri = SpaceUri::parse( + "ats://did:plc:abc123/com.example.forum/main/did:plc:user1/com.example.forum.post/3k2abc", + ) + .unwrap(); + assert_eq!( + uri.space_uri(), + "ats://did:plc:abc123/com.example.forum/main" + ); + } + + #[test] + fn reject_at_scheme() { + let result = SpaceUri::parse("at://did:plc:abc123/com.example.forum/main"); + assert!(result.is_err()); + } + + #[test] + fn reject_too_few_components() { + let result = SpaceUri::parse("ats://did:plc:abc123/com.example.forum"); + assert!(result.is_err()); + } + + #[test] + fn reject_wrong_component_count() { + let result = SpaceUri::parse("ats://did:plc:abc123/com.example.forum/main/did:plc:user1"); + assert!(result.is_err()); + } + + #[test] + fn reject_empty_components() { + let result = SpaceUri::parse("ats:///com.example.forum/main"); + assert!(result.is_err()); + } + + #[test] + fn roundtrip_parse_display() { + let original = "ats://did:plc:abc123/com.example.forum/main"; + let uri = SpaceUri::parse(original).unwrap(); + assert_eq!(uri.to_string(), original); + + let original_record = "ats://did:plc:abc123/com.example.forum/main/did:plc:user1/com.example.forum.post/3k2abc"; + let uri = SpaceUri::parse(original_record).unwrap(); + assert_eq!(uri.to_string(), original_record); + } +} diff --git a/src/spaces/notifications.rs b/src/spaces/notifications.rs new file mode 100644 --- /dev/null +++ b/src/spaces/notifications.rs @@ -0,0 +1,86 @@ +use serde::Deserialize; +use uuid::Uuid; + +use crate::db::DatabaseBackend; +use crate::error::AppError; +use crate::spaces::db; +use crate::spaces::types::*; + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct WriteNotification { + pub space_uri: String, + pub author_did: String, + pub collection: String, + pub rkey: String, + pub action: WriteAction, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum WriteAction { + Create, + Update, + Delete, +} + +/// Process a write notification by queuing a sync pull for the affected member. +/// +/// This marks the member's sync state as pending so the next sync pass picks it up. +pub async fn handle_write_notification( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + space_id: &str, + notification: &WriteNotification, +) -> Result<(), AppError> { + let existing = db::get_sync_state(pool, backend, space_id, ¬ification.author_did).await?; + + let state = SpaceSyncState { + id: existing + .map(|s| s.id) + .unwrap_or_else(|| Uuid::new_v4().to_string()), + space_id: space_id.to_string(), + member_did: notification.author_did.clone(), + cursor: None, + last_synced_at: None, + status: SyncStatus::Pending, + error: None, + }; + + db::upsert_sync_state(pool, backend, &state).await?; + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn write_action_deserializes() { + let action: WriteAction = serde_json::from_str("\"create\"").unwrap(); + assert!(matches!(action, WriteAction::Create)); + + let action: WriteAction = serde_json::from_str("\"update\"").unwrap(); + assert!(matches!(action, WriteAction::Update)); + + let action: WriteAction = serde_json::from_str("\"delete\"").unwrap(); + assert!(matches!(action, WriteAction::Delete)); + } + + #[test] + fn write_notification_deserializes() { + let json = r#"{ + "spaceUri": "ats://did:plc:owner/com.example.forum/main", + "authorDid": "did:plc:alice", + "collection": "com.example.forum.post", + "rkey": "3k2abc", + "action": "create" + }"#; + + let notif: WriteNotification = serde_json::from_str(json).unwrap(); + assert_eq!(notif.author_did, "did:plc:alice"); + assert_eq!(notif.collection, "com.example.forum.post"); + assert!(matches!(notif.action, WriteAction::Create)); + } +} diff --git a/src/spaces/routes.rs b/src/spaces/routes.rs new file mode 100644 --- /dev/null +++ b/src/spaces/routes.rs @@ -0,0 +1,982 @@ +use axum::extract::{Query, State}; +use axum::http::{HeaderMap, StatusCode}; +use axum::response::{IntoResponse, Response}; +use axum::routing::{get, post}; +use axum::{Json, Router}; +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::spaces::types::*; +use crate::spaces::{SpaceUri, db, members}; + +// --------------------------------------------------------------------------- +// Request / response types +// --------------------------------------------------------------------------- + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct CreateSpaceInput { + type_nsid: String, + skey: String, + display_name: Option, + description: Option, + access_mode: Option, + managing_app_did: Option, + config: Option, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct SpaceUriQuery { + space_uri: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct ListSpacesQuery { + owner_did: Option, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct DeleteSpaceInput { + space_uri: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct UpdateSpaceInput { + space_uri: String, + display_name: Option>, + description: Option>, + access_mode: Option, + app_allowlist: Option>>, + app_denylist: Option>>, + managing_app_did: Option>, + config: Option, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct PutRecordInput { + space_uri: String, + collection: String, + rkey: String, + record: serde_json::Value, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct DeleteRecordInput { + space_uri: String, + collection: String, + rkey: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct GetRecordQuery { + space_uri: String, + collection: String, + rkey: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct ListRecordsQuery { + space_uri: String, + collection: Option, + limit: Option, + cursor: Option, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct AddMemberInput { + space_uri: String, + member_did: String, + access: Option, + is_delegation: Option, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct RemoveMemberInput { + space_uri: String, + member_did: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct CreateInviteInput { + space_uri: String, + access: Option, + max_uses: Option, + expires_at: Option, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct RedeemInviteInput { + token: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct RevokeInviteInput { + space_uri: String, + invite_id: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct GetCredentialInput { + space_uri: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct RefreshCredentialInput { + space_uri: String, + credential: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct WriteNotificationInput { + space_uri: String, + author_did: String, + collection: String, + rkey: String, + action: crate::spaces::notifications::WriteAction, +} + +// --------------------------------------------------------------------------- +// Route registration +// --------------------------------------------------------------------------- + +const NS: &str = "dev.happyview"; + +pub fn space_routes() -> Router { + Router::new() + // Space CRUD + .route(&format!("/xrpc/{NS}.space.create"), post(create_space)) + .route(&format!("/xrpc/{NS}.space.get"), get(get_space)) + .route(&format!("/xrpc/{NS}.space.list"), get(list_spaces)) + .route(&format!("/xrpc/{NS}.space.delete"), post(delete_space)) + .route(&format!("/xrpc/{NS}.space.update"), post(update_space)) + // Records + .route(&format!("/xrpc/{NS}.space.putRecord"), post(put_record)) + .route( + &format!("/xrpc/{NS}.space.deleteRecord"), + post(delete_record), + ) + .route(&format!("/xrpc/{NS}.space.getRecord"), get(get_record)) + .route(&format!("/xrpc/{NS}.space.listRecords"), get(list_records)) + // Members + .route(&format!("/xrpc/{NS}.space.listMembers"), get(list_members)) + .route(&format!("/xrpc/{NS}.space.addMember"), post(add_member)) + .route( + &format!("/xrpc/{NS}.space.removeMember"), + post(remove_member), + ) + // Invites + .route( + &format!("/xrpc/{NS}.space.invite.create"), + post(create_invite), + ) + .route( + &format!("/xrpc/{NS}.space.invite.redeem"), + post(redeem_invite), + ) + .route( + &format!("/xrpc/{NS}.space.invite.revoke"), + post(revoke_invite), + ) + .route(&format!("/xrpc/{NS}.space.invite.list"), get(list_invites)) + // Credentials + .route( + &format!("/xrpc/{NS}.space.getCredential"), + post(get_credential), + ) + .route( + &format!("/xrpc/{NS}.space.refreshCredential"), + post(refresh_credential), + ) + // Notifications + .route( + &format!("/xrpc/{NS}.space.writeNotification"), + post(write_notification), + ) +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +fn require_auth(claims: &XrpcClaims) -> Result<&crate::auth::Claims, AppError> { + claims + .0 + .as_ref() + .ok_or_else(|| AppError::Auth("This endpoint requires DPoP authentication".into())) +} + +async fn resolve_space(state: &AppState, space_uri: &str) -> Result { + let uri = SpaceUri::parse(space_uri)?; + db::get_space_by_address( + &state.db, + state.db_backend, + &uri.owner_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.owner_did == did { + return Ok(()); + } + let sql = adapt_sql("SELECT is_super FROM 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 owner can perform this action".into(), + )) +} + +fn extract_space_credential(headers: &HeaderMap) -> Option { + headers + .get("x-space-credential") + .and_then(|v| v.to_str().ok()) + .map(|s| s.to_string()) +} + +async fn require_membership( + state: &AppState, + space: &Space, + did: &str, + require_write: bool, + space_credential: Option<&str>, +) -> Result { + if let Some(token) = space_credential { + let space_uri = format!( + "ats://{}/{}/{}", + space.owner_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.space == space_uri => { + let access = match claims.scope.as_str() { + "write" => SpaceAccess::Write, + _ => SpaceAccess::Read, + }; + if require_write && !access.can_write() { + return Err(AppError::Forbidden( + "Write access is required for this action".into(), + )); + } + return Ok(access); + } + 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])) +} + +// --------------------------------------------------------------------------- +// Space CRUD handlers +// --------------------------------------------------------------------------- + +async fn create_space( + State(state): State, + xrpc_claims: XrpcClaims, + Json(input): Json, +) -> Result { + let claims = require_auth(&xrpc_claims)?; + 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(), + )); + } + + let existing = db::get_space_by_address( + &state.db, + state.db_backend, + &did, + &input.type_nsid, + &input.skey, + ) + .await?; + if existing.is_some() { + return Err(AppError::Conflict( + "A space with this address already exists".into(), + )); + } + + let space = Space { + id: Uuid::new_v4().to_string(), + owner_did: did.clone(), + type_nsid: input.type_nsid, + skey: input.skey, + display_name: input.display_name, + description: input.description, + access_mode: input.access_mode.unwrap_or(AccessMode::DefaultAllow), + app_allowlist: None, + app_denylist: None, + managing_app_did: input.managing_app_did, + config: input.config.unwrap_or_default(), + created_at: now_rfc3339(), + updated_at: now_rfc3339(), + }; + + db::create_space(&state.db, state.db_backend, &space).await?; + + // Auto-add the creator as a write member + let member = SpaceMember { + id: Uuid::new_v4().to_string(), + space_id: space.id.clone(), + member_did: did.clone(), + access: SpaceAccess::Write, + is_delegation: false, + granted_by: Some(did), + created_at: now_rfc3339(), + }; + db::add_member(&state.db, state.db_backend, &member).await?; + + let space_uri = format!( + "ats://{}/{}/{}", + space.owner_did, space.type_nsid, space.skey + ); + let body = serde_json::json!({ + "spaceUri": space_uri, + "space": space, + }); + + let mut response = Json(body).into_response(); + *response.status_mut() = StatusCode::CREATED; + Ok(response) +} + +async fn get_space( + State(state): State, + xrpc_claims: XrpcClaims, + Query(query): Query, +) -> Result, AppError> { + let space = resolve_space(&state, &query.space_uri).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.owner_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.owner_did, space.type_nsid, space.skey + ); + Ok(Json(serde_json::json!({ + "spaceUri": space_uri, + "space": space, + }))) +} + +async fn list_spaces( + State(state): State, + xrpc_claims: XrpcClaims, + Query(query): Query, +) -> Result, AppError> { + let claims = require_auth(&xrpc_claims)?; + let did = claims.did().to_string(); + + 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 spaces_with_uris: Vec = spaces + .into_iter() + .map(|s| { + let uri = format!("ats://{}/{}/{}", s.owner_did, s.type_nsid, s.skey); + serde_json::json!({ "spaceUri": uri, "space": s }) + }) + .collect(); + + Ok(Json(serde_json::json!({ "spaces": spaces_with_uris }))) +} + +async fn delete_space( + State(state): State, + xrpc_claims: XrpcClaims, + Json(input): Json, +) -> Result, AppError> { + let claims = require_auth(&xrpc_claims)?; + let space = resolve_space(&state, &input.space_uri).await?; + require_space_admin(&state, &space, claims.did()).await?; + + db::delete_space(&state.db, state.db_backend, &space.id).await?; + + Ok(Json(serde_json::json!({ "success": true }))) +} + +async fn update_space( + State(state): State, + xrpc_claims: XrpcClaims, + Json(input): Json, +) -> Result, AppError> { + let claims = require_auth(&xrpc_claims)?; + let mut space = resolve_space(&state, &input.space_uri).await?; + require_space_admin(&state, &space, claims.did()).await?; + + if let Some(name) = input.display_name { + space.display_name = name; + } + if let Some(desc) = input.description { + space.description = desc; + } + if let Some(mode) = input.access_mode { + space.access_mode = mode; + } + if let Some(list) = input.app_allowlist { + space.app_allowlist = list; + } + if let Some(list) = input.app_denylist { + space.app_denylist = list; + } + if let Some(did) = input.managing_app_did { + space.managing_app_did = did; + } + if let Some(config) = input.config { + space.config = config; + } + + db::update_space(&state.db, state.db_backend, &space).await?; + + let space_uri = format!( + "ats://{}/{}/{}", + space.owner_did, space.type_nsid, space.skey + ); + Ok(Json(serde_json::json!({ + "spaceUri": space_uri, + "space": space, + }))) +} + +// --------------------------------------------------------------------------- +// Record handlers +// --------------------------------------------------------------------------- + +async fn put_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_uri).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 + ); + + let record = SpaceRecord { + uri: record_uri.clone(), + space_id: space.id, + author_did: did, + collection: input.collection, + rkey: input.rkey, + record: input.record, + cid: cid.clone(), + indexed_at: now_rfc3339(), + }; + + db::upsert_space_record(&state.db, state.db_backend, &record).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, + xrpc_claims: XrpcClaims, + 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 record_uri = format!( + "ats://{}/{}/{}/{}/{}/{}", + space.owner_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())); + } + _ => {} + } + + db::delete_space_record(&state.db, state.db_backend, &record_uri).await?; + + Ok(Json(serde_json::json!({ "success": true }))) +} + +async fn get_record( + State(state): State, + xrpc_claims: XrpcClaims, + headers: HeaderMap, + Query(query): Query, +) -> Result, AppError> { + let claims = require_auth(&xrpc_claims)?; + let space = resolve_space(&state, &query.space_uri).await?; + let cred = extract_space_credential(&headers); + require_membership(&state, &space, claims.did(), false, cred.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()))?; + + Ok(Json(serde_json::json!({ + "uri": record.uri, + "space": query.space_uri, + "collection": record.collection, + "record": record.record, + "cid": record.cid, + }))) +} + +async fn list_records( + State(state): State, + xrpc_claims: XrpcClaims, + headers: HeaderMap, + Query(query): Query, +) -> Result, AppError> { + let claims = require_auth(&xrpc_claims)?; + let space = resolve_space(&state, &query.space_uri).await?; + let cred = extract_space_credential(&headers); + require_membership(&state, &space, claims.did(), false, cred.as_deref()).await?; + + let limit = query.limit.unwrap_or(50).min(100); + let records = db::list_space_records( + &state.db, + state.db_backend, + &space.id, + query.collection.as_deref(), + limit, + query.cursor.as_deref(), + ) + .await?; + + let cursor = records.last().map(|r| r.indexed_at.clone()); + + let records_json: Vec = records + .into_iter() + .map(|r| { + serde_json::json!({ + "uri": r.uri, + "space": query.space_uri, + "collection": r.collection, + "record": r.record, + "cid": r.cid, + }) + }) + .collect(); + + Ok(Json(serde_json::json!({ + "records": records_json, + "cursor": cursor, + }))) +} + +// --------------------------------------------------------------------------- +// Member handlers +// --------------------------------------------------------------------------- + +async fn list_members( + State(state): State, + xrpc_claims: XrpcClaims, + headers: HeaderMap, + Query(query): Query, +) -> Result, AppError> { + let space = resolve_space(&state, &query.space_uri).await?; + + if !space.config.membership_public { + let claims = require_auth(&xrpc_claims)?; + let cred = extract_space_credential(&headers); + require_membership(&state, &space, claims.did(), false, cred.as_deref()).await?; + } + + let resolved = members::resolve_members(&state.db, state.db_backend, &space.id).await?; + + Ok(Json(serde_json::json!({ "members": resolved }))) +} + +async fn add_member( + State(state): State, + xrpc_claims: XrpcClaims, + Json(input): Json, +) -> Result { + let claims = require_auth(&xrpc_claims)?; + let space = resolve_space(&state, &input.space_uri).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?; + if existing.is_some() { + return Err(AppError::Conflict( + "Member already exists in this space".into(), + )); + } + + let member = SpaceMember { + id: Uuid::new_v4().to_string(), + space_id: space.id, + member_did: input.member_did, + access: input.access.unwrap_or(SpaceAccess::Read), + is_delegation: input.is_delegation.unwrap_or(false), + granted_by: Some(claims.did().to_string()), + created_at: now_rfc3339(), + }; + + db::add_member(&state.db, state.db_backend, &member).await?; + + let mut response = Json(serde_json::json!({ "member": member })).into_response(); + *response.status_mut() = StatusCode::CREATED; + Ok(response) +} + +async fn remove_member( + State(state): State, + xrpc_claims: XrpcClaims, + Json(input): Json, +) -> Result, AppError> { + let claims = require_auth(&xrpc_claims)?; + let space = resolve_space(&state, &input.space_uri).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?; + + if !removed { + return Err(AppError::NotFound("Member not found in this space".into())); + } + + Ok(Json(serde_json::json!({ "success": true }))) +} + +// --------------------------------------------------------------------------- +// Invite handlers +// --------------------------------------------------------------------------- + +async fn create_invite( + State(state): State, + xrpc_claims: XrpcClaims, + Json(input): Json, +) -> Result { + let claims = require_auth(&xrpc_claims)?; + let space = resolve_space(&state, &input.space_uri).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 redeem_invite( + State(state): State, + xrpc_claims: XrpcClaims, + Json(input): Json, +) -> Result { + 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(), + member_did: 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.owner_did, s.type_nsid, s.skey)); + + let mut response = Json(serde_json::json!({ + "spaceUri": space_uri, + "access": member.access, + })) + .into_response(); + *response.status_mut() = StatusCode::CREATED; + Ok(response) +} + +async fn revoke_invite( + State(state): State, + xrpc_claims: XrpcClaims, + Json(input): Json, +) -> Result, AppError> { + let claims = require_auth(&xrpc_claims)?; + let space = resolve_space(&state, &input.space_uri).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, + xrpc_claims: XrpcClaims, + Query(query): Query, +) -> Result, AppError> { + let claims = require_auth(&xrpc_claims)?; + let space = resolve_space(&state, &query.space_uri).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 = 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_credential( + State(state): State, + xrpc_claims: XrpcClaims, + 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?; + + require_membership(&state, &space, &did, false, None).await?; + + 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 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?; + + Ok(Json(serde_json::json!({ + "credential": issued.token, + "expiresAt": issued.expires_at, + }))) +} + +async fn refresh_credential( + State(state): State, + xrpc_claims: XrpcClaims, + Json(input): Json, +) -> Result, AppError> { + let _claims = require_auth(&xrpc_claims)?; + let space = resolve_space(&state, &input.space_uri).await?; + + 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( + &state.db, + state.db_backend, + encryption_key, + &space, + &input.credential, + ) + .await?; + + Ok(Json(serde_json::json!({ + "credential": issued.token, + "expiresAt": issued.expires_at, + }))) +} + +// --------------------------------------------------------------------------- +// Notification handlers +// --------------------------------------------------------------------------- + +async fn write_notification( + State(state): State, + xrpc_claims: XrpcClaims, + Json(input): Json, +) -> Result, AppError> { + let claims = require_auth(&xrpc_claims)?; + let space = resolve_space(&state, &input.space_uri).await?; + require_space_admin(&state, &space, claims.did()).await?; + + let notification = crate::spaces::notifications::WriteNotification { + space_uri: input.space_uri, + author_did: input.author_did, + collection: input.collection, + rkey: input.rkey, + action: input.action, + }; + + crate::spaces::notifications::handle_write_notification( + &state.db, + state.db_backend, + &space.id, + ¬ification, + ) + .await?; + + Ok(Json(serde_json::json!({ "success": true }))) +} diff --git a/src/spaces/sync.rs b/src/spaces/sync.rs new file mode 100644 --- /dev/null +++ b/src/spaces/sync.rs @@ -0,0 +1,293 @@ +use uuid::Uuid; + +use crate::db::DatabaseBackend; +use crate::db::now_rfc3339; +use crate::error::AppError; +use crate::profile::resolve_pds_endpoint; +use crate::spaces::types::*; +use crate::spaces::{db, members}; + +/// Sync all members of a space by pulling records from their PDSes. +pub async fn sync_space( + http: &reqwest::Client, + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + plc_url: &str, + space_id: &str, + collections: &[String], +) -> Result { + let resolved = members::resolve_members(pool, backend, space_id).await?; + let mut results = Vec::new(); + + for member in &resolved { + let result = sync_member( + http, + pool, + backend, + plc_url, + space_id, + &member.did, + collections, + ) + .await; + + results.push(MemberSyncResult { + did: member.did.clone(), + records_synced: result.as_ref().map(|r| r.records_synced).unwrap_or(0), + error: result.err().map(|e| e.to_string()), + }); + } + + let total = results.iter().map(|r| r.records_synced).sum(); + + Ok(SyncSpaceResult { + members_processed: results.len(), + total_records_synced: total, + member_results: results, + }) +} + +/// Sync records from a single member's PDS for a given space. +pub async fn sync_member( + http: &reqwest::Client, + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + plc_url: &str, + space_id: &str, + member_did: &str, + collections: &[String], +) -> Result { + let state_id = match db::get_sync_state(pool, backend, space_id, member_did).await? { + Some(s) => s.id, + None => { + let id = Uuid::new_v4().to_string(); + let initial = SpaceSyncState { + id: id.clone(), + space_id: space_id.to_string(), + member_did: member_did.to_string(), + cursor: None, + last_synced_at: None, + status: SyncStatus::Pending, + error: None, + }; + db::upsert_sync_state(pool, backend, &initial).await?; + id + } + }; + + // Mark as syncing + let syncing_state = SpaceSyncState { + id: state_id.clone(), + space_id: space_id.to_string(), + member_did: member_did.to_string(), + cursor: None, + last_synced_at: None, + status: SyncStatus::Syncing, + error: None, + }; + db::upsert_sync_state(pool, backend, &syncing_state).await?; + + let result = pull_member_records( + http, + pool, + backend, + plc_url, + space_id, + member_did, + collections, + ) + .await; + + match result { + Ok(summary) => { + let done = SpaceSyncState { + id: state_id, + space_id: space_id.to_string(), + member_did: member_did.to_string(), + cursor: summary.cursor.clone(), + last_synced_at: Some(now_rfc3339()), + status: SyncStatus::Synced, + error: None, + }; + db::upsert_sync_state(pool, backend, &done).await?; + Ok(summary) + } + Err(e) => { + let err_state = SpaceSyncState { + id: state_id, + space_id: space_id.to_string(), + member_did: member_did.to_string(), + cursor: None, + last_synced_at: Some(now_rfc3339()), + status: SyncStatus::Error, + error: Some(e.to_string()), + }; + db::upsert_sync_state(pool, backend, &err_state).await?; + Err(e) + } + } +} + +async fn pull_member_records( + http: &reqwest::Client, + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + plc_url: &str, + space_id: &str, + member_did: &str, + collections: &[String], +) -> Result { + let pds_url = resolve_pds_endpoint(http, plc_url, member_did).await?; + let mut total_records = 0usize; + let mut last_cursor = None; + + for collection in collections { + let mut cursor: Option = None; + loop { + let (records, next_cursor) = fetch_records_page( + http, + &pds_url, + member_did, + collection, + cursor.as_deref(), + 100, + ) + .await?; + + if records.is_empty() { + break; + } + + for record in &records { + let uri = record["uri"].as_str().unwrap_or(""); + let rkey = extract_rkey(uri); + let cid = record["cid"].as_str().unwrap_or("").to_string(); + let value = record + .get("value") + .cloned() + .unwrap_or(serde_json::Value::Null); + + let space_record_uri = format!("ats://{space_id}/{member_did}/{collection}/{rkey}"); + + let space_record = SpaceRecord { + uri: space_record_uri, + space_id: space_id.to_string(), + author_did: member_did.to_string(), + collection: collection.clone(), + rkey: rkey.to_string(), + record: value, + cid, + indexed_at: now_rfc3339(), + }; + + db::upsert_space_record(pool, backend, &space_record).await?; + total_records += 1; + } + + last_cursor = next_cursor.clone(); + cursor = next_cursor; + + if cursor.is_none() { + break; + } + } + } + + Ok(MemberSyncSummary { + records_synced: total_records, + cursor: last_cursor, + }) +} + +async fn fetch_records_page( + http: &reqwest::Client, + pds_url: &str, + repo: &str, + collection: &str, + cursor: Option<&str>, + limit: u32, +) -> Result<(Vec, Option), AppError> { + let mut url = format!( + "{}/xrpc/com.atproto.repo.listRecords?repo={}&collection={}&limit={}", + pds_url.trim_end_matches('/'), + repo, + collection, + limit, + ); + + if let Some(c) = cursor { + url.push_str(&format!("&cursor={c}")); + } + + let resp = http + .get(&url) + .send() + .await + .map_err(|e| AppError::Internal(format!("PDS request failed: {e}")))?; + + if !resp.status().is_success() { + let status = resp.status(); + return Err(AppError::Internal(format!( + "PDS listRecords failed with {status} for {repo}/{collection}" + ))); + } + + let body: serde_json::Value = resp + .json() + .await + .map_err(|e| AppError::Internal(format!("invalid PDS response: {e}")))?; + + let records = body["records"].as_array().cloned().unwrap_or_default(); + + let next_cursor = body["cursor"].as_str().map(|s| s.to_string()); + + Ok((records, next_cursor)) +} + +fn extract_rkey(uri: &str) -> &str { + uri.rsplit('/').next().unwrap_or("") +} + +// --------------------------------------------------------------------------- +// Result types +// --------------------------------------------------------------------------- + +pub struct SyncSpaceResult { + pub members_processed: usize, + pub total_records_synced: usize, + pub member_results: Vec, +} + +pub struct MemberSyncResult { + pub did: String, + pub records_synced: usize, + pub error: Option, +} + +pub struct MemberSyncSummary { + pub records_synced: usize, + pub cursor: Option, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn extract_rkey_from_at_uri() { + assert_eq!( + extract_rkey("at://did:plc:abc/app.bsky.feed.post/3k2abc"), + "3k2abc" + ); + } + + #[test] + fn extract_rkey_from_empty() { + assert_eq!(extract_rkey(""), ""); + } + + #[test] + fn extract_rkey_no_slash() { + assert_eq!(extract_rkey("singlevalue"), "singlevalue"); + } +} diff --git a/src/spaces/types.rs b/src/spaces/types.rs new file mode 100644 --- /dev/null +++ b/src/spaces/types.rs @@ -0,0 +1,248 @@ +use serde::{Deserialize, Serialize}; +use std::fmt; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum SpaceAccess { + Read, + Write, +} + +impl SpaceAccess { + pub fn as_str(&self) -> &'static str { + match self { + SpaceAccess::Read => "read", + SpaceAccess::Write => "write", + } + } + + pub fn parse(s: &str) -> Option { + match s { + "read" => Some(SpaceAccess::Read), + "write" => Some(SpaceAccess::Write), + _ => None, + } + } + + pub fn can_write(&self) -> bool { + matches!(self, SpaceAccess::Write) + } + + pub fn can_read(&self) -> bool { + true + } +} + +impl fmt::Display for SpaceAccess { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum AccessMode { + DefaultAllow, + DefaultDeny, +} + +impl AccessMode { + pub fn as_str(&self) -> &'static str { + match self { + AccessMode::DefaultAllow => "default_allow", + AccessMode::DefaultDeny => "default_deny", + } + } + + pub fn parse(s: &str) -> Option { + match s { + "default_allow" => Some(AccessMode::DefaultAllow), + "default_deny" => Some(AccessMode::DefaultDeny), + _ => None, + } + } +} + +impl fmt::Display for AccessMode { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Space { + pub id: String, + pub owner_did: String, + pub type_nsid: String, + pub skey: String, + pub display_name: Option, + pub description: Option, + pub access_mode: AccessMode, + pub app_allowlist: Option>, + pub app_denylist: Option>, + pub managing_app_did: Option, + pub config: SpaceConfig, + pub created_at: String, + pub updated_at: String, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct SpaceConfig { + #[serde(default)] + pub membership_public: bool, + #[serde(default)] + pub records_public: bool, + #[serde(flatten)] + pub extra: serde_json::Map, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SpaceMember { + pub id: String, + pub space_id: String, + pub member_did: String, + pub access: SpaceAccess, + pub is_delegation: bool, + pub granted_by: Option, + pub created_at: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ResolvedMember { + pub did: String, + pub access: SpaceAccess, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SpaceRecord { + pub uri: String, + pub space_id: String, + pub author_did: String, + pub collection: String, + pub rkey: String, + pub record: serde_json::Value, + pub cid: String, + pub indexed_at: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SpaceInvite { + pub id: String, + pub space_id: String, + pub token_hash: String, + pub created_by: String, + pub access: SpaceAccess, + pub max_uses: Option, + pub uses: i64, + pub expires_at: Option, + pub revoked: bool, + pub created_at: String, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum SyncStatus { + Pending, + Syncing, + Synced, + Error, +} + +impl SyncStatus { + pub fn as_str(&self) -> &'static str { + match self { + SyncStatus::Pending => "pending", + SyncStatus::Syncing => "syncing", + SyncStatus::Synced => "synced", + SyncStatus::Error => "error", + } + } + + pub fn parse(s: &str) -> Option { + match s { + "pending" => Some(SyncStatus::Pending), + "syncing" => Some(SyncStatus::Syncing), + "synced" => Some(SyncStatus::Synced), + "error" => Some(SyncStatus::Error), + _ => None, + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SpaceSyncState { + pub id: String, + pub space_id: String, + pub member_did: String, + pub cursor: Option, + pub last_synced_at: Option, + pub status: SyncStatus, + pub error: Option, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn space_access_roundtrip() { + assert_eq!(SpaceAccess::parse("read"), Some(SpaceAccess::Read)); + assert_eq!(SpaceAccess::parse("write"), Some(SpaceAccess::Write)); + assert_eq!(SpaceAccess::parse("admin"), None); + + assert_eq!(SpaceAccess::Read.as_str(), "read"); + assert_eq!(SpaceAccess::Write.as_str(), "write"); + } + + #[test] + fn space_access_permissions() { + assert!(SpaceAccess::Read.can_read()); + assert!(!SpaceAccess::Read.can_write()); + assert!(SpaceAccess::Write.can_read()); + assert!(SpaceAccess::Write.can_write()); + } + + #[test] + fn access_mode_roundtrip() { + assert_eq!( + AccessMode::parse("default_allow"), + Some(AccessMode::DefaultAllow) + ); + assert_eq!( + AccessMode::parse("default_deny"), + Some(AccessMode::DefaultDeny) + ); + assert_eq!(AccessMode::parse("open"), None); + } + + #[test] + fn space_config_defaults() { + let config: SpaceConfig = serde_json::from_str("{}").unwrap(); + assert!(!config.membership_public); + assert!(!config.records_public); + } + + #[test] + fn space_config_with_extra_fields() { + let config: SpaceConfig = + serde_json::from_str(r#"{"membership_public": true, "custom_field": 42}"#).unwrap(); + assert!(config.membership_public); + assert!(!config.records_public); + assert_eq!(config.extra.get("custom_field").unwrap(), &42); + } + + #[test] + fn space_access_serialization() { + let json = serde_json::to_string(&SpaceAccess::Read).unwrap(); + assert_eq!(json, "\"read\""); + + let json = serde_json::to_string(&SpaceAccess::Write).unwrap(); + assert_eq!(json, "\"write\""); + + let parsed: SpaceAccess = serde_json::from_str("\"read\"").unwrap(); + assert_eq!(parsed, SpaceAccess::Read); + + let parsed: SpaceAccess = serde_json::from_str("\"write\"").unwrap(); + assert_eq!(parsed, SpaceAccess::Write); + } +} diff --git a/src/xrpc/procedure.rs b/src/xrpc/procedure.rs --- a/src/xrpc/procedure.rs +++ b/src/xrpc/procedure.rs @@ -20,7 +20,7 @@ lexicon: &crate::lexicon::ParsedLexicon, ) -> Result { if let Some(ref script) = lexicon.script { return crate::lua::execute_procedure_script( - state, method, claims, input, params, lexicon, script, + state, method, claims, input, params, lexicon, script, None, ) .await; } diff --git a/src/xrpc/query.rs b/src/xrpc/query.rs --- a/src/xrpc/query.rs +++ b/src/xrpc/query.rs @@ -16,8 +16,10 @@ lexicon: &crate::lexicon::ParsedLexicon, claims: Option<&Claims>, ) -> Result { if let Some(ref script) = lexicon.script { - return crate::lua::execute_query_script(state, method, params, lexicon, script, claims) - .await; + return crate::lua::execute_query_script( + state, method, params, lexicon, script, claims, None, + ) + .await; } // Single-record query: has a `uri` parameter