<?php /** @noinspection PhpUnused */

namespace apexl\Vault\operators\drivers;

use apexl\Vault\enums\SortDirection;
use apexl\Vault\enums\WhereCondition;
use apexl\Vault\enums\WhereType;
use apexl\Vault\exceptions\MongoUnsupportedException;
use apexl\Vault\interfaces\driver;
use apexl\Vault\interfaces\operators;
use app\vendor\apexl\vault\src\Vault\enums\HavingType;
use Closure;
use MongoDB\BSON\Regex;
use MongoDB\Client;
use MongoDB\Collection;
use MongoDB\Database;
use stdClass;

/**
 * MongoDB driver — sibling to `mysql` and `sqlite` under
 * `apexl\Vault\operators\drivers`. Implements both `operators` (connection
 * lifecycle + result fetching) and `driver` (query-builder API) interfaces
 * directly. Does NOT extend `pdo` — Mongo has no PDO wire protocol.
 *
 * Connection details (per-named-vault config entry):
 *   - `driver`   : 'mongodb'         (matched by Vault::initiateVault)
 *   - `uri`      : 'mongodb://...'   (full connection URI)
 *   - `database` : '<db_name>'       (default DB to operate on)
 *   - `options`  : optional array of MongoDB\Client options
 *
 * Build-plan reference: TC plans/vault-mongo.md.
 *
 * Status by phase:
 *
 *   Z1.c — skeleton with throws. ✓
 *   Z1.d — read path: select / where / orWhere / orderBy / limit / fields
 *          / newQuery / reset / *Group* / execute / fetch* / toCountQuery. ✓
 *   Z2.a-c — write path: insert / update / delete + counters strategy
 *          for int id issue + lastInsertId(). (pending)
 *   Z2.d — escape hatches: pipeline / aggregate / ensureIndex / dropIndex
 *          / bulkWrite. (pending)
 *
 * SQL-only primitives (whereRaw / executeRaw / executeSql / subQuery /
 * join / leftJoin / having / groupBy / describe / createTable) throw
 * MongoUnsupportedException permanently — D1 of the plan locks them as
 * out-of-scope.
 */
class mongodb implements operators, driver
{
    protected stdClass $connectionDetails;
    protected ?Client $client = null;

    /** Default database name from connection config. */
    protected string $database = '';

    /** Default options bag from connection config. */
    protected array $options = [];

    // ─── Query-state ───
    // Populated by builder methods; cleared by reset() / newQuery().

    protected string $table = '';
    /** 'find' | 'count' | 'insert' | 'update' | 'delete'. */
    protected string $mode = '';

    /**
     * Linear list of where-clauses at the current scope. Each entry is
     * `['type' => 'AND'|'OR', 'expr' => <mongo filter doc>]`. Compiled to
     * a Mongo filter document by compileScopeFilter() with SQL-grade
     * precedence — OR binds looser than AND, so the list is split into
     * AND-runs at each OR boundary, then those runs are OR'd together.
     *
     *   where(A); orWhere(B); where(C)  → list: [AND A, OR B, AND C]
     *                                   → compile: (A) OR (B AND C)
     */
    protected array $whereClauses = [];

    /** Stack of saved $whereClauses snapshots for nested group support. */
    protected array $groupStack = [];

    protected array $projection = [];
    protected array $sort = [];
    protected ?int $limit = null;
    protected int $offset = 0;
    protected ?array $insertDoc = null;
    protected ?array $updateDoc = null;
    /** @var string[] Field names captured by ->fields([...]); echoed via getQueryFields(). */
    protected array $queryFields = [];

    // ─── Result materialisation ───

    /** Array of associative-array rows, materialised by execute(). */
    protected ?array $resultRows = null;
    /** Walking cursor into $resultRows for sequential fetch() calls. */
    protected int $resultPosition = 0;
    /** True if the most-recent execute() was a count query (sets row[0]['count']). */
    protected bool $wasCount = false;
    /** Counter issued by the most-recent insert. 0 = no insert run yet. */
    protected int $lastInsertId = 0;

    /** Per-collection counters live in this Mongo collection name. */
    public const string COUNTERS_COLLECTION = 'vault_counters';

