use knot_cob::{
Append, Approved, Authorize, ChangePayload, CobError, CobHome, CobStore, Materialized, Record,
WriteContext,
};
use knot_cobs::{
Creation, Erasure, Opening, SocialBody, SocialChange, SocialCob, SocialObject, Ticket,
};
use knot_index::Resolved;
use knot_record::Prose;
use knot_record::attachment::Attachment;
use knot_record::label::LabelToken;
use knot_record::ticket::{TargetBranch, TicketRecord, TicketState};
use knot_types::{
AccountDid, BlobCid, ChangeId, CobId, ContributionPermission, Oid, RecordAddress, RecordBody,
RecordCollection, RecordRkey, RepoDid, SubjectNumber, TypeName, UnixSeconds,
};
use super::drafts::SwapInput;
use super::error::WriteError;
use super::session::wrote;
use super::{COLLABORATE_REMEDY, Done, Ledger, NOT_LABEL_AUTHORITY, Written};
pub(super) fn ready<T>(resolved: Resolved<T>, projection: &'static str) -> Result<T, WriteError> {
match resolved {
Resolved::Ready(value) => Ok(value),
Resolved::Warming => Err(WriteError::warming(format!(
"{projection} projection is still warming, please retry shortly"
))),
}
}
pub(super) fn collection_type(collection: &RecordCollection) -> TypeName {
TypeName::new(collection.as_str()).expect("Record collection is a valid type name")
}
pub(super) fn noun_of(collection: &RecordCollection) -> &'static str {
use knot_record::{artifact, comment, grant, label, manifest, pull, reaction, ticket};
match collection.as_str() {
ticket::TICKET_COLLECTION => "ticket",
comment::COMMENT_COLLECTION => "comment",
reaction::REACTION_COLLECTION => "reaction",
label::LABEL_DEF_COLLECTION => "label def",
pull::PATCH_COLLECTION => "patch",
grant::MEMBER_COLLECTION => "member",
grant::COLLABORATOR_COLLECTION => "collaborator",
grant::BLOCK_COLLECTION => "block",
manifest::MANIFEST_COLLECTION => "manifest",
artifact::ARTIFACT_COLLECTION => "artifact",
_ => "record",
}
}
pub(super) fn rejected(error: CobError) -> WriteError {
match error {
CobError::CreateRejected { permission, .. }
if !permission.policy().allows(permission.role()) =>
{
let sentence = error.to_string();
if permission.moderates().is_allowed() {
WriteError::forbidden(sentence)
} else {
WriteError::forbidden(format!("{sentence}. {COLLABORATE_REMEDY}"))
}
}
other => other.into(),
}
}
pub(super) fn append_denial(
opening: Opening,
verb: &str,
collection: &RecordCollection,
permission: ContributionPermission,
) -> WriteError {
match opening {
Opening::Ownership => WriteError::forbidden(NOT_LABEL_AUTHORITY),
Opening::Contribution if verb == "erase" => WriteError::forbidden(format!(
"Sorry, only a collaborator can erase this {}",
noun_of(collection)
)),
_ if !permission.contributes().is_allowed() => WriteError::forbidden(format!(
"Sorry, you're {permission}, so you can't {verb} this {}; {COLLABORATE_REMEDY}",
noun_of(collection)
)),
_ => WriteError::forbidden(format!(
"Sorry, only the author or a moderator can {verb} this {}",
noun_of(collection)
)),
}
}
pub(crate) fn keep_ref(
record_did: &RepoDid,
address: &RecordAddress,
commit: Oid,
) -> Result<knot_types::RefName, WriteError> {
let encode = |raw: &str| {
raw.replace('%', "%25")
.replace(':', "%3a")
.replace('~', "%7e")
};
knot_types::RefName::new(format!(
"refs/tngl/keep/{}/{}/{}/{}",
encode(record_did.as_str()),
address.collection().as_str(),
encode(address.rkey().as_str()),
commit,
))
.map_err(|_| WriteError::invalid_request("record doesn't map to a usable ref name"))
}
pub(super) enum Doing<'a> {
Open { body: &'a RecordBody },
Edit { object: CobId, body: &'a RecordBody },
Erase { object: CobId },
}
pub(super) fn decode<T: serde::de::DeserializeOwned>(body: &[u8]) -> Result<T, WriteError> {
serde_json::from_slice(body).map_err(|error| WriteError::invalid_request(error.to_string()))
}
#[allow(clippy::too_many_arguments)]
pub(super) fn ticket_body_of(
prose: Prose,
state: TicketState,
tangled_labels: Option<Vec<LabelToken>>,
assignees: Option<Vec<AccountDid>>,
target_branch: Option<TargetBranch>,
blobs: Option<Vec<knot_record::attachment::Attachment>>,
created_at: UnixSeconds,
editor: AccountDid,
) -> Result<(RecordBody, Ticket), WriteError> {
let keys = tangled_labels
.as_ref()
.map(|tokens| tokens.iter().map(|token| token.key().clone()).collect());
let record = TicketRecord::new(
prose,
state,
tangled_labels,
assignees.clone(),
target_branch,
blobs,
created_at,
editor,
)?
.body()?;
Ok((
record,
Ticket {
state,
tangled_labels: keys,
assignees,
},
))
}
impl Ledger<'_> {
pub(super) fn next_number(
&self,
collection: &RecordCollection,
) -> Result<SubjectNumber, WriteError> {
ready(
self.index.next_subject_number(self.repo, collection),
"social",
)?
.ok_or_else(|| WriteError::internal("This repository has numbered past what fits"))
}
pub(super) fn object_at(&self, address: &RecordAddress) -> Result<CobId, WriteError> {
ready(self.index.social_object_at(self.repo, address), "social")?.ok_or_else(|| {
WriteError::not_found(format!("no record at {}", address.at_uri(self.repo)))
})
}
pub(super) fn write<B: SocialBody>(
&self,
cx: &WriteContext<'_>,
address: &RecordAddress,
change: SocialChange<B>,
doing: Doing<'_>,
) -> Result<(CobId, ChangeId), WriteError> {
let cited = self.cited.take();
if !cited.is_empty() {
self.authorized(cx, address, &change, &doing)?;
cited
.into_iter()
.try_for_each(|blob| self.admit_blob(cx, blob))?;
}
let (record, amending, verb) = match doing {
Doing::Open { body } => (Record::version(body.clone()), None, None),
Doing::Edit { object, body, .. } => {
(Record::version(body.clone()), Some(object), Some("edit"))
}
Doing::Erase { object, .. } => (Record::Tombstone, Some(object), Some("erase")),
};
let approved = Approved::signed_by(cx, Materialized { change, record })?;
let store = CobStore::new(self.git);
let (object, tip) = match amending {
None => store
.create_approved::<SocialCob<B>>(
&CobHome::from(self.repo),
&approved,
self.signer,
self.now,
)
.map_err(rejected)
.map(|created| (created.object, created.tip))?,
Some(object) => (
object,
store
.append_approved::<SocialCob<B>>(
&CobHome::from(self.repo),
object,
&approved,
self.signer,
self.now,
)
.map_err(|error| match (verb, error) {
(Some(verb), CobError::AppendRejected { permission, .. }) => {
append_denial(B::OPENING, verb, address.collection(), permission)
}
(_, other) => other.into(),
})?,
),
};
self.index
.refresh_social_object(self.repo, &SocialChange::<B>::type_name(), object)?;
Ok((object, tip))
}
pub(super) fn open_at<B: SocialBody>(
&self,
cx: &WriteContext<'_>,
address: &RecordAddress,
number: SubjectNumber,
parent: Option<RecordAddress>,
record: &RecordBody,
body: B,
) -> Result<Written, WriteError> {
let produces = record.cid();
let change = SocialChange::Open {
creation: Creation {
address: address.clone(),
author: cx.actor().clone(),
number,
parent,
created_at: self.now,
produces,
},
body,
};
let (_, tip) = self.write(cx, address, change, Doing::Open { body: record })?;
Ok(Written {
change: Some(tip),
..wrote(self.repo, address, produces)
})
}
pub(super) fn open_keyed<B: SocialBody>(
&self,
cx: &WriteContext<'_>,
collection: &RecordCollection,
key: impl FnOnce(SubjectNumber) -> Result<RecordAddress, WriteError>,
parent: Option<RecordAddress>,
record: &RecordBody,
body: B,
) -> Result<Written, WriteError> {
let number = self.next_number(collection)?;
let address = key(number)?;
self.open_at(cx, &address, number, parent, record, body)
}
pub(super) fn live_record_at(
&self,
address: &RecordAddress,
) -> Result<SocialObject, WriteError> {
let object =
ready(self.index.social_object_at(self.repo, address), "social")?.ok_or_else(|| {
WriteError::invalid_request(format!(
"{} isn't a record on this knot",
address.at_uri(self.repo)
))
})?;
self.serving_object(address, object)?
.ok_or_else(|| self.deleted(address, WriteError::invalid_request))
}
pub(super) fn deleted(
&self,
address: &RecordAddress,
fail: fn(String) -> WriteError,
) -> WriteError {
fail(format!(
"{} at {} was deleted",
noun_of(address.collection()),
address.at_uri(self.repo)
))
}
pub(super) fn serving_object(
&self,
address: &RecordAddress,
object: CobId,
) -> Result<Option<SocialObject>, WriteError> {
let type_name = collection_type(address.collection());
let (state, _) = knot_cobs::social_state(&CobStore::new(self.git), &type_name, object)?
.ok_or_else(|| {
WriteError::internal(format!(
"this knot didn't register a builder for {type_name}"
))
})?;
Ok(state.object().filter(|live| !live.is_erased()).cloned())
}
pub(super) fn standing(
&self,
address: &RecordAddress,
) -> Result<(CobId, SocialObject), WriteError> {
let object = self.object_at(address)?;
let live = self
.serving_object(address, object)?
.ok_or_else(|| self.deleted(address, WriteError::not_found))?;
Ok((object, live))
}
pub(super) fn standing_version(
&self,
reference: &knot_record::comment::CommentRef,
what: &'static str,
) -> Result<(), WriteError> {
let serving = self
.live_record_at(reference.address())?
.version()
.ok_or_else(|| {
WriteError::internal(format!("The current {what} is missing its version"))
})?;
if serving != reference.cid() {
return Err(WriteError::invalid_request(format!(
"The {what}'s strongRef CID is {}, but the record at {} now has version {serving}",
reference.cid(),
reference.address().at_uri(self.repo),
)));
}
Ok(())
}
pub(super) fn change_of<B: SocialBody>(
&self,
object: CobId,
missing: &'static str,
root: bool,
) -> Result<SocialChange<B>, WriteError> {
let type_name = TypeName::new(B::COLLECTION).expect("collection is a valid type name");
let mut changes = CobStore::new(self.git)
.changes_of(&type_name, object, knot_cob::HistoryModel::Convergent, None)?
.changes
.into_iter();
let change = if root { changes.next() } else { changes.last() }
.ok_or_else(|| WriteError::internal(missing))?;
<SocialChange<B> as ChangePayload>::decode(change.payload())
.map_err(|error| WriteError::internal(error.to_string()))
}
pub(super) fn mint(&self) -> RecordRkey {
RecordRkey::minted(jacquard_common::types::tid::Ticker::new().next(None))
}
pub(super) fn cite_attachments(&self, attachments: &[Attachment]) {
for blob in attachments {
self.cite(blob.cid(), blob.mime().as_str(), blob.size().get());
}
}
pub(super) fn cite(&self, cid: BlobCid, mime: &str, size: u64) {
self.cited.borrow_mut().push((cid, mime.to_owned(), size));
}
/// runs before the blob checks. a refused writer can't restart a blob's grace window.
fn authorized<B: SocialBody>(
&self,
cx: &WriteContext<'_>,
address: &RecordAddress,
change: &SocialChange<B>,
doing: &Doing<'_>,
) -> Result<(), WriteError> {
match doing {
Doing::Open { .. } if !SocialCob::<B>::may_create(cx).is_allowed() => {
Err(rejected(CobError::CreateRejected {
actor: cx.actor().clone(),
type_name: SocialChange::<B>::type_name(),
permission: cx.permission(),
}))
}
Doing::Edit { object, .. } => {
let (state, _) = knot_cobs::social_state(
&CobStore::new(self.git),
&collection_type(address.collection()),
*object,
)?
.ok_or_else(|| WriteError::internal("this knot didn't register a builder"))?;
match SocialCob::<B>::may_append(cx, &state, change) {
Append::Deny => Err(append_denial(
B::OPENING,
"edit",
address.collection(),
cx.permission(),
)),
_ => Ok(()),
}
}
_ => Ok(()),
}
}
fn admit_blob(
&self,
cx: &WriteContext<'_>,
(cid, mime, size): (BlobCid, String, u64),
) -> Result<(), WriteError> {
let io = |error: std::io::Error| WriteError::internal(error.to_string());
match self.blobs.size(&cid).map_err(io)? {
Some(stored) if stored == size => (),
Some(stored) => {
return Err(WriteError::invalid_request(format!(
"The blob {cid} is stored as {stored} bytes, but the record claims {size}. \
Please re-upload it and copy the size `uploadBlob` returned, thanks."
)));
}
None => {
return Err(WriteError::invalid_request(format!(
"The blob {cid} isn't in this repo's blob store. Please upload it with \
`com.atproto.repo.uploadBlob` before the record that references it, thanks."
)));
}
}
if let Some(reads) = self.blobs.sniffed(&cid).filter(|reads| *reads != mime) {
return Err(WriteError::invalid_request(format!(
"The blob {cid} reads as {reads} on this knot, but the record claims {mime}. \
Please use the `mimeType` from the upload, thanks."
)));
}
if cx.permission().contributes().is_allowed() {
self.blobs.touch(&cid).map_err(io)?;
}
Ok(())
}
pub(super) fn edit<B: SocialBody>(
&self,
cx: &WriteContext<'_>,
address: &RecordAddress,
change: SocialChange<B>,
object: CobId,
record: &RecordBody,
) -> Result<Written, WriteError> {
let (_, tip) = self.write(
cx,
address,
change,
Doing::Edit {
object,
body: record,
},
)?;
Ok(Written {
change: Some(tip),
..wrote(self.repo, address, record.cid())
})
}
pub(super) fn erase_of<B: SocialBody>(
&self,
cx: &WriteContext<'_>,
swap: SwapInput,
) -> Result<Done, WriteError> {
let collection = SocialCob::<B>::collection();
let (_, rkey, from) = swap.edit_of(&collection)?;
let address = RecordAddress::new(collection, rkey);
let (object, _) = self.standing(&address)?;
let change: SocialChange<B> = SocialChange::Erase {
erasure: Erasure {
author: cx.actor().clone(),
from,
erased_at: self.now,
},
};
self.write(cx, &address, change, Doing::Erase { object })?;
Ok(Done::Deleted)
}
}