//! Update operators: $set, $unset, $inc, $push, $pull, $rename with
//! dot-path navigation, including MongoDB's three positional forms.
//! Mutates the document's pairs in place, allocating from the document's
//! own arena.
//!
//! A path with a positional segment names a *set* of concrete paths rather
//! than one: `y.$[].b` on a two-element array names `y.0.b` and `y.1.b`.
//! `resolve` turns the former into the latter against the document at hand,
//! and every operator then walks the concrete paths it already knew how to
//! walk. Nothing below `resolve` knows a positional segment exists.

const std = @import("std");
const bson = @import("bson.zig");
const query = @import("query.zig");

pub const UpdateError = error{
    ImmutableId,
    InvalidUpdate,
    /// A path segment that is not a field to create: a non-numeric name
    /// applied to an array, or any name applied to a scalar element a
    /// positional segment selected. mongod's `PathNotViable`.
    PathNotViable,
    /// `$`, `$[]` or `$[<identifier>]` as a path's first segment, where there
    /// is no array above it to descend into.
    PositionalFirst,
    /// More than one `$` in one path. Only the first can be resolved: the
    /// query records one matched element per document, not one per level.
    TooManyPositional,
    /// `$[<identifier>]` naming an identifier that no array filter binds.
    NoArrayFilter,
    /// An array filter whose top-level field name is not `[a-z][a-zA-Z0-9]*`.
    BadArrayFilterIdentifier,
    /// An array filter whose identifier no path in the update mentions. A
    /// filter that selects nothing is a typo, so mongod refuses rather than
    /// quietly ignoring it -- and so a `$[i]` misspelt on one side or the
    /// other is caught by whichever half it broke.
    UnusedArrayFilter,
    /// Two array filters binding the same identifier.
    DuplicateArrayFilter,
    /// An array filter with no top-level field name to bind.
    EmptyArrayFilter,
    /// An array filter whose top-level fields name two different identifiers.
    MultipleArrayFilterIdentifiers,
    /// A positional segment whose array is absent from the document. Array
    /// updates address what is there; unlike `$set` on a plain path, they do
    /// not bring it into being.
    ArrayPathRequired,
    /// A positional segment whose path names something that is not an array.
    NotAnArrayPath,
    /// `$` where the query holds no predicate on the array, or none that any
    /// element satisfies. `$` writes *the element the query matched*, so with
    /// no such predicate there is nothing for it to mean.
    NoPositionalMatch,
    /// `$rename` with a positional source or destination. Both ends of a
    /// rename are refused by mongod: a rename moves a field, and a positional
    /// path names an element rather than a field.
    RenameDynamicSource,
    RenameDynamicDestination,
    OutOfMemory,
};

/// What a refusal was about, so the reply can name it instead of saying "bad
/// update". Mostly borrowed from the update document, which outlives the call;
/// `ArrayPathRequired` and `NotAnArrayPath` name a *resolved* prefix
/// (`y.1.c`), which is allocated from the same arena as the document copy.
pub const Diagnostic = struct {
    path: []const u8 = "",
    segment: []const u8 = "",
    /// A second name, for the one message that mentions two.
    other: []const u8 = "",
};

fn note(diag: ?*Diagnostic, path: []const u8, segment: []const u8) void {
    if (diag) |d| d.* = .{ .path = path, .segment = segment };
}

/// One `arrayFilters` entry: the identifier `$[<identifier>]` spells, and the
/// predicate that decides which elements it selects.
///
/// `used` is written by `validate` as it walks the update's paths, and read
/// once afterwards -- which is the whole of `UnusedArrayFilter`.
pub const ArrayFilter = struct {
    /// Filled in by `validate` from `pairs`, not by the caller.
    ident: []const u8 = "",
    /// The predicate, with `ident` still on the front of every key.
    pairs: []const bson.Pair,
    used: bool = false,
};

pub const Options = struct {
    /// Bound to the `$[<identifier>]` segments of the update's paths.
    array_filters: []ArrayFilter = &.{},
    /// The command's query. `$` resolves against it, so an update carrying one
    /// without this is refused rather than guessing at an element.
    query: ?[]const bson.Pair = null,
    diag: ?*Diagnostic = null,
};

const max_path_segments = 16;

/// The wrapper field an element is matched under. A predicate written against
/// an element (`{"i.b": 3}`, `{"y.b": 3}`) becomes one written against
/// `{elem_key: <element>}`, which is a document the query engine can answer
/// about without knowing anything about arrays or identifiers.
const elem_key = "e";

/// One segment of a parsed path.
const Segment = union(enum) {
    literal: []const u8,
    /// `$[]` -- every element.
    all,
    /// `$[<identifier>]` -- the elements its array filter selects, by index
    /// into `Options.array_filters`.
    filtered: usize,
    /// `$` -- the one element the query matched.
    first,
};

const Path = struct {
    text: []const u8,
    segs: [max_path_segments]Segment,
    n: usize,
};

fn is_positional_segment(seg: []const u8) bool {
    if (std.mem.eql(u8, seg, "$")) return true;
    return seg.len >= 3 and std.mem.startsWith(u8, seg, "$[") and seg[seg.len - 1] == ']';
}

fn has_positional(path: []const u8) bool {
    var it = std.mem.splitScalar(u8, path, '.');
    while (it.next()) |seg| if (is_positional_segment(seg)) return true;
    return false;
}

/// mongod's rule for an array filter identifier, measured on 8.3.7: `aB2` is
/// accepted, `Ab`, `a_b` and `1x` are not.
fn valid_identifier(ident: []const u8) bool {
    if (ident.len == 0) return false;
    if (ident[0] < 'a' or ident[0] > 'z') return false;
    for (ident[1..]) |c| if (!std.ascii.isAlphanumeric(c)) return false;
    return true;
}

fn find_filter(filters: []const ArrayFilter, ident: []const u8) ?usize {
    for (filters, 0..) |f, i| if (std.mem.eql(u8, f.ident, ident)) return i;
    return null;
}

/// Split a path into segments, classifying the positional ones and binding
/// each identifier to its array filter.
///
/// This is the static half of a positional update: it depends on the update
/// and the filters, never on a document, so it answers the same way for every
/// document a multi-update touches.
fn parse_path(text: []const u8, opts: Options) UpdateError!Path {
    var p: Path = .{ .text = text, .segs = undefined, .n = 0 };
    var dollars: usize = 0;
    var it = std.mem.splitScalar(u8, text, '.');
    while (it.next()) |seg| {
        if (p.n >= max_path_segments) return error.InvalidUpdate;
        p.segs[p.n] = seg: {
            if (!is_positional_segment(seg)) break :seg .{ .literal = seg };
            if (p.n == 0) {
                note(opts.diag, text, seg);
                return error.PositionalFirst;
            }
            if (std.mem.eql(u8, seg, "$")) {
                dollars += 1;
                break :seg .first;
            }
            const ident = seg[2 .. seg.len - 1];
            if (ident.len == 0) break :seg .all;
            const found = find_filter(opts.array_filters, ident) orelse {
                note(opts.diag, text, ident);
                return error.NoArrayFilter;
            };
            opts.array_filters[found].used = true;
            break :seg .{ .filtered = found };
        };
        p.n += 1;
    }
    if (dollars > 1) {
        note(opts.diag, text, "$");
        return error.TooManyPositional;
    }
    return p;
}

/// Everything about an update that can be refused without reading a document:
/// the array filters, then every path they bind into.
///
/// Called by `apply`, and separately by the command handlers *before* they
/// scan for matches -- an update naming an identifier nothing binds is
/// refused whether or not it would have matched anything, which is also what
/// makes `UnusedArrayFilter` observable on a filter that matches no document.
pub fn validate(update: []const bson.Pair, opts: Options) UpdateError!void {
    // A replacement carries data, not paths, so nothing here applies to it --
    // which is also why mongod ignores `arrayFilters` alongside one, the
    // single case out of seventeen where this server already agreed with it.
    if (is_replacement(update)) return;
    try bind_array_filters(opts);
    for (update) |op| {
        const ops = doc_pairs(op.value) orelse continue;
        const rename = std.mem.eql(u8, op.key, "$rename");
        for (ops) |p| {
            if (rename) {
                if (has_positional(p.key)) {
                    note(opts.diag, p.key, "");
                    return error.RenameDynamicSource;
                }
                if (p.value == .string and has_positional(p.value.string)) {
                    note(opts.diag, p.value.string, "");
                    return error.RenameDynamicDestination;
                }
                continue;
            }
            _ = try parse_path(p.key, opts);
        }
    }
    for (opts.array_filters) |f| {
        if (f.used) continue;
        note(opts.diag, "", f.ident);
        return error.UnusedArrayFilter;
    }
}

