diff --git a/Cargo.lock b/Cargo.lock index 281d337cb..877864ca6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5649,6 +5649,7 @@ dependencies = [ "http", "http-body", "httpdate", + "ipld-core", "jacquard-axum", "jacquard-common", "jacquard-repo", diff --git a/crates/lexicons/src/lib.rs b/crates/lexicons/src/lib.rs index d5e211d62..d23a7bb71 100644 --- a/crates/lexicons/src/lib.rs +++ b/crates/lexicons/src/lib.rs @@ -5,6 +5,7 @@ extern crate alloc; #[allow( clippy::absurd_extreme_comparisons, clippy::collapsible_if, + clippy::len_zero, clippy::manual_strip, clippy::needless_update, clippy::new_ret_no_self, diff --git a/knot2/crates/knot-index/tests/common/mod.rs b/knot2/crates/knot-index/tests/common/mod.rs index 05552e7ed..b564c5e0e 100644 --- a/knot2/crates/knot-index/tests/common/mod.rs +++ b/knot2/crates/knot-index/tests/common/mod.rs @@ -8,8 +8,7 @@ use knot_cob::{ }; use knot_cobs::{ CollaboratorsChange, Creation, Editing, Erasure, Grant, IssueChange, IssueCob, - IssueStateChange, IssueStateCob, MembersChange, Registration, RegistryChange, Removal, - Transition, + IssueStateChange, MembersChange, Registration, RegistryChange, Removal, Transition, }; use knot_git::{Layout, Repo}; use knot_index::{Index, KeyBudget}; @@ -18,7 +17,7 @@ use knot_runtime::{K256Signer, SeededEntropy}; use knot_types::{ AccountDid, Blocked, ContributionPermission, ContributionPolicy, KnotId, OwnerDid, RecordAddress, RecordBody, RecordCid, RecordCollection, RecordRkey, RepoDid, RepoName, - RepoRkey, RepoRole, SubjectNumber, TypeName, UnixSeconds, + RepoRkey, RepoRole, SubjectNumber, TypeName, UnixSeconds, dag_cbor_cid, }; use serde::Serialize; use serde::de::DeserializeOwned; @@ -48,6 +47,10 @@ pub fn at(seconds: i64) -> UnixSeconds { UnixSeconds::new(seconds) } +fn num(number: u32) -> SubjectNumber { + SubjectNumber::new(number).expect("Subject is numbered from one") +} + pub fn meta_home() -> CobHome { CobHome::from(&KnotId::new("did:web:knot.nel.pet").unwrap()) } @@ -67,7 +70,7 @@ pub fn registration(owner_id: &str, key: &str, repo: &RepoDid, seconds: i64) -> name: RepoName::new(key).unwrap(), repo: repo.clone(), created_at: at(seconds), - policy: knot_types::ContributionPolicy::default(), + policy: ContributionPolicy::default(), } } @@ -84,7 +87,7 @@ pub fn named_registration( name: RepoName::new(display).unwrap(), repo: repo.clone(), created_at: at(seconds), - policy: knot_types::ContributionPolicy::default(), + policy: ContributionPolicy::default(), } } @@ -347,9 +350,16 @@ impl World { text: &str, seconds: i64, ) -> CobId { - let (body, _) = note_record(text); - let change = issue_opening(&tid(key), author, number, text, seconds); - self.open_issue(repo, &change, Record::version(body), seconds) + self.opened( + repo, + key, + author, + number, + None, + knot_cobs::Issue, + text, + seconds, + ) } #[allow(clippy::too_many_arguments)] @@ -362,10 +372,99 @@ impl World { parent: &RecordAddress, state: IssueState, seconds: i64, + ) -> CobId { + self.opened( + repo, + key, + author, + number, + Some(parent), + Transition { state }, + key, + seconds, + ) + } + + #[allow(clippy::too_many_arguments)] + pub fn opened( + &self, + repo: &RepoDid, + key: &str, + author: &AccountDid, + number: u32, + parent: Option<&RecordAddress>, + body: B, + prose: &str, + seconds: i64, ) -> CobId { let (change, record) = - state_opening(&tid(key), author, number, parent.clone(), state, seconds); - self.open_cob::(repo, &change, Record::version(record), seconds) + contribution_of(key, author, number, parent, body, prose, at(seconds)); + self.open_cob::>(repo, &change, Record::version(record), seconds) + } + pub fn erase( + &self, + repo: &RepoDid, + object: CobId, + author: &AccountDid, + from: RecordCid, + seconds: i64, + ) { + self.append_cob::>( + repo, + object, + &erased(author, from, at(seconds)), + Record::Tombstone, + seconds, + ); + } + + #[allow(clippy::too_many_arguments)] + pub fn comment( + &self, + repo: &RepoDid, + key: &str, + author: &AccountDid, + number: u32, + parent: &RecordAddress, + text: &str, + seconds: i64, + ) -> CobId { + self.opened( + repo, + key, + author, + number, + Some(parent), + knot_cobs::Comment { + subject: RecordCid::new(dag_cbor_cid(b"subject")).unwrap(), + reply: None, + }, + text, + seconds, + ) + } + + #[allow(clippy::too_many_arguments)] + pub fn reaction( + &self, + repo: &RepoDid, + key: &str, + author: &AccountDid, + number: u32, + parent: &RecordAddress, + kind: knot_cobs::ReactionKind, + seconds: i64, + ) -> CobId { + self.opened( + repo, + key, + author, + number, + Some(parent), + knot_cobs::Reaction { kind }, + key, + seconds, + ) } } @@ -418,12 +517,20 @@ pub fn state_at(key: &str) -> RecordAddress { state_address(&tid(key)) } +pub fn comment_at(key: &str) -> RecordAddress { + comment_address(&tid(key)) +} + addressed! { note_collection of NoteCob::collection(), note_address issue_collection of knot_record::issue::issue_collection(), issue_address, issue_type_name of IssueChange state_collection of knot_record::issue::issue_state_collection(), state_address, state_type_name of IssueStateChange + comment_collection of knot_record::comment::comment_collection(), comment_address, + comment_type_name of knot_cobs::CommentChange + reaction_collection of knot_record::reaction::reaction_collection(), reaction_address, + reaction_type_name of knot_cobs::ReactionChange } pub fn note_record(text: &str) -> (RecordBody, RecordCid) { @@ -438,53 +545,66 @@ pub fn note_record(text: &str) -> (RecordBody, RecordCid) { (body, cid) } -pub fn opening( - key: &str, +pub fn contribution( + address: RecordAddress, author: &AccountDid, number: u32, parent: Option, - text: &str, - seconds: i64, -) -> (NoteChange, RecordBody) { - let (bytes, produces) = note_record(text); - ( - NoteChange::Open { - creation: Creation { - address: note_address(&tid(key)), - author: author.clone(), - number: SubjectNumber::new(number).expect("Subject is numbered from one"), - parent, - created_at: at(seconds), - produces, - }, - body: NoteBody { - text: text.to_owned(), - }, + produces: RecordCid, + body: B, + seconds: UnixSeconds, +) -> knot_cobs::SocialChange { + knot_cobs::SocialChange::Open { + creation: Creation { + address, + author: author.clone(), + number: num(number), + parent, + created_at: seconds, + produces, }, - bytes, - ) + body, + } } -pub fn issue_opening( - tid: &RecordRkey, +pub fn erased( author: &AccountDid, - number: u32, - text: &str, - seconds: i64, -) -> IssueChange { - IssueChange::Open { - creation: Creation { - address: issue_address(tid), + from: RecordCid, + seconds: UnixSeconds, +) -> knot_cobs::SocialChange { + knot_cobs::SocialChange::Erase { + erasure: Erasure { author: author.clone(), - number: SubjectNumber::new(number).expect("Subject is numbered from one"), - parent: None, - created_at: at(seconds), - produces: note_record(text).1, + from, + erased_at: seconds, }, - body: knot_cobs::Issue, } } +pub fn contribution_of( + key: &str, + author: &AccountDid, + number: u32, + parent: Option<&RecordAddress>, + body: B, + prose: &str, + seconds: UnixSeconds, +) -> (knot_cobs::SocialChange, RecordBody) { + let (bytes, produces) = note_record(prose); + ( + contribution( + RecordAddress::new(RecordCollection::new(B::COLLECTION).unwrap(), tid(key)), + author, + number, + parent.cloned(), + produces, + body, + seconds, + ), + bytes, + ) +} + pub fn issue_editing( author: &AccountDid, from: RecordCid, @@ -501,38 +621,3 @@ pub fn issue_editing( body: knot_cobs::Issue, } } - -pub fn issue_erasing(author: &AccountDid, from: RecordCid, seconds: i64) -> IssueChange { - IssueChange::Erase { - erasure: Erasure { - author: author.clone(), - from, - erased_at: at(seconds), - }, - } -} - -pub fn state_opening( - tid: &RecordRkey, - author: &AccountDid, - number: u32, - parent: RecordAddress, - state: IssueState, - seconds: i64, -) -> (IssueStateChange, RecordBody) { - let (bytes, produces) = note_record(tid.as_str()); - ( - IssueStateChange::Open { - creation: Creation { - address: state_address(tid), - author: author.clone(), - number: SubjectNumber::new(number).expect("Subject is numbered from one"), - parent: Some(parent), - created_at: at(seconds), - produces, - }, - body: Transition { state }, - }, - bytes, - ) -} diff --git a/knot2/crates/knot-index/tests/lifecycle.rs b/knot2/crates/knot-index/tests/lifecycle.rs index 196b20059..f6d04fb09 100644 --- a/knot2/crates/knot-index/tests/lifecycle.rs +++ b/knot2/crates/knot-index/tests/lifecycle.rs @@ -308,7 +308,7 @@ fn a_folded_repo_serves_while_unaccessed_repos_stay_warming() { assert_eq!( index.effective_collaborator(&repo_did("conch"), &acc("lyna")), Resolved::Warming, - "repo the index never folded cannot answer, so it fails closed" + "a repo the index never read can't be checked, so access must be denied" ); } diff --git a/knot2/crates/knot-index/tests/projections.rs b/knot2/crates/knot-index/tests/projections.rs index 1f56fe201..0415998b9 100644 --- a/knot2/crates/knot-index/tests/projections.rs +++ b/knot2/crates/knot-index/tests/projections.rs @@ -15,10 +15,12 @@ use serde::{Deserialize, Serialize}; mod common; use common::{ - NoteCob, World, acc, at, grant, issue_at, issue_collection, issue_editing, issue_erasing, - issue_type_name, meta_home, named_registration, note_address, note_record, opening, own, - registration, repo_did, rkey, state_at, state_collection, state_type_name, tid, + NoteBody, NoteCob, World, acc, at, comment_at, comment_collection, contribution_of, erased, + grant, issue_at, issue_collection, issue_editing, issue_type_name, meta_home, + named_registration, note_address, note_record, own, reaction_collection, registration, + repo_did, rkey, state_at, state_collection, state_type_name, tid, }; +use knot_cobs::ReactionKind; #[derive(Serialize, Deserialize)] #[serde(tag = "op", content = "data", rename_all = "snake_case")] @@ -608,7 +610,7 @@ fn recording_and_retaining_track_who_still_publishes_each_key() { } #[test] -fn a_full_budget_reports_unheld_readings_and_then_refuses_to_grow_at_all() { +fn full_budget_reports_unheld_readings_and_then_stops_growing_entirely() { let world = World::new(); let index = world.index_within(KeyBudget::from_bytes(560)); let fat = |seed: u8| OfferedKey::from_bytes(vec![seed; 48]); @@ -780,7 +782,7 @@ fn an_unfolded_repo_keeps_the_pass_partial_and_an_unopenable_repo_never_grants() ); assert!( !work.complete, - "a partial roll mustn't let the fill call the set complete, or the knot refuses a \ + "a partial roll mustn't let the fill call the set complete, or the knot rejects a \ pusher whose repo it never managed to read" ); @@ -1049,7 +1051,7 @@ fn an_account_the_knot_never_reads_stops_delaying_the_whole_set() { assert!( !pushers_complete(&index, at(0)), "an unread account mustn't count toward a complete set, or the knot refuses the keys \ - it has yet to fetch" + the account publishes" ); assert!( matches!(index.keys().work(at(299), sweep()), Resolved::Ready(work) if work.due.is_empty()), @@ -1673,7 +1675,7 @@ fn assert_projections_agree( objects: &[CobId], addresses: &[RecordAddress], ) { - [issue_collection(), state_collection()] + [issue_collection(), state_collection(), comment_collection(), reaction_collection()] .into_iter() .for_each(|collection| { assert_eq!( @@ -1714,16 +1716,21 @@ fn assert_projections_agree( cold.social_children_of(repo, at), "Incremental fold and cold fold disagree about children of {at}" ); - [issue_collection(), state_collection()] - .into_iter() - .for_each(|collection| { - assert_eq!( - incremental.social_latest_child(repo, at, &collection), - cold.social_latest_child(repo, at, &collection), - "Incremental replay and cold rebuild mustn't disagree about the latest \ + [ + issue_collection(), + state_collection(), + comment_collection(), + reaction_collection(), + ] + .into_iter() + .for_each(|collection| { + assert_eq!( + incremental.social_latest_child(repo, at, &collection), + cold.social_latest_child(repo, at, &collection), + "Incremental replay and cold rebuild mustn't disagree about the latest \ {collection} under {at}" - ); - }); + ); + }); }); } @@ -1818,6 +1825,140 @@ fn backdated_transition_still_ranks_latest_by_write_order() { ); } +#[test] +fn erased_subject_stops_answering_children_and_comment_keeps_serving() { + let (world, repo, issue_object) = kelp_issue(); + let parent = issue_at("3lubrptx57d22"); + let comment_object = world.comment( + &repo, + "3lubrptx57d33", + &acc("nel"), + 1, + &parent, + "kelp reply", + 20, + ); + let index = world.index(); + index.refresh_social(&repo).unwrap(); + assert_eq!( + index.social_children_of(&repo, &parent), + Resolved::Ready(vec![comment_object]), + "while the issue exists, comments on it ought to be listed under it" + ); + assert_eq!( + index.social_latest_child(&repo, &parent, &comment_collection()), + Resolved::Ready(Some(comment_at("3lubrptx57d33"))), + "the newest comment under the issue shoukd probably come back too" + ); + let before = + state_of(&index, &repo, comment_object).expect("the index should contain the comment"); + + world.erase::( + &repo, + issue_object, + &acc("nel"), + version_of(&index, &repo, issue_object).expect("the issue should probably have a version"), + 30, + ); + index.refresh_social(&repo).unwrap(); + + assert_eq!( + index.social_children_of(&repo, &parent), + Resolved::Ready(Vec::new()), + "comments on a deleted issue mustn't be listed" + ); + assert_eq!( + index.social_latest_child(&repo, &parent, &comment_collection()), + Resolved::Ready(None), + "nor should the newest comment be listed" + ); + assert_eq!( + state_of(&index, &repo, comment_object), + Some(before), + "deleting the issue shouldn't change the comment's own record" + ); + assert_eq!( + index.social_object_at(&repo, &comment_at("3lubrptx57d33")), + Resolved::Ready(Some(comment_object)), + "the comment should still be findable by address" + ); +} + +#[test] +fn reaction_index_answers_live_reactions_only() { + let (world, repo, _issue) = kelp_issue(); + let parent = issue_at("3lubrptx57d22"); + let nels_like = world.reaction( + &repo, + "3lubrptx57d44", + &acc("nel"), + 1, + &parent, + ReactionKind::Like, + 20, + ); + let olarens_like = world.reaction( + &repo, + "3lubrptx57d55", + &acc("olaren"), + 2, + &parent, + ReactionKind::Like, + 30, + ); + let nels_heart = world.reaction( + &repo, + "3lubrptx57d66", + &acc("nel"), + 3, + &parent, + ReactionKind::Heart, + 40, + ); + let index = world.index(); + index.refresh_social(&repo).unwrap(); + let cold = world.index(); + cold.refresh_social(&repo).unwrap(); + + for (actor, kind, expected) in [ + (&acc("nel"), ReactionKind::Like, Some(nels_like)), + (&acc("olaren"), ReactionKind::Like, Some(olarens_like)), + (&acc("nel"), ReactionKind::Heart, Some(nels_heart)), + (&acc("olaren"), ReactionKind::Heart, None), + (&acc("nel"), ReactionKind::Eyes, None), + ] { + assert_eq!( + index.social_live_reaction(&repo, &parent, actor, kind), + Resolved::Ready(expected), + "existing reactions have to be found by actor, subject, & kind" + ); + assert_eq!( + index.social_live_reaction(&repo, &parent, actor, kind), + cold.social_live_reaction(&repo, &parent, actor, kind), + "an index rebuilt from scratch must match the incrementally updated index" + ); + } + + world.erase::( + &repo, + nels_like, + &acc("nel"), + version_of(&index, &repo, nels_like).expect("the reaction should have a version"), + 50, + ); + index.refresh_social(&repo).unwrap(); + assert_eq!( + index.social_live_reaction(&repo, &parent, &acc("nel"), ReactionKind::Like), + Resolved::Ready(None), + "a deleted reaction should probably stop being found" + ); + assert_eq!( + index.social_live_reaction(&repo, &parent, &acc("olaren"), ReactionKind::Like), + Resolved::Ready(Some(olarens_like)), + "another actor's same-kind reaction should still be found" + ); +} + #[test] fn untouched_repo_stays_warming_until_write_reaches_answer() { let (world, repo, object) = kelp_issue(); @@ -1857,7 +1998,7 @@ fn untouched_repo_stays_warming_until_write_reaches_answer() { world.append_issue( &repo, object, - &issue_erasing(&acc("nel"), note_record("kelp forest").1, 30), + &erased(&acc("nel"), note_record("kelp forest").1, at(30)), Record::Tombstone, 30, ); @@ -2033,7 +2174,7 @@ fn serving_bytes_answer_live_versions_and_erasure_frees_them() { world.append_issue( &repo, object, - &issue_erasing(&acc("nel"), note_record("kelp forest").1, 30), + &erased(&acc("nel"), note_record("kelp forest").1, at(30)), Record::Tombstone, 30, ); @@ -2108,7 +2249,17 @@ fn unprojected_cob_skips_without_failing_repository() { let repo = repo_did("squid"); world.seed_collaborator(&repo, "olaren"); let object = world.issue(&repo, "3lubrptx57d22", &acc("nel"), 1, "kelp", 10); - let (note, note_body) = opening("3lubrptx57d33", &acc("nel"), 1, None, "foreign", 11); + let (note, note_body) = contribution_of::( + "3lubrptx57d33", + &acc("nel"), + 1, + None, + NoteBody { + text: "foreign".to_owned(), + }, + "foreign", + at(11), + ); let foreign = world.open_cob::(&repo, ¬e, Record::version(note_body), 11); world.on_repo(&repo, foreign, &BadNote::Explode(0), 12); @@ -2355,7 +2506,7 @@ fn a_change_that_opens_no_object_still_costs_the_repository_its_bytes() { world.open_issue( &repo, - &issue_erasing(&acc("nel"), note_record("kelp").1, 20), + &erased(&acc("nel"), note_record("kelp").1, at(20)), Record::Tombstone, 20, ); diff --git a/knot2/crates/knot-xrpc/Cargo.toml b/knot2/crates/knot-xrpc/Cargo.toml index 965fd9fe5..989b93e1e 100644 --- a/knot2/crates/knot-xrpc/Cargo.toml +++ b/knot2/crates/knot-xrpc/Cargo.toml @@ -21,6 +21,7 @@ jacquard-common = { workspace = true } jacquard-repo = { workspace = true } cid = { workspace = true } serde_ipld_dagcbor = { workspace = true } +ipld-core = { workspace = true } serde_bytes = { workspace = true } knot-record = { workspace = true } knot-cob = { workspace = true } diff --git a/knot2/crates/knot-xrpc/src/atproto.rs b/knot2/crates/knot-xrpc/src/atproto.rs index ad4355e57..4e4692425 100644 --- a/knot2/crates/knot-xrpc/src/atproto.rs +++ b/knot2/crates/knot-xrpc/src/atproto.rs @@ -9,6 +9,7 @@ use axum::response::{IntoResponse, Response}; use bytes::Bytes; use cid::Cid; use futures::{StreamExt, TryStreamExt}; +use ipld_core::ipld::Ipld; use crate::query::ValidatedQuery; use jacquard_repo::RepoError; @@ -44,6 +45,38 @@ fn repo_not_found(did: &str) -> XrpcError { ) } +fn record_value(bytes: &[u8]) -> Result { + let decoded: Ipld = serde_ipld_dagcbor::from_slice(bytes) + .map_err(|error| XrpcError::internal(error.to_string()))?; + json_of(&decoded).map_err(XrpcError::internal) +} + +fn json_of(ipld: &Ipld) -> Result { + match ipld { + Ipld::Null => Ok(Value::Null), + Ipld::Bool(boolean) => Ok(Value::Bool(*boolean)), + Ipld::Integer(integer) => serde_json::Number::from_i128(*integer) + .map(Value::Number) + .ok_or_else(|| "an integer in the record is out of range for JSON".to_string()), + Ipld::Float(float) => serde_json::Number::from_f64(*float) + .map(Value::Number) + .ok_or_else(|| "a non-finite float in the record doesn't fit JSON".to_string()), + Ipld::String(string) => Ok(Value::String(string.clone())), + Ipld::Link(cid) => Ok(Value::String(cid.to_string())), + Ipld::Bytes(_) => Err("bytes in the record don't render as JSON".to_string()), + Ipld::List(items) => items + .iter() + .map(json_of) + .collect::, _>>() + .map(Value::Array), + Ipld::Map(fields) => fields + .iter() + .map(|(key, value)| json_of(value).map(|value| (key.clone(), value))) + .collect::, _>>() + .map(Value::Object), + } +} + fn record_not_found(uri: &str) -> XrpcError { XrpcError::named( StatusCode::BAD_REQUEST, @@ -566,8 +599,7 @@ pub(crate) async fn repo_get_record( ); record_not_found(&uri) })?; - let value: Value = serde_ipld_dagcbor::from_slice(&bytes) - .map_err(|error| XrpcError::internal(error.to_string()))?; + let value = record_value(&bytes)?; Result::<(Cid, Value), XrpcError>::Ok((cid, value)) })?; json( @@ -653,8 +685,7 @@ pub(crate) async fn repo_list_records( ); return Ok(None); }; - let value: Value = serde_ipld_dagcbor::from_slice(&bytes) - .map_err(|error| XrpcError::internal(error.to_string()))?; + let value = record_value(&bytes)?; Ok::<_, XrpcError>(Some((rkey, cid, value))) } }) diff --git a/knot2/crates/knot-xrpc/src/social.rs b/knot2/crates/knot-xrpc/src/social.rs index ecc75439e..122c14850 100644 --- a/knot2/crates/knot-xrpc/src/social.rs +++ b/knot2/crates/knot-xrpc/src/social.rs @@ -12,16 +12,21 @@ use knot_cob::{ Approved, ChangePayload, CobError, CobHome, CobStore, Materialized, Record, WriteContext, }; use knot_cobs::{ - Creation, Editing, Erasure, Issue, IssueChange, IssueCob, IssueStateChange, IssueStateCob, - SocialBody, SocialChange, SocialCob, Transition, + Comment, CommentChange, Creation, Editing, Erasure, Issue, IssueChange, IssueStateChange, + IssueStateCob, Reaction, ReactionKind, SocialBody, SocialChange, SocialCob, SocialObject, + Transition, }; use knot_git::Repo; use knot_index::{Resolved, SocialEntry}; use knot_record::chain::{self, RecordOp}; +use knot_record::comment::{ + CommentMarkup, CommentRecord, CommentRef, PullRoundIdx, comment_collection, +}; use knot_record::issue::{ IssueProse, IssueRecord, IssueState, IssueStateRecord, IssueTitle, MarkdownBody, issue_collection, issue_state_collection, }; +use knot_record::reaction::{ReactionRecord, reaction_collection}; use knot_resource::StructureBytes; use knot_runtime::{Clock, HttpTransport, Signer}; use knot_types::{ @@ -84,17 +89,14 @@ struct SwapInput { } impl SwapInput { - fn issue_edit_of( + fn edit_of( self, - on_state: &'static str, + editable: &RecordCollection, ) -> Result<(RepoDid, RecordRkey, RecordCid), XrpcError> { if self.swap_commit.is_some() { return Err(no_swap_commit()); } - if self.collection == issue_state_collection() { - return Err(XrpcError::invalid_request(on_state)); - } - if self.collection != issue_collection() { + if &self.collection != editable { return Err(rejected_collection(&self.collection)); } let from = self @@ -175,12 +177,7 @@ impl IssueDraft { collection: &RecordCollection, ) -> Result { kind_matches(self.kind.as_ref(), collection)?; - if self.repo != *written_to { - return Err(XrpcError::invalid_request(format!( - "the record's repo is {}, and the write is for {}", - self.repo, written_to - ))); - } + on_this_knot(&self.repo, written_to, "record")?; let unusable = |what: &'static str| { move |error| XrpcError::invalid_request(format!("{what} is unusable: {error}")) }; @@ -209,6 +206,151 @@ struct StateDraft { _created_at: serde::de::IgnoredAny, } +#[derive(Deserialize)] +struct StrongRefInput { + uri: AtUri, + cid: crate::out::CidString, +} + +impl StrongRefInput { + fn record(self, what: &'static str) -> Result { + self.cid.record().ok_or_else(|| { + XrpcError::invalid_request(format!( + "a record CID is dag-cbor sha2-256, and {what}'s strongRef cid isn't" + )) + }) + } +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct MarkupInput { + #[serde(rename = "$type", default)] + kind: Option, + text: String, + #[serde(default)] + original: Option, + #[serde(default)] + blobs: Option, +} + +impl MarkupInput { + fn parse(self) -> Result { + kind_matches( + self.kind.as_ref(), + &knot_record::comment::markup_collection(), + )?; + if self.blobs.is_some() { + return Err(XrpcError::invalid_request( + "blobs are attachments, and this knot won't take them; please leave blobs out", + )); + } + let unusable = |what: &'static str| { + move |error| { + XrpcError::invalid_request(format!("comment body's {what} is unusable: {error}")) + } + }; + let text = MarkdownBody::new(self.text).map_err(unusable("text"))?; + let original = self + .original + .map(MarkdownBody::new) + .transpose() + .map_err(unusable("original"))?; + Ok(CommentMarkup::new(text, original)) + } +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct CommentDraft { + #[serde(rename = "$type", default)] + kind: Option, + subject: StrongRefInput, + #[serde(rename = "replyTo", default)] + reply_to: Option, + body: MarkupInput, + #[serde(rename = "pullRoundIdx", default)] + pull_round_idx: Option, + #[serde(default)] + embed: Option, + #[serde(rename = "createdAt")] + _created_at: serde::de::IgnoredAny, +} + +struct ParsedComment { + subject: CommentRef, + reply_to: Option, + markup: CommentMarkup, +} + +impl CommentDraft { + fn parse(self, written_to: &RepoDid) -> Result { + kind_matches(self.kind.as_ref(), &comment_collection())?; + if self.embed.is_some() { + return Err(XrpcError::invalid_request( + "embed is an attachment this knot won't store; please leave embed out", + )); + } + let subject = any_target(&self.subject.uri)?; + if subject.address.collection() != &issue_collection() { + return Err(XrpcError::invalid_request(format!( + "this knot only takes issues as comment subjects for now; the uri's collection \ + is {}", + subject.address.collection() + ))); + } + on_this_knot(&subject.repo, written_to, "subject")?; + if self.pull_round_idx.is_some() { + return Err(XrpcError::invalid_request( + "pullRoundIdx belongs to a pull's submission round; this subject is an issue, \ + so please leave pullRoundIdx out", + )); + } + let reply_to = self + .reply_to + .map(|reply| -> Result { + let target = target(&reply.uri, &comment_collection())?; + on_this_knot(&target.repo, written_to, "reply")?; + Ok(CommentRef::new(target.address, reply.record("the reply")?)) + }) + .transpose()?; + Ok(ParsedComment { + subject: CommentRef::new(subject.address, self.subject.record("the subject")?), + reply_to, + markup: self.body.parse()?, + }) + } +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct ReactionDraft { + #[serde(rename = "$type", default)] + kind: Option, + subject: AtUri, + reaction: ReactionKind, + #[serde(rename = "createdAt")] + _created_at: serde::de::IgnoredAny, +} + +impl ReactionDraft { + fn parse(self, written_to: &RepoDid) -> Result<(RecordAddress, ReactionKind), XrpcError> { + kind_matches(self.kind.as_ref(), &reaction_collection())?; + let subject = any_target(&self.subject)?; + on_this_knot(&subject.repo, written_to, "subject")?; + Ok((subject.address, self.reaction)) + } +} + +fn on_this_knot(repo: &RepoDid, written_to: &RepoDid, what: &'static str) -> Result<(), XrpcError> { + if repo == written_to { + return Ok(()); + } + Err(XrpcError::invalid_request(format!( + "the write is for {written_to}, but the {what}'s repo is {repo}" + ))) +} + #[derive(Serialize)] enum ValidationStatus { #[serde(rename = "valid")] @@ -250,27 +392,20 @@ struct Target { } fn target(uri: &AtUri, collection: &RecordCollection) -> Result { - let repo = RepoDid::new(uri.authority().as_str()) - .map_err(|_| XrpcError::invalid_request("uri authority must be a repository DID"))?; - let from_uri = uri - .collection() - .ok_or_else(|| XrpcError::invalid_request("the uri is missing a collection"))?; - if from_uri.as_str() != collection.as_str() { + let target = any_target(uri)?; + if target.address.collection() != collection { return Err(XrpcError::invalid_request(format!( - "the uri's collection is {from_uri}, and this procedure takes {collection}" + "this procedure takes {collection}; the uri's collection is {}", + target.address.collection() ))); } - let rkey = uri - .rkey() - .ok_or_else(|| XrpcError::invalid_request("the uri is missing a record key")) - .and_then(|rkey| { - RecordRkey::new(rkey.as_ref()) - .map_err(|_| XrpcError::invalid_request("record key isn't a TID this knot minted")) - })?; - Ok(Target { - repo, - address: RecordAddress::new(collection.clone(), rkey), - }) + Ok(target) +} + +fn any_target(uri: &AtUri) -> Result { + let (repo, address) = RecordAddress::parse_at_uri(uri) + .map_err(|error| XrpcError::invalid_request(error.to_string()))?; + Ok(Target { repo, address }) } fn ready(resolved: Resolved, projection: &'static str) -> Result { @@ -484,10 +619,16 @@ impl Ledger<'_, H, C> { change: SocialChange, doing: Doing<'_>, ) -> Result<(CobId, ChangeId, CommitHead), XrpcError> { - let (record, op, amending, replaced) = match doing { + let (record, op, amending, replaced, verb) = match doing { Doing::Open { body } => { let (record, placed) = Self::placed(body); - (record, RecordOp::Create { record: placed }, None, None) + ( + record, + RecordOp::Create { record: placed }, + None, + None, + None, + ) } Doing::Edit { object, from, body } => { let (record, placed) = Self::placed(body); @@ -499,6 +640,7 @@ impl Ledger<'_, H, C> { }, Some(object), Some(object), + Some("edit"), ) } Doing::Erase { object, from } => ( @@ -506,6 +648,7 @@ impl Ledger<'_, H, C> { RecordOp::Delete { from }, Some(object), None, + Some("erase"), ), }; let adding = match &op { @@ -531,13 +674,23 @@ impl Ledger<'_, H, C> { .map(|created| (created.object, created.tip))?, Some(object) => ( object, - store.append_approved::>( - &CobHome::from(self.repo), - object, - &approved, - self.signer, - self.now, - )?, + store + .append_approved::>( + &CobHome::from(self.repo), + object, + &approved, + self.signer, + self.now, + ) + .map_err(|error| match (verb, error) { + (Some(verb), CobError::AppendRejected { .. }) => { + XrpcError::forbidden(format!( + "sorry, only its author or a moderator can {verb} this {}", + noun_of(address.collection()) + )) + } + (_, other) => other.into(), + })?, ), }; let head = self.settled(object, &SocialChange::::type_name(), || { @@ -558,7 +711,7 @@ fn tree_body( version: RecordCid, ) -> Result, XrpcError> { let type_name = knot_types::TypeName::new(collection.as_str()) - .map_err(|_| XrpcError::internal("a social collection is a type name"))?; + .expect("Record collection is a valid type name"); let delta = store.changes_of(&type_name, object, knot_cob::HistoryModel::Convergent, None)?; Ok(delta .changes @@ -757,16 +910,26 @@ pub(crate) async fn create_record( record, } = decode::(&body)?; if rkey.is_some() - && (collection == issue_state_collection() || collection == issue_collection()) + && [ + issue_state_collection(), + issue_collection(), + comment_collection(), + reaction_collection(), + ] + .contains(&collection) { return Err(XrpcError::invalid_request(format!( - "this knot mints the record key for {collection}; leave rkey out" + "this knot mints the record key for {collection}; please leave rkey out" ))); } if collection == issue_state_collection() { state_open(&state, &headers, &method, repo, record).await } else if collection == issue_collection() { issue_open(&state, &headers, &method, repo, record).await + } else if collection == comment_collection() { + comment_open(&state, &headers, &method, repo, record).await + } else if collection == reaction_collection() { + reaction_open(&state, &headers, &method, repo, record).await } else { Err(rejected_collection(&collection)) } @@ -785,12 +948,16 @@ where B: SocialBody + Send + 'static, H: HttpTransport, C: Clock, - F: FnOnce(&Ledger<'_, H, C>, &WriteContext<'_>) -> Result<(), XrpcError> + Send + 'static, + F: FnOnce(&Ledger<'_, H, C>, &WriteContext<'_>) -> Result, XrpcError> + + Send + + 'static, { let produces = record.cid(); let address = RecordAddress::new(collection.clone(), state.rkeys.mint()); on_repo(state, session, move |ledger, cx| { - admit(ledger, cx)?; + if let Some(answered) = admit(ledger, cx)? { + return Ok(answered); + } let change = SocialChange::Open { creation: Creation { address: address.clone(), @@ -826,29 +993,234 @@ async fn issue_open( )? .body()?; open(state, session, collection, None, record, Issue, |_, _| { - Ok(()) + Ok(None) }) .await } +async fn comment_open( + state: &Arc>, + headers: &HeaderMap, + method: &crate::Method, + repo: RepoDid, + record: serde_json::Value, +) -> Result { + let collection = comment_collection(); + let draft = typed::(record, &collection)?.parse(&repo)?; + let session = session(state, headers, method, repo).await?; + let subject = draft.subject.clone(); + let reply = draft.reply_to.clone(); + let record = CommentRecord::new( + session.repo.clone(), + draft.subject, + draft.markup, + draft.reply_to, + None, + session.now, + session.actor.clone(), + )? + .body()?; + let body = Comment { + subject: subject.cid(), + reply: reply.clone(), + }; + open( + state, + session, + comment_collection(), + Some(subject.address().clone()), + record, + body, + move |ledger, _cx| { + standing_version(ledger, &subject, "subject")?; + reply + .as_ref() + .map(|reply| standing_version(ledger, reply, "reply")) + .transpose()?; + Ok(None) + }, + ) + .await +} + +fn standing_version( + ledger: &Ledger<'_, H, C>, + reference: &CommentRef, + what: &'static str, +) -> Result<(), XrpcError> { + let serving = live_record_at(ledger, reference.address())? + .version() + .ok_or_else(|| XrpcError::internal(format!("the current {what} is missing its version")))?; + if serving != reference.cid() { + return Err(XrpcError::invalid_request(format!( + "the {what}'s strongRef cid is {}, but the record at {} now has version {serving}", + reference.cid(), + reference.address().at_uri(ledger.repo), + ))); + } + Ok(()) +} + +async fn reaction_open( + state: &Arc>, + headers: &HeaderMap, + method: &crate::Method, + repo: RepoDid, + record: serde_json::Value, +) -> Result { + let (subject, reaction) = + typed::(record, &reaction_collection())?.parse(&repo)?; + let session = session(state, headers, method, repo).await?; + let record = ReactionRecord::new( + session.repo.clone(), + subject.clone(), + reaction, + session.now, + session.actor.clone(), + )? + .body()?; + open( + state, + session, + reaction_collection(), + Some(subject.clone()), + record, + Reaction { kind: reaction }, + move |ledger, cx| { + live_record_at(ledger, &subject)?; + if !cx.permission().contributes().is_allowed() { + return Ok(None); + } + if let Some(object) = ready( + ledger.state.index.social_live_reaction( + ledger.repo, + &subject, + cx.actor(), + reaction, + ), + "social", + )? { + let address = ready( + ledger.state.index.social_address_of(ledger.repo, object), + "social", + )? + .ok_or_else(|| { + XrpcError::internal("the matching reaction is missing its address") + })?; + let version = live_record_at(ledger, &address)?.version().ok_or_else(|| { + XrpcError::internal("the matching reaction should have a version and doesn't") + })?; + return Ok(Some(wrote(ledger.repo, &address, version, &ledger.head()?))); + } + Ok(None) + }, + ) + .await +} +fn noun_of(collection: &RecordCollection) -> &'static str { + match collection.as_str() { + knot_record::issue::ISSUE_COLLECTION => "issue", + knot_record::comment::COMMENT_COLLECTION => "comment", + knot_record::reaction::REACTION_COLLECTION => "reaction", + _ => "record", + } +} + fn standing( ledger: &Ledger<'_, H, C>, address: &RecordAddress, -) -> Result<(CobId, knot_cobs::SocialObject), XrpcError> { +) -> Result<(CobId, SocialObject), XrpcError> { let object = object_at(ledger.state, ledger.repo, address)?; - let (projected, _) = CobStore::new(ledger.git).materialize::(object)?; - projected + let live = serving_object(ledger, address, object)? + .ok_or_else(|| deleted(ledger, address, XrpcError::not_found))?; + Ok((object, live)) +} + +fn describe(reply: Option<&CommentRef>, repo: &RepoDid) -> String { + reply.map_or_else( + || "without a reply".to_string(), + |reply| format!("{} at {}", reply.cid(), reply.address().at_uri(repo)), + ) +} + +fn live_record_at( + ledger: &Ledger<'_, H, C>, + address: &RecordAddress, +) -> Result { + let object = ready( + ledger.state.index.social_object_at(ledger.repo, address), + "social", + )? + .ok_or_else(|| { + XrpcError::invalid_request(format!( + "{} isn't a record on this knot", + address.at_uri(ledger.repo) + )) + })?; + serving_object(ledger, address, object)? + .ok_or_else(|| deleted(ledger, address, XrpcError::invalid_request)) +} + +fn deleted( + ledger: &Ledger<'_, H, C>, + address: &RecordAddress, + fail: fn(String) -> XrpcError, +) -> XrpcError { + fail(format!( + "{} at {} was deleted", + noun_of(address.collection()), + address.at_uri(ledger.repo) + )) +} + +fn serving_object( + ledger: &Ledger<'_, H, C>, + address: &RecordAddress, + object: CobId, +) -> Result, XrpcError> { + Ok(materialized(ledger, address, object)? + .0 .object() .filter(|live| !live.is_erased()) - .cloned() - .map(|live| (object, live)) + .cloned()) +} + +fn materialized( + ledger: &Ledger<'_, H, C>, + address: &RecordAddress, + object: CobId, +) -> Result<(knot_cobs::SocialState, ChangeId), XrpcError> { + let type_name = TypeName::new(address.collection().as_str()) + .expect("Record collection is a valid type name"); + knot_cobs::social_state(&CobStore::new(ledger.git), &type_name, object) + .map_err(XrpcError::from)? .ok_or_else(|| { - XrpcError::not_found(format!( - "issue at {} was deleted", - address.at_uri(ledger.repo) + XrpcError::internal(format!( + "this knot didn't register a builder for {type_name}" )) }) } + +fn opened_comment( + ledger: &Ledger<'_, H, C>, + object: CobId, +) -> Result { + let type_name = TypeName::new(knot_record::comment::COMMENT_COLLECTION) + .expect("Comment collection is a valid type name"); + let root = CobStore::new(ledger.git) + .changes_of(&type_name, object, knot_cob::HistoryModel::Convergent, None)? + .changes + .into_iter() + .next() + .ok_or_else(|| XrpcError::internal("a comment object's root change went missing"))?; + match CommentChange::decode(root.payload()) { + Ok(SocialChange::Open { body, .. }) => Ok(body), + Ok(_) => Err(XrpcError::internal( + "a comment object's root change isn't the change that created it", + )), + Err(error) => Err(XrpcError::internal(error.to_string())), + } +} pub(crate) async fn put_record( State(state): State>>, headers: HeaderMap, @@ -856,12 +1228,37 @@ pub(crate) async fn put_record( body: Bytes, ) -> Result { let PutRecordInput { swap, record } = decode::(&body)?; - let (repo, rkey, from) = swap.issue_edit_of("a state transition is created, never edited")?; + let SwapInput { collection, .. } = &swap; + match collection.as_str() { + knot_record::issue::ISSUE_STATE_COLLECTION => Err(XrpcError::invalid_request( + "a state transition is created, never edited", + )), + knot_record::reaction::REACTION_COLLECTION => Err(XrpcError::invalid_request( + "createRecord makes a reaction and deleteRecord renoves it; a reaction is never edited", + )), + knot_record::issue::ISSUE_COLLECTION => { + issue_put(&state, &headers, &method, swap, record).await + } + knot_record::comment::COMMENT_COLLECTION => { + comment_put(&state, &headers, &method, swap, record).await + } + _ => Err(rejected_collection(&swap.collection)), + } +} + +async fn issue_put( + state: &Arc>, + headers: &HeaderMap, + method: &crate::Method, + swap: SwapInput, + record: serde_json::Value, +) -> Result { + let (repo, rkey, from) = swap.edit_of(&issue_collection())?; let collection = issue_collection(); let prose = typed::(record, &collection)?.parse(&repo, &collection)?; let address = RecordAddress::new(collection, rkey); - let session = session(&state, &headers, &method, repo).await?; - on_repo(&state, session, move |ledger, cx| { + let session = session(state, headers, method, repo).await?; + on_repo(state, session, move |ledger, cx| { let (object, live) = standing(ledger, &address)?; let record = IssueRecord::new( ledger.repo.clone(), @@ -898,6 +1295,83 @@ pub(crate) async fn put_record( .await } +async fn comment_put( + state: &Arc>, + headers: &HeaderMap, + method: &crate::Method, + swap: SwapInput, + record: serde_json::Value, +) -> Result { + let (repo, rkey, from) = swap.edit_of(&comment_collection())?; + let collection = comment_collection(); + let draft = typed::(record, &collection)?.parse(&repo)?; + let address = RecordAddress::new(collection, rkey); + let session = session(state, headers, method, repo).await?; + on_repo(state, session, move |ledger, cx| { + let (object, live) = standing(ledger, &address)?; + if live.parent() != Some(draft.subject.address()) { + let opened_on = live.parent().map_or_else( + || "wasn't created on a subject".to_string(), + |parent| format!("was created on {}", parent.at_uri(ledger.repo)), + ); + return Err(XrpcError::invalid_request(format!( + "a comment's subject is set at creation: {} {}, and the edit's subject is {}", + address.at_uri(ledger.repo), + opened_on, + draft.subject.address().at_uri(ledger.repo), + ))); + } + let opened = opened_comment(ledger, object)?; + if opened.subject != draft.subject.cid() { + return Err(XrpcError::invalid_request(format!( + "the edit's subject is {}, but this comment was created on {}; subject is set \ + at creation", + draft.subject.cid(), + opened.subject, + ))); + } + if opened.reply != draft.reply_to { + return Err(XrpcError::invalid_request(format!( + "reply is set at creation: the comment was created {}, the edit's reply is {}", + describe(opened.reply.as_ref(), ledger.repo), + describe(draft.reply_to.as_ref(), ledger.repo), + ))); + } + let record = CommentRecord::new( + ledger.repo.clone(), + draft.subject, + draft.markup, + draft.reply_to, + None, + live.created_at(), + cx.actor().clone(), + )? + .body()?; + let produces = record.cid(); + let change: knot_cobs::CommentChange = SocialChange::Edit { + editing: Editing { + author: cx.actor().clone(), + from, + produces, + edited_at: ledger.now, + }, + body: opened, + }; + let (_, _, head) = ledger.write( + cx, + &address, + change, + Doing::Edit { + object, + from, + body: &record, + }, + )?; + Ok(wrote(ledger.repo, &address, produces, &head)) + }) + .await +} + pub(crate) async fn delete_record( State(state): State>>, headers: HeaderMap, @@ -905,14 +1379,42 @@ pub(crate) async fn delete_record( body: Bytes, ) -> Result { let swap = decode::(&body)?; - let (repo, rkey, from) = swap.issue_edit_of( - "a state transition is created, never erased; erasing the issue erases the history", - )?; - let address = RecordAddress::new(issue_collection(), rkey); - let session = session(&state, &headers, &method, repo).await?; - on_repo(&state, session, move |ledger, cx| { + let SwapInput { collection, .. } = &swap; + match collection.as_str() { + knot_record::issue::ISSUE_STATE_COLLECTION => Err(XrpcError::invalid_request( + "a state transition is created, never erased; erasing the issue erases the history", + )), + knot_record::issue::ISSUE_COLLECTION => { + erase_of::(&state, &headers, &method, swap).await + } + knot_record::comment::COMMENT_COLLECTION => { + erase_of::(&state, &headers, &method, swap).await + } + knot_record::reaction::REACTION_COLLECTION => { + erase_of::(&state, &headers, &method, swap).await + } + _ => Err(rejected_collection(&swap.collection)), + } +} + +async fn erase_of( + state: &Arc>, + headers: &HeaderMap, + method: &crate::Method, + swap: SwapInput, +) -> Result +where + B: SocialBody + Send + 'static, + H: HttpTransport, + C: Clock, +{ + let collection = SocialCob::::collection(); + let (repo, rkey, from) = swap.edit_of(&collection)?; + let address = RecordAddress::new(collection, rkey); + let session = session(state, headers, method, repo).await?; + on_repo(state, session, move |ledger, cx| { let (object, _) = standing(ledger, &address)?; - let change: IssueChange = SocialChange::Erase { + let change: SocialChange = SocialChange::Erase { erasure: Erasure { author: cx.actor().clone(), from, @@ -1011,7 +1513,7 @@ async fn state_open( "issue at {issue} is already {wanted}" ))); } - Ok(()) + Ok(None) }, ) .await @@ -1053,21 +1555,13 @@ pub(crate) async fn list_cob_versions( State(state): State>>, ValidatedQuery(query): ValidatedQuery, ) -> Result { - let uri = query.uri; - let collection = uri - .collection() - .ok_or_else(|| XrpcError::invalid_request("the uri is missing a collection")) - .and_then(|collection| { - RecordCollection::new(collection.as_str()).map_err(|_| { - XrpcError::invalid_request("the uri's collection isn't a record collection") - }) - })?; + let Target { repo, address } = any_target(&query.uri)?; + let collection = address.collection().clone(); if !knot_cobs::social_collections().contains(&collection) { return Err(XrpcError::invalid_request(format!( "{collection} isn't a collection this knot keeps version chains for" ))); } - let Target { repo, address } = target(&uri, &collection)?; ready(state.index.ownership_of(&repo), "registry")? .ok_or_else(|| XrpcError::not_found("no such repository on this knot"))?; warm_social(&state, &repo).await?;