narju/

adapter.rs

use rustyline::error::ReadlineError;
use rustyline::DefaultEditor;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::sync::mpsc::{Receiver, Sender};

use crate::floor::host::Str;
use crate::floor::{list, list_elements, pair, Val};
use crate::world::{reply, request, Event};

fn sym(s: &str) -> Val {
    Val::Sym(s.into())
}

fn throw(why: Val) -> Val {
    pair(sym("throw"), why)
}

/// A string as itself, anything else as it prints: a capability taking only
/// strings would make the object language convert before it could say anything.
fn text(v: &Val) -> String {
    match Str::of(v) {
        Some(s) => s.to_string(),
        None => v.to_string(),
    }
}

/// Answer a request, or say nothing if the message was a cast. The reply port
/// closing ends the adapter: the world it was answering is gone.
async fn answer(back: &Sender<Event>, msg: &Val, v: Val) -> bool {
    match request(msg) {
        Some((from, id, _)) => back.send(Event::Post(from, reply(id, v))).await.is_ok(),
        None => true,
    }
}

/// The body of a request, or the whole message when it is a cast.
fn body(msg: &Val) -> Val {
    match request(msg) {
        Some((_, _, body)) => body,
        None => msg.clone(),
    }
}

/// `('host stdout)`: `('write v)` and `('line v)`, answering `'ok`.
pub async fn stdout(mut inbox: Receiver<Val>, back: Sender<Event>) {
    let mut out = tokio::io::stdout();
    while let Some(msg) = inbox.recv().await {
        let v = match list_elements(&body(&msg)).as_deref() {
            Some([verb, arg]) if *verb == sym("write") => put(&mut out, &text(arg)).await,
            Some([verb, arg]) if *verb == sym("line") => {
                put(&mut out, &format!("{}\n", text(arg))).await
            }
            _ => throw(pair(sym("bad-request"), body(&msg))),
        };
        if !answer(&back, &msg, v).await {
            return;
        }
    }
}

async fn put(out: &mut tokio::io::Stdout, s: &str) -> Val {
    match out.write_all(s.as_bytes()).await {
        Ok(()) => match out.flush().await {
            Ok(()) => sym("ok"),
            Err(e) => throw(pair(sym("io"), Str::val(e.to_string()))),
        },
        Err(e) => throw(pair(sym("io"), Str::val(e.to_string()))),
    }
}

/// `('host stdin)`: `('read)` or `('read prompt)`, answering the text or
/// `'eof`. A prompt is ignored on a stream that is not a terminal, and a cast
/// is dropped since there is nowhere to put the line.
pub async fn stdin(mut inbox: Receiver<Val>, back: Sender<Event>) {
    let mut lines = BufReader::new(tokio::io::stdin()).lines();
    while let Some(msg) = inbox.recv().await {
        let Some((from, id, ask)) = request(&msg) else {
            continue;
        };
        let v = match list_elements(&ask).as_deref() {
            Some([verb] | [verb, _]) if *verb == sym("read") => match lines.next_line().await {
                Ok(Some(s)) => Str::val(s),
                Ok(None) => sym("eof"),
                Err(e) => throw(pair(sym("io"), Str::val(e.to_string()))),
            },
            _ => throw(pair(sym("bad-request"), ask)),
        };
        if back.send(Event::Post(from, reply(id, v))).await.is_err() {
            return;
        }
    }
}

/// The interactive console: `('write v)`, `('line v)` and `('read prompt)` on
/// one peer, because ordering between printing and prompting is the whole
/// difficulty. Sharing one inbox makes the channel's order the screen's order;
/// two channels would let a print land mid-line. Reading a tty line blocks, so
/// a thread is occupied either way.
///
/// Registered under both `stdout` and `stdin`, so an object program says the
/// same thing whether its output is a terminal or a pipe.
pub async fn terminal(mut inbox: Receiver<Val>, back: Sender<Event>) {
    let _ = tokio::task::spawn_blocking(move || {
        let Ok(mut rl) = DefaultEditor::new() else {
            return;
        };
        while let Some(msg) = inbox.blocking_recv() {
            let v = match list_elements(&body(&msg)).as_deref() {
                Some([verb, arg]) if *verb == sym("write") => emit(&text(arg)),
                Some([verb, arg]) if *verb == sym("line") => emit(&format!("{}\n", text(arg))),
                Some([verb] | [verb, _]) if *verb == sym("read") => {
                    let prompt = match list_elements(&body(&msg)).as_deref() {
                        Some([_, p]) => text(p),
                        _ => String::new(),
                    };
                    match rl.readline(&prompt) {
                        Ok(line) => {
                            rl.add_history_entry(line.as_str()).ok();
                            Str::val(line)
                        }
                        // Ctrl-c abandons the line, not the session.
                        Err(ReadlineError::Interrupted) => Str::val(""),
                        Err(ReadlineError::Eof) => sym("eof"),
                        Err(e) => throw(pair(sym("io"), Str::val(e.to_string()))),
                    }
                }
                _ => throw(pair(sym("bad-request"), body(&msg))),
            };
            if let Some((from, id, _)) = request(&msg) {
                if back.blocking_send(Event::Post(from, reply(id, v))).is_err() {
                    return;
                }
            }
        }
    })
    .await;
}

