kobold_csv/

parse.rs

//! `KOBOLD.CSV.PARSE.1` + `KOBOLD.CSV.ROUNDTRIP.1` (FAIL-CLOSED) -- parse a COMPACT extract back into
//! record bytes, and prove the round trip.
//!
//! [`parse_into`] takes a `Compact` (header + value rows) CSV produced against `copybook` and re-encodes
//! each field value into its declared PIC bytes, returning one reconstructed record per row. It FAILS CLOSED
//! (a [`Finding`], never a truncation/coercion) on:
//!
//! * a value too long for an alphanumeric field (`VALUE_OVERFLOW`),
//! * a non-numeric value into a numeric field (`NUMERIC_INVALID`), too many integer/fraction digits, a sign
//!   on an unsigned field,
//! * a header whose column names do not match the copybook's leaf field names (`HEADER_MISMATCH`),
//! * a data row whose column count differs from the header (`COLUMN_COUNT`),
//! * a malformed CSV line (the dialect reader's own findings bubble up).
//!
//! [`roundtrip`] proves records -> Compact CSV -> [`parse_into`] -> identical bytes, reporting per-record
//! honestly when a value-only compact extract legitimately does NOT round-trip (e.g. a numeric field whose
//! stored zoned form differs from its canonical render, so re-encoding yields the canonical -- not the
//! original -- bytes).
//!
//! This module is independent of GnuCOBOL/libcob.

use crate::dialect::{parse_row, Dialect};
use crate::export::{export, Mode};
use crate::model::{Copybook, FieldDecl, FieldKind, Finding};

/// `KOBOLD.CSV.PARSE.1` -- reconstruct record bytes from a Compact CSV `csv_text` against `copybook`.
///
/// Returns `Ok(records)` (one `Vec<u8>` of `copybook.record_length()` bytes per data row), or
/// `Err(findings)` listing every problem across the whole input. Fail-closed: on ANY finding no bytes are
/// returned for the offending row's class of error -- the result is all-or-nothing.
pub fn parse_into(
    copybook: &Copybook,
    csv_text: &[u8],
    dialect: &Dialect,
) -> Result<Vec<Vec<u8>>, Vec<Finding>> {
    let leaves = copybook.leaf_fields();
    let mut findings: Vec<Finding> = Vec::new();

    // Split into logical lines on LF; the dialect reader tolerates a trailing CR per line. NOTE: a Compact
    // extract never embeds newlines inside a field value (compact values are decoded, trimmed display text),
    // so a plain LF split is sound here. (Audit/Evidence are not re-parsed by this court.)
    let lines: Vec<&[u8]> = split_lines(csv_text);
    if lines.is_empty() {
        return Err(vec![Finding::new("CSV_EMPTY", "no rows (expected a header row)".to_string())]);
    }

    // Header row.
    let header = match parse_row(lines[0], dialect) {
        Ok(h) => h,
        Err(f) => return Err(vec![f]),
    };
    let expected: Vec<&str> = leaves.iter().map(|f| f.name.as_str()).collect();
    if header.len() != expected.len() || header.iter().zip(&expected).any(|(h, e)| h != e) {
        findings.push(Finding::new(
            "HEADER_MISMATCH",
            format!("header {:?} does not match copybook leaf fields {:?}", header, expected),
        ));
        // Header mismatch is fatal: we cannot trust column->field alignment.
        return Err(findings);
    }

    let total = copybook.record_length();
    let mut records: Vec<Vec<u8>> = Vec::new();

    for (row_idx, line) in lines.iter().enumerate().skip(1) {
        let row = match parse_row(line, dialect) {
            Ok(r) => r,
            Err(mut f) => {
                f.message = format!("row {}: {}", row_idx, f.message);
                findings.push(f);
                continue;
            }
        };
        if row.len() != leaves.len() {
            findings.push(Finding::new(
                "COLUMN_COUNT",
                format!("row {}: {} columns, expected {}", row_idx, row.len(), leaves.len()),
            ));
            continue;
        }
        let mut out = vec![b' '; total];
        let before = findings.len();
        for (f, val) in leaves.iter().zip(&row) {
            encode_leaf(f, val, &mut out, &mut findings, row_idx);
        }
        if findings.len() == before {
            records.push(out);
        }
    }

    if findings.is_empty() {
        Ok(records)
    } else {
        Err(findings)
    }
}

