Repositories

BaseMongoRepository[F, S, E, K] implements Repository — count/find/insert/upsert/update/delete/bulkWrite, batched by batchSize (default 500) — over any MongoCollection[F, S, E], from either codec path.

Reads and writes that take a condition rather than a key come in two spellings, and the name says which: …ByField takes a field and a value, …ByFilter takes a whole Filter. findByField/findByFilter return a List, getByField/getByFilter the same rows as a stream, and updateByField/updateByFilter and deleteByField/deleteByFilter write to every row that matches:

users.exists("1")                                 // F[Boolean], by key
users.updateField("1", ageField, 31)              // one field of one entity, by key
users.getAll                                      // every entity, as a stream
users.findByField(nameField, "alice")             // List[User]
users.getByField(nameField, "alice")              // the same, as a stream
users.updateByField(ageField, 30, birthday)       // every 30-year-old
users.deleteByField(ageField, 30)                 // likewise

A …ByField call is exactly its …ByFilter counterpart over field.equalTo(value) — there for the common case that would otherwise spell out a one-field filter.

A few more cover what services otherwise build by hand:

users.findOneByFilter(adults)                                  // F[Option[User]]
users.estimatedCount                                           // the collection's metadata count
users.updateByKeys(List("1", "2"), birthday)                   // one update, every listed key, batched
users.updateEach(List("1" -> nameField.set("al"), "2" -> birthday)) // each key its own update, one bulk write
users.upsertKeeping(fresh, List(Field.of[User](_.score).path))      // upsert, leaving score as stored
users.findOneAndReplace(user)                                  // replace under user's key, F[Option[User]] after
users.findOneAndDelete("1")                                    // delete under a key, F[Option[User]] it was

upsertKeeping is for an entity some of whose fields another writer owns. Each entity is upserted with $set on every field the codec writes except the kept ones, the key and _id, so an existing document keeps those fields as they were, and a new one is inserted without them. A field the codec leaves out — a None — is left as stored too, not removed. A Repository written by hand rather than through BaseMongoRepository inherits findOneByFilter and estimatedCount built from its own methods; the others need a key and a codec the trait does not have, and throw until it implements them. findOneAndReplace answers the document after the replacement unless its options say returningPrevious, and both take the projection and sort options the collection's own do, the repository's defaultProjection applying when you give none.

bulkWrite takes the same BulkWriteOptions the collection does, and insertMany the same InsertManyOptions. An ordered write is sent in batches, which keeps its meaning — it stops at the first failure and what came before it stays applied. An unordered one is sent as one command, because the point of unordered is that the server applies everything it can, and batching would stop at the first batch that failed instead.

A failure in a later batch is reported as the whole write, the way the driver reports its own internal batches: BulkWriteFailed names the failing command by its position in the sequence you passed, not within its batch, and its result counts what the earlier batches wrote. insertMany and upsertMany batch the same way and report the same way.

findMany(keys) is one $in per batch, so it answers the way a query does: in no particular order, without the keys that matched nothing, and once per document however often its key was asked for. Key the result by pk.key when you need it lined up with the input.

A findOneAndUpdate projection you pass wins over the repository's defaultProjection; the default applies when you pass none.

BaseMongoRepository(collection)                                // over a collection you already have
BaseMongoRepository.create[F, S, E, K](db, "collection")       // reads return the document as stored, _id included
BaseMongoRepository.createDirect[F, S, E, K](db, "collection") // the same over getDirectCollection and a WireCodec
BaseMongoRepository.withoutId[F, S, E, K](db, "collection")    // strips _id from reads — for entities that do not model it
BaseMongoRepository.objectId[F, S, E](db, "collection")        // WithId[ObjectId, E], auto _id round-trip
BaseMongoRepository.direct(db, "collection")                   // a Repository over a WireCodec, typed by where it goes
BaseMongoRepository.objectIdDirect[F, S, E](db, "collection")  // WithId[ObjectId, E] over a WireCodec

direct is the one to reach for when the entity has a WireCodec. It reads and writes through getDirectCollection, so the field names come from the codec's own naming and no FieldNaming is passed a second time, and it answers F[Repository[F, S, E, K]], so the type it is assigned to decides E and K — no type arguments, no .widen:

type PlayerRepository[F[*]] = Repository[F, CatsStream[F], Player, PlayerId]

def players[F[*]: Async](db: MongoDatabase[F, CatsStream[F]]): F[PlayerRepository[F]] =
  BaseMongoRepository.direct(db, "players", projection = Projection.excludeId[Player])

projection is what withoutId sets, and batchSize is the same as everywhere else. objectIdDirect is the direct objectId: a WithId[ObjectId, E] has a WireCodec whenever E is a derived case class, writing _id first and the entity's fields beside it under the entity's naming.

