use std::collections::{BTreeMap, BTreeSet};
use std::path::PathBuf;
use knot_cob::{Change, CobStore, Record};
use knot_cobs::Social as _;
use knot_git::Repo;
use knot_index::{Index, Resolved, SocialEntryState};
use knot_record::grant::{self, GrantCollection};
use knot_types::{
AccountDid, ChangeId, Datetime, MstKey, RecordBody, RecordCid, RepoDid, TypeName,
};
use crate::datastore::keys::{CobStateKey, RepoKey};
use crate::datastore::{Datastore, schema};
use crate::repolock::RepoLock;
use crate::store::RepoStore;
use super::CobState;
use super::chain::short_hash;
/// bump this when the replay changes, or the old cached states keep getting served.
const COB_STATE_VERSION: u32 = 1;
pub(crate) const COB_KIND: &str = "atproto-cob";
const COBS_KIND: &str = "atproto-cobs";
/// the index a write kept current, and the cob tips it was current at
pub(crate) struct SyncedIndex {
pub index: Index,
pub repo: RepoDid,
pub tips: BTreeMap<String, String>,
}
pub(crate) fn cob_tips(source: &Repo) -> anyhow::Result<BTreeMap<String, String>> {
Ok(source
.references()?
.into_iter()
.filter(|reference| reference.name.as_str().starts_with(knot_cob::COBS_PREFIX))
.map(|reference| {
(
reference.name.as_str().to_string(),
reference.target.to_hex(),
)
})
.collect())
}
/// a cob ref moving after the tips are read would cache a state the knot never had.
pub fn cob_state(
index: &Index,
source: &Repo,
did: &RepoDid,
cache: &RepoStore,
records: &Datastore,
lock: &RepoLock,
) -> anyhow::Result<CobState> {
cob_state_synced(index, source, did, cache, records, lock, None)
}
/// an index synced at the very tips read here already holds the repo's social state.
pub(crate) fn cob_state_synced(
index: &Index,
source: &Repo,
did: &RepoDid,
cache: &RepoStore,
records: &Datastore,
lock: &RepoLock,
synced: Option<&BTreeMap<String, String>>,
) -> anyhow::Result<CobState> {
let repo = RepoKey::of(did);
anyhow::ensure!(
lock.key() == repo.as_str(),
"cob state of {did} read under the lock of {}",
lock.key()
);
let tips = cob_tips(source)?;
let key = CobStateKey::new(&format!(
"{COB_STATE_VERSION}-{}",
short_hash(&serde_json::to_vec(&tips)?)
))?;
if let Some(state) = cache
.read_file::<StoredState>(COBS_KIND, key.as_str())
.and_then(|stored| stored.into_state())
{
record_cob_state(records, &repo, &key)?;
return Ok(state);
}
if synced != Some(&tips) {
index.refresh_social(did)?;
}
index.refresh_collaborators(did)?;
let cobs = CobStore::new(source);
let mut state = CobState::default();
let mut cob_keys = BTreeSet::new();
for entry in ready(index.social_entries(did))? {
if matches!(entry.state, SocialEntryState::Contested) {
continue;
}
let type_name = TypeName::new(entry.address.collection().as_str())?;
let tip = knot_cob::cob_ref_name(&type_name, entry.object)
.ok()
.and_then(|name| tips.get(name.as_str()));
let cob_key = tip.map(|tip| {
let serving = entry.state.serving().map(|cid| cid.cid().to_string());
format!(
"{COB_STATE_VERSION}-{type_name}-{}-{tip}-{}",
entry.object.oid().to_hex(),
serving.as_deref().unwrap_or("none")
)
});
let cached = cob_key
.as_deref()
.and_then(|cob_key| cache.read_file::<StoredCob>(COB_KIND, cob_key))
.and_then(StoredCob::into_replayed);
let replayed = match cached {
Some(replayed) => replayed,
None => {
let Some(replayed) = replay(&cobs, &type_name, &entry, did)? else {
continue;
};
if let Some(cob_key) = &cob_key {
cache.publish_file(COB_KIND, cob_key, &StoredCob::of(&replayed))?;
}
replayed
}
};
cob_keys.extend(cob_key);
let key = entry.address.mst_key();
if let Some(versions) = replayed.versions {
state.versions.insert(key.as_str().to_string(), versions);
}
if entry.state.serving().is_some() {
state.records.insert(key, replayed.body);
}
}
let grants = ready(index.collaborator_entries(did))?;
let grants = grant::bodies(
grants
.iter()
.map(|(subject, entry)| (subject, entry.created_at)),
GrantCollection::Collaborator,
);
state
.records
.extend(grants.into_iter().map(|(key, body)| (key, Some(body))));
cache.publish_file(COBS_KIND, key.as_str(), &StoredState::of(&state, cob_keys))?;
record_cob_state(records, &repo, &key)?;
Ok(state)
}
/// the recorded key decides the current state, since a clock step can make mtimes wrong.
pub fn record_cob_state(
records: &Datastore,
repo: &RepoKey,
key: &CobStateKey,
) -> anyhow::Result<()> {
let record = records.record::<schema::CobStatePointer>(repo);
if record.read()?.as_ref() == Some(key) {
return Ok(());
}
record.put(key)
}
/// with no pointer yet, gc keeps every cached state.
pub fn kept_cob_states(
cache: &RepoStore,
records: &Datastore,
repo: &RepoKey,
) -> Option<BTreeSet<PathBuf>> {
let key = records
.record::<schema::CobStatePointer>(repo)
.read()
.ok()??;
let stored = cache.read_file::<StoredState>(COBS_KIND, key.as_str())?;
let mut kept: BTreeSet<PathBuf> = stored
.cobs?
.iter()
.map(|cob_key| cache.file_path(COB_KIND, cob_key))
.collect();
kept.insert(cache.file_path(COBS_KIND, key.as_str()));
Some(kept)
}
pub fn cob_state_dirs(store: &RepoStore) -> [PathBuf; 2] {
[store.kind_dir(COB_KIND), store.kind_dir(COBS_KIND)]
}
struct Replayed {
versions: Option<Vec<Version>>,
body: Option<RecordBody>,
}
fn replay(
cobs: &CobStore,
type_name: &TypeName,
entry: &knot_index::SocialEntry,
did: &RepoDid,
) -> anyhow::Result<Option<Replayed>> {
let history = cobs.changes_of(
type_name,
entry.object,
knot_cob::HistoryModel::Convergent,
None,
);
let delta = match history {
Ok(delta) => delta,
Err(error) if entry.state.erased() => {
tracing::warn!(repo = %did, address = %entry.address.at_uri(did), %error, "erased cob history unreadable, skipping");
return Ok(None);
}
Err(error) => return Err(error.into()),
};
let versions = versions_of(&delta.changes)
.inspect_err(|error| {
let key = entry.address.mst_key();
tracing::warn!(repo = %did, %key, %error, "cob versions unreadable, not listing them");
})
.ok();
let body = entry
.state
.serving()
.and_then(|version| body_of(&delta.changes, version));
Ok(Some(Replayed { versions, body }))
}
pub(crate) fn encode_body(body: &RecordBody) -> String {
use base64::Engine as _;
base64::engine::general_purpose::STANDARD_NO_PAD.encode(body.as_bytes())
}
fn decode_body(text: &str) -> Option<RecordBody> {
use base64::Engine as _;
let bytes = base64::engine::general_purpose::STANDARD_NO_PAD
.decode(text)
.ok()?;
RecordBody::new(bytes).ok()
}
#[derive(serde::Serialize, serde::Deserialize)]
struct StoredCob {
versions: Option<Vec<Version>>,
body: Option<String>,
}
impl StoredCob {
fn of(replayed: &Replayed) -> Self {
Self {
versions: replayed.versions.clone(),
body: replayed.body.as_ref().map(encode_body),
}
}
fn into_replayed(self) -> Option<Replayed> {
let body = match self.body {
Some(text) => Some(decode_body(&text)?),
None => None,
};
Some(Replayed {
versions: self.versions,
body,
})
}
}
#[derive(serde::Serialize, serde::Deserialize)]
pub(crate) struct StoredState {
pub(crate) records: Vec<(String, Option<String>)>,
pub(crate) versions: BTreeMap<String, Vec<Version>>,
/// a missing list means gc can't tell what the state holds.
#[serde(default)]
pub(crate) cobs: Option<BTreeSet<String>>,
}
impl StoredState {
pub(crate) fn of(state: &CobState, cobs: BTreeSet<String>) -> Self {
Self {
records: state
.records
.iter()
.map(|(key, body)| (key.as_str().to_string(), body.as_ref().map(encode_body)))
.collect(),
versions: state.versions.clone(),
cobs: Some(cobs),
}
}
fn into_state(self) -> Option<CobState> {
let records = self
.records
.into_iter()
.map(|(key, body)| {
let body = match body {
Some(text) => Some(decode_body(&text)?),
None => None,
};
Some((MstKey::new(key).ok()?, body))
})
.collect::<Option<_>>()?;
Some(CobState {
records,
versions: self.versions,
})
}
}
fn ready<T>(resolved: Resolved<T>) -> anyhow::Result<T> {
match resolved {
Resolved::Ready(value) => Ok(value),
Resolved::Warming => anyhow::bail!("index projection is still warming"),
}
}
#[derive(Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub(crate) struct Version {
#[serde(default, skip_serializing_if = "Option::is_none")]
cid: Option<String>,
#[serde(rename = "changeId")]
change_id: ChangeId,
editor: AccountDid,
#[serde(rename = "createdAt")]
created_at: Datetime,
}
impl Version {
pub(crate) fn editor(&self) -> &str {
self.editor.as_str()
}
}
pub(crate) fn versions_of(changes: &[Change]) -> anyhow::Result<Vec<Version>> {
changes
.iter()
.map(|change| {
let payload = knot_cobs::decode_social(change.payload())?;
let emission = payload.emission();
Ok(Version {
cid: emission.produces().map(|cid| cid.to_string()),
change_id: change.id,
editor: emission.author().clone(),
created_at: Datetime::new(
chrono::DateTime::from_timestamp(emission.at().get(), 0)
.unwrap_or_default()
.fixed_offset(),
),
})
})
.collect()
}
pub(crate) fn body_of(changes: &[Change], version: RecordCid) -> Option<RecordBody> {
changes
.iter()
.rev()
.find_map(|change| match &change.record {
Record::Version(body) if body.cid() == version => Some(body.clone()),
_ => None,
})
}