use std::collections::BTreeMap;
use std::path::PathBuf;
use serde::{Deserialize, Serialize};
use serde_json::json;
use url::Url;
use crate::boundary::{Surface, TermScope};
use crate::error::{invalid, Error, Result};
use crate::event::EventKind;
use crate::host::{ChangeId, RemoteHost, Sha};
use crate::providers::Providers;
use crate::review::{ReadCost, ReadMarker, ReviewRead};
use crate::stream::Stream;
use crate::workspace::{object, Ref};
use crate::{git, home, ids, publish, release, review, status, store};
pub(crate) mod github;
pub(crate) const STANDINGS_BATCH: usize = 50;
pub(crate) const REVIEW_BATCH: usize = 150;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ChangeStandingsRequest {
pub repo: String,
pub root: String,
pub changes: Vec<Url>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ChangeStanding {
pub url: Url,
pub body: String,
pub base: String,
pub head_branch: String,
pub head: Sha,
pub draft: bool,
pub state: ChangeState,
pub checks: ChecksVerdict,
}
#[non_exhaustive]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "state", rename_all = "kebab-case")]
pub enum ChangeState {
Open,
Merged {
into: String,
commit: Sha,
landing: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
merged_at: Option<String>,
},
Closed {
base_deleted: bool,
},
}
#[non_exhaustive]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "verdict", rename_all = "kebab-case")]
pub enum ChecksVerdict {
Pending {
waiting: Vec<String>,
},
Passed,
PassedWithSkipped {
skipped: Vec<String>,
},
Failed {
failed: Vec<String>,
},
NoneRequired,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ChangeStandings {
pub standings: Vec<ChangeStanding>,
pub cost: ReadCost,
}
pub fn change_standings(
providers: &Providers<'_>,
request: &ChangeStandingsRequest,
) -> Result<ChangeStandings> {
let repository = Repository::of(providers, &request.repo)?;
let root = branch_name(&request.root, "the root")?;
let ids = repository.ids(request.changes.iter())?;
let read = if ids.is_empty() {
ChangeStandings {
standings: Vec::new(),
cost: ReadCost::default(),
}
} else {
repository.host.change_standings(&root, &ids)?
};
repository.answered(&ids, &read.standings)?;
let landed = repository.take_up(&read.standings, &root);
repository.record_read(read.standings.len(), &landed, read.cost);
Ok(read)
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ReviewTarget {
pub change: Url,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub since: Option<ReadMarker>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ReviewStandingsRequest {
pub repo: String,
pub root: String,
pub changes: Vec<ReviewTarget>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ReviewStandings {
pub standings: Vec<ChangeStanding>,
pub reads: Vec<ReviewRead>,
pub cost: ReadCost,
}
pub fn review_standings(
providers: &Providers<'_>,
request: &ReviewStandingsRequest,
) -> Result<ReviewStandings> {
let repository = Repository::of(providers, &request.repo)?;
let root = branch_name(&request.root, "the root")?;
let ids = repository.ids(request.changes.iter().map(|target| &target.change))?;
let targets: Vec<(ChangeId, Option<ReadMarker>)> = ids
.iter()
.cloned()
.zip(request.changes.iter().map(|target| target.since.clone()))
.collect();
let read = if targets.is_empty() {
ReviewStandings {
standings: Vec::new(),
reads: Vec::new(),
cost: ReadCost::default(),
}
} else {
repository.host.review_standings(&root, &targets)?
};
repository.answered(&ids, &read.standings)?;
if read.reads.len() != ids.len() {
return Err(invalid(format!(
"the host answered {} review read(s) for {} change request(s), so which feedback \
belongs to which cannot be read",
read.reads.len(),
ids.len()
)));
}
let landed = repository.take_up(&read.standings, &root);
for review_read in &read.reads {
review::record_url_read(&repository.key, review_read);
}
repository.record_read(read.standings.len(), &landed, read.cost);
Ok(read)
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RetargetRequest {
pub repo: String,
pub change: Url,
pub base: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub body: Option<String>,
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub remove_restored_base: bool,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Retargeted {
pub change: Url,
pub from: String,
pub to: String,
pub outcome: RetargetOutcome,
pub cost: ReadCost,
}
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum RetargetOutcome {
Retargeted,
Unchanged,
}
pub fn retarget_change(providers: &Providers<'_>, request: &RetargetRequest) -> Result<Retargeted> {
let repository = Repository::of(providers, &request.repo)?;
let base = branch_name(&request.base, "the base")?;
let id = repository.id(&request.change)?;
let mut cost = ReadCost::default();
let before = repository.one(&id, &base, &mut cost)?;
let url = before.url.clone();
match &before.state {
ChangeState::Open => {}
ChangeState::Merged { into, .. } => {
return Err(invalid(format!(
"{url} has merged into {into:?}, so its base cannot change. Nothing was written"
)))
}
_ => {
return Err(invalid(format!(
"{url} is closed, and a closed change request's base cannot change: reopen it \
first — `onevcs change reopen` restores a deleted base and reopens it. Nothing \
was written"
)))
}
}
let from = before.base.clone();
let moves = from != base;
if request.remove_restored_base {
if !moves {
return Err(invalid(format!(
"{url} already targets {base:?}, so no base is left behind for \
--remove-restored-base to delete. Nothing was written"
)));
}
if !restored_on(&url, &from)? {
return Err(invalid(format!(
"{from:?} is not a base `onevcs change reopen` restored on {url}, so \
--remove-restored-base will not delete it. Nothing was written"
)));
}
}
if !moves && request.body.is_none() {
repository.record_change(
&url,
EventKind::ChangeRetargeted,
json!({
"change_url": url.to_string(),
"from": from,
"to": base,
"outcome": "unchanged",
"body_written": false,
}),
);
return Ok(Retargeted {
change: url,
to: base,
from,
outcome: RetargetOutcome::Unchanged,
cost,
});
}
let tips = repository.origin_branches()?;
if moves && !tips.contains_key(&base) {
return Err(invalid(format!(
"the origin of {} holds no branch {base:?}, so {url} cannot target it. Nothing was \
written",
repository.key
)));
}
let mut fields = vec![(Surface::Branch, base.as_str())];
if let Some(body) = &request.body {
fields.push((Surface::Body, body.as_str()));
}
crate::boundary::evidence::guard_fields(
providers.hosting,
&repository.key,
&fields,
&TermScope::default(),
)?;
add(
&mut cost,
repository
.host
.retarget_change(&id, &base, request.body.as_deref())?,
);
let after = repository.one(&id, &base, &mut cost)?;
if after.draft != before.draft {
return Err(invalid(format!(
"the host moved {url} onto {base:?} and answered it {} where it was {} — this \
crate never changes a change request's draft state, so the host's answer is \
refused",
draft_word(after.draft),
draft_word(before.draft)
)));
}
if after.base != base {
return Err(invalid(format!(
"the host accepted moving {url} onto {base:?} and reports its base as {:?}",
after.base
)));
}
if request.remove_restored_base {
let expected = tips
.get(&from)
.map(|tip| tip.as_str().to_owned())
.ok_or_else(|| {
invalid(format!(
"{url} moved onto {base:?}, and the restored base {from:?} is already gone \
from the origin, so there was nothing to delete"
))
})?;
let deleted = git::delete_remote_branch(&repository.publication, ORIGIN, &from, &expected)?;
if !deleted.accepted() {
return Err(Error::PushRejected {
reason: format!(
"{url} moved onto {base:?}, and the origin declined to delete the restored \
base {from:?}: {}",
deleted.output().lines().next_back().unwrap_or("").trim()
),
});
}
}
let outcome = if moves {
RetargetOutcome::Retargeted
} else {
RetargetOutcome::Unchanged
};
repository.record_change(
&url,
EventKind::ChangeRetargeted,
json!({
"change_url": url.to_string(),
"from": from,
"to": base,
"outcome": outcome_word(outcome),
"body_written": request.body.is_some(),
}),
);
Ok(Retargeted {
change: url,
from,
to: base,
outcome,
cost,
})
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ReopenRequest {
pub repo: String,
pub change: Url,
pub restore_base_at: Sha,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Reopened {
pub change: Url,
pub restored_base: String,
pub cost: ReadCost,
}
pub fn reopen_change(providers: &Providers<'_>, request: &ReopenRequest) -> Result<Reopened> {
let repository = Repository::of(providers, &request.repo)?;
let id = repository.id(&request.change)?;
let at = git::ObjectId::parse(&request.restore_base_at.0).ok_or_else(|| {
invalid(format!(
"{:?} is not a commit id; pass the full id of the commit to restore the base at",
request.restore_base_at.0
))
})?;
let mut cost = ReadCost::default();
let before = repository.one(&id, "", &mut cost)?;
let url = before.url.clone();
if !matches!(before.state, ChangeState::Closed { .. }) {
return Err(invalid(format!(
"{url} is {}, not closed, so there is nothing to reopen. Nothing was written",
match before.state {
ChangeState::Open => "open",
_ => "merged",
}
)));
}
let base = branch_name(&before.base, "the change request's base")?;
if repository.origin_branches()?.contains_key(&base) {
return Err(invalid(format!(
"the origin of {} still holds {base:?}, the base of {url}, so there is no deleted \
base to restore. Nothing was written",
repository.key
)));
}
let commit = Sha(at.as_str().to_owned());
if !git::has_commit(repository.publication.as_path(), &commit)
&& !git::fetch_commit(&repository.publication, ORIGIN, at.as_str())?
{
return Err(invalid(format!(
"{} is not a commit {} holds or can fetch from its origin, so {base:?} cannot be \
restored at it. Nothing was written",
at.as_str(),
repository.publication.display()
)));
}
let compared = format!(
"{ORIGIN}/{}",
git::default_branch(&repository.publication, ORIGIN)?
);
crate::boundary::evidence::guard_publication(
providers.hosting,
&repository.key,
&crate::boundary::screen::Outgoing {
repo: &repository.publication,
base: Some(compared.as_str()),
tip: at.as_str(),
branch: Some(base.as_str()),
title: None,
body: None,
},
&TermScope::default(),
)?;
let pushed = git::restore_remote_branch(&repository.publication, ORIGIN, &base, at.as_str())?;
if !pushed.accepted() {
return Err(Error::PushRejected {
reason: format!(
"the origin of {} declined restoring {base:?} at {}: {}. Nothing was reopened",
repository.key,
at.as_str(),
pushed.output().lines().next_back().unwrap_or("").trim()
),
});
}
add(&mut cost, repository.host.reopen_change(&id)?);
let after = repository.one(&id, "", &mut cost)?;
if after.state != ChangeState::Open {
return Err(invalid(format!(
"the host accepted reopening {url} and still reports it {}; {base:?} stays restored \
at {}",
match after.state {
ChangeState::Closed { .. } => "closed",
_ => "merged",
},
at.as_str()
)));
}
if after.draft != before.draft {
return Err(invalid(format!(
"the host reopened {url} {} where it was {} — this crate never changes a change \
request's draft state, so the host's answer is refused",
draft_word(after.draft),
draft_word(before.draft)
)));
}
repository.record_change(
&url,
EventKind::ChangeReopened,
json!({
"change_url": url.to_string(),
"restored_base": base,
"at": at.as_str(),
}),
);
Ok(Reopened {
change: url,
restored_base: base,
cost,
})
}
const ORIGIN: &str = "origin";
fn add(total: &mut ReadCost, more: ReadCost) {
total.graphql_points = total.graphql_points.saturating_add(more.graphql_points);
total.rest_requests = total.rest_requests.saturating_add(more.rest_requests);
}
fn draft_word(draft: bool) -> &'static str {
if draft {
"a draft"
} else {
"ready for review"
}
}
fn outcome_word(outcome: RetargetOutcome) -> &'static str {
match outcome {
RetargetOutcome::Retargeted => "retargeted",
RetargetOutcome::Unchanged => "unchanged",
}
}
fn branch_name(value: &str, what: &str) -> Result<String> {
Ref::try_from(value.to_owned())
.map(String::from)
.map_err(|reason| invalid(format!("{what} is {reason}")))
}
struct Repository {
host: Box<dyn RemoteHost>,
key: String,
publication: PathBuf,
registry: crate::registry::Registry,
}
impl Repository {
fn of(providers: &Providers<'_>, repo: &str) -> Result<Self> {
let registry = store::load()?;
let resolution = store::resolve(®istry, repo)?;
let slug = publish::change_host(&resolution.key)?;
Ok(Self {
host: providers.hosting.for_repo(&slug)?,
key: resolution.key,
publication: resolution.publication,
registry,
})
}
fn id(&self, url: &Url) -> Result<ChangeId> {
let parsed = review::ChangeUrl::parse(url.as_str())?;
if parsed.identity != self.key {
return Err(invalid(format!(
"{url} is a change request of {}, not of {}; every change request named must \
be one of the repository the call names",
parsed.identity, self.key
)));
}
Ok(ChangeId(parsed.number))
}
fn ids<'a>(&self, urls: impl Iterator<Item = &'a Url>) -> Result<Vec<ChangeId>> {
urls.map(|url| self.id(url)).collect()
}
fn answered(&self, ids: &[ChangeId], standings: &[ChangeStanding]) -> Result<()> {
let numbers: Vec<Option<String>> = standings
.iter()
.map(|standing| {
review::ChangeUrl::parse(standing.url.as_str())
.ok()
.map(|url| url.number)
})
.collect();
let asked: Vec<Option<String>> = ids.iter().map(|id| Some(id.0.clone())).collect();
if numbers != asked {
return Err(invalid(format!(
"the host answered about {} change request(s) where {} were asked about, or not \
in the order asked, so which standing is which cannot be read",
standings.len(),
ids.len()
)));
}
Ok(())
}
fn one(&self, id: &ChangeId, root: &str, cost: &mut ReadCost) -> Result<ChangeStanding> {
let read = self.host.change_standings(root, std::slice::from_ref(id))?;
self.answered(std::slice::from_ref(id), &read.standings)?;
add(cost, read.cost);
let standing = read.standings.into_iter().next().ok_or_else(|| {
invalid(format!(
"the host answered nothing about change request {}",
id.0
))
})?;
record_head(&standing.url, &standing.head_branch);
Ok(standing)
}
fn origin_branches(&self) -> Result<BTreeMap<String, git::ObjectId>> {
git::remote_branch_tips(&self.publication, ORIGIN)?.ok_or_else(|| {
invalid(format!(
"the origin of {} would not list its branches from {}",
self.key,
self.publication.display()
))
})
}
fn take_up(&self, standings: &[ChangeStanding], root: &str) -> Vec<String> {
let mut landed = Vec::new();
for standing in standings {
record_head(&standing.url, &standing.head_branch);
let ChangeState::Merged {
into,
commit,
landing: true,
merged_at,
} = &standing.state
else {
continue;
};
if into != root {
continue;
}
if self.record_landing(standing, commit, merged_at.as_deref()) {
landed.push(standing.url.to_string());
}
}
landed
}
fn record_landing(
&self,
standing: &ChangeStanding,
commit: &Sha,
merged_at: Option<&str>,
) -> bool {
let warn = |failure: &dyn std::fmt::Display| {
eprintln!(
"onevcs: warning: the host reports {url} merged into the root at {commit}, and \
this host could not record that landing for {branch} ({failure})",
url = standing.url,
commit = commit.0,
branch = standing.head_branch,
);
false
};
let (recorded, opened_on) = match status::landing_record(&self.key, &standing.head_branch) {
Ok(found) => found,
Err(failure) => return warn(&failure),
};
if recorded.as_deref() == Some(commit.0.as_str()) {
return false;
}
let located = match release::for_repository(&self.registry, &self.key) {
Ok(located) => located,
Err(failure) => return warn(&failure),
};
let mut releases = match Stream::releases(&self.key) {
Ok(stream) => stream,
Err(failure) => return warn(&failure),
};
let own = change_stream(standing.url.as_str());
let recording = match &opened_on {
Some(token) => release::LandingStream::Opened(token),
None => release::LandingStream::Own {
token: &own,
identity: &self.key,
},
};
release::reconcile_landing(
&self.registry,
&located,
&standing.head_branch,
Some(recording),
release::Landing {
commit: &commit.0,
at: merged_at.and_then(crate::gate_run::Moment::reported),
},
&standing.url,
&mut releases,
)
}
fn record_read(&self, count: usize, landed: &[String], cost: ReadCost) {
let token = format!("standings-{}", ids::short_digest(&self.key));
match Stream::open(&token) {
Ok(mut stream) => {
stream.label("identity", &self.key);
stream.emit(
EventKind::ChangeStandingsRead,
object(json!({
"repo": self.key,
"count": count,
"landed": landed,
"cost": {
"graphql_points": cost.graphql_points,
"rest_requests": cost.rest_requests,
},
})),
);
}
Err(error) => eprintln!(
"onevcs: warning: the standings read of {} is not recorded: {error}",
self.key
),
}
}
fn record_change(&self, url: &Url, kind: EventKind, payload: serde_json::Value) {
match Stream::open(&change_stream(url.as_str())) {
Ok(mut stream) => {
stream.label("change_url", url.as_str());
stream.label("identity", &self.key);
stream.emit(kind, object(payload));
}
Err(error) => {
eprintln!("onevcs: warning: what was done on {url} is not recorded: {error}")
}
}
}
}
pub(crate) fn change_stream(change_url: &str) -> String {
format!("change-{}", ids::short_digest(change_url))
}
fn restored_on(url: &Url, base: &str) -> Result<bool> {
let token = change_stream(url.as_str());
if !crate::stream::path_for(&token)?.is_file() {
return Ok(false);
}
let mut reader = crate::stream::Reader::open(&token)?;
Ok(reader.records()?.into_iter().any(|record| {
record.read.is_ok_and(|envelope| {
envelope.kind.as_str() == "change-reopened"
&& envelope
.payload
.get("restored_base")
.and_then(serde_json::Value::as_str)
== Some(base)
})
}))
}
fn heads_dir() -> Result<PathBuf> {
Ok(home::root()?.join("change-heads"))
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct HeadRecord {
url: String,
head_branch: String,
at: String,
}
pub(crate) fn record_head(url: &Url, head_branch: &str) {
let written = (|| -> Result<()> {
let directory = heads_dir()?;
home::ensure_dir(&directory)?;
let record = HeadRecord {
url: url.to_string(),
head_branch: head_branch.to_owned(),
at: ids::timestamp(),
};
let text = serde_json::to_string(&record)
.map_err(|e| invalid(format!("a head record could not be written: {e}")))?;
home::atomic_write(
&directory.join(format!("{}.json", ids::short_digest(url.as_str()))),
&text,
)
})();
if let Err(error) = written {
eprintln!("onevcs: warning: the head branch read for {url} is not recorded: {error}");
}
}
pub(crate) fn recorded_head(url: &Url) -> Result<Option<String>> {
let mut candidates: Vec<(String, String)> = status::opened_heads(url.as_str())?;
let path = heads_dir()?.join(format!("{}.json", ids::short_digest(url.as_str())));
match std::fs::read_to_string(&path) {
Ok(text) => {
let record = serde_json::from_str::<HeadRecord>(&text)
.ok()
.filter(|record| {
record.url == url.as_str()
&& ids::is_timestamp(&record.at)
&& Ref::try_from(record.head_branch.clone()).is_ok()
})
.ok_or_else(|| {
invalid(format!(
"the head record at {} is not one this build writes for {url}; remove \
it, and a standings read of {url} writes it again",
path.display()
))
})?;
candidates.push((record.at, record.head_branch));
}
Err(failure) if failure.kind() == std::io::ErrorKind::NotFound => {}
Err(failure) => {
return Err(invalid(format!(
"the head record at {} could not be read: {failure}",
path.display()
)))
}
}
Ok(candidates
.into_iter()
.max_by(|left, right| left.0.cmp(&right.0))
.map(|(_, branch)| branch))
}