    public function __construct(stdClass $connectionDetails)
    {
        $this->connectionDetails = $connectionDetails;
        $this->database = (string) ($connectionDetails->database ?? '');
        $this->options  = (array) ($connectionDetails->options ?? []);

        $uri = (string) ($connectionDetails->uri ?? '');
        if ($uri === '') {
            throw new MongoUnsupportedException(
                'mongodb vault requires a `uri` config entry (e.g. "mongodb://host:27017")'
            );
        }
        $this->client = new Client($uri, $this->options);
    }

    public function __destruct()
    {
        $this->client = null;
    }

    // ─── interfaces\operators (connection lifecycle + result fetching) ───

    /**
     * Materialise the current query against MongoDB. Populates $resultRows
     * with an array of associative-array rows (BSONDocument flattened). The
     * fetch* methods consume from there.
     */
    public function execute(): operators
    {
        if ($this->mode === '') {
            throw new MongoUnsupportedException(
                'execute() called with no query mode set — call select() first'
            );
        }
        if ($this->table === '') {
            throw new MongoUnsupportedException(
                'execute() called with no table — call select($table) first'
            );
        }

        $collection = $this->collection($this->table);
        $filter = $this->compileFilter();

        if ($this->mode === 'count') {
            $count = $collection->countDocuments($filter);
            $this->resultRows = [['count' => $count]];
            $this->resultPosition = 0;
            $this->wasCount = true;
            return $this;
        }

        if ($this->mode === 'find') {
            $opts = $this->buildFindOptions();
            $cursor = $collection->find($filter, $opts);
            $this->resultRows = [];
            foreach ($cursor as $doc) {
                $this->resultRows[] = $this->bsonToArray($doc);
            }
            $this->resultPosition = 0;
            $this->wasCount = false;
            return $this;
        }

        if ($this->mode === 'insert') {
            if ($this->insertDoc === null || $this->insertDoc === []) {
                throw new MongoUnsupportedException(
                    'execute() called on insert query with no fields([...]) supplied'
                );
            }
            // Caller may pass either a single assoc doc OR a list-of-rows
            // (mirroring mysql's fieldsForInsert). Detect via 0-key:
            $docs = isset($this->insertDoc[0]) && is_array($this->insertDoc[0])
                ? $this->insertDoc
                : [$this->insertDoc];

            $lastId = 0;
            foreach ($docs as $doc) {
                // Honour an explicit caller-supplied `id` (rare; mirrors PDO
                // INSERT with explicit PK). Otherwise issue from counters.
                if (!array_key_exists('id', $doc)) {
                    $doc['id'] = $this->nextSequenceFor($this->table);
                }
                $collection->insertOne($doc);
                $lastId = (int) $doc['id'];
            }
            $this->lastInsertId = $lastId;
            $this->resultRows = [];
            $this->resultPosition = 0;
            $this->wasCount = false;
            return $this;
        }

        if ($this->mode === 'update') {
            if ($this->updateDoc === null || $this->updateDoc === []) {
                throw new MongoUnsupportedException(
                    'execute() called on update query with no fields([...]) supplied'
                );
            }
            $collection->updateMany($filter, ['$set' => $this->updateDoc]);
            $this->resultRows = [];
            $this->resultPosition = 0;
            $this->wasCount = false;
            return $this;
        }

        if ($this->mode === 'delete') {
            $collection->deleteMany($filter);
            $this->resultRows = [];
            $this->resultPosition = 0;
            $this->wasCount = false;
            return $this;
        }

        throw new MongoUnsupportedException("execute() mode `{$this->mode}` not yet implemented");
    }

    public function executeRaw(string $sql): array
    {
        throw new MongoUnsupportedException(
            'executeRaw() is SQL-only; use the typed mongo()->pipeline() escape hatch'
        );
    }

    public function fetch(): stdClass|false
    {
        $row = $this->fetchAssoc();
        return $row === false ? false : (object) $row;
    }

    public function fetchColumn(int $ix = 0): mixed
    {
        $row = $this->fetchAssoc();
        if ($row === false) return false;
        // Stable column-by-index requires the row to remember its key order,
        // which is array_values'd here. Acceptable for the common cases
        // (count queries return [['count' => N]] → fetchColumn(0) === N).
        $values = array_values($row);
        return $values[$ix] ?? false;
    }

    public function fetchAll(): array|false
    {
        $rows = $this->fetchAllAssoc();
        if ($rows === false) return false;
        return array_map(static fn(array $r): stdClass => (object) $r, $rows);
    }

