// The JSON-records shared core: everything the JSON family of readers and
// emitters agree on, so the formats cannot drift apart. Three consumers
// meet here — the one-top-level-array records reader, the line-delimited
// reader (one object per line), and the `chart` package (a Vega-Lite
// `data.values` array) — all driving the same records → frame core
// (`frame_from_json_records`) and the same cell mapping (`scalar_to_json`
// / `df_to_json_records`). Per-column type inference mirrors the CSV
// reader — Int → Float → Bool → String, with nulls passing through
// transparently — so the round-trip `read_csv → write_json → read_json`
// preserves dtypes for most frames whose JSON values stay within the four
// concrete dtypes.
//
// Caveat 1 — integral-valued `Float`. JSON has a single number type and
// `Double::to_string` drops the fractional part of a whole value
// (`2.0` → `"2"`), so a `Float` column whose cells are all integral is
// re-inferred as `Int` on read-back (`json_value_as_int` accepts any
// `Number` equal to its truncation). The dtype narrows `Float → Int`;
// the values themselves are unchanged. This is inherent to JSON's
// typeless numbers: the CSV reader keeps `"2.0"` as `Float` only because the
// *text* still carries the decimal point, which a JSON number does not. It is
// the first of the three ways the two readers part company — the other two are
// the caveats below.
//
// Caveat 2 — large `Int64`. JSON numbers (and `@json`'s reader) are
// IEEE-754 `Double`, so an `Int64` whose magnitude exceeds 2^53 has no exact
// `Double` value. To keep it lossless, `scalar_to_json` writes every `Int`
// with its verbatim decimal text (`Json::number(.., repr=..)`), and
// `json_value_as_int` recovers the exact `Int64` from that text on read-back
// (`@json.parse` parks an out-of-`Double`-range integer as `±Infinity` but
// preserves its digits in `Number.repr`). So a full-range `Int64` round-trips
// exactly, dtype and value intact — only an external JSON integer beyond
// `Int64`'s own range (|v| ≥ 2^63) is unrepresentable and infers as `Float`.
// When such a large integer lands in a `Float` column instead (a sibling cell
// is fractional, so inference picks `Float`), `json_value_as_float` likewise
// recovers the nearest finite `Double` from the preserved digits rather than
// corrupting the cell to `Infinity`.
//
// Caveat 3 — all-non-finite / all-null `Float`. `scalar_to_json` maps every
// `NaN` / `±Infinity` to JSON `null` (no JSON literal exists for them), so a
// `Float` column that is *entirely* non-finite and/or null writes as all
// `null`. On read-back, an all-null column carries no dtype signal and infers
// as `String` (the all-null default), narrowing `Float → String`. This is
// inherent to JSON's lack of a `NaN` literal plus schemaless inference; a
// column with at least one finite value keeps its `Float` dtype. (Same applies
// to NDJSON, which shares `scalar_to_json` and the inference core.)
//
// All parsing and stringification goes through the builtin `@json`
// package so we get standards-conformant string escaping, the
// `NaN` / `Infinity` rendering convention, and number formatting for
// free.
//
// One-file-one-test exemption: this shared core has no dedicated
// test file of its own. (doc-guard: unresolved) It is covered through the
// JSON / NDJSON / chart blackbox tests that drive it.

