~ 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

A login and a gateway's grant name knotless's own space types

Merged chadtmiller.com asked to merge into main from chadtmiller.com/knotless:space/scopes
.changeset/space-scopes.mdnew+7−0
View file
@@ -0,0 +1,7 @@
---
default: minor
---
# A login and a gateway's grant name knotless's own space types
`knotless login` asks for `space:dev.knotless.space` and `space:dev.knotless.orgSpace`, each with the collections knotless writes there, in place of every space of every app (`space:*`); `knotless delegate` asks the same of a gateway's grant. A login or grant from before still works, and is asked for again as any older one is.
crates/knotless/src/gateway.rs+6−3
View file
@@ -36,12 +36,15 @@ pub const AWAIT_DELEGATION: &str = "dev.knotless.getDelegation";
/// for an organization, its record of being one and its records of whom it
/// invited, its space, where it keeps those of invitations and hidden
/// members and its members write theirs, and who its private repositories'
/// spaces hold, which members and admins change through it.
/// spaces hold, which members and admins change through it. Each space
/// type is named, as the login's are ([`crate::oauth::SCOPE`]).
pub const SCOPE: &str = "atproto repo:dev.knotless.repo repo:dev.knotless.ref \
repo:dev.knotless.claim repo:dev.knotless.pack repo:dev.knotless.org \
repo:dev.knotless.orgMember blob:*/* \
space:*?authority=self&collection=dev.knotless.repo&collection=dev.knotless.ref\
&collection=dev.knotless.claim&collection=dev.knotless.pack&collection=dev.knotless.orgMember\
space:dev.knotless.space?authority=self&collection=dev.knotless.repo\
&collection=dev.knotless.ref&collection=dev.knotless.claim&collection=dev.knotless.pack\
&collection=dev.knotless.orgMember&manage=create&manage=update \
space:dev.knotless.orgSpace?authority=self&collection=dev.knotless.orgMember\
&collection=dev.knotless.orgAcceptance&manage=create&manage=update";
/// A member joining an organization, which its gateway takes them into.
pub const JOIN_ORG: &str = "dev.knotless.joinOrg";
crates/knotless/src/oauth.rs+52−6
View file
@@ -32,10 +32,11 @@ const CALLBACK: &str = "/callback";
/// What knotless asks for, and nothing more: to write the records it keeps
/// in an account, to upload the blobs that hold its packs and pictures, to call gateways
/// and moderation services as the account (its operator's methods too), and the spaces private
/// repositories live in. A PDS without spaces drops the scope it does not
/// know. The space scope names every space type: naming
/// [`crate::space::SPACE_TYPE`] alone takes its lexicon, which is not
/// published yet.
/// repositories and organizations live in. A PDS without spaces drops the
/// scope it does not know. Each space scope names its type, with the
/// collections that type's lexicon lists ([`space_scope`]): an
/// authorization server resolves the lexicon of each type a scope names,
/// and refuses the login if it cannot.
pub static SCOPE: LazyLock<String> = LazyLock::new(|| {
let mut scope = vec!["atproto".to_string()];
let written = nsid::ALL.iter().chain(&nsid::FORGE).chain([
@@ -100,10 +101,33 @@ pub static SCOPE: LazyLock<String> = LazyLock::new(|| {
crate::account::FORGET,
];
scope.extend(called.map(|lxm| format!("rpc:{lxm}?aud=*")));
scope.push("space:*?authority=*&collection=*&manage=create&manage=update&manage=delete".into());
let manage = "&manage=create&manage=update&manage=delete";
for lexicon in [SPACE_LEXICON, ORG_SPACE_LEXICON] {
scope.push(space_scope(lexicon, "*", manage));
}
scope.join(" ")
});
/// The lexicon of [`crate::space::SPACE_TYPE`], a private repository's.
pub const SPACE_LEXICON: &str = include_str!("../../../lexicons/dev/knotless/space.json");
/// The lexicon of [`crate::org::SPACE_TYPE`], an organization's.
pub const ORG_SPACE_LEXICON: &str = include_str!("../../../lexicons/dev/knotless/orgSpace.json");
/// The scope for the spaces of the type whose lexicon is `lexicon`, under
/// `authority` (`*`, or `self`), to write each collection it lists, and
/// `manage`, the `&manage=` of it.
pub fn space_scope(lexicon: &str, authority: &str, manage: &str) -> String {
let doc: Value = serde_json::from_str(lexicon).expect("a space lexicon is JSON");
let listed = doc["defs"]["main"]["collections"].as_array().into_iter();
let collections: String = listed
.flatten()
.filter_map(Value::as_str)
.map(|c| format!("&collection={c}"))
.collect();
let id = doc["id"].as_str().unwrap_or_default();
format!("space:{id}?authority={authority}{collections}{manage}")
}
/// What a call refused for want of scope says in place of the PDS's
/// words, when the session refused was the caller's own login.
pub const LOGIN_BEHIND: &str =
@@ -797,6 +821,18 @@ mod tests {
assert!(behind(&session, &SCOPE));
session.oauth.as_mut().unwrap().scope = Some(SCOPE.clone());
assert!(!behind(&session, &SCOPE));
// A login that named every space type is behind one that names
// knotless's alone, and a gateway's grant so too: both ask again.
let broad = SCOPE.split(" space:").next().unwrap().to_string()
+ " space:*?authority=*&collection=*&manage=create&manage=update&manage=delete";
session.oauth.as_mut().unwrap().scope = Some(broad);
assert!(behind(&session, &SCOPE));
let granted = gateway::SCOPE.split(" space:").next().unwrap().to_string()
+ " space:*?authority=self&collection=dev.knotless.repo";
session.oauth.as_mut().unwrap().scope = Some(granted);
assert!(behind(&session, gateway::SCOPE));
session.oauth.as_mut().unwrap().scope = Some(gateway::SCOPE.into());
assert!(!behind(&session, gateway::SCOPE));
}
#[test]
@@ -843,7 +879,17 @@ mod tests {
assert!(id.starts_with(
"http://localhost?redirect_uri=http%3A%2F%2F127.0.0.1%2Fcallback&scope=atproto+"
));
assert!(SCOPE.contains(" space:*?authority=*&"));
// Only knotless's own space types, each by name.
assert!(!SCOPE.contains("space:*"));
let named = |kind: &str| {
SCOPE
.matches(&format!(" space:{kind}?authority=*&"))
.count()
};
assert_eq!(named("dev.knotless.space"), 1);
assert_eq!(named("dev.knotless.orgSpace"), 1);
assert!(SCOPE.contains("&collection=dev.knotless.rerun&collection=dev.knotless.orgMember&manage=create&manage=update&manage=delete"));
assert!(SCOPE.ends_with(" space:dev.knotless.orgSpace?authority=*&collection=dev.knotless.orgMember&collection=dev.knotless.orgAcceptance&manage=create&manage=update&manage=delete"));
assert!(SCOPE.contains(" repo:dev.knotless.pull "));
assert!(SCOPE.contains(" repo:dev.knotless.check "));
assert!(SCOPE.contains(" repo:dev.knotless.minutes "));
crates/knotless/src/org.rs+48−167
View file
@@ -51,11 +51,6 @@ use serde_json::{Value, json};
/// The type of the space an organization keeps its invitations and hidden
/// members in, keyed [`RKEY`].
pub const SPACE_TYPE: &str = "dev.knotless.orgSpace";
/// The type that space was before: the NSID of the organization's public
/// record, whose lexicon is no space, so that no scope can name it. What
/// one of that type holds is moved into the organization's space as its
/// gateway next opens it ([`moved`]), and read from there until then.
pub const OLD_SPACE_TYPE: &str = "dev.knotless.org";
const RKEY: &str = "self";
/// The most records one write deletes, as a PDS takes them.
const BATCH: usize = 100;
@@ -71,17 +66,6 @@ pub fn space_uri(org: &str) -> String {
format!("at://{org}/space/{SPACE_TYPE}/{RKEY}")
}
/// The space `org` kept them in before, of [`OLD_SPACE_TYPE`].
pub fn old_space_uri(org: &str) -> String {
format!("at://{org}/space/{OLD_SPACE_TYPE}/{RKEY}")
}
/// Both: what reads the organization's space reads each, until the old
/// one holds nothing.
pub fn space_uris(org: &str) -> [String; 2] {
[space_uri(org), old_space_uri(org)]
}
/// What a member may do.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
@@ -246,11 +230,11 @@ fn inside(ctx: &Ctx, id: &Identity, space: &Space) -> Remote {
}
}
/// The organization's space `uri`, as `reader` reads it: none when it has
/// none, its PDS has no spaces, or `reader` may not read it.
async fn seen(ctx: &Ctx, org: &Identity, reader: &Session, uri: &str) -> Result<Option<Space>> {
/// The organization's space, as `reader` reads it: none when it has none,
/// its PDS has no spaces, or `reader` may not read it.
async fn seen(ctx: &Ctx, org: &Identity, reader: &Session) -> Result<Option<Space>> {
let host = Xrpc::new(ctx.http.clone(), &org.pds);
let joined = Space::join(ctx, &host, uri, &org.did, reader).await;
let joined = Space::join(ctx, &host, &space_uri(&org.did), &org.did, reader).await;
unless_shut(joined.map(Some))
}
@@ -503,27 +487,20 @@ pub async fn gateway(ctx: &Ctx, id: &Identity) -> Option<String> {
/// public of one it took in is of a member in public; in its space, of a
/// hidden one.
///
/// Answers too the organization's spaces, as `org` reads them: its space,
/// with what the old one held moved into it first ([`moved`]).
async fn recorded(ctx: &Ctx, id: &Identity, org: &Session) -> Result<(Roster, Spaces)> {
/// Answers too the organization's space, as `org` reads it.
async fn recorded(ctx: &Ctx, id: &Identity, org: &Session) -> Result<(Roster, Option<Space>)> {
let open = public(ctx, id);
// Read at once: an account that is no organization keeps none of the
// others, and what reading them said is set aside.
let [uri, old] = space_uris(&id.did);
let (named, shown, space, old) = tokio::join!(
// Read at once: an account that is no organization keeps neither of the
// other two, and what reading them said is set aside.
let (named, shown, space) = tokio::join!(
open.get_record::<OrgRecord>(nsid::ORG, RKEY),
member_records(&open),
seen(ctx, id, org, &uri),
seen(ctx, id, org, &old),
seen(ctx, id, org)
);
let named = named?.map(|r| r.value);
let (shown, space, old) = match &named {
Some(_) => (shown?, space?, old?),
None => (Vec::new(), None, None),
};
let space = match &old {
Some(old) => moved(ctx, id, org, old, space).await?,
None => space,
let (shown, space) = match &named {
Some(_) => (shown?, space?),
None => (Vec::new(), None),
};
let kept = match &space {
Some(space) => member_records(&inside(ctx, id, space)).await?,
@@ -563,7 +540,7 @@ async fn recorded(ctx: &Ctx, id: &Identity, org: &Session) -> Result<(Roster, Sp
gateway: named.and_then(|o| o.gateway),
entries,
};
Ok((roster, Spaces { space, old }))
Ok((roster, space))
}
/// The organization `org` is, read as itself from its own records alone
@@ -574,88 +551,6 @@ pub async fn roster_as(ctx: &Ctx, org: &Session) -> Result<Roster> {
Ok(roster)
}
/// An organization's spaces, as it reads them.
struct Spaces {
/// Where it keeps its invitations and hidden members, if it has it.
space: Option<Space>,
/// The one of [`OLD_SPACE_TYPE`] it kept them in before, if it has it,
/// where hidden members' own records of joining may be still.
old: Option<Space>,
}
/// Moves what the organization `org` is keeps in its old space into its
/// space, as every change through its gateway does first ([`moved`]):
/// done once it passes, and nothing to do after.
pub async fn move_space(ctx: &Ctx, org: &Session) -> Result<()> {
roster_as(ctx, org).await.map(drop)
}
/// Moves what the organization `org` keeps in `old`, its space of
/// [`OLD_SPACE_TYPE`], into `space`, its space now, made if it has none:
/// the accounts the old one holds, then its records of those it invited,
/// each written in the new before it goes from the old, so that a reader
/// of both, as each is meanwhile, never misses one. An old one with no
/// such record left has nothing to move, and is left as it is. Answers
/// the space moved into.
///
/// The hidden members' own records of joining stay where they are, as
/// only they write in their repo there: they count from the old space,
/// which keeps them as members, until each joins again ([`join`]). The old
/// space goes only with the organization ([`delete_as`]).
async fn moved(
ctx: &Ctx,
id: &Identity,
org: &Session,
old: &Space,
space: Option<Space>,
) -> Result<Option<Space>> {
let from = inside(ctx, id, old);
let records = member_records(&from).await?;
if records.is_empty() {
return Ok(space);
}
let xrpc = Xrpc::new(ctx.http.clone(), &id.pds);
let admin = Admin {
xrpc: &xrpc,
session: org,
};
admin.create_of(SPACE_TYPE, RKEY).await?;
let made = seen(ctx, id, org, &space_uri(&id.did)).await?;
let space = made.context("the organization's space was made, and cannot be read")?;
let there = admin.members(&space.uri).await?;
for m in admin.members(&old.uri).await? {
// One the new space holds is as it was put there since.
if !there.iter().any(|t| t.did == m.did) {
admin
.put_member_as(&space.uri, &m.did, m.read, m.write)
.await?;
}
}
let to = inside(ctx, id, &space);
let kept = member_records(&to).await?;
// One in both was written in the new since: that one counts.
let unkept = records
.iter()
.filter(|r| !kept.iter().any(|k| k.member == r.member));
let write = |r: &OrgMemberRecord| {
let value = records::typed(nsid::ORG_MEMBER, r);
remote::create(nsid::ORG_MEMBER, &r.member, value)
};
let writes: Vec<Value> = unkept.map(write).collect();
let gone = records
.iter()
.map(|r| remote::delete(nsid::ORG_MEMBER, &r.member));
let gone: Vec<Value> = gone.collect();
let auth = org.auth();
for batch in writes.chunks(BATCH) {
to.apply(batch.to_vec(), &auth).await?;
}
for batch in gone.chunks(BATCH) {
from.apply(batch.to_vec(), &auth).await?;
}
Ok(Some(space))
}
/// What inviting someone, or changing their role, did.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
@@ -701,7 +596,7 @@ pub struct Joined {
}
/// The organization `session` is, read as itself, with the places it keeps
/// its records: its space as it opened, if it had one, and its old one.
/// its records: its space as it opened, if it had one.
struct Own<'a> {
ctx: &'a Ctx,
session: &'a Session,
@@ -709,13 +604,12 @@ struct Own<'a> {
open: Remote,
roster: Roster,
space: Option<Space>,
old: Option<Space>,
}
impl<'a> Own<'a> {
async fn open(ctx: &'a Ctx, session: &'a Session) -> Result<Own<'a>> {
let id = identity::resolve(ctx, &session.did).await?;
let (roster, Spaces { space, old }) = recorded(ctx, &id, session).await?;
let (roster, space) = recorded(ctx, &id, session).await?;
let open = public(ctx, &id);
Ok(Own {
ctx,
@@ -724,7 +618,6 @@ impl<'a> Own<'a> {
open,
roster,
space,
old,
})
}
@@ -748,8 +641,7 @@ impl<'a> Own<'a> {
Err(err) if space::refused(&err) && !space::scope_missing(&err) => Ok(None),
made => {
made?;
let uri = space_uri(&self.id.did);
seen(self.ctx, &self.id, self.session, &uri).await
seen(self.ctx, &self.id, self.session).await
}
}
}
@@ -919,7 +811,7 @@ pub async fn invite_as(
}
/// Takes `who` (a DID) out of the organization `org` is, as the
/// organization: out of its records, its spaces, and its private
/// organization: out of its records, its space, and its private
/// repositories' spaces, but for those shared with them on their own.
pub async fn remove_as(ctx: &Ctx, org: &Session, who: &str) -> Result<Removed> {
let own = Own::open(ctx, org).await?;
@@ -932,7 +824,7 @@ pub async fn remove_as(ctx: &Ctx, org: &Session, who: &str) -> Result<Removed> {
for place in [Some(own.open.clone()), kept].into_iter().flatten() {
own.unname(&place, who).await?;
}
for space in [&own.space, &own.old].into_iter().flatten() {
if let Some(space) = &own.space {
own.admin().remove_member(&space.uri, who).await?;
}
if admitted {
@@ -965,23 +857,19 @@ pub async fn admit_as(
bail!("{name} has not invited you: ask one of its admins to (`knotless org add`)");
};
// Read at once: their join record, wherever they wrote it, and its
// record of them, in public and in its space. One in its old space
// stands until they join again.
// record of them, in public and in its space.
let member = identity::resolve(ctx, who).await?;
let kept = own.space.as_ref().map(|s| inside(ctx, &own.id, s));
let theirs = own.space.as_ref().map(|s| inside(ctx, &member, s));
let older = own.old.as_ref().map(|s| inside(ctx, &member, s));
let shared = public(ctx, &member);
let (open_join, hidden_join, old_join, shown, held) = tokio::join!(
let (open_join, hidden_join, shown, held) = tokio::join!(
joined(&shared, &own.id.did),
OptionFuture::from(theirs.as_ref().map(|t| joined(t, &own.id.did))),
OptionFuture::from(older.as_ref().map(|t| joined(t, &own.id.did))),
member_record(&own.open, who),
OptionFuture::from(kept.as_ref().map(|k| member_record(k, who))),
);
let (open_join, shown) = (open_join?, shown?);
let hidden_join = hidden_join.transpose()?.unwrap_or(false);
let hidden_join = hidden_join || old_join.transpose()?.unwrap_or(false);
let held = held.transpose()?.flatten();
// Moving the record, they may keep it in both places a moment: the
// one they ask for counts.
@@ -1028,7 +916,7 @@ pub async fn admit_as(
}
/// Ends the organization `org` is, as the organization: takes out each
/// member and invitation, then deletes its spaces, and its records in
/// member and invitation, then deletes its space, and its records in
/// public, a batch at a time.
pub async fn delete_as(ctx: &Ctx, org: &Session) -> Result<Vec<Entry>> {
let own = Own::open(ctx, org).await?;
@@ -1038,7 +926,7 @@ pub async fn delete_as(ctx: &Ctx, org: &Session) -> Result<Vec<Entry>> {
own.reach(&entry.did, false).await?;
}
}
for space in [&own.space, &own.old].into_iter().flatten() {
if let Some(space) = &own.space {
own.admin().delete(&space.uri).await?;
}
let named = member_records(&own.open).await?;
@@ -1066,13 +954,6 @@ async fn ask<T: serde::de::DeserializeOwned>(
lxm: &str,
body: Value,
) -> Result<T> {
let gateway = acts_through(roster)?;
let (xrpc, auth) = gateway::client(ctx, me, gateway, lxm).await?;
xrpc.procedure(lxm, Some(&body), Some(&auth)).await
}
/// The gateway the organization `roster` is of acts through, or why none.
fn acts_through(roster: &Roster) -> Result<&str> {
let Some(gateway) = &roster.gateway else {
bail!(
"{} acts through no gateway: whoever holds its login runs `knotless delegate {}`",
@@ -1080,7 +961,8 @@ fn acts_through(roster: &Roster) -> Result<&str> {
roster.name
);
};
Ok(gateway)
let (xrpc, auth) = gateway::client(ctx, me, gateway, lxm).await?;
xrpc.procedure(lxm, Some(&body), Some(&auth)).await
}
/// `org` (a handle or DID; yours when none), as `me` sees it, and with the
@@ -1226,9 +1108,8 @@ pub async fn delete(ctx: &Ctx) -> Result<(Roster, Vec<Entry>)> {
/// Joins `org` (a handle or DID) as whoever is logged in, seen by anyone
/// or by its members alone: writes the join record where `visibility` says,
/// taking back one in any other place, and has the organization's gateway
/// take the rest in. A hidden one goes in the organization's space, so
/// joining again moves one kept in its old one ([`OLD_SPACE_TYPE`]).
/// taking back one in the other place, and has the organization's gateway
/// take the rest in.
pub async fn join(ctx: &Ctx, org: &str, visibility: Visibility) -> Result<Joined> {
let me = session::me(ctx).await?;
let mine = identity::resolve(ctx, &me.did).await?;
@@ -1238,20 +1119,22 @@ pub async fn join(ctx: &Ctx, org: &str, visibility: Visibility) -> Result<Joined
bail!("{} is not an organization", roster.name);
}
let open = public(ctx, &mine);
let [hidden, old] = space_uris(&org_id.did).map(|uri| Remote {
space: Some(Space::owner(&uri, &me)),
let hidden = Remote {
space: Some(Space::owner(&space_uri(&org_id.did), &me)),
..open.clone()
});
let place = match visibility {
Visibility::Public => &open,
Visibility::Hidden => &hidden,
};
let (place, other) = match visibility {
Visibility::Public => (&open, &hidden),
Visibility::Hidden => (&hidden, &open),
};
let record = OrgAcceptanceRecord {
subject: org_id.did.clone(),
created_at: records::now(),
};
let value = records::typed(nsid::ORG_ACCEPTANCE, &record);
let found = joined(place, &org_id.did).await;
// Both places at once: the other's record, should it be there, goes
// once the gateway took this one in.
let (found, there) = tokio::join!(joined(place, &org_id.did), joined(other, &org_id.did));
let was = matches!(found, Ok(true));
let wrote = match found {
Ok(true) => Ok(()),
@@ -1271,17 +1154,15 @@ pub async fn join(ctx: &Ctx, org: &str, visibility: Visibility) -> Result<Joined
wrote?;
let body = json!({"org": org_id.did, "visibility": visibility});
let asked: Result<Joined> = ask(ctx, &me, &roster, gateway::JOIN_ORG, body).await;
// Refused, the new record joins nothing, so it goes; taken in, those in
// the other places do, which only then are no longer the membership.
let stale = match &asked {
Ok(_) => [&open, &hidden, &old].map(|p| Some(p).filter(|p| !std::ptr::eq(*p, place))),
Err(_) => [(!was).then_some(place), None, None],
// Refused, the new record joins nothing, so it goes; taken in, the one
// in the other place does, which only then is no longer the membership.
let (stale, place) = match &asked {
Ok(_) => (there.unwrap_or(false), other),
Err(_) => (!was, place),
};
for place in stale.into_iter().flatten() {
if joined(place, &org_id.did).await.unwrap_or(false) {
let gone = vec![remote::delete(nsid::ORG_ACCEPTANCE, &org_id.did)];
place.apply(gone, &me.auth()).await?;
}
if stale {
let gone = vec![remote::delete(nsid::ORG_ACCEPTANCE, &org_id.did)];
place.apply(gone, &me.auth()).await?;
}
asked
}
@@ -1298,12 +1179,12 @@ pub async fn leave(ctx: &Ctx, org: &str) -> Result<(String, Option<String>)> {
let name = identity::handles(ctx, [org_id.did.as_str()]).await;
let name = name.get(&org_id.did).cloned().unwrap_or(org_id.did.clone());
let open = public(ctx, &mine);
let [hidden, old] = space_uris(&org_id.did).map(|uri| Remote {
space: Some(Space::owner(&uri, &me)),
let hidden = Remote {
space: Some(Space::owner(&space_uri(&org_id.did), &me)),
..open.clone()
});
};
let mut left = false;
for place in [&open, &hidden, &old] {
for place in [&open, &hidden] {
if joined(place, &org_id.did).await.unwrap_or(false) {
let gone = vec![remote::delete(nsid::ORG_ACCEPTANCE, &org_id.did)];
place.apply(gone, &me.auth()).await?;
crates/knotless-serve/src/index.rs+8−60
View file
@@ -2765,27 +2765,15 @@ impl Index {
/// public records when empty) name, the first invited first, each with
/// whether they keep a record of joining it there.
pub fn org_roster(&self, org: &str, space: &str) -> rusqlite::Result<Vec<Invitee>> {
self.org_roster_with(org, space, space)
}
/// [`Index::org_roster`], a record of joining kept in `also` counting
/// too: an organization's old space, for its space, as a hidden member
/// keeps theirs there until they join again.
pub fn org_roster_with(
&self,
org: &str,
space: &str,
also: &str,
) -> rusqlite::Result<Vec<Invitee>> {
let db = self.db.lock().unwrap();
let mut stmt = db.prepare_cached(
"SELECT m.did, m.role, m.invited_at, m.expires_at, m.joined_at,
EXISTS (SELECT 1 FROM membership j
WHERE j.space IN (m.space, ?3) AND j.did = m.did AND j.org = m.org)
WHERE j.space = m.space AND j.did = m.did AND j.org = m.org)
FROM org_member m WHERE m.space = ?1 AND m.org = ?2
ORDER BY m.invited_at, m.did",
)?;
let rows = stmt.query_map([space, org, also], invitee)?;
let rows = stmt.query_map([space, org], invitee)?;
rows.collect()
}
@@ -2807,46 +2795,18 @@ impl Index {
Ok(found.optional()?.flatten())
}
/// The organizations whose public records name `gateway` the one they
/// act through, by DID.
pub fn orgs_through(&self, gateway: &str) -> rusqlite::Result<Vec<String>> {
let db = self.db.lock().unwrap();
let mut stmt = db.prepare_cached(
"SELECT did FROM org WHERE space = '' AND gateway = ?1 ORDER BY did
-- scans: org, of a row an organization, once as the server starts",
)?;
stmt.query_map([gateway], |row| row.get(0))?.collect()
}
/// How many records of joining an organization are in an old space of
/// one ([`knotless::org::OLD_SPACE_TYPE`]): hidden members' who have
/// not joined again since it moved.
pub fn old_org_joins(&self) -> rusqlite::Result<u64> {
let db = self.db.lock().unwrap();
let old = format!("/space/{}/self", knotless::org::OLD_SPACE_TYPE);
db.query_row(
"SELECT count(*) FROM membership WHERE space = 'at://' || org || ?1
-- scans: membership, of a row a membership, once as the server starts",
[old],
|row| row.get(0),
)
}
/// The organizations `did` is a member of, by DID: in public, or with
/// `hidden` those whose spaces name them taken in, where they keep a
/// record of joining: its space, or its old one, either.
/// record of joining.
pub fn orgs_of(&self, did: &str, hidden: bool) -> rusqlite::Result<Vec<String>> {
let db = self.db.lock().unwrap();
let space = |kind: &str| {
format!("CASE WHEN ?2 THEN 'at://' || m.org || '/space/{kind}/self' ELSE '' END")
};
let spaces = [knotless::org::SPACE_TYPE, knotless::org::OLD_SPACE_TYPE].map(space);
let spaces = spaces.join(", ");
let mut stmt = db.prepare_cached(&format!(
"SELECT DISTINCT m.org FROM org_member m
JOIN membership j ON j.space IN ({spaces}) AND j.did = m.did AND j.org = m.org
WHERE m.did = ?1 AND m.joined_at IS NOT NULL AND m.space IN ({spaces})
"SELECT m.org FROM org_member m
JOIN membership j ON j.space = m.space AND j.did = m.did AND j.org = m.org
WHERE m.did = ?1 AND m.joined_at IS NOT NULL
AND m.space = CASE WHEN ?2 THEN 'at://' || m.org || '/space/{}/self' ELSE '' END
ORDER BY m.org",
knotless::org::SPACE_TYPE
))?;
stmt.query_map(params![did, hidden], |row| row.get(0))?
.collect()
@@ -5504,18 +5464,6 @@ mod tests {
.put_from_space(&space, "did:h", nsid::ORG_ACCEPTANCE, "did:o", &joined)
.unwrap();
assert_eq!(named(&space), [pair("did:h", true)]);
// One kept in its old space counts for its space too, as a hidden
// member's does until they join again.
let old = knotless::org::old_space_uri("did:o");
index
.delete_from_space(&space, "did:h", nsid::ORG_ACCEPTANCE, "did:o")
.unwrap();
index
.put_from_space(&old, "did:h", nsid::ORG_ACCEPTANCE, "did:o", &joined)
.unwrap();
assert_eq!(named(&space), [pair("did:h", false)]);
let roster = index.org_roster_with("did:o", &space, &old).unwrap();
assert!(roster[0].joined);
assert_eq!(named("").len(), 2);
assert_eq!(index.orgs_of("did:h", true).unwrap(), ["did:o"]);
assert!(index.orgs_of("did:h", false).unwrap().is_empty());
crates/knotless-serve/src/lib.rs+0−2
View file
@@ -1171,8 +1171,6 @@ pub async fn run(opts: Options, ctx: Ctx) -> Result<()> {
tokio::spawn(jetstream::watch(app.clone(), url));
}
tokio::spawn(spaces::follow(app.clone(), SPACES));
let moving = app.clone();
tokio::spawn(async move { moving.move_org_spaces().await });
tokio::spawn(languages::follow(app.clone()));
tokio::spawn(telemetry::summaries(app.clone(), SUMMARY));
let listener = tokio::net::TcpListener::bind(&opts.listen).await?;
crates/knotless-serve/src/orgs.rs+33−89
View file
@@ -29,19 +29,10 @@ impl App {
return None;
}
let open = index.org_roster(org, "").unwrap_or_default();
// Its space and its old one, each a record of joining in the other
// counting too, as while it moves: the one in its space counts.
let [space, old] = org::space_uris(org);
let mut kept = Vec::new();
for (space, also) in [(&space, &old), (&old, &space)] {
let roster = self.spaces.as_ref();
let roster = roster.and_then(|i| i.org_roster_with(org, space, also).ok());
let roster = roster.unwrap_or_default().into_iter();
let unkept: Vec<Invitee> = roster
.filter(|m| !kept.iter().any(|k: &Invitee| k.did == m.did))
.collect();
kept.extend(unkept);
}
let space = org::space_uri(org);
let kept = self.spaces.as_ref();
let kept = kept.and_then(|i| i.org_roster(org, &space).ok());
let kept = kept.unwrap_or_default();
// Moving between the two, a record may be in both a moment.
let mut listed = open.clone();
let unlisted = kept.iter().filter(|k| !open.iter().any(|m| m.did == k.did));
@@ -56,15 +47,12 @@ impl App {
}
/// Whether this server keeps an index of the organization `org`'s
/// space, or its old one, synced at least once, which alone says who
/// is a hidden member. One with nothing in it never is, and holds no
/// one to say.
/// space, synced at least once, which alone says who is a hidden
/// member. One with nothing in it never is, and holds no one to say.
pub fn org_kept(&self, org: &str) -> bool {
let synced = |uri: &String| {
let synced = self.spaces.as_ref().map(|i| i.space_synced(uri));
matches!(synced, Some(Ok(true)))
};
org::space_uris(org).iter().any(synced)
let uri = org::space_uri(org);
let synced = self.spaces.as_ref().map(|i| i.space_synced(&uri));
matches!(synced, Some(Ok(true)))
}
/// Where `did` stands in the organization `org`, if it invited them.
@@ -115,21 +103,19 @@ impl App {
}
}
/// Keeps an index of the organization `org`'s space, and of its old
/// one, synced as `viewer`, when none is kept: read from the index from
/// then on, as a private repository's is ([`App::keep`]). A viewer who
/// may not read one, or one there is not, leaves none, and is not
/// asked again for a while.
/// Keeps an index of the organization `org`'s space, synced as
/// `viewer`, when none is kept: read from the index from then on, as a
/// private repository's is ([`App::keep`]). A viewer who may not read
/// it leaves none, and is not asked again for a while.
pub async fn keep_org(&self, org: &str, viewer: &str) {
for uri in org::space_uris(org) {
let unkept = self.spaces.as_ref();
let unkept = unkept.filter(|i| matches!(i.syncer_of(&uri), Ok(None)));
let tried = unkept.filter(|_| self.discovering(&format!("{uri} {viewer}")));
if let Some(index) = tried
&& self.keep(&uri, viewer).await.is_err()
{
self.unfollow(index, &uri);
}
let uri = org::space_uri(org);
let unkept = self.spaces.as_ref();
let unkept = unkept.filter(|i| matches!(i.syncer_of(&uri), Ok(None)));
let tried = unkept.filter(|_| self.discovering(&format!("{uri} {viewer}")));
if let Some(index) = tried
&& self.keep(&uri, viewer).await.is_err()
{
self.unfollow(index, &uri);
}
}
@@ -191,66 +177,24 @@ impl App {
}
/// Takes out of the indexes `member`'s record of joining `org` in the
/// places they left, having joined with `visibility`: in public when they
/// moved to its space, else in its spaces, the old one too. One who
/// joined its space again leaves its old one.
/// place they left, having joined with `visibility`: in public when they
/// moved to its space, else in its space.
pub fn unjoined(&self, org: &str, member: &str, visibility: Visibility) {
let [space, old] = org::space_uris(org).map(Some);
let left = match visibility == Visibility::Hidden {
true => [(self.index.as_ref(), None), (self.spaces.as_ref(), old)],
false => [(self.spaces.as_ref(), space), (self.spaces.as_ref(), old)],
let moved_out = visibility == Visibility::Hidden;
let (index, space) = match moved_out {
true => (self.index.as_ref(), None),
false => (self.spaces.as_ref(), Some(org::space_uri(org))),
};
for (index, space) in left {
let gone =
index.map(|i| i.delete_in(space.as_deref(), member, nsid::ORG_ACCEPTANCE, org));
let _ = gone;
}
let gone = index.map(|i| i.delete_in(space.as_deref(), member, nsid::ORG_ACCEPTANCE, org));
let _ = gone;
}
/// Follows `org`'s space in `index` and syncs it, as the organization,
/// and its old one too while that is followed, which an organization
/// made since never had.
/// Follows `org`'s space in `index` and syncs it, as the organization.
async fn resync(&self, index: &Index, org: &str) {
let uri = org::space_uri(org);
let _ = index.follow_space(&uri, org);
let ctx = crate::login::syncer_ctx(self, org);
let [uri, old] = org::space_uris(org);
let old = Some(old).filter(|old| matches!(index.syncer_of(old), Ok(Some(_))));
for uri in [Some(uri), old].into_iter().flatten() {
let _ = index.follow_space(&uri, org);
let _ = crate::spaces::sync(&ctx, index, &uri).await;
}
}
/// Moves each organization this gateway acts for out of its old space
/// ([`org::move_space`]), as it starts, alone among changes to it, and
/// writes what moved into the indexes; then logs how many it moved and
/// how many it could not, which it tries again as it next starts, or
/// as one is next changed, and how many hidden members' records of
/// joining are in old spaces still. None it could not, it is done.
pub async fn move_org_spaces(&self) {
let (Some(gateway), Some(index)) = (self.gateway.as_ref(), self.index.as_ref()) else {
return;
};
let orgs = index.orgs_through(&gateway.did).unwrap_or_default();
let (mut moved, mut failed) = (0, 0);
for org in orgs {
let lock = self.org_lock(&org);
let _held = lock.lock().await;
let done = async {
let id = knotless::identity::resolve(&self.ctx, &org).await?;
let session = knotless::session::ensure(&self.ctx, &id).await?;
org::move_space(&self.ctx, &session).await
};
let done = done.await;
crate::telemetry::org_space_moved(&org, &done);
moved += u64::from(done.is_ok());
failed += u64::from(done.is_err());
self.reindex_org(&org, &org).await;
}
let spaces = self.spaces.as_ref();
let left = spaces
.and_then(|i| i.old_org_joins().ok())
.unwrap_or_default();
crate::telemetry::org_spaces_moved(moved, failed, left);
let _ = crate::spaces::sync(&ctx, index, &uri).await;
}
}
crates/knotless-serve/src/spaces.rs+1−5
View file
@@ -227,11 +227,7 @@ async fn from_authority(
let aud = format!("{}{}", did.unwrap_or_default(), space::APPVIEW);
let caller = crate::gateway::caller_for(app, headers, &aud, lxm).await?;
let theirs = space.starts_with(&space::uri(&caller, ""));
if !theirs
&& !knotless::org::space_uris(&caller)
.iter()
.any(|uri| uri == space)
{
if !theirs && space != knotless::org::space_uri(&caller) {
let why = "only a private repository's authority tells of its space";
return Err(AppError(StatusCode::FORBIDDEN, why.into()));
}
crates/knotless-serve/src/telemetry.rs+0−33
View file
@@ -325,39 +325,6 @@ pub fn labeler<T>(did: &str, done: anyhow::Result<T>) {
}
}
/// Logs, at warn, why the organization `org` could not be moved out of
/// its old space as this server started, if it could not: it is tried
/// again as it is next changed, and as this server next starts.
pub fn org_space_moved(org: &str, done: &anyhow::Result<()>) {
if let Err(err) = done {
let fields = Fields {
did: Some(org),
error: Some(format!("{err:#}")),
..Fields::default()
};
telemetry::note(Level::WARN, "org space not moved", &fields);
}
}
/// Logs, at info, how moving organizations out of their old spaces went
/// as this server started: how many it moved, or found moved (`count`),
/// and how many it could not (`failed`); none failed, their part is done.
/// Then how many hidden members' records of joining are in old spaces
/// still (`count`), each moved as its member joins again.
pub fn org_spaces_moved(moved: u64, failed: u64, joins: u64) {
let fields = Fields {
count: Some(moved),
failed: Some(failed),
..Fields::default()
};
telemetry::note(Level::INFO, "org spaces moved", &fields);
let fields = Fields {
count: Some(joins),
..Fields::default()
};
telemetry::note(Level::INFO, "org joins in old spaces", &fields);
}
/// Logs that the index was opened at startup, taking `took`.
pub fn opened(index: Option<&crate::index::Index>, took: Duration) {
if let Some(index) = index {
crates/knotless-serve/tests/gateway/orgs.rs+0−100
View file
@@ -504,103 +504,3 @@ fn private_invitations_heard_of_from_clients() {
let said = heard();
assert!(said.contains(o) && said.contains("as member"), "{said}");
}
/// An organization kept as knotless kept one before: its invitations and
/// hidden members in a space of the old type, with a hidden member's
/// record of joining there. As the gateway starts, it moves the
/// organization's part into its space, each member standing where they
/// stood; a member's own record moves as they join again; and moving
/// again moves nothing.
#[test]
fn organizations_move_out_of_old_spaces() {
let (gw, [org, member, invitee, other]) = served();
if !net().spaces {
return;
}
let (o, m, i) = (
org.account.handle.as_str(),
member.account.handle.as_str(),
invitee.account.handle.as_str(),
);
ok(&org, &["delegate", o, &gw.did]);
let did = org.account.did.as_str();
let old = knotless::org::old_space_uri(did);
let policy = json!({"$type": "com.atproto.memberspace.policy"});
let made = json!({"spaceType": "dev.knotless.org", "skey": "self", "policy": policy.clone()});
org.call("com.atproto.space.createSpace", made);
for (who, read) in [(&member, true), (&invitee, false)] {
let put = json!({"space": old, "did": who.account.did, "read": read, "write": true});
org.call("com.atproto.memberspace.putMember", put);
}
let at = "2026-10-30T00:00:00.000Z";
let named = json!({"$type": "dev.knotless.org", "gateway": gw.did, "createdAt": at});
org.put_record("dev.knotless.org", "self", named.clone());
let hidden = json!({"$type": "dev.knotless.orgMember", "member": member.account.did,
"roles": [{"role": "member"}], "invitedAt": at, "joinedAt": at});
org.put_space_record(&old, "dev.knotless.orgMember", &member.account.did, hidden);
let asked = json!({"$type": "dev.knotless.orgMember", "member": invitee.account.did,
"roles": [{"role": "member"}], "invitedAt": at, "expiresAt": "2099-01-01T00:00:00.000Z"});
org.put_space_record(
&old,
"dev.knotless.orgMember",
&invitee.account.did,
asked.clone(),
);
// A move cut off part way: the new space made, holding the invitee
// and its record of them, which counts over the old one's.
let space = knotless::org::space_uri(did);
let made = json!({"spaceType": "dev.knotless.orgSpace", "skey": "self", "policy": policy});
org.call("com.atproto.space.createSpace", made);
let put = json!({"space": space, "did": invitee.account.did, "read": false, "write": true});
org.call("com.atproto.memberspace.putMember", put);
org.put_space_record(
&space,
"dev.knotless.orgMember",
&invitee.account.did,
asked,
);
let joined = json!({"$type": "dev.knotless.orgAcceptance", "subject": did, "createdAt": at});
member.put_space_record(&old, "dev.knotless.orgAcceptance", did, joined);
// One that names this gateway and never delegated to it is not moved.
other.put_record("dev.knotless.org", "self", named);
reindexed(&gw, &other);
// Followed, as a server kept it before.
let index = gw.app.spaces.as_ref().unwrap();
index.follow_space(&old, did).unwrap();
reindexed(&gw, &org);
let members = || ok(&org, &["org", "members"]);
let stood = [
format!("{m}\tmember\thidden"),
format!("{i}\tmember\tinvited"),
];
let stands = |shown: &str| stood.iter().all(|s| shown.contains(s.as_str()));
let shown = members();
assert!(stands(&shown), "{shown}");
gw._rt.block_on(gw.app.move_org_spaces());
let roster = |uri: &str| index.org_roster(did, uri).unwrap().len();
assert_eq!((roster(&space), roster(&old)), (2, 0));
let shown = members();
assert!(stands(&shown), "{shown}");
assert_eq!(index.old_org_joins().unwrap(), 1);
// Moved, it moves nothing.
gw._rt.block_on(gw.app.move_org_spaces());
assert_eq!((roster(&space), roster(&old)), (2, 0));
// Joining again, a hidden member's record of it moves too; an
// invitee joins in its space.
for who in [&member, &invitee] {
assert!(ok(who, &["org", "join", o, "--hidden"]).starts_with("You're a member"));
}
reindexed(&gw, &org);
assert_eq!(index.old_org_joins().unwrap(), 0);
let shown = members();
assert!(shown.contains(&format!("{i}\tmember\thidden")), "{shown}");
assert!(shown.contains(&stood[0]), "{shown}");
// Ended, it takes both spaces with it.
ok(&org, &["org", "delete", "--yes"]);
reindexed(&gw, &org);
let how = "`knotless org add <handle>` invites someone, and makes it one";
assert_eq!(members(), format!("{o} is not an organization: {how}."));
}
docs/design/access.md+9−10
View file
docs/design/gateway.md+8−46
View file
docs/design/serve.md+1−3
View file
harness/pds.mjs+31−1
View file