/// Split `text` into logical lines on LF, dropping a single trailing empty line (the terminator after the
/// last row). Each returned slice excludes the LF; a trailing CR is left for the dialect reader to strip.
fn split_lines(text: &[u8]) -> Vec<&[u8]> {
    let mut lines = Vec::new();
    let mut start = 0;
    for i in 0..text.len() {
        if text[i] == 0x0a {
            lines.push(&text[start..i]);
            start = i + 1;
        }
    }
    if start < text.len() {
        lines.push(&text[start..]);
    }
    lines
}

/// Encode one leaf field value into `out` at its offset, fail-closed.
fn encode_leaf(d: &FieldDecl, value: &str, out: &mut [u8], findings: &mut Vec<Finding>, row: usize) {
    match &d.kind {
        FieldKind::Alphanumeric => encode_alnum(d, value, out, findings, row),
        FieldKind::Numeric { scale, signed } => {
            encode_numeric(d, value, *scale, *signed, out, findings, row)
        }
        FieldKind::Group(_) => {} // leaves never include groups
    }
}

/// Place `bytes` at absolute `offset`, fail-closed if it would not fit.
fn place(out: &mut [u8], offset: usize, bytes: &[u8], findings: &mut Vec<Finding>, name: &str, row: usize) {
    let end = offset + bytes.len();
    if end > out.len() {
        findings.push(Finding::new(
            "FIELD_OUT_OF_RANGE",
            format!(
                "row {}: field {}: writing [{}..{}] exceeds record length {}",
                row, name, offset, end, out.len()
            ),
        ));
        return;
    }
    out[offset..end].copy_from_slice(bytes);
}

/// Encode an alphanumeric value: pad with trailing spaces to the field length; fail closed if longer.
fn encode_alnum(d: &FieldDecl, value: &str, out: &mut [u8], findings: &mut Vec<Finding>, row: usize) {
    let mut bytes = Vec::with_capacity(value.len());
    for ch in value.chars() {
        let cp = ch as u32;
        if cp > 0xff {
            findings.push(Finding::new(
                "ALNUM_NON_BYTE",
                format!("row {}: field {}: char U+{:04X} not representable in one byte", row, d.name, cp),
            ));
            return;
        }
        bytes.push(cp as u8);
    }
    if bytes.len() > d.length {
        findings.push(Finding::new(
            "VALUE_OVERFLOW",
            format!(
                "row {}: field {}: value of {} bytes overflows field length {} (fail-closed, no truncation)",
                row,
                d.name,
                bytes.len(),
                d.length
            ),
        ));
        return;
    }
    let mut buf = vec![b' '; d.length];
    buf[..bytes.len()].copy_from_slice(&bytes);
    place(out, d.offset, &buf, findings, &d.name, row);
}

/// Encode a numeric value into zoned-decimal display digits per `scale`/`signed`. Fail-closed on a
/// non-numeric value, too many integer/fraction digits, or a sign on an unsigned field.
fn encode_numeric(
    d: &FieldDecl,
    value: &str,
    scale: usize,
    signed: bool,
    out: &mut [u8],
    findings: &mut Vec<Finding>,
    row: usize,
) {
    let mut s = value.trim();
    let mut negative = false;
    if let Some(rest) = s.strip_prefix('-') {
        negative = true;
        s = rest;
    } else if let Some(rest) = s.strip_prefix('+') {
        s = rest;
    }
    if negative && !signed {
        findings.push(Finding::new(
            "SIGN_ON_UNSIGNED",
            format!("row {}: field {}: negative value into unsigned PIC {}", row, d.name, d.pic),
        ));
        return;
    }

    let (int_str, frac_str) = match s.split_once('.') {
        Some((i, f)) => (i, f),
        None => (s, ""),
    };
    if int_str.is_empty() && frac_str.is_empty() {
        findings.push(Finding::new(
            "NUMERIC_EMPTY",
            format!("row {}: field {}: empty numeric value", row, d.name),
        ));
        return;
    }
    for (label, part) in [("integer", int_str), ("fraction", frac_str)] {
        if !part.chars().all(|c| c.is_ascii_digit()) {
            findings.push(Finding::new(
                "NUMERIC_INVALID",
                format!("row {}: field {}: non-numeric {} part {:?} (fail-closed)", row, d.name, label, part),
            ));
            return;
        }
    }
    if frac_str.len() > scale {
        findings.push(Finding::new(
            "FRACTION_OVERFLOW",
            format!(
                "row {}: field {}: {} fraction digits exceed scale {} (fail-closed, no rounding)",
                row,
                d.name,
                frac_str.len(),
                scale
            ),
        ));
        return;
    }

    let int_digits = d.length.saturating_sub(scale);
    let int_trimmed = int_str.trim_start_matches('0');
    if int_trimmed.len() > int_digits {
        findings.push(Finding::new(
            "VALUE_OVERFLOW",
            format!(
                "row {}: field {}: integer part {:?} needs {} digits, field has {} (fail-closed)",
                row,
                d.name,
                int_str,
                int_trimmed.len(),
                int_digits
            ),
        ));
        return;
    }

    let mut digits = String::with_capacity(d.length);
    for _ in 0..(int_digits - int_trimmed.len()) {
        digits.push('0');
    }
    digits.push_str(int_trimmed);
    digits.push_str(frac_str);
    for _ in 0..(scale - frac_str.len()) {
        digits.push('0');
    }

    let mut bytes: Vec<u8> = digits.into_bytes();
    if bytes.len() != d.length {
        findings.push(Finding::new(
            "NUMERIC_LENGTH",
            format!("row {}: field {}: built {} digits, declared length {}", row, d.name, bytes.len(), d.length),
        ));
        return;
    }
    if signed {
        if let Some(last) = bytes.last_mut() {
            *last = overpunch_byte(*last, negative);
        }
    }
    place(out, d.offset, &bytes, findings, &d.name, row);
}