///|
/// Shared "records → `DataFrame`" core for the JSON-records and NDJSON
/// readers. Both produce an `Array[Json]` of candidate records — the
/// array reader from one top-level `[...]`, the NDJSON reader from one
/// object per line — and from there the pipeline is identical:
///   1. Validate every element is a `Json::Object`, pulling the inner
///      `Map` out for downstream passes (so inference / build don't
///      have to re-match).
///   2. Collect headers in first-seen order across **all** records, so
///      a sparse record (missing some keys) still contributes its own
///      columns and gets nulls for the missing fields.
///   3. Infer each column's dtype over the first `infer_schema_rows`
///      rows in priority order `Int → Float → Bool → String` (a null cell
///      within them is skipped but still counts toward the window);
///      the first dtype that accepts every probed cell wins, and an
///      all-null column defaults to `String` (same convention as the
///      CSV reader — no information to disambiguate).
///   4. Materialise each column into a typed `Series` and wrap through
///      `DataFrame::from_parts`, under the record count as the height.
///
/// An empty record set collapses to a 0×0 frame — the same shape both
/// readers' empty-input paths produce. Records that are all empty objects
/// (`[{}, {}, ...]`) instead keep their count as the `N×0` frame: they are
/// rows, they just have no fields. A cell that doesn't
/// fit its column's inferred dtype is handled per `on_parse_error`
/// (raise `ParseError(Cell(...))`, or downgrade the cell to a null), shared
/// by both
/// readers so JSON-array and NDJSON input agree on the policy.
///
/// `projection` narrows which columns are built: `None` builds every column
/// (the eager `read_*` behaviour), `Some(names)` builds only the columns whose
/// header name appears in `names`, in first-seen header order, leaving the rest
/// uninferred and unbuilt — the projection-pushdown path the lazy `scan_ndjson`
/// source uses. Record validation and header collection run *before* the
/// filter, so a projected read still fails on a non-object record exactly as a
/// full read does; only a parse error confined to a dropped column goes
/// unraised, because that column is never built.
///
/// `source_lines` (parallel to `records`) lets a line-oriented caller number
/// build-stage type errors by physical file line: `Some(lines)` reports
/// `line N` (the NDJSON reader, whose skipped blank lines shift the record
/// index off the line), `None` reports `record N` (the JSON-array reader,
/// whose records have no line of their own).
fn frame_from_json_records(
  records : Array[Json],
  infer_schema_rows : Int,
  on_parse_error : OnParseError,
  projection : Array[String]?,
  source_lines : Array[Int]?,
  pruner? : RowPruner,
) -> @frame.DataFrame raise @types.DataError {
  if records.is_empty() {
    return @frame.DataFrame::DataFrame([])
  }
  let objs : Array[Map[String, Json]] = Array(capacity=records.length())
  let headers : Array[String] = []
  let header_set : Map[String, Unit] = Map([])
  for i, rec in records {
    match rec {
      Object(obj) => {
        // `@json.parse` already collapsed any duplicate keys within this one
        // object (last value wins) before we see the `Map`, so `{"a":1,"a":2}`
        // reads as a single `a = 2` with no diagnostic — there is no per-object
        // `DuplicateColumn` signal (cross-record headers are collected first-seen).
        for k, _ in obj {
          if !header_set.contains(k) {
            header_set[k] = ()
            headers.push(k)
          }
        }
        objs.push(obj)
      }
      _ => {
        // Carry the locator, like every other error in these readers: in a
        // large file "which record" is the whole diagnostic. A 1-based
        // **file line** when the caller supplies a per-record line map (the
        // NDJSON reader, whose blank-line skipping makes the record index
        // diverge from the physical line), or a 1-based **record** number for
        // the JSON-array reader (where records have no line of their own).
        let (loc, n) = json_cell_parse_location(source_lines, i)
        raise @types.DataError::ParseError(
          @types.ParseErrorDetail::Message(
            "\{locator_text(loc, n)}: expected each record to be a JSON object",
          ),
        )
      }
    }
  }
  // Build one typed column per header, skipping any the projection drops; the
  // surviving columns keep first-seen header order, so a projected read is
  // column-for-column a sub-frame of the full read. Records that carry no
  // fields at all (`[{}, {}]`) leave no header to build from and read as the
  // `N×0` frame — the records are still records.
  // Inference always walks every record, even when a predicate will drop some,
  // so a pruned read locks in the dtypes an eager read-then-filter would.
  let build = fn(
    name : String,
    from : Array[Map[String, Json]],
    lines : Array[Int]?,
  ) -> @series.Series raise @types.DataError {
    let dt = infer_json_column_dtype(objs, name, infer_schema_rows)
    build_json_column(name, dt, from, on_parse_error, lines)
  }
  // Build one column over the records `keep` selects (`None` = every record),
  // mapping each surviving record's original line through `keep` so a parse
  // error in a kept record still reports its true file line. Inference runs
  // over every record either way. A named column the file lacks is not
  // built, so a pruner's callback raises
  // `ColumnNotFound` exactly as the eager filter would.
  let build_over = fn(
    name : String,
    _col : Int,
    keep : Array[Int]?,
  ) -> @series.Series raise @types.DataError {
    let records = kept_records(keep, objs)
    let lines = source_lines.map(l => kept_records(keep, l))
    build(name, records, lines)
  }
  assemble_pruned_frame(headers, objs.length(), projection, pruner, build_over)
}

///|
/// Materialise `df` as an array of flat JSON record objects — one
/// `Json::object` per row, keys in column-declaration order, every cell
/// mapped through `scalar_to_json` (so nulls and non-finite floats land on
/// JSON `null`). Shared by the records-shaped emitters: `format_json`
/// wraps the array in one top-level `[...]`, `format_ndjson` stringifies
/// one object per line, and the chart package's `format_vega_lite` inlines
/// it as a Vega-Lite `data.values` array — so the cell conventions stay
/// identical across all of them without re-implementing the row walk.
/// That last consumer is cross-package, which is why this is the one
/// symbol here that leaves the package: an `#internal(engine)` seam
/// (hidden from the interface, flagged for downstream callers), keeping
/// the cell mapping single-sourced instead of copied into `chart`.
#doc(hidden)
#internal(engine, "MoonFrame chart export shares the JSON records cell conventions")
pub fn df_to_json_records(df : @frame.DataFrame) -> Array[Json] {
  let names = df.columns()
  let nrows = df.nrows()
  // Materialise the whole frame once (column-major) through the shared
  // `to_scalar_matrix` so a record is assembled by plain `cols[c][r]` reads
  // rather than per-cell `item`.
  let cols = df.to_scalar_matrix()
  Array::makei(nrows, r => {
    let obj : Map[String, Json] = Map([])
    for c, name in names {
      obj[name] = scalar_to_json(cols[c][r])
    }
    Json::object(obj)
  })
}

