//! Streaming parser for Jetstream V2 operator timestamp imports.
//!
//! The source is an uncompressed RFC 4180 CSV so later phases can seek back
//! to an accepted row by byte offset. Parsing is forward-only and bounded by
//! `max_row_bytes`; strings in `Row` and `Reject` borrow parser scratch and are
//! valid only for the duration of their callback.

const std = @import("std");
const zat = @import("zat");

pub const max_row_bytes = 64 * 1024;
pub const default_reject_sample_limit: usize = 100;
pub const max_diagnostic_value_len: usize = 256;

pub const Scope = enum(u8) {
    all_versions = 0,
    specific_version = 1,
};

/// Atmos' CID representation is deliberately fixed to the formats accepted
/// by upstream: CIDv1, dag-cbor/raw, SHA-256. This also gives the rule store a
/// stable allocation-free key representation.
pub const Cid = struct {
    codec: u8,
    hash: [32]u8,

    pub fn eql(a: Cid, b: Cid) bool {
        return a.codec == b.codec and std.mem.eql(u8, &a.hash, &b.hash);
    }
};

pub const Row = struct {
    offset: u64,
    did: []const u8,
    collection: []const u8,
    rkey: []const u8,
    scope: Scope,
    cid: ?Cid,
    timestamp_micros: i64,
};

pub const RejectReason = enum {
    malformed_csv,
    missing_field,
    bad_uri,
    uri_authority_not_did,
    uri_missing_collection_or_rkey,
    bad_timestamp,
    non_positive_timestamp,
    unknown_scope,
    specific_version_missing_cid,
    bad_cid,
    row_too_long,
};

pub const Reject = struct {
    row_number: u64,
    offset: u64,
    reason: RejectReason,
    field: []const u8,
    value: []const u8,
};

pub const OwnedReject = struct {
    row_number: u64,
    offset: u64,
    reason: RejectReason,
    field: []u8,
    value: []u8,
};

pub const Stats = struct {
    rows_total: u64 = 0,
    rows_valid: u64 = 0,
    rows_rejected: u64 = 0,
    rejects_by_reason: [@typeInfo(RejectReason).@"enum".field_names.len]u64 = @splat(0),
    reject_sample: std.ArrayList(OwnedReject) = .empty,

    pub fn rejected(self: *const Stats, reason: RejectReason) u64 {
        return self.rejects_by_reason[@backingInt(reason)];
    }

    pub fn deinit(self: *Stats, allocator: std.mem.Allocator) void {
        for (self.reject_sample.items) |item| {
            allocator.free(item.field);
            allocator.free(item.value);
        }
        self.reject_sample.deinit(allocator);
        self.* = undefined;
    }
};

pub const Options = struct {
    context: ?*anyopaque,
    on_row: *const fn (?*anyopaque, Row) anyerror!void,
    on_reject: ?*const fn (?*anyopaque, Reject) void = null,
    /// Zero selects the upstream default; `no_reject_sample` disables it.
    reject_sample_limit: usize = 0,

    pub const no_reject_sample = std.math.maxInt(usize);
};

const Columns = struct {
    uri: u3,
    timestamp: u3,
    scope: ?u3,
    cid: ?u3,
    count: u3,
};

const Record = struct {
    bytes: []const u8,
    offset: u64,
    footprint: u64,
    too_long: bool,
};

const FieldSet = struct {
    values: [5][]const u8 = @splat(""),
    count: usize = 0,
};

/// Positioned Phase-C reader for offsets produced by the streaming parser.
/// It owns one file handle and private scratch, so separate instances can be
/// used concurrently without sharing a seek cursor. Every read independently
/// verifies the row boundary, 64 KiB contract, CSV shape, and field semantics.
pub const RowReader = struct {
    io: std.Io,
    file: std.Io.File,
    size: u64,
    columns: Columns,
    raw: [max_row_bytes]u8 = undefined,
    decoded: [max_row_bytes]u8 = undefined,

    pub fn open(io: std.Io, path: []const u8) !RowReader {
        const file = try std.Io.Dir.cwd().openFile(io, path, .{});
        errdefer file.close(io);
        const stat = try file.stat(io);
        var self: RowReader = .{
            .io = io,
            .file = file,
            .size = stat.size,
            .columns = undefined,
        };
        const count = try file.readPositionalAll(io, &self.raw, 0);
        var reader: std.Io.Reader = .fixed(self.raw[0..count]);
        var position: u64 = 0;
        const record = (try readRecord(&reader, &self.raw, &position)) orelse
            return error.InvalidHeader;
        if (record.too_long or record.footprint >= max_row_bytes) return error.InvalidHeader;
        self.columns = parseHeader(decodeRecord(record.bytes, &self.decoded) catch
            return error.InvalidHeader) catch return error.InvalidHeader;
        return self;
    }

    pub fn close(self: *RowReader) void {
        self.file.close(self.io);
        self.* = undefined;
    }

    pub fn read(self: *RowReader, offset: u64) !Row {
        if (offset == 0 or offset >= self.size) return error.CorruptOffset;
        var previous: [1]u8 = undefined;
        if (try self.file.readPositionalAll(self.io, &previous, offset - 1) != 1 or previous[0] != '\n')
            return error.CorruptOffset;

        const span: usize = @intCast(@min(self.size - offset, max_row_bytes));
        const count = try self.file.readPositionalAll(self.io, self.raw[0..span], offset);
        var reader: std.Io.Reader = .fixed(self.raw[0..count]);
        var position = offset;
        const record = (try readRecord(&reader, &self.raw, &position)) orelse
            return error.CorruptOffset;
        if (record.too_long or record.footprint >= max_row_bytes)
            return error.CorruptOffset;
        const fields = decodeRecord(record.bytes, &self.decoded) catch
            return error.CorruptOffset;
        if (fields.count != self.columns.count) return error.CorruptOffset;
        return switch (validate(self.columns, fields, offset)) {
            .row => |row| row,
            .reject => error.CorruptOffset,
        };
    }
};

