// Step 3 -- stop touching the bytes: fields become views into the mapping.
//
// Step 2 still ran every byte of every field through `cur += c`. This step
// hands the caller {pointer, length} pairs -- std::string_view -- aimed
// straight into the mmap'd file. Nothing is copied, nothing is allocated,
// nothing is even written; the parser's only output is coordinates.
//
// The two decisions that make zero-copy work:
//
//   * Lifetime. A string_view into a mapping is valid exactly as long as the
//     mapping. Here the Mapped object outlives every use of every view, so
//     the borrow is safe BY SCOPE, not by discipline. (An API that let views
//     escape the mapping's lifetime would be a bug factory; ours processes
//     rows inside the parse call.)
//   * Escapes are deferred. A quoted field with doubled quotes ("she said
//     ""hi""") cannot be viewed verbatim -- unescaping needs a write
//     somewhere. So the view keeps the RAW bytes and a needs_unescape flag;
//     only a consumer that actually wants the cooked text pays for the copy
//     (unescape_into, used by the correctness check). In bench.csv that is
//     ~1% of fields; the other 99% are never rewritten at all.
//
// PROOF, not promise: the benchmark parses the file twice and reports heap
// allocations during the second pass. The number is zero -- every vector has
// reached capacity, and views allocate nothing.
//
// Build:  c++ -std=c++20 -O2 step3_zerocopy.cpp -o step3
#include "csv_test_common.hpp"
#include <fcntl.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>
#include <string>
#include <string_view>
#include <vector>

struct Mapped {
    const char* data = nullptr;
    size_t size = 0;
    Mapped(const char* path) {
        int fd = ::open(path, O_RDONLY);
        if (fd < 0) { std::perror(path); std::exit(1); }
        struct stat st{};
        ::fstat(fd, &st);
        size = size_t(st.st_size);
        data = (const char*)::mmap(nullptr, size, PROT_READ, MAP_PRIVATE, fd, 0);
        ::close(fd);
        if (data == MAP_FAILED) { std::perror("mmap"); std::exit(1); }
    }
    ~Mapped() { if (data) ::munmap((void*)data, size); }
};

struct Field {
    std::string_view raw;      // quoted fields: content between the quotes,
    bool needs_unescape;       // doubled quotes still doubled
};

// Collapse "" -> " into `out`. Only called when needs_unescape is set.
static void unescape_into(std::string_view raw, std::string& out) {
    out.clear();
    for (size_t i = 0; i < raw.size(); ++i) {
        out += raw[i];
        if (raw[i] == '"') ++i;          // skip the second quote of a pair
    }
}

// One pass; on_record receives the reused field vector for each record.
template <typename OnRecord>
static void parse_view(const char* p, const char* end, OnRecord on_record) {
    std::vector<Field> rec;
    rec.reserve(16);
    while (p < end) {
        rec.clear();
        for (;;) {                                   // one field per iteration
            bool quoted = (p < end && *p == '"');
            const char* begin;
            bool esc = false;
            if (quoted) {
                begin = ++p;                          // content starts after "
                while (p < end) {
                    if (*p == '"') {
                        if (p + 1 < end && p[1] == '"') { esc = true; p += 2; }
                        else break;                   // closing quote
                    } else ++p;
                }
                rec.push_back({{begin, size_t(p - begin)}, esc});
                if (p < end) ++p;                     // step over closing "
            } else {
                begin = p;
                while (p < end && *p != ',' && *p != '\r' && *p != '\n') ++p;
                rec.push_back({{begin, size_t(p - begin)}, false});
            }
            if (p >= end) break;                      // EOF ends the record
            char c = *p++;
            if (c == ',') continue;                   // next field
            if (c == '\r' && p < end && *p == '\n') ++p;
            break;                                    // \r, \r\n or \n: record done
        }
        on_record(rec);
    }
}

static std::vector<std::vector<std::string>> parse_all_materialized(const char* path) {
    Mapped m(path);
    std::vector<std::vector<std::string>> rows;
    std::string scratch;
    parse_view(m.data, m.data + m.size, [&](const std::vector<Field>& rec) {
        std::vector<std::string> out;
        for (const Field& f : rec) {
            if (f.needs_unescape) { unescape_into(f.raw, scratch); out.push_back(scratch); }
            else out.emplace_back(f.raw);
        }
        rows.push_back(std::move(out));
    });
    return rows;
}

int main() {
    if (!run_edge_checks("edge.csv", parse_all_materialized)) return 1;

    Mapped m("bench.csv");
    const double mb = m.size / 1e6;

    BenchResult res;
    for (int rep = 0; rep < 3; ++rep) {
        unsigned long long a0 = allocs_now();
        double t0 = now_secs();
        unsigned long long rows = 0, fields = 0, byte_sum = 0;
        parse_view(m.data, m.data + m.size, [&](const std::vector<Field>& rec) {
            ++rows;
            fields += rec.size();
            for (const Field& f : rec) byte_sum += f.raw.size();
        });
        double dt = now_secs() - t0;
        if (dt < res.best_secs) res.best_secs = dt;
        res.rows = rows; res.fields = fields; res.byte_sum = byte_sum;
        res.last_pass_allocs = allocs_now() - a0;    // second/third pass: 0
    }
    print_bench("step3_zerocopy", res, mb);
    return 0;
}