flow-mumu 0.2.0-rc.9

// src/read.rs
//
// flow:read — produce a zero-arg transform that yields **fixed-size byte chunks** from a file.
//
// New signature (chunked I/O):
//   flow:read(chunk_size_bytes, "path/to/file")
//   - `chunk_size_bytes` (int, > 0) is required (can be supplied via partials/placeholders).
//   - Each call pulls up to `chunk_size_bytes` from the file (raw bytes),
//     converted to a String using UTF-8 *lossy* decoding (replacement for invalid sequences).
//   - When EOF is reached, subsequent calls return Err("NO_MORE_DATA").
//
// Examples:
//   tf = flow:read(1024, "data.bin")   // zero-arg transform yielding ~1KB strings
//   flow:to_array(tf)                  // collects to a StrArray if fully drained
//
// Notes:
// - Strings are used for compatibility with existing array tooling.
//   If exact binary processing is required later, we can add a bytes variant.

use std::fs::File;
use std::io::{BufReader, Read};

use std::sync::{Arc, Mutex};

use mumu::parser::interpreter::{DynamicFnInfo, Interpreter};
use mumu::parser::types::{FunctionValue, Value};

use crate::common::is_placeholder;

/// Register `flow:read` with the interpreter.
pub fn register_flow_read(interp: &mut Interpreter) {
    let func = Arc::new(Mutex::new(flow_read_bridge));
    // Marked pure for parity; it performs I/O, but purity only informs heuristics.
    let info = DynamicFnInfo::new(func, true);
    interp.register_dynamic_function_ex("flow:read", info);

    interp.set_variable(
        "flow:read",
        Value::Function(Box::new(FunctionValue::Named("flow:read".to_string()))),
    );
}

/// Bridge entry supporting partials:
///   0 args                 => read(_, _)
///   1 arg (int)            => read(size, _)
///   1 arg (path)           => read(_, path)
///   2 args (size, path)    => finalize -> transform
fn flow_read_bridge(_interp: &mut Interpreter, args: Vec<Value>) -> Result<Value, String> {
    match args.len() {
        0 => Ok(make_partial(None, None)),
        1 => {
            let a = &args[0];
            if is_placeholder(a) {
                Ok(make_partial(None, None))
            } else if let Some(sz) = take_size_maybe(a)? {
                Ok(make_partial(Some(sz), None))
            } else {
                // treat as path
                Ok(make_partial(None, Some(a.clone())))
            }
        }
        2 => {
            let a_size = take_size_maybe(&args[0])?;
            let a_path = if is_placeholder(&args[1]) {
                None
            } else {
                Some(args[1].clone())
            };

            match (a_size, a_path) {
                (Some(sz), Some(pv)) => finalize_read(sz, pv),
                (sz, pv) => Ok(make_partial(sz, pv)),
            }
        }
        n => Err(format!(
            "flow:read => expected up to 2 arguments (size, path), got {}",
            n
        )),
    }
}

/// Partial wrapper so users may supply `_` and fill later.
fn make_partial(size_opt: Option<i32>, path_opt: Option<Value>) -> Value {
    use mumu::parser::types::FunctionValue::RustClosure;

    Value::Function(Box::new(RustClosure(
        "flow:read-partial".to_string(),
        Arc::new(Mutex::new(move |_interp: &mut Interpreter, new_args: Vec<Value>| {
            let mut s = size_opt;
            let mut p = path_opt.clone();

            for a in new_args {
                if s.is_none() {
                    if is_placeholder(&a) {
                        // keep None
                    } else if let Some(sz) = take_size_maybe(&a)? {
                        s = Some(sz);
                    } else {
                        // not a size => treat as path if slot empty
                        if p.is_none() {
                            p = Some(a);
                        } else {
                            return Err("flow:read => unexpected extra argument".to_string());
                        }
                    }
                    continue;
                }
                if p.is_none() {
                    if is_placeholder(&a) {
                        // remain None
                    } else {
                        p = Some(a);
                    }
                    continue;
                }
                return Err("flow:read => partial => too many arguments".to_string());
            }

            match (s, p.clone()) {
                (Some(sz), Some(pv)) => finalize_read(sz, pv),
                _ => Ok(make_partial(s, p)),
            }
        })),
        0,
    )))
}