/// Parse a timestamp import in one forward-only pass. Only structural header
/// errors, quote corruption, read failures, allocation failures, and the row
/// callback abort the import. Ordinary data errors are counted and skipped.
pub fn parse(
    allocator: std.mem.Allocator,
    reader: *std.Io.Reader,
    opts: Options,
) !Stats {
    var stats: Stats = .{};
    parseInto(allocator, reader, opts, &stats) catch |err| {
        stats.deinit(allocator);
        return err;
    };
    return stats;
}

/// Variant used by the durable importer so partial counts remain available
/// when a structural error terminates the file. `stats` is initialized before
/// any read and always belongs to the caller, on success or error.
pub fn parseInto(
    allocator: std.mem.Allocator,
    reader: *std.Io.Reader,
    opts: Options,
    stats: *Stats,
) !void {
    stats.* = .{};

    var raw: [max_row_bytes]u8 = undefined;
    var decoded: [max_row_bytes]u8 = undefined;
    var input_offset: u64 = 0;

    const header_record = (try readRecord(reader, &raw, &input_offset)) orelse
        return error.InvalidHeader;
    if (header_record.too_long) return error.InvalidHeader;
    const header_fields = decodeRecord(header_record.bytes, &decoded) catch
        return error.InvalidHeader;
    const columns = parseHeader(header_fields) catch return error.InvalidHeader;

    var row_number: u64 = 0;
    while (try readRecord(reader, &raw, &input_offset)) |record| {
        row_number += 1;
        if (record.too_long or record.footprint >= max_row_bytes) {
            try noteReject(allocator, stats, opts, .{
                .row_number = row_number,
                .offset = record.offset,
                .reason = .row_too_long,
                .field = "",
                .value = "row reaches or exceeds the 64 KiB limit",
            });
            continue;
        }

        const fields = decodeRecord(record.bytes, &decoded) catch |err| switch (err) {
            error.MalformedCsv => {
                try noteReject(allocator, stats, opts, .{
                    .row_number = row_number,
                    .offset = record.offset,
                    .reason = .malformed_csv,
                    .field = "",
                    .value = record.bytes,
                });
                continue;
            },
        };
        if (fields.count != columns.count) {
            try noteReject(allocator, stats, opts, .{
                .row_number = row_number,
                .offset = record.offset,
                .reason = .missing_field,
                .field = "",
                .value = record.bytes,
            });
            continue;
        }

        const result = validate(columns, fields, record.offset);
        switch (result) {
            .reject => |reject| try noteReject(allocator, stats, opts, .{
                .row_number = row_number,
                .offset = record.offset,
                .reason = reject.reason,
                .field = reject.field,
                .value = reject.value,
            }),
            .row => |row| {
                stats.rows_total += 1;
                stats.rows_valid += 1;
                try opts.on_row(opts.context, row);
            },
        }
    }
}

const ValidationReject = struct {
    reason: RejectReason,
    field: []const u8,
    value: []const u8,
};

const Validation = union(enum) {
    row: Row,
    reject: ValidationReject,
};

