//! 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,
    /// `$inc` or `$mul` against a stored field that is not a number, and by an
    /// operand that is not one. Both are mongod's `TypeMismatch` (14) rather
    /// than the `BadValue` most bad updates answer, and each has its own
    /// sentence, so they are two errors.
    NotNumericField,
    NotNumericOperand,
    /// `$addToSet` or `$pullAll` against a field that is not an array.
    /// `BadValue` (2) on mongod, where `$pop` against one is `TypeMismatch`.
    NotAnArrayField,
    /// `$pop` against a field that is not an array. Same mistake as
    /// `NotAnArrayField`, different code -- measured, not derived.
    NotAnArrayPathElement,
    /// `$pop` by anything that is not 1 or -1. `FailedToParse` (9).
    BadPopArgument,
    /// `$pullAll` by something that is not an array of values.
    PullAllNeedsArray,
    /// `$each` that is not an array.
    BadEach,
    /// `$currentDate` handed a document that is not `{$type: date|timestamp}`.
    BadCurrentDateType,
    /// `$currentDate` handed something that is neither a bool nor a document.
    BadCurrentDateOperand,
    /// A `$push` modifier that is unknown, or whose operand is not what it
    /// takes: `$slice`/`$position` want a number, `$sort` a number or a
    /// one-field document.
    BadPushModifier,
    /// 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,
    /// Whether this application is building a document an upsert will insert.
    /// `$setOnInsert` is the one operator that asks.
    inserting: bool = false,
    /// The clock `$currentDate` reads, in Unix milliseconds. There is no
    /// fallback to a global one: this file has no `io` to read the real clock
    /// through, and a caller that forgets gets the epoch rather than a value
    /// that changes between runs. The command handlers pass the same clock
    /// `ttl_sweep` and the cursor sweep already use.
    now_ms: i64 = 0,
    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 {
    // Every operator's argument is a document of path/operand pairs, so the
    // shape is checked once here rather than at the top of each.
    const ops = doc_pairs(value) orelse return error.InvalidUpdate;
    if (std.mem.eql(u8, op, "$set")) return op_set(arena, pairs, ops, opts);
    // `$setOnInsert` is `$set` on the branch that inserts and nothing at all
    // on the branch that updates -- including its `_id` rule, which is why it
    // does not go through `op_set`: writing `_id` is allowed on a document
    // being built and refused on one being rewritten. Measured.
    if (std.mem.eql(u8, op, "$setOnInsert")) {
        if (!opts.inserting) return;
        return op_set_on_insert(arena, pairs, ops, opts);
    }
    if (std.mem.eql(u8, op, "$currentDate")) return op_current_date(arena, pairs, ops, opts);
    if (std.mem.eql(u8, op, "$unset")) return op_unset(arena, pairs, ops, opts);
    if (std.mem.eql(u8, op, "$inc")) return op_arith(arena, pairs, ops, opts, .add);
    if (std.mem.eql(u8, op, "$mul")) return op_arith(arena, pairs, ops, opts, .mul);
    if (std.mem.eql(u8, op, "$min")) return op_extremum(arena, pairs, ops, opts, .lt);
    if (std.mem.eql(u8, op, "$max")) return op_extremum(arena, pairs, ops, opts, .gt);
    if (std.mem.eql(u8, op, "$push")) return op_push(arena, pairs, ops, opts);
    if (std.mem.eql(u8, op, "$pull")) return op_pull(arena, pairs, ops, opts);
    if (std.mem.eql(u8, op, "$addToSet")) return op_add_to_set(arena, pairs, ops, opts);
    if (std.mem.eql(u8, op, "$pop")) return op_pop(arena, pairs, ops, opts);
    if (std.mem.eql(u8, op, "$pullAll")) return op_pull_all(arena, pairs, ops, opts);
    if (std.mem.eql(u8, op, "$rename")) return op_rename(arena, pairs, ops, opts);
    return error.InvalidUpdate;
}

fn op_set(
    arena: std.mem.Allocator,
    pairs: *std.ArrayListUnmanaged(bson.Pair),
    ops: []const bson.Pair,
    opts: Options,
) UpdateError!void {
    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, opts.diag);
        }
    }
}

fn op_set_on_insert(
    arena: std.mem.Allocator,
    pairs: *std.ArrayListUnmanaged(bson.Pair),
    ops: []const bson.Pair,
    opts: Options,
) UpdateError!void {
    for (ops) |p| {
        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, opts.diag);
        }
    }
}

/// `$currentDate`: write the clock, as a date or as a BSON timestamp.
///
/// The operand says which. A bool -- **either** bool, measured: `false` writes
/// a date too, the value is ignored -- means a date; `{$type: "date"}` and
/// `{$type: "timestamp"}` say so; anything else is refused.
fn op_current_date(
    arena: std.mem.Allocator,
    pairs: *std.ArrayListUnmanaged(bson.Pair),
    ops: []const bson.Pair,
    opts: Options,
) UpdateError!void {
    const now = opts.now_ms;
    for (ops) |p| {
        const value: bson.Value = switch (try current_date_kind(p.value, opts, p.key)) {
            .date => .{ .datetime = now },
            // A BSON timestamp is seconds in the high 32 bits and an ordinal
            // in the low ones. mongod fills the ordinal from the oplog; a
            // standalone has no oplog, so it is 1 -- the same answer twice in
            // one second is the same timestamp, which nothing here reads.
            .timestamp => .{ .timestamp = (@as(u64, @intCast(@divFloor(now, 1000))) << 32) | 1 },
        };
        for (try resolve(arena, pairs.items, p.key, opts)) |segs| {
            try set_path(arena, pairs, segs, value, p.key, opts.diag);
        }
    }
}