create is the default, and it is the right one far more often than it looks: none of the four codec backends rejects a document carrying an _id the entity does not model — medeia, zio-bson, calypso and a bridged WireCodec all decode it and ignore the field. So withoutId is not needed to make such an entity readable.

Reach for it when a codec you wrote is strict about unknown fields, or when you would rather not carry _id over the wire at all. Never reach for it for an entity keyed on _id: the projection strips the key, and it comes back missing.

insertOne returns the entity's K, and insertMany the K of every entity, in insertion order and across batches. A repository carries a PrimaryKey[E, K], so it can name what it just inserted rather than hand back a raw BsonValue for you to decode. MongoCollection has no key type and keeps returning the driver's InsertOneResult/InsertManyResult — that is where the server's own _id still lives.

insertIfAbsent inserts an entity only when nothing holds its key yet, and answers whether it did. upsert overwrites what is there; insertOne fails with MongoError.DuplicateKey, and inside a transaction that failure ends the transaction. insertIfAbsent is an upsert that only writes on insert — every field of the entity under $setOnInsert, matched by key — so a key that is already there leaves the stored entity untouched and the transaction intact:

users.insertIfAbsent(User("1", "alice", 30)) // F[Boolean]: true when it inserted, false when the key was taken

replace swaps the entity stored under the same key, and never inserts. It takes an optional condition the stored entity must still meet, which is what optimistic concurrency needs. Keep a version on the entity, replace only while the version is the one you read, and treat a matchedCount of zero as "someone else wrote first":

users.replace(user.copy(name = "alicia", version = user.version + 1), versionField.equalTo(user.version))
// matchedCount == 0 → it changed (or went) under you: read it again and retry

An operator update under the same guard is updateByFilter with the key and the version in the filter. Adding an array element only when no element with the same id is there yet is the same shape — the condition goes in the filter, and a miss means it was already there:

orders.updateByFilter(orderId.equalTo("1") && !items.elemMatch(sku.equalTo("abc")), Update.push(items, item))

Paging goes through Page:

import mongo4s.repositories.Page

users.findByFilter(adults, Page.sortedBy(Sort.asc(nameField)).skipping(20).taking(10))
users.getByFilter(adults, Page.first(100)) // same, as a stream

Skip-based paging re-scans what it skips, so it degrades on deep pages — and a row deleted earlier in the collection shifts everything after it, so a walk can miss rows it never saw. findPage avoids both by asking for what comes after a key rather than for an offset:

val firstPage = users.findPage(100)                    // KeysetPage(items, next)
val nextPage  = users.findPage(100, after = firstPage.next)

users.findPage(100, after = Some(lastKey), filter = adults) // narrowed, still paged by key
users.findPage(20, descending = true)                       // newest key first

A page is a KeysetPage[E, K]: its items, and next — the key to pass as after for the page that follows, or None when there is none. The repository asks for one row more than the page holds to know which, so the last page is reported as last rather than discovered by fetching an empty one.

The repository already knows the key's fields, so it builds both halves itself: the sort that gives the page an order, and the comparison that steps past the cursor. For a compound key that comparison is lexicographic — an $or over "first field greater, or first equal and second greater, or …" — which is the part that is easy to get subtly wrong by hand. ensureKeyIndex builds the index that makes the whole walk a range scan.

The cursor is the key of the last row you handled, so a crash resumes from there rather than from an offset that no longer means the same thing. The walk is only as sound as the key: it has to be unique, it has to compare the way the server orders it — keep one BSON type per key field, since a mixed-type field sorts by type before value — and it should be indexed, which is what ensureKeyIndex is for.

ensureKeyIndex builds the unique index the PrimaryKey describes. Nothing builds it for you: indexes are often owned by migration scripts rather than by the application, so the repository leaves the decision where it belongs. A key on _id needs nothing, since the server indexes _id uniquely on its own; any other key without that index is not unique — two inserts with the same key both land, and so do two insertIfAbsent calls that race. WithId[Id, E] wraps an entity with a separately-typed id (type Oid[E] = WithId[ObjectId, E]) and ships its own PrimaryKey/BsonDocumentCodec instances, for entities that don't carry their own id field — it lives in mongo4s-core, so it's available whether or not you use the repository layer.

Like MongoCollection, every Repository method takes (using session: Option[ClientSession] = None) — a repository call can join a transaction the same way a raw collection call can.

BaseMongoRepository is open: adding a domain query means declaring it against collection/Filter/Field directly, not reimplementing what's already there.

Counters

