use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
use anyhow::Context as _;
use bytes::Bytes;
use cid::Cid;
use knot_git::Repo;
use knot_record::chain;
use knot_record::git_refs;
use knot_record::grant::GrantCollection;
use knot_record::manifest::{ManifestMeta, ManifestRecord};
use knot_runtime::Signer;
use knot_types::{
AccountDid, MstKey, RecordAddress, RecordCid, RecordCollection, RecordRkey, UnixSeconds,
};
use crate::cas::atomic_write;
use crate::plan::repo_key;
use crate::store::RepoStore;
use super::chain::{
DueRecords, block_on, created_at, manifest_key, open_chain, read_back, short_hash, unix_now,
};
use super::cob_state::Version;
use super::{ChainState, RepoSource, SettledRepo};
use crate::xrpc::Rows;
/// bump this when the pooled files change, or an old seal gets served.
pub(crate) const SEAL_VERSION: u32 = 3;
pub(crate) const POOLED: [&str; 3] = ["records", "lists", "versions"];
const POOL_KIND: &str = "atproto";
pub(crate) const SEAL_KIND: &str = "atproto-seal";
const LAST_SEALED: &str = "sealed";
const SETTLED: &str = "settled.json";
#[derive(serde::Serialize, serde::Deserialize)]
pub(crate) struct Sealed {
commit: String,
blobs: BTreeSet<String>,
versions: BTreeMap<String, Vec<Version>>,
rows: Rows,
files: BTreeMap<String, u64>,
}
pub fn settle(
chain_root: &Path,
signer: &dyn Signer,
source: &RepoSource,
) -> anyhow::Result<SettledRepo> {
settle_at(
chain_root,
signer,
source,
unix_now(),
&mut crate::firehose::Framing::new(&source.did),
)
}
pub(crate) fn settle_at(
chain_root: &Path,
signer: &dyn Signer,
source: &RepoSource,
now: UnixSeconds,
framing: &mut crate::firehose::Framing,
) -> anyhow::Result<SettledRepo> {
let chain_repo = open_chain(chain_root, &source.did)?;
let versions_path = chain_root
.join(repo_key(source.did.as_str()))
.join("versions.json");
let Some(cobs) = &source.cobs else {
let stored = std::fs::read(&versions_path).unwrap_or_default();
let versions = serde_json::from_slice(&stored).unwrap_or_default();
return sealed(chain_root, &chain_repo, source, versions);
};
atomic_write(&versions_path, &serde_json::to_vec(&cobs.versions)?)
.with_context(|| format!("write {}", versions_path.display()))?;
let export = Repo::open(&source.export_dir)
.with_context(|| format!("open export {}", source.export_dir.display()))?;
let mut due: BTreeMap<MstKey, chain::Due> = git_refs::desired_record(&export)?
.into_iter()
.map(|(key, body)| (key, chain::Due::Write(body)))
.collect();
// the chain's leaves walk the whole tree, and that's only worth it for an unreadable body
let mut leaves = None;
for (key, body) in &cobs.records {
let Some(body) = body else {
let leaves = match &leaves {
Some(leaves) => leaves,
None => leaves.insert(block_on(chain::leaves(&chain_repo))?),
};
match leaves.get(key).and_then(|cid| RecordCid::new(*cid)) {
Some(kept) => {
tracing::warn!(repo = %source.did, %key, "cob record body unreadable, keeping the chain's record");
due.insert(key.clone(), chain::Due::Keep(kept));
}
None => {
tracing::warn!(repo = %source.did, %key, "cob record body unreadable and unpublished, skipping");
}
}
continue;
};
due.insert(key.clone(), chain::Due::Write(body.clone()));
}
if !due.contains_key(&manifest_key()) {
let editor =
AccountDid::new(source.owner.as_str()).context("owner is not an account did")?;
let manifest = ManifestRecord::new(
source.owner.clone(),
source.name.clone(),
ManifestMeta::default(),
created_at(chain_root, &source.did)?,
editor,
)?;
due.insert(manifest_key(), chain::Due::Write(manifest.body()?));
}
let collections: Vec<_> = std::iter::once(git_refs::git_refs_collection())
.chain(knot_cobs::social_collections())
.chain([GrantCollection::Collaborator.records()])
.collect();
let settled_path = chain_root.join(repo_key(source.did.as_str())).join(SETTLED);
let written = Settled::written(&due);
let applied = match &written {
Some(written) => {
let clock = (now, &mut *framing);
apply_changes(
&chain_repo,
source,
signer,
clock,
&settled_path,
written,
&due,
)
.unwrap_or_else(|error| {
tracing::warn!(repo = %source.did, "settling only the changed records: {error:#}");
false
})
}
None => false,
};
if !applied {
block_on(chain::settle(
&chain_repo,
&source.did,
&collections,
&DueRecords(due),
signer,
framing,
now,
))?;
}
match (written, chain::tip(&chain_repo)?) {
(Some(records), Some(tip)) => {
let settled = Settled {
tip: tip.git.to_hex(),
records,
};
atomic_write(&settled_path, &serde_json::to_vec(&settled)?)
.with_context(|| format!("write {}", settled_path.display()))?;
}
_ => remove_settled(&settled_path)?,
}
sealed(chain_root, &chain_repo, source, cobs.versions.clone())
}
/// the records the last settle left on the chain, and the chain tip they're on
#[derive(serde::Serialize, serde::Deserialize)]
struct Settled {
tip: String,
records: BTreeMap<String, String>,
}
impl Settled {
/// a kept record's body is unreadable, so only the full settle knows what's on the chain
fn written(due: &BTreeMap<MstKey, chain::Due>) -> Option<BTreeMap<String, String>> {
due.iter()
.map(|(key, due)| match due {
chain::Due::Write(body) => {
Some((key.as_str().to_string(), body.cid().cid().to_string()))
}
chain::Due::Keep(_) => None,
})
.collect()
}
}
fn remove_settled(path: &Path) -> anyhow::Result<()> {
match std::fs::remove_file(path) {
Err(error) if error.kind() != std::io::ErrorKind::NotFound => {
Err(error).with_context(|| format!("remove {}", path.display()))
}
_ => Ok(()),
}
}
/// writes only what changed since the last settle, when the chain is still where that one left it.
/// false leaves the chain alone for the full settle.
fn apply_changes(
chain_repo: &Repo,
source: &RepoSource,
signer: &dyn Signer,
(now, framing): (UnixSeconds, &mut crate::firehose::Framing),
settled_path: &Path,
written: &BTreeMap<String, String>,
due: &BTreeMap<MstKey, chain::Due>,
) -> anyhow::Result<bool> {
let Some(last) = std::fs::read(settled_path)
.ok()
.and_then(|bytes| serde_json::from_slice::<Settled>(&bytes).ok())
else {
return Ok(false);
};
let Some(tip) = chain::tip(chain_repo)? else {
return Ok(false);
};
if last.tip != tip.git.to_hex() {
return Ok(false);
}
let address = |key: &str| -> anyhow::Result<RecordAddress> {
let (collection, rkey) = key
.split_once('/')
.context("a record key without a collection")?;
Ok(RecordAddress::new(
RecordCollection::new(collection)?,
RecordRkey::new(rkey)?,
))
};
let from = |cid: &str| -> anyhow::Result<RecordCid> {
RecordCid::new(cid.parse()?).context("a settled record cid isn't a record cid")
};
let mut ops = Vec::new();
for (key, due) in due {
let chain::Due::Write(record) = due else {
return Ok(false);
};
let op = match last.records.get(key.as_str()) {
None => chain::RecordOp::Create {
record: record.clone(),
},
Some(cid) if *cid == written[key.as_str()] => continue,
Some(cid) => chain::RecordOp::Update {
record: record.clone(),
from: from(cid)?,
},
};
ops.push((address(key.as_str())?, op));
}
for (key, cid) in &last.records {
if !written.contains_key(key) {
ops.push((address(key)?, chain::RecordOp::Delete { from: from(cid)? }));
}
}
if ops.len() > chain::MAX_OPS_PER_APPLY {
return Ok(false);
}
block_on(chain::apply(
chain_repo,
&source.did,
&ops,
signer,
framing,
now,
))?;
#[cfg(test)]
APPLIED.set(APPLIED.get() + 1);
Ok(true)
}
#[cfg(test)]
thread_local! {
pub(crate) static APPLIED: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}
fn sealed(
chain_root: &Path,
chain_repo: &Repo,
source: &RepoSource,
versions: BTreeMap<String, Vec<Version>>,
) -> anyhow::Result<SettledRepo> {
let tip = chain::tip(chain_repo)?.context("chain has no tip after settling")?;
let commit = chain::commit_cid(&chain::commit_bytes(chain_repo, &tip)?)?;
let key = format!(
"{SEAL_VERSION}-{commit}-{}",
short_hash(&serde_json::to_vec(&versions)?)
);
let dir = source.served.pool(POOL_KIND);
let last = chain_root
.join(repo_key(source.did.as_str()))
.join(LAST_SEALED);
let last_key = std::fs::read_to_string(&last).ok();
let settled = |sealed: Sealed| SettledRepo {
did: source.did.clone(),
commit,
rev: tip.rev.to_string(),
entry_key: key.clone(),
pool: dir.clone(),
blobs: sealed.blobs,
rows: sealed.rows,
};
let read = |key: &str| {
source
.served
.read_file::<Sealed>(SEAL_KIND, key)
.filter(|_| !source.verify)
};
if let Some(sealed) = read(&key).filter(|sealed| Pool::holds(&dir, sealed)) {
if last_key.as_deref() != Some(key.as_str()) {
atomic_write(&last, key.as_bytes())?;
}
return Ok(settled(sealed));
}
let prior = last_key
.and_then(|last| read(&last))
.map(|sealed| Prior::of(sealed, &dir));
let state = read_back(chain_repo, &source.did, versions, prior.as_ref())?;
let mut pool = Pool {
dir: &dir,
verify: source.verify,
files: BTreeMap::new(),
};
let rows = seal(&mut pool, &state, prior.as_ref())?;
let sealed = Sealed {
commit: commit.to_string(),
blobs: crate::blobs::referenced(state.records.iter().map(|record| &record.answer["value"])),
versions: state.versions,
rows,
files: pool.files,
};
source.served.publish_file(SEAL_KIND, &key, &sealed)?;
atomic_write(&last, key.as_bytes())?;
Ok(settled(sealed))
}
pub fn served_dirs(served: &RepoStore) -> (PathBuf, PathBuf) {
(served.pool(POOL_KIND), served.kind_dir(SEAL_KIND))
}
pub fn last_seal(chain_root: &Path, repo_key: &str) -> Option<String> {
std::fs::read_to_string(chain_root.join(repo_key).join(LAST_SEALED)).ok()
}
pub fn kept_served(
served: &RepoStore,
seals: &BTreeSet<String>,
tips: &BTreeSet<String>,
) -> Option<BTreeSet<PathBuf>> {
let pool = served.pool(POOL_KIND);
let mut kept: BTreeSet<PathBuf> = tips.iter().map(|tip| pool.join(tip)).collect();
for key in seals {
let sealed = served.read_file::<Sealed>(SEAL_KIND, key)?;
kept.insert(served.file_path(SEAL_KIND, key));
kept.insert(pool.join(&sealed.commit));
kept.extend(sealed.files.keys().map(|file| pool.join(file)));
}
Some(kept)
}
pub(crate) struct Pool<'a> {
pub(crate) dir: &'a Path,
pub(crate) verify: bool,
pub(crate) files: BTreeMap<String, u64>,
}
impl Pool<'_> {
pub(crate) fn put(&mut self, file: &str, bytes: &[u8]) -> anyhow::Result<()> {
if self.files.contains_key(file) {
return Ok(());
}
let path = self.dir.join(file);
let held = if self.verify {
std::fs::read(&path).is_ok_and(|held| held == bytes)
} else {
length(&path) == Some(bytes.len() as u64)
};
if !held {
atomic_write(&path, bytes)
.and_then(|()| crate::cas::seal_file(&path))
.with_context(|| format!("write {}", path.display()))?;
}
self.files.insert(file.to_string(), bytes.len() as u64);
Ok(())
}
pub(crate) fn keep(&mut self, prior: &Prior, file: &str) -> bool {
let Some(&len) = prior.files.get(file) else {
return false;
};
let kept = length(&self.dir.join(file)) == Some(len);
if kept {
self.files.insert(file.to_string(), len);
}
kept
}
pub(crate) fn holds(dir: &Path, sealed: &Sealed) -> bool {
let head = dir.join(&sealed.commit);
POOLED.iter().all(|shared| head.join(shared).is_dir())
&& sealed
.files
.iter()
.all(|(file, &len)| length(&dir.join(file)) == Some(len))
}
}
fn length(path: &Path) -> Option<u64> {
std::fs::metadata(path).ok().map(|metadata| metadata.len())
}
fn seal(pool: &mut Pool, state: &ChainState, prior: Option<&Prior>) -> anyhow::Result<Rows> {
crate::xrpc::seal_repo(pool, state, prior)
}
pub(crate) struct Prior {
pub(crate) blocks: std::sync::Arc<BTreeMap<Cid, Bytes>>,
pub(crate) versions: BTreeMap<String, Vec<Version>>,
pub(crate) version_rows: std::collections::HashMap<String, Vec<(String, String)>>,
pub(crate) files: BTreeMap<String, u64>,
}
impl Prior {
pub(crate) fn of(sealed: Sealed, dir: &Path) -> Self {
let car = format!("{}/repo.car", sealed.commit);
let blocks = std::fs::read(dir.join(&car))
.ok()
.filter(|bytes| sealed.files.get(&car) == Some(&(bytes.len() as u64)))
.and_then(|bytes| knot_record::car_blocks(&bytes).ok())
.map(|car| std::sync::Arc::new(car.blocks))
.unwrap_or_default();
let mut version_rows = std::collections::HashMap::<String, Vec<_>>::new();
for (row, file) in sealed.rows.version {
let uri = row.split_once('|').map_or(row.as_str(), |(uri, _)| uri);
version_rows
.entry(uri.to_string())
.or_default()
.push((row.clone(), file));
}
Self {
blocks,
versions: sealed.versions,
version_rows,
files: sealed.files,
}
}
}