johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! Landing pulls and running their checks, from the pages:
//! `dev.knotless.getMergeStatus`, `mergePull`, `revertPull`, `queuePull`,
//! `leaveQueue`, `setAutoMerge`, `retargetPull`, `requestRerun`, `listWorkflows`,
//! `listRunners`, `addRunner` and `removeRunner`. Each write is made as the
//! account logged in to the pages, with the session the server keeps for
//! it ([`crate::login`]), through the same code as `knotless pull merge`,
//! `revert`, `queue`, `retarget`, `rerun` and `runners`.
use super::repo::found;
use super::repos::refused;
use super::stars::logged_in;
use super::*;
use crate::viewer::Viewer;
use anyhow::Context;
use axum::Extension;
use knotless::checks::Check;
use knotless::env::Ctx;
use knotless::landing;
use knotless::manage;
use knotless::records::RerunVar;
/// The repository as the `knotless` crate names it.
fn url_of(repo: &Repo) -> RepoUrl {
RepoUrl {
authority: repo.did.clone(),
name: repo.name.clone(),
}
}
/// The account logged in to the pages, for a write from them that `act`s,
/// acting as itself, and the repository `uri` names.
async fn acting(
app: &App,
viewer: Viewer,
headers: &HeaderMap,
act: &str,
uri: &str,
) -> Result<(Ctx, Repo)> {
let did = logged_in(app, viewer, headers, act)?;
let repo = found(app, uri).await?;
Ok((crate::login::ctx_for(app, &did), repo))
}
/// Indexes, at once, the newest queue and state records the repository's
/// owner and collaborators wrote about it, as their repos have them, so
/// the page after a merge, or a pull queued or retargeted, shows it.
async fn freshen(app: &App, repo: &Repo) -> anyhow::Result<()> {
let index = app.index.as_ref().context("no index")?;
let t = forge::Target::open(&app.ctx, &url_of(repo)).await?;
for e in t.fresh_entries(&app.ctx).await {
let (did, collection, rkey) = forge::parse_uri(&e.uri)?;
index.put(did, collection, rkey, &e.value)?;
}
Ok(())
}
/// A pull of a repository, as a query or a write names it.
#[derive(Deserialize)]
pub struct PullParams {
repo: String,
pull: String,
}
/// `dev.knotless.getMergeStatus`: for someone who asks, as the pages' login
/// or a service token says, read as the pages read the repository for
/// them: a private one as that viewer, a public one as this server.
pub async fn get_merge_status(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
Query(p): Query<PullParams>,
) -> Result<Response> {
let why = "log in to see whether a pull can merge, and what you may do to it";
let did = viewer
.0
.ok_or(Shown(StatusCode::UNAUTHORIZED, why.into()))?;
let repo = found(&app, &p.repo).await?;
let mut found = mergeable(&app, &repo, &p.pull, &did).await?;
if found.may.queue && undelegated(&app, &repo) {
found.queue_blocked = Some("undelegated");
}
// Who set it, and who pushed since, as the pages show an account.
let mut said = serde_json::json!(found);
for key in ["by", "pushedBy"] {
if let Some(did) = said["autoMerge"][key].as_str().map(str::to_string) {
said["autoMerge"][key] = serde_json::json!(super::actor::actor(&app, &did).await);
}
}
Ok(answer(said))
}
/// Whether the open pull `pull` on `repo` can merge now, as
/// [`landing::mergeable`] decides it, read from what this server holds
/// rather than from the PDSes again: the repository as its mirror last
/// synced, the pull, its reviews and its checks as the index has them.
async fn mergeable(app: &App, repo: &Repo, pull: &str, who: &str) -> Result<landing::Mergeable> {
let mut made = assembled(app, repo).await?;
if !made.items.iter().any(|i| i.uri == pull) {
// Opened a moment ago, perhaps, before the index heard of it.
super::threads::reindex(app, repo, pull).await?;
made = assembled(app, repo).await?;
}
// As the mirror last synced, which a write from the pages syncs again.
let record = repo.synced.clone();
let (entries, items) = (&made.entries, &made.items);
let thread = forge::thread_in(&record, &repo.did, entries, items, pull);
let item = thread.map_err(not_found)?.item;
let (Some(base), Some(head), "open") = (&item.base, &item.head, item.state.as_str()) else {
let why = format!("{pull} is {}", item.state);
return Err(Shown(StatusCode::BAD_REQUEST, why));
};
let (git, base_sha, head_sha, _) = super::pull_range(app, repo, &item, base, head).await?;
let sides = forge::Sides {
git,
refname: format!("refs/heads/{base}"),
base: base_sha,
head: head_sha,
};
let url = url_of(repo);
let facts = forge::Facts {
url: &url,
owner: &repo.did,
key: &repo.key,
record: &record,
private: repo.in_space.is_some(),
};
let ctx = &repo.reader;
let gauged = landing::gauge(ctx, facts, items, &item, &sides, who).await;
let gauged = gauged.map_err(AppError::upstream)?;
let checks = match gauged.workflows.is_empty() {
true => Vec::new(),
false => checks_of(app, repo, &record, pull)?,
};
let words = knotless::fix::Words::Pages;
Ok(landing::decide(ctx, facts, &item, &sides, gauged, &checks, who, words).await)
}
/// The checks `record`'s runners ran of `pull` on `repo`, newest first, as
/// the index its pulls were read from has them: as many as a merge goes by, those of its head now
/// and of the heads just before it.
fn checks_of(app: &App, repo: &Repo, record: &RepoRecord, pull: &str) -> Result<Vec<Check>> {
let index = super::indexed(app, repo).expect("its pulls were read from an index");
let runners: Vec<String> = record.runners.iter().map(|r| r.did.clone()).collect();
let filter = CheckFilter {
pull: Some(pull),
..CheckFilter::default()
};
let found = index.checks(&repo.did, &repo.key, &runners, filter, CHECKS_READ, None);
let (entries, _) = found
.map_err(anyhow::Error::from)
.map_err(AppError::internal)?;
let checks = entries.into_iter().filter_map(|e| {
Some(Check {
record: serde_json::from_value(e.value).ok()?,
uri: e.uri,
runner: e.author,
log_text: None,
space: repo.in_space.clone(),
})
});
Ok(checks.collect())
}
/// How many of a pull's checks, newest first, its merge status reads: a
/// push runs one of each of its workflows, so these cover its head and
/// the pushes before it many times over.
const CHECKS_READ: usize = 200;
/// Whether the merge queue of `repo` lands through this server's gateway,
/// as no runner it names may push and it records no other gateway, while
/// its owner has not delegated writes here: then it lands nothing.
fn undelegated(app: &App, repo: &Repo) -> bool {
let record = &repo.synced;
let lands = app.gateway.as_ref().zip(record.lander(&repo.did));
lands.is_some_and(|(gw, lander)| {
let pushes = lander == repo.did || record.is_collaborator(lander);
let here = record.gateway.as_deref().is_none_or(|g| g == gw.did);
!pushes && here && !crate::gateway::writes_for(app, &repo.did)
})
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct MergeInput {
repo: String,
pull: String,
#[serde(default)]
keep_branch: bool,
}
/// `dev.knotless.mergePull#closedIssue`.
#[derive(Serialize)]
pub struct ClosedIssue {
uri: String,
}
/// `dev.knotless.mergePull`'s answer.
#[derive(Serialize)]
pub struct Merged {
commit: String,
#[serde(skip_serializing_if = "Option::is_none")]
deleted: Option<String>,
closed: Vec<ClosedIssue>,
}
/// `dev.knotless.mergePull`: as `knotless pull merge`, waiting for nothing.
pub async fn merge_pull(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
headers: HeaderMap,
axum::Json(input): axum::Json<MergeInput>,
) -> Result<Response> {
let (ctx, repo) = acting(&app, viewer, &headers, "merge", &input.repo).await?;
let (url, gate) = (url_of(&repo), forge::Gate::default());
let merged = forge::merge(&ctx, &url, &input.pull, input.keep_branch, gate).await;
let merged = merged.map_err(refused)?;
// The base branch moved: its mirror is read again, as after a push.
app.touch(&repo.did);
let _ = freshen(&app, &repo).await;
let closed = merged.closed.into_iter().map(|uri| ClosedIssue { uri });
Ok(answer(Merged {
commit: merged.sha,
deleted: merged.deleted,
closed: closed.collect(),
}))
}
#[derive(Deserialize)]
pub struct RevertInput {
repo: String,
pull: String,
reason: Option<String>,
}
/// `dev.knotless.revertPull`'s answer.
#[derive(Serialize)]
pub struct Reverted {
uri: String,
branch: String,
commit: String,
}
/// `dev.knotless.revertPull`: as `knotless pull revert`, indexing the pull it
/// opens at once, so the page it lands on shows it.
pub async fn revert_pull(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
headers: HeaderMap,
axum::Json(input): axum::Json<RevertInput>,
) -> Result<Response> {
let (ctx, repo) = acting(&app, viewer, &headers, "revert a pull", &input.repo).await?;
let reason = input.reason.as_deref();
let reverted = knotless::revert::revert(&ctx, &url_of(&repo), &input.pull, reason).await;
let reverted = reverted.map_err(refused)?;
// The new branch is read again, as after a push.
app.touch(&repo.did);
super::threads::reindex(&app, &repo, &reverted.pull).await?;
Ok(answer(Reverted {
uri: reverted.pull,
branch: reverted.branch,
commit: reverted.commit,
}))
}
/// `dev.knotless.queuePull`: as `knotless pull merge --queue`.
pub async fn queue_pull(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
headers: HeaderMap,
axum::Json(input): axum::Json<PullParams>,
) -> Result<Response> {
let (ctx, repo) = acting(&app, viewer, &headers, "queue a pull", &input.repo).await?;
if undelegated(&app, &repo) {
let why = "its merge queue lands through this server once its owner delegates writes to it, under Delegated writes in their settings";
return Err(Shown(StatusCode::BAD_REQUEST, why.into()));
}
let queued = knotless::queue::enqueue(&ctx, &url_of(&repo), &input.pull).await;
let place = queued.map_err(refused)?;
let _ = freshen(&app, &repo).await;
Ok(answer(place))
}
#[derive(Deserialize)]
pub struct AutoMergeInput {
repo: String,
pull: String,
#[serde(default)]
undo: bool,
}
/// `dev.knotless.setAutoMerge`: as `knotless pull merge --auto`, or with
/// undo `--auto --undo`.
pub async fn set_auto_merge(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
headers: HeaderMap,
axum::Json(input): axum::Json<AutoMergeInput>,
) -> Result<Response> {
let act = "set a pull to merge when its checks pass";
let (ctx, repo) = acting(&app, viewer, &headers, act, &input.repo).await?;
let url = url_of(&repo);
let commit = match input.undo {
true => knotless::auto_merge::cancel(&ctx, &url, &input.pull)
.await
.map(|()| None),
false => knotless::auto_merge::set(&ctx, &url, &input.pull)
.await
.map(Some),
};
let commit = commit.map_err(refused)?;
// Its author's cancelling is in their repo, which freshening skips.
let _ = freshen(&app, &repo).await;
super::threads::reindex(&app, &repo, &input.pull).await?;
let said = commit.map(|commit| serde_json::json!({"commit": commit}));
Ok(answer(said.unwrap_or(serde_json::json!({}))))
}
/// `dev.knotless.leaveQueue`: as `knotless queue leave`.
pub async fn leave_queue(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
headers: HeaderMap,
axum::Json(input): axum::Json<PullParams>,
) -> Result<Response> {
let act = "take a pull out of the queue";
let (ctx, repo) = acting(&app, viewer, &headers, act, &input.repo).await?;
let left = knotless::queue::leave(&ctx, &url_of(&repo), &input.pull).await;
left.map_err(refused)?;
let _ = freshen(&app, &repo).await;
Ok(answer(serde_json::json!({})))
}
#[derive(Deserialize)]
pub struct RetargetInput {
repo: String,
pull: String,
base: Option<String>,
head: Option<String>,
}
/// `dev.knotless.retargetPull`: as `knotless pull retarget`.
pub async fn retarget_pull(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
headers: HeaderMap,
axum::Json(input): axum::Json<RetargetInput>,
) -> Result<Response> {
let (ctx, repo) = acting(&app, viewer, &headers, "retarget a pull", &input.repo).await?;
let (base, head) = (input.base.as_deref(), input.head.as_deref());
let moved = forge::retarget(&ctx, &url_of(&repo), &input.pull, base, head).await;
moved.map_err(refused)?;
let _ = freshen(&app, &repo).await;
Ok(answer(serde_json::json!({})))
}
#[derive(Deserialize)]
pub struct RerunInput {
repo: String,
#[serde(rename = "ref")]
refname: Option<String>,
pull: Option<String>,
workflow: Option<String>,
#[serde(default)]
env: Vec<RerunVar>,
}
/// `dev.knotless.requestRerun`: as `knotless run rerun`.
pub async fn request_rerun(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
headers: HeaderMap,
axum::Json(input): axum::Json<RerunInput>,
) -> Result<Response> {
let (ctx, repo) = acting(&app, viewer, &headers, "run checks", &input.repo).await?;
let (refname, pull) = (input.refname.as_deref(), input.pull.as_deref());
let workflow = input.workflow.as_deref();
let asked = manage::rerun(&ctx, &url_of(&repo), refname, pull, workflow, input.env).await;
let uri = asked.map_err(refused)?;
Ok(answer(serde_json::json!({ "uri": uri })))
}
#[derive(Deserialize)]
pub struct WorkflowsParams {
repo: String,
#[serde(rename = "ref")]
refname: Option<String>,
}
/// `dev.knotless.listWorkflows`: as `knotless workflow list`.
pub async fn list_workflows(
State(app): State<Arc<App>>,
Query(p): Query<WorkflowsParams>,
) -> Result<Response> {
let repo = found(&app, &p.repo).await?;
let name = p.refname.as_deref();
let listed = landing::workflows(&repo.reader, &url_of(&repo), name).await;
let (refname, workflows) = listed.map_err(not_found)?;
Ok(answer(
serde_json::json!({ "ref": refname, "workflows": workflows }),
))
}
/// `dev.knotless.listRunners#runners`.
#[derive(Serialize)]
pub struct Runners {
runners: Vec<ActorView>,
#[serde(skip_serializing_if = "Option::is_none")]
warning: Option<String>,
/// The minutes they count for the repository's owner.
#[serde(skip_serializing_if = "Vec::is_empty")]
minutes: Vec<MinutesView>,
/// The keys they publish to seal secrets to, of those that publish one.
#[serde(skip_serializing_if = "Vec::is_empty")]
keys: Vec<RunnerKey>,
}
/// `dev.knotless.listRunners#runnerKey`.
#[derive(Serialize)]
pub struct RunnerKey {
runner: String,
key: String,
}
/// The answer naming `runners`, the repository `repo`'s, with what
/// they count for its owner and the keys they publish.
async fn runners_answer(
app: &App,
repo: &Repo,
runners: Vec<ActorView>,
warning: Option<String>,
) -> Response {
let dids: Vec<String> = runners.iter().map(|r| r.did.clone()).collect();
let minutes = minutes(app, &repo.uri(), &repo.did, &dids).await;
let keys = app.index.as_ref().and_then(|i| i.runner_keys(&dids).ok());
let keys = keys.unwrap_or_default().into_iter();
answer(Runners {
runners,
warning,
minutes,
keys: keys
.map(|(runner, key)| RunnerKey { runner, key })
.collect(),
})
}
/// The runners `url` names, read as `ctx` reads it, each as an account.
async fn runners(app: &App, ctx: &Ctx, url: &RepoUrl) -> Result<Vec<ActorView>> {
let dids = manage::runners(ctx, url)
.await
.map_err(AppError::upstream)?;
Ok(futures::future::join_all(dids.iter().map(|did| actor(app, did))).await)
}
#[derive(Deserialize)]
pub struct RepoParams {
repo: String,
}
/// `dev.knotless.listRunners`: as `knotless runner list`.
pub async fn list_runners(
State(app): State<Arc<App>>,
Query(p): Query<RepoParams>,
) -> Result<Response> {
let repo = found(&app, &p.repo).await?;
let runners = runners(&app, &repo.reader, &url_of(&repo)).await?;
Ok(runners_answer(&app, &repo, runners, None).await)
}
#[derive(Deserialize)]
pub struct RunnerInput {
repo: String,
runner: String,
}
/// `dev.knotless.addRunner`: as `knotless runner add`.
pub async fn add_runner(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
headers: HeaderMap,
axum::Json(input): axum::Json<RunnerInput>,
) -> Result<Response> {
naming(&app, viewer, &headers, input, true).await
}
/// `dev.knotless.removeRunner`: as `knotless runner remove`.
pub async fn remove_runner(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
headers: HeaderMap,
axum::Json(input): axum::Json<RunnerInput>,
) -> Result<Response> {
naming(&app, viewer, &headers, input, false).await
}
/// Names `input.runner` among the repository's runners, or with `add`
/// false no longer, and answers with those it names after.
async fn naming(
app: &Arc<App>,
viewer: Viewer,
headers: &HeaderMap,
input: RunnerInput,
add: bool,
) -> Result<Response> {
// Its owner's, or an organization's admin's, as its other settings.
let act = "change runners";
let (repo, ctx) = super::settings::owned(app, viewer, headers, &input.repo, act).await?;
let url = url_of(&repo);
let named = manage::set_runner(&ctx, &url, &input.runner, add).await;
let warning = named.map_err(refused)?;
// What waits for its checks reads the runners from its record as
// synced: synced again as it is next read, as after any other write
// from the pages.
app.touch(&repo.did);
let runners = runners(app, &ctx, &url).await?;
Ok(runners_answer(app, &repo, runners, warning).await)
}