use std::collections::BTreeMap;
use std::future::Future;
use std::path::Path;
use std::sync::Arc;
use anyhow::Context as _;
use bytes::Bytes;
use cid::Cid;
use ipld_core::ipld::Ipld;
use jacquard_repo::mst::Mst;
use jacquard_repo::storage::BlockStore;
use knot_git::Repo;
use knot_record::blocks::GitBlocks;
use knot_record::chain::{self, ChainTip};
use knot_record::manifest;
use knot_types::{MstKey, RepoDid, ServiceDid, UnixSeconds};
use serde_json::json;
use crate::cas::atomic_write;
use crate::plan::repo_key;
use super::cob_state::Version;
use super::pool::Prior;
use super::{ChainState, ServedRecord};
pub fn unix_now() -> UnixSeconds {
let seconds = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |elapsed| elapsed.as_secs());
UnixSeconds::new(seconds as i64)
}
pub(crate) fn open_chain(chain_root: &Path, did: &RepoDid) -> anyhow::Result<Repo> {
let path = chain_root.join(repo_key(did.as_str())).join("chain.git");
if path.exists() {
Repo::open(&path)
} else {
Repo::create(&path)
}
.with_context(|| format!("open atproto chain {}", path.display()))
}
pub fn created_at(chain_root: &Path, did: &RepoDid) -> anyhow::Result<UnixSeconds> {
let path = chain_root.join(repo_key(did.as_str())).join("created_at");
if let Some(seconds) = std::fs::read_to_string(&path)
.ok()
.and_then(|text| text.trim().parse::<i64>().ok())
{
return Ok(UnixSeconds::new(seconds));
}
let now = unix_now();
atomic_write(&path, now.get().to_string().as_bytes())
.with_context(|| format!("write {}", path.display()))?;
Ok(now)
}
pub(crate) fn manifest_key() -> MstKey {
MstKey::new(format!(
"{}/{}",
manifest::MANIFEST_COLLECTION,
manifest::MANIFEST_RKEY
))
.expect("the manifest self key fits mst")
}
pub fn chain_head(chain_root: &Path, did: &RepoDid) -> anyhow::Result<(Cid, String)> {
let chain_repo = open_chain(chain_root, did)?;
let tip = chain::tip(&chain_repo)?.context("chain has no tip")?;
let commit = chain::commit_cid(&chain::commit_bytes(&chain_repo, &tip)?)?;
Ok((commit, tip.rev.to_string()))
}
pub async fn published_spindle(
chain_root: &Path,
did: &RepoDid,
) -> anyhow::Result<Option<ServiceDid>> {
#[derive(serde::Deserialize)]
struct Manifest {
spindle: Option<ServiceDid>,
}
let chain_repo = open_chain(chain_root, did)?;
let Some(tip) = chain::tip(&chain_repo)? else {
return Ok(None);
};
let tree = chain::load_tree(&chain_repo, &tip);
let Some(cid) = tree.get(manifest_key().as_str()).await? else {
return Ok(None);
};
let Some(bytes) = GitBlocks::at(&chain_repo, tip.blocks).get(&cid).await? else {
return Ok(None);
};
Ok(serde_ipld_dagcbor::from_slice::<Manifest>(&bytes)?.spindle)
}
pub(crate) struct DueRecords(pub(crate) BTreeMap<MstKey, chain::Due>);
impl chain::Records for DueRecords {
fn due(&self) -> &BTreeMap<MstKey, chain::Due> {
&self.0
}
}
pub(crate) fn read_back(
chain_repo: &Repo,
did: &RepoDid,
versions: BTreeMap<String, Vec<Version>>,
prior: Option<&Prior>,
) -> anyhow::Result<ChainState> {
let tip = chain::tip(chain_repo)?.context("chain has no tip after settling")?;
let commit_bytes = chain::commit_bytes(chain_repo, &tip)?;
let commit = chain::commit_cid(&commit_bytes)?;
let known = prior
.map(|prior| Arc::clone(&prior.blocks))
.unwrap_or_default();
let (mut blocks, collections, records) = block_on(whole_tree(chain_repo, &tip, did, known))?;
blocks.insert(commit, Bytes::from(commit_bytes));
Ok(ChainState {
did: did.clone(),
commit,
rev: tip.rev.to_string(),
collections,
records,
versions,
car: knot_record::car_bytes(commit, &blocks),
})
}
async fn whole_tree(
repo: &Repo,
tip: &ChainTip,
did: &RepoDid,
known: Arc<BTreeMap<Cid, Bytes>>,
) -> anyhow::Result<(BTreeMap<Cid, Bytes>, Vec<String>, Vec<ServedRecord>)> {
let store = Arc::new(KnownFirst {
known,
chain: Arc::new(GitBlocks::at(repo, tip.blocks)),
});
let tree = Mst::load(Arc::clone(&store), tip.data, None);
let get = async |cid: &Cid| store.get(cid).await;
let mut nodes = tree.collect_node_cids().await?;
nodes.push(tip.data);
let leaves = tree.leaves().await?;
let mut blocks = BTreeMap::new();
for cid in nodes {
let block = get(&cid)
.await?
.with_context(|| format!("chain block {cid} is missing"))?;
blocks.insert(cid, block);
}
let mut collections = Vec::new();
let mut records = Vec::new();
for (key, cid) in leaves {
let (collection, rkey) = key.as_str().split_once('/').unwrap_or((key.as_str(), ""));
if let Some(block) = get(&cid).await? {
let value = record_value(&block).with_context(|| format!("record {key}"))?;
records.push(ServedRecord {
collection: collection.to_string(),
rkey: rkey.to_string(),
answer: json!({
"uri": format!("at://{did}/{key}"),
"cid": cid.to_string(),
"value": value,
}),
});
blocks.insert(cid, block);
}
if collections.last().map(String::as_str) != Some(collection) {
collections.push(collection.to_string());
}
}
Ok((blocks, collections, records))
}
/// the prior seal's blocks, then the chain's. a known block is used only when it hashes to its cid
#[derive(Clone)]
struct KnownFirst {
known: Arc<BTreeMap<Cid, Bytes>>,
chain: Arc<GitBlocks>,
}
impl KnownFirst {
fn known(&self, cid: &Cid) -> Option<Bytes> {
use sha2::{Digest as _, Sha256};
const SHA2_256: u64 = 0x12;
let block = self.known.get(cid)?;
let hash = cid.hash();
let named = hash.code() == SHA2_256 && hash.digest() == Sha256::digest(block).as_slice();
#[cfg(test)]
if named {
KNOWN_READS.set(KNOWN_READS.get() + 1);
}
named.then(|| block.clone())
}
}
#[cfg(test)]
thread_local! {
pub(crate) static KNOWN_READS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}
impl BlockStore for KnownFirst {
async fn get(&self, cid: &Cid) -> jacquard_repo::error::Result<Option<Bytes>> {
match self.known(cid) {
Some(block) => Ok(Some(block)),
None => self.chain.get(cid).await,
}
}
async fn put(&self, data: &[u8]) -> jacquard_repo::error::Result<Cid> {
self.chain.put(data).await
}
async fn has(&self, cid: &Cid) -> jacquard_repo::error::Result<bool> {
Ok(self.get(cid).await?.is_some())
}
async fn put_many(
&self,
blocks: impl IntoIterator<Item = (Cid, Bytes)> + Send,
) -> jacquard_repo::error::Result<()> {
self.chain.put_many(blocks).await
}
async fn get_many(&self, cids: &[Cid]) -> jacquard_repo::error::Result<Vec<Option<Bytes>>> {
let mut blocks = Vec::with_capacity(cids.len());
for cid in cids {
blocks.push(self.get(cid).await?);
}
Ok(blocks)
}
async fn apply_commit(
&self,
commit: jacquard_repo::repo::CommitData,
) -> jacquard_repo::error::Result<()> {
self.chain.apply_commit(commit).await
}
}
fn record_value(bytes: &[u8]) -> anyhow::Result<serde_json::Value> {
let decoded: Ipld = serde_ipld_dagcbor::from_slice(bytes)?;
json_of(&decoded)
}
/// cids go out as `$link` and bytes as `$bytes`, the atproto json spellings.
fn json_of(ipld: &Ipld) -> anyhow::Result<serde_json::Value> {
use serde_json::Value;
Ok(match ipld {
Ipld::Null => Value::Null,
Ipld::Bool(boolean) => Value::Bool(*boolean),
Ipld::Integer(integer) => Value::Number(
serde_json::Number::from_i128(*integer)
.context("an integer in the record is out of range for JSON")?,
),
Ipld::Float(float) => Value::Number(
serde_json::Number::from_f64(*float)
.context("a non-finite float in the record doesn't fit JSON")?,
),
Ipld::String(string) => Value::String(string.clone()),
Ipld::Link(cid) => json!({ "$link": cid.to_string() }),
Ipld::Bytes(bytes) => {
use base64::Engine as _;
json!({ "$bytes": base64::engine::general_purpose::STANDARD_NO_PAD.encode(bytes) })
}
Ipld::List(items) => Value::Array(items.iter().map(json_of).collect::<Result<_, _>>()?),
Ipld::Map(fields) => Value::Object(
fields
.iter()
.map(|(key, value)| json_of(value).map(|value| (key.clone(), value)))
.collect::<Result<_, _>>()?,
),
})
}
pub(crate) fn record_file(collection: &str, rkey: &str, cid: &str) -> String {
format!(
"records/{}.json",
short_hash(format!("{collection}/{rkey}@{cid}").as_bytes())
)
}
pub(crate) fn short_hash(bytes: &[u8]) -> String {
use sha2::{Digest as _, Sha256};
knot_types::lowercase_hex(&Sha256::digest(bytes))[..32].to_string()
}
/// the chain's trees spawn tasks, which need a runtime to run on.
pub(crate) fn block_on<T>(future: impl Future<Output = T>) -> T {
tokio::runtime::Builder::new_current_thread()
.build()
.expect("build the chain runtime")
.block_on(future)
}
pub(crate) fn write_json(path: &Path, value: &serde_json::Value) -> anyhow::Result<()> {
atomic_write(path, &serde_json::to_vec(value)?)
.with_context(|| format!("write {}", path.display()))
}