fn validate(columns: Columns, fields: FieldSet, offset: u64) Validation {
    const uri_text = trim(fields.values[columns.uri]);
    if (uri_text.len == 0) return rejected(.missing_field, "uri", "");
    const timestamp_text = trim(fields.values[columns.timestamp]);
    if (timestamp_text.len == 0) return rejected(.missing_field, "timestamp", "");

    const uri = zat.AtUri.parse(uri_text) orelse
        return rejected(.bad_uri, "uri", uri_text);
    const did = zat.Did.parse(uri.authority()) orelse
        return rejected(.uri_authority_not_did, "uri", uri_text);
    const collection = uri.collection() orelse
        return rejected(.uri_missing_collection_or_rkey, "uri", uri_text);
    const rkey = uri.rkey() orelse
        return rejected(.uri_missing_collection_or_rkey, "uri", uri_text);

    const timestamp_micros = parseRfc3339Micros(timestamp_text) orelse
        return rejected(.bad_timestamp, "timestamp", timestamp_text);
    if (timestamp_micros <= 0)
        return rejected(.non_positive_timestamp, "timestamp", timestamp_text);

    var scope: Scope = .all_versions;
    if (columns.scope) |index| {
        const value = trim(fields.values[index]);
        if (value.len != 0) {
            if (std.mem.eql(u8, value, "all_versions")) {
                scope = .all_versions;
            } else if (std.mem.eql(u8, value, "specific_version")) {
                scope = .specific_version;
            } else {
                return rejected(.unknown_scope, "scope", value);
            }
        }
    }

    var cid: ?Cid = null;
    if (scope == .specific_version) {
        const cid_text = if (columns.cid) |index| trim(fields.values[index]) else "";
        if (cid_text.len == 0)
            return rejected(.specific_version_missing_cid, "cid", "");
        cid = parseCid(cid_text) orelse return rejected(.bad_cid, "cid", cid_text);
    }

    return .{ .row = .{
        .offset = offset,
        .did = did.str(),
        .collection = collection,
        .rkey = rkey,
        .scope = scope,
        .cid = cid,
        .timestamp_micros = timestamp_micros,
    } };
}

fn rejected(reason: RejectReason, field: []const u8, value: []const u8) Validation {
    return .{ .reject = .{ .reason = reason, .field = field, .value = value } };
}

fn noteReject(
    allocator: std.mem.Allocator,
    stats: *Stats,
    opts: Options,
    reject: Reject,
) !void {
    var diagnostic_buffer: [max_diagnostic_value_len + 3]u8 = undefined;
    const value = truncateDiagnostic(reject.value, &diagnostic_buffer);
    const bounded: Reject = .{
        .row_number = reject.row_number,
        .offset = reject.offset,
        .reason = reject.reason,
        .field = reject.field,
        .value = value,
    };
    stats.rows_total += 1;
    stats.rows_rejected += 1;
    stats.rejects_by_reason[@backingInt(reject.reason)] += 1;
    if (opts.on_reject) |callback| callback(opts.context, bounded);

    const limit = if (opts.reject_sample_limit == 0)
        default_reject_sample_limit
    else if (opts.reject_sample_limit == Options.no_reject_sample)
        0
    else
        opts.reject_sample_limit;
    if (stats.reject_sample.items.len >= limit) return;

    const field = try allocator.dupe(u8, bounded.field);
    errdefer allocator.free(field);
    const owned_value = try allocator.dupe(u8, bounded.value);
    errdefer allocator.free(owned_value);
    try stats.reject_sample.append(allocator, .{
        .row_number = bounded.row_number,
        .offset = bounded.offset,
        .reason = bounded.reason,
        .field = field,
        .value = owned_value,
    });
}

fn parseHeader(fields: FieldSet) !Columns {
    if (fields.count == 0 or fields.count > 4) return error.InvalidHeader;
    var uri: ?u3 = null;
    var timestamp: ?u3 = null;
    var scope: ?u3 = null;
    var cid: ?u3 = null;

    for (fields.values[0..fields.count], 0..) |raw, index| {
        const name = trim(raw);
        const slot: *?u3 = if (std.ascii.eqlIgnoreCase(name, "uri"))
            &uri
        else if (std.ascii.eqlIgnoreCase(name, "timestamp"))
            &timestamp
        else if (std.ascii.eqlIgnoreCase(name, "scope"))
            &scope
        else if (std.ascii.eqlIgnoreCase(name, "cid"))
            &cid
        else
            return error.InvalidHeader;
        if (slot.* != null) return error.InvalidHeader;
        slot.* = @intCast(index);
    }
    return .{
        .uri = uri orelse return error.InvalidHeader,
        .timestamp = timestamp orelse return error.InvalidHeader,
        .scope = scope,
        .cid = cid,
        .count = @intCast(fields.count),
    };
}