/// Read each array filter's identifier off its top-level field names.
///
/// `{"i.b": 3}` binds `i`; `{"i.b": 3, "i.c": 1}` also binds `i` and is legal;
/// `{"i.b": 3, "j.b": 1}` names two and is not. Measured, all four.
fn bind_array_filters(opts: Options) UpdateError!void {
    for (opts.array_filters) |*f| {
        var ident: ?[]const u8 = null;
        for (f.pairs) |p| {
            const name = p.key[0 .. std.mem.indexOfScalar(u8, p.key, '.') orelse p.key.len];
            if (ident) |first| {
                if (std.mem.eql(u8, first, name)) continue;
                if (opts.diag) |d| d.* = .{ .segment = first, .other = name };
                return error.MultipleArrayFilterIdentifiers;
            }
            ident = name;
        }
        const name = ident orelse return error.EmptyArrayFilter;
        if (!valid_identifier(name)) {
            note(opts.diag, "", name);
            return error.BadArrayFilterIdentifier;
        }
        f.ident = name;
    }
    for (opts.array_filters, 0..) |f, i| {
        for (opts.array_filters[i + 1 ..]) |g| {
            if (!std.mem.eql(u8, f.ident, g.ident)) continue;
            note(opts.diag, "", f.ident);
            return error.DuplicateArrayFilter;
        }
    }
}

/// Whether an update document is a *replacement* rather than a set of
/// operators. MongoDB decides on the first field and nothing else: a
/// `$`-prefixed one means operators. An empty document is a replacement, and a
/// legal one -- it strips every field but `_id`.
///
/// This distinction is the whole of `replaceOne`, `findOneAndReplace` and
/// `bulkWrite`'s `replaceOne`. Without it `apply` rejected every update whose
/// first key was not `$`-prefixed, so all three failed with "bad update".
pub fn is_replacement(pairs: []const bson.Pair) bool {
    if (pairs.len == 0) return true;
    return !is_operator_key(pairs[0].key);
}

fn is_operator_key(key: []const u8) bool {
    return key.len > 0 and key[0] == '$';
}

/// Apply an update document to `doc`: a replacement, or a set of operators.
pub fn apply(
    doc: *bson.Document,
    update: *const bson.Document,
    opts: Options,
) UpdateError!void {
    if (is_replacement(update.pairs)) return apply_replacement(doc, update.pairs);
    // Up front rather than at the point of use, because one update names
    // several paths: validating as we walk would refuse the third path having
    // already rewritten what the first two named.
    try validate(update.pairs, opts);
    const arena = doc.arena.allocator();
    var pairs = try copy_to_list(bson.Pair, arena, doc.pairs);
    for (update.pairs) |op| {
        // The first field decided this is an operator update, so a field that
        // is not one is a mixed document -- which MongoDB rejects rather than
        // guessing at.
        if (!is_operator_key(op.key)) return error.InvalidUpdate;
        try apply_operator(arena, &pairs, op.key, op.value, opts);
    }
    doc.pairs = try pairs.toOwnedSlice(arena);
}

/// The concrete paths one update path names in `root`.
///
/// A path with no positional segment names itself, and is returned unresolved
/// -- including when nothing along it exists, because `$set` on a plain path
/// creates what it needs. A positional segment is different: it addresses
/// elements that are already there, so an absent or non-array path is a
/// refusal rather than a creation.
fn resolve(
    arena: std.mem.Allocator,
    root: []const bson.Pair,
    text: []const u8,
    opts: Options,
) UpdateError![]const []const []const u8 {
    const path = try parse_path(text, opts);
    var out: std.ArrayListUnmanaged([]const []const u8) = .empty;
    var walk = Walk{ .arena = arena, .opts = opts, .path = &path, .out = &out };
    try walk.descend(.{ .doc = root }, 0);
    return out.items;
}

/// Expands one parsed path into the concrete paths it names, depth first, so
/// the results come out in document order.
const Walk = struct {
    arena: std.mem.Allocator,
    opts: Options,
    path: *const Path,
    out: *std.ArrayListUnmanaged([]const []const u8),
    /// The concrete segments chosen so far. Positional ones hold the index
    /// they resolved to, which is what makes `y.1.c` the prefix a refusal
    /// names.
    buf: [max_path_segments][]const u8 = undefined,

    fn descend(self: *Walk, container: ?bson.Value, i: usize) UpdateError!void {
        if (i == self.path.n) {
            try self.out.append(self.arena, try self.arena.dupe([]const u8, self.buf[0..i]));
            return;
        }
        switch (self.path.segs[i]) {
            .literal => |lit| {
                self.buf[i] = lit;
                try self.descend(child_value(container, lit), i + 1);
            },
            else => try self.spread(container, i),
        }
    }

    /// A positional segment: one step down, several ways.
    fn spread(self: *Walk, container: ?bson.Value, i: usize) UpdateError!void {
        const value = container orelse {
            note(self.opts.diag, try self.prefix_text(i), "");
            return error.ArrayPathRequired;
        };
        const arr = switch (value) {
            .array => |a| a,
            else => {
                note(self.opts.diag, try self.prefix_text(i), "");
                return error.NotAnArrayPath;
            },
        };
        switch (self.path.segs[i]) {
            .all => for (arr, 0..) |elem, k| try self.element(elem, k, i),
            .filtered => |fi| {
                const f = self.opts.array_filters[fi];
                const pred = try rewrite_keys(self.arena, f.pairs, f.ident.len);
                for (arr, 0..) |elem, k| {
                    if (matches_element(self.arena, .{ .wrapped = pred }, elem)) {
                        try self.element(elem, k, i);
                    }
                }
            },
            .first => {
                const k = try self.first_match(arr, i);
                try self.element(arr[k], k, i);
            },
            .literal => unreachable,
        }
    }

    fn element(self: *Walk, elem: bson.Value, k: usize, i: usize) UpdateError!void {
        // A scalar element with more path below it is not a field to create.
        // Without this the walk would hand `y.1.b` to `set_path`, which would
        // replace the `7` at `y.1` with `{b: 9}` -- the same class of silent
        // destruction the positional forms themselves used to cause.
        if (i + 1 < self.path.n) switch (self.path.segs[i + 1]) {
            .literal => |lit| switch (elem) {
                .doc, .array => {},
                else => {
                    note(self.opts.diag, self.path.text, lit);
                    return error.PathNotViable;
                },
            },
            else => {},
        };
        self.buf[i] = try std.fmt.allocPrint(self.arena, "{d}", .{k});
        try self.descend(elem, i + 1);
    }

    /// The resolved path above segment `i`: `y`, or `y.1.c` under a positional
    /// segment that already chose an element. This is the path a refusal
    /// names, and mongod names the same one.
    fn prefix_text(self: *Walk, i: usize) UpdateError![]const u8 {
        return std.mem.join(self.arena, ".", self.buf[0..i]);
    }

    /// Which element the query matched, for `$`.
    ///
    /// The query is written against the *unresolved* path -- `{"y.b": 3}`
    /// selects an element of `y` however deep a positional segment above it
    /// went -- so the predicates are gathered by the path with its positional
    /// segments elided, and then run against the concrete array here.
    fn first_match(self: *Walk, arr: []const bson.Value, i: usize) UpdateError!usize {
        const q = self.opts.query orelse return self.no_match();
        var preds: std.ArrayListUnmanaged(ElemPred) = .empty;
        try collect_preds(self.arena, q, try self.query_prefix(i), &preds);
        if (preds.items.len == 0) return self.no_match();
        for (arr, 0..) |elem, k| {
            var all = true;
            for (preds.items) |p| {
                if (!matches_element(self.arena, p, elem)) {
                    all = false;
                    break;
                }
            }
            if (all) return k;
        }
        // Two predicates on one array need not agree on an element, and the
        // document still matched: `{"y.b": 3, "y.c": 2}` matches
        // `[{b: 3, c: 1}, {b: 1, c: 2}]` without either element satisfying
        // both. mongod answers 1 here; this answers 0. Recorded in PLAN §6
        // rather than guessed at -- the rule mongod uses is an artefact of
        // which predicate last wrote its match position.
        for (arr, 0..) |elem, k| {
            for (preds.items) |p| if (matches_element(self.arena, p, elem)) return k;
        }
        return self.no_match();
    }

    fn no_match(self: *Walk) UpdateError {
        note(self.opts.diag, self.path.text, "$");
        return error.NoPositionalMatch;
    }

    fn query_prefix(self: *Walk, i: usize) UpdateError![]const u8 {
        var parts: [max_path_segments][]const u8 = undefined;
        var n: usize = 0;
        for (self.path.segs[0..i]) |seg| switch (seg) {
            .literal => |lit| {
                parts[n] = lit;
                n += 1;
            },
            else => {},
        };
        return std.mem.join(self.arena, ".", parts[0..n]);
    }
};

fn child_value(container: ?bson.Value, key: []const u8) ?bson.Value {
    return switch (container orelse return null) {
        .doc => |sub| bson.get_pair(sub, key),
        .array => |arr| blk: {
            const index = parse_index(key) orelse break :blk null;
            break :blk if (index < arr.len) arr[index] else null;
        },
        else => null,
    };
}

