boxmaker/docs/plans/M5a/files/crates/memoryd/src/serve.rs

//! Serving: three sockets, one request per connection, each answered by one frame (M5a spec,
//! section 7). One lock covers the files and the index, so writes never interleave.

use std::os::unix::net::{UnixListener, UnixStream};
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::Duration;

use proto::{
    Empty, Envelope, ErrorCode, MemoryAdminReply, MemoryArgs, MemoryBlockReply, MemoryCall,
    MemoryCommand, Message, PROTOCOL_VERSION, RecallArgs, Reindexed, Timestamp, ToolResponse,
    WireError, read_frame, write_frame,
};

use crate::config::Config;
use crate::edit::{self, Context};
use crate::files::Store;
use crate::index::{self, Index};
use crate::recall::Recalled;
use crate::{journal, recall, review};

/// How long a peer may take to send its request.
pub const READ_WAIT: Duration = Duration::from_secs(10);

/// Where log lines go: stderr in `main`, a buffer in tests.
pub type Log = Arc<dyn Fn(&str) + Send + Sync>;

struct State {
    index: Index,
}

pub struct Memoryd {
    config: Config,
    store: Store,
    state: Mutex<State>,
    log: Log,
}

/// A started `memoryd`: its three listeners, bound and waiting.
pub struct Running {
    memoryd: Arc<Memoryd>,
    listeners: Vec<(UnixListener, Which)>,
}

#[derive(Clone, Copy)]
enum Which {
    Loop,
    Broker,
    Admin,
}

/// Create what is missing under `memory/`, open the index and catch it up, then bind the three
/// sockets. The error names what failed; the caller points at `memoryd-start-failed`.
pub fn start(config: Config, log: Log) -> Result<Running, String> {
    // 1. `Store::new(&config.memory_dir())`, then `store.init(&config.memory.soul_starter)?`.
    // 2. `Index::open(&config.index_path(), config.index.max_class)?`, then
    //    `index.catch_up(&config.sessions_dir(), &store.path("notes.md"))?`; log each problem as
    //    format!("memoryd: not indexed: {problem}").
    // 3. `crate::socket::listen` the loop, broker and admin sockets of the config, in that order,
    //    paired with Which::Loop, Which::Broker and Which::Admin.
    // 4. Ok(Running { memoryd: Arc::new(Memoryd { config, store, state: Mutex::new(State { index
    //    }), log }), listeners }).
    todo!()
}

impl Running {
    /// Accept on every socket, one thread per connection, until a listener fails.
    pub fn run(self) -> Result<(), String> {
        // An mpsc channel of String. For each (listener, which): spawn a thread (with
        // `std::thread::Builder::new().spawn`, its error -> Err(format!("cannot start a thread:
        // {e}"))) that loops over `listener.incoming()`: Ok(stream) -> spawn one more thread
        // running `memoryd.connection(stream, which)`, and if that spawn fails, log
        // format!("memoryd: cannot start a thread: {e}") and go on; Err(e) -> send format!("a
        // listener failed: {e}") and return. Drop the sender, then wait on `rx.recv()`: a message
        // -> Err(it); all senders gone -> Ok(()).
        todo!()
    }
}

fn error(detail: String) -> Message {
    Message::Error(WireError {
        code: ErrorCode::Internal,
        detail,
    })
}

fn bad(detail: &str) -> Message {
    Message::Error(WireError {
        code: ErrorCode::BadMessage,
        detail: detail.to_string(),
    })
}