/// `('host files)`: `('read path)`, answering the text, and `('root)`,
/// answering where fetched modules were put.
///
/// Reading a file is a capability rather than a verb, so a task with no route
/// to this peer cannot open one - which is what makes loading a module by name
/// an ordinary message. `('root)` is the host's answer to give, since a build
/// system may stage sources somewhere unpredictable; it grants no authority,
/// as `('read path)` already takes any path.
pub async fn files(root: String, mut inbox: Receiver<Val>, back: Sender<Event>) {
    while let Some(msg) = inbox.recv().await {
        let Some((from, id, ask)) = request(&msg) else {
            continue;
        };
        let v = match list_elements(&ask).as_deref() {
            Some([verb, path]) if *verb == sym("read") => {
                let name = text(path);
                match tokio::fs::read_to_string(&name).await {
                    Ok(s) => Str::val(s),
                    // Named, because a caller does not always choose the path:
                    // `need` reads a lock file the program never wrote down.
                    Err(e) => throw(pair(sym("io"), Str::val(format!("{name}: {e}")))),
                }
            }
            Some([verb]) if *verb == sym("root") => Str::val(root.clone()),
            _ => throw(pair(sym("bad-request"), ask)),
        };
        if back.send(Event::Post(from, reply(id, v))).await.is_err() {
            return;
        }
    }
}

/// `('host disk)`: `('write path text)`, answering `'ok`.
///
/// A second peer rather than a second verb on [`files`], because a peer is the
/// unit of granting and a host may want reads without writes. It creates
/// nothing but the file: a missing directory is an error, since a mistyped
/// path would otherwise get a tree instead of a complaint.
pub async fn disk(mut inbox: Receiver<Val>, back: Sender<Event>) {
    while let Some(msg) = inbox.recv().await {
        let Some((from, id, ask)) = request(&msg) else {
            continue;
        };
        let v = match list_elements(&ask).as_deref() {
            Some([verb, path, body]) if *verb == sym("write") => {
                let name = text(path);
                match tokio::fs::write(&name, text(body)).await {
                    Ok(()) => sym("ok"),
                    Err(e) => throw(pair(sym("io"), Str::val(format!("{name}: {e}")))),
                }
            }
            _ => throw(pair(sym("bad-request"), ask)),
        };
        if back.send(Event::Post(from, reply(id, v))).await.is_err() {
            return;
        }
    }
}

/// A repository's place under the store root: its url without the scheme.
/// Derived once here and written into the lock as a field, so `naj/prelude.naj`
/// and `flake.nix` join strings rather than reimplement the rule.
///
/// Checked rather than trusted, since it comes from a file the program was
/// given and ends up as an argument to `git`. Not about quoting - there is no
/// shell - but a segment beginning with `-` would be read as an option and
/// `..` would place the source outside the store. `https` only, which rules
/// out git's scp-like syntax and a `file://` url naming somewhere unreadable.
fn store_path(url: &str) -> Result<&str, String> {
    let rest = url
        .strip_prefix("https://")
        .ok_or_else(|| format!("{url}: want an https url"))?;
    if rest.split('/').count() < 2 {
        return Err(format!("{url}: want a host and a path"));
    }
    for part in rest.split('/') {
        if part.is_empty() || part.starts_with('-') || part.starts_with('.') {
            return Err(format!("{url}: bad url"));
        }
        if !part
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.'))
        {
            return Err(format!("{url}: bad url"));
        }
    }
    Ok(rest)
}

/// Checked for the reason [`store_path`] is: a revision names a directory in
/// the store.
fn rev(s: &str) -> Result<&str, String> {
    if s.len() == 40 && s.chars().all(|c| c.is_ascii_hexdigit()) {
        Ok(s)
    } else {
        Err(format!("{s}: not an object name"))
    }
}

