glazer

// vim:ts=2:sw=2:et
//-----------------------------------------------------------------------------
// CSV decoding/encoding (RFC 4180-style), producing/consuming native Erlang
// terms in a single pass.
//
// Decoding:
//   - Without `headers`: a list of rows, each row a list of binary fields.
//   - With `headers`: the first row becomes the column names, and each
//     subsequent row decodes to a map (or proplist) keyed by those names.
//
// Encoding accepts the same shapes: a list of rows (each row a list of
// binaries/atoms/integers/floats), or — with `headers` — a list of maps
// whose values are written out in the given column order.
//
// Quoting follows RFC 4180: a field is quoted if it contains the delimiter,
// a quote character, or a line break; embedded quotes are doubled.
//-----------------------------------------------------------------------------
#pragma once

#include <cctype>
#include <cmath>
#include <charconv>
#include <cstdint>
#include <optional>
#include <string>
#include <string_view>
#include <vector>

#include <erl_nif.h>

#include "fast_float.hpp"
#include "glazer_atoms.hpp"
#include "glazer_bigint.hpp"
#include "glazer_common.hpp"

namespace glz {

//-----------------------------------------------------------------------------
// Per-column field type conversion (`{fields, [Type, ...]}` decode option)
//-----------------------------------------------------------------------------

enum class CsvFieldType {
  none,          // leave as binary (default)
  integer,
  floatp,        // {float, Precision}
  boolean,
  datetime,      // {datetime, InputFormat}
  binary,
  charlist,
  existing_atom,
  atom_list,     // {atom, ExistingAtoms}
};

// What to do with a field that fails to convert to the requested type.
enum class CsvOnFailure {
  binary,         // leave the original binary (default)
  raise,          // fail the whole decode with {invalid_field_value, Row, Col}
  default_value,  // use CsvFieldSpec::default_term
  null,           // use the configured null term (am_null)
};

struct CsvFieldSpec {
  CsvFieldType type        = CsvFieldType::none;
  int          precision   = 0; // for floatp
  int64_t      scale       = 1; // for floatp
  std::string  format;          // for datetime (strptime-like)
  std::vector<std::string> atoms; // for {atom, ExistingAtoms}
  CsvOnFailure on_failure   = CsvOnFailure::binary;
  ERL_NIF_TERM default_term = 0;  // for empty fields, or on_failure == default_value
  bool         has_default  = false;
};

//-----------------------------------------------------------------------------
// Minimal strptime-like parser, used to turn a `{datetime, InputFormat}`
// field into Unix epoch seconds (UTC).
//
// Supported directives: %Y %y %m %d %H %M %S %f %z, and literal `%%`.
// Any other character in the format must match the input literally; a space
// in the format matches a run of one-or-more whitespace characters in the
// input (as with strptime).
//-----------------------------------------------------------------------------

namespace datetime {

// Days since 1970-01-01 for the given proleptic-Gregorian civil date.
// Howard Hinnant's `days_from_civil` algorithm (public domain).
inline int64_t days_from_civil(int64_t y, unsigned m, unsigned d)
{
  y -= m <= 2;
  const int64_t era = (y >= 0 ? y : y - 399) / 400;
  const unsigned yoe = static_cast<unsigned>(y - era * 400);          // [0, 399]
  const unsigned doy = (153 * (m + (m > 2 ? -3 : 9)) + 2) / 5 + d - 1; // [0, 365]
  const unsigned doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;          // [0, 146096]
  return era * 146097 + static_cast<int64_t>(doe) - 719468;
}

inline bool parse_uint(const char*& p, const char* end, int max_digits, int& out)
{
  const char* start = p;
  out = 0;
  while (p < end && (p - start) < max_digits && *p >= '0' && *p <= '9') {
    out = out * 10 + (*p - '0');
    ++p;
  }
  return p > start;
}

// Parses `input` according to `format` (a strptime-like format string) and
// returns the corresponding Unix epoch time in seconds (UTC), or
// `std::nullopt` if the input doesn't match the format.
// NOTE: we could use std::get_time(), but it's locale-dependent and doesn't
// support fractional seconds or timezone offsets, so we implement our own
inline std::optional<int64_t> parse(std::string_view input, std::string_view format)
{
  const char* p   = input.data();
  const char* end = p + input.size();
  const char* f   = format.data();
  const char* fe  = f + format.size();

  int year = 1970, month = 1, day = 1, hour = 0, minute = 0, second = 0;
  bool have_date = false;
  int tz_offset_sec = 0;

  while (f < fe) {
    char fc = *f;

    if (fc == '%' && f + 1 < fe) {
      char spec = f[1];
      f += 2;
      switch (spec) {
        case 'Y': {
          int v;
          if (!parse_uint(p, end, 4, v)) return std::nullopt;
          year = v; have_date = true;
          break;
        }
        case 'y': {
          int v;
          if (!parse_uint(p, end, 2, v)) return std::nullopt;
          year = (v <= 68) ? 2000 + v : 1900 + v; have_date = true;
          break;
        }
        case 'm': {
          int v;
          if (!parse_uint(p, end, 2, v) || v < 1 || v > 12) return std::nullopt;
          month = v; have_date = true;
          break;
        }
        case 'd': {
          int v;
          if (!parse_uint(p, end, 2, v) || v < 1 || v > 31) return std::nullopt;
          day = v; have_date = true;
          break;
        }
        case 'H': {
          int v;
          if (!parse_uint(p, end, 2, v) || v > 23) return std::nullopt;
          hour = v;
          break;
        }
        case 'M': {
          int v;
          if (!parse_uint(p, end, 2, v) || v > 59) return std::nullopt;
          minute = v;
          break;
        }
        case 'S': {
          int v;
          if (!parse_uint(p, end, 2, v) || v > 60) return std::nullopt;
          second = v;
          break;
        }
        case 'f': {
          // Fractional seconds — consume digits, discard.
          int v;
          if (!parse_uint(p, end, 9, v)) return std::nullopt;
          break;
        }
        case 'z': {
          if (p < end && (*p == 'Z' || *p == 'z')) { ++p; tz_offset_sec = 0; break; }
          if (p >= end || (*p != '+' && *p != '-'))
            return std::nullopt;
          auto neg = *p == '-';
          ++p;
          int hh, mm = 0;
          if (!parse_uint(p, end, 2, hh))
            return std::nullopt;
          if (p < end && *p == ':') ++p;
          if (p < end && *p >= '0' && *p <= '9' && !parse_uint(p, end, 2, mm))
            return std::nullopt;
          tz_offset_sec = (hh * 3600 + mm * 60) * (neg ? -1 : 1);
          break;
        }
        case '%':
          if (p >= end || *p != '%')
            return std::nullopt;
          ++p;
          break;
        default:
          return std::nullopt;
      }
      continue;
    }

    if (fc == ' ') {
      if (p >= end || !std::isspace(static_cast<uint8_t>(*p))) [[unlikely]]
        return std::nullopt;
      while (p < end && std::isspace(static_cast<uint8_t>(*p))) ++p;
      ++f;
      continue;
    }

    if (p >= end || *p != fc) [[unlikely]]
      return std::nullopt;
    ++p; ++f;
  }

  if (p != end)   return std::nullopt; // trailing input not consumed by format
  if (!have_date) return std::nullopt;

  auto days = days_from_civil(year, static_cast<unsigned>(month), static_cast<unsigned>(day));
  auto secs = days * 86400 + hour * 3600 + minute * 60 + second - tz_offset_sec;
  return secs;
}

} // namespace datetime

//-----------------------------------------------------------------------------
// Options
//-----------------------------------------------------------------------------

enum class CsvReturnKind { list, map, tuple };

struct CsvDecodeOpts {
  char delimiter           = ',';
  bool headers             = false;          // bare `headers` atom or {headers, ...}
  bool label_atom          = false;          // {headers, atom}
  bool label_existing_atom = false;          // {headers, existing_atom}
  bool label_charlist      = false;          // {headers, charlist}
  CsvReturnKind return_kind = CsvReturnKind::list; // {return, list | map | tuple}
  bool explicit_headers    = false;          // {headers, [List]}: no header row in data
  ERL_NIF_TERM null_term   = 0;
  std::vector<ERL_NIF_TERM> header_list;    // populated when explicit_headers is true
  std::vector<CsvFieldSpec> fields;
  size_t skip_rows         = 0;             // {skip, N}: skip first N data rows
  size_t limit             = 0;             // {limit, N}: max data rows (0 = unlimited)
};

// Parses a `field_type()`:
//   integer | {float, Precision} | boolean | {datetime, InputFormat}
//   | binary | charlist | existing_atom | {atom, ExistingAtoms :: list(atom())}
static bool parse_csv_field_type(ErlNifEnv* env, ERL_NIF_TERM term, CsvFieldSpec& spec)
{
  if (enif_is_identical(term, AM_INTEGER))       { spec.type = CsvFieldType::integer;       return true; }
  if (enif_is_identical(term, AM_BOOLEAN))       { spec.type = CsvFieldType::boolean;       return true; }
  if (enif_is_identical(term, AM_BINARY))        { spec.type = CsvFieldType::binary;        return true; }
  if (enif_is_identical(term, AM_CHARLIST))      { spec.type = CsvFieldType::charlist;      return true; }
  if (enif_is_identical(term, AM_EXISTING_ATOM)) { spec.type = CsvFieldType::existing_atom; return true; }

  int arity; const ERL_NIF_TERM* tp;
  if (!enif_get_tuple(env, term, &arity, &tp) || arity != 2) return false;

  if (enif_is_identical(tp[0], AM_FLOAT)) {
    int precision;
    if (!enif_get_int(env, tp[1], &precision) || precision < 0) return false;
    spec.type      = CsvFieldType::floatp;
    spec.precision = precision;
    spec.scale     = glz::power(10LL, static_cast<size_t>(precision));
    return true;
  }

  if (enif_is_identical(tp[0], AM_DATETIME)) {
    ErlNifBinary bin;
    if (!enif_inspect_binary(env, tp[1], &bin)) return false;
    spec.type   = CsvFieldType::datetime;
    spec.format.assign(reinterpret_cast<const char*>(bin.data), bin.size);
    return true;
  }

  if (enif_is_identical(tp[0], AM_ATOM)) {
    if (!enif_is_list(env, tp[1])) return false;
    std::vector<std::string> atoms;
    ERL_NIF_TERM ahead, atail = tp[1];
    char buf[256];
    while (enif_get_list_cell(env, atail, &ahead, &atail)) {
      std::string_view sv;
      if (!atom_to_sv(env, ahead, buf, sizeof(buf), sv)) return false;
      atoms.emplace_back(sv);
    }
    spec.type  = CsvFieldType::atom_list;
    spec.atoms = std::move(atoms);
    return true;
  }

  return false;
}

// Parses a single element of `{fields, [FieldType, ...]}`, where:
//   FieldType :: field_type()
//             | #{type => field_type(), default => Term, on_failure => binary | raise | default | null}
//   field_type() :: integer | {float, Precision} | boolean | {datetime, InputFormat}
//   | binary | charlist | existing_atom | {atom, ExistingAtoms :: list(atom())}
static bool parse_csv_field_spec(ErlNifEnv* env, ERL_NIF_TERM term, CsvFieldSpec& spec)
{
  if (!enif_is_map(env, term))
    return parse_csv_field_type(env, term, spec);

  ERL_NIF_TERM type_term;
  if (!enif_get_map_value(env, term, AM_TYPE, &type_term)) return false;
  if (!parse_csv_field_type(env, type_term, spec)) return false;

  ERL_NIF_TERM default_term;
  if (enif_get_map_value(env, term, AM_DEFAULT, &default_term)) {
    spec.default_term = default_term;
    spec.has_default  = true;
  }

  ERL_NIF_TERM on_failure_term;
  if (enif_get_map_value(env, term, AM_ON_FAILURE, &on_failure_term)) {
    if      (enif_is_identical(on_failure_term, AM_BINARY))  spec.on_failure = CsvOnFailure::binary;
    else if (enif_is_identical(on_failure_term, AM_RAISE))   spec.on_failure = CsvOnFailure::raise;
    else if (enif_is_identical(on_failure_term, AM_DEFAULT)) spec.on_failure = CsvOnFailure::default_value;
    else if (enif_is_identical(on_failure_term, AM_NULL))    spec.on_failure = CsvOnFailure::null;
    else return false;
  }

  return true;
}

static bool parse_csv_decode_opts(ErlNifEnv* env, ERL_NIF_TERM list, CsvDecodeOpts& opts)
{
  ERL_NIF_TERM head, tail = list;
  while (enif_get_list_cell(env, tail, &head, &tail)) {
    if (enif_is_identical(head, AM_HEADERS)) { opts.headers = true; continue; }

    int arity; const ERL_NIF_TERM* tp;
    if (!enif_get_tuple(env, head, &arity, &tp) || arity != 2) continue;

    if (enif_is_identical(tp[0], AM_DELIMITER)) {
      int cp;
      if (!enif_get_int(env, tp[1], &cp) || cp <= 0 || cp > 0x7F) return false;
      opts.delimiter = static_cast<char>(cp);
    } else if (enif_is_identical(tp[0], AM_NULL_TERM)) {
      if (!enif_is_atom(env, tp[1])) return false;
      opts.null_term = tp[1];
    } else if (enif_is_identical(tp[0], AM_RETURN)) {
      if      (enif_is_identical(tp[1], AM_MAP))   opts.return_kind = CsvReturnKind::map;
      else if (enif_is_identical(tp[1], AM_TUPLE)) opts.return_kind = CsvReturnKind::tuple;
      else if (enif_is_identical(tp[1], AM_LIST))  opts.return_kind = CsvReturnKind::list;
    } else if (enif_is_identical(tp[0], AM_HEADERS)) {
      // {headers, [List]}  — explicit column names, no header row in data
      // {headers, binary | string} — 1st row → binary keys
      // {headers, atom}            — 1st row → atom keys
      // {headers, existing_atom}   — 1st row → existing-atom keys
      // {headers, charlist}        — 1st row → charlist keys
      if (enif_is_list(env, tp[1])) {
        opts.headers          = true;
        opts.explicit_headers = true;
        ERL_NIF_TERM hhead, htail = tp[1];
        while (enif_get_list_cell(env, htail, &hhead, &htail))
          opts.header_list.push_back(hhead);
      } else if (enif_is_identical(tp[1], AM_BINARY) || enif_is_identical(tp[1], AM_STRING)) {
        opts.headers          = true;
        opts.label_atom          = false;
        opts.label_existing_atom = false;
        opts.label_charlist      = false;
      } else if (enif_is_identical(tp[1], AM_LABEL_ATOM)) {
        opts.headers          = true;
        opts.label_atom       = true;
      } else if (enif_is_identical(tp[1], AM_EXISTING_ATOM)) {
        opts.headers             = true;
        opts.label_existing_atom = true;
      } else if (enif_is_identical(tp[1], AM_CHARLIST)) {
        opts.headers         = true;
        opts.label_charlist  = true;
      }
    } else if (enif_is_identical(tp[0], AM_SKIP)) {
      // {skip, N} or {skip, {From, To}} (1-based, inclusive)
      ErlNifUInt64 n;
      if (enif_get_uint64(env, tp[1], &n)) {
        opts.skip_rows = static_cast<size_t>(n);
      } else {
        int arity2; const ERL_NIF_TERM* tp2;
        if (enif_get_tuple(env, tp[1], &arity2, &tp2) && arity2 == 2) {
          ErlNifUInt64 from, to;
          if (enif_get_uint64(env, tp2[0], &from) && enif_get_uint64(env, tp2[1], &to)
              && from >= 1 && to >= from) {
            opts.skip_rows = static_cast<size_t>(from - 1);
            opts.limit     = static_cast<size_t>(to - from + 1);
          }
        }
      }
    } else if (enif_is_identical(tp[0], AM_LIMIT)) {
      ErlNifUInt64 n;
      if (enif_get_uint64(env, tp[1], &n))
        opts.limit = static_cast<size_t>(n);
    } else if (enif_is_identical(tp[0], AM_FIELDS)) {
      if (!enif_is_list(env, tp[1])) return false;
      ERL_NIF_TERM fhead, ftail = tp[1];
      std::vector<CsvFieldSpec> fields;
      while (enif_get_list_cell(env, ftail, &fhead, &ftail)) {
        CsvFieldSpec spec;
        if (!parse_csv_field_spec(env, fhead, spec)) return false;
        fields.push_back(spec);
      }
      opts.fields = std::move(fields);
    }
  }
  return true;
}

struct CsvEncodeOpts {
  char        delimiter = ',';
  bool        headers   = false;
  bool        explicit_headers = false;       // {headers, [List]}: explicit column order
  std::vector<ERL_NIF_TERM> header_list;       // populated when explicit_headers is true
  std::string_view line_ending = "\r\n";
};

static bool parse_csv_encode_opts(ErlNifEnv* env, ERL_NIF_TERM list, CsvEncodeOpts& opts)
{
  ERL_NIF_TERM head, tail = list;
  while (enif_get_list_cell(env, tail, &head, &tail)) {
    if (enif_is_identical(head, AM_HEADERS)) { opts.headers = true; continue; }

    int arity; const ERL_NIF_TERM* tp;
    if (!enif_get_tuple(env, head, &arity, &tp) || arity != 2) continue;

    if (enif_is_identical(tp[0], AM_DELIMITER)) {
      int cp;
      if (!enif_get_int(env, tp[1], &cp) || cp <= 0 || cp > 0x7F) return false;
      opts.delimiter = static_cast<char>(cp);
    } else if (enif_is_identical(tp[0], AM_LINE_ENDING)) {
      if      (enif_is_identical(tp[1], AM_LF))   opts.line_ending = "\n";
      else if (enif_is_identical(tp[1], AM_CRLF)) opts.line_ending = "\r\n";
      else return false;
    } else if (enif_is_identical(tp[0], AM_HEADERS)) {
      // {headers, [Names]} — explicit column order/names for map-row encoding
      if (!enif_is_list(env, tp[1])) return false;
      opts.headers          = true;
      opts.explicit_headers = true;
      ERL_NIF_TERM hhead, htail = tp[1];
      while (enif_get_list_cell(env, htail, &hhead, &htail))
        opts.header_list.push_back(hhead);
    }
  }
  return true;
}

//-----------------------------------------------------------------------------
// SIMD helpers — field scanners used by both decoder and encoder.
//
// find_field_end:   first of delimiter | '\n' | '\r'  (unquoted-field boundary)
// find_csv_special: first of delimiter | '"' | '\n' | '\r'  (needs-quoting check)
// find_byte (glazer_common.hpp): first '"'            (quoted-field / encoder)
//
// find_field_end and find_csv_special cascade AVX2 → SSE2 → scalar.
// find_byte is the shared single-byte scanner from glazer_common.hpp.
//-----------------------------------------------------------------------------

static const char* find_field_end(const char* p, const char* end, char delim) noexcept
{
#if defined(__AVX2__)
  {
    const __m256i vd = _mm256_set1_epi8(delim);
    const __m256i vn = _mm256_set1_epi8('\n');
    const __m256i vr = _mm256_set1_epi8('\r');
    while (p + 32 <= end) {
      __m256i  v    = _mm256_loadu_si256(reinterpret_cast<const __m256i*>(p));
      uint32_t mask = (uint32_t)_mm256_movemask_epi8(_mm256_or_si256(
        _mm256_or_si256(_mm256_cmpeq_epi8(v, vd), _mm256_cmpeq_epi8(v, vn)),
        _mm256_cmpeq_epi8(v, vr)));
      if (mask) return p + __builtin_ctz(mask);
      p += 32;
    }
  }
#endif
#if defined(__SSE2__)
  {
    const __m128i vd = _mm_set1_epi8(delim);
    const __m128i vn = _mm_set1_epi8('\n');
    const __m128i vr = _mm_set1_epi8('\r');
    while (p + 16 <= end) {
      __m128i  v    = _mm_loadu_si128(reinterpret_cast<const __m128i*>(p));
      unsigned mask = (unsigned)_mm_movemask_epi8(_mm_or_si128(
        _mm_or_si128(_mm_cmpeq_epi8(v, vd), _mm_cmpeq_epi8(v, vn)),
        _mm_cmpeq_epi8(v, vr)));
      if (mask) return p + __builtin_ctz(mask);
      p += 16;
    }
  }
#endif
  while (p < end && *p != delim && *p != '\n' && *p != '\r') ++p;
  return p;
}

// Finds the first byte in [p, end) that requires the field to be quoted:
// the delimiter, a double-quote, a newline, or a carriage return.
static const char* find_csv_special(const char* p, const char* end, char delim) noexcept
{
#if defined(__AVX2__)
  {
    const __m256i vd = _mm256_set1_epi8(delim);
    const __m256i vq = _mm256_set1_epi8('"');
    const __m256i vn = _mm256_set1_epi8('\n');
    const __m256i vr = _mm256_set1_epi8('\r');
    while (p + 32 <= end) {
      __m256i  v    = _mm256_loadu_si256(reinterpret_cast<const __m256i*>(p));
      uint32_t mask = (uint32_t)_mm256_movemask_epi8(_mm256_or_si256(_mm256_or_si256(
        _mm256_or_si256(_mm256_cmpeq_epi8(v, vd), _mm256_cmpeq_epi8(v, vq)),
        _mm256_cmpeq_epi8(v, vn)), _mm256_cmpeq_epi8(v, vr)));
      if (mask) return p + __builtin_ctz(mask);
      p += 32;
    }
  }
#endif
#if defined(__SSE2__)
  {
    const __m128i vd = _mm_set1_epi8(delim);
    const __m128i vq = _mm_set1_epi8('"');
    const __m128i vn = _mm_set1_epi8('\n');
    const __m128i vr = _mm_set1_epi8('\r');
    while (p + 16 <= end) {
      __m128i  v    = _mm_loadu_si128(reinterpret_cast<const __m128i*>(p));
      unsigned mask = (unsigned)_mm_movemask_epi8(_mm_or_si128(_mm_or_si128(
        _mm_or_si128(_mm_cmpeq_epi8(v, vd), _mm_cmpeq_epi8(v, vq)),
        _mm_cmpeq_epi8(v, vn)), _mm_cmpeq_epi8(v, vr)));
      if (mask) return p + __builtin_ctz(mask);
      p += 16;
    }
  }
#endif
  while (p < end && *p != delim && *p != '"' && *p != '\n' && *p != '\r') ++p;
  return p;
}

//-----------------------------------------------------------------------------
// Decoder
//-----------------------------------------------------------------------------

struct CsvDecoder {
  ErlNifEnv*           m_env;
  const CsvDecodeOpts& m_opts;
  const char*          m_p;
  const char*          m_end;

