// 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. Polars'
// `read_ndjson` / `write_ndjson` and pandas' `read_json(lines=True)` /
// `to_json(lines=True)` speak the same format.
//
// 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)
  // 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[Int] = []
  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, one line at a time so the diagnostic
    // can name the physical line (see `parse_json_str` for the why-ahead-of-
    // parse). NDJSON objects are top-level per line, so the object number in
    // the message is that line's own 1-based object count.
    if options.reject_duplicate_keys {
      match first_duplicate_json_key(line) {
        Some((n, key)) =>
          raise @types.DataError::ParseError(
            @types.ParseErrorDetail::Message(
              "\{locator_text(@types.CellParseLocation::Line, i + 1)}: duplicate key \"\{key}\" in JSON object \{n}",
            ),
          )
        None => ()
      }
    }
    let json = @json.parse(line) catch {
      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?,
  )
}

///|
/// 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 and Polars' `write_ndjson`). 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. 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)
}