boxmaker/crates/gatewayd/src/serve/handle.rs

//! What the event loop does with a post, a finished turn, and the time between: routing, starting
//! turns, typing, and catching up after a gap.

use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};

use crate::approvals::Judged;
use crate::deliver::{LOOP_DOWN, Notice, deliver};
use crate::mm::{Post, typing};
use crate::serve::approvals::Asker;
use crate::serve::{Gateway, Stop};
use crate::sessions::{BUSY, Batch, Ignored, Pushed, Route};
use crate::state::InFlight;
use crate::ws::WsError;
use crate::ws::conn::Ws;

/// Now, in Mattermost's milliseconds.
fn now_ms() -> i64 {
    // `let ms = SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_millis()).unwrap_or(0);`
    // then `i64::try_from(ms).unwrap_or(i64::MAX)`.
    let ms = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_millis())
        .unwrap_or(0);
    i64::try_from(ms).unwrap_or(i64::MAX)
}

impl Gateway {
    /// Post in a thread; a failure is logged, not fatal.
    pub(super) fn post(&self, channel: &str, root: &str, text: &str) {
        // `if let Err(e) = self.client.create_post(channel, root, text)`: log
        //    "gatewayd: cannot post in <channel> (thread <root>): <e>", with `(self.log)(&line)`.
        // Nothing else: posting records nothing in the state.
        if let Err(e) = self.client.create_post(channel, root, text) {
            (self.log)(&format!(
                "gatewayd: cannot post in {channel} (thread {root}): {e}"
            ));
        }
    }

    /// Is this channel one whose posts `gatewayd` keeps track of?
    fn tracked(&self, channel: &str, channel_type: &str) -> bool {
        // `channel_type == "D" || self.config.allow.channels.iter().any(|c| c == channel)`.
        channel_type == "D" || self.config.allow.channels.iter().any(|c| c == channel)
    }