  CsvDecoder(ErlNifEnv* e, const CsvDecodeOpts& o, const char* data, size_t size)
    : m_env(e), m_opts(o), m_p(data), m_end(data + size) {}

  static bool is_eol(char c) { return c == '\n' || c == '\r'; }

  void skip_eol() {
    if (m_p < m_end && *m_p == '\r') ++m_p;
    if (m_p < m_end && *m_p == '\n') ++m_p;
  }

  // Unescapes doubled quotes by copying the runs *between* them in bulk,
  // rather than byte-by-byte — for a field with a handful of escaped quotes
  // this is a handful of memcpy calls instead of one push_back per byte.
  ERL_NIF_TERM make_field_term(std::string_view sv, bool has_quote_escape)
  {
    if (!has_quote_escape) return make_binary(m_env, sv);

    std::string buf;
    buf.reserve(sv.size());
    size_t start = 0;
    for (size_t i = 0; i < sv.size(); ++i) {
      if (sv[i] == '"' && i + 1 < sv.size() && sv[i+1] == '"') {
        buf.append(sv.data() + start, i + 1 - start); // include first quote
        start = i + 2; // skip the second quote
        ++i;
      }
    }
    buf.append(sv.data() + start, sv.size() - start);
    return make_binary(m_env, buf);
  }

