JSON Lines

use ndjson.qweruse ndjson_parse, ndjson_write from ndjson      // Quillo 0.6: only these

serde_json = { version = "1", features = ["preserve_order"] }

// An app's request log kept as JSON Lines: read a page at a time, searched, appended to and exported.////   quillo run ndjson.qwe      then open http://localhost:8080/page?offset=15use "ndjson.qwer"// One page of the log, 10 lines at a time; the broken line is reported, not fatal.route GET /page ? { offset: int = 0 } {    let page = ndjson_read_file("ndjson_sample.jsonl", offset, 10)?    { first: page.rows[0].at, count: len(page.rows), next: page.next, done: page.done, bad: page.bad }}// The errors one user met, without loading the whole file.route GET /errors = ndjson_find("ndjson_sample.jsonl", { level: "error", "user.id": 101 }, 5)? | map((r) => { at: r.at, path: r.path })// A log that only grows, and an export made from it.route GET /export {    write_file("data/events.jsonl", "")    ndjson_append_file("data/events.jsonl", [{ event: "signup", user: "ada" }])?    ndjson_append_file("data/events.jsonl", [{ event: "order", user: "ada", total: 42.5 }, { event: "signup", user: "grace" }])?    let text = ndjson_write(ndjson_parse(read_file("data/events.jsonl"))? | filter((e) => e.event == "signup"))?    { lines: ndjson_count("data/events.jsonl")?, signups: text }}

