gatewayd: approvals, the registry and the judging of commands

Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)
2026-09-24 04:48:06 -07:00
parent d9f8fa9331
commit c5df86d13d
3 changed files with 300 additions and 0 deletions
@@ -2,3 +2,136 @@
 //! approvals this `gatewayd` has posted and the commands that answer them.

 pub mod text;
+
+use std::collections::HashMap;
+use std::sync::mpsc::Sender;
+use std::sync::{Mutex, MutexGuard};
+
+use proto::ApprovalReply;
+
+use crate::sessions::ApprovalCommand::{self, Approve, Deny, Malformed};
+
+pub const NOT_APPROVER: &str = "only an approver can answer approvals";
+pub const USAGE: &str = "use !approve <id> or !deny <id> <reason>";
+
+/// One approval posted here and not yet answered.
+#[derive(Debug)]
+pub struct Posted {
+    pub approval: u64,
+    /// The session whose turn waits for it.
+    pub session: String,
+    /// The thread it is answered in.
+    pub channel: String,
+    pub root: String,
+    /// The last post's `create_at`: only a command after it counts.
+    pub shown_at: i64,
+    /// False when the arguments were too long to show: only `bxctl` answers it.
+    pub answerable: bool,
+    /// Where an accepted answer goes: the thread of its `gateway.sock` connection.
+    pub answer: Sender<ApprovalReply>,
+}
+
+/// What a command came to.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Judged {
+    /// The answer went to `brokerd`'s connection; the approval post will show the outcome.
+    Sent,
+    /// Refused: this line goes in the thread.
+    Reply(String),
+}
+
+/// The approvals posted here, by id.
+#[derive(Debug, Default)]
+pub struct Registry {
+    posted: Mutex<HashMap<u64, Posted>>,
+}
+
+impl Registry {
+    pub fn new() -> Registry {
+        Registry::default()
+    }
+
+    fn lock(&self) -> MutexGuard<'_, HashMap<u64, Posted>> {
+        self.posted.lock().unwrap_or_else(|p| p.into_inner())
+    }
+
+    pub fn add(&self, posted: Posted) {
+        self.lock().insert(posted.approval, posted);
+    }
+
+    /// Take it out: it ended, or its connection is done with it.
+    pub fn remove(&self, id: u64) -> Option<Posted> {
+        self.lock().remove(&id)
+    }
+
+    pub fn holds(&self, id: u64) -> bool {
+        self.lock().contains_key(&id)
+    }
+
+    /// The sessions with an approval waiting here: their turns are waiting, not working.
+    pub fn waiting_sessions(&self) -> Vec<String> {
+        self.lock().values().map(|p| p.session.clone()).collect()
+    }
+
+    /// Judge a command from `user`, posted as `post` at `at` in the thread `root`, in the order of
+    /// the spec's section 7. An accepted answer is taken out and sent; nothing else changes.
+    pub fn judge(
+        &self,
+        command: &ApprovalCommand,
+        user: &str,
+        root: &str,
+        post: &str,
+        at: i64,
+        approvers: &[String],
+    ) -> Judged {
+        // In this order; each refusal is `Judged::Reply(<text>)` and changes nothing:
+        // 1. `user` not in `approvers`: NOT_APPROVER.
+        // 2. The command: Approve(id) is (id, approved true, reason None); Deny(id, reason) is (id,
+        //    false, reason.clone()); Malformed: USAGE.
+        // 3. `let mut posted = self.lock();` Not in `posted`: "approval {id} is not pending here".
+        // 4. `!entry.answerable`: "approval {id} can only be answered with bxctl approvals".
+        // 5. `entry.root != root`: "approval {id} is answered in its own thread".
+        // 6. `at <= entry.shown_at`: "approval {id} was answered before it was shown in full".
+        // 7. Take it out (`posted.remove(&id)`) and send `ApprovalReply { approval: id, approved,
+        //    by: user, post, reason }` on its `answer`: Ok -> `Judged::Sent`; Err (the connection
+        //    is gone) -> "approval {id} is not pending here".
+        if !approvers.iter().any(|a| a.as_str() == user) {
+            return Judged::Reply(NOT_APPROVER.to_string());
+        }
+        let (id, approved, reason) = match command {
+            Approve(id) => (*id, true, None),
+            Deny(id, reason) => (*id, false, reason.clone()),
+            Malformed => return Judged::Reply(USAGE.to_string()),
+        };
+        let mut posted = self.lock();
+        let entry = match posted.get(&id) {
+            Some(entry) => entry,
+            None => return Judged::Reply(format!("approval {id} is not pending here")),
+        };
+        let answer = entry.answer.clone();
+        if !entry.answerable {
+            return Judged::Reply(format!(
+                "approval {id} can only be answered with bxctl approvals"
+            ));
+        }
+        if entry.root != root {
+            return Judged::Reply(format!("approval {id} is answered in its own thread"));
+        }
+        if at <= entry.shown_at {
+            return Judged::Reply(format!(
+                "approval {id} was answered before it was shown in full"
+            ));
+        }
+        posted.remove(&id);
+        match answer.send(ApprovalReply {
+            approval: id,
+            approved,
+            by: user.to_string(),
+            post: post.to_string(),
+            reason,
+        }) {
+            Ok(()) => Judged::Sent,
+            Err(_) => Judged::Reply(format!("approval {id} is not pending here")),
+        }
+    }
+}
@@ -0,0 +1,166 @@
+//! Which commands answer an approval, in the order of the M4b spec, section 7; everything else is
+//! answered with its reason and changes nothing. Do not edit.
+
+use std::sync::mpsc::{self, Receiver};
+
+use gatewayd::approvals::{Judged, NOT_APPROVER, Posted, Registry, USAGE};
+use gatewayd::sessions::ApprovalCommand::{self, Approve, Deny, Malformed};
+use proto::ApprovalReply;
+
+const KYLE: &str = "k0000000000000000000000000";
+const HERMES: &str = "h0000000000000000000000000";
+const DM: &str = "d0000000000000000000000000";
+const ROOT: &str = "r0000000000000000000000000";
+const ELSEWHERE: &str = "e0000000000000000000000000";
+const CMD: &str = "c0000000000000000000000001";
+
+fn approvers() -> Vec<String> {
+    vec![KYLE.to_string()]
+}
+
+/// A registry with approval 42 posted in ROOT, shown at 100.
+fn posted(answerable: bool) -> (Registry, Receiver<ApprovalReply>) {
+    let registry = Registry::new();
+    let (tx, rx) = mpsc::channel();
+    registry.add(Posted {
+        approval: 42,
+        session: format!("mm-{ROOT}"),
+        channel: DM.to_string(),
+        root: ROOT.to_string(),
+        shown_at: 100,
+        answerable,
+        answer: tx,
+    });
+    (registry, rx)
+}
+
+fn judge(r: &Registry, command: &ApprovalCommand, user: &str, root: &str, at: i64) -> Judged {
+    r.judge(command, user, root, CMD, at, &approvers())
+}
+
+#[test]
+fn an_approver_in_the_thread_after_the_post_answers_it() {
+    let (r, rx) = posted(true);
+    assert_eq!(judge(&r, &Approve(42), KYLE, ROOT, 101), Judged::Sent);
+    assert_eq!(
+        rx.try_recv().unwrap(),
+        ApprovalReply {
+            approval: 42,
+            approved: true,
+            by: KYLE.to_string(),
+            post: CMD.to_string(),
+            reason: None
+        }
+    );
+    assert!(!r.holds(42), "answered once");
+    assert_eq!(
+        judge(&r, &Approve(42), KYLE, ROOT, 102),
+        Judged::Reply("approval 42 is not pending here".to_string())
+    );
+    assert!(rx.try_recv().is_err(), "a second answer goes nowhere");
+}
+
+#[test]
+fn a_refusal_carries_its_reason() {
+    let (r, rx) = posted(true);
+    assert_eq!(
+        judge(
+            &r,
+            &Deny(42, Some("not on a Friday".to_string())),
+            KYLE,
+            ROOT,
+            101
+        ),
+        Judged::Sent
+    );
+    let got = rx.try_recv().unwrap();
+    assert_eq!(
+        (got.approved, got.reason.as_deref()),
+        (false, Some("not on a Friday"))
+    );
+    let (r, rx) = posted(true);
+    assert_eq!(judge(&r, &Deny(42, None), KYLE, ROOT, 101), Judged::Sent);
+    assert_eq!(rx.try_recv().unwrap().reason, None);
+}
+
+#[test]
+fn everything_else_is_refused_with_its_reason_in_order() {
+    let cases: &[(&ApprovalCommand, &str, &str, i64, &str)] = &[
+        (&Approve(42), HERMES, ROOT, 101, NOT_APPROVER),
+        (&Malformed, HERMES, ROOT, 101, NOT_APPROVER),
+        (&Malformed, KYLE, ROOT, 101, USAGE),
+        (
+            &Approve(7),
+            KYLE,
+            ROOT,
+            101,
+            "approval 7 is not pending here",
+        ),
+        (
+            &Approve(42),
+            KYLE,
+            ELSEWHERE,
+            101,
+            "approval 42 is answered in its own thread",
+        ),
+        (
+            &Approve(42),
+            KYLE,
+            ROOT,
+            100,
+            "approval 42 was answered before it was shown in full",
+        ),
+        (
+            &Approve(42),
+            KYLE,
+            ROOT,
+            50,
+            "approval 42 was answered before it was shown in full",
+        ),
+        (
+            &Approve(42),
+            KYLE,
+            ELSEWHERE,
+            50,
+            "approval 42 is answered in its own thread",
+        ),
+    ];
+    for (command, user, root, at, want) in cases {
+        let (r, rx) = posted(true);
+        assert_eq!(
+            judge(&r, command, user, root, *at),
+            Judged::Reply(want.to_string()),
+            "{command:?} {user} {root} {at}"
+        );
+        assert!(r.holds(42), "still pending after {command:?}");
+        assert!(rx.try_recv().is_err(), "nothing sent for {command:?}");
+    }
+}
+
+#[test]
+fn too_long_to_show_is_answered_only_in_bxctl() {
+    let (r, rx) = posted(false);
+    assert_eq!(
+        judge(&r, &Approve(42), KYLE, ROOT, 101),
+        Judged::Reply("approval 42 can only be answered with bxctl approvals".to_string())
+    );
+    assert!(r.holds(42) && rx.try_recv().is_err());
+}
+
+#[test]
+fn an_answer_whose_connection_is_gone_is_not_pending() {
+    let (r, rx) = posted(true);
+    drop(rx);
+    assert_eq!(
+        judge(&r, &Approve(42), KYLE, ROOT, 101),
+        Judged::Reply("approval 42 is not pending here".to_string())
+    );
+}
+
+#[test]
+fn the_waiting_sessions_and_removal() {
+    let (r, _rx) = posted(true);
+    assert_eq!(r.waiting_sessions(), [format!("mm-{ROOT}")]);
+    assert_eq!(r.remove(42).map(|p| p.approval), Some(42));
+    assert!(r.waiting_sessions().is_empty() && !r.holds(42) && r.remove(42).is_none());
+}