fn current_date_kind(
    v: bson.Value,
    opts: Options,
    path: []const u8,
) UpdateError!enum { date, timestamp } {
    switch (v) {
        .bool => return .date,
        .doc => |spec| {
            const t = bson.get_pair(spec, "$type") orelse {
                note(opts.diag, path, "");
                return error.BadCurrentDateType;
            };
            if (t == .string) {
                if (std.mem.eql(u8, t.string, "date")) return .date;
                if (std.mem.eql(u8, t.string, "timestamp")) return .timestamp;
            }
            note(opts.diag, path, "");
            return error.BadCurrentDateType;
        },
        else => {
            if (opts.diag) |d| d.* = .{ .path = path, .other = v.type_name() };
            return error.BadCurrentDateOperand;
        },
    }
}

fn op_unset(
    arena: std.mem.Allocator,
    pairs: *std.ArrayListUnmanaged(bson.Pair),
    ops: []const bson.Pair,
    opts: Options,
) UpdateError!void {
    for (ops) |p| {
        for (try resolve(arena, pairs.items, p.key, opts)) |segs| {
            unset_path(arena, pairs, segs);
        }
    }
}

/// `$inc` and `$mul`, which differ only in the operation and in what an absent
/// field starts from: `$inc` from 0 because adding leaves the operand, `$mul`
/// from 0 because multiplying does too -- so `{$mul: {gone: 5}}` writes 0, not
/// 5. Measured; the natural guess is the other one.
fn op_arith(
    arena: std.mem.Allocator,
    pairs: *std.ArrayListUnmanaged(bson.Pair),
    ops: []const bson.Pair,
    opts: Options,
    comptime kind: enum { add, mul },
) UpdateError!void {
    for (ops) |p| {
        // The operand is checked before the paths are resolved, so an operand
        // that is not a number is one answer for the whole update rather than
        // one per element a positional segment reached.
        const op_name = if (kind == .add) "$inc" else "$mul";
        if (!p.value.is_number()) {
            note(opts.diag, p.key, op_name);
            return error.NotNumericOperand;
        }
        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()) {
                if (opts.diag) |d| d.* = .{ .path = p.key, .segment = op_name, .other = current.type_name() };
                return error.NotNumericField;
            }
            const result = switch (kind) {
                .add => try numeric_add(current, p.value),
                .mul => try numeric_mul(current, p.value),
            };
            try set_path(arena, pairs, segs, result, p.key, opts.diag);
        }
    }
}

/// `$min` and `$max`, which are not numeric operators at all: they compare in
/// BSON canonical order, so `{$min: {s: 5}}` on `s: "b"` writes 5 because a
/// number ranks below a string. An absent field is always written -- there is
/// nothing to be smaller or larger than.
fn op_extremum(
    arena: std.mem.Allocator,
    pairs: *std.ArrayListUnmanaged(bson.Pair),
    ops: []const bson.Pair,
    opts: Options,
    comptime want: std.math.Order,
) UpdateError!void {
    for (ops) |p| {
        for (try resolve(arena, pairs.items, p.key, opts)) |segs| {
            if (get_value(pairs.items, segs)) |current| {
                if (bson.compare(p.value, current) != want) continue;
            }
            try set_path(arena, pairs, segs, try bson.copy_value(arena, p.value), p.key, opts.diag);
        }
    }
}

/// `$push`'s modifiers, in the order mongod applies them.
///
/// Absent `$each` there are no modifiers at all: `{$push: {t: {$slice: 1}}}`
/// pushes the document `{$slice: 1}` as a value. Measured, and it is what
/// makes `$each` the flag rather than a member of the set.
const PushModifiers = struct {
    each: []const bson.Value,
    /// Where the new elements go. Negative counts back from the end.
    position: ?i64 = null,
    /// Ascending/descending over whole elements, or a document naming a field
    /// of them.
    sort: ?bson.Value = null,
    /// Keep the first n, or with a negative n the last -n. Applied last, and
    /// on its own it truncates without adding anything.
    slice: ?i64 = null,
};

fn parse_push_modifiers(spec: []const bson.Pair, opts: Options, path: []const u8) UpdateError!PushModifiers {
    var m = PushModifiers{ .each = &.{} };
    for (spec) |p| {
        if (std.mem.eql(u8, p.key, "$each")) {
            m.each = switch (p.value) {
                .array => |a| a,
                else => {
                    note(opts.diag, path, "$push");
                    return error.BadEach;
                },
            };
            continue;
        }
        const numeric: ?i64 = if (p.value.is_number()) @intFromFloat(@trunc(p.value.as_f128())) else null;
        if (std.mem.eql(u8, p.key, "$position")) {
            m.position = numeric orelse return bad_modifier(opts, path, "$position");
            continue;
        }
        if (std.mem.eql(u8, p.key, "$slice")) {
            m.slice = numeric orelse return bad_modifier(opts, path, "$slice");
            continue;
        }
        if (std.mem.eql(u8, p.key, "$sort")) {
            m.sort = p.value;
            continue;
        }
        // An unknown `$`-prefixed key beside `$each` is a typo, not a field of
        // a document being pushed: the document is `$each`'s elements, not
        // this one.
        return bad_modifier(opts, path, p.key);
    }
    return m;
}

fn bad_modifier(opts: Options, path: []const u8, name: []const u8) UpdateError {
    note(opts.diag, path, name);
    return error.BadPushModifier;
}

fn op_push(
    arena: std.mem.Allocator,
    pairs: *std.ArrayListUnmanaged(bson.Pair),
    ops: []const bson.Pair,
    opts: Options,
) UpdateError!void {
    for (ops) |p| {
        const mods: ?PushModifiers = if (each_of(p.value) != null)
            try parse_push_modifiers(p.value.doc, opts, p.key)
        else
            null;
        for (try resolve(arena, pairs.items, p.key, opts)) |segs| {
            var items: std.ArrayListUnmanaged(bson.Value) = .empty;
            defer items.deinit(arena);
            if (get_value(pairs.items, segs)) |current| switch (current) {
                .array => |arr| try items.appendSlice(arena, arr),
                .null => {},
                else => return error.InvalidUpdate, // non-array field
            };
            if (mods) |m| {
                try insert_each(arena, &items, m);
                if (m.sort) |key| try sort_elements(items.items, key);
                if (m.slice) |n| {
                    const kept = slice_range(items.items.len, n);
                    std.mem.copyForwards(bson.Value, items.items[0..kept.len], kept.of(items.items));
                    items.shrinkRetainingCapacity(kept.len);
                }
            } else {
                try items.append(arena, try bson.copy_value(arena, p.value));
            }
            try set_path(arena, pairs, segs, .{ .array = try items.toOwnedSlice(arena) }, p.key, opts.diag);
        }
    }
}