/// Read one RFC 4180 logical record. Blank physical lines are skipped exactly
/// like Go's encoding/csv; the returned offset remains before skipped lines so
/// seeking there and decoding again still yields the same row.
fn readRecord(reader: *std.Io.Reader, storage: []u8, input_offset: *u64) !?Record {
    const State = enum { field_start, unquoted, quoted, after_quote, after_quote_cr };
    const offset = input_offset.*;
    var state: State = .field_start;
    var length: usize = 0;
    var footprint: u64 = 0;
    var too_long = false;

    while (true) {
        const byte = reader.takeByte() catch |err| switch (err) {
            error.EndOfStream => {
                if (state == .quoted) return error.UnbalancedQuote;
                if (state == .after_quote_cr) return error.UnbalancedQuote;
                if (footprint == 0) return null;
                if (isBlankRecord(storage[0..length])) return null;
                return .{
                    .bytes = storage[0..length],
                    .offset = offset,
                    .footprint = footprint,
                    .too_long = too_long,
                };
            },
            error.ReadFailed => return error.ReadFailed,
        };
        input_offset.* += 1;
        footprint += 1;
        if (length < storage.len) {
            storage[length] = byte;
            length += 1;
        } else {
            too_long = true;
        }

        switch (state) {
            .field_start => switch (byte) {
                '"' => state = .quoted,
                ',' => {},
                '\n' => {
                    if (isBlankRecord(storage[0..length])) {
                        length = 0;
                        state = .field_start;
                        continue;
                    }
                    return .{ .bytes = storage[0..length], .offset = offset, .footprint = footprint, .too_long = too_long };
                },
                else => state = .unquoted,
            },
            .unquoted => switch (byte) {
                '"' => return error.UnbalancedQuote,
                ',' => state = .field_start,
                '\n' => {
                    if (isBlankRecord(storage[0..length])) {
                        length = 0;
                        state = .field_start;
                        continue;
                    }
                    return .{ .bytes = storage[0..length], .offset = offset, .footprint = footprint, .too_long = too_long };
                },
                else => {},
            },
            .quoted => {
                if (byte == '"') state = .after_quote;
            },
            .after_quote => switch (byte) {
                '"' => state = .quoted,
                ',' => state = .field_start,
                '\r' => state = .after_quote_cr,
                '\n' => return .{ .bytes = storage[0..length], .offset = offset, .footprint = footprint, .too_long = too_long },
                else => return error.UnbalancedQuote,
            },
            .after_quote_cr => if (byte == '\n')
                return .{ .bytes = storage[0..length], .offset = offset, .footprint = footprint, .too_long = too_long }
            else
                return error.UnbalancedQuote,
        }
    }
}

fn isBlankRecord(bytes: []const u8) bool {
    return bytes.len == 0 or std.mem.eql(u8, bytes, "\n") or std.mem.eql(u8, bytes, "\r\n");
}

/// Decode a structurally checked record into at most five fields. Five is
/// enough to prove an invalid field count because the schema has at most four.
fn decodeRecord(raw_record: []const u8, storage: []u8) error{MalformedCsv}!FieldSet {
    var raw = raw_record;
    if (raw.len > 0 and raw[raw.len - 1] == '\n') raw = raw[0 .. raw.len - 1];
    if (raw.len > 0 and raw[raw.len - 1] == '\r') raw = raw[0 .. raw.len - 1];

    var result: FieldSet = .{};
    var in_pos: usize = 0;
    var out_pos: usize = 0;
    while (true) {
        const field_start = out_pos;
        if (in_pos < raw.len and raw[in_pos] == '"') {
            in_pos += 1;
            while (true) {
                if (in_pos >= raw.len) return error.MalformedCsv;
                const byte = raw[in_pos];
                if (byte == '"') {
                    if (in_pos + 1 < raw.len and raw[in_pos + 1] == '"') {
                        storage[out_pos] = '"';
                        out_pos += 1;
                        in_pos += 2;
                        continue;
                    }
                    in_pos += 1;
                    break;
                }
                // encoding/csv normalizes CRLF within quoted fields to LF.
                if (byte == '\r' and in_pos + 1 < raw.len and raw[in_pos + 1] == '\n') {
                    in_pos += 1;
                    continue;
                }
                storage[out_pos] = byte;
                out_pos += 1;
                in_pos += 1;
            }
            if (in_pos < raw.len and raw[in_pos] != ',') return error.MalformedCsv;
        } else {
            while (in_pos < raw.len and raw[in_pos] != ',') : (in_pos += 1) {
                if (raw[in_pos] == '"') return error.MalformedCsv;
                storage[out_pos] = raw[in_pos];
                out_pos += 1;
            }
        }

        if (result.count < result.values.len)
            result.values[result.count] = storage[field_start..out_pos];
        result.count += 1;
        if (in_pos >= raw.len) break;
        in_pos += 1; // comma
    }
    return result;
}

fn parseCid(text: []const u8) ?Cid {
    if (text.len < 2 or text[0] != 'b') return null;
    var scratch: [128]u8 = undefined;
    var fba = std.heap.FixedBufferAllocator.init(&scratch);
    const raw = zat.multibase.base32lower.decode(fba.allocator(), text[1..]) catch return null;
    // Atmos accepts only minimally encoded CIDv1, dag-cbor/raw, SHA-256/32.
    if (raw.len != 36 or raw[0] != 1 or (raw[1] != 0x71 and raw[1] != 0x55) or
        raw[2] != 0x12 or raw[3] != 0x20) return null;
    var hash: [32]u8 = undefined;
    @memcpy(&hash, raw[4..36]);
    return .{ .codec = raw[1], .hash = hash };
}