/// A predicate about one array element.
const ElemPred = union(enum) {
    /// Keys rewritten onto `elem_key`, matched against `{e: <element>}`.
    wrapped: []const bson.Pair,
    /// `$elemMatch`'s own document, matched against the element directly.
    direct: []const bson.Pair,
};

fn matches_element(arena: std.mem.Allocator, pred: ElemPred, elem: bson.Value) bool {
    switch (pred) {
        .wrapped => |pairs| {
            const wrapper = [_]bson.Pair{.{ .key = elem_key, .value = elem }};
            return query.matches(
                arena,
                &.{ .arena = undefined, .pairs = pairs },
                &.{ .arena = undefined, .pairs = &wrapper },
            ) catch false;
        },
        .direct => |pairs| {
            const sub = switch (elem) {
                .doc => |d| d,
                else => return false,
            };
            return query.matches(
                arena,
                &.{ .arena = undefined, .pairs = pairs },
                &.{ .arena = undefined, .pairs = sub },
            ) catch false;
        },
    }
}

/// Re-point a predicate's keys from whatever they were written against onto
/// `elem_key`: `i.b` with `prefix_len` 1 becomes `e.b`, and `i` becomes `e`.
fn rewrite_keys(
    arena: std.mem.Allocator,
    pairs: []const bson.Pair,
    prefix_len: usize,
) UpdateError![]const bson.Pair {
    const out = try arena.alloc(bson.Pair, pairs.len);
    for (pairs, 0..) |p, i| {
        out[i] = .{
            .key = try std.mem.concat(arena, u8, &.{ elem_key, p.key[prefix_len..] }),
            .value = p.value,
        };
    }
    return out;
}

/// The query's predicates about elements of the array at `prefix`.
///
/// `$and` is descended into because a driver writes one whenever two
/// predicates share a field; `$or` is not, because an element satisfying one
/// branch says nothing about the document having matched through it.
fn collect_preds(
    arena: std.mem.Allocator,
    q: []const bson.Pair,
    prefix: []const u8,
    out: *std.ArrayListUnmanaged(ElemPred),
) UpdateError!void {
    for (q) |p| {
        if (std.mem.eql(u8, p.key, "$and")) {
            const branches = switch (p.value) {
                .array => |a| a,
                else => continue,
            };
            for (branches) |b| switch (b) {
                .doc => |sub| try collect_preds(arena, sub, prefix, out),
                else => {},
            };
            continue;
        }
        if (!std.mem.startsWith(u8, p.key, prefix)) continue;
        const rest = p.key[prefix.len..];
        if (rest.len != 0 and rest[0] != '.') continue;
        if (rest.len == 0) {
            if (p.value == .doc) {
                if (bson.get_pair(p.value.doc, "$elemMatch")) |em| {
                    if (em == .doc) {
                        try out.append(arena, .{ .direct = em.doc });
                        continue;
                    }
                }
            }
        }
        try out.append(arena, .{ .wrapped = try rewrite_keys(arena, &.{p}, prefix.len) });
    }
}

/// Replace every field of `doc` with `replacement`'s, except `_id`.
///
/// `_id` is immutable, so it survives and keeps its position at the front (which
/// is also where MongoDB stores it, and what the `_id_` index descends on). A
/// replacement carrying an `_id` is allowed only when it is the *same* `_id`;
/// anything else is an attempt to change a document's identity by rewriting it.
///
/// `doc` may legitimately have no `_id` yet: that is the upsert path, where the
/// caller has seeded the document from the filter's equalities and `insert`
/// generates an ObjectId afterwards. Then the replacement's own `_id`, if it has
/// one, is what the new document gets.
fn apply_replacement(doc: *bson.Document, replacement: []const bson.Pair) UpdateError!void {
    const arena = doc.arena.allocator();
    const old_id = doc.get("_id");

    var pairs: std.ArrayListUnmanaged(bson.Pair) = .empty;
    errdefer pairs.deinit(arena);
    try pairs.ensureTotalCapacity(arena, replacement.len + 1);

    if (old_id) |id| {
        try pairs.append(arena, .{
            .key = try arena.dupe(u8, "_id"),
            .value = try bson.copy_value(arena, id),
        });
    }
    for (replacement) |p| {
        // A replacement is data, not instructions. `{$set: {...}}` reaching here
        // would mean the first field was not an operator and a later one was,
        // i.e. a mixed document.
        if (is_operator_key(p.key)) return error.InvalidUpdate;
        if (std.mem.eql(u8, p.key, "_id")) {
            if (old_id) |id| {
                if (bson.compare(id, p.value) != .eq) return error.ImmutableId;
                continue; // already at the front
            }
            // No stored `_id`: this replacement supplies it.
            try pairs.insert(arena, 0, .{
                .key = try arena.dupe(u8, "_id"),
                .value = try bson.copy_value(arena, p.value),
            });
            continue;
        }
        try pairs.append(arena, .{
            .key = try arena.dupe(u8, p.key),
            .value = try bson.copy_value(arena, p.value),
        });
    }
    doc.pairs = try pairs.toOwnedSlice(arena);
}

fn apply_operator(
    arena: std.mem.Allocator,
    pairs: *std.ArrayListUnmanaged(bson.Pair),
    op: []const u8,
    value: bson.Value,
    opts: Options,
) UpdateError!void {
    const diag = opts.diag;
    if (std.mem.eql(u8, op, "$set")) {
        const ops = doc_pairs(value) orelse return error.InvalidUpdate;
        for (ops) |p| {
            if (std.mem.eql(u8, p.key, "_id")) return error.ImmutableId;
            for (try resolve(arena, pairs.items, p.key, opts)) |segs| {
                try set_path(arena, pairs, segs, try bson.copy_value(arena, p.value), p.key, diag);
            }
        }
        return;
    }
    if (std.mem.eql(u8, op, "$unset")) {
        const ops = doc_pairs(value) orelse return error.InvalidUpdate;
        for (ops) |p| {
            for (try resolve(arena, pairs.items, p.key, opts)) |segs| {
                unset_path(arena, pairs, segs);
            }
        }
        return;
    }
    if (std.mem.eql(u8, op, "$inc")) {
        const ops = doc_pairs(value) orelse return error.InvalidUpdate;
        for (ops) |p| {
            for (try resolve(arena, pairs.items, p.key, opts)) |segs| {
                const current = get_value(pairs.items, segs) orelse bson.Value{ .int32 = 0 };
                if (!current.is_number() or !p.value.is_number()) return error.InvalidUpdate;
                const sum = try numeric_add(current, p.value);
                try set_path(arena, pairs, segs, sum, p.key, diag);
            }
        }
        return;
    }
    if (std.mem.eql(u8, op, "$push")) {
        const ops = doc_pairs(value) orelse return error.InvalidUpdate;
        for (ops) |p| {
            for (try resolve(arena, pairs.items, p.key, opts)) |segs| {
                const current_opt = get_value(pairs.items, segs);
                var items: std.ArrayListUnmanaged(bson.Value) = .empty;
                defer items.deinit(arena);
                if (current_opt) |current| {
                    switch (current) {
                        .array => |arr| try items.appendSlice(arena, arr),
                        .null => {},
                        else => return error.InvalidUpdate, // non-array field
                    }
                }
                if (p.value == .doc) {
                    if (bson.get_pair(p.value.doc, "$each")) |each| {
                        const arr = switch (each) {
                            .array => |a| a,
                            else => return error.InvalidUpdate,
                        };
                        for (arr) |item| try items.append(arena, try bson.copy_value(arena, item));
                        try set_path(arena, pairs, segs, .{ .array = try items.toOwnedSlice(arena) }, p.key, diag);
                        continue;
                    }
                }
                try items.append(arena, try bson.copy_value(arena, p.value));
                try set_path(arena, pairs, segs, .{ .array = try items.toOwnedSlice(arena) }, p.key, diag);
            }
        }
        return;
    }
    if (std.mem.eql(u8, op, "$pull")) {
        const ops = doc_pairs(value) orelse return error.InvalidUpdate;
        for (ops) |p| {
            for (try resolve(arena, pairs.items, p.key, opts)) |segs| {
                const current = get_value(pairs.items, segs) orelse continue;
                const arr = switch (current) {
                    .array => |a| a,
                    else => return error.InvalidUpdate,
                };
                var items: std.ArrayListUnmanaged(bson.Value) = .empty;
                defer items.deinit(arena);
                for (arr) |elem| {
                    if (!pull_matches(arena, p.value, elem)) {
                        try items.append(arena, elem);
                    }
                }
                try set_path(arena, pairs, segs, .{ .array = try items.toOwnedSlice(arena) }, p.key, diag);
            }
        }
        return;
    }
    // `$rename` alone keeps the plain split: `validate` has already refused a
    // positional path on either end of it, which is what mongod does too.
    if (std.mem.eql(u8, op, "$rename")) {
        const ops = doc_pairs(value) orelse return error.InvalidUpdate;
        for (ops) |p| {
            if (p.value != .string) return error.InvalidUpdate;
            if (std.mem.eql(u8, p.key, "_id") or std.mem.eql(u8, p.value.string, "_id")) return error.ImmutableId;
            var old_segs: [max_path_segments][]const u8 = undefined;
            const old_n = split_path(p.key, &old_segs) orelse return error.InvalidUpdate;
            const v = get_value(pairs.items, old_segs[0..old_n]) orelse continue; // no-op when absent
            unset_path(arena, pairs, old_segs[0..old_n]);
            var new_segs: [max_path_segments][]const u8 = undefined;
            const new_n = split_path(p.value.string, &new_segs) orelse return error.InvalidUpdate;
            try set_path(arena, pairs, new_segs[0..new_n], v, p.value.string, diag);
        }
        return;
    }
    return error.InvalidUpdate;
}

