// NDJSON (Newline-Delimited JSON, a.k.a. JSON Lines / `.ndjson` /
// `.jsonl`) serialisation for `DataFrame`. The wire shape is one flat
// JSON object per line (`{...}\n{...}\n...`) — the streaming-friendly
// sibling of the single-array "records" layout.
//
// Everything *after* the framing is shared with the JSON-records reader:
// per-line objects are collected into the same `Array[Json]` and handed
// to `frame_from_json_records`, so header collection (first-seen order
// across all rows, sparse rows contribute their own columns), per-column
// type inference (`Int → Float → Bool → String`, nulls passing through),
// and typed-column construction behave identically. The writer reuses
// `scalar_to_json`, so the same cell conventions hold — notably a
// non-finite `Float` (`NaN` / `±Infinity`) is emitted as `null` to keep
// each line valid JSON, and an `Int` keeps its dtype and round-trips exactly
// across the full `Int64` range — it is written as verbatim decimal text and
// recovered from the number's preserved `repr` (the shared core's
// large-`Int64` caveat), even though the `@json` number model is `Double`.
//
// Reading is lenient about line endings: lines are split on `\n`, a
// trailing `\r` (CRLF input) is tolerated as JSON whitespace by
// `@json.parse`, and blank / whitespace-only lines are skipped — so the
// writer's trailing newline (and any incidental blank lines) round-trip
// without producing phantom rows.

///|
/// Parse an NDJSON string (one JSON object per line) into a `DataFrame`.
///
/// Pipeline:
///   1. Split `content` on `\n`. Each non-blank line is parsed with
///      `@json.parse`; a malformed line surfaces as
///      `ParseError(Message("line N: ..."))` (1-based). Blank / whitespace-only
///      lines are skipped, so the writer's trailing newline — and any
///      incidental blank lines — never produce phantom records.
///   2. The parsed values are handed to `frame_from_json_records`,
///      which validates each is an object, collects headers in
///      first-seen order across all records (sparse records → null
///      cells), infers one dtype per column, and materialises the
///      frame under per-column rules shared with `parse_json_str`.
///
/// Empty input — or input that is entirely blank lines — yields a 0×0
/// frame, matching `parse_json_str`'s empty-array behaviour.
///
/// Raises:
/// - `ParseError(Message(...))` — a malformed line or a line whose value is
///   not a JSON object.
/// - `ParseError(Cell(...))` — a non-null cell does not fit its inferred
///   dtype, unless `options.on_parse_error` is `Null`, which downgrades that
///   cell to a null instead.
/// - Whatever `DataFrame::from_parts` propagates.
///
/// `projection` narrows the read to the named columns — the same knob the
/// lazy `scan_ndjson` source's projection pushdown rides. `None` (the
/// default) builds every column; `Some(names)` builds only the columns whose
/// header name appears in `names`, in first-seen header order. Every line is
/// still parsed and record validation runs before the column filter, so a
/// projected read fails on a malformed or non-object line exactly as a full
/// read does; the one behavioural difference is the point of the knob — a
/// parse error confined to a *dropped* column is not raised, that column is
/// never built.
pub fn parse_ndjson_str(
  content : String,
  options~ : JsonReadOptions = JsonReadOptions::JsonReadOptions(),
  projection? : Array[String],
) -> @frame.DataFrame raise @types.DataError {
  parse_ndjson_str_projected(content, options, projection, None)
}

