boxmaker/crates/bxctl/src/memory.rs

//! The owner's memory commands over `memoryd`'s owner socket: `reindex`, and `memory log`,
//! `undo`, `pending`, `accept`, `reject` (M5a spec, section 10), and `sync` (M5c spec, section 5).

use std::io::Write;
use std::os::unix::net::UnixStream;
use std::path::{Path, PathBuf};
use std::time::Duration;

use bxcore::{
    Empty, Envelope, MemoryAdmin, MemoryCommand, Message, PROTOCOL_VERSION, read_frame, write_frame,
};

use crate::cli::{Command, UsageError};
use crate::escape::escape_json_text;

/// How long `bxctl` waits for `memoryd`: a reindex reads every session log.
pub const MEMORY_TIMEOUT: Duration = Duration::from_secs(300);

pub fn default_socket(home: &Path) -> PathBuf {
    home.join("run").join("owner-memory").join("admin.sock")
}

/// What to ask `memoryd`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Ask {
    Reindex,
    Admin(MemoryCommand, String),
}

/// `bxctl reindex [--memory-socket PATH]`.
pub fn parse_reindex(args: &[String], home: &Path) -> Result<Command, UsageError> {
    let (socket, rest) = socket_flag(args, home)?;
    if !rest.is_empty() {
        return Err(UsageError);
    }
    Ok(Command::Memory {
        socket,
        ask: Ask::Reindex,
    })
}

/// `bxctl memory log [N] | undo <id> | pending | accept <id> | reject <id> | sync`, each with an
/// optional `--memory-socket PATH`.
pub fn parse_memory(args: &[String], home: &Path) -> Result<Command, UsageError> {
    let (socket, rest) = socket_flag(args, home)?;
    let rest: Vec<&str> = rest.iter().map(String::as_str).collect();
    let ask = match rest.as_slice() {
        ["log"] => Ask::Admin(MemoryCommand::Log, String::new()),
        ["log", n] => {
            if n.parse::<usize>().is_err() {
                return Err(UsageError);
            }
            Ask::Admin(MemoryCommand::Log, n.to_string())
        }
        ["pending"] => Ask::Admin(MemoryCommand::Pending, String::new()),
        ["undo", id] => Ask::Admin(MemoryCommand::Undo, id.to_string()),
        ["accept", id] => Ask::Admin(MemoryCommand::Accept, id.to_string()),
        ["reject", id] => Ask::Admin(MemoryCommand::Reject, id.to_string()),
        ["sync"] => Ask::Admin(MemoryCommand::Sync, String::new()),
        _ => return Err(UsageError),
    };
    Ok(Command::Memory { socket, ask })
}

/// Take `--memory-socket PATH` out of `args`, once at most; the rest in order.
fn socket_flag(args: &[String], home: &Path) -> Result<(PathBuf, Vec<String>), UsageError> {
    let mut socket = default_socket(home);
    let mut rest = Vec::with_capacity(args.len());
    let mut seen = false;
    let mut i = 0;
    while i < args.len() {
        if args[i] == "--memory-socket" {
            if seen {
                return Err(UsageError);
            }
            seen = true;
            socket = PathBuf::from(args.get(i + 1).cloned().ok_or(UsageError)?);
            i += 2;
        } else {
            rest.push(args[i].clone());
            i += 1;
        }
    }
    Ok((socket, rest))
}

/// `n` with commas between thousands.
fn grouped(n: u64) -> String {
    let digits = n.to_string();
    let mut out = String::with_capacity(digits.len() + digits.len() / 3);
    for (i, c) in digits.chars().enumerate() {
        if i > 0 && (digits.len() - i).is_multiple_of(3) {
            out.push(',');
        }
        out.push(c);
    }
    out
}

/// Ask `memoryd` and print its answer. `Err` is the line for stderr.
pub fn run(socket: &Path, ask: &Ask, timeout: Duration, out: &mut dyn Write) -> Result<(), String> {
    let unreachable =
        |e: &dyn std::fmt::Display| format!("cannot reach memoryd at {}: {e}", socket.display());
    let mut stream = UnixStream::connect(socket).map_err(|e| unreachable(&e))?;
    stream
        .set_read_timeout(Some(timeout))
        .map_err(|e| unreachable(&e))?;
    stream
        .set_write_timeout(Some(timeout))
        .map_err(|e| unreachable(&e))?;

    let msg = match ask {
        Ask::Reindex => Message::Reindex(Empty {}),
        Ask::Admin(command, arg) => Message::MemoryAdmin(MemoryAdmin {
            command: *command,
            arg: arg.clone(),
        }),
    };
    let env = Envelope {
        v: PROTOCOL_VERSION,
        id: 1,
        r#final: true,
        msg,
    };
    write_frame(&mut stream, &env).map_err(|e| unreachable(&e))?;

    let reply = read_frame(&mut stream).map_err(|e| format!("memoryd did not answer: {e}"))?;
    let text = match reply.msg {
        Message::Reindexed(r) => format!(
            "reindexed: {} logs, {} records, {} notes",
            grouped(r.logs),
            grouped(r.records),
            grouped(r.notes)
        ),
        Message::MemoryAdminReply(reply) => reply.text,
        Message::Error(e) => return Err(format!("memoryd: {}", escape_json_text(&e.detail))),
        _ => return Err("memoryd answered with the wrong message".to_string()),
    };

    writeln!(out, "{text}").map_err(|e| e.to_string())
}