fn doc_pairs(v: bson.Value) ?[]const bson.Pair {
    return switch (v) {
        .doc => |pairs| pairs,
        else => null,
    };
}

fn split_path(path: []const u8, out: *[max_path_segments][]const u8) ?usize {
    var n: usize = 0;
    var it = std.mem.splitScalar(u8, path, '.');
    while (it.next()) |seg| {
        if (n >= max_path_segments) return null;
        out[n] = seg;
        n += 1;
    }
    return n;
}

fn parse_index(seg: []const u8) ?usize {
    return std.fmt.parseInt(usize, seg, 10) catch null;
}

/// Shallow-copy a slice into a growable list backed by `arena`.
fn copy_to_list(
    comptime T: type,
    arena: std.mem.Allocator,
    items: []const T,
) UpdateError!std.ArrayListUnmanaged(T) {
    var out: std.ArrayListUnmanaged(T) = .empty;
    errdefer out.deinit(arena);
    try out.appendSlice(arena, items);
    return out;
}

const find_pair = bson.get_pair_index;

fn get_value(pairs: []const bson.Pair, segs: []const []const u8) ?bson.Value {
    const idx = find_pair(pairs, segs[0]) orelse return null;
    if (segs.len == 1) return pairs[idx].value;
    return switch (pairs[idx].value) {
        .doc => |sub| get_value(sub, segs[1..]),
        .array => |arr| blk: {
            const index = parse_index(segs[1]) orelse break :blk null;
            if (index >= arr.len) break :blk null;
            if (segs.len == 2) break :blk arr[index];
            break :blk switch (arr[index]) {
                .doc => |sub| get_value(sub, segs[2..]),
                else => null,
            };
        },
        else => null,
    };
}

fn set_path(
    arena: std.mem.Allocator,
    pairs: *std.ArrayListUnmanaged(bson.Pair),
    segs: []const []const u8,
    value: bson.Value,
    path: []const u8,
    diag: ?*Diagnostic,
) UpdateError!void {
    if (segs.len == 1) {
        if (find_pair(pairs.items, segs[0])) |idx| {
            pairs.items[idx].value = value;
        } else {
            try pairs.append(arena, .{ .key = try arena.dupe(u8, segs[0]), .value = value });
        }
        return;
    }
    const idx = find_pair(pairs.items, segs[0]) orelse {
        const is_array = parse_index(segs[1]) != null;
        try pairs.append(arena, .{ .key = try arena.dupe(u8, segs[0]), .value = if (is_array) .{ .array = &.{} } else .{ .doc = &.{} } });
        return set_path(arena, pairs, segs, value, path, diag);
    };
    switch (pairs.items[idx].value) {
        .doc => |sub| {
            var sub_pairs = try copy_to_list(bson.Pair, arena, sub);
            defer sub_pairs.deinit(arena);
            try set_path(arena, &sub_pairs, segs[1..], value, path, diag);
            pairs.items[idx].value = .{ .doc = try sub_pairs.toOwnedSlice(arena) };
        },
        .array => |arr| {
            const index = parse_index(segs[1]) orelse {
                // A non-numeric segment under an array is never a field to
                // create. This branch used to read "treat as non-array:
                // replace with a doc" and did exactly that -- `y.nope.b`
                // turned `y: [{b: 3}]` into `y: {nope: {b: 2}}`, discarding
                // every element and answering ok: 1. mongod refuses with
                // PathNotViable and leaves the document alone.
                //
                // The positional spellings took this same branch and are the
                // reason it was found; they are refused earlier, by
                // `reject_positional`. What reaches here is the rest of the
                // class: a plain field name, and a `$`-prefixed one that is
                // not positional.
                note(diag, path, segs[1]);
                return error.PathNotViable;
            };
            var items = try copy_to_list(bson.Value, arena, arr);
            defer items.deinit(arena);
            if (index >= items.items.len) {
                try items.appendNTimes(arena, .null, index + 1 - items.items.len);
            }
            if (segs.len == 2) {
                items.items[index] = value;
            } else {
                switch (items.items[index]) {
                    .doc => |sub| {
                        var sub_pairs = try copy_to_list(bson.Pair, arena, sub);
                        defer sub_pairs.deinit(arena);
                        try set_path(arena, &sub_pairs, segs[2..], value, path, diag);
                        items.items[index] = .{ .doc = try sub_pairs.toOwnedSlice(arena) };
                    },
                    else => {
                        var sub_pairs: std.ArrayListUnmanaged(bson.Pair) = .empty;
                        defer sub_pairs.deinit(arena);
                        try set_path(arena, &sub_pairs, segs[2..], value, path, diag);
                        items.items[index] = .{ .doc = try sub_pairs.toOwnedSlice(arena) };
                    },
                }
            }
            pairs.items[idx].value = .{ .array = try items.toOwnedSlice(arena) };
        },
        else => {
            var sub_pairs: std.ArrayListUnmanaged(bson.Pair) = .empty;
            defer sub_pairs.deinit(arena);
            try set_path(arena, &sub_pairs, segs[1..], value, path, diag);
            pairs.items[idx].value = .{ .doc = try sub_pairs.toOwnedSlice(arena) };
        },
    }
}

fn unset_path(
    arena: std.mem.Allocator,
    pairs: *std.ArrayListUnmanaged(bson.Pair),
    segs: []const []const u8,
) void {
    if (segs.len == 1) {
        if (find_pair(pairs.items, segs[0])) |idx| {
            _ = pairs.orderedRemove(idx);
        }
        return;
    }
    const idx = find_pair(pairs.items, segs[0]) orelse return;
    switch (pairs.items[idx].value) {
        .doc => |sub| {
            var sub_pairs = copy_to_list(bson.Pair, arena, sub) catch return;
            unset_path(arena, &sub_pairs, segs[1..]);
            pairs.items[idx].value = .{ .doc = sub_pairs.items };
        },
        // Unsetting an element leaves a null in its place rather than
        // shortening the array: measured on both `$unset: {"y.0": ""}` and
        // `$unset: {"y.$[]": ""}`, which is the same rule reached two ways.
        .array => |arr| {
            const index = parse_index(segs[1]) orelse return;
            if (index >= arr.len) return;
            var items = copy_to_list(bson.Value, arena, arr) catch return;
            if (segs.len == 2) {
                items.items[index] = .null;
            } else switch (items.items[index]) {
                .doc => |sub| {
                    var sub_pairs = copy_to_list(bson.Pair, arena, sub) catch return;
                    unset_path(arena, &sub_pairs, segs[2..]);
                    items.items[index] = .{ .doc = sub_pairs.items };
                },
                else => return,
            }
            pairs.items[idx].value = .{ .array = items.items };
        },
        else => {},
    }
}

fn pull_matches(arena: std.mem.Allocator, condition: bson.Value, elem: bson.Value) bool {
    switch (condition) {
        .doc => |cond_pairs| {
            const elem_doc = switch (elem) {
                .doc => |pairs| pairs,
                else => return false,
            };
            if (query.all_operator_keys(cond_pairs)) {
                // Operator condition against the element's value at each
                // operator's field — treat element doc as the doc.
                var ok = true;
                for (cond_pairs) |p| {
                    const actuals = bson.get_pair(elem_doc, p.key[1..]);
                    const a: bson.Value = actuals orelse .null;
                    if (!(query.value_matches_operator(arena, p.key, p.value, a) catch false)) ok = false;
                }
                return ok;
            }
            return (query.matches(arena, &.{ .arena = undefined, .pairs = cond_pairs }, &.{ .arena = undefined, .pairs = elem_doc }) catch false);
        },
        else => return bson.compare(condition, elem) == .eq,
    }
}