//! JSON Lines (NDJSON): one value per line, read a page at a time from files of any size, appended to, filtered and written, for logs, exports and imports.//! version: 1.0.0//!//! [dependencies]//! serde_json = { version = "1", features = ["preserve_order"] }use serde_json::{json, Value};use std::io::{BufRead, BufReader, Write};/// A line longer than this (16 MiB) is an error rather than a value.const MAX_LINE: usize = 16 * 1024 * 1024;fn open(path: &str) -> Result<BufReader<std::fs::File>, String> {    std::fs::File::open(path).map(BufReader::new).map_err(|e| format!("{}: {}", path, e))}/// Each line of the reader in turn: (line number from 1, text), blank lines skipped.fn lines(r: impl BufRead) -> impl Iterator<Item = Result<(i64, String), String>> {    r.lines().enumerate().filter_map(|(i, l)| match l {        Ok(l) if l.trim().is_empty() => None,        Ok(l) if l.len() > MAX_LINE => Some(Err(format!("line {} is over 16 MiB", i + 1))),        Ok(l) => Some(Ok((i as i64 + 1, l))),        Err(e) => Some(Err(format!("line {}: {}", i + 1, e))),    })}fn value(n: i64, line: &str) -> Result<Value, String> {    serde_json::from_str(line.trim_start_matches('\u{feff}')).map_err(|e| format!("line {}: {}", n, e))}fn one_line(v: &Value) -> Result<String, String> {    serde_json::to_string(v).map_err(|e| e.to_string())}/// The values of NDJSON text, one per line (blank lines skipped). A line that is not JSON is an error naming it.pub fn ndjson_parse(text: &str) -> Result<Vec<Value>, String> {    lines(text.as_bytes()).map(|l| l.and_then(|(n, s)| value(n, &s))).collect()}/// Values as NDJSON text: each on one line, with a line break after the last.pub fn ndjson_write(rows: Vec<serde_json::Value>) -> Result<String, String> {    let mut out = String::new();    for r in &rows {        out.push_str(&one_line(r)?);        out.push('\n');    }    Ok(out)}/// One page of a file, read line by line so the file's size does not matter: `{ rows, next, done, bad }`./// Skips `offset` values, takes up to `limit` (1 to 100,000); `next` is the offset of the next page./// Lines that are not JSON do not stop it: each is in `bad` as `{ line, error }`, and counts toward the offset.pub fn ndjson_read_file(path: &str, offset: i64, limit: i64) -> Result<Value, String> {    let limit = limit.clamp(1, 100_000) as usize;    let offset = offset.max(0) as usize;    let mut rows = vec![];    let mut bad = vec![];    let mut seen = 0usize;    let mut done = true;    for l in lines(open(path)?) {        let (n, s) = l?;        if seen < offset {            seen += 1;            continue;        }        if rows.len() + bad.len() == limit {            done = false;            break;        }        seen += 1;        match value(n, &s) {            Ok(v) => rows.push(v),            Err(e) => bad.push(json!({ "line": n, "error": e })),        }    }    Ok(json!({ "rows": rows, "next": seen, "done": done, "bad": bad }))}/// Adds `rows` at the end of the file (made, with its folders, when missing), one per line: the number of lines added./// A log that only grows: nothing before is read or written again.pub fn ndjson_append_file(path: &str, rows: Vec<serde_json::Value>) -> Result<i64, String> {    if let Some(dir) = std::path::Path::new(path).parent().filter(|d| !d.as_os_str().is_empty()) {        std::fs::create_dir_all(dir).map_err(|e| format!("{}: {}", dir.display(), e))?;    }    let mut f = std::fs::OpenOptions::new().create(true).append(true).open(path).map_err(|e| format!("{}: {}", path, e))?;    // A file that does not end in a line break gets one first, so the new row starts a line.    let len = f.metadata().map(|m| m.len()).unwrap_or(0);    if len > 0 {        let mut last = [0u8; 1];        use std::io::{Read, Seek, SeekFrom};        let mut r = std::fs::File::open(path).map_err(|e| e.to_string())?;        r.seek(SeekFrom::Start(len - 1)).map_err(|e| e.to_string())?;        r.read_exact(&mut last).map_err(|e| e.to_string())?;        if last[0] != b'\n' {            f.write_all(b"\n").map_err(|e| e.to_string())?;        }    }    let mut buf = String::new();    for r in &rows {        buf.push_str(&one_line(r)?);        buf.push('\n');    }    f.write_all(buf.as_bytes()).map_err(|e| format!("{}: {}", path, e))?;    Ok(rows.len() as i64)}/// Writes `rows` as a new NDJSON file at `path` (replacing it; folders made): the number of lines.pub fn ndjson_write_file(path: &str, rows: Vec<serde_json::Value>) -> Result<i64, String> {    if let Some(dir) = std::path::Path::new(path).parent().filter(|d| !d.as_os_str().is_empty()) {        std::fs::create_dir_all(dir).map_err(|e| format!("{}: {}", dir.display(), e))?;    }    let f = std::fs::File::create(path).map_err(|e| format!("{}: {}", path, e))?;    let mut w = std::io::BufWriter::new(f);    for r in &rows {        w.write_all(one_line(r)?.as_bytes()).map_err(|e| e.to_string())?;        w.write_all(b"\n").map_err(|e| e.to_string())?;    }    w.flush().map_err(|e| e.to_string())?;    Ok(rows.len() as i64)}/// How many values the file holds (blank lines not counted), read line by line.pub fn ndjson_count(path: &str) -> Result<i64, String> {    let mut n = 0;    for l in lines(open(path)?) {        l?;        n += 1;    }    Ok(n)}fn get<'a>(v: &'a Value, path: &str) -> Option<&'a Value> {    path.split('.').try_fold(v, |v, k| match v {        Value::Object(m) => m.get(k),        Value::Array(a) => k.parse::<usize>().ok().and_then(|i| a.get(i)),        _ => None,    })}/// The values of the file whose fields equal those in `filter` (`{ level: "error", "user.id": 42 }`: dots go into maps),/// up to `limit` (1 to 100,000), read line by line; lines that are not JSON are passed over.pub fn ndjson_find(path: &str, filter: std::collections::HashMap<String, serde_json::Value>, limit: i64) -> Result<Vec<Value>, String> {    let limit = limit.clamp(1, 100_000) as usize;    let mut out = vec![];    for l in lines(open(path)?) {        let (n, s) = l?;        let Ok(v) = value(n, &s) else { continue };        if filter.iter().all(|(k, want)| get(&v, k) == Some(want)) {            out.push(v);            if out.len() == limit {                break;            }        }    }    Ok(out)}