boxmaker/crates/gatewayd/src/sessions.rs

//! Which posts `gatewayd` acts on, which session each belongs to, commands, and the queue of
//! messages per session (M4a spec, section 7). Pure: the state file and the network are elsewhere.

use std::collections::{BTreeSet, HashMap};

use proto::SessionId;

use crate::mm::Post;

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";

/// Names that name nobody: every agent in the channel would answer them.
const EVERYONE: [&str; 3] = ["channel", "here", "all"];

/// Why a post was not acted on. Only `NotAllowed` is logged, by user id and post id.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Ignored {
    Own,
    System,
    NotAllowed,
    NotForUs,
}

/// Where an answer goes: a channel, and the root of the thread in it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Thread {
    pub channel: String,
    pub root: String,
}

/// A message for a session.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Message {
    pub session: SessionId,
    pub thread: Thread,
    /// A reply in a thread: the session should exist already.
    pub resume: bool,
    pub text: String,
    /// A thread in a channel or group message that this Boxmaker now takes part in.
    pub joins_thread: bool,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Route {
    Ignore(Ignored),
    /// Answer in the thread without a turn (a command).
    Reply {
        thread: Thread,
        text: String,
    },
    Queue(Message),
    /// `!approve` or `!deny`, for the approvals module to judge.
    Approval {
        thread: Thread,
        command: ApprovalCommand,
        /// Who wrote it, which post it is, and when (Mattermost's `create_at`).
        user: String,
        post: String,
        at: i64,
    },
}

/// An approval command as typed: exactly `!approve <id>`, `!deny <id>` or `!deny <id> <reason>`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ApprovalCommand {
    Approve(u64),
    Deny(u64, Option<String>),
    /// `!approve` or `!deny` in any other form.
    Malformed,
}

/// The longest reason a `!deny` may give, in characters.
pub const MAX_REASON: usize = proto::MAX_REASON;

/// `!approve …` or `!deny …` as an approval command; `None` for any other text.
pub fn approval_command(text: &str) -> Option<ApprovalCommand> {
    let after = text.strip_prefix('!')?;
    let (word, rest) = match after.split_once(' ') {
        Some((word, rest)) => (word, Some(rest)),
        None => (after, None),
    };
    if !matches!(word, "approve" | "deny") {
        return None;
    }
    let id = |s: &str| -> Option<u64> {
        if s.is_empty() || s.len() > 20 || !s.bytes().all(|b| b.is_ascii_digit()) {
            return None;
        }
        s.parse().ok()
    };
    let parsed = match (word, rest) {
        ("approve", Some(rest)) => id(rest).map(ApprovalCommand::Approve),
        ("deny", Some(rest)) => match rest.split_once(' ') {
            None => id(rest).map(|n| ApprovalCommand::Deny(n, None)),
            Some((n, reason)) => {
                let fits = !reason.is_empty() && reason.chars().count() <= MAX_REASON;
                id(n)
                    .filter(|_| fits)
                    .map(|n| ApprovalCommand::Deny(n, Some(reason.to_string())))
            }
        },
        _ => None,
    };
    Some(parsed.unwrap_or(ApprovalCommand::Malformed))
}

/// The text after a leading mention of this bot (`me`, lower-case), and the spaces after it. The
/// mention is matched as `named` matches names; nothing else is removed. A message that would be
/// left empty is kept as it was.
pub fn strip_mention<'a>(message: &'a str, me: &str) -> &'a str {
    let Some(after) = message.strip_prefix('@') else {
        return message;
    };
    let len = after
        .find(|c: char| !(c.is_ascii_alphanumeric() || matches!(c, '.' | '-' | '_')))
        .unwrap_or(after.len());
    let raw = after.get(..len).unwrap_or_default();
    let name = raw.trim_end_matches('.');
    if name.is_empty() || !name.eq_ignore_ascii_case(me) {
        return message;
    }
    let rest = after
        .get(name.len()..)
        .unwrap_or_default()
        .trim_start_matches(' ');
    if rest.is_empty() { message } else { rest }
}