    public function fetchAllAssoc(): array|false
    {
        if ($this->resultRows === null) return false;
        // Reset walker position — the contract is "fetchAll = whole set,
        // not the remainder after sequential fetch()". Mirrors PDO semantics.
        $this->resultPosition = count($this->resultRows);
        return $this->resultRows;
    }

    public function fetchAssoc(): array|false
    {
        if ($this->resultRows === null) return false;
        if ($this->resultPosition >= count($this->resultRows)) return false;
        return $this->resultRows[$this->resultPosition++];
    }

    public function isCount(): bool
    {
        return $this->wasCount;
    }

    /** @return string[] */
    public function getQueryFields(): array
    {
        return $this->queryFields;
    }

    // ─── interfaces\driver — read-path builder ───

    public function select(string $table): driver
    {
        $this->reset();
        $this->table = $table;
        $this->mode = 'find';
        return $this;
    }

    public function selectDistinct(string $table): driver
    {
        // Faithful 'distinct' across arbitrary projections needs $group in
        // an aggregation pipeline. Z1.d ships select(); the distinct
        // variant is queued for when the first consumer needs it (none
        // today) — at which point execute() learns a 'distinct' mode.
        throw new MongoUnsupportedException(
            'selectDistinct() — pending; consumer should reach for mongo()->aggregate() with $group'
        );
    }

    /**
     * Polymorphic on $mode — mirrors the SQL driver:
     *
     *   select:  fields(['col_a', 'col_b'])         → projection
     *   insert:  fields(['col_a' => 'v1', ...])     → insert doc
     *            fields([['col_a' => ...], [...]])  → bulk insert docs
     *   update:  fields(['col_a' => 'v1', ...])     → $set doc
     */
    public function fields(?array $fields = null): driver
    {
        if ($this->mode === 'insert') {
            $this->insertDoc = $fields ?? [];
            return $this;
        }
        if ($this->mode === 'update') {
            $this->updateDoc = $fields ?? [];
            return $this;
        }
        // Default: select-mode projection.
        $this->projection = [];
        $this->queryFields = [];
        if ($fields === null) {
            return $this;
        }
        foreach ($fields as $field) {
            $field = (string) $field;
            // Strip table-qualifier (`table.field`) — Mongo projection has
            // no concept of joins, so we project on the bare field name.
            if (str_contains($field, '.')) {
                [, $bare] = explode('.', $field, 2);
            } else {
                $bare = $field;
            }
            $this->projection[$bare] = 1;
            $this->queryFields[] = $bare;
        }
        // Hide _id by default if the caller has narrowed projection — they
        // asked for specific fields; the implicit _id leak would be a
        // surprise. Z2.a inserts always populate `id` from counters so
        // consumers should be using that anyway.
        if ($this->projection !== [] && !array_key_exists('_id', $this->projection)) {
            $this->projection['_id'] = 0;
        }
        return $this;
    }

    public function where(
        string $field,
        $value,
        WhereCondition|string $condition = WhereCondition::EQUALS,
        WhereType|string $type = WhereType::AND,
    ): driver {
        if (is_string($condition)) {
            $condition = WhereCondition::from($condition);
        }
        if (is_string($type)) {
            $type = WhereType::from($type);
        }
        $this->whereClauses[] = [
            'type' => $type === WhereType::OR ? 'OR' : 'AND',
            'expr' => $this->buildClause($field, $value, $condition),
        ];
        return $this;
    }

    public function orWhere(
        string $field,
        $value,
        WhereCondition|string $condition = WhereCondition::EQUALS
    ): driver {
        return $this->where($field, $value, $condition, WhereType::OR);
    }

    public function orderBy(string|array $fieldOrMultipleOrderBys, SortDirection $sort = SortDirection::ASC)
    {
        if (is_array($fieldOrMultipleOrderBys)) {
            // Caller form: orderBy(['field' => SortDirection::ASC, ...])
            foreach ($fieldOrMultipleOrderBys as $field => $dir) {
                $this->sort[(string) $field] = $this->sortDirectionToMongo(
                    $dir instanceof SortDirection ? $dir : SortDirection::ASC
                );
            }
        } else {
            $this->sort[$fieldOrMultipleOrderBys] = $this->sortDirectionToMongo($sort);
        }
        return $this;
    }

    public function limit(int $limit, int $offset = 0)
    {
        $this->limit = $limit > 0 ? $limit : null;
        $this->offset = max(0, $offset);
        return $this;
    }