///|
/// Map a `@types.Scalar` into the corresponding `Json` value. `Int` is
/// written as a JSON number carrying its verbatim decimal text as the
/// number's `repr` (`Json::number(v.to_double(), repr=v.to_string())`), so the
/// emitted token is the exact integer even when its magnitude exceeds 2^53 and
/// the `Double` value alone would round — `json_value_as_int` recovers the
/// exact `Int64` from that text on read-back. A finite `Float` becomes a JSON
/// number. A
/// non-finite `Float` (`NaN` / `±Infinity`) has no JSON literal —
/// `@json.stringify` would emit the bare tokens `NaN` / `Infinity` /
/// `-Infinity`, which is invalid JSON that neither
/// `parse_json_str` nor a standards-compliant parser can read
/// back — so it is mapped to `null` (matching pandas' `to_json`),
/// keeping the emitted text valid JSON. `Bool` lands on the `true` /
/// `false` variants; `String` passes through and is escaped by the
/// stringifier; `Null` lands on JSON `null`.
fn scalar_to_json(s : @types.Scalar) -> Json {
  match s {
    @types.Scalar::Null => Json::null()
    @types.Scalar::Int(v) => Json::number(v.to_double(), repr=v.to_string())
    @types.Scalar::Float(v) =>
      if v.is_nan() || v.is_inf() {
        Json::null()
      } else if v == 0.0 && v.reinterpret_as_int64() != 0L {
        // A repr-less number renders through `Double`'s Show, which drops
        // the sign of negative zero — the token `0` would read back as
        // `+0.0` (and `1.0 / x` flip from `-inf` to `+inf`). Pin the exact
        // token instead; the reader's `parse_float_opt("-0.0")` keeps the
        // sign.
        Json::number(v, repr="-0.0")
      } else {
        Json::number(v)
      }
    @types.Scalar::Bool(v) => Json::boolean(v)
    @types.Scalar::String(v) => Json::string(v)
    // A date leaves as its ISO spelling (the same rendering the CSV writer
    // and `Scalar::to_string` use), so the two interchange formats agree.
    @types.Scalar::Date(_) => Json::string(s.to_string())
  }
}

///|
/// Fetch the value of `name` from `obj`. A missing field on a sparse
/// record is treated as a null cell — every record was already
/// validated as a JSON object by `frame_from_json_records` (the shared entry
/// both the JSON and NDJSON readers pass through) before
/// reaching this helper, so the lookup boils down to a single map
/// access.
fn lookup_field(obj : Map[String, Json], name : String) -> Json {
  match obj.get(name) {
    Some(v) => v
    None => Json::null()
  }
}

///|
/// Walk the first `limit` records of `name` and pick the narrowest
/// dtype that accepts every non-null cell. Priority order matches
/// the CSV reader: `Int → Float → Bool → String`. A column whose
/// probe window is entirely null falls back to `String` (no signal).
/// A column that mixes dtypes (e.g. `1` then `"hi"`) also lands on
/// `String`, where `build_json_column` re-renders numeric cells via their
/// string form. Adapts the JSON cell source (`Json` values looked up by
/// key) onto the shared `infer_dtype` core.
fn infer_json_column_dtype(
  objs : Array[Map[String, Json]],
  name : String,
  limit : Int,
) -> InferredDtype {
  infer_dtype(
    objs.length(),
    limit~,
    cell=i => lookup_field(objs[i], name),
    is_null=j => j is Null,
    is_int=j => json_value_as_int(j) is Some(_),
    is_float=j => json_value_as_float(j) is Some(_),
    is_bool=j => json_value_as_bool(j) is Some(_),
  )
}