fn numeric_add(a: bson.Value, b: bson.Value) UpdateError!bson.Value {
    if (a == .double or b == .double) {
        const sum: f64 = @floatCast(a.as_f128() + b.as_f128());
        return .{ .double = sum };
    }
    if (a == .int64 or b == .int64) {
        const av: i64 = switch (a) {
            .int32 => |i| i,
            .int64 => |i| i,
            else => unreachable,
        };
        const bv: i64 = switch (b) {
            .int32 => |i| i,
            .int64 => |i| i,
            else => unreachable,
        };
        const sum = std.math.add(i64, av, bv) catch return error.InvalidUpdate;
        return .{ .int64 = sum };
    }
    const sum: i64 = @as(i64, a.int32) + b.int32;
    if (sum >= std.math.minInt(i32) and sum <= std.math.maxInt(i32)) {
        return .{ .int32 = @intCast(sum) };
    }
    return .{ .int64 = sum };
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

const testing = std.testing;

fn doc_of(pairs: []const bson.Pair) bson.Document {
    return .{ .arena = undefined, .pairs = pairs };
}

test "$set, $inc, $unset, $rename" {
    var doc = bson.Document{ .arena = std.heap.ArenaAllocator.init(testing.allocator), .pairs = &.{} };
    defer doc.arena.deinit();
    doc.pairs = try doc.arena.allocator().dupe(bson.Pair, &.{
        .{ .key = "a", .value = .{ .int32 = 1 } },
        .{ .key = "user", .value = .{ .doc = &.{
            .{ .key = "name", .value = .{ .string = "bob" } },
            .{ .key = "age", .value = .{ .int32 = 30 } },
        } } },
        .{ .key = "gone", .value = .{ .int32 = 9 } },
    });

    try apply(&doc, &doc_of(&.{
        .{ .key = "$set", .value = .{ .doc = &.{
            .{ .key = "user.name", .value = .{ .string = "alice" } },
            .{ .key = "user.city", .value = .{ .string = "NYC" } },
            .{ .key = "new", .value = .{ .int32 = 5 } },
        } } },
        .{ .key = "$inc", .value = .{ .doc = &.{.{ .key = "user.age", .value = .{ .int32 = 2 } }} } },
        .{ .key = "$unset", .value = .{ .doc = &.{.{ .key = "gone", .value = .{ .string = "" } }} } },
        .{ .key = "$rename", .value = .{ .doc = &.{.{ .key = "new", .value = .{ .string = "renamed" } }} } },
    }), .{});

    const user = bson.get_pair(doc.pairs, "user").?;
    try testing.expectEqualStrings("alice", user.doc[0].value.string);
    try testing.expectEqual(@as(i64, 32), bson.get_pair(user.doc, "age").?.int32);
    try testing.expectEqualStrings("NYC", bson.get_pair(user.doc, "city").?.string);
    try testing.expect(bson.get_pair(doc.pairs, "gone") == null);
    try testing.expect(bson.get_pair(doc.pairs, "new") == null);
    try testing.expectEqual(@as(i64, 5), bson.get_pair(doc.pairs, "renamed").?.int32);
}

test "$push and $pull" {
    var doc = bson.Document{ .arena = std.heap.ArenaAllocator.init(testing.allocator), .pairs = &.{} };
    defer doc.arena.deinit();
    doc.pairs = try doc.arena.allocator().dupe(bson.Pair, &.{
        .{ .key = "tags", .value = .{ .array = &.{ .{ .string = "a" }, .{ .string = "b" } } } },
    });

    try apply(&doc, &doc_of(&.{
        .{ .key = "$push", .value = .{ .doc = &.{.{ .key = "tags", .value = .{ .string = "c" } }} } },
    }), .{});
    try testing.expectEqual(@as(usize, 3), bson.get_pair(doc.pairs, "tags").?.array.len);
    try testing.expectEqualStrings("c", bson.get_pair(doc.pairs, "tags").?.array[2].string);

    try apply(&doc, &doc_of(&.{
        .{ .key = "$pull", .value = .{ .doc = &.{.{ .key = "tags", .value = .{ .string = "b" } }} } },
    }), .{});
    const tags = bson.get_pair(doc.pairs, "tags").?.array;
    try testing.expectEqual(@as(usize, 2), tags.len);
    try testing.expectEqualStrings("a", tags[0].string);
    try testing.expectEqualStrings("c", tags[1].string);

    try apply(&doc, &doc_of(&.{
        .{ .key = "$push", .value = .{ .doc = &.{.{ .key = "tags", .value = .{ .doc = &.{.{ .key = "$each", .value = .{ .array = &.{ .{ .string = "x" }, .{ .string = "y" } } } }} } }} } },
    }), .{});
    try testing.expectEqual(@as(usize, 4), bson.get_pair(doc.pairs, "tags").?.array.len);
}

test "$set nested creation and _id protection" {
    var doc = bson.Document{ .arena = std.heap.ArenaAllocator.init(testing.allocator), .pairs = &.{} };
    defer doc.arena.deinit();
    doc.pairs = try doc.arena.allocator().dupe(bson.Pair, &.{
        .{ .key = "_id", .value = .{ .int32 = 1 } },
    });

    try apply(&doc, &doc_of(&.{
        .{ .key = "$set", .value = .{ .doc = &.{
            .{ .key = "a.b.c", .value = .{ .int32 = 42 } },
            .{ .key = "arr.1", .value = .{ .string = "x" } },
        } } },
    }), .{});
    const a = bson.get_pair(doc.pairs, "a").?;
    const b = bson.get_pair(a.doc, "b").?;
    try testing.expectEqual(@as(i64, 42), bson.get_pair(b.doc, "c").?.int32);
    const arr = bson.get_pair(doc.pairs, "arr").?.array;
    try testing.expectEqual(@as(usize, 2), arr.len);
    try testing.expectEqualStrings("x", arr[1].string);

    try testing.expectError(error.ImmutableId, apply(&doc, &doc_of(&.{
        .{ .key = "$set", .value = .{ .doc = &.{.{ .key = "_id", .value = .{ .int32 = 2 } }} } },
    }), .{}));
}

test "a replacement keeps _id and drops every other field" {
    // Mutation check: seed `apply_replacement`'s list from `doc.pairs` (as the
    // operator path does) instead of starting empty. Red -- `gone` survives, and
    // a replacement that does not remove fields is not a replacement.
    var doc = bson.Document{ .arena = std.heap.ArenaAllocator.init(testing.allocator), .pairs = &.{} };
    defer doc.arena.deinit();
    doc.pairs = try doc.arena.allocator().dupe(bson.Pair, &.{
        .{ .key = "_id", .value = .{ .int32 = 7 } },
        .{ .key = "gone", .value = .{ .int32 = 1 } },
        .{ .key = "also_gone", .value = .{ .string = "x" } },
    });

    try apply(&doc, &doc_of(&.{
        .{ .key = "fresh", .value = .{ .int32 = 42 } },
    }), .{});

    try testing.expectEqual(@as(usize, 2), doc.pairs.len);
    // _id survives, and stays at the front where it is stored.
    try testing.expectEqualStrings("_id", doc.pairs[0].key);
    try testing.expectEqual(@as(i64, 7), doc.pairs[0].value.int32);
    try testing.expectEqual(@as(i64, 42), bson.get_pair(doc.pairs, "fresh").?.int32);
    try testing.expect(bson.get_pair(doc.pairs, "gone") == null);
    try testing.expect(bson.get_pair(doc.pairs, "also_gone") == null);
}

test "an empty replacement leaves a document holding only its _id" {
    var doc = bson.Document{ .arena = std.heap.ArenaAllocator.init(testing.allocator), .pairs = &.{} };
    defer doc.arena.deinit();
    doc.pairs = try doc.arena.allocator().dupe(bson.Pair, &.{
        .{ .key = "_id", .value = .{ .int32 = 7 } },
        .{ .key = "a", .value = .{ .int32 = 1 } },
    });

    // Legal, and the reason `is_replacement` treats an empty document as one
    // rather than as a no-op set of operators.
    try apply(&doc, &doc_of(&.{}), .{});
    try testing.expectEqual(@as(usize, 1), doc.pairs.len);
    try testing.expectEqualStrings("_id", doc.pairs[0].key);
}

test "a replacement may repeat the _id it is replacing, but not change it" {
    // Mutation check: drop the `bson.compare` in `apply_replacement`. Red on the
    // second half -- a replacement would silently rewrite a document's identity,
    // and since the `_id_` index is keyed on it, the stored entry and the stored
    // document would disagree.
    var doc = bson.Document{ .arena = std.heap.ArenaAllocator.init(testing.allocator), .pairs = &.{} };
    defer doc.arena.deinit();
    const original = [_]bson.Pair{
        .{ .key = "_id", .value = .{ .int32 = 7 } },
        .{ .key = "a", .value = .{ .int32 = 1 } },
    };
    doc.pairs = try doc.arena.allocator().dupe(bson.Pair, &original);

    // The same _id, restated: accepted.
    try apply(&doc, &doc_of(&.{
        .{ .key = "_id", .value = .{ .int32 = 7 } },
        .{ .key = "b", .value = .{ .int32 = 2 } },
    }), .{});
    try testing.expectEqual(@as(i64, 7), doc.pairs[0].value.int32);
    try testing.expectEqual(@as(i64, 2), bson.get_pair(doc.pairs, "b").?.int32);
    // And only once, not twice.
    try testing.expectEqual(@as(usize, 2), doc.pairs.len);

    // A different _id: refused.
    try testing.expectError(error.ImmutableId, apply(&doc, &doc_of(&.{
        .{ .key = "_id", .value = .{ .int32 = 8 } },
    }), .{}));

    // Equal across numeric types is the same _id, matching the canonical key
    // encoding the _id_ index descends on.
    try apply(&doc, &doc_of(&.{
        .{ .key = "_id", .value = .{ .double = 7.0 } },
        .{ .key = "c", .value = .{ .int32 = 3 } },
    }), .{});
    try testing.expectEqual(@as(i64, 3), bson.get_pair(doc.pairs, "c").?.int32);
}

test "a replacement supplies the _id when the document has none" {
    // The upsert path: the caller seeds a document from the filter's equalities
    // and the replacement carries the _id.
    //
    // Mutation check: make the `old_id == null` arm `continue` without adding the
    // pair. Red -- the upserted document loses the _id the client asked for and
    // gets a generated ObjectId instead, so a retried upsert inserts twice.
    var doc = bson.Document{ .arena = std.heap.ArenaAllocator.init(testing.allocator), .pairs = &.{} };
    defer doc.arena.deinit();

    try apply(&doc, &doc_of(&.{
        .{ .key = "a", .value = .{ .int32 = 1 } },
        .{ .key = "_id", .value = .{ .int32 = 99 } },
    }), .{});

    try testing.expectEqual(@as(usize, 2), doc.pairs.len);
    try testing.expectEqualStrings("_id", doc.pairs[0].key);
    try testing.expectEqual(@as(i64, 99), doc.pairs[0].value.int32);
}

test "a mixed update document is refused from either side" {
    // Mutation check: drop the `is_operator_key` guard in `apply`'s operator loop
    // (first case) or in `apply_replacement`'s loop (second). Each leaves one
    // half green and the other red.
    var doc = bson.Document{ .arena = std.heap.ArenaAllocator.init(testing.allocator), .pairs = &.{} };
    defer doc.arena.deinit();

    // Starts with an operator, so operators are expected throughout.
    try testing.expectError(error.InvalidUpdate, apply(&doc, &doc_of(&.{
        .{ .key = "$set", .value = .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 1 } }} } },
        .{ .key = "plain", .value = .{ .int32 = 1 } },
    }), .{}));

    // Starts with data, so it is a replacement and an operator has no meaning.
    try testing.expectError(error.InvalidUpdate, apply(&doc, &doc_of(&.{
        .{ .key = "plain", .value = .{ .int32 = 1 } },
        .{ .key = "$set", .value = .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 1 } }} } },
    }), .{}));
}