fn insert_each(
    arena: std.mem.Allocator,
    items: *std.ArrayListUnmanaged(bson.Value),
    m: PushModifiers,
) UpdateError!void {
    const at: usize = if (m.position) |pos| blk: {
        if (pos >= 0) break :blk @min(@as(usize, @intCast(pos)), items.items.len);
        // Counted back from the end, and clamped at the front rather than
        // wrapping: `$position: -99` on a two-element array is 0.
        const back: usize = @intCast(-pos);
        break :blk items.items.len -| back;
    } else items.items.len;
    for (m.each, 0..) |item, i| {
        try items.insert(arena, at + i, try bson.copy_value(arena, item));
    }
}

/// Which elements a `$slice` keeps. A non-negative `n` keeps the first `n`; a
/// negative one keeps the **last** `-n`, which is the shape a capped log uses
/// and the half that is easy to get backwards.
fn slice_range(len: usize, n: i64) struct {
    start: usize,
    len: usize,

    fn of(self: @This(), items: []bson.Value) []bson.Value {
        return items[self.start .. self.start + self.len];
    }
} {
    if (n >= 0) return .{ .start = 0, .len = @min(len, @as(usize, @intCast(n))) };
    const keep = @min(len, @as(usize, @intCast(-n)));
    return .{ .start = len - keep, .len = keep };
}

/// `$sort: 1` orders whole elements; `$sort: {a: 1}` orders on a field of
/// them, which is only meaningful when they are documents -- an element that
/// is not one, or that lacks the field, sorts as null, the same rank a missing
/// field has everywhere else here.
const ElementSort = struct {
    path: ?[]const u8,
    descending: bool,

    fn key_of(self: ElementSort, v: bson.Value) bson.Value {
        const path = self.path orelse return v;
        const sub = switch (v) {
            .doc => |d| d,
            else => return .null,
        };
        var segs: [max_path_segments][]const u8 = undefined;
        const n = split_path(path, &segs) orelse return .null;
        return get_value(sub, segs[0..n]) orelse .null;
    }

    fn less(self: ElementSort, a: bson.Value, b: bson.Value) bool {
        const order = bson.compare(self.key_of(a), self.key_of(b));
        return if (self.descending) order == .gt else order == .lt;
    }
};

fn sort_elements(items: []bson.Value, key: bson.Value) UpdateError!void {
    const spec: ElementSort = switch (key) {
        .doc => |pairs| blk: {
            if (pairs.len != 1) return error.BadPushModifier;
            break :blk .{ .path = pairs[0].key, .descending = is_descending(pairs[0].value) };
        },
        else => blk: {
            if (!key.is_number()) return error.BadPushModifier;
            break :blk .{ .path = null, .descending = is_descending(key) };
        },
    };
    std.mem.sort(bson.Value, items, spec, ElementSort.less);
}

fn is_descending(v: bson.Value) bool {
    return v.is_number() and v.as_f128() < 0;
}

fn op_pull(
    arena: std.mem.Allocator,
    pairs: *std.ArrayListUnmanaged(bson.Pair),
    ops: []const bson.Pair,
    opts: Options,
) UpdateError!void {
    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, opts.diag);
        }
    }
}

/// `$addToSet`: append what the array does not already hold.
///
/// "Already hold" is `bson.compare` equality, which is exactly mongod's: an
/// int32 `2` and a double `2.0` are one value, and two documents with the same
/// fields in a different order are two -- because `compare_docs` walks the
/// pairs positionally and tie-breaks on the key.
fn op_add_to_set(
    arena: std.mem.Allocator,
    pairs: *std.ArrayListUnmanaged(bson.Pair),
    ops: []const bson.Pair,
    opts: Options,
) UpdateError!void {
    for (ops) |p| {
        // `$each` is a modifier only here and under `$push`; anywhere else a
        // document operand is the value being added.
        const candidates: []const bson.Value = if (each_of(p.value)) |each| switch (each) {
            .array => |a| a,
            else => {
                note(opts.diag, p.key, "$addToSet");
                return error.BadEach;
            },
        } else &.{p.value};

        for (try resolve(arena, pairs.items, p.key, opts)) |segs| {
            var items: std.ArrayListUnmanaged(bson.Value) = .empty;
            defer items.deinit(arena);
            if (get_value(pairs.items, segs)) |current| switch (current) {
                .array => |arr| try items.appendSlice(arena, arr),
                .null => {},
                else => {
                    note(opts.diag, p.key, "$addToSet");
                    return error.NotAnArrayField;
                },
            };
            for (candidates) |c| {
                if (holds(items.items, c)) continue;
                try items.append(arena, try bson.copy_value(arena, c));
            }
            try set_path(arena, pairs, segs, .{ .array = try items.toOwnedSlice(arena) }, p.key, opts.diag);
        }
    }
}

fn holds(items: []const bson.Value, v: bson.Value) bool {
    for (items) |item| if (bson.compare(item, v) == .eq) return true;
    return false;
}

/// The `$each` of a modifier document, or null when the operand is a value.
///
/// Presence of `$each` is what makes an operand a modifier document at all --
/// measured: `{$push: {t: {$slice: 1}}}` pushes `{$slice: 1}` as a value.
fn each_of(v: bson.Value) ?bson.Value {
    const doc = doc_pairs(v) orelse return null;
    return bson.get_pair(doc, "$each");
}