/// Extract a positive chunk size if `v` is an Int/Long/Float convertible.
/// Returns Ok(Some(size)) when recognized, Ok(None) when not a size-like value.
fn take_size_maybe(v: &Value) -> Result<Option<i32>, String> {
    let size = match v {
        Value::Int(i) => *i,
        Value::Long(l) => {
            if *l < i64::from(i32::MIN) || *l > i64::from(i32::MAX) {
                return Err("flow:read => chunk size out of range for i32".to_string());
            }
            *l as i32
        }
        Value::Float(f) => *f as i32,
        _ => return Ok(None),
    };
    if size <= 0 {
        return Err(format!("flow:read => chunk size must be > 0, got {}", size));
    }
    Ok(Some(size))
}

/// Convert a Value into a filesystem path string.
fn take_path_from_value(v: Value) -> Result<String, String> {
    match v {
        Value::SingleString(s) => Ok(s),
        Value::StrArray(arr) if arr.len() == 1 => Ok(arr[0].clone()),
        other => Err(format!(
            "flow:read => filename must be a string (or single-element StrArray), got {:?}",
            other
        )),
    }
}

/// Finalize: open the file and return a transform that yields byte chunks.
fn finalize_read(chunk_size: i32, v: Value) -> Result<Value, String> {
    let path = take_path_from_value(v)?;
    let file =
        File::open(&path).map_err(|e| format!("flow:read => cannot open '{}': {}", path, e))?;
    let reader = BufReader::new(file);

    Ok(build_read_transform(path, chunk_size as usize, reader))
}

/// Shared state for the transform closure.
struct ReadEnv {
    /// Only used in verbose logs (path is cloned into env for diagnostics).
    path: String,
    /// Buffered reader over the file.
    reader: BufReader<File>,
    /// Whether EOF has been reached.
    done: bool,
    /// Fixed chunk size in bytes.
    chunk_size: usize,
    /// Reusable buffer.
    buf: Vec<u8>,
}

/// Build the zero-arg transform closure that yields the next chunk each call.
fn build_read_transform(path: String, chunk_size: usize, reader: BufReader<File>) -> Value {
    use mumu::parser::types::FunctionValue::RustClosure;

    let env = Arc::new(Mutex::new(ReadEnv {
        path,
        reader,
        done: false,
        chunk_size,
        buf: vec![0u8; chunk_size.max(1)],
    }));

    let clos = RustClosure(
        "flow:read-transform(bytes)".to_string(),
        Arc::new(Mutex::new(move |interp: &mut Interpreter, _args: Vec<Value>| {
            let mut guard = env
                .lock()
                .map_err(|_| "flow:read => lock error".to_string())?;

            if guard.done {
                if interp.is_verbose() {
                    eprintln!("[flow:read] '{}' => EOF (already done)", guard.path);
                }
                return Err("NO_MORE_DATA".to_string());
            }

            // Move the buffer out to avoid aliasing two mutable borrows of `guard`.
            let mut buf = std::mem::take(&mut guard.buf);
            if buf.len() < guard.chunk_size {
                buf.resize(guard.chunk_size, 0u8);
            }

            // Read up to chunk_size bytes
            let n = match guard.reader.read(&mut buf) {
                Ok(0) => {
                    guard.done = true;
                    // Put the buffer back before returning
                    guard.buf = buf;
                    if interp.is_verbose() {
                        eprintln!("[flow:read] '{}' => EOF", guard.path);
                    }
                    return Err("NO_MORE_DATA".to_string());
                }
                Ok(nread) => nread,
                Err(e) => {
                    // Put the buffer back before returning error
                    guard.buf = buf;
                    return Err(format!("flow:read => I/O error: {}", e));
                }
            };

            // Put the buffer back for reuse on next tick
            let s = String::from_utf8_lossy(&buf[..n]).to_string();
            guard.buf = buf;

            if interp.is_verbose() {
                eprintln!(
                    "[flow:read] '{}' => chunk {} bytes (requested {})",
                    guard.path, n, guard.chunk_size
                );
            }

            Ok(Value::SingleString(s))
        })),
        0,
    );

    Value::Function(Box::new(clos))
}