/// Map an ASCII digit byte + sign to its zoned overpunch byte (the inverse of export's `overpunch`).
fn overpunch_byte(digit: u8, negative: bool) -> u8 {
    let n = digit.wrapping_sub(b'0');
    if n > 9 {
        return digit;
    }
    match (negative, n) {
        (false, 0) => b'{',
        (false, k) => b'A' + (k - 1),
        (true, 0) => b'}',
        (true, k) => b'J' + (k - 1),
    }
}

/// The per-record outcome of a [`roundtrip`] proof.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RoundtripRecord {
    /// The zero-based record index.
    pub index: usize,
    /// True iff `parse_into` reproduced this record's exact original bytes.
    pub identical: bool,
    /// The original record bytes.
    pub original: Vec<u8>,
    /// The bytes reconstructed from the Compact extract (empty if the row failed closed).
    pub reconstructed: Vec<u8>,
    /// Findings explaining a non-roundtrip (re-encode failures, or the canonical-form note below).
    pub findings: Vec<Finding>,
}

/// The full result of a [`roundtrip`] proof over a record set.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RoundtripReport {
    /// Per-record outcomes, in order.
    pub records: Vec<RoundtripRecord>,
}

impl RoundtripReport {
    /// True iff EVERY record reproduced identically through the Compact extract.
    pub fn all_identical(&self) -> bool {
        self.records.iter().all(|r| r.identical)
    }
}