/// `$pop`: remove one element from an end. `1` is the last, `-1` the first.
///
/// An empty array and an absent field are both no-ops rather than errors, so
/// the only refusals are the argument and the field's type.
fn op_pop(
    arena: std.mem.Allocator,
    pairs: *std.ArrayListUnmanaged(bson.Pair),
    ops: []const bson.Pair,
    opts: Options,
) UpdateError!void {
    for (ops) |p| {
        if (!p.value.is_number()) {
            note(opts.diag, p.key, "");
            return error.BadPopArgument;
        }
        const n = p.value.as_f128();
        if (n != 1 and n != -1) {
            note(opts.diag, p.key, "");
            return error.BadPopArgument;
        }
        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 => {
                    note(opts.diag, p.key, "$pop");
                    return error.NotAnArrayPathElement;
                },
            };
            if (arr.len == 0) continue;
            const kept = if (n == 1) arr[0 .. arr.len - 1] else arr[1..];
            try set_path(arena, pairs, segs, .{ .array = try arena.dupe(bson.Value, kept) }, p.key, opts.diag);
        }
    }
}

/// `$pullAll`: remove every element equal to any of the listed values.
///
/// The difference from `$pull` is the whole of it: `$pull` takes a *predicate*
/// and `$pullAll` takes values, compared whole. `{$pull: {t: {a: 1}}}` matches
/// elements having `a: 1`; `{$pullAll: {t: [{a: 1}]}}` matches elements that
/// *are* `{a: 1}`.
fn op_pull_all(
    arena: std.mem.Allocator,
    pairs: *std.ArrayListUnmanaged(bson.Pair),
    ops: []const bson.Pair,
    opts: Options,
) UpdateError!void {
    for (ops) |p| {
        const wanted = switch (p.value) {
            .array => |a| a,
            else => {
                if (opts.diag) |d| d.* = .{ .path = p.key, .other = p.value.type_name() };
                return error.PullAllNeedsArray;
            },
        };
        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 => {
                    note(opts.diag, p.key, "$pullAll");
                    return error.NotAnArrayField;
                },
            };
            var items: std.ArrayListUnmanaged(bson.Value) = .empty;
            defer items.deinit(arena);
            for (arr) |elem| {
                if (holds(wanted, elem)) continue;
                try items.append(arena, elem);
            }
            try set_path(arena, pairs, segs, .{ .array = try items.toOwnedSlice(arena) }, p.key, opts.diag);
        }
    }
}

/// `$rename` alone keeps the plain split: `validate` has already refused a
/// positional path on either end of it, which is what mongod does too.
fn op_rename(
    arena: std.mem.Allocator,
    pairs: *std.ArrayListUnmanaged(bson.Pair),
    ops: []const bson.Pair,
    opts: Options,
) UpdateError!void {
    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, opts.diag);
    }
}

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 };
}

/// The same widening ladder as `numeric_add`: a double anywhere makes the
/// answer a double, two int32s stay int32 unless the product does not fit,
/// and an int64 overflowing is a refusal rather than a wrap.
fn numeric_mul(a: bson.Value, b: bson.Value) UpdateError!bson.Value {
    if (a == .double or b == .double) {
        const product: f64 = @floatCast(a.as_f128() * b.as_f128());
        return .{ .double = product };
    }
    if (a == .int64 or b == .int64) {
        const av: i64 = as_int64(a);
        const bv: i64 = as_int64(b);
        const product = std.math.mul(i64, av, bv) catch return error.InvalidUpdate;
        return .{ .int64 = product };
    }
    const product: i64 = @as(i64, a.int32) * b.int32;
    if (product >= std.math.minInt(i32) and product <= std.math.maxInt(i32)) {
        return .{ .int32 = @intCast(product) };
    }
    return .{ .int64 = product };
}

fn as_int64(v: bson.Value) i64 {
    return switch (v) {
        .int32 => |i| i,
        .int64 => |i| i,
        else => unreachable,
    };
}

// ---------------------------------------------------------------------------
// 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);
}

test "$mul multiplies, and starts a missing field from zero" {
    // The measured surprise: `{$mul: {gone: 5}}` writes 0, not 5. Mutation
    // check: start `op_arith` from `.{ .int32 = 1 }` for `.mul` and the second
    // half goes red -- which is the reading anyone would reach for.
    var doc = try doc_with(testing.allocator, &.{
        .{ .key = "a", .value = .{ .int32 = 5 } },
        .{ .key = "d", .value = .{ .double = 2.5 } },
    });
    defer doc.arena.deinit();
    try apply(&doc, &doc_of(&.{
        .{ .key = "$mul", .value = .{ .doc = &.{
            .{ .key = "a", .value = .{ .int32 = 2 } },
            .{ .key = "d", .value = .{ .int32 = 2 } },
            .{ .key = "gone", .value = .{ .int32 = 5 } },
        } } },
    }), .{});
    try testing.expectEqual(@as(i32, 10), doc.get("a").?.int32);
    try testing.expectEqual(@as(f64, 5.0), doc.get("d").?.double);
    try testing.expectEqual(@as(i32, 0), doc.get("gone").?.int32);
}

test "$mul widens an int32 product that does not fit" {
    var doc = try doc_with(testing.allocator, &.{
        .{ .key = "a", .value = .{ .int32 = 2000000000 } },
    });
    defer doc.arena.deinit();
    try apply(&doc, &doc_of(&.{
        .{ .key = "$mul", .value = .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 2 } }} } },
    }), .{});
    try testing.expectEqual(@as(i64, 4000000000), doc.get("a").?.int64);
}