  // Reads one field starting at m_p. Advances m_p past the field (not past
  // the following delimiter/EOL). Sets `eol`/`eof` flags for the caller and
  // `more_fields` to false if the field was the last one on the line.
  ERL_NIF_TERM read_field(bool& ok)
  {
    ok = true;
    if (m_p < m_end && *m_p == '"') {
      ++m_p; // opening quote
      const char* start = m_p;
      bool has_escape = false;
      for (;;) {
        // SIMD: jump directly to the next '"' (or end) in one scan.
        m_p = find_byte(m_p, m_end, '"');
        if (m_p >= m_end) { ok = false; return 0; } // unterminated quoted field
        if (m_p + 1 < m_end && m_p[1] == '"') { has_escape = true; m_p += 2; continue; }
        break; // closing quote
      }
      std::string_view sv(start, static_cast<size_t>(m_p - start));
      ++m_p; // closing quote
      return make_field_term(sv, has_escape);
    }

    // SIMD: advance past bytes that are not the delimiter, '\n', or '\r'.
    const char* start = m_p;
    m_p = find_field_end(m_p, m_end, m_opts.delimiter);
    return make_binary(m_env, std::string_view(start, static_cast<size_t>(m_p - start)));
  }

  // Reads one record (row) into `fields`. Returns false at end of input with
  // no fields read (clean EOF). On a malformed quoted field, sets `err`.
  template<size_t N>
  bool read_record(SmallTermVec<N>& fields, bool& err)
  {
    err = false;
    fields.set_size(0);
    if (m_p >= m_end) return false;

    // Skip blank lines.
    while (m_p < m_end && is_eol(*m_p)) skip_eol();
    if (m_p >= m_end) return false;

    for (;;) {
      bool ok;
      auto field = read_field(ok);
      if (!ok) [[unlikely]] { err = true; return false; }
      fields.push_back(field);

      if (m_p < m_end && *m_p == m_opts.delimiter) { ++m_p; continue; }
      if (m_p < m_end && is_eol(*m_p)) { skip_eol(); }
      break;
    }
    return true;
  }