///|
/// Try to read `j` as an `Int64`. For a finite `Number` the check is
/// value-based: accept it when it is exactly integer-valued
/// (`n == n.trunc()`) and within `Int64` range (checked by
/// `double_fits_int64`, since `Int64::MAX` isn't exactly representable
/// as a `Double`), so a token like `1.0e2` lands as `Int(100)`, matching
/// pandas / polars on equivalent CSV. A `Number` whose magnitude overflowed
/// the `Double` is recovered from its preserved `repr` digits — the
/// large-`Int64` round-trip rule, stated in full once as the file header's
/// Caveat 2. Booleans, strings, and null are intentionally rejected — JSON
/// has strong types and silently promoting `true → 1` would surprise
/// callers used to pandas / polars semantics.
fn json_value_as_int(j : Json) -> Int64? {
  match j {
    // The `repr`-digit recovery is the file header's Caveat 2; a present
    // `repr` means an overflowed integer, recovered exactly from its text.
    Number(_, repr=Some(text)) => parse_int_opt(text)
    Number(n, repr=None) =>
      if n != n.trunc() {
        None
      } else if !@numeric.double_fits_int64(n) {
        None
      } else {
        Some(n.to_int64())
      }
    _ => None
  }
}

///|
/// Try to read `j` as a `Double`. Accepts any `Number`. Booleans,
/// strings, and null are rejected for the same reason as
/// `json_value_as_int`.
fn json_value_as_float(j : Json) -> Double? {
  match j {
    // `@json.parse` parks a `Double`-overflowing integer as `±Infinity` but
    // keeps its exact digits in `repr`; re-parsing those digits recovers the
    // nearest finite `Double` — as `parse_float_opt` does — rather
    // than the lossy Infinity. `parse_float_opt` routes through that same
    // `@string.parse_double` (and accepts the non-finite forms), so a magnitude
    // that overflows `Double` itself folds back to the original Infinity `n`. A
    // finite `Number` (`repr=None`) takes its value directly.
    Number(n, repr=Some(text)) => Some(parse_float_opt(text, true).unwrap_or(n))
    Number(n, repr=None) => Some(n)
    _ => None
  }
}

///|
/// Try to read `j` as a `Bool`. Only the JSON `true` / `false`
/// variants are accepted; `Number(1)` / `Number(0)` are deliberately
/// **not** treated as booleans, matching `parse_csv_str`'s
/// `parse_bool_opt` rule (numeric `0` / `1` round-trip as `Int`,
/// which the inference order picks up first).
fn json_value_as_bool(j : Json) -> Bool? {
  match j {
    True => Some(true)
    False => Some(false)
    _ => None
  }
}

///|
/// Render any JSON value as a flat string for the `String`-fallback
/// column builder. Numbers render via `Double::to_string` (a whole-valued
/// number with no preserved `repr` loses its fractional part — `2.0` → `"2"`,
/// the String-fallback analogue of the module's integral-`Float` Caveat 1),
/// booleans render `true` / `false`, strings pass through verbatim, and any
/// composite (`Array` / `Object`) falls through to `@json.stringify`.
/// `Null` is filtered out upstream — the shared String-column builder
/// yields `None` for a null cell without invoking this helper —
/// so the catch-all only fires on composites in practice.
fn json_value_to_string_fallback(j : Json) -> String {
  match j {
    True => "true"
    False => "false"
    // Prefer the preserved `repr` so a `Double`-overflowing integer renders as
    // its exact digits rather than "Infinity" when it lands in a String column.
    Number(n, repr~) =>
      match repr {
        Some(text) => text
        None => n.to_string()
      }
    String(s) => s
    _ => (j : Json).stringify()
  }
}

///|
/// Preserve the source coordinate as data for a typed-cell parse error.
fn json_cell_parse_location(
  source_lines : Array[Int]?,
  i : Int,
) -> (@types.CellParseLocation, Int) {
  match source_lines {
    Some(lines) => (@types.CellParseLocation::Line, lines[i])
    None => (@types.CellParseLocation::Record, i + 1)
  }
}

///|
/// Materialise one column from the record array using the dtype
/// chosen by inference. Mirrors the CSV reader's `build_column`: each
/// branch walks the records in row order and converts the cell into
/// the typed array slot; a value past the inference window that fails
/// to match the locked-in dtype is handled per `on_parse_error`
/// (`Raise` → `ParseError(Cell(...))`, `Null` → a null cell).
fn build_json_column(
  name : String,
  dtype : InferredDtype,
  objs : Array[Map[String, Json]],
  on_parse_error : OnParseError,
  source_lines : Array[Int]?,
) -> @series.Series raise @types.DataError {
  let n = objs.length()
  let cell = fn(i : Int) -> Json { lookup_field(objs[i], name) }
  let isnull = fn(c : Json) -> Bool { c is Null }
  build_inferred_column(
    name,
    dtype,
    n,
    cell~,
    is_null=isnull,
    parse_int=json_value_as_int,
    parse_float=json_value_as_float,
    parse_bool=json_value_as_bool,
    render=json_value_to_string_fallback,
    locate=(dt, i, c) => {
      let (location, position) = json_cell_parse_location(source_lines, i)
      @types.DataError::ParseError(
        @types.ParseErrorDetail::Cell(
          location,
          name,
          position,
          dt,
          json_value_to_string_fallback(c),
        ),
      )
    },
    on_parse_error~,
  )
}