test "$inc and $mul refuse a non-number, on either side" {
    // TypeMismatch, not the BadValue the rest of a bad update answers, and the
    // field and the operand are different sentences on mongod -- so they are
    // different errors here.
    var doc = try doc_with(testing.allocator, &.{
        .{ .key = "a", .value = .{ .int32 = 5 } },
        .{ .key = "s", .value = .{ .string = "b" } },
    });
    defer doc.arena.deinit();
    var diag: Diagnostic = .{};
    for ([_][]const u8{ "$inc", "$mul" }) |op| {
        try testing.expectError(error.NotNumericField, apply(&doc, &doc_of(&.{
            .{ .key = op, .value = .{ .doc = &.{.{ .key = "s", .value = .{ .int32 = 2 } }} } },
        }), .{ .diag = &diag }));
        try testing.expectEqualStrings("string", diag.other);
        try testing.expectError(error.NotNumericOperand, apply(&doc, &doc_of(&.{
            .{ .key = op, .value = .{ .doc = &.{.{ .key = "a", .value = .{ .string = "x" } }} } },
        }), .{ .diag = &diag }));
        try testing.expectEqualStrings(op, diag.segment);
    }
    // Neither refusal wrote anything.
    try testing.expectEqual(@as(i32, 5), doc.get("a").?.int32);
    try testing.expectEqualStrings("b", doc.get("s").?.string);
}

test "$min and $max compare in BSON order, not numerically" {
    // The load-bearing case: `s` holds a string and the operand is a number,
    // and there is still a defined answer because a number ranks below a
    // string. Mutation check: make `op_extremum` require both to be numbers
    // and the two `s` rows go red.
    var doc = try doc_with(testing.allocator, &.{
        .{ .key = "a", .value = .{ .int32 = 5 } },
        .{ .key = "s", .value = .{ .string = "b" } },
        .{ .key = "n", .value = .null },
    });
    defer doc.arena.deinit();
    try apply(&doc, &doc_of(&.{
        .{ .key = "$min", .value = .{ .doc = &.{
            .{ .key = "a", .value = .{ .int32 = 7 } }, // higher: not written
            .{ .key = "s", .value = .{ .int32 = 5 } }, // a number is below a string
            .{ .key = "gone", .value = .{ .int32 = 7 } }, // absent: always written
        } } },
    }), .{});
    try testing.expectEqual(@as(i32, 5), doc.get("a").?.int32);
    try testing.expectEqual(@as(i32, 5), doc.get("s").?.int32);
    try testing.expectEqual(@as(i32, 7), doc.get("gone").?.int32);

    try apply(&doc, &doc_of(&.{
        .{ .key = "$max", .value = .{ .doc = &.{
            .{ .key = "a", .value = .{ .int32 = 3 } }, // lower: not written
            .{ .key = "n", .value = .{ .int32 = 1 } }, // a number is above null
        } } },
    }), .{});
    try testing.expectEqual(@as(i32, 5), doc.get("a").?.int32);
    try testing.expectEqual(@as(i32, 1), doc.get("n").?.int32);
}

test "$min treats an int and an equal double as the same value" {
    // Equal is not less, so nothing is written and the stored type survives.
    var doc = try doc_with(testing.allocator, &.{.{ .key = "a", .value = .{ .int32 = 5 } }});
    defer doc.arena.deinit();
    try apply(&doc, &doc_of(&.{
        .{ .key = "$min", .value = .{ .doc = &.{.{ .key = "a", .value = .{ .double = 5.0 } }} } },
    }), .{});
    try testing.expect(doc.get("a").? == .int32);
}

test "$mul reaches every element a positional segment names" {
    var doc = try doc_with(testing.allocator, &.{
        .{ .key = "t", .value = .{ .array = &.{
            .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 2 } }} },
            .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 3 } }} },
        } } },
    });
    defer doc.arena.deinit();
    try apply(&doc, &doc_of(&.{
        .{ .key = "$mul", .value = .{ .doc = &.{.{ .key = "t.$[].a", .value = .{ .int32 = 10 } }} } },
    }), .{});
    const t = doc.get("t").?.array;
    try testing.expectEqual(@as(i32, 20), t[0].doc[0].value.int32);
    try testing.expectEqual(@as(i32, 30), t[1].doc[0].value.int32);
}

test "$addToSet appends only what the array does not hold" {
    var doc = try doc_with(testing.allocator, &.{
        .{ .key = "t", .value = .{ .array = &.{ .{ .int32 = 1 }, .{ .int32 = 2 } } } },
    });
    defer doc.arena.deinit();
    try apply(&doc, &doc_of(&.{
        .{ .key = "$addToSet", .value = .{ .doc = &.{
            .{ .key = "t", .value = .{ .doc = &.{
                .{ .key = "$each", .value = .{ .array = &.{ .{ .int32 = 2 }, .{ .int32 = 3 }, .{ .int32 = 3 } } } },
            } } },
            .{ .key = "gone", .value = .{ .int32 = 1 } },
        } } },
    }), .{});
    // 2 is already there, and the two 3s in one `$each` are one value: the
    // candidates are checked against the array *as it grows*.
    const t = doc.get("t").?.array;
    try testing.expectEqual(@as(usize, 3), t.len);
    try testing.expectEqual(@as(i32, 3), t[2].int32);
    // An absent field becomes a one-element array rather than an error.
    try testing.expectEqual(@as(usize, 1), doc.get("gone").?.array.len);
}

test "$addToSet identity is BSON equality, field order included" {
    // Two rows, opposite answers, one comparator. Mutation check: compare with
    // anything that ignores key order and the second half goes red -- mongod
    // stores both spellings of the same document.
    var doc = try doc_with(testing.allocator, &.{
        .{ .key = "n", .value = .{ .array = &.{.{ .int32 = 2 }} } },
        .{ .key = "d", .value = .{ .array = &.{.{ .doc = &.{
            .{ .key = "a", .value = .{ .int32 = 1 } },
            .{ .key = "b", .value = .{ .int32 = 2 } },
        } }} } },
    });
    defer doc.arena.deinit();
    try apply(&doc, &doc_of(&.{
        .{ .key = "$addToSet", .value = .{ .doc = &.{
            // An int32 2 and a double 2.0 are one value.
            .{ .key = "n", .value = .{ .double = 2.0 } },
            // The same fields in the other order are two.
            .{ .key = "d", .value = .{ .doc = &.{
                .{ .key = "b", .value = .{ .int32 = 2 } },
                .{ .key = "a", .value = .{ .int32 = 1 } },
            } } },
        } } },
    }), .{});
    try testing.expectEqual(@as(usize, 1), doc.get("n").?.array.len);
    try testing.expectEqual(@as(usize, 2), doc.get("d").?.array.len);
}

