johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! A repository's settings, as its owner changes them from the pages:
//! `dev.knotless.getRepoSettings`, and `updateRepo`, `renameRepo`,
//! `compactRepo`, `transferRepo`, `archiveRepo`, `acceptTransfer` and
//! `deleteRepo`, each
//! through the same code as the command it is named for. Only the owner
//! can make these changes, or, for an organization's repository, an admin
//! who joined it, as the organization through the gateway it delegated to
//! here; renaming, transferring and deleting it stay the organization's
//! own. This
//! is checked here, not only in the pages.
use super::repos::{RepoMade, refused, rewrote, wrote, wrote_of};
use super::stars::logged_in;
use super::*;
use crate::viewer::Viewer;
use axum::Extension;
use knotless::env::Ctx;
use knotless::manage;
use knotless::storage::Storage;
use std::time::Instant;
#[derive(Deserialize)]
pub struct SettingsParams {
repo: String,
}
/// `dev.knotless.getRepoSettings#triagerView`.
#[derive(Serialize)]
pub struct TriagerView {
actor: ActorView,
scopes: Vec<knotless::records::Scope>,
}
impl TriagerView {
pub(super) async fn new(app: &App, t: knotless::records::Triager) -> TriagerView {
TriagerView {
actor: actor(app, &t.did).await,
scopes: t.scopes,
}
}
}
/// `dev.knotless.getRepoSettings`'s answer.
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Settings {
uri: String,
#[serde(skip_serializing_if = "Option::is_none")]
description: Option<String>,
#[serde(skip_serializing_if = "Vec::is_empty")]
links: Vec<knotless::records::ProfileLink>,
#[serde(skip_serializing_if = "Vec::is_empty")]
topics: Vec<knotless::records::Topic>,
#[serde(skip_serializing_if = "Option::is_none")]
default_branch: Option<String>,
rules: Vec<knotless::records::Rule>,
/// Whether it names a runner, which the merge queue needs to land it.
#[serde(skip_serializing_if = "std::ops::Not::not")]
runners: bool,
labels: Vec<knotless::records::Label>,
collaborators: Vec<ActorView>,
triagers: Vec<TriagerView>,
#[serde(skip_serializing_if = "Option::is_none")]
gateway: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
transfer: Option<RepoViewBasic>,
/// The gateway its owner delegated writes to from these pages, which
/// sharing it for writing names: none until they have.
#[serde(skip_serializing_if = "Option::is_none")]
delegated: Option<String>,
/// For a private repository, to its owner, the members who read it but
/// do not push to it.
#[serde(skip_serializing_if = "Option::is_none")]
readers: Option<Vec<ActorView>>,
/// The accounts its rules name as gates, each once, in the order
/// they are first named: for the pages to show by handle.
#[serde(skip_serializing_if = "Vec::is_empty")]
gates: Vec<ActorView>,
}
/// The repository `repo` names, as a URL by its owner's DID.
fn url_of(repo: &Repo) -> RepoUrl {
RepoUrl {
authority: repo.did.clone(),
name: repo.name.clone(),
}
}
/// `did`'s repository `name` under that name elsewhere, as views name it.
async fn basic(app: &App, to: &RepoRef) -> RepoViewBasic {
RepoViewBasic {
did: to.did().to_string(),
handle: handle(app, to.did()).await,
name: to.name().to_string(),
}
}
/// `dev.knotless.getRepoSettings`.
pub async fn get_repo_settings(
State(app): State<Arc<App>>,
Query(p): Query<SettingsParams>,
) -> Result<Response> {
let repo = repo::found(&app, &p.repo).await?;
// As the index has it, which a write from the pages updates at once,
// else as the mirror last synced.
let record = repo.record(&app).unwrap_or(repo.synced.clone());
let collaborators = record.collaborators.iter().map(|c| actor(&app, &c.did));
let collaborators = futures::future::join_all(collaborators).await;
let mut gated: Vec<&str> = Vec::new();
let named = (record.rules.iter()).filter_map(|r| r.reviews.as_ref());
for gate in named.flat_map(|r| &r.gates) {
if !gated.contains(&gate.did.as_str()) {
gated.push(&gate.did);
}
}
let gates = futures::future::join_all(gated.into_iter().map(|did| actor(&app, did))).await;
// Who reads a private one but does not push, as its record says, which
// sharing it writes and the appview is shown at once: its owner's word,
// shown to its owner alone, as only they may list its space's members.
let mut readers = None;
if repo.in_space.is_some() {
let owner = crate::private::viewer().is_some_and(|v| v == repo.did);
let writers: Vec<&str> = record.writers().collect();
let named = record.readers.iter().map(|r| r.did.as_str());
let mut read: Vec<&str> = Vec::new();
for did in named.chain(record.runners.iter().map(|r| r.did.as_str())) {
if owner && !writers.contains(&did) && !read.contains(&did) {
read.push(did);
}
}
readers =
Some(futures::future::join_all(read.into_iter().map(|did| actor(&app, did))).await);
}
let mut triagers = Vec::new();
for t in record.triagers {
triagers.push(TriagerView::new(&app, t).await);
}
let transfer = match &record.transfer {
Some(to) => Some(basic(&app, to).await),
None => None,
};
let owners = crate::login::ctx_for(&app, &repo.did);
let delegated = knotless::gateway::delegated(&owners, &repo.did);
Ok(answer(Settings {
uri: repo.uri(),
description: record.description,
links: record.links,
topics: record.topics,
default_branch: record.default_branch,
rules: record.rules,
runners: !record.runners.is_empty(),
labels: record.labels,
collaborators,
triagers,
gateway: record.gateway,
transfer,
delegated,
readers,
gates,
}))
}
/// The most repositories [`Kept`] keeps for: once full, it starts again,
/// as reading one again costs only that page a wait.
const KEPT_MOST: usize = 4096;
/// What reading a repository's PDS found, by two names for it (a public
/// one's owner and key; a private one's space, and none), with how
/// far it had moved then and when it was read: good until it moves on,
/// or, unless `watching` says something tells of it moving, for `ttl`.
pub(crate) struct Kept<T>(Mutex<HashMap<(String, String), Found<T>>>);
/// What was read, with how far its repository had moved, and when.
type Found<T> = (u64, Instant, T);
/// Each repository's storage as last counted ([`get_repo_storage`]).
pub(crate) type Counts = Kept<Storage>;
impl<T> Default for Kept<T> {
fn default() -> Self {
Kept(Mutex::default())
}
}
impl<T: Clone> Kept<T> {
/// What was read of `first` and `second`, if it has not moved since,
/// `seen` being how far it has now, and it is not older than `ttl`
/// when `watching` is unset.
fn get(
&self,
first: &str,
second: &str,
seen: u64,
watching: bool,
ttl: Duration,
) -> Option<T> {
let kept = self.0.lock().unwrap();
let (then, at, found) = kept.get(&(first.to_string(), second.to_string()))?;
let fresh = *then == seen && (watching || at.elapsed() < ttl);
fresh.then(|| found.clone())
}
fn put(&self, first: &str, second: &str, seen: u64, found: T, most: usize) {
let mut kept = self.0.lock().unwrap();
if kept.len() >= most {
kept.clear();
}
kept.insert((first.into(), second.into()), (seen, Instant::now(), found));
}
/// Has all that was read under `first` read again, after a write here
/// that nothing has told of yet.
pub(crate) fn forget(&self, first: &str) {
self.0.lock().unwrap().retain(|(f, _), _| f != first);
}
}
/// What a repository's storage is kept under: a public one's owner and
/// key, or a private one's space.
fn counted_as(repo: &Repo) -> (String, String) {
match &repo.in_space {
Some(space) => (space.clone(), String::new()),
None => (repo.did.clone(), repo.key.clone()),
}
}
/// `dev.knotless.getRepoStorage`: counted from its pack records, which a
/// sync no longer reads, so apart from its settings, and kept until the
/// repository moves: a public one's as the firehose tells of it, a
/// private one's as a sync of its space finds it.
pub async fn get_repo_storage(
State(app): State<Arc<App>>,
Query(p): Query<SettingsParams>,
) -> Result<Response> {
let repo = repo::found(&app, &p.repo).await?;
let url = url_of(&repo);
// A private one is read as the member asking, whom finding it proved
// one; a public one as this server.
let private = crate::private::viewer().filter(|_| repo.in_space.is_some());
let reader = match &private {
Some(viewer) => crate::login::ctx_for(&app, viewer),
None => app.ctx.clone(),
};
// How far it has moved, taken before counting, so a move meanwhile
// has it counted again next time.
let (first, second) = counted_as(&repo);
let (seen, watching) = match &repo.in_space {
Some(space) => (app.moves_of(space), app.space_followed(space)),
None => (
app.stamp(&repo.did, Some(&repo.key)).1,
app.watching.load(Ordering::Relaxed),
),
};
let kept = app.counts.get(&first, &second, seen, watching, app.ttl);
if let Some(storage) = kept {
return Ok(answer(storage));
}
let counted = knotless::storage::storage(&reader, &url).await;
let storage = counted.map_err(AppError::upstream)?;
app.counts
.put(&first, &second, seen, storage.clone(), KEPT_MOST);
Ok(answer(storage))
}
/// The repository `uri` names, for a change only its owner, logged in to
/// the pages, may make, `act`; and a context acting as them.
pub(super) async fn owner_only(
app: &App,
viewer: Viewer,
headers: &HeaderMap,
uri: &str,
act: &str,
) -> Result<(Repo, Ctx)> {
let did = logged_in(app, viewer, headers, act)?;
let repo = repo::found(app, uri).await?;
if repo.did != did {
let why = format!("only {}/{}'s owner can {act}", repo.authority, repo.name);
return Err(Shown(StatusCode::FORBIDDEN, why));
}
Ok((repo, crate::login::ctx_for(app, &did)))
}
/// The repository `uri` names, for a change to its settings, `act`, that
/// its owner may make, or an admin who joined the organization that owns
/// it; and a context acting as the owner: for an admin, the organization's
/// session that its gateway, this server, keeps. An admin asks with the
/// pages' login, or with a service token from their own PDS for the
/// method, as `knotless` does; a token, which the viewer middleware checked,
/// is only ever taken as asking for the gateway to write, never for the
/// pages' session of whoever it names.
pub(super) async fn owned(
app: &App,
viewer: Viewer,
headers: &HeaderMap,
uri: &str,
act: &str,
) -> Result<(Repo, Ctx)> {
// As the viewer middleware, which verifies only a bearer token.
let token = headers
.get(header::AUTHORIZATION)
.and_then(|v| v.to_str().ok())
.is_some_and(|v| v.starts_with("Bearer "));
let did = match viewer.0.clone().filter(|_| token) {
Some(did) => did,
None => logged_in(app, viewer, headers, act)?,
};
let repo = repo::found(app, uri).await?;
if repo.did == did && !token {
return Ok((repo, crate::login::ctx_for(app, &did)));
}
if !app.org_admits(&repo.did, &did) {
let why = format!("only {}/{}'s owner can {act}", repo.authority, repo.name);
return Err(Shown(StatusCode::FORBIDDEN, why));
}
crate::gateway::organization(app, &repo.did).await?;
Ok((repo, app.ctx.clone()))
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct UpdateInput {
repo: String,
description: Option<String>,
links: Option<Vec<knotless::records::ProfileLink>>,
topics: Option<Vec<knotless::records::Topic>>,
default_branch: Option<String>,
}
/// `dev.knotless.updateRepo`'s answer.
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Described {
#[serde(skip_serializing_if = "Option::is_none")]
description: Option<String>,
#[serde(skip_serializing_if = "Vec::is_empty")]
links: Vec<knotless::records::ProfileLink>,
#[serde(skip_serializing_if = "Vec::is_empty")]
topics: Vec<knotless::records::Topic>,
#[serde(skip_serializing_if = "Option::is_none")]
default_branch: Option<String>,
}
/// `dev.knotless.updateRepo`.
pub async fn update_repo(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
headers: HeaderMap,
axum::Json(input): axum::Json<UpdateInput>,
) -> Result<Response> {
let act = "change its settings";
let (repo, ctx) = owned(&app, viewer, &headers, &input.repo, act).await?;
let change = manage::Description {
description: input.description.as_deref(),
links: input.links,
topics: input
.topics
.map(|t| t.into_iter().map(|t| t.name).collect()),
default_branch: input.default_branch.as_deref(),
};
let url = url_of(&repo);
let changed = manage::describe(&ctx, &url, change).await;
let now = changed.map_err(refused)?;
rewrote(&app, &repo).await?;
Ok(answer(Described {
description: now.description,
links: now.links,
topics: now.topics,
default_branch: now.default_branch,
}))
}
#[derive(Deserialize)]
pub struct RenameInput {
repo: String,
name: String,
}
/// `dev.knotless.renameRepo`.
pub async fn rename_repo(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
headers: HeaderMap,
axum::Json(input): axum::Json<RenameInput>,
) -> Result<Response> {
// Renaming it leaves a redirect, which an organization's grant to its
// gateway does not reach: its own, as moving and deleting it are.
let (repo, ctx) = owner_only(&app, viewer, &headers, &input.repo, "rename it").await?;
let renamed = manage::rename(&ctx, &url_of(&repo), input.name.trim()).await;
let to = renamed.map_err(refused)?;
let ops = [
("delete", nsid::REPO, repo.name.as_str()),
("create", nsid::REPO, to.name.as_str()),
("create", nsid::REDIRECT, repo.name.as_str()),
];
wrote_of(&app, &repo, &ops).await?;
Ok(answer(RepoMade::new(&repo.did, &to.name)))
}
#[derive(Deserialize)]
pub struct RepoInput {
repo: String,
}
/// `dev.knotless.compactRepo`'s answer.
#[derive(Serialize)]
pub struct Compacted {
packs: usize,
folded: usize,
claims: usize,
refs: usize,
}
/// `dev.knotless.compactRepo`.
pub async fn compact_repo(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
headers: HeaderMap,
axum::Json(input): axum::Json<RepoInput>,
) -> Result<Response> {
let (repo, ctx) = owned(&app, viewer, &headers, &input.repo, "compact it").await?;
let compacted = knotless::compact::compact_repo(&ctx, &url_of(&repo)).await;
let c = compacted.map_err(refused)?;
app.counts.forget(&counted_as(&repo).0);
wrote(&app, &repo.did, &[]).await?;
Ok(answer(Compacted {
packs: c.before,
folded: c.removed,
claims: c.claims,
refs: c.refs,
}))
}
#[derive(Deserialize)]
pub struct TransferInput {
repo: String,
to: Option<String>,
name: Option<String>,
}
/// `dev.knotless.transferRepo`'s answer.
#[derive(Serialize)]
pub struct Offered {
#[serde(skip_serializing_if = "Option::is_none")]
transfer: Option<RepoViewBasic>,
}
/// `dev.knotless.transferRepo`.
pub async fn transfer_repo(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
headers: HeaderMap,
axum::Json(input): axum::Json<TransferInput>,
) -> Result<Response> {
let (repo, ctx) = owner_only(&app, viewer, &headers, &input.repo, "transfer it").await?;
let url = url_of(&repo);
let name = input
.name
.as_deref()
.map(str::trim)
.filter(|n| !n.is_empty());
let to = match input.to.as_deref().map(str::trim) {
Some(who) => {
let at = format!("knotless://{who}/{}", name.unwrap_or(&repo.name));
Some(RepoUrl::parse(&at).map_err(refused)?)
}
None => None,
};
let offered = manage::transfer(&ctx, &url, to.as_ref()).await;
offered.map_err(refused)?;
rewrote(&app, &repo).await?;
// Read back as its owner, who can read a private one's record.
let remote = Remote::connect(&ctx, &url).await;
let now = remote.map_err(AppError::upstream)?.repo().await;
let offer = now
.map_err(AppError::upstream)?
.and_then(|r| r.value.transfer);
let transfer = match &offer {
Some(to) => Some(basic(&app, to).await),
None => None,
};
Ok(answer(Offered { transfer }))
}
#[derive(Deserialize)]
pub struct ArchiveInput {
repo: String,
#[serde(default)]
unarchive: bool,
}
/// `dev.knotless.archiveRepo`'s answer.
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Archived {
#[serde(skip_serializing_if = "Option::is_none")]
archived_at: Option<String>,
}
/// `dev.knotless.archiveRepo`.
pub async fn archive_repo(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
headers: HeaderMap,
axum::Json(input): axum::Json<ArchiveInput>,
) -> Result<Response> {
let (repo, ctx) = owned(&app, viewer, &headers, &input.repo, "archive it").await?;
let archived = manage::archive(&ctx, &url_of(&repo), !input.unarchive).await;
let archived_at = archived.map_err(refused)?;
rewrote(&app, &repo).await?;
Ok(answer(Archived { archived_at }))
}
/// `dev.knotless.acceptTransfer`.
pub async fn accept_transfer(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
headers: HeaderMap,
axum::Json(input): axum::Json<RepoInput>,
) -> Result<Response> {
let did = logged_in(&app, viewer, &headers, "take it up")?;
let repo = repo::found(&app, &input.repo).await?;
let ctx = crate::login::ctx_for(&app, &did);
let accepted = forge::accept(&ctx, &url_of(&repo)).await;
let to = accepted.map_err(refused)?;
wrote(&app, &did, &[("create", nsid::REPO, to.name.as_str())]).await?;
// Readers follow it from the old account now.
app.touch(&repo.did);
Ok(answer(RepoMade::new(&did, &to.name)))
}
/// `dev.knotless.deleteRepo`'s answer.
#[derive(Serialize)]
pub struct Deleted {
uri: String,
}
/// `dev.knotless.deleteRepo`.
pub async fn delete_repo(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
headers: HeaderMap,
axum::Json(input): axum::Json<RepoInput>,
) -> Result<Response> {
let (repo, ctx) = owner_only(&app, viewer, &headers, &input.repo, "delete it").await?;
manage::delete(&ctx, &url_of(&repo))
.await
.map_err(refused)?;
let gone = [("delete", nsid::REPO, repo.name.as_str())];
wrote_of(&app, &repo, &gone).await?;
// Read off the index at once, its record gone from it: what was kept
// of it goes too, once its host says no record has its key.
crate::accounts::repo_deleted(&app, &repo.did, &repo.key);
Ok(answer(Deleted { uri: repo.uri() }))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn kept_until_the_repository_moves() {
let counts = Counts::default();
let one = Storage {
packs: 1,
..Storage::default()
};
let day = Duration::from_secs(86_400);
assert_eq!(counts.get("d", "k", 0, true, day), None);
counts.put("d", "k", 3, one.clone(), 2);
assert_eq!(counts.get("d", "k", 3, true, day), Some(one.clone()));
// Moved on, or nothing watched for longer than `ttl`, and it is
// read again.
assert_eq!(counts.get("d", "k", 4, true, day), None);
assert_eq!(counts.get("d", "k", 3, false, Duration::ZERO), None);
assert_eq!(counts.get("d", "k", 3, false, day), Some(one.clone()));
// Another repository's is its own.
assert_eq!(counts.get("d", "j", 3, true, day), None);
// Forgotten, all under its first name go.
counts.put("d", "j", 3, one.clone(), 4);
counts.put("e", "k", 3, one.clone(), 4);
counts.forget("d");
assert_eq!(counts.get("d", "k", 3, true, day), None);
assert_eq!(counts.get("d", "j", 3, true, day), None);
assert_eq!(counts.get("e", "k", 3, true, day), Some(one.clone()));
// Full, it starts again.
counts.put("d", "a", 0, one.clone(), 2);
counts.put("d", "b", 0, one.clone(), 2);
counts.put("d", "c", 0, one.clone(), 2);
assert_eq!(counts.get("d", "a", 0, true, day), None);
assert_eq!(counts.get("d", "c", 0, true, day), Some(one));
}
}