/// A subprocess rather than a library: an HTTP client, TLS stack and hashing
/// crate for a step that runs once when a lock file is written. `tokio::process`,
/// so nothing here holds a thread.
async fn git(args: &[&str]) -> Result<String, String> {
    let out = tokio::process::Command::new("git")
        .args(args)
        .output()
        .await
        .map_err(|e| format!("git: {e}"))?;
    if out.status.success() {
        String::from_utf8(out.stdout).map_err(|e| format!("git: {e}"))
    } else {
        Err(format!(
            "git {}: {}",
            args.join(" "),
            String::from_utf8_lossy(&out.stderr).trim()
        ))
    }
}

/// `('host modules)`: the impure half of the module system.
///
///   `('fetch url ref)`    `(rev path)` - the revision `ref` names and
///                         where under the root its source now is
///
/// The only thing in the tree that reaches the network. Registered only by
/// `naj --lock`, so an ordinary program cannot reach it: what a program may do
/// is what the world it runs in registered. Resolution never comes here - it
/// reads the store through [`files`] - which is what lets a program run
/// offline.
///
/// A revision is a content address, so a store entry never changes and an
/// existing one is answered without touching the network. `.git` is removed
/// afterwards, since nix populates the same layout with a plain source tree.
pub async fn modules(root: String, mut inbox: Receiver<Val>, back: Sender<Event>) {
    while let Some(msg) = inbox.recv().await {
        let Some((from, id, ask)) = request(&msg) else {
            continue;
        };
        let v = match list_elements(&ask).as_deref() {
            Some([verb, url, want]) if *verb == sym("fetch") => {
                match fetch(&root, &text(url), &text(want)).await {
                    Ok((at, where_)) => list(&[Str::val(at), Str::val(where_)]),
                    Err(why) => throw(pair(sym("fetch"), Str::val(why))),
                }
            }
            _ => throw(pair(sym("bad-request"), ask)),
        };
        if back.send(Event::Post(from, reply(id, v))).await.is_err() {
            return;
        }
    }
}

/// Resolving is a separate step from fetching because it decides the directory:
/// naming the entry after the revision is what makes it immutable, and a
/// `--branch` clone would leave the name to a ref that can move underneath it.
async fn fetch(root: &str, url: &str, want: &str) -> Result<(String, String), String> {
    let path = store_path(url)?;
    let at = match rev(want) {
        Ok(known) => known.to_string(),
        Err(_) => {
            let line = git(&["ls-remote", url, want]).await?;
            let head = line
                .split_whitespace()
                .next()
                .ok_or(format!("{url}: no ref {want}"))?;
            rev(head)?.to_string()
        }
    };

    let dest = format!("{root}/{path}/{at}");
    if tokio::fs::metadata(&dest).await.is_ok() {
        return Ok((at, path.to_string()));
    }

    tokio::fs::create_dir_all(&dest)
        .await
        .map_err(|e| format!("{dest}: {e}"))?;
    git(&["init", "-q", &dest]).await?;
    git(&["-C", &dest, "fetch", "-q", "--depth", "1", url, &at]).await?;
    git(&["-C", &dest, "checkout", "-q", "FETCH_HEAD"]).await?;
    tokio::fs::remove_dir_all(format!("{dest}/.git"))
        .await
        .map_err(|e| format!("{dest}/.git: {e}"))?;
    Ok((at, path.to_string()))
}