test "$pop takes one element off an end, and is quiet when there is none" {
    var doc = try doc_with(testing.allocator, &.{
        .{ .key = "a", .value = .{ .array = &.{ .{ .int32 = 1 }, .{ .int32 = 2 }, .{ .int32 = 3 } } } },
        .{ .key = "b", .value = .{ .array = &.{ .{ .int32 = 1 }, .{ .int32 = 2 } } } },
        .{ .key = "e", .value = .{ .array = &.{} } },
    });
    defer doc.arena.deinit();
    try apply(&doc, &doc_of(&.{
        .{ .key = "$pop", .value = .{ .doc = &.{
            .{ .key = "a", .value = .{ .int32 = 1 } }, // the last
            .{ .key = "b", .value = .{ .double = -1.0 } }, // the first, and -1.0 is -1
            .{ .key = "e", .value = .{ .int32 = 1 } }, // empty: no-op
            .{ .key = "gone", .value = .{ .int32 = 1 } }, // absent: no-op
        } } },
    }), .{});
    try testing.expectEqual(@as(usize, 2), doc.get("a").?.array.len);
    try testing.expectEqual(@as(i32, 2), doc.get("a").?.array[1].int32);
    try testing.expectEqual(@as(usize, 1), doc.get("b").?.array.len);
    try testing.expectEqual(@as(i32, 2), doc.get("b").?.array[0].int32);
    try testing.expectEqual(@as(usize, 0), doc.get("e").?.array.len);
    try testing.expect(doc.get("gone") == null);
}

test "$pop refuses an argument that is not one of its two values" {
    var doc = try doc_with(testing.allocator, &.{
        .{ .key = "t", .value = .{ .array = &.{.{ .int32 = 1 }} } },
    });
    defer doc.arena.deinit();
    for ([_]bson.Value{ .{ .int32 = 2 }, .{ .int32 = 0 }, .{ .string = "x" } }) |bad| {
        try testing.expectError(error.BadPopArgument, apply(&doc, &doc_of(&.{
            .{ .key = "$pop", .value = .{ .doc = &.{.{ .key = "t", .value = bad }} } },
        }), .{}));
    }
    try testing.expectEqual(@as(usize, 1), doc.get("t").?.array.len);
}

test "$pullAll removes values, where $pull removes matches" {
    // The whole difference between the two, in one document: `{a: 1}` as a
    // `$pull` argument is a predicate and as a `$pullAll` element is a value.
    // Mutation check: route `$pullAll` through `pull_matches` and the second
    // half goes red -- `{a: 1, b: 2}` would be pulled by a predicate too.
    var doc = try doc_with(testing.allocator, &.{
        .{ .key = "n", .value = .{ .array = &.{
            .{ .int32 = 1 }, .{ .int32 = 2 }, .{ .int32 = 3 }, .{ .int32 = 2 },
        } } },
        .{ .key = "d", .value = .{ .array = &.{
            .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 1 } }} },
            .{ .doc = &.{
                .{ .key = "a", .value = .{ .int32 = 1 } },
                .{ .key = "b", .value = .{ .int32 = 2 } },
            } },
        } } },
    });
    defer doc.arena.deinit();
    try apply(&doc, &doc_of(&.{
        .{ .key = "$pullAll", .value = .{ .doc = &.{
            .{ .key = "n", .value = .{ .array = &.{.{ .int32 = 2 }} } },
            .{ .key = "d", .value = .{ .array = &.{
                .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 1 } }} },
            } } },
            .{ .key = "gone", .value = .{ .array = &.{.{ .int32 = 1 }} } },
        } } },
    }), .{});
    try testing.expectEqual(@as(usize, 2), doc.get("n").?.array.len);
    // Only the element that *is* `{a: 1}` went; the one that merely matches
    // that predicate stayed.
    try testing.expectEqual(@as(usize, 1), doc.get("d").?.array.len);
    try testing.expectEqual(@as(usize, 2), doc.get("d").?.array[0].doc.len);
    try testing.expect(doc.get("gone") == null);
}

test "the array operators refuse a field that is not an array" {
    var doc = try doc_with(testing.allocator, &.{.{ .key = "t", .value = .{ .int32 = 5 } }});
    defer doc.arena.deinit();
    try testing.expectError(error.NotAnArrayField, apply(&doc, &doc_of(&.{
        .{ .key = "$addToSet", .value = .{ .doc = &.{.{ .key = "t", .value = .{ .int32 = 1 } }} } },
    }), .{}));
    try testing.expectError(error.NotAnArrayPathElement, apply(&doc, &doc_of(&.{
        .{ .key = "$pop", .value = .{ .doc = &.{.{ .key = "t", .value = .{ .int32 = 1 } }} } },
    }), .{}));
    try testing.expectError(error.NotAnArrayField, apply(&doc, &doc_of(&.{
        .{ .key = "$pullAll", .value = .{ .doc = &.{.{ .key = "t", .value = .{ .array = &.{.{ .int32 = 1 }} } }} } },
    }), .{}));
    try testing.expectError(error.PullAllNeedsArray, apply(&doc, &doc_of(&.{
        .{ .key = "$pullAll", .value = .{ .doc = &.{.{ .key = "t", .value = .{ .int32 = 1 } }} } },
    }), .{}));
    try testing.expectEqual(@as(i32, 5), doc.get("t").?.int32);
}