/// Go time.Parse(time.RFC3339) semantics needed by the import format: fixed
/// four-digit civil date, seconds, optional 1-9 digit fraction, and Z or a
/// numeric offset. The result truncates nanoseconds to Unix microseconds.
fn parseRfc3339Micros(text: []const u8) ?i64 {
    if (text.len < 20) return null;
    if (text[4] != '-' or text[7] != '-' or text[10] != 'T' or
        text[13] != ':' or text[16] != ':') return null;
    const year = parseDigits(text[0..4]) orelse return null;
    const month = parseDigits(text[5..7]) orelse return null;
    const day = parseDigits(text[8..10]) orelse return null;
    const hour = parseDigits(text[11..13]) orelse return null;
    const minute = parseDigits(text[14..16]) orelse return null;
    const second = parseDigits(text[17..19]) orelse return null;
    if (month < 1 or month > 12 or day < 1 or day > daysInMonth(year, month) or
        hour > 23 or minute > 59 or second > 59) return null;

    var pos: usize = 19;
    var micros_fraction: i64 = 0;
    if (pos < text.len and (text[pos] == '.' or text[pos] == ',')) {
        pos += 1;
        const start = pos;
        var digits: usize = 0;
        while (pos < text.len and std.ascii.isDigit(text[pos])) : (pos += 1) {
            digits += 1;
            if (digits <= 6) micros_fraction = micros_fraction * 10 + (text[pos] - '0');
        }
        if (pos == start) return null;
        while (digits < 6) : (digits += 1) micros_fraction *= 10;
    }

    var offset_seconds: i64 = 0;
    if (pos < text.len and text[pos] == 'Z') {
        pos += 1;
    } else {
        if (pos + 6 != text.len or (text[pos] != '+' and text[pos] != '-') or text[pos + 3] != ':') return null;
        const zone_hour = parseDigits(text[pos + 1 .. pos + 3]) orelse return null;
        const zone_minute = parseDigits(text[pos + 4 .. pos + 6]) orelse return null;
        if (zone_hour > 24 or zone_minute > 60) return null;
        offset_seconds = @as(i64, zone_hour) * 3600 + @as(i64, zone_minute) * 60;
        if (text[pos] == '-') offset_seconds = -offset_seconds;
        pos += 6;
    }
    if (pos != text.len) return null;

    const days = daysFromCivil(@intCast(year), @intCast(month), @intCast(day));
    const seconds = days * std.time.s_per_day + @as(i64, hour) * 3600 +
        @as(i64, minute) * 60 + @as(i64, second) - offset_seconds;
    const whole_micros = std.math.mul(i64, seconds, std.time.us_per_s) catch return null;
    return std.math.add(i64, whole_micros, micros_fraction) catch return null;
}

fn parseDigits(text: []const u8) ?u16 {
    if (text.len == 0) return null;
    var value: u16 = 0;
    for (text) |byte| {
        if (!std.ascii.isDigit(byte)) return null;
        value = value * 10 + (byte - '0');
    }
    return value;
}

fn daysInMonth(year: u16, month: u16) u16 {
    return switch (month) {
        1, 3, 5, 7, 8, 10, 12 => 31,
        4, 6, 9, 11 => 30,
        2 => if (@mod(year, 4) == 0 and (@mod(year, 100) != 0 or @mod(year, 400) == 0)) 29 else 28,
        else => 0,
    };
}

// Howard Hinnant's civil-date transform, relative to 1970-01-01.
fn daysFromCivil(year_in: i64, month_in: i64, day: i64) i64 {
    var year = year_in;
    if (month_in <= 2) year -= 1;
    const era = @divFloor(year, 400);
    const yoe = year - era * 400;
    const month = month_in + (if (month_in > 2) @as(i64, -3) else 9);
    const doy = @divFloor(153 * month + 2, 5) + day - 1;
    const doe = yoe * 365 + @divFloor(yoe, 4) - @divFloor(yoe, 100) + doy;
    return era * 146097 + doe - 719468;
}

fn trim(value: []const u8) []const u8 {
    return std.mem.trim(u8, value, " \t\r\n\x0b\x0c");
}

fn truncateDiagnostic(value: []const u8, buffer: *[max_diagnostic_value_len + 3]u8) []const u8 {
    if (value.len <= max_diagnostic_value_len) return value;
    @memcpy(buffer[0..max_diagnostic_value_len], value[0..max_diagnostic_value_len]);
    @memcpy(buffer[max_diagnostic_value_len..], "…");
    return buffer;
}

