boxmaker/crates/gatewayd/src/gateway.rs

//! `gateway.sock`: the listener, and one thread per `brokerd` connection (M4b spec, sections 4
//! and 6). A connection thread speaks the protocol only; everything that touches Mattermost or the
//! state file is asked of the serve loop, which alone owns them.

use std::os::unix::fs::{DirBuilderExt, FileTypeExt, PermissionsExt};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::Path;
use std::sync::Arc;
use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender, TryRecvError};
use std::time::Duration;

use bxcore::{
    ApprovalPosted, ApprovalReply, Envelope, ErrorCode, FrameError, GatewayOutcome, Message,
    PROTOCOL_VERSION, PendingApproval, WireError, read_frame, write_frame,
};

/// A log line printer, shared by the threads.
pub type Log = Arc<dyn Fn(&str) + Send + Sync>;

/// What a connection thread asks of the serve loop.
#[derive(Debug)]
pub enum ToLoop {
    /// Post this approval; an accepted answer comes back on `answer`, and the post ids (or why
    /// it could not be posted) on `shown`.
    Show {
        info: PendingApproval,
        answer: Sender<ApprovalReply>,
        shown: Sender<Result<Vec<String>, String>>,
    },
    /// It ended: edit its last post, and forget it.
    Ended { approval: u64, ending: Ending },
}

/// How an approval ended, as far as this `gatewayd` knows.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Ending {
    /// Answered here, by `by`, and `brokerd` said what came of it.
    Answered {
        by: String,
        reason: Option<String>,
        outcome: GatewayOutcome,
    },
    /// `brokerd` closed first, or said nothing after the answer.
    NoLonger,
}

/// How long to wait: for the first frame, for the loop to post, for `brokerd`'s outcome; and how
/// often to look for an answer meanwhile.
#[derive(Debug, Clone, Copy)]
pub struct Timing {
    pub first_frame: Duration,
    pub posting: Duration,
    pub outcome: Duration,
    pub tick: Duration,
}

impl Default for Timing {
    fn default() -> Self {
        Timing {
            first_frame: Duration::from_secs(10),
            posting: Duration::from_secs(120),
            outcome: Duration::from_secs(30),
            tick: Duration::from_millis(100),
        }
    }
}

/// Listen on `socket`: its directory made 0700, a stale socket removed, the socket made 0600.
pub fn listen(socket: &Path) -> Result<UnixListener, String> {
    // 1. The directory: `socket.parent()`, which must not be empty, else "cannot prepare <socket>:
    //    a socket needs a directory".
    // 2. `std::fs::DirBuilder::new().recursive(true).mode(0o700).create(dir)`, then
    //    `set_permissions(dir, 0o700)`; an error is "cannot prepare <dir>: <e>".
    // 3. `std::fs::symlink_metadata(socket)`: a socket (`file_type().is_socket()`) is removed
    //    (stale); anything else there is "cannot listen on <socket>: something else is there";
    //    nothing there is fine.
    // 4. `UnixListener::bind(socket)`, then `set_permissions(socket, 0o600)`; an error is "cannot
    //    listen on <socket>: <e>".
    // 5. Ok(listener). Paths are shown with `.display()`.
    let dir = match socket.parent() {
        Some(dir) => dir,
        None => {
            return Err(format!(
                "cannot prepare {}: a socket needs a directory",
                socket.display()
            ));
        }
    };
    if let Err(e) = std::fs::DirBuilder::new()
        .recursive(true)
        .mode(0o700)
        .create(dir)
    {
        return Err(format!("cannot prepare {}: {e}", dir.display()));
    }
    if let Err(e) = std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700)) {
        return Err(format!("cannot prepare {}: {e}", dir.display()));
    }
    match std::fs::symlink_metadata(socket) {
        Ok(meta) if meta.file_type().is_socket() => {
            if let Err(e) = std::fs::remove_file(socket) {
                return Err(format!("cannot listen on {}: {e}", socket.display()));
            }
        }
        Ok(_) => {
            return Err(format!(
                "cannot listen on {}: something else is there",
                socket.display()
            ));
        }
        Err(_) => {}
    }
    let listener = match UnixListener::bind(socket) {
        Ok(listener) => listener,
        Err(e) => return Err(format!("cannot listen on {}: {e}", socket.display())),
    };
    if let Err(e) = std::fs::set_permissions(socket, std::fs::Permissions::from_mode(0o600)) {
        return Err(format!("cannot listen on {}: {e}", socket.display()));
    }
    Ok(listener)
}

/// Accept connections on a thread of their own, each served on its own thread.
pub fn serve(
    listener: UnixListener,
    to_loop: Sender<ToLoop>,
    timing: Timing,
    log: Log,
) -> std::io::Result<()> {
    // Spawn (with `std::thread::Builder`, `?`) a thread that loops over `listener.incoming()`,
    // skipping errors. For each stream: clone `to_loop` and `log`, and spawn a thread running
    // `connection(stream, &to_loop, timing, &log)`; if that spawn fails, log "gatewayd: cannot
    // start a thread for an approval: <e>". Then Ok(()).
    std::thread::Builder::new()
        .name("gateway-accept".to_string())
        .spawn(move || {
            for stream in listener.incoming() {
                match stream {
                    Ok(stream) => {
                        let to_loop = to_loop.clone();
                        let thread_log = log.clone();
                        if let Err(e) = std::thread::Builder::new()
                            .name("gateway-connection".to_string())
                            .spawn(move || connection(stream, &to_loop, timing, &thread_log))
                        {
                            log(&format!(
                                "gatewayd: cannot start a thread for an approval: {e}"
                            ));
                        }
                    }
                    Err(_) => continue,
                }
            }
        })?;
    Ok(())
}