MongoDB has no sequences. The usual stand-in is a collection of counters, one document per name, moved by a single findOneAndUpdate that increments and upserts — one round trip, atomic on the server, and the first call creates the counter. MongoCounters is that, given a database and a collection name:

import mongo4s.repositories.MongoCounters

for
  counters <- MongoCounters.create[IO, S](database)            // the "counters" collection, or name your own
  number   <- counters.next("orders")                           // 1, then 2, 3, …
  block    <- counters.next("ids", step = 100L)                 // reserve a block of 100: this is its last value
  current  <- counters.current("orders")                        // Option[Long], without moving it
  _        <- counters.reset("orders", 1000L)                   // the next one is 1001
yield number

Each counter is a document {_id: name, value: n}, so the key is the counter's name and the _id index is all it needs. next hands every value out exactly once however many callers race for it, and like everything else it takes a session, so a number can be drawn inside a transaction. Counters[F] is the trait, for a double of your own; FakeCounters in the testkit is the real implementation over a FakeMongoCollection.

For unit tests, FakeMongoCollection (in mongo4s-testkit, a published module — add it as "now.sendline" %% "mongo4s-testkit" % "6.0.0" % Test) implements MongoCollection in memory. The exact same Filter/Update/Field AST the real driver interprets is interpreted against an in-memory buffer instead, so repository logic is testable without a running MongoDB.

Filters, updates, sorting, paging and projections are simulated, and so is a subset of aggregate: $match, $sort, $skip, $limit, $project, $unset, $count, $sortByCount, $fill and $group with $sum, $avg, $min, $max, $first, $last, $push, $firstN, $lastN, $maxN, $minN, $top, $bottom, $topN, $bottomN, $stdDevPop, $stdDevSamp and $mergeObjects. That covers the pipelines most service code writes, so aggregating no longer forces a test to reach for Docker.

distinct is simulated too, array fields flattened into their elements the way the server does, and so is $slice in a projection. Filters follow the server's array semantics — a filter on an array field matches its elements, a dotted path descends through an array of documents but not into an array inside it, a numeric segment (items.0.name) picks that position as well as a field of that name, and $size, $type, $mod, $elemMatch and the $bits operators look through arrays as the comparisons do — $regex searches rather than anchors and honours its options, a regular expression inside $in matches the strings it matches, $eq null matches a missing field, $all on a scalar field matches its one value, $all: [] matches nothing, $elemMatch and pullWhere take a condition on a scalar element as well as on a document one, $inc keeps the width the server keeps, and an insert stamps an _id and refuses a duplicate one — through insertOne, insertMany (ordered or not), a bulkWrite insert, and an upsert that inserts, so wasUpserted and upsertedIds say what the server would say. A sort on updateOne/replaceOne picks the document the server would pick.

Numbers compare the way the server compares them: an Int64 of 5 equals an Int32 of 5 and a Double of 5.0, Longs past Double's precision sort exactly, and NaN sorts below every number. Strings compare by their UTF-8 bytes, as the server's binary comparison does, so a character outside the Basic Multilingual Plane sorts after U+FFFF rather than among the surrogates. Values of different types compare in BSON's order — null below numbers, numbers below strings, and so on up to dates — so a range filter matches only values of its operand's type, element by element through an array; a sort treats a missing field as null and an array by its smallest element ascending and its largest descending; $min/$max replace across types; and a group's $min/$max skip null. A field holding null — which every codec but the direct one writes for a None — sorts and filters as it does on the server rather than failing the test.

A projection through an array keeps the field in each element, as the server does — an element without it becomes {} and a scalar element is left out of an inclusion — and distinct reads a dotted path through an array of documents. distinct and $group treat numbers of different widths as one value, first decodes only the document it returns, and a write error inside a bulkWrite — a changed _id, a path the update cannot reach — is a BulkWriteFailed naming the command, as a duplicate is.

A duplicate in insertMany or a bulkWrite is a BulkWriteFailed naming the command, after the writes before it — and, unordered, the writes after it too. A unique index is enforced, sparse and partial ones included, and is named as the server names it (email_1), so dropIndex finds it. listIndexes describes each index as the server does — v, key, name and its options, after the _id_ index — creating one that already exists changes nothing, one that clashes with an existing index's name or keys fails with the server's code, createIndexes builds all or none, and dropIndex refuses an index that is not there, and _id_. A unique index on _id alone is refused with the server's code too, and so is a partial index whose filter holds an operator the server does not accept there. An insert stores the _id first, as the server does. A replacement keeps the _id of the document it replaces, an update that changes nothing counts as matched but not modified, and so does a replacement identical to the stored document, _id included, and a findOneAnd* whose projection leaves a document the codec cannot read fails, as the real collection does, rather than answering None.