/// `('host clock)`: nothing in the floor can read a clock and nothing in a task
/// can wait for anything but a message.
///
///   `('now)`          milliseconds since the epoch
///   `('sleep ms)`     `'ok`, that much later
///   `('after ms to msg)`  `msg` is posted to `to`, that much later
///
/// `after` is a cast, since the task that arms a timer is usually not the one
/// that should hear it fire. It is what a timeout is built from: the dispatch
/// loop arms one carrying the reply it wants, and the duplicate-reply path
/// drops whichever loses (`naj/prelude.naj`).
///
/// Each timer is a task that sleeps and then sends, so nothing is polled. It
/// holds a port, which keeps a world with a timer outstanding from concluding
/// that nothing could wake it.
pub async fn clock(mut inbox: Receiver<Val>, back: Sender<Event>) {
    while let Some(msg) = inbox.recv().await {
        let ask = body(&msg);
        // A sleep is a timer whose message is the reply, so it is armed and
        // left: an adapter that waited would answer nobody until it was over.
        if let Some([verb, Val::Num(ms)]) = list_elements(&ask).as_deref() {
            if *verb == sym("sleep") {
                if let Some((from, id, _)) = request(&msg) {
                    let (ms, port) = (*ms.max(&0) as u64, back.clone());
                    tokio::spawn(async move {
                        tokio::time::sleep(std::time::Duration::from_millis(ms)).await;
                        let _ = port.send(Event::Post(from, reply(id, sym("ok")))).await;
                    });
                }
                continue;
            }
        }
        let v = match list_elements(&ask).as_deref() {
            Some([verb]) if *verb == sym("now") => Val::Num(now_ms()),
            Some([verb, Val::Num(ms), to, note]) if *verb == sym("after") => {
                let (ms, to, note) = (*ms.max(&0) as u64, to.clone(), note.clone());
                let port = back.clone();
                tokio::spawn(async move {
                    tokio::time::sleep(std::time::Duration::from_millis(ms)).await;
                    let _ = port.send(Event::Post(to, note)).await;
                });
                sym("ok")
            }
            _ => throw(pair(sym("bad-request"), ask)),
        };
        if !answer(&back, &msg, v).await {
            return;
        }
    }
}

fn now_ms() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0)
}