    /// One new post, live or caught up.
    pub(super) fn handle_post(&mut self, post: &Post, channel_type: &str) -> Result<(), Stop> {
        // 1. `if !self.tracked(&post.channel_id, channel_type) || self.state.seen(&post.id)`:
        //    return Ok(()).
        // 2. `self.state.handled(&post.id, &post.channel_id, post.create_at)?;` before anything
        //    else: after a crash, a post is never answered twice.
        // 3. Route it, with the state answering "is this thread ours?":
        //    `let state = &self.state;`
        //    `let route = self.router.route(post, channel_type, &|root| state.knows_thread(root));`
        //    then `match route`:
        //    - Route::Ignore(Ignored::NotAllowed): log
        //      "gatewayd: ignored post <post.id> from <post.user_id>: not allowed"
        //      (never the text).
        //    - Route::Ignore(_): nothing.
        //    - Route::Reply { thread, text }: `self.post(&thread.channel, &thread.root, &text)`.
        //    - Route::Queue(message): if `message.joins_thread`,
        //      `self.state.join_thread(&message.thread.root)?`; then
        //      `match self.queues.push(message)`:
        //      Pushed::Start(batch) -> `self.start(batch)?`; Pushed::Waiting -> nothing;
        //      Pushed::Full(thread) -> `self.post(&thread.channel, &thread.root, BUSY)`.
        // 4. Ok(()).
        if !self.tracked(&post.channel_id, channel_type) || self.state.seen(&post.id) {
            return Ok(());
        }
        self.state
            .handled(&post.id, &post.channel_id, post.create_at)?;
        let state = &self.state;
        let route = self
            .router
            .route(post, channel_type, &|root| state.knows_thread(root));
        match route {
            Route::Ignore(Ignored::NotAllowed) => {
                (self.log)(&format!(
                    "gatewayd: ignored post {} from {}: not allowed",
                    post.id, post.user_id
                ));
            }
            Route::Ignore(_) => {}
            Route::Reply { thread, text } => self.post(&thread.channel, &thread.root, &text),
            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) => {
                self.authors.insert(
                    message.session.clone(),
                    (post.user_id.clone(), channel_type.to_string()),
                );
                if message.joins_thread {
                    self.state.join_thread(&message.thread.root)?;
                }
                match self.queues.push(message) {
                    Pushed::Start(batch) => self.start(batch)?,
                    Pushed::Waiting => {}
                    Pushed::Full(thread) => self.post(&thread.channel, &thread.root, BUSY),
                }
            }
        }
        Ok(())
    }

    /// Record the turn as in flight and run it on its own thread.
    fn start(&mut self, batch: Batch) -> Result<(), Stop> {
        // 1. `self.state.start_turn(InFlight { session: batch.session.as_str().to_string(),
        //    channel: batch.thread.channel.clone(), root: batch.thread.root.clone() })?;`
        // 2. Clone what the thread takes, before the `move`:
        //    `let (client, socket, log, done) = (Arc::clone(&self.client),
        //    self.loop_socket.clone(), Arc::clone(&self.log), self.done_tx.clone());`
        //    `let (session, thread) = (batch.session.clone(), batch.thread.clone());`
        // 3. `let spawned = std::thread::Builder::new().spawn(move || {
        //    deliver(client.as_ref(), &socket, &batch, &|line| log(line));
        //    let _ = done.send(batch.session); });`
        // 4. `if let Err(e) = spawned`: log "gatewayd: cannot start a thread for <session>: <e>"
        //    (`session.as_str()`), `self.post(&thread.channel, &thread.root, LOOP_DOWN)`, and
        //    `let _ = self.done_tx.send(session);` so the session does not stay busy.
        // 5. Ok(()).
        self.state.start_turn(InFlight {
            session: batch.session.as_str().to_string(),
            channel: batch.thread.channel.clone(),
            root: batch.thread.root.clone(),
        })?;
        if let Some((user, channel_type)) = self.authors.get(&batch.session) {
            let asker = Asker {
                user: user.clone(),
                channel_type: channel_type.clone(),
                prompt: batch.text.clone(),
            };
            self.askers
                .insert(batch.session.as_str().to_string(), asker);
        }
        let (registry, wait) = (Arc::clone(&self.registry), self.notice);
        let (client, socket, log, done) = (
            Arc::clone(&self.client),
            self.loop_socket.clone(),
            Arc::clone(&self.log),
            self.done_tx.clone(),
        );
        let (session, thread) = (batch.session.clone(), batch.thread.clone());
        let speaker = self.speaker(&batch);
        let spawned = std::thread::Builder::new().spawn(move || {
            let posted = |id: u64| registry.was_posted(id);
            let notice = Notice {
                posted: &posted,
                wait,
            };
            deliver(
                client.as_ref(),
                &socket,
                &batch,
                speaker.as_deref(),
                &|line| log(line),
                &notice,
            );
            let _ = done.send(batch.session);
        });
        if let Err(e) = spawned {
            (self.log)(&format!(
                "gatewayd: cannot start a thread for {}: {e}",
                session.as_str()
            ));
            self.post(&thread.channel, &thread.root, LOOP_DOWN);
            let _ = self.done_tx.send(session);
        }
        Ok(())
    }

    /// The turn's speaker: the username of the one person who wrote all of the batch, as
    /// Mattermost gives it (M5c spec, section 11). `None` when two or more people did, or when
    /// the name cannot be found: never the user id, never a guess. A name found is kept; a
    /// failure is logged and not kept, so the next turn looks again.
    fn speaker(&mut self, batch: &Batch) -> Option<String> {
        let user = batch.sender()?;
        if let Some(name) = self.usernames.get(user) {
            return Some(name.clone());
        }
        match self.client.username(user) {
            Ok(name) => {
                self.usernames.insert(user.to_string(), name.clone());
                Some(name)
            }
            Err(e) => {
                (self.log)(&format!(
                    "gatewayd: no username for {user}: {e}; the turn has no speaker"
                ));
                None
            }
        }
    }

    /// Turns that ended: out of flight, who asked forgotten, and the next batch of each session
    /// started. Never blocks.
    pub(super) fn finished(&mut self) -> Result<(), Stop> {
        while let Ok(session) = self.done_rx.try_recv() {
            self.state.end_turn(session.as_str())?;
            self.askers.remove(session.as_str());
            match self.queues.finish(&session) {
                Some(batch) => self.start(batch)?,
                None => {
                    self.authors.remove(&session);
                }
            }
        }
        Ok(())
    }

    /// Show this bot as typing in every thread with a turn running.
    pub(super) fn typing(&mut self, ws: &mut Ws) -> Result<(), WsError> {
        // `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))?;
        }
        Ok(())
    }

    /// Answer what arrived while `gatewayd` was away: the direct channel with each allowed user,
    /// and each allowed channel. Written for you: it is the glue.
    pub(super) fn catch_up(&mut self) -> Result<(), Stop> {
        let mut channels = Vec::new();
        for user in self.config.allow.users.clone() {
            match self.client.direct_channel(&self.me.id, &user) {
                Ok(id) => channels.push((id, "D")),
                Err(e) => (self.log)(&format!("gatewayd: no direct channel with {user}: {e}")),
            }
        }
        channels.extend(self.config.allow.channels.iter().map(|c| (c.clone(), "O")));
        for (channel, channel_type) in channels {
            self.catch_up_channel(&channel, channel_type)?;
        }
        Ok(())
    }

    /// One channel's posts since its mark. A channel seen for the first time starts from now.
    fn catch_up_channel(&mut self, channel: &str, channel_type: &str) -> Result<(), Stop> {
        // 1. `let Some(since) = self.state.since(channel) else { ... }`: with no mark,
        //    `self.state.mark(channel, now_ms())?` and return Ok(()).
        // 2. `self.client.posts_since(channel, since)`: an Err(e) is logged
        //    "gatewayd: cannot catch up <channel>: <e>" and returns Ok(()).
        // 3. When `found.full`, log "gatewayd: <channel>: too many posts to catch up; some may be
        //    missed".
        // 4. `for post in &found.posts { self.handle_post(post, channel_type)?; }` and Ok(()).
        let Some(since) = self.state.since(channel) else {
            self.state.mark(channel, now_ms())?;
            return Ok(());
        };
        let found = match self.client.posts_since(channel, since) {
            Ok(found) => found,
            Err(e) => {
                (self.log)(&format!("gatewayd: cannot catch up {channel}: {e}"));
                return Ok(());
            }
        };
        if found.full {
            (self.log)(&format!(
                "gatewayd: {channel}: too many posts to catch up; some may be missed"
            ));
        }
        for post in &found.posts {
            self.handle_post(post, channel_type)?;
        }
        Ok(())
    }
}