test "$push applies its modifiers in mongod's order: position, sort, slice" {
    // The order is the whole of it. `$sort` runs over the array *after* the
    // new elements are in it, and `$slice` runs last, over the sorted result.
    var doc = try doc_with(testing.allocator, &.{
        .{ .key = "t", .value = .{ .array = &.{ .{ .int32 = 3 }, .{ .int32 = 1 } } } },
    });
    defer doc.arena.deinit();
    try apply(&doc, &doc_of(&.{
        .{ .key = "$push", .value = .{ .doc = &.{.{ .key = "t", .value = .{ .doc = &.{
            .{ .key = "$each", .value = .{ .array = &.{.{ .int32 = 2 }} } },
            .{ .key = "$sort", .value = .{ .int32 = 1 } },
            .{ .key = "$slice", .value = .{ .int32 = 2 } },
        } } }} } },
    }), .{});
    const t = doc.get("t").?.array;
    try testing.expectEqual(@as(usize, 2), t.len);
    try testing.expectEqual(@as(i32, 1), t[0].int32);
    try testing.expectEqual(@as(i32, 2), t[1].int32);
}

test "$slice keeps the first n, or with a negative n the last" {
    // Mutation check: keep the first `-n` elements for a negative slice and
    // the second half goes red. It is the half a capped log depends on.
    const cases = [_]struct { n: i32, want: [3]i32, len: usize }{
        .{ .n = 3, .want = .{ 1, 2, 3 }, .len = 3 },
        .{ .n = -3, .want = .{ 2, 3, 4 }, .len = 3 },
        .{ .n = 0, .want = .{ 0, 0, 0 }, .len = 0 },
        .{ .n = 10, .want = .{ 1, 2, 3 }, .len = 4 },
    };
    for (cases) |c| {
        var doc = try doc_with(testing.allocator, &.{
            .{ .key = "t", .value = .{ .array = &.{ .{ .int32 = 1 }, .{ .int32 = 2 } } } },
        });
        defer doc.arena.deinit();
        const spec = [2]bson.Pair{
            .{ .key = "$each", .value = .{ .array = &.{ .{ .int32 = 3 }, .{ .int32 = 4 } } } },
            .{ .key = "$slice", .value = .{ .int32 = c.n } },
        };
        const arg = [1]bson.Pair{.{ .key = "t", .value = .{ .doc = &spec } }};
        try apply(&doc, &doc_of(&.{.{ .key = "$push", .value = .{ .doc = &arg } }}), .{});
        const t = doc.get("t").?.array;
        try testing.expectEqual(c.len, t.len);
        for (t, 0..) |v, i| {
            if (i < 3) try testing.expectEqual(c.want[i], v.int32);
        }
    }
}

test "$position inserts where it says, counting back from the end when negative" {
    const cases = [_]struct { pos: i32, want: [3]i32 }{
        .{ .pos = 0, .want = .{ 9, 1, 2 } },
        .{ .pos = 1, .want = .{ 1, 9, 2 } },
        .{ .pos = 99, .want = .{ 1, 2, 9 } },
        .{ .pos = -1, .want = .{ 1, 9, 2 } },
        .{ .pos = -99, .want = .{ 9, 1, 2 } }, // clamped at the front, not wrapped
    };
    for (cases) |c| {
        var doc = try doc_with(testing.allocator, &.{
            .{ .key = "t", .value = .{ .array = &.{ .{ .int32 = 1 }, .{ .int32 = 2 } } } },
        });
        defer doc.arena.deinit();
        const spec = [2]bson.Pair{
            .{ .key = "$each", .value = .{ .array = &.{.{ .int32 = 9 }} } },
            .{ .key = "$position", .value = .{ .int32 = c.pos } },
        };
        const arg = [1]bson.Pair{.{ .key = "t", .value = .{ .doc = &spec } }};
        try apply(&doc, &doc_of(&.{.{ .key = "$push", .value = .{ .doc = &arg } }}), .{});
        const t = doc.get("t").?.array;
        try testing.expectEqual(@as(usize, 3), t.len);
        for (t, 0..) |v, i| try testing.expectEqual(c.want[i], v.int32);
    }
}

test "$sort orders on a field of the elements" {
    var doc = try doc_with(testing.allocator, &.{
        .{ .key = "t", .value = .{ .array = &.{
            .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 3 } }} },
            .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 1 } }} },
        } } },
    });
    defer doc.arena.deinit();
    try apply(&doc, &doc_of(&.{
        .{ .key = "$push", .value = .{ .doc = &.{.{ .key = "t", .value = .{ .doc = &.{
            .{ .key = "$each", .value = .{ .array = &.{
                .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 2 } }} },
            } } },
            .{ .key = "$sort", .value = .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 1 } }} } },
        } } }} } },
    }), .{});
    const t = doc.get("t").?.array;
    try testing.expectEqual(@as(usize, 3), t.len);
    for (t, 1..) |v, want| try testing.expectEqual(@as(i32, @intCast(want)), v.doc[0].value.int32);
}

test "without $each there are no modifiers, only a value" {
    // The measured rule that makes `$each` the flag: `{$slice: 1}` on its own
    // is a document to push, not an instruction. Mutation check: treat any
    // `$`-prefixed key as a modifier and this stores nothing where mongod
    // stores a document.
    var doc = try doc_with(testing.allocator, &.{
        .{ .key = "t", .value = .{ .array = &.{.{ .int32 = 1 }} } },
    });
    defer doc.arena.deinit();
    try apply(&doc, &doc_of(&.{
        .{ .key = "$push", .value = .{ .doc = &.{.{ .key = "t", .value = .{ .doc = &.{
            .{ .key = "$slice", .value = .{ .int32 = 1 } },
        } } }} } },
    }), .{});
    const t = doc.get("t").?.array;
    try testing.expectEqual(@as(usize, 2), t.len);
    try testing.expect(t[1] == .doc);
}