  // Attempts the type-specific conversion of `sv` per `spec`. Returns 0 on a
  // value that doesn't match the requested type.
  ERL_NIF_TERM try_convert(std::string_view sv, const CsvFieldSpec& spec)
  {
    switch (spec.type) {
      case CsvFieldType::integer: {
        if (sv.empty()) return 0;
        ERL_NIF_TERM r = glz::BigInt::decode(m_env, sv.data(), sv.data() + sv.size());
        return r;
      }
      case CsvFieldType::floatp: {
        if (sv.empty())
          return 0;
        double d;
        auto [ep, ec] = glz::fast_float::from_chars(sv.data(), sv.data() + sv.size(), d);
        if (ec != std::errc{} || ep != sv.data() + sv.size()) [[unlikely]]
          return 0;
        d = std::round(d * spec.scale) / spec.scale;
        return enif_make_double(m_env, d);
      }
      case CsvFieldType::boolean: {
        if (sv == "true"  || sv == "TRUE"  || sv == "True")  return AM_TRUE;
        if (sv == "false" || sv == "FALSE" || sv == "False") return AM_FALSE;
        return 0;
      }
      case CsvFieldType::datetime: {
        auto secs = datetime::parse(sv, spec.format);
        if (!secs) return 0;
        return enif_make_int64(m_env, *secs);
      }
      case CsvFieldType::charlist: {
        std::vector<ERL_NIF_TERM> cps;
        cps.reserve(sv.size());
        const char* p = sv.data();
        const char* e = p + sv.size();
        while (p < e) cps.push_back(enif_make_uint(m_env, decode_utf8(p, e)));
        return enif_make_list_from_array(m_env, cps.data(), unsigned(cps.size()));
      }
      case CsvFieldType::existing_atom: {
        ERL_NIF_TERM t;
        return enif_make_existing_atom_len(m_env, sv.data(), sv.size(), &t, ERL_NIF_LATIN1)
             ? t : 0;
      }
      case CsvFieldType::atom_list: {
        for (auto& name : spec.atoms) {
          if (sv == name) {
            ERL_NIF_TERM t;
            return enif_make_existing_atom_len(m_env, sv.data(), sv.size(), &t, ERL_NIF_LATIN1)
                 ? t : 0;
          }
        }
        return 0;
      }
      default:
        return 0;
    }
  }