test "is_replacement decides on the first field only" {
    try testing.expect(is_replacement(&.{}));
    try testing.expect(is_replacement(&.{.{ .key = "a", .value = .{ .int32 = 1 } }}));
    try testing.expect(!is_replacement(&.{.{ .key = "$set", .value = .{ .doc = &.{} } }}));
    // An empty key cannot be an operator, so it reads as data -- and the
    // replacement path then stores it, which is what MongoDB does with it.
    try testing.expect(is_replacement(&.{.{ .key = "", .value = .{ .int32 = 1 } }}));
}

fn doc_with(gpa: std.mem.Allocator, pairs: []const bson.Pair) !bson.Document {
    var doc = bson.Document{ .arena = std.heap.ArenaAllocator.init(gpa), .pairs = &.{} };
    doc.pairs = try doc.arena.allocator().dupe(bson.Pair, pairs);
    return doc;
}

/// A document with one array field, rebuilt per case so a refusal can be
/// checked against untouched bytes.
fn array_doc(arena: std.mem.Allocator) !bson.Document {
    return doc_with(arena, &.{
        .{ .key = "y", .value = .{ .array = &.{
            .{ .doc = &.{.{ .key = "b", .value = .{ .int32 = 3 } }} },
            .{ .doc = &.{.{ .key = "b", .value = .{ .int32 = 1 } }} },
        } } },
    });
}

/// `$set` of one path, the shape most of the cases below need. Owns its two
/// pairs: a helper *returning* the document would hand back a slice of its own
/// dead frame.
const SetOne = struct {
    inner: [1]bson.Pair,
    outer: [1]bson.Pair = undefined,

    fn init(path: []const u8, value: bson.Value) SetOne {
        return .{ .inner = .{.{ .key = path, .value = value }} };
    }

    fn doc(self: *SetOne) bson.Document {
        self.outer = .{.{ .key = "$set", .value = .{ .doc = &self.inner } }};
        return doc_of(&self.outer);
    }
};

fn apply_set(
    doc: *bson.Document,
    path: []const u8,
    value: bson.Value,
    opts: Options,
) UpdateError!void {
    var u = SetOne.init(path, value);
    return apply(doc, &u.doc(), opts);
}

/// The `b` of `y[k]`, which is what most of these updates move.
fn y_b(doc: *const bson.Document, k: usize) !i32 {
    const y = doc.get("y") orelse return error.TestUnexpectedResult;
    if (y != .array or k >= y.array.len) return error.TestUnexpectedResult;
    const elem = y.array[k];
    if (elem != .doc) return error.TestUnexpectedResult;
    return (bson.get_pair(elem.doc, "b") orelse return error.TestUnexpectedResult).int32;
}

fn expect_y_untouched(doc: *const bson.Document) !void {
    const y = doc.get("y") orelse return error.TestUnexpectedResult;
    // The array is still an array. Before this refusal existed it was a
    // document keyed by the path segment's literal text, and everything in it
    // was gone.
    try testing.expect(y == .array);
    try testing.expectEqual(@as(usize, 2), y.array.len);
    try testing.expectEqual(@as(i32, 3), y.array[0].doc[0].value.int32);
    try testing.expectEqual(@as(i32, 1), y.array[1].doc[0].value.int32);
}

test "$[] writes every element of the array" {
    // The all-positional operator, which the pinned crud corpus does not
    // contain a single case of. Mutation check: make `spread`'s `.all` arm
    // stop after the first element and the second half goes red.
    var doc = try array_doc(testing.allocator);
    defer doc.arena.deinit();
    try apply_set(&doc, "y.$[].b", .{ .int32 = 9 }, .{});
    try testing.expectEqual(@as(i32, 9), try y_b(&doc, 0));
    try testing.expectEqual(@as(i32, 9), try y_b(&doc, 1));
}

test "every operator goes through a positional segment, not just $set" {
    // The destruction this replaced was below the operator, in the shared path
    // walk: `$inc` through `$[]` stored its operand instead of incrementing.
    // So the walk has to be below the operator too.
    var doc = try array_doc(testing.allocator);
    defer doc.arena.deinit();
    try apply(&doc, &doc_of(&.{
        .{ .key = "$inc", .value = .{ .doc = &.{.{ .key = "y.$[].b", .value = .{ .int32 = 10 } }} } },
    }), .{});
    try testing.expectEqual(@as(i32, 13), try y_b(&doc, 0));
    try testing.expectEqual(@as(i32, 11), try y_b(&doc, 1));

    // `$unset` through an element removes the field; `$unset` *of* an element
    // leaves a null in its place rather than shortening the array.
    try apply(&doc, &doc_of(&.{
        .{ .key = "$unset", .value = .{ .doc = &.{.{ .key = "y.$[].b", .value = .{ .string = "" } }} } },
    }), .{});
    try testing.expectEqual(@as(usize, 0), doc.get("y").?.array[0].doc.len);
    try apply(&doc, &doc_of(&.{
        .{ .key = "$unset", .value = .{ .doc = &.{.{ .key = "y.$[]", .value = .{ .string = "" } }} } },
    }), .{});
    try testing.expect(doc.get("y").?.array[0] == .null);
    try testing.expect(doc.get("y").?.array[1] == .null);
}

test "$[] over an empty array writes nothing at all" {
    // Zero concrete paths, so the operator never runs -- which is how the
    // document comes back byte-identical and the reply says modified 0.
    var doc = try doc_with(testing.allocator, &.{
        .{ .key = "y", .value = .{ .array = &.{} } },
    });
    defer doc.arena.deinit();
    try apply_set(&doc, "y.$[].b", .{ .int32 = 9 }, .{});
    try testing.expectEqual(@as(usize, 0), doc.get("y").?.array.len);
}