    public function newQuery(): driver
    {
        $this->reset();
        return $this;
    }

    public function reset(): void
    {
        $this->table = '';
        $this->mode = '';
        $this->whereClauses = [];
        $this->groupStack = [];
        $this->projection = [];
        $this->sort = [];
        $this->limit = null;
        $this->offset = 0;
        $this->insertDoc = null;
        $this->updateDoc = null;
        $this->queryFields = [];
        $this->resultRows = null;
        $this->resultPosition = 0;
        $this->wasCount = false;
    }

    public function addSubQueryFields(array $fields): void
    {
        throw new MongoUnsupportedException('addSubQueryFields() is SQL-shaped; use mongo()->pipeline()');
    }

    public function resetSubQueryFields(): void
    {
        // No-op while subQuery isn't supported.
    }

    public function resetOrderBy(): void
    {
        $this->sort = [];
    }

    public function resetFields(): void
    {
        $this->projection = [];
        $this->queryFields = [];
    }

    public function startWhereGroup(): driver
    {
        // Save the current scope's clause list onto the stack; start a
        // fresh accumulator for the nested group. endWhereGroup compiles
        // the nested list into a single filter doc and pushes it onto
        // the outer scope as one clause (AND-joined).
        $this->groupStack[] = $this->whereClauses;
        $this->whereClauses = [];
        return $this;
    }

    public function endWhereGroup(): driver
    {
        if ($this->groupStack === []) {
            throw new MongoUnsupportedException(
                'endWhereGroup() called with no matching startWhereGroup()'
            );
        }
        $nested = $this->compileScopeFilter($this->whereClauses);
        $this->whereClauses = array_pop($this->groupStack);
        if ($nested !== []) {
            // Nested group attaches as one AND clause to the outer scope.
            // SQL convention is `WHERE (A OR B) AND C` — outer ops still
            // accumulate normally after the group closes.
            $this->whereClauses[] = ['type' => 'AND', 'expr' => $nested];
        }
        return $this;
    }

    public function orWhereMultiple(array $groupedClauses, WhereType|string $type = WhereType::AND): driver
    {
        // Caller form: orWhereMultiple([['field', value, condition?], ...], type)
        // Treat as: an OR group of the given clauses, attached to the
        // current scope via $type.
        if (is_string($type)) {
            $type = WhereType::from($type);
        }
        $clauses = [];
        foreach ($groupedClauses as $c) {
            if (!is_array($c) || count($c) < 2) {
                continue;
            }
            $field = (string) $c[0];
            $value = $c[1];
            $cond  = $c[2] ?? WhereCondition::EQUALS;
            if (is_string($cond)) {
                $cond = WhereCondition::from($cond);
            }
            $clauses[] = $this->buildClause($field, $value, $cond);
        }
        if ($clauses === []) return $this;
        $this->whereClauses[] = [
            'type' => $type === WhereType::OR ? 'OR' : 'AND',
            'expr' => count($clauses) === 1 ? $clauses[0] : ['$or' => $clauses],
        ];
        return $this;
    }

    public function getConnection(): mixed
    {
        return $this->client;
    }

    public function toCountQuery(): driver
    {
        // Switch the current find query to a countDocuments call on
        // execute(). limit/offset/projection are NOT honoured by count
        // (Mongo's countDocuments ignores them).
        $this->mode = 'count';
        return $this;
    }

    public function keyIsAutoIncrementing(string $keyField, string $table): bool
    {
        // Mongo has no SQL-style auto-increment. The counters strategy (Z2.c)
        // issues int ids explicitly, NOT via collection metadata.
        return false;
    }

    // ─── interfaces\driver — write path (Z2.a-c) ───

    public function insert(string $table): driver
    {
        $this->reset();
        $this->table = $table;
        $this->mode = 'insert';
        return $this;
    }

    public function update(string $table): driver
    {
        $this->reset();
        $this->table = $table;
        $this->mode = 'update';
        return $this;
    }

    public function delete(string $table): driver
    {
        $this->reset();
        $this->table = $table;
        $this->mode = 'delete';
        return $this;
    }

    public function lastInsertId(): int
    {
        return $this->lastInsertId;
    }

    public function executeSql(string $sql, array $params = []): driver
    {
        throw new MongoUnsupportedException('executeSql() is SQL-only — no Mongo equivalent');
    }