  // Converts one field term (a binary) per `spec`. Returns 0 to signal
  // {invalid_field_value, Row, Col} (only when `spec.on_failure == raise`);
  // otherwise always returns a valid term, falling back per `on_failure`.
  ERL_NIF_TERM convert_field(ERL_NIF_TERM term, const CsvFieldSpec& spec)
  {
    if (spec.type == CsvFieldType::none || spec.type == CsvFieldType::binary)
      return term;

    ErlNifBinary bin;
    if (!enif_inspect_binary(m_env, term, &bin))
      return 0;

    auto sv = std::string_view(reinterpret_cast<const char*>(bin.data), bin.size);

    if (sv.empty() && spec.has_default)
      return spec.default_term;

    ERL_NIF_TERM r = try_convert(sv, spec);
    if (r) [[likely]] return r;

    switch (spec.on_failure) {
      case CsvOnFailure::binary:        return term;
      case CsvOnFailure::raise:         return 0;
      case CsvOnFailure::default_value: return spec.has_default ? spec.default_term : term;
      case CsvOnFailure::null:          return m_opts.null_term;
    }
    return term;
  }

  // Converts `fields` in place per `m_opts.fields` (positional by column
  // index; columns beyond the spec list, or with the `none`/default spec,
  // are left as binaries). Returns the 1-based column index of the first
  // field that fails to convert, or 0 on success.
  template<size_t N>
  size_t convert_fields(SmallTermVec<N>& fields)
  {
    size_t n = std::min(fields.size(), m_opts.fields.size());
    for (size_t i = 0; i < n; ++i) {
      auto converted = convert_field(fields[i], m_opts.fields[i]);
      if (!converted) [[unlikely]]
        return i + 1;
      fields.data()[i] = converted;
    }
    return 0;
  }