///|
/// Shared core behind `parse_ndjson_str` and the NDJSON scan driver's read: the full
/// line-split / per-line parse pipeline `parse_ndjson_str` runs, with
/// one extra knob — `projection`. `None` builds every column; `Some(names)`
/// builds only the named columns (in first-seen header order), the
/// projection-pushdown path the lazy `scan_ndjson` source uses. Every line is
/// still parsed (streaming is deferred), and record validation runs before the
/// column filter — so a projected read fails on a malformed or non-object line
/// exactly as a full read does; only a parse error confined to a dropped column
/// goes unraised — that column is never built.
fn parse_ndjson_str_projected(
  content : String,
  options : JsonReadOptions,
  projection : Array[String]?,
  pruner : RowPruner?,
) -> @frame.DataFrame raise @types.DataError {
  // Strip a leading UTF-8 BOM (shared with the CSV / JSON readers) before the
  // line split, so a BOM on the first record line doesn't make `@json.parse`
  // reject it at column 0 while later lines parse — an asymmetric failure on
  // Excel-saved files.
  let content = strip_utf8_bom(content)
  // Normalise CR-family line endings to `\n` before the split: a lone
  // `\r` is a record terminator for classic-Mac exports, and without
  // this the whole file reads as one line and fails at its first `{`.
  // CRLF needs no help of its own (`@json.parse` takes the trailing
  // `\r` as whitespace), but folding it here keeps the physical line
  // count the diagnostics carry identical across the three endings.
  let content = content
    .replace_all(old="\r\n", new="\n")
    .replace_all(old="\r", new="\n")
  // Materialise the line views so the loop can carry a 1-based index
  // for diagnostics (the same `for i, x in ...` idiom the other readers
  // use). The views borrow `content`, so this is cheap.
  let lines = content.split("\n").to_array()
  let records : Array[Json] = []
  // Parallel to `records`: the 1-based physical file line each kept record came
  // from, so a build-stage type error reports the true line despite the blank
  // lines skipped between records (which shift the record index off the line).
  let record_lines : Array[Int32] = []
  for i, line in lines {
    // Skip blank / whitespace-only lines: the trailing `\n` the writer
    // emits leaves an empty final segment, and hand-written NDJSON may
    // carry stray blank lines. `@json.parse` tolerates a trailing `\r`
    // (CRLF) and surrounding spaces as JSON whitespace, so a real
    // record needs no trimming before parsing.
    if line.is_blank() {
      continue
    }
    // The strict duplicate-key reading rides the parser itself; off by
    // default it costs nothing (see `parse_json_str`).
    let json = @json.parse(
      line,
      reject_duplicate_keys=options.reject_duplicate_keys,
    ) catch {
      // Each record parses alone, so the parser's line is always 1 — the
      // locator's physical line is the only line worth naming (see
      // `duplicate_key_detail`).
      DuplicateKey(pos, key) =>
        raise @types.DataError::ParseError(
          @types.ParseErrorDetail::Message(
            "\{locator_text(@types.CellParseLocation::Line, i + 1)}: \{duplicate_key_detail(key, pos.line, pos.column, single_line=true)}",
          ),
        )
      err =>
        raise @types.DataError::ParseError(
          @types.ParseErrorDetail::Message(
            "\{locator_text(@types.CellParseLocation::Line, i + 1)}: invalid JSON: \{err}",
          ),
        )
    }
    records.push(json)
    record_lines.push(i + 1)
  }
  frame_from_json_records(
    records,
    options.infer_schema_rows,
    options.on_parse_error,
    projection,
    Some(record_lines),
    pruner?,
    allow_nonfinite=options.allow_nonfinite_floats,
    dtypes=options.dtypes,
  )
}

///|
/// Read an NDJSON file. `options` defaults to `JsonReadOptions::JsonReadOptions()`, so
/// `read_ndjson(path)` is the default-options read. Filesystem errors surface
/// as `raise IoError(message)`.
pub fn read_ndjson(
  path : String,
  options~ : JsonReadOptions = JsonReadOptions::JsonReadOptions(),
) -> @frame.DataFrame raise @types.DataError {
  let content = read_text(path)
  parse_ndjson_str(content, options~)
}

///|
/// Render a `DataFrame` as an NDJSON string: one flat JSON object per
/// row, each terminated by `\n` (including the last, matching the CSV
/// writer's per-row LF). A 0-row frame
/// renders the empty string.
///
/// Per-cell conventions are shared with `format_json` via
/// `scalar_to_json`: `Null → null`; `Int` → JSON number (no fractional part;
/// exact across the full `Int64` range via the preserved `repr`); finite
/// `Float` → JSON number, non-finite (`NaN` /
/// `±Infinity`) → `null` so each line stays valid JSON; `Bool` →
/// `true` / `false`; `String` → escaped JSON string; `Date` /
/// `Timestamp` / `Decimal` → JSON string holding the value-form rendering.
/// Keys appear in
/// `df.columns()` order, preserved by the linked-hash-map `Map` backing
/// each record object.
pub fn format_ndjson(df : @frame.DataFrame) -> String {
  // Share the column-major record walk with `format_json` /
  // `format_vega_lite`; NDJSON differs only in
  // the framing — one object per line, each terminated by `\n`.
  let buf = StringBuilder()
  for obj in df_to_json_records(df) {
    buf.write_string(obj.stringify())
    buf.write_char('\n')
  }
  buf.to_string()
}

///|
/// Write a `DataFrame` to an NDJSON file. Filesystem errors surface as
/// `raise IoError(message)`. A string cell (or column name) holding an
/// unpaired UTF-16 surrogate is refused with `raise InvalidOperation` —
/// the UTF-8 file encoding would swallow the following code unit and
/// corrupt the line framing.
pub fn write_ndjson(
  path : String,
  df : @frame.DataFrame,
) -> Unit raise @types.DataError {
  let content = format_ndjson(df)
  write_text(path, content)
}