const Collector = struct {
    const TestRow = struct {
        offset: u64,
        did: []u8,
        collection: []u8,
        rkey: []u8,
        scope: Scope,
        cid: ?Cid,
        timestamp_micros: i64,
    };

    rows: std.ArrayList(TestRow) = .empty,
    rejects_seen: usize = 0,

    fn onRow(raw: ?*anyopaque, row: Row) !void {
        const self: *Collector = @ptrCast(@alignCast(raw.?));
        const allocator = std.testing.allocator;
        const did = try allocator.dupe(u8, row.did);
        errdefer allocator.free(did);
        const collection = try allocator.dupe(u8, row.collection);
        errdefer allocator.free(collection);
        const rkey = try allocator.dupe(u8, row.rkey);
        errdefer allocator.free(rkey);
        try self.rows.append(allocator, .{
            .offset = row.offset,
            .did = did,
            .collection = collection,
            .rkey = rkey,
            .scope = row.scope,
            .cid = row.cid,
            .timestamp_micros = row.timestamp_micros,
        });
    }

    fn onReject(raw: ?*anyopaque, _: Reject) void {
        const self: *Collector = @ptrCast(@alignCast(raw.?));
        self.rejects_seen += 1;
    }

    fn deinit(self: *Collector) void {
        for (self.rows.items) |row| {
            std.testing.allocator.free(row.did);
            std.testing.allocator.free(row.collection);
            std.testing.allocator.free(row.rkey);
        }
        self.rows.deinit(std.testing.allocator);
    }
};

fn collect(source: []const u8, collector: *Collector, sample_limit: usize) !Stats {
    var reader: std.Io.Reader = .fixed(source);
    return parse(std.testing.allocator, &reader, .{
        .context = collector,
        .on_row = Collector.onRow,
        .on_reject = Collector.onReject,
        .reject_sample_limit = sample_limit,
    });
}

test "valid rows preserve offsets, defaults, quoted fields, and exact CID" {
    const testing = std.testing;
    const valid_cid = "bafyreigbtj4x7ip5legnfznufuopl4sg4knzc2cof6duas4b3q2fy6swua";
    const row_one = "at://did:plc:alice/app.bsky.feed.post/r1,2022-01-02T03:04:05Z,,\r\n";
    const source = "uri,timestamp,scope,cid\r\n" ++ row_one ++
        "\"at://did:plc:bob/app.bsky.feed.like/r2\",2023-06-07T10:09:10+02:00,specific_version," ++ valid_cid ++ "\r\n";
    var collector: Collector = .{};
    defer collector.deinit();
    var stats = try collect(source, &collector, 0);
    defer stats.deinit(testing.allocator);

    try testing.expectEqual(@as(u64, 2), stats.rows_valid);
    try testing.expectEqual(@as(usize, 2), collector.rows.items.len);
    const first = collector.rows.items[0];
    try testing.expectEqualStrings("did:plc:alice", first.did);
    try testing.expectEqual(@as(i64, 1_641_092_645_000_000), first.timestamp_micros);
    try testing.expectEqual("uri,timestamp,scope,cid\r\n".len, first.offset);
    try testing.expect(std.mem.startsWith(u8, source[first.offset..], row_one));
    try testing.expectEqual(Scope.all_versions, first.scope);
    try testing.expectEqual(null, first.cid);
    const second = collector.rows.items[1];
    try testing.expectEqual(@as(i64, 1_686_125_350_000_000), second.timestamp_micros);
    try testing.expectEqual(Scope.specific_version, second.scope);
    try testing.expectEqual(@as(u8, 0x71), second.cid.?.codec);
}