  ERL_NIF_TERM make_header_key(ERL_NIF_TERM bin_term)
  {
    if (!m_opts.label_atom && !m_opts.label_existing_atom && !m_opts.label_charlist)
      return bin_term;

    ErlNifBinary bin;
    if (!enif_inspect_binary(m_env, bin_term, &bin)) return bin_term;
    auto sv = std::string_view(reinterpret_cast<const char*>(bin.data), bin.size);

    if (m_opts.label_atom)
      return enif_make_atom_len(m_env, sv.data(), sv.size());

    if (m_opts.label_existing_atom) {
      ERL_NIF_TERM t;
      return enif_make_existing_atom_len(m_env, sv.data(), sv.size(), &t, ERL_NIF_LATIN1)
           ? t : bin_term;
    }

    // charlist: decode UTF-8 bytes to a list of Unicode codepoints
    std::vector<ERL_NIF_TERM> cps;
    const char* p   = sv.data();
    const char* end = p + sv.size();
    while (p < end)
      cps.push_back(enif_make_uint(m_env, decode_utf8(p, end)));
    return enif_make_list_from_array(m_env, cps.data(), unsigned(cps.size()));
  }

  // Decodes the entire input.
  // Returns:
  //   - success: {true, #{headers => nil|[binary()|atom()], data => [[term()]]|[map()]}}
  //   - failure: {false, atom | binary}
  //
  // When `headers` is set the first row is extracted as the `headers` value
  // (applying `{headers, atom|existing_atom|charlist}` if requested). Subsequent rows are
  // emitted as field lists (default), as tuples when {return, tuple} is set,
  // or as maps keyed by the header names when {return, map} is also set.
  // duplicate_header is returned if a header row contains duplicate column
  // names and map output is requested.
  std::tuple<bool, ERL_NIF_TERM> decode()
  {
    SmallTermVec<64> fields;
    SmallTermVec<64> header;
    std::vector<ERL_NIF_TERM> rows;
    size_t row_num  = 0;
    // Hoist as_map before any goto so no initialization is jumped over.
    const bool as_map = m_opts.return_kind == CsvReturnKind::map && m_opts.headers;
    const bool as_tuple = m_opts.return_kind == CsvReturnKind::tuple;

    bool err = false;
    auto rec = read_record(fields, err);

    if (!rec) [[unlikely]]
      goto DONE;

    // Populate header: explicit list beats reading a header row from the data.
    if (m_opts.explicit_headers) {
      for (auto h : m_opts.header_list)
        header.push_back(h);
      // Do NOT consume a data row as the header.
    } else if (m_opts.headers) {
      for (auto field : fields)
        header.push_back(make_header_key(field));
      rec = read_record(fields, err);
    }

    // Skip the requested number of leading data rows.
    {
      size_t skipped = 0;
      while (rec && skipped < m_opts.skip_rows) {
        rec = read_record(fields, err);
        ++skipped;
      }
    }

    // Data rows: maps when {return, map} + headers, field lists otherwise.
    while (rec && (m_opts.limit == 0 || row_num < m_opts.limit)) {
      ++row_num;
      if (!m_opts.fields.empty()) {
        if (size_t col = convert_fields(fields))
          return std::make_tuple(false, enif_make_tuple3(m_env, AM_INVALID_FIELD_VALUE,
                                                           enif_make_uint64(m_env, row_num),
                                                           enif_make_uint64(m_env, col)));
      }
      if (as_map) {
        ERL_NIF_TERM map = fields.to_erl_map(m_env, header);
        if (!map) [[unlikely]]
          return std::make_tuple(false, AM_DUPLICATE_HEADER);
        rows.push_back(map);
      } else if (as_tuple) {
        rows.push_back(fields.to_erl_tuple(m_env));
      } else {
        rows.push_back(fields.to_erl_list(m_env));
      }
      rec = read_record(fields, err);
    }

  DONE:
    if (err) [[unlikely]]
      return std::make_tuple(false, AM_UNTERMINATED_QUOTED_FIELD);

    ERL_NIF_TERM headers_term = m_opts.headers
      ? enif_make_list_from_array(m_env, header.data(), unsigned(header.size()))
      : AM_NIL;
    ERL_NIF_TERM data_term =
      enif_make_list_from_array(m_env, rows.data(), unsigned(rows.size()));

    ERL_NIF_TERM keys[2] = { AM_HEADERS, AM_DATA };
    ERL_NIF_TERM vals[2] = { headers_term, data_term };
    ERL_NIF_TERM result;
    enif_make_map_from_arrays(m_env, keys, vals, 2, &result);

    return std::make_tuple(true, result);
  }
};

//-----------------------------------------------------------------------------
// Encoder
//-----------------------------------------------------------------------------

struct CsvEncoder {
  ErlNifEnv*           m_env;
  const CsvEncodeOpts& m_opts;
  OutBuf&              m_out;
  std::string          m_err;