impl Memoryd {
    fn lock(&self) -> MutexGuard<'_, State> {
        self.state.lock().unwrap_or_else(|p| p.into_inner())
    }

    fn connection(&self, mut stream: UnixStream, which: Which) {
        // `set_read_timeout(Some(READ_WAIT))` (ignore its result). `read_frame`; on error just
        // return. The reply is `self.handle(request.msg, which)`, written with `write_frame` in an
        // Envelope with v PROTOCOL_VERSION, the request's id, final true (ignore the result).
        todo!()
    }

    /// The answer to one request on one socket. A message sent to the wrong socket is refused.
    fn handle(&self, msg: Message, which: Which) -> Message {
        match (which, msg) {
            (Which::Loop, Message::MemoryBlock(_)) => self.block(),
            (Which::Loop, Message::SessionUpdated(update)) => {
                self.session_updated(update.session.as_str())
            }
            (Which::Broker, Message::MemoryCall(call)) => Message::ToolResponse(self.call(call)),
            (Which::Admin, Message::Reindex(_)) => self.reindex(),
            (Which::Admin, Message::MemoryAdmin(admin)) => self.admin(admin.command, &admin.arg),
            _ => bad("memoryd does not take this message on this socket"),
        }
    }

    fn block(&self) -> Message {
        // Hold the lock. `self.store.block()`: Ok(block) -> log each (file, refused) of
        // block.left_out as format!("memoryd: an entry of {} was left out of the block: {}",
        // file.name(), refused.sentence()), and answer Message::MemoryBlockReply(MemoryBlockReply {
        // text: block.text }); Err(e) -> error(e).
        todo!()
    }

    fn session_updated(&self, session: &str) -> Message {
        // Hold the lock. `index::session_logs(&self.config.sessions_dir(), session)`; its error ->
        // error(e). For each (epoch, path): `state.index.index_log(path, session, *epoch)`, and log
        // an error as format!("memoryd: not indexed: {e}"). Answer Message::Ok(Empty {}).
        todo!()
    }

    /// A decided call from `brokerd`: `recall` or `memory`, under the one lock. `Failed` carries the
    /// sentence for the model.
    fn call(&self, call: MemoryCall) -> ToolResponse {
        // Hold the lock. By the request's tool: proto::memory::RECALL -> `self.recall(&state,
        // &call)`; proto::memory::MEMORY -> `self.memory(&state, &call)`; anything else ->
        // Err("memoryd runs only recall and memory"). Ok(found) -> ToolResponse::Result { content:
        // found.text, class: found.class, untrusted: found.untrusted, truncated: false };
        // Err(message) -> ToolResponse::Failed { message }.
        todo!()
    }

    /// `recall`: search, labelled by what it found.
    fn recall(&self, state: &State, call: &MemoryCall) -> Result<Recalled, String> {
        // Parse the request's arguments as RecallArgs (an error -> format!("recall's arguments:
        // {e}")), then `recall::recall(&state.index, &self.store, &args, session,
        // Timestamp::now())`.
        todo!()
    }

    /// `memory`: staged for review, or applied and journalled; then what it appended to `notes.md`
    /// is indexed. Its answer is labelled private and trusted: it is `memoryd`'s own sentence.
    fn memory(&self, state: &State, call: &MemoryCall) -> Result<Recalled, String> {
        // 1. Parse the arguments as MemoryArgs (an error -> format!("memory's arguments: {e}")).
        // 2. A Context: session, call (request.call.0), taint and untrusted from the MemoryCall,
        //    now Timestamp::now().
        // 3. When `self.config.memory.review` is on, the action is not a Note, and the call is not
        //    untrusted: `review::submit(&self.store, &args, &request.arguments, &ctx)`. Otherwise
        //    `edit::apply(&self.store, &args, &ctx)`, then `journal::record(&self.store, &ctx,
        //    &outcome)?`, answering outcome.answer.
        // 4. `self.index_notes(state)`, whatever the answer.
        // 5. The answer as a Recalled { text, class Private, untrusted false }.
        todo!()
    }

    /// Index what was appended to `notes.md`; a failure is logged, and the next start catches up.
    fn index_notes(&self, state: &State) {
        // Read `self.store.path("notes.md")` (an error names the path) and
        // `state.index.index_notes(&text)`; log an error as format!("memoryd: notes not indexed:
        // {e}").
        todo!()
    }

    fn reindex(&self) -> Message {
        // Hold the lock (mutably). `index::rebuild(&self.config.index_path(), max_class,
        // &sessions_dir, &self.store.path("notes.md"))`; its error -> error(e); log each problem as
        // format!("memoryd: not indexed: {problem}"). Then `Index::open` the same path again and
        // put it in `state.index` (the old connection still points at the replaced file); its error
        // -> error(e). Answer Message::Reindexed { logs, records, notes } from the counts.
        todo!()
    }

    fn admin(&self, command: MemoryCommand, arg: &str) -> Message {
        // Hold the lock; now = Timestamp::now(). Log: an empty arg means 10, otherwise it must
        // parse as usize, else Err(format!("log takes a count; {arg:?} is not one")); then
        // `journal::log(&self.store, n)`. Undo -> `journal::undo(&self.store, arg, now)`. Pending
        // -> `review::list`. Accept -> `review::accept(.., arg, now)`. Reject ->
        // `review::reject(.., arg, now)`. Then `self.index_notes(&state)`. Ok(text) ->
        // Message::MemoryAdminReply(MemoryAdminReply { text }); Err(e) -> error(e).
        todo!()
    }
}