test "bad data rows are exhaustively rejected without aborting later rows" {
    const testing = std.testing;
    const source =
        "uri,timestamp,scope,cid\n" ++
        "at://did:plc:ok/app.bsky.feed.post/r1,2022-01-02T03:04:05Z,,\n" ++
        "not-a-uri,2022-01-02T03:04:05Z,,\n" ++
        "at://alice.example/app.bsky.feed.post/r,2022-01-02T03:04:05Z,,\n" ++
        "at://did:plc:pathless,2022-01-02T03:04:05Z,,\n" ++
        "at://did:plc:time/app.bsky.feed.post/r,not-a-time,,\n" ++
        "at://did:plc:epoch/app.bsky.feed.post/r,1970-01-01T00:00:00Z,,\n" ++
        "at://did:plc:scope/app.bsky.feed.post/r,2022-01-02T03:04:05Z,bogus,\n" ++
        ",2022-01-02T03:04:05Z,,\n" ++
        "at://did:plc:missing/app.bsky.feed.post/r,,,\n" ++
        "at://did:plc:cid/app.bsky.feed.post/r,2022-01-02T03:04:05Z,specific_version,\n" ++
        "at://did:plc:badcid/app.bsky.feed.post/r,2022-01-02T03:04:05Z,specific_version,nope\n" ++
        "at://did:plc:fields/app.bsky.feed.post/r,2022-01-02T03:04:05Z\n" ++
        "at://did:plc:later/app.bsky.feed.post/r2,2023-01-02T03:04:05Z,,\n";
    var collector: Collector = .{};
    defer collector.deinit();
    var stats = try collect(source, &collector, 4);
    defer stats.deinit(testing.allocator);

    try testing.expectEqual(@as(u64, 13), stats.rows_total);
    try testing.expectEqual(@as(u64, 2), stats.rows_valid);
    try testing.expectEqual(@as(u64, 11), stats.rows_rejected);
    try testing.expectEqual(@as(usize, 11), collector.rejects_seen);
    try testing.expectEqual(@as(usize, 4), stats.reject_sample.items.len);
    try testing.expectEqual(@as(u64, 3), stats.rejected(.missing_field));
    try testing.expectEqual(@as(u64, 1), stats.rejected(.bad_uri));
    try testing.expectEqual(@as(u64, 1), stats.rejected(.uri_authority_not_did));
    try testing.expectEqual(@as(u64, 1), stats.rejected(.uri_missing_collection_or_rkey));
    try testing.expectEqual(@as(u64, 1), stats.rejected(.bad_timestamp));
    try testing.expectEqual(@as(u64, 1), stats.rejected(.non_positive_timestamp));
    try testing.expectEqual(@as(u64, 1), stats.rejected(.unknown_scope));
    try testing.expectEqual(@as(u64, 1), stats.rejected(.specific_version_missing_cid));
    try testing.expectEqual(@as(u64, 1), stats.rejected(.bad_cid));
    try testing.expectEqualStrings("did:plc:later", collector.rows.items[1].did);
}

test "header is strict, order independent, and case insensitive" {
    const testing = std.testing;
    const invalid = [_][]const u8{
        "",
        "timestamp,scope\n",
        "uri,scope\n",
        "uri,uri,timestamp\n",
        "uri,timestamp,timestmp\n",
    };
    for (invalid) |source| {
        var reader: std.Io.Reader = .fixed(source);
        var collector: Collector = .{};
        defer collector.deinit();
        try testing.expectError(error.InvalidHeader, parse(testing.allocator, &reader, .{
            .context = &collector,
            .on_row = Collector.onRow,
        }));
    }

    var collector: Collector = .{};
    defer collector.deinit();
    var stats = try collect(
        " Timestamp , URI , CID , Scope \n" ++
            "2022-01-02T03:04:05Z,at://did:plc:a/app.bsky.feed.post/r,,all_versions\n",
        &collector,
        Options.no_reject_sample,
    );
    defer stats.deinit(testing.allocator);
    try testing.expectEqual(@as(u64, 1), stats.rows_valid);
}

test "quoted multiline rows and escaped quotes are RFC 4180 decoded" {
    const testing = std.testing;
    // A newline inside the invalid timestamp remains one logical rejected row;
    // the escaped quote is decoded and diagnosed, and the next row survives.
    const source = "uri,timestamp\n" ++
        "at://did:plc:a/app.bsky.feed.post/r,\"bad\r\n\"\"timestamp\"\n" ++
        "at://did:plc:b/app.bsky.feed.post/r,2022-01-02T03:04:05.123456789Z\n";
    var collector: Collector = .{};
    defer collector.deinit();
    var stats = try collect(source, &collector, 0);
    defer stats.deinit(testing.allocator);
    try testing.expectEqual(@as(u64, 1), stats.rows_rejected);
    try testing.expectEqual(@as(u64, 1), stats.rows_valid);
    try testing.expectEqual(@as(i64, 1_641_092_645_123_456), collector.rows.items[0].timestamp_micros);
}

test "unbalanced or bare quote fails loudly after prior rows" {
    const testing = std.testing;
    const source = "uri,timestamp\n" ++
        "at://did:plc:a/app.bsky.feed.post/r,2022-01-02T03:04:05Z\n" ++
        "at://did:plc:b/app.bsky.feed.post/\"r,2022-01-02T03:04:05Z\n" ++
        "at://did:plc:c/app.bsky.feed.post/r,2022-01-02T03:04:05Z\n";
    var reader: std.Io.Reader = .fixed(source);
    var collector: Collector = .{};
    defer collector.deinit();
    try testing.expectError(error.UnbalancedQuote, parse(testing.allocator, &reader, .{
        .context = &collector,
        .on_row = Collector.onRow,
    }));
    try testing.expectEqual(@as(usize, 1), collector.rows.items.len);
}