fn emit(s: &str) -> Val {
    use std::io::Write;
    let mut out = std::io::stdout();
    match out.write_all(s.as_bytes()).and_then(|()| out.flush()) {
        Ok(()) => sym("ok"),
        Err(e) => throw(pair(sym("io"), Str::val(e.to_string()))),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::floor::list;

    #[test]
    fn a_url_becomes_its_place_in_the_store() {
        assert_eq!(
            store_path("https://github.com/octocat/Hello-World"),
            Ok("github.com/octocat/Hello-World")
        );
        assert_eq!(
            store_path("https://git.avery.garden/thorn/narju"),
            Ok("git.avery.garden/thorn/narju")
        );
    }

    #[test]
    fn a_url_that_would_leave_the_store_is_refused() {
        for bad in [
            "https://forge/../etc/passwd",
            "https://forge/-x/y",
            "https://forge/a/.ssh",
            "http://forge/a/b",
            "git@github.com:a/b",
            "file:///etc",
            "https://forge",
            "https://forge//b",
        ] {
            assert!(store_path(bad).is_err(), "{bad}");
        }
    }
    fn clocked() -> (Sender<Val>, Receiver<Event>) {
        let (to, inbox) = tokio::sync::mpsc::channel(8);
        let (back, out) = tokio::sync::mpsc::channel(8);
        tokio::spawn(clock(inbox, back));
        (to, out)
    }

    fn ask(id: i64, body: Val) -> Val {
        list(&[sym("req"), sym("caller"), Val::Num(id), body])
    }

    async fn posted(out: &mut Receiver<Event>) -> (Val, Val) {
        match out.recv().await {
            Some(Event::Post(to, msg)) => (to, msg),
            _ => panic!("nothing was posted"),
        }
    }

    #[tokio::test]
    async fn the_clock_says_what_time_it_is() {
        let (to, mut out) = clocked();
        to.send(ask(1, list(&[sym("now")]))).await.expect("sent");
        let (_, answer) = posted(&mut out).await;
        match list_elements(&answer).as_deref() {
            Some([tag, id, Val::Num(ms)]) if *tag == sym("reply") => {
                assert_eq!(*id, Val::Num(1));
                assert!(*ms > 1_700_000_000_000, "{ms} is not a time");
            }
            _ => panic!("not a reply: {answer}"),
        }
    }

    #[tokio::test]
    async fn a_timer_posts_the_message_it_was_given_where_it_was_told() {
        let (to, mut out) = clocked();
        let note = list(&[
            sym("reply"),
            Val::Num(7),
            pair(sym("throw"), sym("timeout")),
        ]);
        to.send(list(&[
            sym("after"),
            Val::Num(1),
            sym("elsewhere"),
            note.clone(),
        ]))
        .await
        .expect("sent");
        assert_eq!(posted(&mut out).await, (sym("elsewhere"), note));
    }

    #[tokio::test]
    async fn a_sleep_does_not_stop_the_clock_answering_anyone_else() {
        let (to, mut out) = clocked();
        to.send(ask(1, list(&[sym("sleep"), Val::Num(80)])))
            .await
            .expect("sent");
        to.send(ask(2, list(&[sym("sleep"), Val::Num(1)])))
            .await
            .expect("sent");
        assert_eq!(
            posted(&mut out).await,
            (sym("caller"), reply(Val::Num(2), sym("ok")))
        );
        assert_eq!(
            posted(&mut out).await,
            (sym("caller"), reply(Val::Num(1), sym("ok")))
        );
    }

    #[tokio::test]
    async fn a_verb_the_clock_does_not_know_is_refused() {
        let (to, mut out) = clocked();
        to.send(ask(1, list(&[sym("nonsense")])))
            .await
            .expect("sent");
        assert_eq!(
            posted(&mut out).await,
            (
                sym("caller"),
                reply(
                    Val::Num(1),
                    throw(pair(sym("bad-request"), list(&[sym("nonsense")])))
                )
            )
        );
    }

    #[tokio::test]
    async fn a_call_that_runs_out_of_time_raises_and_leaves_the_loop_usable() {
        let mut t = crate::tower::Tower::load().expect("loads");
        let mut entry = |src: &str| {
            let forms = crate::floor::read::read(src).expect("reads");
            t.task(&crate::surface::desugar_body(&forms).expect("desugars"))
                .expect("no fault")
        };

        let driver = entry(
            "(task-loop self
               (lambda h (st msg)
                 (let ((deaf (spawn (lambda (me)
                                      (task-loop me
                                                 (lambda h2 (s2 m2)
                                                   (if (eq? s2 'asked) (cons 'stop 'ok) 'asked))
                                                 0)))))
                   (let ((late (attempt (lambda () (call-within 20 deaf '(hello))))))
                     (let ((now (call-within 10000 '(host clock) '(now))))
                       (begin (send deaf 'bye)
                              (cons 'stop (list late now)))))))
               0)",
        );

        let mut w = crate::world::World::new(50_000_000);
        let d = w.heap().spawn(driver);
        w.heap().post(&crate::sched::addr(d), sym("go"));

        let (tx, rx) = tokio::sync::mpsc::channel(8);
        w.serve("clock", tx);
        tokio::spawn(clock(rx, w.port()));

        let ended = w.run().await.expect("no bug");
        let Some((id, Val::Pair(tag, answered))) = ended.iter().find(|(id, _)| *id == d) else {
            panic!("the driver did not end: {ended:?}")
        };
        assert_eq!((*id, &**tag), (d, &sym("ok")));
        match list_elements(answered).as_deref() {
            Some([late, Val::Num(_)]) => {
                assert_eq!(*late, pair(sym("throw"), sym("timeout")));
            }
            _ => panic!("not what the driver answered: {answered}"),
        }
    }

    fn written() -> (Sender<Val>, Receiver<Event>) {
        let (to, inbox) = tokio::sync::mpsc::channel(8);
        let (back, out) = tokio::sync::mpsc::channel(8);
        tokio::spawn(disk(inbox, back));
        (to, out)
    }

    #[tokio::test]
    async fn a_write_puts_the_text_where_it_was_told() {
        let path = std::env::temp_dir().join("narju-disk-writes.txt");
        let (to, mut out) = written();
        to.send(ask(
            1,
            list(&[
                sym("write"),
                Str::val(path.to_string_lossy().into_owned()),
                Str::val("hi\n"),
            ]),
        ))
        .await
        .expect("sent");
        let (_, answer) = posted(&mut out).await;
        assert_eq!(answer, list(&[sym("reply"), Val::Num(1), sym("ok")]));
        assert_eq!(std::fs::read_to_string(&path).expect("written"), "hi\n");
        std::fs::remove_file(&path).expect("removed");
    }

    #[tokio::test]
    async fn a_write_makes_no_directory_to_write_into() {
        let dir = std::env::temp_dir().join("narju-disk-absent");
        let (to, mut out) = written();
        to.send(ask(
            2,
            list(&[
                sym("write"),
                Str::val(dir.join("f.txt").to_string_lossy().into_owned()),
                Str::val(""),
            ]),
        ))
        .await
        .expect("sent");
        let (_, answer) = posted(&mut out).await;
        match list_elements(&answer).as_deref() {
            Some([_, _, Val::Pair(tag, why)]) => {
                assert_eq!(**tag, sym("throw"));
                assert!(
                    matches!(&**why, Val::Pair(t, _) if **t == sym("io")),
                    "{why}"
                );
            }
            _ => panic!("not a refusal: {answer}"),
        }
        assert!(!dir.exists(), "the directory was made");
    }
}