test "a $push modifier that is not one, or is handed the wrong thing, is refused" {
    var doc = try doc_with(testing.allocator, &.{
        .{ .key = "t", .value = .{ .array = &.{.{ .int32 = 1 }} } },
    });
    defer doc.arena.deinit();
    const bad = [_]bson.Pair{
        .{ .key = "$bogus", .value = .{ .int32 = 1 } },
        .{ .key = "$slice", .value = .{ .string = "x" } },
        .{ .key = "$position", .value = .{ .string = "x" } },
    };
    for (bad) |m| {
        const spec = [2]bson.Pair{
            .{ .key = "$each", .value = .{ .array = &.{.{ .int32 = 3 }} } },
            m,
        };
        const arg = [1]bson.Pair{.{ .key = "t", .value = .{ .doc = &spec } }};
        try testing.expectError(error.BadPushModifier, apply(
            &doc,
            &doc_of(&.{.{ .key = "$push", .value = .{ .doc = &arg } }}),
            .{},
        ));
    }
    try testing.expectEqual(@as(usize, 1), doc.get("t").?.array.len);
}

test "$setOnInsert writes only on the branch that inserts" {
    // Mutation check: drop the `opts.inserting` guard and the first half goes
    // red -- an update would gain a field that is only supposed to exist on a
    // document nobody had before.
    var updating = try doc_with(testing.allocator, &.{
        .{ .key = "_id", .value = .{ .int32 = 1 } },
        .{ .key = "a", .value = .{ .int32 = 100 } },
    });
    defer updating.arena.deinit();
    try apply(&updating, &doc_of(&.{
        .{ .key = "$setOnInsert", .value = .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 1 } }} } },
        .{ .key = "$set", .value = .{ .doc = &.{.{ .key = "b", .value = .{ .int32 = 2 } }} } },
    }), .{});
    try testing.expectEqual(@as(i32, 100), updating.get("a").?.int32);
    try testing.expectEqual(@as(i32, 2), updating.get("b").?.int32);

    var inserting = try doc_with(testing.allocator, &.{.{ .key = "k", .value = .{ .int32 = 1 } }});
    defer inserting.arena.deinit();
    try apply(&inserting, &doc_of(&.{
        .{ .key = "$setOnInsert", .value = .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 1 } }} } },
    }), .{ .inserting = true });
    try testing.expectEqual(@as(i32, 1), inserting.get("a").?.int32);
}

test "$setOnInsert may write _id, where $set may not" {
    // The one place the immutability rule does not apply: a document being
    // built has no identity yet to change.
    var doc = try doc_with(testing.allocator, &.{.{ .key = "k", .value = .{ .int32 = 1 } }});
    defer doc.arena.deinit();
    try apply(&doc, &doc_of(&.{
        .{ .key = "$setOnInsert", .value = .{ .doc = &.{.{ .key = "_id", .value = .{ .int32 = 9 } }} } },
    }), .{ .inserting = true });
    try testing.expectEqual(@as(i32, 9), doc.get("_id").?.int32);

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

test "$currentDate writes the clock it was handed" {
    // The clock is a parameter, so the result is a value rather than a moving
    // target -- and the server passes the same one `ttl_sweep` reads.
    var doc = try doc_with(testing.allocator, &.{.{ .key = "a", .value = .{ .int32 = 1 } }});
    defer doc.arena.deinit();
    try apply(&doc, &doc_of(&.{
        .{ .key = "$currentDate", .value = .{ .doc = &.{
            .{ .key = "d", .value = .{ .bool = true } },
            // Measured: `false` writes a date too. The boolean says "a date",
            // not "whether".
            .{ .key = "f", .value = .{ .bool = false } },
            .{ .key = "e", .value = .{ .doc = &.{.{ .key = "$type", .value = .{ .string = "date" } }} } },
            .{ .key = "t", .value = .{ .doc = &.{.{ .key = "$type", .value = .{ .string = "timestamp" } }} } },
        } } },
    }), .{ .now_ms = 1_700_000_000_123 });
    try testing.expectEqual(@as(i64, 1_700_000_000_123), doc.get("d").?.datetime);
    try testing.expectEqual(@as(i64, 1_700_000_000_123), doc.get("f").?.datetime);
    try testing.expectEqual(@as(i64, 1_700_000_000_123), doc.get("e").?.datetime);
    // Seconds in the high 32 bits, an ordinal in the low ones.
    try testing.expectEqual(@as(u64, (1_700_000_000 << 32) | 1), doc.get("t").?.timestamp);
}

test "$currentDate refuses an operand that names no type" {
    var doc = try doc_with(testing.allocator, &.{.{ .key = "a", .value = .{ .int32 = 1 } }});
    defer doc.arena.deinit();
    var diag: Diagnostic = .{};
    try testing.expectError(error.BadCurrentDateType, apply(&doc, &doc_of(&.{
        .{ .key = "$currentDate", .value = .{ .doc = &.{
            .{ .key = "d", .value = .{ .doc = &.{.{ .key = "$type", .value = .{ .string = "nope" } }} } },
        } } },
    }), .{ .diag = &diag }));
    try testing.expectError(error.BadCurrentDateType, apply(&doc, &doc_of(&.{
        .{ .key = "$currentDate", .value = .{ .doc = &.{
            .{ .key = "d", .value = .{ .doc = &.{.{ .key = "a", .value = .{ .int32 = 1 } }} } },
        } } },
    }), .{ .diag = &diag }));
    try testing.expectError(error.BadCurrentDateOperand, apply(&doc, &doc_of(&.{
        .{ .key = "$currentDate", .value = .{ .doc = &.{.{ .key = "d", .value = .{ .int32 = 1 } }} } },
    }), .{ .diag = &diag }));
    try testing.expectEqualStrings("int", diag.other);
    try testing.expect(doc.get("d") == null);
}