  CsvEncoder(ErlNifEnv* e, const CsvEncodeOpts& o, OutBuf& out)
    : m_env(e), m_opts(o), m_out(out) {}

  void push_field_raw(std::string_view sv)
  {
    const char* p   = sv.data();
    const char* end = p + sv.size();

    // SIMD: check whether quoting is needed at all.
    if (find_csv_special(p, end, m_opts.delimiter) == end) { m_out.push(sv); return; }

    // Field contains a special byte — wrap in double quotes and double any
    // embedded '"' characters. SIMD locates each '"' for bulk-copy runs.
    m_out.push('"');
    for (;;) {
      const char* q = find_byte(p, end, '"');
      m_out.push(p, q - p);    // bulk-copy bytes up to (not including) the quote
      if (q >= end) break;
      m_out.push("\"\"", 2);   // double the embedded quote
      p = q + 1;
    }
    m_out.push('"');
  }

  // Encodes one field term (binary, atom, integer, or float). Returns false
  // for unsupported term types.
  bool encode_field(ERL_NIF_TERM term)
  {
    switch (enif_term_type(m_env, term)) {
      case ERL_NIF_TERM_TYPE_BITSTRING: {
        ErlNifBinary bin;
        if (!enif_inspect_binary(m_env, term, &bin)) return false;
        push_field_raw({reinterpret_cast<const char*>(bin.data), bin.size});
        return true;
      }
      case ERL_NIF_TERM_TYPE_INTEGER:
        return glz::BigInt::encode(m_env, term, m_out);

      case ERL_NIF_TERM_TYPE_FLOAT: {
        double d;
        if (!enif_get_double(m_env, term, &d)) return false;
        char buf[32];
        auto [e, ec] = std::to_chars(buf, buf+32, d, std::chars_format::general);
        if (ec != std::errc{}) return false;
        m_out.push(buf, e - buf);
        return true;
      }
      case ERL_NIF_TERM_TYPE_ATOM: {
        char buf[256];
        std::string_view sv;
        if (!atom_to_sv(m_env, term, buf, sizeof(buf), sv)) return false;
        push_field_raw(sv);
        return true;
      }
      default:
        return false;
    }
  }