    // ─── PERMANENT throws (SQL-only primitives, no Mongo equivalent) ───

    public function describe(string $table): driver
    {
        throw new MongoUnsupportedException(
            'describe() is SQL-DDL; Mongo collections are schemaless. Use validator rules instead.'
        );
    }

    public function createTable(
        string $table,
        stdClass $colList,
        bool $existsCheck = true,
        string $engine = 'INNODB'
    ): driver {
        throw new MongoUnsupportedException(
            'createTable() is SQL-DDL; use MongoMigrationHelper::ensureIndex from a Phinx migration'
        );
    }

    public function having(
        string $field,
        $value,
        WhereCondition|string $condition = WhereCondition::EQUALS,
        HavingType $type = HavingType::AND,
    ): driver {
        throw new MongoUnsupportedException(
            'having() — SQL aggregation primitive. Use mongo()->aggregate() with $group + $match for equivalent.'
        );
    }

    public function orHaving(
        string $field,
        $value,
        WhereCondition|string $condition = WhereCondition::EQUALS,
    ): driver {
        throw new MongoUnsupportedException('orHaving() — see having()');
    }

    public function whereRaw(
        string $condition,
        array $params = [],
        WhereType|string $type = WhereType::AND
    ): driver {
        throw new MongoUnsupportedException(
            'whereRaw() is SQL-only. Restructure with where()/orWhere(), or use mongo()->pipeline().'
        );
    }

    public function orWhereRaw(string $condition, array $params = []): driver
    {
        throw new MongoUnsupportedException('orWhereRaw() is SQL-only — see whereRaw()');
    }

    public function join(
        string $table,
        string|array $field1,
        ?string $field2 = null,
        WhereCondition|string $condition = WhereCondition::EQUALS,
    ): driver {
        throw new MongoUnsupportedException(
            'join() — Mongo $lookup is available via mongo()->pipeline() on Operators only'
        );
    }

    public function leftJoin(
        string $table,
        string|array $field1,
        ?string $field2 = null,
        WhereCondition|string $condition = WhereCondition::EQUALS
    ): driver {
        throw new MongoUnsupportedException('leftJoin() — see join()');
    }

    public function subQuery(Closure $callback, ?string $alias = null): string
    {
        throw new MongoUnsupportedException(
            'subQuery() is SQL-shaped; use mongo()->pipeline() with $lookup pipeline form for cross-collection cases.'
        );
    }

    public function groupBy(string|array $field): driver
    {
        throw new MongoUnsupportedException(
            'groupBy() — SQL aggregation primitive. Use mongo()->aggregate() with $group for equivalent.'
        );
    }

    // ─── Mongo-only escape hatches (typed; not on the driver interface) ───
    // Reached via mongoEntityDatabaseOperator's typed mongo() accessor.

    /**
     * Run an aggregation pipeline against $collection. Returns the
     * materialised array of result documents (each flattened to a plain
     * assoc array, like execute()). Caller owns the stage array; we just
     * forward verbatim.
     *
     * @param  string                           $collection target collection
     * @param  list<array<string,mixed>>        $stages     pipeline stages
     * @return list<array<string,mixed>>
     */
    public function pipeline(string $collection, array $stages): array
    {
        $cursor = $this->collection($collection)->aggregate($stages);
        $rows = [];
        foreach ($cursor as $doc) {
            $rows[] = $this->bsonToArray($doc);
        }
        return $rows;
    }

    /**
     * Alias of pipeline() — matches MongoDB driver naming.
     *
     * @param  list<array<string,mixed>>        $stages
     * @return list<array<string,mixed>>
     */
    public function aggregate(string $collection, array $stages): array
    {
        return $this->pipeline($collection, $stages);
    }

    /**
     * Idempotent index creation. $keys is the Mongo index spec
     * (e.g. ['contact_id' => 1, 'sent_at' => -1]). $opts forwarded —
     * common keys: ['unique' => true], ['name' => '...'],
     * ['expireAfterSeconds' => N] (TTL).
     */
    public function ensureIndex(string $collection, array $keys, array $opts = []): void
    {
        $this->collection($collection)->createIndex($keys, $opts);
    }

    public function dropIndex(string $collection, string $indexName): void
    {
        $this->collection($collection)->dropIndex($indexName);
    }