test "nested $[] is a cross product over both levels" {
    var doc = try doc_with(testing.allocator, &.{
        .{ .key = "y", .value = .{ .array = &.{
            .{ .doc = &.{.{ .key = "c", .value = .{ .array = &.{
                .{ .doc = &.{.{ .key = "d", .value = .{ .int32 = 1 } }} },
                .{ .doc = &.{.{ .key = "d", .value = .{ .int32 = 2 } }} },
            } } }} },
            .{ .doc = &.{.{ .key = "c", .value = .{ .array = &.{
                .{ .doc = &.{.{ .key = "d", .value = .{ .int32 = 3 } }} },
            } } }} },
        } } },
    });
    defer doc.arena.deinit();
    try apply_set(&doc, "y.$[].c.$[].d", .{ .int32 = 0 }, .{});
    const y = doc.get("y").?.array;
    for (y) |outer| {
        for (outer.doc[0].value.array) |inner| {
            try testing.expectEqual(@as(i32, 0), inner.doc[0].value.int32);
        }
    }
}

test "an array update addresses an array that is there, and refuses one that is not" {
    // The difference between a positional segment and a plain one: `$set` on
    // `a.b.c` creates what it needs, and this does not. Both refusals are code
    // 2 on mongod, with different text, so they are different errors here.
    var missing = try doc_with(testing.allocator, &.{.{ .key = "z", .value = .{ .int32 = 1 } }});
    defer missing.arena.deinit();
    var diag: Diagnostic = .{};
    try testing.expectError(
        error.ArrayPathRequired,
        apply_set(&missing, "y.$[].b", .{ .int32 = 9 }, .{ .diag = &diag }),
    );
    try testing.expectEqualStrings("y", diag.path);

    var scalar = try doc_with(testing.allocator, &.{.{ .key = "y", .value = .{ .int32 = 5 } }});
    defer scalar.arena.deinit();
    try testing.expectError(
        error.NotAnArrayPath,
        apply_set(&scalar, "y.$[].b", .{ .int32 = 9 }, .{ .diag = &diag }),
    );
    try testing.expectEqualStrings("y", diag.path);

    // Under a positional segment the refusal names the *resolved* prefix, so
    // it says which element was missing the array. mongod names the same one.
    var nested = try doc_with(testing.allocator, &.{
        .{ .key = "y", .value = .{ .array = &.{
            .{ .doc = &.{.{ .key = "c", .value = .{ .array = &.{} } }} },
            .{ .doc = &.{.{ .key = "b", .value = .{ .int32 = 2 } }} },
        } } },
    });
    defer nested.arena.deinit();
    try testing.expectError(
        error.ArrayPathRequired,
        apply_set(&nested, "y.$[].c.$[].d", .{ .int32 = 0 }, .{ .diag = &diag }),
    );
    try testing.expectEqualStrings("y.1.c", diag.path);
}

test "a scalar element with more path below it is PathNotViable" {
    // Mutation check: delete the `.literal` arm of `Walk.element` and this
    // update answers ok having replaced the `7` with `{b: 9}` -- a smaller
    // version of the destruction the whole walk exists to have stopped.
    var doc = try doc_with(testing.allocator, &.{
        .{ .key = "y", .value = .{ .array = &.{
            .{ .doc = &.{.{ .key = "b", .value = .{ .int32 = 1 } }} },
            .{ .int32 = 7 },
        } } },
    });
    defer doc.arena.deinit();
    try testing.expectError(
        error.PathNotViable,
        apply_set(&doc, "y.$[].b", .{ .int32 = 9 }, .{}),
    );
    // And the first element, which the walk had already resolved, is untouched
    // too: the refusal is taken during resolution, before any write.
    try testing.expectEqual(@as(i32, 1), try y_b(&doc, 0));
    try testing.expect(doc.get("y").?.array[1] == .int32);
}

test "$[<identifier>] writes only the elements its array filter selects" {
    var doc = try array_doc(testing.allocator);
    defer doc.arena.deinit();
    var filters = [_]ArrayFilter{
        .{ .pairs = &.{.{ .key = "i.b", .value = .{ .int32 = 3 } }} },
    };
    try apply_set(&doc, "y.$[i].b", .{ .int32 = 2 }, .{ .array_filters = &filters });
    try testing.expectEqual(@as(i32, 2), try y_b(&doc, 0));
    try testing.expectEqual(@as(i32, 1), try y_b(&doc, 1));
    try testing.expect(filters[0].used);

    // The identifier as the leaf replaces the whole element, and a filter on
    // the identifier itself is how an array of scalars is addressed.
    var scalars = try doc_with(testing.allocator, &.{
        .{ .key = "y", .value = .{ .array = &.{ .{ .int32 = 1 }, .{ .int32 = 2 }, .{ .int32 = 3 }, .{ .int32 = 2 } } } },
    });
    defer scalars.arena.deinit();
    var by_value = [_]ArrayFilter{
        .{ .pairs = &.{.{ .key = "i", .value = .{ .int32 = 2 } }} },
    };
    try apply_set(&scalars, "y.$[i]", .{ .int32 = 9 }, .{ .array_filters = &by_value });
    const y = scalars.get("y").?.array;
    try testing.expectEqual(@as(i32, 1), y[0].int32);
    try testing.expectEqual(@as(i32, 9), y[1].int32);
    try testing.expectEqual(@as(i32, 3), y[2].int32);
    try testing.expectEqual(@as(i32, 9), y[3].int32);
}

test "an array filter that selects nothing leaves the document alone" {
    // Not an error: matched, modified nothing. The reply's `modifiedCount` is
    // the only place this shows, which is why the corpus records outcomes.
    var doc = try array_doc(testing.allocator);
    defer doc.arena.deinit();
    var filters = [_]ArrayFilter{
        .{ .pairs = &.{.{ .key = "i.b", .value = .{ .int32 = 4 } }} },
    };
    try apply_set(&doc, "y.$[i].b", .{ .int32 = 2 }, .{ .array_filters = &filters });
    try expect_y_untouched(&doc);
}

test "an identifier and its array filter each refuse the other's absence" {
    var doc = try array_doc(testing.allocator);
    defer doc.arena.deinit();
    var diag: Diagnostic = .{};

    // A path naming an identifier nothing binds.
    var one = [_]ArrayFilter{.{ .pairs = &.{.{ .key = "i.b", .value = .{ .int32 = 3 } }} }};
    try testing.expectError(
        error.NoArrayFilter,
        apply_set(&doc, "y.$[k].b", .{ .int32 = 2 }, .{ .array_filters = &one, .diag = &diag }),
    );
    try testing.expectEqualStrings("k", diag.segment);
    try testing.expectEqualStrings("y.$[k].b", diag.path);

    // A filter no path names. Mutation check: drop the `used` loop at the end
    // of `validate` and this goes green -- and a misspelt identifier on the
    // filter side becomes a silent no-op update.
    var unused = [_]ArrayFilter{.{ .pairs = &.{.{ .key = "i.b", .value = .{ .int32 = 3 } }} }};
    try testing.expectError(
        error.UnusedArrayFilter,
        apply_set(&doc, "y.b", .{ .int32 = 2 }, .{ .array_filters = &unused, .diag = &diag }),
    );
    try testing.expectEqualStrings("i", diag.segment);
    try expect_y_untouched(&doc);
}

test "an array filter names exactly one identifier, spelled the one legal way" {
    var doc = try array_doc(testing.allocator);
    defer doc.arena.deinit();
    var diag: Diagnostic = .{};

    var empty = [_]ArrayFilter{.{ .pairs = &.{} }};
    try testing.expectError(
        error.EmptyArrayFilter,
        apply_set(&doc, "y.$[i].b", .{ .int32 = 2 }, .{ .array_filters = &empty }),
    );

    var two = [_]ArrayFilter{.{ .pairs = &.{
        .{ .key = "i.b", .value = .{ .int32 = 3 } },
        .{ .key = "j.b", .value = .{ .int32 = 1 } },
    } }};
    try testing.expectError(
        error.MultipleArrayFilterIdentifiers,
        apply_set(&doc, "y.$[i].b", .{ .int32 = 2 }, .{ .array_filters = &two, .diag = &diag }),
    );
    try testing.expectEqualStrings("i", diag.segment);
    try testing.expectEqualStrings("j", diag.other);

    var dup = [_]ArrayFilter{
        .{ .pairs = &.{.{ .key = "i.b", .value = .{ .int32 = 3 } }} },
        .{ .pairs = &.{.{ .key = "i.b", .value = .{ .int32 = 1 } }} },
    };
    try testing.expectError(
        error.DuplicateArrayFilter,
        apply_set(&doc, "y.$[i].b", .{ .int32 = 2 }, .{ .array_filters = &dup }),
    );

    // Two fields naming the *same* identifier are one predicate, not two
    // identifiers -- measured, and the reason the check is on the name rather
    // than on the count of fields.
    var same = [_]ArrayFilter{.{ .pairs = &.{
        .{ .key = "i.b", .value = .{ .int32 = 3 } },
        .{ .key = "i.c", .value = .{ .int32 = 1 } },
    } }};
    try apply_set(&doc, "y.$[i].b", .{ .int32 = 2 }, .{ .array_filters = &same });
    try expect_y_untouched(&doc); // no element has both, so nothing moved

    for ([_][]const u8{ "1x", "Ab", "a_b" }) |bad| {
        var f = [_]ArrayFilter{.{ .pairs = &.{.{ .key = bad, .value = .{ .int32 = 3 } }} }};
        try testing.expectError(
            error.BadArrayFilterIdentifier,
            apply_set(&doc, "y.$[i].b", .{ .int32 = 2 }, .{ .array_filters = &f, .diag = &diag }),
        );
        try testing.expectEqualStrings(bad, diag.segment);
    }
}

