~ 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

An organization's space has a type of its own, dev.knotless.orgSpace

Merged chadtmiller.com asked to merge into main from chadtmiller.com/knotless:space/org-space
.changeset/org-space-type.mdnew+7−0
View file
@@ -0,0 +1,7 @@
---
default: patch
---
# An organization's space has a type of its own, `dev.knotless.orgSpace`
An organization keeps its invitations and hidden members in a space of the type `dev.knotless.orgSpace`, which has its own lexicon, so that a login can name it. Its gateway moves an organization out of its old space, of the type `dev.knotless.org`, as it starts and as the organization next changes, keeping each member where they stood. A hidden member's record of joining stays in the old space, and counts, until they join again: `knotless org join <org> --hidden` moves it.
crates/knotless/src/org.rs+168−49
View file
@@ -50,7 +50,12 @@ 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.org";
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;
@@ -66,6 +71,17 @@ 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")]
@@ -230,11 +246,11 @@ fn inside(ctx: &Ctx, id: &Identity, space: &Space) -> Remote {
}
}
/// 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>> {
/// 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>> {
let host = Xrpc::new(ctx.http.clone(), &org.pds);
let joined = Space::join(ctx, &host, &space_uri(&org.did), &org.did, reader).await;
let joined = Space::join(ctx, &host, uri, &org.did, reader).await;
unless_shut(joined.map(Some))
}
@@ -487,20 +503,27 @@ 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 space, as `org` reads it.
async fn recorded(ctx: &Ctx, id: &Identity, org: &Session) -> Result<(Roster, Option<Space>)> {
/// 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)> {
let open = public(ctx, id);
// 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!(
// 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!(
open.get_record::<OrgRecord>(nsid::ORG, RKEY),
member_records(&open),
seen(ctx, id, org)
seen(ctx, id, org, &uri),
seen(ctx, id, org, &old),
);
let named = named?.map(|r| r.value);
let (shown, space) = match &named {
Some(_) => (shown?, space?),
None => (Vec::new(), None),
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 kept = match &space {
Some(space) => member_records(&inside(ctx, id, space)).await?,
@@ -540,7 +563,7 @@ async fn recorded(ctx: &Ctx, id: &Identity, org: &Session) -> Result<(Roster, Op
gateway: named.and_then(|o| o.gateway),
entries,
};
Ok((roster, space))
Ok((roster, Spaces { space, old }))
}
/// The organization `org` is, read as itself from its own records alone
@@ -551,6 +574,88 @@ 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")]
@@ -596,7 +701,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.
/// its records: its space as it opened, if it had one, and its old one.
struct Own<'a> {
ctx: &'a Ctx,
session: &'a Session,
@@ -604,12 +709,13 @@ 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, space) = recorded(ctx, &id, session).await?;
let (roster, Spaces { space, old }) = recorded(ctx, &id, session).await?;
let open = public(ctx, &id);
Ok(Own {
ctx,
@@ -618,6 +724,7 @@ impl<'a> Own<'a> {
open,
roster,
space,
old,
})
}
@@ -641,7 +748,8 @@ impl<'a> Own<'a> {
Err(err) if space::refused(&err) && !space::scope_missing(&err) => Ok(None),
made => {
made?;
seen(self.ctx, &self.id, self.session).await
let uri = space_uri(&self.id.did);
seen(self.ctx, &self.id, self.session, &uri).await
}
}
}
@@ -811,7 +919,7 @@ pub async fn invite_as(
}
/// Takes `who` (a DID) out of the organization `org` is, as the
/// organization: out of its records, its space, and its private
/// organization: out of its records, its spaces, 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?;
@@ -824,7 +932,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?;
}
if let Some(space) = &own.space {
for space in [&own.space, &own.old].into_iter().flatten() {
own.admin().remove_member(&space.uri, who).await?;
}
if admitted {
@@ -857,19 +965,23 @@ 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.
// record of them, in public and in its space. One in its old space
// stands until they join again.
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, shown, held) = tokio::join!(
let (open_join, hidden_join, old_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.
@@ -916,7 +1028,7 @@ pub async fn admit_as(
}
/// Ends the organization `org` is, as the organization: takes out each
/// member and invitation, then deletes its space, and its records in
/// member and invitation, then deletes its spaces, 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?;
@@ -926,7 +1038,7 @@ pub async fn delete_as(ctx: &Ctx, org: &Session) -> Result<Vec<Entry>> {
own.reach(&entry.did, false).await?;
}
}
if let Some(space) = &own.space {
for space in [&own.space, &own.old].into_iter().flatten() {
own.admin().delete(&space.uri).await?;
}
let named = member_records(&own.open).await?;
@@ -954,6 +1066,13 @@ 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 {}`",
@@ -961,8 +1080,7 @@ async fn ask<T: serde::de::DeserializeOwned>(
roster.name
);
};
let (xrpc, auth) = gateway::client(ctx, me, gateway, lxm).await?;
xrpc.procedure(lxm, Some(&body), Some(&auth)).await
Ok(gateway)
}
/// `org` (a handle or DID; yours when none), as `me` sees it, and with the
@@ -1108,8 +1226,9 @@ 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 the other place, and has the organization's gateway
/// take the rest in.
/// 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`]).
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?;
@@ -1119,22 +1238,20 @@ 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 = Remote {
space: Some(Space::owner(&space_uri(&org_id.did), &me)),
let [hidden, old] = space_uris(&org_id.did).map(|uri| Remote {
space: Some(Space::owner(&uri, &me)),
..open.clone()
};
let (place, other) = match visibility {
Visibility::Public => (&open, &hidden),
Visibility::Hidden => (&hidden, &open),
});
let place = match visibility {
Visibility::Public => &open,
Visibility::Hidden => &hidden,
};
let record = OrgAcceptanceRecord {
subject: org_id.did.clone(),
created_at: records::now(),
};
let value = records::typed(nsid::ORG_ACCEPTANCE, &record);
// 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 found = joined(place, &org_id.did).await;
let was = matches!(found, Ok(true));
let wrote = match found {
Ok(true) => Ok(()),
@@ -1154,15 +1271,17 @@ 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, 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),
// 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],
};
if stale {
let gone = vec![remote::delete(nsid::ORG_ACCEPTANCE, &org_id.did)];
place.apply(gone, &me.auth()).await?;
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?;
}
}
asked
}
@@ -1179,12 +1298,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 = Remote {
space: Some(Space::owner(&space_uri(&org_id.did), &me)),
let [hidden, old] = space_uris(&org_id.did).map(|uri| Remote {
space: Some(Space::owner(&uri, &me)),
..open.clone()
};
});
let mut left = false;
for place in [&open, &hidden] {
for place in [&open, &hidden, &old] {
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/tests/common/mod.rs+7−1
View file
@@ -577,8 +577,14 @@ impl Sandbox {
}
fn space_call(&self, method: &str, body: serde_json::Value) {
self.call(&format!("com.atproto.space.{method}"), body);
}
/// Calls the procedure `nsid` of the account's PDS as the account, as
/// any client could, which must pass.
pub fn call(&self, nsid: &str, body: serde_json::Value) {
reqwest::blocking::Client::new()
.post(format!("{}/xrpc/com.atproto.space.{method}", net().pds))
.post(format!("{}/xrpc/{nsid}", net().pds))
.bearer_auth(self.token())
.json(&body)
.send()
crates/knotless-serve/src/index.rs+60−8
View file
@@ -2765,15 +2765,27 @@ 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 = m.space AND j.did = m.did AND j.org = m.org)
WHERE j.space IN (m.space, ?3) 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], invitee)?;
let rows = stmt.query_map([space, org, also], invitee)?;
rows.collect()
}
@@ -2795,18 +2807,46 @@ 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.
/// record of joining: its space, or its old one, either.
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 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
"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})
ORDER BY m.org",
knotless::org::SPACE_TYPE
))?;
stmt.query_map(params![did, hidden], |row| row.get(0))?
.collect()
@@ -5464,6 +5504,18 @@ 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+2−0
View file
@@ -1171,6 +1171,8 @@ 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+89−33
View file
@@ -29,10 +29,19 @@ impl App {
return None;
}
let open = index.org_roster(org, "").unwrap_or_default();
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();
// 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);
}
// 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));
@@ -47,12 +56,15 @@ impl App {
}
/// Whether this server keeps an index of the organization `org`'s
/// 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.
/// 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.
pub fn org_kept(&self, org: &str) -> bool {
let uri = org::space_uri(org);
let synced = self.spaces.as_ref().map(|i| i.space_synced(&uri));
matches!(synced, Some(Ok(true)))
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)
}
/// Where `did` stands in the organization `org`, if it invited them.
@@ -103,19 +115,21 @@ impl App {
}
}
/// 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.
/// 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.
pub async fn keep_org(&self, org: &str, viewer: &str) {
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);
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);
}
}
}
@@ -177,24 +191,66 @@ impl App {
}
/// Takes out of the indexes `member`'s record of joining `org` in the
/// place they left, having joined with `visibility`: in public when they
/// moved to its space, else in its space.
/// 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.
pub fn unjoined(&self, org: &str, member: &str, visibility: Visibility) {
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))),
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 gone = index.map(|i| i.delete_in(space.as_deref(), member, nsid::ORG_ACCEPTANCE, org));
let _ = gone;
for (index, space) in left {
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.
/// 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.
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 _ = crate::spaces::sync(&ctx, index, &uri).await;
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);
}
}
crates/knotless-serve/src/spaces.rs+5−1
View file
@@ -227,7 +227,11 @@ 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 && space != knotless::org::space_uri(&caller) {
if !theirs
&& !knotless::org::space_uris(&caller)
.iter()
.any(|uri| uri == space)
{
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+33−0
View file
@@ -325,6 +325,39 @@ 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+103−3
View file
@@ -214,7 +214,7 @@ fn organizations_invite_and_members_join() {
let rkey = third.account.did.as_str();
match net().spaces {
true => {
let space = format!("at://{}/space/dev.knotless.org/self", org.account.did);
let space = knotless::org::space_uri(&org.account.did);
org.put_space_record(&space, "dev.knotless.orgMember", rkey, old);
}
false => {
@@ -400,7 +400,7 @@ fn organizations_keep_hidden_members() {
// keeping that repository for a reader, it follows that space as them.
let (app, did) = (&gw.app, &org.account.did);
let index = app.spaces.as_ref().unwrap();
let org_space = format!("at://{did}/space/dev.knotless.org/self");
let org_space = knotless::org::space_uri(did);
app.unfollow(index, &org_space);
// One who may not read that space asking leaves it unkept.
assert_eq!(ok(&stranger, &["org", "members", o]), "");
@@ -468,7 +468,7 @@ fn organizations_keep_hidden_members() {
// Gone while no gateway answers, their join record is all the index
// goes by as it hears of it: they push to none of it, and no reader
// counts them, though its records name them until they are taken out.
let space = format!("at://{}/space/dev.knotless.org/self", org.account.did);
let space = knotless::org::space_uri(&org.account.did);
let did = &org.account.did;
member.delete_space_record(&space, "dev.knotless.orgAcceptance", did);
reindexed(&gw, &org);
@@ -504,3 +504,103 @@ 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+11−8
View file
@@ -64,12 +64,15 @@ then asks again at its next login.
The space scope names every space type. An authorization server asked
for a space type by name resolves that type's lexicon as the account
approves, and refuses the login if it cannot. A private repository's
type has one, `dev.knotless.space`, published with knotless's other
lexicons, which lists the collections its space holds. An organization's
space has none: its type, `dev.knotless.org`, is the NSID of the
organization's record, whose lexicon is no space. A PDS without spaces
drops the scope it does not know.
approves, and refuses the login if it cannot. Each of knotless's space
types has one, published with its other lexicons, which lists the
collections its space holds: `dev.knotless.space`, a private
repository's, and `dev.knotless.orgSpace`, an organization's. An
organization's space was of the type `dev.knotless.org` before, the NSID
of its record, whose lexicon is no space, so no scope could name it;
each is moved to the new type
([Moving out of the old space](gateway.md#moving-out-of-the-old-space)).
A PDS without spaces drops the scope it does not know.
## Private repositories
@@ -158,8 +161,8 @@ keeps a `dev.knotless.orgMember` record there for each of them, so its
readers know who of them pushes to it. A member shared it on their own as well, for
reading or writing, stays when they leave; unsharing a member leaves
them in while they are one. The organization's member list is a space of
its own, `at://<org>/space/dev.knotless.org/self`, which only it and its
members read.
its own, `at://<org>/space/dev.knotless.orgSpace/self`, which only it and
its members read.
A fork of a private repository is private, and is shared for reading with
the source's owner so that they can take pulls from it, made from the
docs/design/gateway.md+44−3
View file
docs/design/serve.md+3−1
View file
lexicons/dev/knotless/orgSpace.jsonnew+13−0
View file
web/lexicons/README.md+4−4
View file
web/package.json+1−1
View file