test "parseInto preserves partial statistics on fatal structure error" {
    const testing = std.testing;
    const source = "uri,timestamp\n" ++
        "at://did:plc:a/app.bsky.feed.post/r,2022-01-02T03:04:05Z\n" ++
        "\"unterminated\n";
    var reader: std.Io.Reader = .fixed(source);
    var collector: Collector = .{};
    defer collector.deinit();
    var stats: Stats = .{};
    defer stats.deinit(testing.allocator);
    try testing.expectError(error.UnbalancedQuote, parseInto(testing.allocator, &reader, .{
        .context = &collector,
        .on_row = Collector.onRow,
    }, &stats));
    try testing.expectEqual(@as(u64, 1), stats.rows_total);
    try testing.expectEqual(@as(u64, 1), stats.rows_valid);
}

test "row footprint is bounded and blank lines remain part of seek offset" {
    const testing = std.testing;
    var source: std.ArrayList(u8) = .empty;
    defer source.deinit(testing.allocator);
    try source.appendSlice(testing.allocator, "uri,timestamp\n\n\r\n");
    const expected_offset = "uri,timestamp\n".len;
    try source.appendSlice(testing.allocator, "at://did:plc:huge/app.bsky.feed.post/");
    try source.appendNTimes(testing.allocator, 'x', max_row_bytes);
    try source.appendSlice(testing.allocator, ",2022-01-02T03:04:05Z\n");
    try source.appendSlice(testing.allocator, "at://did:plc:ok/app.bsky.feed.post/r,2022-01-02T03:04:05Z\n");

    var collector: Collector = .{};
    defer collector.deinit();
    var stats = try collect(source.items, &collector, 0);
    defer stats.deinit(testing.allocator);
    try testing.expectEqual(@as(u64, 1), stats.rejected(.row_too_long));
    try testing.expectEqual(@as(u64, 1), stats.rows_valid);
    try testing.expectEqual(@as(u64, 2), stats.rows_total);
    try testing.expectEqual(@as(u64, expected_offset), stats.reject_sample.items[0].offset);
}

test "RFC3339 conversion matches Go edge semantics" {
    const testing = std.testing;
    try testing.expectEqual(@as(?i64, 1_641_092_645_123_456), parseRfc3339Micros("2022-01-02T03:04:05.1234567890Z"));
    try testing.expectEqual(@as(?i64, 1_641_092_645_100_000), parseRfc3339Micros("2022-01-02T03:04:05,1Z"));
    try testing.expectEqual(@as(?i64, 1_641_006_245_000_000), parseRfc3339Micros("2022-01-02T03:04:05+24:00"));
    try testing.expectEqual(@as(?i64, 1_641_089_045_000_000), parseRfc3339Micros("2022-01-02T03:04:05+00:60"));
    try testing.expectEqual(@as(?i64, null), parseRfc3339Micros("2022-01-02t03:04:05Z"));
    try testing.expectEqual(@as(?i64, null), parseRfc3339Micros("2022-01-02T03:04:05z"));
    try testing.expectEqual(@as(?i64, null), parseRfc3339Micros("2022-01-02T03:04:60Z"));
}

test "reject sample and callback are independently bounded" {
    const testing = std.testing;
    var source: std.ArrayList(u8) = .empty;
    defer source.deinit(testing.allocator);
    try source.appendSlice(testing.allocator, "uri,timestamp\n");
    for (0..150) |_| try source.appendSlice(testing.allocator, "bad,bad\n");
    var collector: Collector = .{};
    defer collector.deinit();
    var stats = try collect(source.items, &collector, 10);
    defer stats.deinit(testing.allocator);
    try testing.expectEqual(@as(usize, 150), collector.rejects_seen);
    try testing.expectEqual(@as(usize, 10), stats.reject_sample.items.len);
}

test "streaming parse performs no heap allocation when reject sampling is disabled" {
    const testing = std.testing;
    const Counter = struct {
        count: usize = 0,
        fn onRow(raw: ?*anyopaque, _: Row) !void {
            const self: *@This() = @ptrCast(@alignCast(raw.?));
            self.count += 1;
        }
    };

    var source: std.ArrayList(u8) = .empty;
    defer source.deinit(testing.allocator);
    try source.appendSlice(testing.allocator, "uri,timestamp\n");
    for (0..10_000) |index| {
        var row_buffer: [128]u8 = undefined;
        const row = try std.fmt.bufPrint(
            &row_buffer,
            "at://did:plc:bounded/app.bsky.feed.post/{d},2022-01-02T03:04:05Z\n",
            .{index},
        );
        try source.appendSlice(testing.allocator, row);
    }

    var reader: std.Io.Reader = .fixed(source.items);
    var counter: Counter = .{};
    var stats = try parse(testing.failing_allocator, &reader, .{
        .context = &counter,
        .on_row = Counter.onRow,
        .reject_sample_limit = Options.no_reject_sample,
    });
    defer stats.deinit(testing.failing_allocator);
    try testing.expectEqual(@as(usize, 10_000), counter.count);
    try testing.expectEqual(@as(u64, 10_000), stats.rows_valid);
}