test "$ writes the one element the query matched" {
    // The operator that needs something the document alone does not hold.
    var doc = try doc_with(testing.allocator, &.{
        .{ .key = "y", .value = .{ .array = &.{
            .{ .doc = &.{.{ .key = "b", .value = .{ .int32 = 1 } }} },
            .{ .doc = &.{.{ .key = "b", .value = .{ .int32 = 1 } }} },
            .{ .doc = &.{.{ .key = "b", .value = .{ .int32 = 2 } }} },
        } } },
    });
    defer doc.arena.deinit();
    // Two elements qualify and only the first is written. Mutation check: make
    // `first_match` return the last index instead and this goes red both ways.
    try apply_set(&doc, "y.$.b", .{ .int32 = 7 }, .{
        .query = &.{.{ .key = "y.b", .value = .{ .int32 = 1 } }},
    });
    try testing.expectEqual(@as(i32, 7), try y_b(&doc, 0));
    try testing.expectEqual(@as(i32, 1), try y_b(&doc, 1));
    try testing.expectEqual(@as(i32, 2), try y_b(&doc, 2));
}

test "$ takes the element from a predicate on any field of it, including an operator" {
    var doc = try array_doc(testing.allocator);
    defer doc.arena.deinit();
    // `{y.b: {$lt: 2}}` matches the second element, not the first.
    try apply_set(&doc, "y.$.b", .{ .int32 = 7 }, .{
        .query = &.{.{ .key = "y.b", .value = .{ .doc = &.{
            .{ .key = "$lt", .value = .{ .int32 = 2 } },
        } } }},
    });
    try testing.expectEqual(@as(i32, 3), try y_b(&doc, 0));
    try testing.expectEqual(@as(i32, 7), try y_b(&doc, 1));

    // `$elemMatch` names the element directly, and is the one predicate shape
    // that is matched against the element rather than through a wrapper.
    var em = try array_doc(testing.allocator);
    defer em.arena.deinit();
    try apply_set(&em, "y.$.b", .{ .int32 = 7 }, .{
        .query = &.{.{ .key = "y", .value = .{ .doc = &.{
            .{ .key = "$elemMatch", .value = .{ .doc = &.{.{ .key = "b", .value = .{ .int32 = 1 } }} } },
        } } }},
    });
    try testing.expectEqual(@as(i32, 3), try y_b(&em, 0));
    try testing.expectEqual(@as(i32, 7), try y_b(&em, 1));
}

test "$ with no predicate on the array is refused, and writes nothing" {
    // The query matched the document by `_id`, so nothing recorded which
    // element `$` meant. mongod refuses this, and the document is what makes
    // that the only safe answer: any element would be a guess.
    var doc = try array_doc(testing.allocator);
    defer doc.arena.deinit();
    try testing.expectError(error.NoPositionalMatch, apply_set(&doc, "y.$.b", .{ .int32 = 7 }, .{
        .query = &.{.{ .key = "_id", .value = .{ .int32 = 1 } }},
    }));
    try expect_y_untouched(&doc);

    // And with no query at all -- the upsert path, where the document being
    // built never matched anything.
    try testing.expectError(
        error.NoPositionalMatch,
        apply_set(&doc, "y.$.b", .{ .int32 = 7 }, .{}),
    );
    try expect_y_untouched(&doc);
}

test "a positional segment is refused in first position, and $ twice in one path" {
    var doc = try array_doc(testing.allocator);
    defer doc.arena.deinit();
    var diag: Diagnostic = .{};
    for ([_][]const u8{ "$", "$[]", "$[i]" }) |path| {
        try testing.expectError(
            error.PositionalFirst,
            apply_set(&doc, path, .{ .int32 = 2 }, .{ .diag = &diag }),
        );
        try testing.expectEqualStrings(path, diag.segment);
    }
    // Two `$` cannot both be resolved: the query records one matched element
    // per document, not one per level.
    try testing.expectError(
        error.TooManyPositional,
        apply_set(&doc, "y.$.c.$.d", .{ .int32 = 0 }, .{ .diag = &diag }),
    );
    try testing.expectEqualStrings("y.$.c.$.d", diag.path);
    try expect_y_untouched(&doc);
}

test "$rename refuses a positional path on either end" {
    // A rename moves a field; a positional path names an element. mongod
    // refuses both ends with its own message for each, and the destination is
    // the *value* of the pair, which is the end easiest to forget.
    var doc = try array_doc(testing.allocator);
    defer doc.arena.deinit();
    var diag: Diagnostic = .{};
    try testing.expectError(error.RenameDynamicSource, apply(&doc, &doc_of(&.{
        .{ .key = "$rename", .value = .{ .doc = &.{
            .{ .key = "y.$[].b", .value = .{ .string = "z" } },
        } } },
    }), .{ .diag = &diag }));
    try testing.expectEqualStrings("y.$[].b", diag.path);

    try testing.expectError(error.RenameDynamicDestination, apply(&doc, &doc_of(&.{
        .{ .key = "$rename", .value = .{ .doc = &.{
            .{ .key = "y", .value = .{ .string = "z.$[i]" } },
        } } },
    }), .{ .diag = &diag }));
    try testing.expectEqualStrings("z.$[i]", diag.path);
    try expect_y_untouched(&doc);
}

test "nothing in the update is applied when one of its paths is refused" {
    // The static half of the refusal is taken before the first write, so the
    // good path in this update does not land either. Mutation check: move the
    // `validate` call inside the operator loop and `ok` appears.
    var doc = try array_doc(testing.allocator);
    defer doc.arena.deinit();
    try testing.expectError(error.NoArrayFilter, apply(&doc, &doc_of(&.{
        .{ .key = "$set", .value = .{ .doc = &.{
            .{ .key = "ok", .value = .{ .int32 = 1 } },
            .{ .key = "y.$[i].b", .value = .{ .int32 = 2 } },
        } } },
    }), .{}));
    try testing.expect(doc.get("ok") == null);
    try expect_y_untouched(&doc);
}

test "a non-numeric segment under an array is PathNotViable, not a new field" {
    // The rest of the class the positional forms belonged to. `y.nope.b` is
    // not a positional operator and is refused for a different reason with a
    // different code -- measured on mongod 8.3.7, which answers 28 here and 2
    // for the positional forms.
    const cases = [_][]const u8{ "y.nope.b", "y.nope", "y.$x.b" };
    for (cases) |path| {
        var doc = try array_doc(testing.allocator);
        defer doc.arena.deinit();
        var diag: Diagnostic = .{};
        try testing.expectError(error.PathNotViable, apply(&doc, &doc_of(&.{
            .{ .key = "$set", .value = .{ .doc = &.{.{ .key = path, .value = .{ .int32 = 2 } }} } },
        }), .{ .diag = &diag }));
        try expect_y_untouched(&doc);
    }
}

test "a numeric segment still addresses an array element" {
    // The regression guard for the refusal above: indexed paths are the one
    // way into an array that does work, and they must keep working, including
    // the null padding past the end.
    var doc = try array_doc(testing.allocator);
    defer doc.arena.deinit();
    try apply(&doc, &doc_of(&.{
        .{ .key = "$set", .value = .{ .doc = &.{
            .{ .key = "y.0.b", .value = .{ .int32 = 7 } },
            .{ .key = "y.3.b", .value = .{ .int32 = 8 } },
        } } },
    }), .{});
    const y = doc.get("y").?;
    try testing.expectEqual(@as(usize, 4), y.array.len);
    try testing.expectEqual(@as(i32, 7), y.array[0].doc[0].value.int32);
    try testing.expect(y.array[2] == .null);
    try testing.expectEqual(@as(i32, 8), y.array[3].doc[0].value.int32);
}

test "a replacement is not a path, so it is not refused" {
    // The one case out of seventeen where this server already agreed with
    // mongod: `arrayFilters` alongside a replacement is ignored by both, and
    // a replacement field named like a path is data, not a path.
    var doc = try array_doc(testing.allocator);
    defer doc.arena.deinit();
    try apply(&doc, &doc_of(&.{
        .{ .key = "z", .value = .{ .int32 = 1 } },
    }), .{});
    try testing.expect(doc.get("y") == null);
    try testing.expectEqual(@as(i32, 1), doc.get("z").?.int32);
}