The fake streams through the same RsBridge the real collection does, and reads the collection when the stream runs, not when it is built — a stream set up before an insert sees the insert. aggregate, distinct and selectAs stream as well as find.

One fake can be shared between fibers. Each operation runs start to finish under one lock, so concurrent inserts all land, a duplicate _id or unique key lets exactly one of them through, $incs from many fibers add up, and findOneAndUpdate hands each document to one caller — what a test of a counter or a work queue relies on. That is stricter than the server, which promises atomicity per document: every answer the fake gives is one MongoDB could give, but a race across documents that the server could expose will not show here.

Every update operator is simulated — $set, $unset, $inc, $mul, $min, $max, $rename, $currentDate (a date or a timestamp), $push with its $each/$position/$slice/$sort modifiers, $pull by value or by condition, $pullAll, $pop, $addToSet and $bit — and $setOnInsert correctly does nothing on an update that matches. A path reaches an element by its position (items.0.qty), padding the array with nulls to a position past its end, and a path that runs into a value that is neither a document nor an array is refused with the server's code 28; $inc and $mul of a value that is not a number are refused with its code 14. A negative $position counts from the end, $pull reads a document as a query on each element, extra fields and all, and $pull, $pullAll and $addToSet compare numbers across widths. The positional operators are not: each, matched and where paths are refused by name, rather than written to a field called $[].

Everything else fails with UnsupportedOperationException naming what was asked for, rather than quietly answering wrong — a fake that lies is worse than no fake. The failure is raised where the real operation would fail — in F, or in the stream — so attempt sees it. That includes watch, explain, renameCollection — the fake holds a single namespace, with nowhere to move it — a write for the client-level bulkWrite, for the same reason, $text and ranking by $meta textScore, every aggregation expression and so $expr, Atlas Search's indexes and stages, a tailable cursor, $jsonSchema, the geospatial operators, Filter.Raw, Stage.Raw, Update.Raw, a pipeline update, a collation anywhere it is accepted, an update carrying arrayFilters, a unique index over an array field, every aggregation stage outside the list above, $median and $percentile, whose approximate method only the server knows, and the $addToSet accumulator, which is refused on purpose: MongoDB leaves the order of its result undefined, so no fake can be faithful to it. $inc, $mul, $sum and $avg over a Decimal128 are refused as well, since the rounding is the server's to define.

Two kinds of option are not refused. A sort on updateMany throws IllegalArgumentException, as the real collection does, because the server has no such option. And the options that only tune how the server runs a query — hint, maxTime, batchSize, comment, allowDiskUse, maxAwaitTime, noCursorTimeout — are accepted and ignored, since they cannot change the answer; so is let, whose variables only an expression could read, and expressions are refused.

Upserts insert on a miss the way the server does, whether they replace or update — updateOne, updateMany, findOneAndUpdate, findOneAndReplace, replaceOne and their bulkWrite commands. An update builds the new document from the filter's equality conditions — $eq, a one-value $in or $or, every value of an $all, through $and — then applies its operators, $setOnInsert included, and stamps an _id first if none came from the filter. Two equalities on one path are refused with the server's code 54, $all with several values among them. A replacement keeps only the _id from the filter.

One caveat carries over from find: where MongoDB does not define an order, neither does the fake, and the two need not agree. $group emits its buckets in the order their keys were first seen, and distinct the order it first met each value. End the pipeline with $sort, or sort the values yourself, if the order matters — as you would have to against a real server anyway.

FakeRepository is the same idea one layer up: a real BaseMongoRepository over a FakeMongoCollection, so the repository logic under test is the one that ships. repository.fake reaches the collection underneath, for seeding and for snapshot. Both take the runtime's Effect and RsBridge from scope — the same givens a real client uses.

A fake stores documents, so it needs a BsonDocumentCodec and the naming a query is spelled with; with a WireCodec, direct takes both from it, the way getDirectCollection does. FakeMongoCollection[IO, S, User](codec, naming = …) is the long form:

val users = FakeRepository.direct[IO, S, User, String]()
val raw   = FakeMongoCollection.direct[IO, S, User]()

FakeRepository is open, so a service's own repository trait — the Repository plus the finders that service adds — can be tested over the real logic rather than a hand-written double:

val players: PlayerRepository[IO] =
  new FakeRepository[IO, S, Player, String](FakeMongoCollection.direct()) with PlayerRepository[IO]:
    def adults: IO[List[Player]] = findByFilter(Field.of[Player](_.age).gte(18))

The fake takes a session and ignores it: each operation is atomic on its own, and nothing groups several of them. A withTransaction body over the fake runs, but an error in it does not undo the writes that came before, so a test of rollback needs a real replica set.