    /**
     * Bulk write — caller passes the operations array verbatim per
     * the MongoDB PHP library's bulkWrite contract:
     *
     *   [
     *     ['insertOne' => [$doc]],
     *     ['updateMany' => [$filter, $update, $opts]],
     *     ['deleteOne'  => [$filter]],
     *     ...
     *   ]
     *
     * Returns the affected row count (insertedCount + modifiedCount +
     * deletedCount + upsertedCount). For per-op counts the caller can
     * use the typed library directly via mongo()->getConnection().
     */
    public function bulkWrite(string $collection, array $ops): int
    {
        $result = $this->collection($collection)->bulkWrite($ops);
        return $result->getInsertedCount()
             + $result->getModifiedCount()
             + $result->getDeletedCount()
             + $result->getUpsertedCount();
    }

    /**
     * Issue the next sequence number for a collection — atomic
     * findAndModify($inc) against the vault_counters collection.
     * Each collection's counter is `{_id: <collection_name>, seq: <int>}`.
     * Returns 1 on first call per collection. Persistent across processes.
     */
    public function nextSequenceFor(string $collection): int
    {
        $countersCol = $this->client
            ->selectDatabase($this->database)
            ->selectCollection(self::COUNTERS_COLLECTION);
        $doc = $countersCol->findOneAndUpdate(
            ['_id' => $collection],
            ['$inc' => ['seq' => 1]],
            [
                'upsert'         => true,
                'returnDocument' => \MongoDB\Operation\FindOneAndUpdate::RETURN_DOCUMENT_AFTER,
            ],
        );
        return (int) ($doc['seq'] ?? 1);
    }

    // ─── Internal builders ───

    /**
     * Get the collection handle for the current query's table. Lazy —
     * Mongo creates the collection on first write.
     */
    protected function collection(string $name): Collection
    {
        return $this->client->selectDatabase($this->database)->selectCollection($name);
    }

    /**
     * Compile the current filter state into a single Mongo filter document.
     */
    protected function compileFilter(): array
    {
        return $this->compileScopeFilter($this->whereClauses);
    }

    /**
     * Compile a linear `[{type:AND|OR, expr:{...}}, ...]` list into a Mongo
     * filter document with SQL-grade precedence: OR binds looser than AND,
     * so we split the list into AND-runs at each OR-typed entry, then OR
     * the runs together.
     *
     *   [AND A]                      → {A}                      (just A)
     *   [AND A, AND B]               → {$and: [A, B]}           (A AND B)
     *   [AND A, OR B]                → {$or: [A, B]}            (A OR B)
     *   [AND A, AND B, OR C]         → {$or: [{$and:[A,B]}, C]} ((A AND B) OR C)
     *   [AND A, OR B, AND C]         → {$or: [A, {$and:[B,C]}]} (A OR (B AND C))
     *   [AND A, OR B, OR C]          → {$or: [A, B, C]}
     *
     * @param  list<array{type:string,expr:array}> $clauses
     * @return array<string,mixed>
     */
    protected function compileScopeFilter(array $clauses): array
    {
        if ($clauses === []) return [];

        // 1. Split into AND-runs at each OR boundary. The very first clause
        //    is always treated as starting a new run regardless of its type
        //    (SQL `WHERE A` is just A; the leading boolean is meaningless).
        $runs = [];
        $current = [];
        foreach ($clauses as $i => $c) {
            if ($i > 0 && $c['type'] === 'OR') {
                // Close the previous run + start a new one.
                if ($current !== []) $runs[] = $current;
                $current = [];
            }
            $current[] = $c['expr'];
        }
        if ($current !== []) $runs[] = $current;

        // 2. Wrap each run in $and (or unwrap if length 1).
        $orMembers = array_map(
            static fn(array $run): array => count($run) === 1 ? $run[0] : ['$and' => $run],
            $runs,
        );

        // 3. Wrap the OR-runs in $or (or unwrap if length 1).
        return count($orMembers) === 1 ? $orMembers[0] : ['$or' => $orMembers];
    }

