~ knotless Search/
↑↓ move↵ open⇧↵ all results

chadtmiller.com/knotless
Switch repository
Loading…

git that lives in your Atmosphere account

knotless.dev
1 collaborator

knotless 0.13.0 Latest 2 packages, 1 image

1
10

The pages list the shared steps repositories publish

Merged chadtmiller.com asked to merge into main from chadtmiller.com/knotless:web/steps
crates/knotless-serve/src/index/shared.rsnew+284−0
View file
@@ -0,0 +1,284 @@
//! What repositories share for others to use, kept by kind, with the
//! repositories that use each: the directory the pages show
//! (`dev.knotless.listSteps`). Shared steps are the one kind yet: a
//! release of the package `@step/<name>` whose one file is a step file
//! ([`knotless::uses`]). Each such release is kept as it is indexed, its
//! file read after ([`crate::shared::follow`]), and shown once it is
//! found to be one. Who uses each is read from the lock on each public
//! repository's default branch, as what it is written in is counted
//! ([`crate::languages`]).
use super::*;
/// The kinds of thing shared, as `shared.kind` names them.
pub const STEP: &str = "step";
/// The tables: each release of something shared, by its record's URI,
/// with its file as read, as JSON, once it is (`state` 0 until then, 1
/// for one that is what its package says, 2 for one that is not); and
/// each repository's uses of them, a row for each thing it uses, at
/// whatever version.
pub(super) const TABLES: &str = "CREATE TABLE IF NOT EXISTS shared (
uri TEXT PRIMARY KEY, kind TEXT NOT NULL, did TEXT NOT NULL, repo_key TEXT NOT NULL,
name TEXT NOT NULL, created_at TEXT NOT NULL, state INTEGER NOT NULL, value TEXT);
CREATE INDEX IF NOT EXISTS shared_name ON shared (kind, state, name, did, repo_key);
CREATE INDEX IF NOT EXISTS shared_repo ON shared (did, repo_key, kind, state, name);
CREATE INDEX IF NOT EXISTS shared_unread ON shared (state) WHERE state = 0;
CREATE TABLE IF NOT EXISTS shared_use (
did TEXT NOT NULL, key TEXT NOT NULL, kind TEXT NOT NULL, of_did TEXT NOT NULL,
of_key TEXT NOT NULL, name TEXT NOT NULL,
PRIMARY KEY (did, key, kind, of_did, of_key, name));
CREATE INDEX IF NOT EXISTS shared_use_of ON shared_use (kind, of_did, of_key, name);";
/// The releases of steps an index from before the table holds, once, as
/// the table is made: kept to be read, as new ones are.
pub(super) const UNREAD: &str = "INSERT OR IGNORE INTO shared
SELECT uri, 'step', did, repo_key, substr(json_extract(value, '$.package'), 7),
created_at, 0, NULL
FROM release
-- scans: release, once, as the table of what is shared is made
WHERE substr(json_extract(value, '$.package'), 1, 6) = '@step/'";
/// The kind and name of what `package` releases, if it is shared: a step,
/// `@step/<name>`, with a name a step may have.
pub fn kind_of(package: &str) -> Option<(&'static str, &str)> {
let name = package.strip_prefix("@step/")?;
knotless::workflow::valid_name(name).then_some((STEP, name))
}
/// A thing shared, by its kind, the owner and key of its repository, and
/// its name.
#[derive(Debug, Clone, PartialEq)]
pub struct Shared {
pub kind: String,
pub did: String,
pub key: String,
pub name: String,
}
/// A release of a thing shared, read: its record's URI, the record, and
/// what was read of its file.
pub type Version = (String, ReleaseRecord, Value);
/// Where a page of things shared starts: after the one it names, by
/// name, owner and key, as `cursor` gives them.
fn after(cursor: Option<&str>, prefix: &str) -> (String, String, String) {
let mut parts = cursor.unwrap_or_default().splitn(3, ' ');
match (parts.next(), parts.next(), parts.next()) {
(Some(name), Some(did), Some(key)) => (name.into(), did.into(), key.into()),
_ => (prefix.into(), String::new(), String::new()),
}
}
/// The cursor a page ending at `last` gives.
pub fn cursor(last: &Shared) -> String {
format!("{} {} {}", last.name, last.did, last.key)
}
impl Index {
/// Keeps the release at `uri` of the thing `package` names, if it is
/// shared, to have its file read; and asks for it to be read. Called
/// as the release is indexed, with its database.
pub(super) fn share(
&self,
db: &Connection,
uri: &str,
did: &str,
release: &ReleaseRecord,
) -> rusqlite::Result<()> {
let Some((kind, name)) = kind_of(&release.package) else {
return Ok(());
};
let kept = db.execute(
"INSERT OR IGNORE INTO shared VALUES (?1, ?2, ?3, ?4, ?5, ?6, 0, NULL)",
params![uri, kind, did, release.repo, name, release.created_at],
);
let _ = self.read.send(uri.to_string());
kept.map(|_| ())
}
/// What [`Index::share`] asks to be read, for the one that reads it:
/// none after the first take.
pub fn reading(&self) -> Option<mpsc::UnboundedReceiver<String>> {
self.reading.lock().unwrap().take()
}
/// The releases of things shared whose files are yet to be read.
pub fn unread(&self) -> rusqlite::Result<Vec<String>> {
let db = self.db.lock().unwrap();
let mut stmt = db.prepare_cached("SELECT uri FROM shared WHERE state = 0")?;
let rows = stmt.query_map([], |row| row.get(0))?;
rows.collect()
}
/// The release at `uri` of a thing shared, with the thing, if the
/// index has both.
pub fn shared_release(&self, uri: &str) -> rusqlite::Result<Option<(Shared, ReleaseRecord)>> {
let db = self.db.lock().unwrap();
let mut stmt = db.prepare_cached(
"SELECT s.kind, s.did, s.repo_key, s.name, r.value FROM shared s
JOIN release r ON r.uri = s.uri WHERE s.uri = ?1",
)?;
let found = stmt.query_row([uri], |row| {
let text: String = row.get(4)?;
let thing = Shared {
kind: row.get(0)?,
did: row.get(1)?,
key: row.get(2)?,
name: row.get(3)?,
};
Ok((thing, text))
});
let found = found.optional()?;
Ok(found.and_then(|(thing, text)| Some((thing, serde_json::from_str(&text).ok()?))))
}
/// Keeps what was read of the release at `uri`'s file: what it is, or
/// none for one that is not what its package says.
pub fn set_shared(&self, uri: &str, value: Option<&Value>) -> rusqlite::Result<()> {
let db = self.db.lock().unwrap();
let (state, text) = match value {
Some(v) => (1, Some(v.to_string())),
None => (2, None),
};
let mut stmt =
db.prepare_cached("UPDATE shared SET state = ?2, value = ?3 WHERE uri = ?1")?;
stmt.execute(params![uri, state, text]).map(|_| ())
}
/// Keeps `used` as what `did`'s repository with `key` uses, in place of
/// what it did.
pub fn set_uses(&self, did: &str, key: &str, used: &[Shared]) -> rusqlite::Result<()> {
let mut db = self.db.lock().unwrap();
let tx = db.transaction()?;
tx.execute(
"DELETE FROM shared_use WHERE did = ?1 AND key = ?2",
[did, key],
)?;
for s in used {
tx.execute(
"INSERT OR IGNORE INTO shared_use VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
params![did, key, s.kind, s.did, s.key, s.name],
)?;
}
tx.commit()
}
/// The key of `did`'s repository named `name`, if the index has it.
pub fn key_named(&self, did: &str, name: &str) -> rusqlite::Result<Option<String>> {
let db = self.db.lock().unwrap();
let mut stmt =
db.prepare_cached("SELECT key FROM repo WHERE did = ?1 AND name = ?2 LIMIT 1")?;
stmt.query_row([did, name], |row| row.get(0)).optional()
}
/// A page of at most `limit` things of `kind` shared, in the order of
/// their names: those of `did`'s repository `key`, or everyone's, whose
/// names start with `prefix`, after the cursor an earlier page gave;
/// and the cursor for the next, if any. One of an account, repository
/// or release taken down, or whose repository the index lacks, is left
/// out.
pub fn shared(
&self,
kind: &str,
repo: Option<(&str, &str)>,
prefix: &str,
limit: u32,
from: Option<&str>,
) -> rusqlite::Result<(Vec<Shared>, Option<String>)> {
let (name, did, key) = after(from, prefix);
let (of_did, of_key) = repo.unzip();
let db = self.db.lock().unwrap();
// Each way in by an index of its own: one repository's, or
// everyone's, by name.
let sql = match repo {
Some(_) => {
"SELECT DISTINCT s.name, s.did, s.repo_key FROM shared s
WHERE s.did = ?7 AND s.repo_key = ?8 AND s.kind = ?1 AND s.state = 1
AND (s.name, s.did, s.repo_key) > (?2, ?3, ?4) AND s.name < ?5
AND s.uri NOT IN down AND s.did NOT IN down
AND EXISTS (SELECT 1 FROM repo r WHERE r.did = s.did
AND r.key = s.repo_key AND r.uri NOT IN down)
ORDER BY s.name, s.did, s.repo_key LIMIT ?6"
}
None => {
"SELECT DISTINCT s.name, s.did, s.repo_key FROM shared s
WHERE ?7 IS NULL AND ?8 IS NULL AND s.kind = ?1 AND s.state = 1
AND (s.name, s.did, s.repo_key) > (?2, ?3, ?4) AND s.name < ?5
AND s.uri NOT IN down AND s.did NOT IN down
AND EXISTS (SELECT 1 FROM repo r WHERE r.did = s.did
AND r.key = s.repo_key AND r.uri NOT IN down)
ORDER BY s.name, s.did, s.repo_key LIMIT ?6"
}
};
let mut stmt = db.prepare_cached(sql)?;
// The names from the prefix up to the last that could start so.
let end = format!("{prefix}{}", char::MAX);
let args = params![kind, name, did, key, end, limit + 1, of_did, of_key];
let thing = |row: &rusqlite::Row| -> rusqlite::Result<Shared> {
Ok(Shared {
kind: kind.to_string(),
name: row.get(0)?,
did: row.get(1)?,
key: row.get(2)?,
})
};
let rows = stmt.query_map(args, thing)?;
let mut found = rows.collect::<rusqlite::Result<Vec<_>>>()?;
let more = found.len() > limit as usize;
found.truncate(limit as usize);
let next = found.last().filter(|_| more).map(cursor);
Ok((found, next))
}
/// Each release of `thing` whose file is one, as read, leaving out
/// those taken down.
pub fn shared_versions(&self, thing: &Shared) -> rusqlite::Result<Vec<Version>> {
let db = self.db.lock().unwrap();
let mut stmt = db.prepare_cached(
"SELECT s.uri, r.value, s.value FROM shared s JOIN release r ON r.uri = s.uri
WHERE s.did = ?1 AND s.repo_key = ?2 AND s.kind = ?3 AND s.state = 1
AND s.name = ?4 AND s.uri NOT IN down",
)?;
let args = params![thing.did, thing.key, thing.kind, thing.name];
let texts = |row: &rusqlite::Row| -> rusqlite::Result<(String, String, String)> {
Ok((row.get(0)?, row.get(1)?, row.get(2)?))
};
let rows = stmt.query_map(args, texts)?;
let rows = rows.collect::<rusqlite::Result<Vec<_>>>()?;
let read = rows.into_iter().filter_map(|(uri, record, value)| {
let record = serde_json::from_str(&record).ok()?;
Some((uri, record, serde_json::from_str(&value).ok()?))
});
Ok(read.collect())
}
/// How many public repositories use `thing`, and at most `limit` of
/// them, by owner and name, leaving out those taken down.
pub fn users(
&self,
thing: &Shared,
limit: u32,
) -> rusqlite::Result<(u64, Vec<(String, String)>)> {
let db = self.db.lock().unwrap();
let args = params![thing.kind, thing.did, thing.key, thing.name, limit];
let mut count = db.prepare_cached(
"SELECT COUNT(*) FROM shared_use u
WHERE u.kind = ?1 AND u.of_did = ?2 AND u.of_key = ?3 AND u.name = ?4
AND u.did NOT IN down AND EXISTS (SELECT 1 FROM repo r
WHERE r.did = u.did AND r.key = u.key AND r.uri NOT IN down)",
)?;
let n = count.query_row(&args[..4], |row| row.get(0))?;
let mut some = db.prepare_cached(
"SELECT u.did, (SELECT r.name FROM repo r WHERE r.did = u.did AND r.key = u.key
AND r.uri NOT IN down LIMIT 1)
FROM shared_use u
WHERE u.kind = ?1 AND u.of_did = ?2 AND u.of_key = ?3 AND u.name = ?4
AND u.did NOT IN down AND EXISTS (SELECT 1 FROM repo r
WHERE r.did = u.did AND r.key = u.key AND r.uri NOT IN down)
LIMIT ?5",
)?;
let rows = some.query_map(args, |row| Ok((row.get(0)?, row.get(1)?)))?;
Ok((n, rows.collect::<rusqlite::Result<Vec<_>>>()?))
}
}
crates/knotless-serve/src/web/steps.rsnew+294−0
View file
@@ -0,0 +1,294 @@
//! Shared steps, as the index keeps what repositories share
//! ([`crate::shared`]): `dev.knotless.listSteps`, the directory, and
//! `dev.knotless.getStep`, one step with its versions and who uses it.
use super::repo::invalid;
use super::*;
use crate::index::{STEP, Shared, Version};
use knotless::records::ReleaseRecord;
use serde_json::Value;
/// Steps a page of the directory shows unless asked for more or fewer.
const PAGE: u32 = 50;
/// The most repositories using a step that its page names.
const USED_BY: u32 = 30;
/// `dev.knotless.defs#stepView`.
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct StepView {
uri: String,
name: String,
repo: RepoViewBasic,
#[serde(skip_serializing_if = "Option::is_none")]
description: Option<String>,
version: String,
secret_count: usize,
use_count: u64,
created_at: String,
}
/// `dev.knotless.defs#stepViewDetailed`.
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct StepViewDetailed {
uri: String,
name: String,
repo: RepoViewBasic,
#[serde(skip_serializing_if = "Option::is_none")]
description: Option<String>,
version: String,
commit: String,
path: String,
uses: String,
inputs: Vec<Value>,
needs: Vec<Named>,
secrets: Vec<Named>,
versions: Vec<StepVersionView>,
use_count: u64,
used_by: Vec<RepoViewBasic>,
created_at: String,
}
/// `dev.knotless.defs#stepCommandView` and `#stepSecretView`.
#[derive(Serialize)]
pub struct Named {
name: String,
}
/// `dev.knotless.defs#stepVersionView`.
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct StepVersionView {
uri: String,
version: String,
commit: String,
created_at: String,
}
/// A step's versions, the highest first; a version that is not one
/// last.
fn highest_first(versions: &mut [Version]) {
let order = |r: &ReleaseRecord| std::cmp::Reverse(knotless::release::version(&r.version).ok());
versions.sort_by_key(|a| order(&a.1));
}
/// The version shown as a step's: its highest with no pre-release part,
/// or its highest when each has one. `versions` are the highest first.
fn latest(versions: &[Version]) -> Option<&Version> {
let release =
|v: &&Version| knotless::release::version(&v.1.version).is_ok_and(|s| s.pre.is_empty());
versions.iter().find(release).or(versions.first())
}
/// What a workflow's `uses` names `name` of `owner`'s repository `repo`
/// by to take `version`'s major version: before 1.0.0, its minor one.
fn uses(owner: &str, repo: &str, name: &str, version: &str) -> String {
let spec = match knotless::release::version(version) {
Ok(v) if v.major == 0 => format!("0.{}", v.minor),
Ok(v) => v.major.to_string(),
Err(_) => version.to_string(),
};
format!("{owner}/{repo}/{name}@{spec}")
}
/// What a step file says, as [`crate::shared`] kept it, by key.
fn said(value: &Value, key: &str) -> Vec<String> {
let names = value[key].as_array().into_iter().flatten();
names
.filter_map(|n| n.as_str().map(str::to_string))
.collect()
}
/// The index, or why there are no steps without one.
fn indexed(app: &App) -> Result<&Index> {
let why = || Shown(StatusCode::NOT_FOUND, "this server keeps no index".into());
app.index.as_deref().ok_or_else(why)
}
/// The owner and key of the repository `uri` names, its
/// `dev.knotless.repo` record by handle or DID.
async fn repo_of(app: &App, index: &Index, uri: &str) -> Result<(String, String)> {
let rest = uri.strip_prefix("at://").unwrap_or_default();
let parts: Vec<&str> = rest.split('/').collect();
let [authority, nsid::REPO, name] = parts[..] else {
let why = format!("{uri} is not at://<authority>/{}/<name>", nsid::REPO);
return Err(invalid(why));
};
let id = knotless::identity::resolve(&app.ctx, authority).await;
let did = id.map_err(not_found)?.did;
let key = index.key_named(&did, name).ok().flatten();
let key = key.ok_or_else(|| not_found(anyhow::anyhow!("no repository {uri}")))?;
Ok((did, key))
}
/// `thing`'s repository, as it is shown: none when the index lacks it.
async fn repo_view(app: &App, index: &Index, thing: &Shared) -> Option<RepoViewBasic> {
let name = index.name(&thing.did, &thing.key).ok().flatten()?;
Some(RepoViewBasic {
did: thing.did.clone(),
handle: handle(app, &thing.did).await,
name,
})
}
/// `thing`, at its latest version, as the directory shows it: none when
/// it has no version that is one.
async fn view(app: &App, index: &Index, thing: Shared) -> Option<StepView> {
let mut versions = index.shared_versions(&thing).ok()?;
highest_first(&mut versions);
let (uri, record, value) = latest(&versions)?;
let (use_count, _) = index.users(&thing, 0).ok()?;
Some(StepView {
uri: uri.clone(),
description: value["description"].as_str().map(str::to_string),
version: record.version.clone(),
secret_count: said(value, "secrets").len(),
use_count,
created_at: record.created_at.clone(),
repo: repo_view(app, index, &thing).await?,
name: thing.name,
})
}
#[derive(Deserialize)]
pub struct ListParams {
repo: Option<String>,
q: Option<String>,
limit: Option<u32>,
cursor: Option<String>,
}
#[derive(Serialize)]
pub struct Steps {
steps: Vec<StepView>,
#[serde(skip_serializing_if = "Option::is_none")]
cursor: Option<String>,
}
/// `dev.knotless.listSteps`.
pub(super) async fn list_steps(
State(app): State<Arc<App>>,
Query(p): Query<ListParams>,
) -> Result<Response> {
let index = indexed(&app)?;
let repo = match &p.repo {
Some(uri) => Some(repo_of(&app, index, uri).await?),
None => None,
};
let limit = p.limit.unwrap_or(PAGE).clamp(1, 100);
let prefix = p.q.as_deref().unwrap_or_default().trim();
let of = repo.as_ref().map(|(did, key)| (did.as_str(), key.as_str()));
let page = index.shared(STEP, of, prefix, limit, p.cursor.as_deref());
let (found, cursor) = page
.map_err(anyhow::Error::from)
.map_err(AppError::internal)?;
let each = found.into_iter().map(|thing| view(&app, index, thing));
let steps = futures::future::join_all(each).await;
Ok(answer(Steps {
steps: steps.into_iter().flatten().collect(),
cursor,
}))
}
#[derive(Deserialize)]
pub struct GetParams {
repo: String,
name: String,
}
/// `dev.knotless.getStep`.
pub(super) async fn get_step(
State(app): State<Arc<App>>,
Query(p): Query<GetParams>,
) -> Result<Response> {
let index = indexed(&app)?;
let (did, key) = repo_of(&app, index, &p.repo).await?;
let thing = Shared {
kind: STEP.into(),
did,
key,
name: p.name,
};
let none = || not_found(anyhow::anyhow!("{} has no step {}", p.repo, thing.name));
let mut versions = index.shared_versions(&thing).unwrap_or_default();
highest_first(&mut versions);
let (uri, record, value) = latest(&versions).ok_or_else(none)?.clone();
let repo = repo_view(&app, index, &thing).await.ok_or_else(none)?;
let (use_count, users) = index.users(&thing, USED_BY).unwrap_or_default();
let app = &app;
let each = users.into_iter().map(|(did, name)| async move {
RepoViewBasic {
handle: handle(app, &did).await,
did,
name,
}
});
let used_by = futures::future::join_all(each).await;
let named = |key: &str| said(&value, key).into_iter().map(|name| Named { name });
let versions = versions.into_iter().map(|(uri, r, _)| StepVersionView {
uri,
version: r.version,
commit: r.commit.unwrap_or_default(),
created_at: r.created_at,
});
Ok(answer(StepViewDetailed {
uri,
uses: uses(&repo.handle, &repo.name, &thing.name, &record.version),
path: format!("{}/{}.yml", knotless::uses::STEPS, thing.name),
description: value["description"].as_str().map(str::to_string),
version: record.version,
commit: record.commit.unwrap_or_default(),
inputs: value["inputs"].as_array().cloned().unwrap_or_default(),
needs: named("needs").collect(),
secrets: named("secrets").collect(),
versions: versions.collect(),
use_count,
used_by,
created_at: record.created_at,
repo,
name: thing.name,
}))
}
#[cfg(test)]
mod tests {
use super::*;
fn version(v: &str) -> Version {
let record = ReleaseRecord {
repo: "k".into(),
package: "@step/s".into(),
version: v.into(),
commit: None,
check: None,
notes: None,
files: Vec::new(),
created_at: String::new(),
};
(v.into(), record, Value::Null)
}
#[test]
fn the_latest_is_the_highest_released_or_else_the_highest() {
let mut versions: Vec<Version> = ["1.0.0", "1.10.0", "2.0.0-rc.1", "1.2.0", "x"]
.into_iter()
.map(version)
.collect();
highest_first(&mut versions);
let order: Vec<&str> = versions.iter().map(|v| v.0.as_str()).collect();
assert_eq!(order, ["2.0.0-rc.1", "1.10.0", "1.2.0", "1.0.0", "x"]);
assert_eq!(latest(&versions).unwrap().0, "1.10.0");
let pre = [version("0.1.0-rc.1")];
assert_eq!(latest(&pre).unwrap().0, "0.1.0-rc.1");
assert!(latest(&[]).is_none());
}
#[test]
fn uses_names_the_major_version() {
assert_eq!(uses("a.test", "ci", "wait", "1.2.0"), "a.test/ci/wait@1");
assert_eq!(uses("a.test", "ci", "wait", "0.3.1"), "a.test/ci/wait@0.3");
assert_eq!(uses("a.test", "ci", "wait", "x"), "a.test/ci/wait@x");
}
}
crates/knotless-serve/src/index.rs+41−5
View file
@@ -27,6 +27,9 @@ use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::sync::{broadcast, mpsc};
mod shared;
pub use shared::{STEP, Shared, Version, kind_of};
/// The kinds of preference kept, in `pref`: how the pages are drawn, and
/// which notifications an account hears of, which the queries for them
/// read by name.
@@ -55,6 +58,10 @@ pub struct Index {
recount: mpsc::UnboundedSender<(String, String)>,
/// What counts them takes this, once ([`crate::languages::follow`]).
counting: Mutex<Option<mpsc::UnboundedReceiver<(String, String)>>>,
/// The releases of things shared whose files are to be read, by URI;
/// what reads them takes the other end, once ([`crate::shared::follow`]).
read: mpsc::UnboundedSender<String>,
reading: Mutex<Option<mpsc::UnboundedReceiver<String>>>,
/// When what each repository's pulls and issues are made of last
/// changed, for [`Index::assembled`] to know what it kept still holds.
stamps: Stamps,
@@ -366,6 +373,12 @@ pub(crate) fn connect(path: &Path) -> rusqlite::Result<(Connection, bool)> {
let collaborated = db
.prepare("SELECT 1 FROM sqlite_master WHERE name = 'collaborator'")?
.exists([])?;
let shared = db
.prepare("SELECT 1 FROM sqlite_master WHERE name = 'shared'")?
.exists([])?;
let used = db
.prepare("SELECT 1 FROM sqlite_master WHERE name = 'shared_use'")?
.exists([])?;
// A repository is known by its owner and key; its name, and the
// URIs others wrote naming it, can change.
db.execute_batch(&format!(
@@ -509,7 +522,9 @@ pub(crate) fn connect(path: &Path) -> rusqlite::Result<(Connection, bool)> {
CREATE INDEX IF NOT EXISTS label_uri ON label (uri);
DROP VIEW IF EXISTS temp.down;
CREATE TABLE IF NOT EXISTS down (subject TEXT PRIMARY KEY);
{shared}
{triggers}",
shared = shared::TABLES,
triggers = down_triggers(),
))?;
// What is down, from what makes it so, as it stands now: the
@@ -556,6 +571,17 @@ pub(crate) fn connect(path: &Path) -> rusqlite::Result<(Connection, bool)> {
if !collaborated {
let _ = db.execute(COLLABORATORS, [None::<&str>]);
}
// The releases of what is shared, from the releases, once: an index
// from before had them only there.
if !shared {
let _ = db.execute(shared::UNREAD, []);
}
// What each repository uses is read as what it is written in is
// counted: an index from before counted them without, so each is
// counted again as a list next shows it.
if !used {
let _ = db.execute("DELETE FROM language", []);
}
Ok((db, rebuilt))
}
@@ -594,7 +620,7 @@ const CACHE: &str = "DROP TABLE IF EXISTS repo; DROP TABLE IF EXISTS forge;
DROP TABLE IF EXISTS reaction; DROP TABLE IF EXISTS release; DROP TABLE IF EXISTS mention;
DROP TABLE IF EXISTS reviewer;
DROP TABLE IF EXISTS org; DROP TABLE IF EXISTS org_member; DROP TABLE IF EXISTS membership;
DROP TABLE IF EXISTS collaborator;";
DROP TABLE IF EXISTS collaborator; DROP TABLE IF EXISTS shared;";
/// A label that has not lapsed.
const LIVE: &str = "(exp IS NULL OR exp > strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))";
@@ -663,6 +689,8 @@ const FORGET_REPO: &[&str] = &[
"DELETE FROM forge WHERE repo_did = ?1 AND repo_key = ?2",
"DELETE FROM check_run WHERE repo_did = ?1 AND repo_key = ?2",
"DELETE FROM release WHERE did = ?1 AND repo_key = ?2",
"DELETE FROM shared WHERE did = ?1 AND repo_key = ?2",
"DELETE FROM shared_use WHERE did = ?1 AND key = ?2",
"DELETE FROM star WHERE repo_did = ?1 AND repo_key = ?2",
"DELETE FROM push WHERE did = ?1 AND key = ?2",
"DELETE FROM language WHERE did = ?1 AND key = ?2",
@@ -694,6 +722,8 @@ const PURGE: &[&str] = &[
"DELETE FROM follow WHERE did = ?1",
"DELETE FROM reaction WHERE did = ?1 -- scans: reaction, once, as an account is forgotten",
"DELETE FROM release WHERE did = ?1",
"DELETE FROM shared WHERE did = ?1",
"DELETE FROM shared_use WHERE did = ?1",
"DELETE FROM theme WHERE did = ?1",
"DELETE FROM org WHERE did = ?1 OR substr(space, 1, length(?2)) = ?2",
"DELETE FROM org_member WHERE org = ?1 OR substr(space, 1, length(?2)) = ?2",
@@ -731,6 +761,7 @@ impl Index {
let (db, rebuilt) = connect(path)?;
crate::telemetry::watch(&db);
let (recount, counting) = mpsc::unbounded_channel();
let (read, reading) = mpsc::unbounded_channel();
Ok(Index {
db: Mutex::new(db),
rebuilt,
@@ -738,6 +769,8 @@ impl Index {
asked: Mutex::default(),
recount,
counting: Mutex::new(Some(counting)),
read,
reading: Mutex::new(Some(reading)),
stamps: Stamps::new(STAMPS_MAX),
kept: Mutex::new(Kept::new(KEPT_ENTRIES)),
})
@@ -928,10 +961,12 @@ impl Index {
&& knotless::release::version(&r.version).is_ok()
});
match release {
Some(release) => db.execute(
"INSERT OR REPLACE INTO release VALUES (?1, ?2, ?3, ?4, ?5)",
params![uri, did, release.repo, release.created_at, text],
),
Some(release) => db
.execute(
"INSERT OR REPLACE INTO release VALUES (?1, ?2, ?3, ?4, ?5)",
params![uri, did, release.repo, release.created_at, text],
)
.and_then(|n| self.share(&db, &uri, did, &release).map(|_| n)),
None => Ok(0),
}
} else if let Some(repo) = named("repo") {
@@ -1142,6 +1177,7 @@ impl Index {
db.execute("DELETE FROM minutes WHERE uri = ?1", [&uri])?;
db.execute("DELETE FROM runner_key WHERE uri = ?1", [&uri])?;
db.execute("DELETE FROM release WHERE uri = ?1", [&uri])?;
db.execute("DELETE FROM shared WHERE uri = ?1", [&uri])?;
db.execute(
"DELETE FROM profile_record WHERE did = ?1 AND collection = ?2 AND rkey = ?3",
[did, collection, rkey],
crates/knotless-serve/src/languages.rs+15−4
View file
@@ -10,6 +10,8 @@
//!
//! What a public repository is mostly written in, which lists of them
//! show, is kept in the index ([`follow`]), so a list reads no repository.
//! The shared steps its lock pins are read from the same tree as it is
//! counted ([`crate::shared`]).
use crate::App;
use crate::objects::{self, Objects};
@@ -260,15 +262,24 @@ pub async fn follow(app: Arc<App>) {
}
}
/// Counts `did`'s repository with `key` again, and keeps the answer.
/// Counts `did`'s repository with `key` again, and keeps the answer; and
/// what shared steps its lock pins, read from the same tree
/// ([`crate::shared::read_uses`]).
async fn recount(app: &App, did: &str, key: &str) -> Option<()> {
let index = app.index.as_ref()?;
index.answered(did, key);
let name = index.name(did, key).ok().flatten()?;
let git = app.mirror(did, &name).await.ok()?;
let objects = objects::open(&git.dir).ok()?;
let value = serde_json::to_string(&main_of(&objects)).ok()?;
index.set_language(did, key, &value).ok()
// Read, and let go of, before what is awaited: a reader belongs to
// one thread.
let (value, lock) = {
let objects = objects::open(&git.dir).ok()?;
let value = serde_json::to_string(&main_of(&objects)).ok()?;
(value, objects.blob("HEAD", knotless::uses::LOCK).ok())
};
index.set_language(did, key, &value).ok()?;
crate::shared::read_uses(app, index, did, key, lock.as_deref()).await;
Some(())
}
#[cfg(test)]
crates/knotless-serve/src/lib.rs+2−0
View file
@@ -36,6 +36,7 @@ pub mod plans;
pub mod private;
pub mod releases;
mod secrets;
pub mod shared;
mod smart;
pub mod spaces;
pub mod status;
@@ -1172,6 +1173,7 @@ pub async fn run(opts: Options, ctx: Ctx) -> Result<()> {
}
tokio::spawn(spaces::follow(app.clone(), SPACES));
tokio::spawn(languages::follow(app.clone()));
tokio::spawn(shared::follow(app.clone()));
tokio::spawn(telemetry::summaries(app.clone(), SUMMARY));
let listener = tokio::net::TcpListener::bind(&opts.listen).await?;
let addr = listener.local_addr()?;
crates/knotless-serve/src/plans.rs+3−1
View file
@@ -32,7 +32,7 @@ pub const ENV: &str = "KNOTLESS_PLANS";
/// The tables that grow with what people do, as many rows as prod might
/// hold: a scan of one is a scan of the network. The rest are said to be
/// small, as a scan of them is.
pub const GROWING: [&str; 14] = [
pub const GROWING: [&str; 16] = [
"forge",
"check_run",
"comment",
@@ -47,6 +47,8 @@ pub const GROWING: [&str; 14] = [
"release",
"reviewer",
"theme",
"shared",
"shared_use",
];
/// The rows a growing table is said to hold, and a small one.
crates/knotless-serve/src/shared.rsnew+138−0
View file
@@ -0,0 +1,138 @@
//! What repositories share for others to use, as the index keeps it
//! (`index::shared`): the files of their releases, read and kept as what
//! they are, and which repositories use each.
//!
//! A shared step is a release of the package `@step/<name>` whose one
//! file is a step file, published with the commit it was made from, as
//! `knotless uses update` takes one. Its file is read once, from the
//! release's blobs on its owner's PDS, checked against its record and kept
//! ([`Records`]), as a release's files never change. One that is not a
//! step file, or names no commit, is kept as not one, and never shown.
//!
//! Who uses a step is read from `.knotless/uses.lock` on each public
//! repository's default branch, as what it is written in is counted
//! ([`crate::languages::follow`]): once a list shows it, and again as its
//! refs move. A repository is counted once for each step it pins, at
//! whatever versions.
use crate::App;
use crate::index::{Index, STEP, Shared};
use crate::releases::Records;
use knotless::url::RepoUrl;
use knotless::uses::{self, Use};
use serde_json::{Value, json};
use std::sync::Arc;
/// Reads the files of the releases of things shared as the index asks
/// ([`Index::reading`]): first those it kept before and has yet to read,
/// then each as it is indexed, one at a time.
pub async fn follow(app: Arc<App>) {
let taken = app.index.clone().and_then(|i| Some((i.reading()?, i)));
let Some((mut asked, index)) = taken else {
return;
};
for uri in index.unread().unwrap_or_default() {
read(&app, &index, &uri).await;
}
while let Some(uri) = asked.recv().await {
read(&app, &index, &uri).await;
}
}
/// Reads the file of the release at `uri` and keeps what it is. One that
/// cannot be read now, its owner's PDS away, is left to be read when
/// next the server starts.
async fn read(app: &App, index: &Index, uri: &str) -> Option<()> {
let (thing, release) = index.shared_release(uri).ok().flatten()?;
let name = index.name(&thing.did, &thing.key).ok().flatten()?;
let found = match (&release.files[..], &release.commit) {
([file], Some(_)) => {
let url = RepoUrl::parse(&format!("knotless://{}/{name}", thing.did)).ok()?;
let records = Records::new(app.ctx.clone(), url, release.package.clone());
let bytes = records.read(file).await.ok()?;
step(&file.name, &bytes)
}
_ => None,
};
index.set_shared(uri, found.as_ref()).ok()
}
/// The step file `bytes`, named `file`, as the index keeps it: what it
/// does, takes and asks for. None for one that is not a step file.
pub fn step(file: &str, bytes: &[u8]) -> Option<Value> {
let parsed = uses::parse_step(file, bytes).ok()?;
let inputs = parsed.inputs.iter().map(|(name, input)| {
let mut shown = json!({"name": name, "required": input.required});
if !input.description.is_empty() {
shown["description"] = input.description.clone().into();
}
if let Some(default) = &input.default {
shown["default"] = default.clone().into();
}
shown
});
let description = Some(parsed.description).filter(|d| !d.trim().is_empty());
Some(json!({
"description": description,
"inputs": inputs.collect::<Vec<_>>(),
"needs": parsed.needs,
"secrets": parsed.secrets,
}))
}
/// Keeps what `did`'s repository with `key` uses, from `lock`, the lock
/// file on its default branch: nothing, with none, or one that does not
/// parse. A step whose repository the index does not have is passed by,
/// for a count after it does.
pub async fn read_uses(app: &App, index: &Index, did: &str, key: &str, lock: Option<&[u8]>) {
let lock = lock.and_then(|text| uses::parse_lock(text).ok());
let mut used = Vec::new();
for named in lock.unwrap_or_default().keys() {
if let Some(thing) = step_of(app, index, named).await {
used.push(thing);
}
}
let _ = index.set_uses(did, key, &used);
}
/// The step a lock's `uses` names, if the index has its repository.
async fn step_of(app: &App, index: &Index, named: &str) -> Option<Shared> {
let used = Use::parse(named).ok()?;
let owner = knotless::identity::resolve(&app.ctx, &used.owner).await;
let did = owner.ok()?.did;
let key = index.key_named(&did, &used.repo).ok().flatten()?;
Some(Shared {
kind: STEP.into(),
did,
key,
name: used.step,
})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_step_file_is_kept_as_what_it_takes_and_asks_for() {
let file = b"description: Waits.\ninputs:\n workflows: {description: Which, required: true}\n minutes: {default: \"25\"}\nneeds: [curl]\nsecrets: [TOKEN]\nsteps:\n - run: echo\n";
let kept = step("wait.yml", file).unwrap();
assert_eq!(
kept,
json!({
"description": "Waits.",
"inputs": [
{"name": "minutes", "required": false, "default": "25"},
{"name": "workflows", "required": true, "description": "Which"},
],
"needs": ["curl"],
"secrets": ["TOKEN"],
})
);
assert_eq!(
step("x.yml", b"steps: []\n").unwrap()["description"],
Value::Null
);
assert!(step("x.yml", b"not: a step\n").is_none());
}
}
crates/knotless-serve/src/web.rs+3−0
View file
@@ -66,6 +66,7 @@ mod repos;
mod runs;
mod settings;
mod stars;
mod steps;
mod themes;
mod threads;
mod views;
@@ -167,6 +168,8 @@ pub fn queries() -> Router<Arc<App>> {
post(controls::remove_triager),
)
.route("/xrpc/dev.knotless.listThemes", get(themes::list_themes))
.route("/xrpc/dev.knotless.listSteps", get(steps::list_steps))
.route("/xrpc/dev.knotless.getStep", get(steps::get_step))
.route("/xrpc/dev.knotless.createTheme", post(themes::create_theme))
.route("/xrpc/dev.knotless.deleteTheme", post(themes::delete_theme))
.route("/xrpc/dev.knotless.followActor", post(follow_actor))
crates/knotless-serve/tests/web.rs+157−0
View file
@@ -3105,3 +3105,160 @@ fn badges_count_where_an_account_lives() {
assert_eq!(kinds, ["carry-your-own", "moved-house"]);
wait_until(|| index.home(did).unwrap().unwrap().0 == Home::default());
}
#[test]
fn the_steps_shared() {
let site = serve(true);
let owner = person(account(), &site);
let handle = owner.account.handle.clone();
let url = owner.url("steps");
ok(&owner, &["create", &url]);
let work = owner.work("w", 1);
owner.git(&work, &["push", "-q", &url, "main"]);
let commit = owner.git(&work, &["rev-parse", "HEAD"]);
// Steps, each released as its package with the commit it is at; a
// file that is no step, and a release naming no commit, are passed by.
let file = |name: &str, text: &str| {
let path = owner.path(name);
std::fs::write(&path, text).unwrap();
path.to_string_lossy().into_owned()
};
let wait = file(
"wait.yml",
"description: Waits for this commit's checks.\ninputs:\n workflows: {description: Which must pass, required: true}\n minutes: {default: \"25\"}\nneeds: [curl]\nsecrets: [TOKEN]\nsteps:\n - run: echo\n",
);
let lint = file("lint.yml", "steps:\n - run: cargo clippy\n");
let broken = file("broken.yml", "not: a step\n");
let publish = |package: &str, version: &str, path: &str, commit: Option<&str>| {
let mut args = vec!["publish", &url, package, version, path];
args.extend(commit.map(|c| ["--commit", c]).into_iter().flatten());
ok(&owner, &args);
};
publish("@step/wait", "1.0.0", &wait, Some(&commit));
let repo = format!("at://{handle}/dev.knotless.repo/steps");
let releases = || site.json(&xrpc("listReleases", &[("repo", &repo)]))["releases"].clone();
wait_until(|| releases().as_array().unwrap().len() == 1);
// What was indexed before they start is read first.
let rt = &site._rt;
rt.spawn(knotless_serve::shared::follow(site.app.clone()));
rt.spawn(knotless_serve::languages::follow(site.app.clone()));
publish("@step/wait", "1.1.0", &wait, Some(&commit));
publish("@step/wait", "2.0.0-rc.1", &wait, Some(&commit));
publish("@step/lint", "0.3.0", &lint, Some(&commit));
publish("@step/broken", "1.0.0", &broken, Some(&commit));
publish("@step/loose", "1.0.0", &lint, None);
wait_until(|| releases().as_array().unwrap().len() == 6);
let list = |params: &[(&str, &str)]| site.json(&xrpc("listSteps", params));
let names = |listed: &Value| -> Vec<String> {
let steps = listed["steps"].as_array().unwrap().iter();
steps.map(|s| s["name"].as_str().unwrap().into()).collect()
};
wait_until(|| names(&list(&[])).len() == 2 && list(&[])["steps"][1]["version"] == "1.1.0");
let all = list(&[]);
assert_eq!(names(&all), ["lint", "wait"]);
let shown = &all["steps"][1];
assert_eq!(shown["repo"]["handle"], handle.as_str());
assert_eq!(shown["repo"]["name"], "steps");
assert_eq!(shown["description"], "Waits for this commit's checks.");
assert_eq!(shown["secretCount"], 1);
assert_eq!(shown["useCount"], 0);
assert!(all["steps"][0]["description"].is_null());
assert!(all["cursor"].is_null());
// A page at a time, by name; by how a name starts; one repository's.
let first = list(&[("limit", "1")]);
assert_eq!(names(&first), ["lint"]);
let cursor = first["cursor"].as_str().unwrap().to_string();
let rest = list(&[("limit", "1"), ("cursor", &cursor)]);
assert_eq!(names(&rest), ["wait"]);
assert!(rest["cursor"].is_null());
assert_eq!(names(&list(&[("q", "wa")])), ["wait"]);
assert!(names(&list(&[("q", "x")])).is_empty());
assert_eq!(names(&list(&[("repo", &repo)])).len(), 2);
let other = owner.url("user");
ok(&owner, &["create", &other]);
let used_in = format!("at://{handle}/dev.knotless.repo/user");
let status =
|nsid: &str, params: &[(&str, &str)]| site.get(&xrpc(nsid, params)).status().as_u16();
wait_until(|| status("listSteps", &[("repo", &used_in)]) == 200);
assert!(names(&list(&[("repo", &used_in)])).is_empty());
assert_eq!(status("listSteps", &[("repo", "at://nope")]), 400);
let nowhere = format!("at://{handle}/dev.knotless.repo/nowhere");
assert_eq!(status("listSteps", &[("repo", &nowhere)]), 404);
let unknown = "at://nobody.invalid/dev.knotless.repo/x";
assert_eq!(status("listSteps", &[("repo", unknown)]), 404);
// One step: what it takes and asks for, and its versions, the highest
// first.
let get = |name: &str| site.json(&xrpc("getStep", &[("repo", &repo), ("name", name)]));
let step = get("wait");
assert_eq!(step["version"], "1.1.0");
assert_eq!(step["commit"], commit.as_str());
assert_eq!(step["path"], ".knotless/steps/wait.yml");
assert_eq!(step["uses"], format!("{handle}/steps/wait@1"));
assert_eq!(
step["inputs"],
serde_json::json!([
{"name": "minutes", "required": false, "default": "25"},
{"name": "workflows", "required": true, "description": "Which must pass"},
])
);
assert_eq!(step["needs"], serde_json::json!([{"name": "curl"}]));
assert_eq!(step["secrets"], serde_json::json!([{"name": "TOKEN"}]));
let versions: Vec<&str> = step["versions"]
.as_array()
.unwrap()
.iter()
.map(|v| v["version"].as_str().unwrap())
.collect();
assert_eq!(versions, ["2.0.0-rc.1", "1.1.0", "1.0.0"]);
assert_eq!(get("lint")["uses"], format!("{handle}/steps/lint@0.3"));
for missing in ["broken", "loose", "nothing"] {
assert_eq!(
status("getStep", &[("repo", &repo), ("name", missing)]),
404
);
}
// Used, as a repository's lock on its default branch pins it: read as
// what the repository is written in is counted, once a list shows it.
// Pinned twice, it is one repository using it.
let user = owner.work("u", 1);
std::fs::create_dir_all(user.join(".knotless")).unwrap();
let pin = format!("{{version: 1.1.0, commit: {commit}}}");
let lock = format!(
"{handle}/steps/wait@1: {pin}\n{handle}/steps/wait@1.1: {pin}\n{handle}/nowhere/wait@1: {pin}\nnobody.invalid/x/y@1: {pin}\nnot-a-use: {pin}\n"
);
owner.commit_file(&user, ".knotless/uses.lock", &lock);
owner.git(&user, &["push", "-q", &other, "main"]);
let users = || {
site.profile(&handle);
get("wait")["useCount"].clone()
};
wait_until(|| users() == 1);
let step = get("wait");
assert_eq!(
step["usedBy"],
serde_json::json!([{"did": owner.account.did, "handle": handle, "name": "user"}])
);
assert_eq!(list(&[("q", "wait")])["steps"][0]["useCount"], 1);
// Taken down, a step's releases leave it.
let index = site.app.index.clone().unwrap();
for v in step["versions"].as_array().unwrap() {
let uri = v["uri"].as_str().unwrap();
let subject = serde_json::json!({"$type": "com.atproto.repo.strongRef", "uri": uri});
index
.take_down("did:web:here", uri, &subject, true)
.unwrap();
}
assert_eq!(names(&list(&[])), ["lint"]);
// With no index, there are none.
let plain = serve(false);
assert_eq!(plain.get(&xrpc("listSteps", &[])).status().as_u16(), 404);
plain
._rt
.block_on(knotless_serve::shared::follow(plain.app.clone()));
}
docs/design/reusable-workflows.md+14−4
View file
@@ -234,14 +234,24 @@ Once it lands, the default branch's runs use the steps with its
secrets, and the reviewed diff of the lock is the approval. For the rest
of who runs whose code, see [security](security.md).
## Finding a step
The pages list every shared step the appview has indexed at `/steps`,
by name, each with its repository, what it does, its latest version, how
many public repositories pin it and how many secrets it asks for. A
step's own page, `/steps/<owner>/<repo>/<step>`, gives the lines to paste
into a workflow, its inputs, the secrets and commands it asks for, each
version with the commit it was published at, and the repositories that
use it. A repository that publishes steps links to them from its release
line, and each step's release links to its page. How the appview finds
them, and counts their uses from each repository's lock, is in
[serve](serve.md#shared-steps).
## Not built yet
- Outputs. Steps pass state to later steps through files in the shared
container. This was left out of the first version on purpose.
- A directory of steps on the pages, indexed by knotless-serve from step
files at release commits, with use counts from the locks on default
branches. It is meant to hold other kinds of shared thing, such as
themes, later.
- Themes and other kinds of shared thing in the directory of steps.
- Private step repositories. A step's repository is read through the
runner's mirror as a public one is, and shared steps have been tested
only with public ones.
docs/design/serve.md+33−0
View file
docs/reference/api.md+2−0
View file
lexicons/dev/knotless/defs.json+75−0
View file
lexicons/dev/knotless/getStep.jsonnew+26−0
View file
lexicons/dev/knotless/listSteps.jsonnew+34−0
View file
web/src/lib/organisms/ReleaseCard.svelte+6−1
View file
web/src/lib/organisms/RepoFacts.svelte+4−1
View file
web/src/lib/organisms/StepList.sveltenew+31−0
View file
web/src/lib/organisms/StepSheet.sveltenew+81−0
View file
web/src/lib/icons.ts+3−2
View file
web/src/lib/steps.tsnew+34−0
View file
web/src/routes/steps/[authority]/[repo]/[step]/+page.sveltenew+24−0
View file
web/src/routes/steps/[authority]/[repo]/[step]/+page.tsnew+11−0
View file
web/src/routes/steps/+page.sveltenew+50−0
View file
web/src/routes/steps/+page.tsnew+24−0
View file
web/src/app.css+28−0
View file