pub struct Router {
    me_id: String,
    me_name: String,
    users: BTreeSet<String>,
    channels: BTreeSet<String>,
}

/// Every `@name` in a message, lower-cased: `a-z`, `0-9`, `.`, `-` and `_` after an `@`, without
/// trailing dots.
pub fn named(message: &str) -> Vec<String> {
    // Find each "@". The name after it is the longest run of ASCII letters, digits, ".", "-" and
    // "_", with trailing "." removed and lower-cased. Skip empty names. Continue after the name.
    let mut names = Vec::new();
    let bytes = message.as_bytes();
    let mut i = 0;
    while i < bytes.len() {
        if bytes[i] != b'@' {
            i += 1;
            continue;
        }
        let start = i + 1;
        let mut j = start;
        while j < bytes.len() {
            let c = bytes[j];
            if c.is_ascii_alphanumeric() || c == b'.' || c == b'-' || c == b'_' {
                j += 1;
            } else {
                break;
            }
        }
        if j > start {
            let trimmed = message[start..j].trim_end_matches('.').to_lowercase();
            if !trimmed.is_empty() {
                names.push(trimmed);
            }
        }
        i = j;
    }
    names
}

impl Router {
    pub fn new(me_id: &str, me_name: &str, users: &[String], channels: &[String]) -> Router {
        // Store the ids, the username lower-cased, and the two lists as sets.
        Router {
            me_id: me_id.to_string(),
            me_name: me_name.to_lowercase(),
            users: users.iter().cloned().collect(),
            channels: channels.iter().cloned().collect(),
        }
    }

    /// Is a post in a channel or group message for this Boxmaker? `known` says whether it has a
    /// session for a thread root.
    fn for_us(&self, post: &Post, known: &dyn Fn(&str) -> bool) -> bool {
        // Named this bot -> true. Otherwise true only for a reply (root_id not empty) in a known
        // thread (`known(root_id)`) that names nobody but channel, here or all.
        let names = named(&post.message);
        if names.iter().any(|n| n == &self.me_name) {
            return true;
        }
        if post.root_id.is_empty() || !known(&post.root_id) {
            return false;
        }
        names.iter().all(|n| EVERYONE.contains(&n.as_str()))
    }

    /// What to do with a new post (a post seen before is dropped by the caller first).
    pub fn route(&self, post: &Post, channel_type: &str, known: &dyn Fn(&str) -> bool) -> Route {
        // Section 7 of the spec, in this order: own post -> Ignore(Own); kind not empty ->
        // Ignore(System); user not allowed -> Ignore(NotAllowed); then the channel: "D" is always
        // ours; "O", "P" or "G" is ours when its id is allowed and `for_us`; anything else ->
        // Ignore(NotForUs). The thread root is root_id, or the post id when root_id is empty. Then
        // commands: Then a leading mention of this bot is stripped (M4b spec, section 7), and:
        // "!!..." drops one "!" and goes on as a message; `approval_command` -> Approval; any
        // other "!" -> Reply UNKNOWN_COMMAND. Then Queue: session
        // "mm-<root>" (if SessionId::new fails, Ignore(NotForUs)), resume = root_id not empty,
        // joins_thread = not "D".
        if post.user_id == self.me_id {
            return Route::Ignore(Ignored::Own);
        }
        if !post.kind.is_empty() {
            return Route::Ignore(Ignored::System);
        }
        if !self.users.contains(&post.user_id) {
            return Route::Ignore(Ignored::NotAllowed);
        }
        let channel_ok = match channel_type {
            "D" => true,
            "O" | "P" | "G" => self.channels.contains(&post.channel_id) && self.for_us(post, known),
            _ => false,
        };
        if !channel_ok {
            return Route::Ignore(Ignored::NotForUs);
        }

        let root = if post.root_id.is_empty() {
            post.id.clone()
        } else {
            post.root_id.clone()
        };
        let session = match SessionId::new(&format!("mm-{root}")) {
            Ok(session) => session,
            Err(_) => return Route::Ignore(Ignored::NotForUs),
        };
        let thread = Thread {
            channel: post.channel_id.clone(),
            root: root.clone(),
        };

        let message = strip_mention(&post.message, &self.me_name);
        if let Some(rest) = message.strip_prefix("!!") {
            return Route::Queue(Message {
                session,
                thread,
                resume: !post.root_id.is_empty(),
                text: format!("!{rest}"),
                joins_thread: channel_type != "D",
            });
        }
        if let Some(command) = approval_command(message) {
            return Route::Approval {
                thread,
                command,
                user: post.user_id.clone(),
                post: post.id.clone(),
                at: post.create_at,
            };
        }
        if message.starts_with('!') {
            return Route::Reply {
                thread,
                text: UNKNOWN_COMMAND.to_string(),
            };
        }

        Route::Queue(Message {
            session,
            thread,
            resume: !post.root_id.is_empty(),
            text: message.to_string(),
            joins_thread: channel_type != "D",
        })
    }
}