    /**
     * Translate a single (field, value, condition) tuple into the matching
     * Mongo filter clause.
     */
    protected function buildClause(string $field, mixed $value, WhereCondition $condition): array
    {
        // Strip table-qualifier — same reasoning as fields().
        if (str_contains($field, '.')) {
            [, $bare] = explode('.', $field, 2);
            $field = $bare;
        }

        return match ($condition) {
            WhereCondition::EQUALS                => [$field => $value],
            WhereCondition::NOT_EQUALS            => [$field => ['$ne' => $value]],
            WhereCondition::GREATER_THAN          => [$field => ['$gt' => $value]],
            WhereCondition::GREATER_THAN_OR_EQUAL => [$field => ['$gte' => $value]],
            WhereCondition::LESS_THAN             => [$field => ['$lt' => $value]],
            WhereCondition::LESS_THAN_OR_EQUAL    => [$field => ['$lte' => $value]],
            // {$eq: null} matches both null-valued AND absent keys (Mongo
            // convention) which mirrors SQL NULL semantics for our use cases.
            WhereCondition::IS_NULL               => [$field => null],
            WhereCondition::IS_NOT_NULL           => [$field => ['$ne' => null]],
            WhereCondition::IN                    => [$field => ['$in'  => is_array($value) ? array_values($value) : [$value]]],
            WhereCondition::NOT_IN                => [$field => ['$nin' => is_array($value) ? array_values($value) : [$value]]],
            WhereCondition::REGEXP                => [$field => new Regex((string) $value)],
            WhereCondition::LIKE                  => [$field => new Regex($this->sqlLikeToRegex((string) $value), 'i')],
            WhereCondition::BETWEEN               => [$field => ['$gte' => $value[0] ?? null, '$lte' => $value[1] ?? null]],
            // JSON_CONTAINS is SQL-specific. Mongo native arrays use $in
            // semantics; for the rare consumer that wants an exact Mongo
            // translation, reach for the typed mongo() escape hatch.
            WhereCondition::JSON_CONTAINS         => throw new MongoUnsupportedException(
                'JSON_CONTAINS is SQL-only; Mongo arrays use $in / $elemMatch — use mongo()->aggregate() if needed'
            ),
        };
    }

    /**
     * SQL LIKE → Mongo regex. `%` → `.*`, `_` → `.`. Other regex special
     * chars escaped. Wrapped in `^...$` to mirror SQL's full-string
     * semantics. Caller passes the case-insensitive flag separately.
     */
    protected function sqlLikeToRegex(string $like): string
    {
        // Escape regex-significant chars BEFORE substituting % and _.
        $escaped = preg_quote($like, '/');
        // preg_quote escapes % and _ on some PHP versions; safe to
        // double-handle.
        $escaped = str_replace(['\\%', '\\_'], ['.*', '.'], $escaped);
        $escaped = str_replace(['%', '_'], ['.*', '.'], $escaped);
        return '^' . $escaped . '$';
    }

    /**
     * @return array<string, mixed>  options array for MongoDB\Collection::find
     */
    protected function buildFindOptions(): array
    {
        $opts = [];
        if ($this->projection !== []) {
            $opts['projection'] = $this->projection;
        }
        if ($this->sort !== []) {
            $opts['sort'] = $this->sort;
        }
        if ($this->limit !== null) {
            $opts['limit'] = $this->limit;
        }
        if ($this->offset > 0) {
            $opts['skip'] = $this->offset;
        }
        return $opts;
    }

    protected function sortDirectionToMongo(SortDirection $dir): int
    {
        return match ($dir) {
            SortDirection::ASC,
            SortDirection::NULL_FIRST,
            SortDirection::NULL_FIRST_ASC,
            SortDirection::NULL_LAST_ASC => 1,
            SortDirection::DESC,
            SortDirection::NULL_LAST,
            SortDirection::NULL_FIRST_DESC,
            SortDirection::NULL_LAST_DESC => -1,
        };
    }

    /**
     * Flatten a Mongo BSONDocument (or array) into a plain assoc array.
     * Recurses into nested documents and arrays. ObjectId is cast to its
     * 24-char hex string for stable JSON / comparison.
     */
    protected function bsonToArray(mixed $doc): array
    {
        $arr = is_array($doc) ? $doc : iterator_to_array((array) $doc);
        $out = [];
        foreach ($arr as $k => $v) {
            if ($v instanceof \MongoDB\BSON\ObjectId) {
                $out[$k] = (string) $v;
            } elseif ($v instanceof \MongoDB\Model\BSONDocument || $v instanceof \MongoDB\Model\BSONArray) {
                $out[$k] = $this->bsonToArray($v);
            } elseif (is_object($v) || is_array($v)) {
                $out[$k] = $this->bsonToArray($v);
            } else {
                $out[$k] = $v;
            }
        }
        return $out;
    }
}