  bool encode_row(ERL_NIF_TERM row)
  {
    ERL_NIF_TERM head, tail = row;
    bool first = true;
    while (enif_get_list_cell(m_env, tail, &head, &tail)) {
      if (!first) m_out.push(m_opts.delimiter);
      first = false;
      if (!encode_field(head)) { m_err = "cannot encode CSV field"; return false; }
    }
    m_out.push(m_opts.line_ending);
    return true;
  }

  bool encode_map_row(ERL_NIF_TERM map, const std::vector<ERL_NIF_TERM>& header)
  {
    for (size_t i = 0; i < header.size(); ++i) {
      if (i > 0) m_out.push(m_opts.delimiter);
      ERL_NIF_TERM val;
      if (!enif_get_map_value(m_env, map, header[i], &val)) continue; // missing key -> empty field
      if (!encode_field(val)) { m_err = "cannot encode CSV field"; return false; }
    }
    m_out.push(m_opts.line_ending);
    return true;
  }

  // `term` is a list of rows (each a list of fields), or — with `headers` —
  // a list of maps.
  bool encode(ERL_NIF_TERM term)
  {
    if (!enif_is_list(m_env, term)) { m_err = "expected a list of rows"; return false; }

    if (enif_is_empty_list(m_env, term)) return true;

    if (m_opts.headers) {
      std::vector<ERL_NIF_TERM> header;

      if (m_opts.explicit_headers) {
        // Use the explicitly given column order/names.
        header = m_opts.header_list;
      } else {
        // Determine column order from the first row's map keys.
        ERL_NIF_TERM head, tail = term;
        enif_get_list_cell(m_env, tail, &head, &tail);
        if (enif_term_type(m_env, head) != ERL_NIF_TERM_TYPE_MAP) {
          m_err = "headers option requires rows to be maps";
          return false;
        }

        ERL_NIF_TERM key, val;
        ErlNifMapIterator it;
        enif_map_iterator_create(m_env, head, &it, ERL_NIF_MAP_ITERATOR_FIRST);
        while (enif_map_iterator_get_pair(m_env, &it, &key, &val)) {
          header.push_back(key);
          enif_map_iterator_next(m_env, &it);
        }
        enif_map_iterator_destroy(m_env, &it);
      }

      // Emit header row.
      for (size_t i = 0; i < header.size(); ++i) {
        if (i > 0) m_out.push(m_opts.delimiter);
        if (!encode_field(header[i])) { m_err = "cannot encode CSV header"; return false; }
      }
      m_out.push(m_opts.line_ending);

      ERL_NIF_TERM row, rest = term;
      while (enif_get_list_cell(m_env, rest, &row, &rest)) {
        if (enif_term_type(m_env, row) != ERL_NIF_TERM_TYPE_MAP) {
          m_err = "headers option requires rows to be maps";
          return false;
        }
        if (!encode_map_row(row, header)) return false;
      }
      return true;
    }

    ERL_NIF_TERM row, rest = term;
    while (enif_get_list_cell(m_env, rest, &row, &rest)) {
      if (!enif_is_list(m_env, row)) { m_err = "expected each row to be a list"; return false; }
      if (!encode_row(row)) return false;
    }
    return true;
  }
};

} // namespace glz