/// A turn to send: every message that was waiting, joined with a blank line.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Batch {
    pub session: SessionId,
    pub thread: Thread,
    pub resume: bool,
    pub text: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Pushed {
    /// No turn was running: send this one now.
    Start(Batch),
    /// A turn is running; the message waits for the next.
    Waiting,
    /// Too many are waiting: the message is dropped, answer `BUSY`.
    Full(Thread),
}

struct Pending {
    thread: Thread,
    waiting: Vec<String>,
}

/// The sessions with a turn running, and the messages waiting for each.
pub struct Queues {
    limit: usize,
    running: HashMap<SessionId, Pending>,
}

impl Queues {
    /// `limit` is the most messages that may wait per session.
    pub fn new(limit: usize) -> Queues {
        // An empty map.
        Queues {
            limit,
            running: HashMap::new(),
        }
    }

    pub fn push(&mut self, message: Message) -> Pushed {
        // If the session is running: with `limit` messages already waiting -> Full(thread); else
        // add the text to waiting -> Waiting. Otherwise insert it as running (its later batches
        // resume: true) and return Start with this message alone.
        let limit = self.limit;
        if let Some(pending) = self.running.get_mut(&message.session) {
            if pending.waiting.len() >= limit {
                return Pushed::Full(pending.thread.clone());
            }
            pending.waiting.push(message.text.clone());
            return Pushed::Waiting;
        }
        let batch = Batch {
            session: message.session.clone(),
            thread: message.thread.clone(),
            resume: message.resume,
            text: message.text.clone(),
        };
        self.running.insert(
            message.session,
            Pending {
                thread: message.thread.clone(),
                waiting: Vec::new(),
            },
        );
        Pushed::Start(batch)
    }

    /// A session's turn ended: the next batch, or `None`, and the session is no longer running.
    pub fn finish(&mut self, session: &SessionId) -> Option<Batch> {
        // Not running -> None. Nothing waiting -> remove it, None. Otherwise take every waiting
        // text, joined with "\n\n", as the next Batch (it stays running).
        let mut pending = self.running.remove(session)?;
        if pending.waiting.is_empty() {
            return None;
        }
        let batch = Batch {
            session: session.clone(),
            thread: pending.thread.clone(),
            resume: true,
            text: pending.waiting.join("\n\n"),
        };
        pending.waiting.clear();
        self.running.insert(session.clone(), pending);
        Some(batch)
    }

    /// How many sessions have a turn running.
    pub fn running(&self) -> usize {
        // How many sessions are running.
        self.running.len()
    }

    /// The threads with a turn running, for showing this bot as typing in them.
    pub fn threads(&self) -> Vec<Thread> {
        // The thread of every running session.
        self.running.values().map(|p| p.thread.clone()).collect()
    }
}