Invites and suggestions are cases; the responsible follow the content

Five things the first cut of accounts-from-content left open.

The first person and the responsible people are set up side by side,
and both try again a minute apart: a seed that waited twenty minutes on
Kanidm used to hold everyone else up, and a responsible person who
failed waited for the next restart.

The responsible are gone over again whenever the content changes.
watch_for_reload publishes portal.content.reloaded once new content is
in place, and seed.rs listens.

Being responsible ends when the content stops saying so: someone no
page names any more leaves RESPONSIBLE_GROUP in the same pass, if
portal is who put them there (Invited.already_member). The account
stays.

An invite keeps itself current. portal_invites has an expired state;
signing in closes a person's invites, a sweep moves the ones whose link
ran out, and a desk's move from expired back to open makes a fresh link
and mails it. Invites made at start are on file with the desks'.

A suggested edit is a case in the built-in portal_edit_suggestions
bucket, the person who made it is mailed at each move, and the card
lists what they sent and where it stands.

Found on the way, against Kanidm 1.11.2: a link's end comes as seconds
since the epoch, which portal did not read, so invite mails said
"until" and nothing; and a mail search answered with a 5xx was taken
for a refusal.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GT37Z1Xtfd9pUuQtMTg6Yt

src/seed.rs

@@ -16,20 +16,30 @@
 //! Once, not on every start: a done seed is remembered in the
 //! `portal_seed` bucket, keyed by address and group, so a restart or an
 //! upgrade never mails a second link. Changing `SEED_ADMIN_EMAIL` or
-//! `SEED_GROUP` seeds the new pair. Kanidm and gdo may still be coming
-//! up alongside portal, so a failed attempt is retried for a while.
+//! `SEED_GROUP` seeds the new pair.
 //!
 //! **Everyone the content names as responsible.** A page's
 //! `responsible: { name, contact }` is a person answering for that
-//! question, so they get an account: at every start, each responsible
-//! address Kanidm does not already hold is given one, and mailed that
-//! they are listed as responsible for asking those questions, with the
-//! link to sign in. Signed in, their own pages offer them a way to
-//! suggest changes. `RESPONSIBLE_GROUP` names a group they all join -
-//! on the vel, `redaktor`, which is also who may sign in to the project
-//! Gitea - and people who already had an account are added to it
-//! without a mail. A desk is still a desk invite. Each address is
-//! onboarded once per group (`portal_seed`).
+//! question, so they get an account: at start, and again whenever the
+//! content changes, each responsible address Kanidm does not already
+//! hold is given one, and mailed that they are listed as responsible
+//! for asking those questions, with the link to sign in. Signed in,
+//! their own pages offer them a way to suggest changes.
+//! `RESPONSIBLE_GROUP` names a group they all join - on the vel,
+//! `redaktor`, which is also who may sign in to the project Gitea - and
+//! people who already had an account are added to it without a mail.
+//! A desk is still a desk invite. Each address is onboarded once per
+//! group (`portal_seed`).
+//!
+//! **And no longer.** Someone the content has stopped naming leaves
+//! `RESPONSIBLE_GROUP` in the same pass, if portal is who put them
+//! there. Their account is theirs and stays.
+//!
+//! Both run side by side and try again a minute apart, twenty times:
+//! Kanidm and gdo may still be coming up alongside portal. Every
+//! invite made here is put on file in `portal_invites` like a desk's,
+//! so a link that runs out unused shows up where someone can send it
+//! again.
 #![cfg(feature = "ssr")]
 
 use std::collections::{BTreeMap, BTreeSet, HashMap};
@@ -92,10 +102,58 @@ pub fn contact_email(contact: &str) -> Option<String> {
     (c.contains('@') && !c.contains(char::is_whitespace)).then_some(c)
 }
 
-/// The first person in, then everyone the content names as responsible.
+/// A failed attempt is tried again this many times, this far apart:
+/// Kanidm and gdo may still be coming up beside portal.
+const ATTEMPTS: u32 = 20;
+const APART: std::time::Duration = std::time::Duration::from_secs(60);
+
+/// One pass over the responsible people at a time: a reload may arrive
+/// while the pass from the start is still at it, and two at once would
+/// both find the same person missing and invite them twice.
+static PASS: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
+
+/// The first person in and everyone the content names as responsible,
+/// side by side, so neither waits on the other's retries - and the
+/// responsible again whenever the content changes, for the life of the
+/// process.
 pub async fn run(state: crate::server::AppState) {
-    seed_admin(&state).await;
-    responsible(&state).await;
+    // Subscribed before the first pass, so content that changes while
+    // it runs is not missed.
+    let reloads = state.nats.subscribe(crate::content::CONTENT_RELOADED_SUBJECT).await;
+    tokio::join!(seed_admin(&state), responsible_until_done(&state), async {
+        let mut reloads = match reloads {
+            Ok(reloads) => reloads,
+            Err(e) => {
+                tracing::error!("could not listen for changed content: {e} - a new responsible person waits for the next start");
+                return;
+            }
+        };
+        use futures::StreamExt;
+        while reloads.next().await.is_some() {
+            responsible_until_done(&state).await;
+        }
+    });
+}
+
+/// Passes over the responsible people until one leaves nobody behind.
+/// Each pass reads the content afresh, so what it retries is what the
+/// content says by then.
+async fn responsible_until_done(state: &crate::server::AppState) {
+    for attempt in 1..=ATTEMPTS {
+        let left = {
+            let _one_at_a_time = PASS.lock().await;
+            responsible(state).await
+        };
+        if left == 0 {
+            return;
+        }
+        if attempt == ATTEMPTS {
+            tracing::error!(left, "some responsible people could not be set up, for good - until the content changes or portal starts again");
+            return;
+        }
+        tracing::warn!(left, attempt, "some responsible people could not be set up, trying again in a minute");
+        tokio::time::sleep(APART).await;
+    }
 }
 
 async fn seed_bucket(state: &crate::server::AppState) -> Option<async_nats::jetstream::kv::Store> {
@@ -111,35 +169,64 @@ async fn seed_bucket(state: &crate::server::AppState) -> Option<async_nats::jets
     }
 }
 
