use std::sync::Arc;
use anyhow::Context as _;
use knot_index::Index;
use knot_types::{BranchName, ContributionPolicy, MstKey, RepoDid};
use serde_json::{Value, json};
use crate::cli::{
CollaboratorCommand, IssueCommand, LabelAction, LabelCommand, PullCommand, RenameArgs,
SettingsArgs,
};
use crate::cob_write::{Method, Settings};
use crate::config::ValidatedConfig;
use crate::writes::{self, Actor};
const TICKET: &str = "org.tangled.track.ticket";
const COMMENT: &str = "org.tangled.feed.comment";
const LABEL: &str = "org.tangled.label.definition";
const COLLABORATOR: &str = "org.tangled.graph.collaborator";
fn now() -> String {
chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
}
pub(crate) fn resolve(config: &ValidatedConfig, repo: &str) -> anyhow::Result<RepoDid> {
if repo.starts_with("did:") {
return Ok(RepoDid::new(repo)?);
}
let layout = config.layout()?;
let index = Index::new(layout.meta_path(&config.knot_did)?, layout);
index.rebuild().context("rebuild registry index")?;
let mut named = crate::registry::matching_bare_name(&index, repo)?.into_iter();
let first = named
.next()
.with_context(|| format!("no hosted repo {repo}"))?;
anyhow::ensure!(
named.next().is_none(),
"{repo} names more than one repo here, use the repo's DID"
);
Ok(first)
}
/// the cid goes out as `swapRecord`. the edit fails if the record moved.
fn live(
config: &ValidatedConfig,
repo: &RepoDid,
collection: &str,
rkey: &str,
) -> anyhow::Result<(String, Value)> {
let layout = config.layout()?;
let index = Index::new(layout.meta_path(&config.knot_did)?, config.layout()?);
index.rebuild().context("rebuild registry index")?;
let repo_key = crate::plan::repo_key(repo.as_str());
// without the lock, a write between reading the cob tips and replaying them caches stale state
let lock = crate::repolock::acquire(
&config.output_root,
&repo_key,
crate::repolock::LockMode::Lifetime,
)?;
let store =
crate::store::RepoStore::open(&crate::repolock::repo_dir(&config.output_root, &repo_key));
let state = crate::atproto::cob_state(
&index,
&layout.open(repo)?,
repo,
&store,
&crate::datastore::Datastore::at(&config.output_root),
&lock,
)?;
let key = MstKey::new(format!("{collection}/{rkey}"))?;
let body = state
.records
.get(&key)
.and_then(Option::as_ref)
.with_context(|| format!("no live {collection}/{rkey}"))?;
Ok((
body.cid().to_string(),
serde_ipld_dagcbor::from_slice(body.as_bytes())?,
))
}
async fn write(
config: &Arc<ValidatedConfig>,
repo: &RepoDid,
method: Method,
body: Value,
) -> anyhow::Result<Value> {
let published = writes::perform(
config,
repo,
Actor::Owner,
method,
body.to_string().as_bytes(),
)
.await
.map_err(|error| anyhow::anyhow!("{}", error.body()))?;
announce(config, &published).await;
Ok(published.response())
}
pub(crate) async fn announce(config: &Arc<ValidatedConfig>, published: &writes::Published) {
if !published.changed.is_empty() {
crate::crawl::announce(config, &published.changed).await;
}
}
fn body_text(
body: Option<String>,
file: Option<std::path::PathBuf>,
) -> anyhow::Result<Option<String>> {
match (body, file) {
(Some(body), _) => Ok(Some(body)),
(None, Some(file)) => Ok(Some(
std::fs::read_to_string(&file).with_context(|| format!("read {}", file.display()))?,
)),
(None, None) => Ok(None),
}
}
async fn put(
config: &Arc<ValidatedConfig>,
repo: &RepoDid,
collection: &str,
rkey: &str,
edit: impl FnOnce(&mut Value) -> anyhow::Result<()>,
) -> anyhow::Result<Value> {
let (cid, mut record) = live(config, repo, collection, rkey)?;
edit(&mut record)?;
write(
config,
repo,
Method::PutRecord,
json!({ "repo": repo, "collection": collection, "rkey": rkey, "swapRecord": cid, "record": record }),
)
.await
}
pub async fn issue(config: Arc<ValidatedConfig>, command: IssueCommand) -> anyhow::Result<()> {
let printed = match command {
IssueCommand::New {
repo,
title,
body,
body_file,
} => {
let repo = resolve(&config, &repo)?;
let mut record = json!({ "title": title, "createdAt": now() });
if let Some(text) = body_text(body, body_file)? {
record["body"] = text.into();
}
write(
&config,
&repo,
Method::CreateRecord,
json!({ "repo": repo, "collection": TICKET, "record": record }),
)
.await?
}
IssueCommand::Comment {
repo,
number,
body,
body_file,
reply_to,
} => {
let repo = resolve(&config, &repo)?;
let (cid, _) = live(&config, &repo, TICKET, &number)?;
let text =
body_text(body, body_file)?.context("a comment needs --body or --body-file")?;
let mut record = json!({
"subject": { "uri": format!("at://{repo}/{TICKET}/{number}"), "cid": cid },
"body": { "text": text },
"createdAt": now(),
});
if let Some(reply) = reply_to {
let (reply_cid, _) = live(&config, &repo, COMMENT, &reply)?;
record["replyTo"] =
json!({ "uri": format!("at://{repo}/{COMMENT}/{reply}"), "cid": reply_cid });
}
write(
&config,
&repo,
Method::CreateRecord,
json!({ "repo": repo, "collection": COMMENT, "record": record }),
)
.await?
}
IssueCommand::Close {
repo,
number,
completed,
} => {
let state = if completed { "completed" } else { "closed" };
let repo = resolve(&config, &repo)?;
put(&config, &repo, TICKET, &number, |record| {
record["state"] = state.into();
Ok(())
})
.await?
}
IssueCommand::Reopen { repo, number } => {
let repo = resolve(&config, &repo)?;
put(&config, &repo, TICKET, &number, |record| {
record["state"] = "open".into();
Ok(())
})
.await?
}
IssueCommand::Label {
repo,
number,
action,
label,
} => label_ticket(&config, &repo, &number, action, &label).await?,
IssueCommand::Edit {
repo,
number,
title,
body,
body_file,
} => {
let repo = resolve(&config, &repo)?;
let text = body_text(body, body_file)?;
put(&config, &repo, TICKET, &number, |record| {
if let Some(title) = title {
record["title"] = title.into();
}
if let Some(text) = text {
record["body"] = text.into();
}
Ok(())
})
.await?
}
};
println!("{printed}");
Ok(())
}
pub async fn label(config: Arc<ValidatedConfig>, command: LabelCommand) -> anyhow::Result<()> {
let printed = match command {
LabelCommand::Add {
repo,
name,
color,
slug,
} => {
let repo = resolve(&config, &repo)?;
let mut record = json!({ "name": name, "createdAt": now() });
if let Some(color) = color {
record["color"] = color.into();
}
write(
&config,
&repo,
Method::CreateRecord,
json!({ "repo": repo, "collection": LABEL, "rkey": slug.unwrap_or(name), "record": record }),
)
.await?
}
LabelCommand::Rm { repo, slug } => {
let repo = resolve(&config, &repo)?;
let (cid, _) = live(&config, &repo, LABEL, &slug)?;
write(
&config,
&repo,
Method::DeleteRecord,
json!({ "repo": repo, "collection": LABEL, "rkey": slug, "swapRecord": cid }),
)
.await?
}
};
println!("{printed}");
Ok(())
}
async fn label_ticket(
config: &Arc<ValidatedConfig>,
repo: &str,
number: &str,
action: LabelAction,
label: &str,
) -> anyhow::Result<Value> {
let repo = resolve(config, repo)?;
put(config, &repo, TICKET, number, |record| {
let mut tokens = record["tangledLabels"]
.as_array()
.cloned()
.unwrap_or_default();
let has = tokens.iter().any(|token| token["key"] == label);
match (action, has) {
(LabelAction::Add, true) => anyhow::bail!("{number} already has the label {label}"),
(LabelAction::Rm, false) => anyhow::bail!("{number} doesn't have the label {label}"),
(LabelAction::Add, false) => tokens.push(json!({ "key": label })),
(LabelAction::Rm, true) => tokens.retain(|token| token["key"] != label),
}
// edits replace the record. leaving tangledLabels out clears them
if tokens.is_empty() {
if let Some(fields) = record.as_object_mut() {
fields.remove("tangledLabels");
}
} else {
record["tangledLabels"] = tokens.into();
}
Ok(())
})
.await
}
pub async fn collaborator(
config: Arc<ValidatedConfig>,
command: CollaboratorCommand,
) -> anyhow::Result<()> {
let printed = match command {
CollaboratorCommand::Add { repo, did } => {
let repo = resolve(&config, &repo)?;
let record = json!({ "$type": COLLABORATOR, "createdAt": now() });
write(
&config,
&repo,
Method::CreateRecord,
json!({ "repo": repo, "collection": COLLABORATOR, "rkey": did, "record": record }),
)
.await?
}
CollaboratorCommand::Rm { repo, did } => {
let repo = resolve(&config, &repo)?;
write(
&config,
&repo,
Method::DeleteRecord,
json!({ "repo": repo, "collection": COLLABORATOR, "rkey": did }),
)
.await?
}
};
println!("{printed}");
Ok(())
}
pub async fn pull(config: Arc<ValidatedConfig>, command: PullCommand) -> anyhow::Result<()> {
match command {
PullCommand::Label {
repo,
number,
action,
label,
} => {
let labelled = label_ticket(&config, &repo, &number, action, &label).await?;
println!("{labelled}");
}
PullCommand::New {
repo,
title,
body,
body_file,
target,
branch,
from,
} => {
let repo = resolve(&config, &repo)?;
let source = from
.as_deref()
.map(RepoDid::new)
.transpose()?
.unwrap_or_else(|| repo.clone());
let branch = BranchName::new(branch)?;
let git = config.layout()?.open(&repo)?;
let head = if source == repo {
git.find_ref(&branch.head_ref())?
.with_context(|| format!("no branch {branch}"))?
} else {
let head = crate::fork::branch_tip(&config, &source, &branch)
.await
.map_err(|error| anyhow::anyhow!("{}", error.body()))?;
crate::fork::fetch_head(&config, &repo, &source, head)
.await
.map_err(|error| anyhow::anyhow!("{}", error.body()))?;
head
};
let target_tip = git
.find_ref(&BranchName::new(&target)?.head_ref())?
.with_context(|| format!("no branch {target}"))?;
let base = git
.merge_base(target_tip, head)?
.with_context(|| format!("{branch} shares no history with {target}"))?;
let mut record = json!({ "title": title, "targetBranch": target, "createdAt": now() });
if let Some(text) = body_text(body, body_file)? {
record["body"] = text.into();
}
let opened = write(
&config,
&repo,
Method::CreateRecord,
json!({ "repo": repo, "collection": TICKET, "record": record }),
)
.await?;
let rkey = opened["uri"]
.as_str()
.and_then(|uri| uri.rsplit('/').next())
.context("ticket uri")?;
let patch = write(
&config,
&repo,
Method::SubmitPatch,
json!({
"repo": repo,
"patch": {
"ticket": rkey,
"base": { "oid": base },
"head": { "oid": head },
"sourceRepo": source,
"sourceBranch": branch,
},
}),
)
.await?;
println!("{opened}\n{patch}");
}
}
Ok(())
}
pub async fn settings(config: Arc<ValidatedConfig>, args: SettingsArgs) -> anyhow::Result<()> {
let repo = resolve(&config, &args.repo)?;
let patch = Settings {
name: None,
description: args.description,
website: args.website,
topics: (!args.topics.is_empty()).then_some(args.topics),
policy: args
.policy
.map(|policy| serde_json::from_value::<ContributionPolicy>(json!(policy)))
.transpose()?,
spindle: if args.no_spindle {
Some(None)
} else {
args.spindle
.map(knot_types::ServiceDid::new)
.transpose()?
.map(Some)
},
source: None,
};
let published = writes::perform_with(&config, &repo, Actor::Owner, |ledger, session| {
ledger.settings(session, &patch)
})
.await
.map_err(|error| anyhow::anyhow!("{}", error.body()))?;
announce(&config, &published).await;
println!("{}", published.response());
Ok(())
}
/// new declaration first, old one last, and the new one is undone if the knot refuses.
pub async fn rename(config: Arc<ValidatedConfig>, args: RenameArgs) -> anyhow::Result<()> {
const DECLARATION: &str = "org.tangled.repo.declaration";
let repo = resolve(&config, &args.repo)?;
let new_name = knot_types::RepoName::new(&args.new_name).context("invalid repo name")?;
let to = knot_types::RepoRkey::from_repo_name(&new_name).context("invalid repo rkey")?;
let layout = config.layout()?;
let index = Index::new(layout.meta_path(&config.knot_did)?, layout);
index.rebuild().context("rebuild registry index")?;
let ownership = match index.ownership_of(&repo) {
knot_index::Resolved::Ready(Some(ownership)) => ownership,
knot_index::Resolved::Ready(None) => anyhow::bail!("no hosted repo {repo}"),
knot_index::Resolved::Warming => anyhow::bail!("the registry is still warming, try again"),
};
if let knot_index::Resolved::Ready(Some(existing)) = index.resolve_repo(&ownership.owner, &to)
&& existing != repo
{
anyhow::bail!("the record key {to} already belongs to another repository");
}
let from = ownership.rkey;
let patch = Settings {
name: Some(new_name.clone()),
..Settings::default()
};
if from == to {
let published = writes::perform_with(&config, &repo, Actor::Owner, |ledger, session| {
ledger.settings(session, &patch)
})
.await
.map_err(|error| anyhow::anyhow!("{}", error.body()))?;
announce(&config, &published).await;
println!("{}", published.response());
return Ok(());
}
let owner_account = knot_types::AccountDid::new(ownership.owner.to_string())
.context("invalid owner DID in registry")?;
let mut oauth_session = crate::oauth::ensure_auth_for_owner(&config, &owner_account, true)
.await?
.context("OAuth session required for moving the repository declaration")?;
if crate::oauth::get_record_oauth(&mut oauth_session, DECLARATION, to.as_str())
.await?
.is_some()
{
anyhow::bail!("A repository named \"{new_name}\" already exists in your account.");
}
let current = crate::oauth::get_record_oauth(&mut oauth_session, DECLARATION, from.as_str())
.await?
.context("This repository has no declaration in your account to move.")?;
let current_cid = current["cid"].as_str().map(str::to_string);
let moved = crate::oauth::create_record_oauth(
&mut oauth_session,
DECLARATION,
to.as_str(),
¤t["value"],
)
.await?;
let moved_cid = moved["cid"].as_str().map(str::to_string);
let published = match writes::perform_with(&config, &repo, Actor::Owner, |ledger, session| {
ledger.settings(session, &patch)
})
.await
{
Ok(published) => published,
Err(error) => {
let _ = crate::oauth::delete_record_oauth(
&mut oauth_session,
DECLARATION,
to.as_str(),
moved_cid.as_deref(),
)
.await;
anyhow::bail!("{}", error.body());
}
};
announce(&config, &published).await;
crate::oauth::delete_record_oauth(
&mut oauth_session,
DECLARATION,
from.as_str(),
current_cid.as_deref(),
)
.await?;
println!("{}", published.response());
Ok(())
}
#[cfg(test)]
#[path = "tests.rs"]
mod tests;