fn frame(id: u64, last: bool, msg: Message) -> Envelope {
    Envelope {
        v: PROTOCOL_VERSION,
        id,
        r#final: last,
        msg,
    }
}

/// Every frame `brokerd` sends, read whole on a thread of its own; shutting the stream down ends it.
fn frames(stream: &UnixStream) -> std::io::Result<Receiver<Result<Envelope, FrameError>>> {
    // As in `brokerd`'s courier: `let mut reader = stream.try_clone()?;`, a channel, and a thread
    // (`std::thread::Builder`, `?`) looping `let frame = read_frame(&mut reader); let last =
    // frame.is_err(); if tx.send(frame).is_err() || last { return; }`. Then `Ok(rx)`.
    let mut reader = stream.try_clone()?;
    let (tx, rx) = mpsc::channel();
    std::thread::Builder::new()
        .name("gateway-frame".to_string())
        .spawn(move || {
            loop {
                let frame = read_frame(&mut reader);
                let last = frame.is_err();
                if tx.send(frame).is_err() || last {
                    return;
                }
            }
        })?;
    Ok(rx)
}

/// One `brokerd` connection, frames 1 to 4 of the spec's section 4.
pub fn connection(mut stream: UnixStream, to_loop: &Sender<ToLoop>, timing: Timing, log: &Log) {
    // 1. The request, within `first_frame`.
    let _ = stream.set_read_timeout(Some(timing.first_frame));
    let info = match read_frame(&mut stream) {
        Ok(Envelope {
            id: 1,
            r#final: true,
            msg: Message::ApprovalRequest(req),
            ..
        }) => req.approval,
        _ => return,
    };
    let _ = stream.set_read_timeout(None);
    let id = info.approval;
    // 2. The loop posts it.
    let (answer_tx, answer_rx) = mpsc::channel();
    let (shown_tx, shown_rx) = mpsc::channel();
    if to_loop
        .send(ToLoop::Show {
            info,
            answer: answer_tx,
            shown: shown_tx,
        })
        .is_err()
    {
        return;
    }
    let posts = match shown_rx.recv_timeout(timing.posting) {
        Ok(Ok(posts)) => posts,
        Ok(Err(why)) => {
            let e = WireError {
                code: ErrorCode::Internal,
                detail: why,
            };
            let _ = write_frame(&mut stream, &frame(1, true, Message::Error(e)));
            return;
        }
        Err(_) => {
            let e = WireError {
                code: ErrorCode::Internal,
                detail: "gatewayd did not post it in time".to_string(),
            };
            let _ = write_frame(&mut stream, &frame(1, true, Message::Error(e)));
            return;
        }
    };
    let rx = match frames(&stream) {
        Ok(rx) => rx,
        Err(_) => return end(to_loop, id, Ending::NoLonger),
    };
    if write_frame(
        &mut stream,
        &frame(1, false, Message::ApprovalPosted(ApprovalPosted { posts })),
    )
    .is_err()
    {
        return end(to_loop, id, Ending::NoLonger);
    }
    // 3. An answer from an approver, or brokerd closing first.
    let mut answers = Some(answer_rx);
    let reply = loop {
        if let Some(answers_rx) = &answers {
            match answers_rx.recv_timeout(timing.tick) {
                Ok(reply) => break reply,
                Err(RecvTimeoutError::Timeout) => {}
                Err(RecvTimeoutError::Disconnected) => answers = None,
            }
        } else {
            std::thread::sleep(timing.tick);
        }
        match rx.try_recv() {
            Err(TryRecvError::Empty) => {}
            // brokerd closed (the approval ended elsewhere), or sent what it may not.
            _ => {
                let _ = stream.shutdown(std::net::Shutdown::Both);
                return end(to_loop, id, Ending::NoLonger);
            }
        }
    };
    let (by, reason) = (reply.by.clone(), reply.reason.clone());
    if write_frame(&mut stream, &frame(1, true, Message::ApprovalReply(reply))).is_err() {
        return end(to_loop, id, Ending::NoLonger);
    }
    // 4. What came of it.
    let ending = match rx.recv_timeout(timing.outcome) {
        Ok(Ok(Envelope {
            id: 2,
            r#final: true,
            msg: Message::ApprovalOutcome(o),
            ..
        })) => Ending::Answered {
            by,
            reason,
            outcome: o.outcome,
        },
        _ => {
            (log)(&format!(
                "gatewayd: approval {id}: brokerd said nothing after the answer"
            ));
            Ending::NoLonger
        }
    };
    let _ = stream.shutdown(std::net::Shutdown::Both);
    end(to_loop, id, ending);
}

fn end(to_loop: &Sender<ToLoop>, approval: u64, ending: Ending) {
    // `let _ = to_loop.send(ToLoop::Ended { approval, ending });`
    let _ = to_loop.send(ToLoop::Ended { approval, ending });
}