boxmaker/crates/gatewayd/src/deliver.rs

//! One turn on `loop.sock` and its answer in the thread (M4a spec, section 8). Typing is shown by
//! `serve`, which owns the WebSocket; this module only sends the turn and posts what comes back.

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,
    write_frame,
};

use crate::mm::MmError;
use crate::mm::rest::Client;
use crate::sessions::{Batch, Thread};

/// The longest reply we post, in characters (a post holds at most 16,383).
pub const MAX_POST: usize = 16_000;
pub const LOOP_DOWN: &str = "Boxmaker's loop is not running (see docs/runbook.md#loop-unavailable)";
pub const EMPTY_ANSWER: &str = "(the answer was empty)";

/// Somewhere to post: Mattermost, or a test's record.
pub trait Poster: Send + Sync {
    fn post(&self, channel: &str, root: &str, text: &str) -> Result<(), MmError>;
}

impl Poster for Client {
    fn post(&self, channel: &str, root: &str, text: &str) -> Result<(), MmError> {
        self.create_post(channel, root, text).map(|_| ())
    }
}

/// How a turn ended.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Outcome {
    Answer(String),
    Refused(WireError),
    /// `loop.sock` could not be reached or closed early; why, for the log.
    LoopDown(String),
}

pub fn approval_text(approval: u64) -> String {
    format!(
        "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
    // (`serde_json::to_value(code)` is a JSON string, e.g. "no_such_session").
    let code = match serde_json::to_value(error.code) {
        Ok(serde_json::Value::String(code)) => code,
        _ => return format!("Error: <unknown>: {}", error.detail),
    };
    format!("Error: {}: {}", code, error.detail)
}

/// An answer in posts of at most `MAX_POST` characters: each cut at the last newline before the
/// limit (the newline is dropped), or at the limit when there is none.
pub fn split_answer(text: &str) -> Vec<String> {
    // Blank (only whitespace) -> [EMPTY_ANSWER]. Otherwise, while the rest is longer than MAX_POST
    // *characters*: take the first MAX_POST characters; if they hold a newline past position 0, cut
    // at the last one and drop that newline; else cut at MAX_POST characters. The last part is the
    // rest.
    if text.trim().is_empty() {
        return vec![EMPTY_ANSWER.to_string()];
    }

    let mut parts = Vec::new();
    let mut rest = text;
    while rest.chars().count() > MAX_POST {
        let cut = rest
            .char_indices()
            .take(MAX_POST)
            .filter(|&(i, c)| i != 0 && c == '\n')
            .last()
            .map(|(i, _)| i);
        let (end, skip) = match cut {
            // A newline past position 0: drop it by cutting before and skipping past it.
            Some(bi) => (bi, bi + 1),
            // No newline: cut at the byte index of the MAX_POST-th character, never inside one.
            None => {
                let byte = rest
                    .char_indices()
                    .nth(MAX_POST)
                    .map(|(i, _)| i)
                    .unwrap_or(rest.len());
                (byte, byte)
            }
        };
        parts.push(rest[..end].to_string());
        rest = &rest[skip..];
    }
    parts.push(rest.to_string());
    parts
}

/// Send one turn and read it to its end; `on_event` sees every event.
fn one_turn(
    socket: &Path,
    batch: &Batch,
    resume: bool,
    on_event: &mut dyn FnMut(&TurnEvent),
) -> Outcome {
    // 1. Connect to the socket (else LoopDown("cannot connect to <path>: <e>")).
    let mut stream = match UnixStream::connect(socket) {
        Ok(stream) => stream,
        Err(e) => {
            return Outcome::LoopDown(format!("cannot connect to {}: {}", socket.display(), e));
        }
    };

    // 2. `write_frame` one Envelope: v PROTOCOL_VERSION, id 1, final true, msg Turn (else
    //    LoopDown).
    let turn = Turn {
        session: batch.session.clone(),
        content: batch.text.clone(),
        resume,
    };
    if let Err(e) = write_frame(
        &mut stream,
        &Envelope {
            v: PROTOCOL_VERSION,
            id: 1,
            r#final: true,
            msg: Message::Turn(turn),
        },
    ) {
        return Outcome::LoopDown(format!("cannot send the turn: {}", e));
    }

    // 3. `read_frame` until the end: (id 1, not final, TurnEvent) -> on_event; (1, final, TurnDone)
    //    -> Answer with its content; (1, final, Error) -> Refused; a read error -> LoopDown("the
    //    turn ended early: <e>"); anything else -> LoopDown("an unexpected frame").
    loop {
        let env = match read_frame(&mut stream) {
            Ok(env) => env,
            Err(e) => return Outcome::LoopDown(format!("the turn ended early: {}", e)),
        };
        if env.id != 1 {
            return Outcome::LoopDown("an unexpected frame".to_string());
        }
        match (env.r#final, &env.msg) {
            (false, Message::TurnEvent(e)) => on_event(e),
            (true, Message::TurnDone(d)) => return Outcome::Answer(d.content.clone()),
            (true, Message::Error(w)) => return Outcome::Refused(w.clone()),
            _ => return Outcome::LoopDown("an unexpected frame".to_string()),
        }
    }
}

/// A turn for a batch. A reply in a thread `loopd` does not know creates the session, as
/// `bxctl chat --session` does.
pub fn run_turn(socket: &Path, batch: &Batch, on_event: &mut dyn FnMut(&TurnEvent)) -> Outcome {
    // `one_turn` with batch.resume. When that is Refused with NoSuchSession and batch.resume was
    // true, one more `one_turn` with resume false. Otherwise the first outcome.
    match one_turn(socket, batch, batch.resume, on_event) {
        Outcome::Refused(err) if err.code == ErrorCode::NoSuchSession && batch.resume => {
            one_turn(socket, batch, false, on_event)
        }
        outcome => outcome,
    }
}

/// Post one thing in a thread, logging (via `log`) when Mattermost refuses it, and carry on.
fn post(poster: &dyn Poster, thread: &Thread, text: String, log: &dyn Fn(&str)) {
    if let Err(e) = poster.post(&thread.channel, &thread.root, &text) {
        log(&format!(
            "gatewayd: cannot post in {} (thread {}): {}",
            thread.channel, thread.root, e
        ));
    }
}

/// 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),
    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
            && !notice.arrives(*approval)
        {
            post(poster, &batch.thread, approval_text(*approval), log);
        }
    });

    match outcome {
        Outcome::Answer(content) => {
            for part in split_answer(&content) {
                post(poster, &batch.thread, part, log);
            }
        }
        Outcome::Refused(err) => post(poster, &batch.thread, error_text(&err), log),
        Outcome::LoopDown(why) => {
            log(&format!("gatewayd: {}: {}", batch.session.as_str(), why));
            post(poster, &batch.thread, LOOP_DOWN.to_string(), log);
        }
    }
}