/// `KOBOLD.CSV.ROUNDTRIP.1` -- prove records -> Compact CSV -> [`parse_into`] -> identical bytes.
///
/// A Compact extract carries only the DECODED value (not the raw zoned form). For alphanumeric fields and
/// for numeric fields already in canonical form this round-trips exactly. When a numeric field's STORED form
/// differs from its canonical render (e.g. an unsigned field whose bytes carried a stray sign overpunch, or
/// leading-space padding), the re-encode yields the canonical bytes -- legitimately NOT the original -- and
/// that record is reported `identical: false` with a `NON_CANONICAL_STORAGE` finding rather than pretending.
pub fn roundtrip(copybook: &Copybook, records: &[&[u8]], dialect: &Dialect) -> RoundtripReport {
    let csv = export(copybook, records, Mode::Compact, dialect);
    let parsed = parse_into(copybook, csv.as_bytes(), dialect);

    let mut out = Vec::with_capacity(records.len());
    match parsed {
        Ok(recs) => {
            for (i, orig) in records.iter().enumerate() {
                let recon = recs.get(i).cloned().unwrap_or_default();
                let identical = recon.as_slice() == *orig;
                let mut findings = Vec::new();
                if !identical {
                    findings.push(Finding::new(
                        "NON_CANONICAL_STORAGE",
                        format!(
                            "record {}: compact extract round-trips to canonical bytes which differ from the \
                             stored form (a value-only extract cannot preserve a non-canonical zoned/padded \
                             representation)",
                            i
                        ),
                    ));
                }
                out.push(RoundtripRecord {
                    index: i,
                    identical,
                    original: orig.to_vec(),
                    reconstructed: recon,
                    findings,
                });
            }
        }
        Err(findings) => {
            // A re-encode failed closed somewhere; attribute the whole batch's findings to each record so the
            // caller sees the failure without us guessing per-row attribution beyond the messages themselves.
            for (i, orig) in records.iter().enumerate() {
                out.push(RoundtripRecord {
                    index: i,
                    identical: false,
                    original: orig.to_vec(),
                    reconstructed: Vec::new(),
                    findings: findings.clone(),
                });
            }
        }
    }

    RoundtripReport { records: out }
}

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

    fn signed_copybook() -> Copybook {
        Copybook {
            record_name: "CUST".into(),
            encoding: "ascii".into(),
            fields: vec![
                FieldDecl::alnum("NAME", "X(4)", 0, 4),
                FieldDecl::numeric("AMT", "S9(3)V99", 4, 5, 2, true),
            ],
        }
    }

    fn unsigned_copybook() -> Copybook {
        Copybook {
            record_name: "R".into(),
            encoding: "ascii".into(),
            fields: vec![
                FieldDecl::alnum("NAME", "X(4)", 0, 4),
                FieldDecl::numeric("AMT", "9(3)V99", 4, 5, 2, false),
            ],
        }
    }

    #[test]
    fn compact_roundtrip_identical_bytes() {
        // KOBOLD.CSV.ROUNDTRIP.1: canonical records survive bytes -> Compact CSV -> parse_into -> bytes.
        let cb = signed_copybook();
        // AMT = -12.50 zoned: "0125" + overpunch of '0' negative = '}' -> "0125}"
        let recs: Vec<&[u8]> = vec![b"JOHN0125}", b"JANE0007A"]; // 'A' = +1 overpunch on last digit
        let report = roundtrip(&cb, &recs, &Dialect::csv());
        assert!(report.all_identical(), "report: {:?}", report);
    }

    #[test]
    fn parse_into_reconstructs() {
        let cb = unsigned_copybook();
        let csv = b"NAME,AMT\nAL,12.50\n";
        let recs = parse_into(&cb, csv, &Dialect::csv()).expect("parse");
        assert_eq!(recs.len(), 1);
        assert_eq!(&recs[0], b"AL  01250");
    }

    #[test]
    fn fail_closed_overflow_value() {
        let cb = unsigned_copybook();
        let csv = b"NAME,AMT\nTOOLONG,12.50\n"; // NAME too long for X(4)
        let findings = parse_into(&cb, csv, &Dialect::csv()).expect_err("must fail closed");
        assert!(findings.iter().any(|f| f.code == "VALUE_OVERFLOW"));
    }

    #[test]
    fn fail_closed_nonnumeric() {
        let cb = unsigned_copybook();
        let csv = b"NAME,AMT\nAL,1X.50\n"; // non-numeric integer part
        let findings = parse_into(&cb, csv, &Dialect::csv()).expect_err("must fail closed");
        assert!(findings.iter().any(|f| f.code == "NUMERIC_INVALID"));
    }

    #[test]
    fn fail_closed_header_mismatch() {
        let cb = unsigned_copybook();
        let csv = b"NAME,BALANCE\nAL,12.50\n"; // wrong second column name
        let findings = parse_into(&cb, csv, &Dialect::csv()).expect_err("must fail closed");
        assert_eq!(findings[0].code, "HEADER_MISMATCH");
    }

    #[test]
    fn fail_closed_wrong_column_count() {
        let cb = unsigned_copybook();
        let csv = b"NAME,AMT\nAL,12.50,EXTRA\n";
        let findings = parse_into(&cb, csv, &Dialect::csv()).expect_err("must fail closed");
        assert!(findings.iter().any(|f| f.code == "COLUMN_COUNT"));
    }

    #[test]
    fn non_canonical_storage_reported_honestly() {
        let cb = unsigned_copybook();
        // AMT stored as "  099" (leading spaces) renders to "0.99" but re-encodes to "00099": not identical.
        let recs: Vec<&[u8]> = vec![b"AL    099"];
        let report = roundtrip(&cb, &recs, &Dialect::csv());
        assert!(!report.all_identical());
        assert_eq!(report.records[0].findings[0].code, "NON_CANONICAL_STORAGE");
    }
}