johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! Organizations, as this server's indexes know them, which is how every
//! reader learns of them ([`knotless::org`] asks `dev.knotless.getOrg`): the
//! appview's, of everyone's public records, names each organization's
//! members in public; this server's own index of spaces names the rest,
//! from the organization's space, where its invitations and hidden members
//! are, and from each of its private repositories' spaces, where it names
//! the members who push to that one. The gateway writes what it changes
//! into both as it changes it ([`App::reindex_org`]), so a member pushes
//! the moment they are taken in, and not a moment after they leave.
use crate::App;
use crate::index::{Index, Invitee};
use knotless::org::{self, Role, Standing, Visibility};
use knotless::records::{RepoRecord, nsid};
use knotless::remote::Remote;
/// An account an organization invited, and where it stands.
pub struct Stood {
pub invitee: Invitee,
pub standing: Standing,
}
impl App {
/// Every account the organization `org` invited, the first invited
/// first, each with where it stands: none when it is no organization.
pub fn org_standings(&self, org: &str) -> Option<Vec<Stood>> {
let index = self.index.as_ref()?;
if !index.is_org(org).unwrap_or(false) {
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();
// 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));
listed.extend(unlisted.cloned());
listed.sort_by(|a, b| (&a.invited_at, &a.did).cmp(&(&b.invited_at, &b.did)));
let find = |among: &[Invitee], did: &str| among.iter().find(|m| m.did == did).cloned();
let stood = listed.into_iter().map(|m| Stood {
standing: standing(&m, find(&open, &m.did), find(&kept, &m.did)),
invitee: m,
});
Some(stood.collect())
}
/// 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.
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)))
}
/// Where `did` stands in the organization `org`, if it invited them.
pub fn org_standing(&self, org: &str, did: &str) -> Option<Stood> {
let all = self.org_standings(org).unwrap_or_default();
all.into_iter().find(|s| s.invitee.did == did)
}
/// Whether `did` may act for the organization `org`: it is the
/// organization, or an admin who joined.
pub fn org_admits(&self, org: &str, did: &str) -> bool {
let stood = self.org_standing(org, did);
let admin = |s: &Stood| Role::of(s.invitee.role.as_deref()) == Role::Admin;
did == org || stood.is_some_and(|s| admin(&s) && s.standing.joined())
}
/// The members of the organization `owner` is who may push to its
/// repository in `space`, or a public one with none: its members in
/// public, and for a private one, those the record in its space names,
/// who are members still, hidden or not, when this server keeps an
/// index of the organization's space; else every one it names, as a
/// reader outside the organization counts them.
pub fn org_members(&self, owner: &str, space: Option<&str>) -> Vec<String> {
let all = self.org_standings(owner).unwrap_or_default();
let indexed = self.org_kept(owner);
let joined = all.iter().filter(|s| s.standing.joined());
let joined: std::collections::HashSet<&str> =
joined.map(|s| s.invitee.did.as_str()).collect();
let still = |did: &str| !indexed || joined.contains(did);
let open = all.iter().filter(|s| s.standing == Standing::Public);
let mut members: Vec<String> = open.map(|s| s.invitee.did.clone()).collect();
let kept = space.zip(self.spaces.as_ref());
let kept = kept.and_then(|(space, i)| i.org_roster(owner, space).ok());
let pushers = kept.into_iter().flatten().filter(|m| still(&m.did));
members.extend(pushers.map(|m| m.did));
let mut seen = std::collections::HashSet::new();
members.retain(|m| seen.insert(m.clone()));
members
}
/// `record`, of a repository of `owner`'s in `space` if private, with
/// the members of the organization `owner` is who may push to it
/// ([`App::org_members`]), as `RepoRecord::members` holds them.
pub fn with_members(&self, owner: &str, space: Option<&str>, record: RepoRecord) -> RepoRecord {
RepoRecord {
members: self.org_members(owner, space),
..record
}
}
/// 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) {
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);
}
}
/// Follows the space of the organization that owns the private
/// repository in `space`, as `syncer`, who reads that repository, when
/// none syncs it yet: its readers among the organization's members read
/// its space too, which says who of them is a member still.
pub fn follow_org(&self, space: &str, syncer: &str) {
let owner = space.trim_start_matches("at://").split('/').next();
let owner = owner.unwrap_or_default();
let uri = org::space_uri(owner);
let unkept = self.spaces.as_ref();
let unkept = unkept.filter(|i| matches!(i.syncer_of(&uri), Ok(None)));
let org = unkept.filter(|_| uri != space && self.org_standings(owner).is_some());
if let Some(index) = org {
let _ = index.follow_space(&uri, syncer);
self.due.lock().unwrap().insert(uri);
self.told.notify_one();
}
}
/// Writes into the indexes what this gateway just changed of the
/// organization `org`, as `org`, about `member`: its public records, of
/// being one and of the member, and the member's public join record,
/// read again from their PDSes, and its space, synced as the
/// organization. What the event stream and
/// the spaces' own syncing bring later then changes nothing.
pub async fn reindex_org(&self, org: &str, member: &str) {
let reread = self
.index
.as_ref()
.map(|index| self.reread(index, org, member));
let synced = self.spaces.as_ref().map(|index| self.resync(index, org));
// Apart, as each waits on PDSes.
tokio::join!(
futures::future::OptionFuture::from(reread),
futures::future::OptionFuture::from(synced)
);
}
/// Reads `org`'s public records, of being one and of `member`, and
/// `member`'s public join record into `index` again: one that is gone
/// goes from it, and one that could not be read is left as it was.
async fn reread(&self, index: &Index, org: &str, member: &str) {
let found = [
(org, nsid::ORG, "self"),
(org, nsid::ORG_MEMBER, member),
(member, nsid::ORG_ACCEPTANCE, org),
];
let reads = found.map(|(did, collection, rkey)| record(self, did, collection, rkey));
let reads = futures::future::join_all(reads).await;
for ((did, collection, rkey), read) in found.into_iter().zip(reads) {
let _ = match read {
Ok(Some(value)) => index.put(did, collection, rkey, &value),
Ok(None) => index.delete(did, collection, rkey),
Err(_) => Ok(()),
};
}
}
/// 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.
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 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.
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;
}
}
/// `did`'s record `collection`/`rkey` in their public records, as their PDS
/// has it now: none when there is none.
async fn record(
app: &App,
did: &str,
collection: &str,
rkey: &str,
) -> anyhow::Result<Option<serde_json::Value>> {
let url = knotless::url::RepoUrl::parse(&format!("knotless://{did}/_"))?;
let remote = Remote::connect(&app.ctx, &url).await?;
let found = remote.get_record::<serde_json::Value>(collection, rkey);
Ok(found.await?.map(|r| r.value))
}
/// Where `m` stands: a member in public when the organization's public
/// record (`open`) names them taken in and they keep a public record of
/// joining; a hidden one when its space's record (`kept`) does and they
/// keep one there; else invited, until it lapses.
fn standing(m: &Invitee, open: Option<Invitee>, kept: Option<Invitee>) -> Standing {
let took = |m: Option<Invitee>| m.is_some_and(|m| m.joined_at.is_some() && m.joined);
let lapsed = org::lapsed(&m.invited_at, m.expires_at.as_deref());
let found = [
(took(open), Standing::Public),
(took(kept), Standing::Hidden),
(lapsed, Standing::Lapsed),
];
let found = found.into_iter().find(|(is, _)| *is);
found.map_or(Standing::Invited, |(_, standing)| standing)
}