use std::collections::BTreeSet;
use std::sync::Arc;
use cid::Cid;
use knot_cob::CobHome;
use knot_git::{Layout, Repo};
use knot_index::{Index, Resolved};
use knot_types::{AccountDid, RepoDid};
use crate::atproto::SyncedIndex;
use crate::blobs::Store;
use crate::cob_write::{Done, Ledger, Method, Session, WriteError};
use crate::config::ValidatedConfig;
use crate::repolock::{self, LockMode};
pub enum Actor {
Account(AccountDid),
Owner,
}
pub struct Published {
pub done: Done,
pub commit: Cid,
pub rev: String,
pub changed: Vec<String>,
}
impl Published {
pub fn response(&self) -> serde_json::Value {
let commit = serde_json::json!({ "cid": self.commit.to_string(), "rev": self.rev });
match &self.done {
Done::Written(written) => {
let mut body = serde_json::json!({
"uri": written.uri.as_str(),
"cid": written.cid.to_string(),
"commit": commit,
"validationStatus": "valid",
});
if let Some(change) = &written.change {
body["x-tngl-change"] = change.to_string().into();
}
body
}
Done::Deleted => serde_json::json!({ "commit": commit }),
Done::Granted => serde_json::json!({}),
}
}
}
const UNPUBLISHED: &str = "the knot saved this but couldn't publish it. it shows up once the knot \
publishes again, so don't send it twice";
fn internal(error: impl std::fmt::Display) -> WriteError {
WriteError::internal(format!("{error:#}"))
}
pub async fn perform(
config: &Arc<ValidatedConfig>,
repo: &RepoDid,
actor: Actor,
method: Method,
body: &[u8],
) -> Result<Published, WriteError> {
if let Some((source, head)) = crate::cob_write::fork_head(method, repo, body)? {
// the fetch leaves objects in the repo even when the write fails, so it waits on the grant.
if let Actor::Account(actor) = &actor {
session_of(config, repo, actor.clone())?.require_patch_opening()?;
}
crate::fork::fetch_head(config, repo, &source, head).await?;
}
perform_with(config, repo, actor, |ledger, session| {
ledger.apply(session, method, body)
})
.await
}
pub async fn perform_with(
config: &Arc<ValidatedConfig>,
repo: &RepoDid,
actor: Actor,
write: impl FnOnce(&Ledger<'_>, &Session) -> Result<Done, WriteError>,
) -> Result<Published, WriteError> {
let (done, synced) = write_locked(config, repo, actor, write)?;
if matches!(done, Done::Granted) {
if let Err(error) = crate::git_ssh::spawn_detached(config, ["keys", "sync", "--new"]) {
tracing::warn!("couldn't refresh the ssh keys after a grant: {error:#}");
}
}
let changed = publish_stored(config, repo, Some(&synced)).await?;
let (commit, rev) =
crate::atproto::chain_head(&config.output_root.join("atproto"), repo).map_err(internal)?;
Ok(Published {
done,
commit,
rev,
changed,
})
}
pub(crate) fn rebuilt_index(
config: &ValidatedConfig,
) -> Result<(Layout, std::path::PathBuf, Index), WriteError> {
let layout = config.layout().map_err(internal)?;
let meta_path = layout.meta_path(&config.knot_did).map_err(internal)?;
let index = Index::new(&meta_path, layout.clone());
index.rebuild()?;
Ok((layout, meta_path, index))
}
fn admins(config: &ValidatedConfig) -> BTreeSet<AccountDid> {
BTreeSet::from([config.owner.clone()])
}
fn session_of(
config: &ValidatedConfig,
repo: &RepoDid,
actor: AccountDid,
) -> Result<Session, WriteError> {
let (.., index) = rebuilt_index(config)?;
crate::cob_write::session(&index, &admins(config), actor, repo.clone())
}
pub fn require_contributor(
config: &ValidatedConfig,
repo: &RepoDid,
actor: AccountDid,
) -> Result<(), WriteError> {
session_of(config, repo, actor)?.require_contribution()
}
fn write_locked(
config: &ValidatedConfig,
repo: &RepoDid,
actor: Actor,
write: impl FnOnce(&Ledger<'_>, &Session) -> Result<Done, WriteError>,
) -> Result<(Done, SyncedIndex), WriteError> {
let (layout, meta_path, index) = rebuilt_index(config)?;
let _lock = repolock::acquire(
&config.output_root,
&crate::plan::repo_key(repo.as_str()),
LockMode::Exclusive,
)
.map_err(internal)?;
let git = layout
.open(repo)
.map_err(|_| WriteError::not_found("No such repository on this knot"))?;
let meta = Repo::open(&meta_path)?;
let signer = crate::identity::load_knot_signer(config).map_err(internal)?;
let actor = match actor {
Actor::Account(actor) => actor,
Actor::Owner => match index.ownership_of(repo) {
Resolved::Ready(Some(ownership)) => {
AccountDid::new(ownership.owner.as_str()).map_err(internal)?
}
_ => return Err(WriteError::not_found("No such repository on this knot")),
},
};
let now = crate::atproto::unix_now();
let session = crate::cob_write::session(&index, &admins(config), actor, repo.clone())?;
let guard = index.guard_social(repo)?;
let home = CobHome::from(&config.knot_did);
let chain_root = config.output_root.join("atproto");
let ledger = Ledger {
index: &index,
git: &git,
meta: &meta,
home: &home,
repo,
signer: &signer,
blobs: &Store::of(&config.output_root, repo),
chain_root: &chain_root,
cited: Default::default(),
now,
};
let done = write(&ledger, &session)?;
let tips = crate::atproto::cob_tips(&git).map_err(internal)?;
crate::commit_graph::refresh(git.path());
drop((ledger, guard));
let synced = SyncedIndex {
index,
repo: repo.clone(),
tips,
};
Ok((done, synced))
}
pub(crate) async fn publish_stored(
config: &Arc<ValidatedConfig>,
repo: &RepoDid,
synced: Option<&SyncedIndex>,
) -> Result<Vec<String>, WriteError> {
publish_scoped_with(config, Some(repo.as_str()), synced)
.await
.map_err(|error| {
tracing::error!(%repo, "stored changes weren't published: {error:#}");
WriteError::internal_notice(UNPUBLISHED)
})
}
pub(crate) async fn publish(
config: &Arc<ValidatedConfig>,
repo: &RepoDid,
) -> anyhow::Result<Vec<String>> {
publish_scoped(config, Some(repo.as_str())).await
}
/// None rebuilds the whole knot, which is what a delete needs.
pub(crate) async fn publish_scoped(
config: &Arc<ValidatedConfig>,
repo_hint: Option<&str>,
) -> anyhow::Result<Vec<String>> {
publish_scoped_with(config, repo_hint, None).await
}
async fn publish_scoped_with(
config: &Arc<ValidatedConfig>,
repo_hint: Option<&str>,
synced: Option<&SyncedIndex>,
) -> anyhow::Result<Vec<String>> {
match crate::generate::run_with(config.clone(), repo_hint, synced).await? {
crate::generate::BuildOutcome::Unchanged { .. } => Ok(Vec::new()),
crate::generate::BuildOutcome::Built {
generation_id,
pin: _pin,
..
} => crate::git_ssh::activate_with_merge(config.clone(), &generation_id, repo_hint).await,
}
}
/// deregister first. a crash between the two leaves an orphan directory.
pub(crate) fn retire(config: &ValidatedConfig, repo: &RepoDid) -> Result<(), WriteError> {
let (layout, meta_path, index) = rebuilt_index(config)?;
let _lock = repolock::acquire(
&config.output_root,
&crate::plan::repo_key(repo.as_str()),
LockMode::Exclusive,
)
.map_err(internal)?;
let meta = Repo::open(&meta_path)?;
crate::registry::drop_pending_genesis(&meta, repo).map_err(internal)?;
if let Resolved::Ready(Some(target)) = index.ownership_of(repo) {
let store = knot_cob::CobStore::new(&meta);
let [object] = store.list::<knot_cobs::RepoRegistryCob>()?[..] else {
return Err(WriteError::internal(
"the meta repo lacks one repo registry",
));
};
let signer = crate::identity::load_knot_signer(config).map_err(internal)?;
knot_cobs::deregister_repo(
&store,
&CobHome::from(&config.knot_did),
object,
target,
repo.clone(),
&signer,
crate::atproto::unix_now(),
)?;
}
layout.remove(repo)?;
Ok(())
}