gatewayd: approvals answered from Mattermost; typing paused; the notice

Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)

crates/gatewayd/src/approvals/mod.rs

@@ -3,7 +3,7 @@
 
 pub mod text;
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::sync::mpsc::Sender;
 use std::sync::{Mutex, MutexGuard};
 
@@ -44,6 +44,8 @@ pub enum Judged {
 #[derive(Debug, Default)]
 pub struct Registry {
     posted: Mutex<HashMap<u64, Posted>>,
+    /// Every approval ever posted here, answered or not: for the notice.
+    ever: Mutex<HashSet<u64>>,
 }
 
 impl Registry {
@@ -56,6 +58,10 @@ impl Registry {
     }
 
     pub fn add(&self, posted: Posted) {
+        self.ever
+            .lock()
+            .unwrap_or_else(|p| p.into_inner())
+            .insert(posted.approval);
         self.lock().insert(posted.approval, posted);
     }
 
@@ -73,6 +79,14 @@ impl Registry {
         self.lock().values().map(|p| p.session.clone()).collect()
     }
 
+    /// Was approval `id` ever posted here?
+    pub fn was_posted(&self, id: u64) -> bool {
+        self.ever
+            .lock()
+            .unwrap_or_else(|p| p.into_inner())
+            .contains(&id)
+    }
+
     /// 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(

crates/gatewayd/src/deliver.rs

@@ -3,6 +3,7 @@
 
 use std::os::unix::net::UnixStream;
 use std::path::Path;
+use std::time::{Duration, Instant};
 
 use proto::{
     Envelope, ErrorCode, Message, PROTOCOL_VERSION, Turn, TurnEvent, WireError, read_frame,
@@ -40,10 +41,36 @@ pub enum Outcome {
 
 pub fn approval_text(approval: u64) -> String {
     format!(
-        "waiting for approval {approval}: approve or deny it with `bxctl` (Mattermost approvals arrive in M4b)"
+        "waiting for approval {approval}: brokerd has not sent it here; answer it with bxctl approvals"
     )
 }
 
+/// How long a turn waits for `brokerd` to bring its approval here before posting the notice.
+pub const NOTICE_WAIT: Duration = Duration::from_secs(10);
+
+/// Whether an approval was brought here (and so is posted already), and how long to wait for it.
+pub struct Notice<'a> {
+    pub posted: &'a dyn Fn(u64) -> bool,
+    pub wait: Duration,
+}
+
+impl Notice<'_> {
+    fn arrives(&self, id: u64) -> bool {
+        let until = Instant::now() + self.wait;
+        loop {
+            if (self.posted)(id) {
+                return true;
+            }
+            if Instant::now() >= until {
+                return false;
+            }
+            std::thread::sleep(
+                Duration::from_millis(50).min(until.saturating_duration_since(Instant::now())),
+            );
+        }
+    }
+}
+
 /// "Error: <code>: <detail>"; the detail carries `loopd`'s runbook pointer when there is one.
 pub fn error_text(error: &WireError) -> String {
     // "Error: <code>: <detail>", where <code> is the snake_case name serde gives the ErrorCode
@@ -173,14 +200,23 @@ fn post(poster: &dyn Poster, thread: &Thread, text: String, log: &dyn Fn(&str))
 }
 
 /// Run a batch's turn and post what comes of it in its thread. A post that fails is logged.
-pub fn deliver(poster: &dyn Poster, socket: &Path, batch: &Batch, log: &dyn Fn(&str)) {
-    // Post in the batch thread. An ApprovalPending event posts `approval_text(approval)` at once.
-    // Then: Answer -> every part of `split_answer`, in order; Refused -> `error_text`;
+pub fn deliver(
+    poster: &dyn Poster,
+    socket: &Path,
+    batch: &Batch,
+    log: &dyn Fn(&str),
+    notice: &Notice,
+) {
+    // Post in the batch thread. An ApprovalPending event waits up to `notice.wait` for `brokerd` to
+    // bring the approval through `gateway.sock`; only when it does not, it posts the notice. Then:
+    // Answer -> every part of `split_answer`, in order; Refused -> `error_text`;
     // LoopDown(why) -> log "gatewayd: <session>: <why>" and post LOOP_DOWN. A post that fails is
     // logged, exactly "gatewayd: cannot post in <channel> (thread <root>): <error>", and the rest
     // goes on.
     let outcome = run_turn(socket, batch, &mut |event| {
-        if let TurnEvent::ApprovalPending { approval, .. } = event {
+        if let TurnEvent::ApprovalPending { approval, .. } = event
+            && !notice.arrives(*approval)
+        {
             post(poster, &batch.thread, approval_text(*approval), log);
         }
     });

crates/gatewayd/src/serve/approvals.rs

@@ -141,7 +141,7 @@ impl Gateway {
             Ending::NoLonger => Ended::NoLonger,
             Ending::Answered {
                 by,
-                reason: _,
+                reason: refusal,
                 outcome,
             } => {
                 let by = match self.client.username(&by) {
@@ -151,7 +151,10 @@ impl Gateway {
                 match outcome {
                     GatewayOutcome::Runs {} => Ended::Runs { by },
                     GatewayOutcome::Denied { reason } => Ended::Denied { by, reason },
-                    GatewayOutcome::Refused {} => Ended::Refused { by, reason: None },
+                    GatewayOutcome::Refused {} => Ended::Refused {
+                        by,
+                        reason: refusal,
+                    },
                     GatewayOutcome::Gone {} => Ended::NoLonger,
                 }
             }

crates/gatewayd/src/serve/handle.rs

@@ -4,10 +4,11 @@
 use std::sync::Arc;
 use std::time::{SystemTime, UNIX_EPOCH};
 
-use crate::deliver::{LOOP_DOWN, deliver};
+use crate::approvals::Judged;
+use crate::deliver::{LOOP_DOWN, Notice, deliver};
 use crate::mm::{Post, typing};
 use crate::serve::{Gateway, Stop};
-use crate::sessions::{BUSY, Batch, Ignored, M4B_COMMAND, Pushed, Route};
+use crate::sessions::{BUSY, Batch, Ignored, Pushed, Route};
 use crate::state::InFlight;
 use crate::ws::WsError;
 use crate::ws::conn::Ws;
@@ -81,8 +82,26 @@ impl Gateway {
             }
             Route::Ignore(_) => {}
             Route::Reply { thread, text } => self.post(&thread.channel, &thread.root, &text),
-            // Until the approvals module is wired in (task 15), M4a's notice.
-            Route::Approval { thread, .. } => self.post(&thread.channel, &thread.root, M4B_COMMAND),
+            Route::Approval {
+                thread,
+                command,
+                user,
+                post: command_post,
+                at,
+            } => {
+                let approvers = self.config.approvers();
+                let judged = self.registry.judge(
+                    &command,
+                    &user,
+                    &thread.root,
+                    &command_post,
+                    at,
+                    approvers,
+                );
+                if let Judged::Reply(text) = judged {
+                    self.post(&thread.channel, &thread.root, &text);
+                }
+            }
             Route::Queue(message) => {
                 if message.joins_thread {
                     self.state.join_thread(&message.thread.root)?;
@@ -117,6 +136,7 @@ impl Gateway {
             channel: batch.thread.channel.clone(),
             root: batch.thread.root.clone(),
         })?;
+        let (registry, wait) = (Arc::clone(&self.registry), self.notice);
         let (client, socket, log, done) = (
             Arc::clone(&self.client),
             self.loop_socket.clone(),
@@ -125,7 +145,12 @@ impl Gateway {
         );
         let (session, thread) = (batch.session.clone(), batch.thread.clone());
         let spawned = std::thread::Builder::new().spawn(move || {
-            deliver(client.as_ref(), &socket, &batch, &|line| log(line));
+            let posted = |id: u64| registry.was_posted(id);
+            let notice = Notice {
+                posted: &posted,
+                wait,
+            };
+            deliver(client.as_ref(), &socket, &batch, &|line| log(line), &notice);
             let _ = done.send(batch.session);
         });
         if let Err(e) = spawned {
@@ -156,9 +181,14 @@ impl Gateway {
 
     /// Show this bot as typing in every thread with a turn running.
     pub(super) fn typing(&mut self, ws: &mut Ws) -> Result<(), WsError> {
-        // `for thread in self.queues.threads() { self.seq += 1;
-        //    ws.send_text(&typing(self.seq, &thread.channel, &thread.root))?; }` then Ok(()).
+        // `let waiting = self.registry.waiting_sessions();` then for each thread, none that waits
+        // for its approval: A turn waiting for its approval is not working: no typing in its thread.
+        // Otherwise `self.seq += 1; ws.send_text(...)`; then Ok(()).
+        let waiting = self.registry.waiting_sessions();
         for thread in self.queues.threads() {
+            if waiting.contains(&format!("mm-{}", thread.root)) {
+                continue;
+            }
             self.seq += 1;
             ws.send_text(&typing(self.seq, &thread.channel, &thread.root))?;
         }

crates/gatewayd/src/serve/mod.rs

@@ -77,6 +77,8 @@ pub struct Tuning {
     pub poll: Duration,
     /// Connecting and each REST read.
     pub rest_timeout: Duration,
+    /// How long a turn waits for its approval to come through `gateway.sock` before the notice.
+    pub notice: Duration,
 }
 
 impl Default for Tuning {
@@ -86,6 +88,7 @@ impl Default for Tuning {
             backoff: secs.to_vec(),
             poll: Duration::from_millis(200),
             rest_timeout: Duration::from_secs(30),
+            notice: crate::deliver::NOTICE_WAIT,
         }
     }
 }
@@ -112,6 +115,8 @@ pub(crate) struct Gateway {
     gateway_rx: Receiver<ToLoop>,
     /// The last post of each approval posted here and not yet ended.
     open: HashMap<u64, approvals::Open>,
+    /// How long a turn waits for its approval to come through `gateway.sock` before the notice.
+    notice: Duration,
 }
 
 /// The wait before attempt `n` (from 0) after a loss.
@@ -167,7 +172,7 @@ pub fn run(config: Config, token: Secret, tuning: Tuning, log: Log, stop: &Atomi
         return Stop::Start(format!("cannot start the gateway.sock thread: {e}"));
     }
     let client = Arc::new(Client::new(connector, token, tuning.rest_timeout));
-    let mut g = Gateway::new(config, client, state, log, gateway_rx);
+    let mut g = Gateway::new(config, client, state, log, gateway_rx, tuning.notice);
     let mut failures = 0;
     loop {
         if stop.load(Ordering::SeqCst) {
@@ -235,6 +240,7 @@ impl Gateway {
         state: State,
         log: Log,
         gateway_rx: Receiver<ToLoop>,
+        notice: Duration,
     ) -> Gateway {
         let (done_tx, done_rx) = mpsc::channel();
         let limit = usize::try_from(config.limits.queue).unwrap_or(usize::MAX);
@@ -258,6 +264,7 @@ impl Gateway {
             registry: Arc::new(Registry::new()),
             gateway_rx,
             open: HashMap::new(),
+            notice,
         }
     }
 

crates/gatewayd/src/sessions.rs

@@ -7,7 +7,6 @@ use proto::SessionId;
 
 use crate::mm::Post;
 
-pub const M4B_COMMAND: &str = "approvals over Mattermost arrive in M4b; use `bxctl approvals`";
 pub const UNKNOWN_COMMAND: &str = "unknown command; the commands are !approve and !deny";
 pub const BUSY: &str = "busy: too many messages are waiting in this conversation";