-async fn responsible(state: &crate::server::AppState) {
+fn seed_admin_email() -> Option<String> {
+    std::env::var("SEED_ADMIN_EMAIL").ok().map(|e| e.trim().to_lowercase()).filter(|e| e.contains('@'))
+}
+
+/// Who portal put in the responsible group and the content no longer
+/// names: their key in the seed bucket, and the account. The first
+/// person is never among them - with `SEED_GROUP` the same group, their
+/// place there is from the seed, not from a page.
+pub fn no_longer_responsible(
+    on_file: &[(String, crate::people::Invited)],
+    people: &BTreeMap<String, (String, Vec<String>)>,
+    seed_admin: Option<&str>,
+) -> Vec<(String, crate::people::Invited)> {
+    on_file
+        .iter()
+        .filter(|(_, invited)| {
+            let email = invited.email.trim().to_lowercase();
+            !people.contains_key(&email) && seed_admin != Some(email.as_str())
+        })
+        .cloned()
+        .collect()
+}
+
+/// One pass: an account for everyone the content names, and out of the
+/// group for everyone it has stopped naming. Returns how many people
+/// it could not finish with.
+async fn responsible(state: &crate::server::AppState) -> usize {
     let people = responsible_people(&state.questions.load());
-    if people.is_empty() {
-        return;
+    let group = std::env::var("RESPONSIBLE_GROUP").ok().map(|g| g.trim().to_string()).filter(|g| !g.is_empty());
+    if people.is_empty() && group.is_none() {
+        return 0;
     }
     let Some(kanidm) = crate::people::Kanidm::from_env() else {
-        tracing::info!(people = people.len(), "the content names responsible people, but invites are not configured (KANIDM_API_TOKEN) - no accounts made");
-        return;
+        if !people.is_empty() {
+            tracing::info!(people = people.len(), "the content names responsible people, but invites are not configured (KANIDM_API_TOKEN) - no accounts made");
+        }
+        return 0;
     };
-    let Some(store) = seed_bucket(state).await else { return };
-    let group = std::env::var("RESPONSIBLE_GROUP").ok().map(|g| g.trim().to_string()).filter(|g| !g.is_empty());
-    for (email, (name, questions)) in people {
-        let key = seed_key(&email, group.as_deref().unwrap_or("responsible"));
+    let Some(store) = seed_bucket(state).await else { return 0 };
+    let mut left = 0;
+    for (email, (name, questions)) in &people {
+        let key = seed_key(email, group.as_deref().unwrap_or("responsible"));
         if let Ok(Some(_)) = store.get(&key).await {
             continue;
         }
-        let name = if name.is_empty() { email.split('@').next().unwrap_or("").to_string() } else { name };
+        let name = if name.is_empty() { email.split('@').next().unwrap_or("").to_string() } else { name.clone() };
         // With a group, one call makes the account if there is none and
         // adds them either way; without one, only a missing account is
         // made. Either way a person is mailed only when an account was
         // made for them: someone who had one already has their way in.
         let made = match &group {
-            Some(group) => kanidm.invite(&name, &email, group).await,
-            None => match kanidm.has_person_with_mail(&email).await {
+            Some(group) => kanidm.invite(&name, email, group).await,
+            None => match kanidm.has_person_with_mail(email).await {
                 Ok(true) => {
                     let _ = store.put(&key, "existing".into()).await;
                     continue;
                 }
-                Ok(false) => kanidm.account(&name, &email).await,
+                Ok(false) => kanidm.account(&name, email).await,
                 Err(e) => Err(e),
             },
         };
@@ -150,13 +237,50 @@ async fn responsible(state: &crate::server::AppState) {
                     let _ = store.put(&key, bytes.into()).await;
                 }
                 if invited.created {
-                    crate::mail::on_responsible(state, &invited, &questions).await;
+                    crate::people::record(state, &invited, questions).await;
+                    crate::mail::on_responsible(state, &invited, questions).await;
                 }
                 tracing::info!(user = %invited.username, created = invited.created, group = group.as_deref().unwrap_or(""), "a person the content names as responsible has their account");
             }
-            Err(e) => tracing::warn!(%email, "could not set up a responsible person, leaving them for the next start: {e}"),
+            Err(e) => {
+                left += 1;
+                tracing::warn!(%email, "could not set up a responsible person: {e}");
+            }
         }
     }
+
+    // Being responsible ends when the content stops saying so. The
+    // account stays - it is theirs - but the place in the group, and
+    // whatever else that group opens, goes with the page. Only what
+    // portal gave: someone who was in the group before is left there.
+    let Some(group) = group else { return left };
+    let prefix = seed_key("", &group);
+    let keys: Vec<String> = {
+        use futures::TryStreamExt;
+        match store.keys().await {
+            Ok(keys) => keys.try_collect().await.unwrap_or_default(),
+            Err(_) => Vec::new(),
+        }
+    };
+    let mut on_file = Vec::new();
+    for key in keys.into_iter().filter(|k| k.starts_with(&prefix)) {
+        let Ok(Some(bytes)) = store.get(&key).await else { continue };
+        let Ok(invited) = serde_json::from_slice::<crate::people::Invited>(&bytes) else { continue };
+        on_file.push((key, invited));
+    }
+    for (key, invited) in no_longer_responsible(&on_file, &people, seed_admin_email().as_deref()) {
+        if !invited.already_member {
+            if let Err(e) = kanidm.remove_member(&group, &invited.username).await {
+                left += 1;
+                tracing::warn!(user = %invited.username, group, "could not take someone no longer responsible out of the group: {e}");
+                continue;
+            }
+        }
+        // Forgotten, so that being named again is being set up again.
+        let _ = store.purge(&key).await;
+        tracing::info!(user = %invited.username, group, removed = !invited.already_member, "the content no longer names this person as responsible");
+    }
+    left
 }
 
 /// The first person in, once, if `SEED_ADMIN_EMAIL` asks for one.
@@ -190,15 +314,14 @@ async fn seed_admin(state: &crate::server::AppState) {
         return;
     };
 
-    // Kanidm may be starting beside us: a minute apart, for twenty.
     let mut attempt = 0;
     let invited = loop {
         attempt += 1;
         match kanidm.invite(&name, &email, &group).await {
             Ok(invited) => break invited,
-            Err(e) if attempt < 20 => {
+            Err(e) if attempt < ATTEMPTS => {
                 tracing::warn!(group, attempt, "seeding the first person failed, trying again in a minute: {e}");
-                tokio::time::sleep(std::time::Duration::from_secs(60)).await;
+                tokio::time::sleep(APART).await;
             }
             Err(e) => {
                 tracing::error!(group, "seeding the first person failed for good: {e}");
@@ -212,6 +335,9 @@ async fn seed_admin(state: &crate::server::AppState) {
             tracing::error!(group, "the first person is invited, but it could not be remembered: {e} - a restart would invite them again");
         }
     }
+    if invited.created {
+        crate::people::record(state, &invited, &[]).await;
+    }
     crate::mail::on_invite(state, &invited).await;
     tracing::info!(group, user = %invited.username, created = invited.created, "the first person is invited");
 }
@@ -250,6 +376,33 @@ mod tests {
         assert!(most_privileged_group(&none).is_none());
     }
 
+    fn on_file(email: &str, already_member: bool) -> (String, crate::people::Invited) {
+        (
+            seed_key(email, "redaktor"),
+            crate::people::Invited {
+                username: email.split('@').next().unwrap().to_string(),
+                display_name: String::new(),
+                email: email.to_string(),
+                group: "redaktor".to_string(),
+                reset_link: None,
+                link_expires_ms: None,
+                created: true,
+                already_member,
+            },
+        )
+    }
+
+    #[test]
+    fn whoever_the_content_stopped_naming_leaves_but_never_the_first_person() {
+        let people: BTreeMap<String, (String, Vec<String>)> = [("kari@vel.no".to_string(), ("Kari".to_string(), vec!["Hva?".to_string()]))].into_iter().collect();
+        let filed = vec![on_file("kari@vel.no", false), on_file("Ola@vel.no", false), on_file("leder@vel.no", false)];
+        let gone = no_longer_responsible(&filed, &people, Some("leder@vel.no"));
+        assert_eq!(gone.iter().map(|(_, i)| i.username.as_str()).collect::<Vec<_>>(), vec!["Ola"]);
+        assert_eq!(gone[0].0, "redaktor.ola_vel_no");
+        // Nobody named at all: everyone portal put there leaves.
+        assert_eq!(no_longer_responsible(&filed, &BTreeMap::new(), None).len(), 3);
+    }
+
     #[test]
     fn everyone_named_responsible_is_found_once_by_their_address() {
         let mut qs: HashMap<String, Question> = HashMap::new();