Core concepts

MongoClient[F, S] → MongoDatabase[F, S] → MongoCollection[F, S, A] mirror the driver's own hierarchy, wrapped in your effect F[_] and stream type S[_]:

trait MongoCollection[F[*], S[*], A]:
  def insertOne(document: A, options: InsertOneOptions = InsertOneOptions.default)(using session: Option[ClientSession] = None): F[InsertOneResult]

  def find(filter: Filter[A] = Filter.all)(using session: Option[ClientSession] = None): FindQuery[F, S, A]

  def updateOne(filter: Filter[A], update: Update[A], options: UpdateOptions = UpdateOptions.default)(using session: Option[ClientSession] = None): F[UpdateResult]

  def deleteOne(filter: Filter[A], options: DeleteOptions = DeleteOptions.default)(using session: Option[ClientSession] = None): F[DeleteResult]

  def findOneAndUpdate(filter: Filter[A], update: Update[A], options: FindOneAndUpdateOptions[A] = FindOneAndUpdateOptions.default[A])(using session: Option[ClientSession] = None): F[Option[A]]

  def aggregate[B](pipeline: Seq[Stage[A]])(using session: Option[ClientSession] = None)(using BsonDocumentDecoder[B]): AggregateQuery[F, S, B]

  def distinct[C, B](field: Field[A, C], filter: Filter[A] = Filter.all)(using session: Option[ClientSession] = None)(using DistinctOf[C, B], BsonDecoder[B]): DistinctQuery[F, S, B]

  def createIndex(index: Index[A])(using session: Option[ClientSession] = None): F[String]

  def watch(options: WatchOptions[A] = WatchOptions.default[A])(using session: Option[ClientSession] = None)(using Streamable[S, ChangeEvent[A]]): S[ChangeEvent[A]]
// count, estimatedCount, insertMany, updateMany, deleteMany, bulkWrite, createIndexes, listIndexes, dropIndex,
// dropIndexes, renameCollection, drop, ...

Every method that reaches the server takes an optional ClientSession and defaults to None — every one the driver has a session overload for, which leaves out estimatedCount and the search-index methods — so none of that is visible until you opt into a transaction.

Field selectors

Field.of[E, A](_.someField) is a macro that reads a field selector at compile time — no strings, no reflection — and gives you a typed path to build filters, updates, and sorts:

val adults = Field.of[User, Int](_.age).gte(18)
val named  = Field.of[User, String](_.name).equalTo("Jenna") && adults
val setAge = Field.of[User, Int](_.age).set(27)
val city   = Field.of[Order, String](_.address.city).equalTo("Barcelona") // dotted paths from nested selectors

Each segment is checked against the case class it is selected from, so _.name.length and _.items.head.sku are compile errors rather than paths that render fine and match nothing. Naming only the entity lets the field's type come from the selector — Field.of[User](_.age) is a Field[User, Int] — which is what a repository hoisting its fields into vals wants.

Selector-derived names are spelled through the collection's FieldNaming. Names that are already what the document stores — a map key, an array position, _id, a shape with no case class — go through at, / or Field.stored, and are used verbatim:

val totals = Field.of[Order, Map[String, Int]](_.totals)
val eur    = totals.at("EUR")                         // "totals" is renamed, "EUR" is not
val first: Field[Order, Item] = itemsField / "0"      // any other stored segment
val id     = Field.stored[Order, ObjectId]("_id")

A stored segment is never renamed: under snakeCase, addressField / "zipCode" renders address.zipCode, whatever the codec wrote. To go on into the fields of a case class held in an Option, or in the elements of a collection, / takes that class's own Field instead, and every segment it adds is derived:

val anySku: Field[Order, String] = itemsField / Field.of[Item, String](_.sku) // "items.sku"

The same / crosses an Option[Address] and an Option[List[Item]]; a field holding neither does not compile. Through an array the path reads the way MongoDB reads it — a filter matches when any element does, a sort takes the smallest or largest — so it serves filters, sorts, projections and indexes. An update names the elements it changes with each, matched or where. The array operators (contains, push, elemMatch, …) apply to an optional collection as they do to the collection itself.

Filters

ageField.gte(13) && ageField.lte(19)     // $and
ageField.notIn(List(40, 41, 42))         // $nin
tagsField.contains("urgent")             // array membership
tagsField.containsAll(List("a", "b"))    // $all
tagsField.hasSize(3)                     // $size
nameField.regex("^A")                    // $regex
scoreField.exists                        // $exists
scoreField.hasType(BsonTypeName.Int)     // $type
scoreField.hasAnyType(BsonTypeName.Int, BsonTypeName.Long) // $type over several types
totalField.mod(4, 1)                     // $mod
flagsField.bitsAllSet(4)                 // $bitsAllSet — also $bitsAnySet, $bitsAllClear, $bitsAnyClear
Filter.text[User]("scala")               // $text
Filter.text[User]("Café", TextSearchOptions.default.withCaseSensitive(true)) // $text, $caseSensitive
Filter.expr(totalField.expr > limitField.expr) // $expr, typed — see Aggregation expressions
Filter.expr[User](someBsonDocument)      // $expr, raw
Filter.jsonSchema[User](schema)          // $jsonSchema
locationField.near(berlin, maxDistance = Some(1500)) // $near — also $nearSphere
locationField.within(GeoShape.Within(area))          // $geoWithin
areaField.intersects(point)                          // $geoIntersects

The geospatial operators take GeoJSON Geometry values — Point, LineString, Polygon and the Multi* forms, with longitude first as MongoDB expects. A polygon's rings have to close, so a ring that does not repeat its first point is refused rather than sent for the server to reject. $geoWithin also takes the legacy 2d shapes through GeoShape.Centre, CenterSphere and Box.

$near sorts by distance and needs a geospatial index; distances are in metres for GeoJSON. The server refuses it inside $or and inside an aggregation $match — Stage.geoNear, as a pipeline's first stage, is the tool there (see Nearest first). $geoWithin needs neither an index nor a sort, so it is the cheaper choice when you only want containment.

The comparisons have symbolic aliases where they read better — ===, =!=, >, >=, <, <= — spelling the same Filter as equalTo/notEqualTo/gt/gte/lt/lte.

An operator that only means something for one kind of field compiles only on that kind: regex on text — a String, an Option[String] or a collection of strings — mod on a number, hasSize on a collection, the $bits operators on an integer (Int, Long or either in an Option). On any other field it is a compile error rather than a filter the server accepts and that matches nothing.

hasType takes a BsonTypeName, not a string. The values are MongoDB's own $type aliases — "object" for a document, "binData", "javascriptWithScope" — which are easy to misspell into a filter that silently matches nothing, so the enum spells them for you. BsonTypeName.of(bsonValue) names the type of a value you already have.

itemsField.elemMatch(...) is the one worth calling out: conditions combined on the array path alone can be satisfied by different elements, and $elemMatch is how you require one element to satisfy all of them.

val bigOrder = itemsField.elemMatch(
  Field.of[Item, String](_.sku).equalTo("abc") && Field.of[Item, Int](_.quantity).gt(2)
)

An array of scalars has no fields to name, so Field.self[A] stands for the element itself. A condition on it renders as the bare operator, and conditions joined with && merge into one document — which is how $elemMatch takes a range that a single comparison could not:

val inRange = scoresField.elemMatch(Field.self[Int].gte(80) && Field.self[Int].lt(85))
// {"scores": {"$elemMatch": {"$gte": 80, "$lt": 85}}}

Filter.and/or fold Filter.all and Filter.none away rather than emitting a one-element $and. That is what makes an empty list safe: field.in(Nil) is Filter.none, not "match everything". A chain — a && b && c — renders as one flat $and rather than each one nested inside the next.

The empty lists answer what the server answers: in(Nil) matches nothing, notIn(Nil) matches everything, and containsAll(Nil) matches nothing, the way $all: [] does — no document is ever counted as containing all of an empty list.

Matching nothing renders as {"_id": {"$in": []}}, and so do Filter.Or(Nil) and !Filter.all. Every collection has an index on _id, so the server answers it without scanning the collection, whichever field the empty in was on.

Updates

ageField.set(31)                  // $set
ageField.inc(1)                   // $inc — also $mul, $min, $max
nameField.unset                   // $unset
scoreField.set(None)              // $unset too, the way the codec leaves a None out; scoreField.setNull writes null
tagsField.push("vip")             // $push — also $pull, $addToSet, and the $each variants
itemsField.pullWhere(qtyField.lt(1))         // $pull by a condition on the elements
scoresField.pullWhere(Field.self[Int].gte(6)) // ...or on a scalar element itself
Update.setOnInsert(nameField, "") // $setOnInsert, for merge-style upserts
Update.rename(oldField, newField) // $rename — also currentDate, popFirst/popLast
Update.pullAll(tagsField, List("a", "b"))     // $pullAll; Update.addAllToSet is $addToSet with $each
Update.currentTimestamp(syncedField) // $currentDate as a BSON timestamp, on a Field[E, BsonTimestamp]
flagsField.bitOr(4)               // $bit — also bitAnd, bitXor

$bit works on integers only, so it compiles on an Int or Long field — or either in an Option — and takes a mask of the field's own width. A missing field counts as zero, the way the server counts it.

Numeric operators only apply to numeric fields, so Update.inc(nameField, 1) does not compile. An Option-typed field takes the unwrapped value — scoreField.inc(5L) on a Field[User, Option[Long]] — because None has no numeric encoding, and the obvious stand-in, $inc by zero, is a write that quietly does nothing.

A numeric operator takes the field's own type, so on a Field[User, Int] only an Int compiles — inc(1L) and mul(1.5) do not. That keeps mongo4s from being the one that changes the stored width. The server can still change it: an $inc that overflows an Int32 is stored as an Int64, silently, and a value that then no longer fits the modelled Int is a decode error naming the value — 2147483711 is out of range for Int. Model a counter that can grow as Long, and read a field whose stored width you do not control as the widest type it can reach.

$min and $max take any type MongoDB orders — a date, a string, an ObjectId — not only a number, because that is what the server compares. $inc and $mul stay numeric, since arithmetic is what they do.

Update.combine/and merge operators of the same name into one sub-document, so setting two fields produces one $set. Update.Raw carries operators the AST does not model and merges the same way:

Update.set(nameField, "bob").and(Update.Raw[User](BsonDocument("$unset", BsonDocument("legacyNickname", BsonString("")))))
// {"$set": {"name": "bob"}, "$unset": {"legacyNickname": ""}}

Rendering never mutates the document you handed to Raw, so the same value can be reused across updates and rendered under more than one FieldNaming.

One operator document holds one change per path, so merging cannot keep two: Update.inc(n, 1).and(Update.inc(n, 2)) would send only the second. An update that changes a path twice is refused when it renders, rather than losing the first change. So is one whose paths overlap across operators — set(name) with unset(name), or unset(notes) with a set on notes.0.rank — which the server would refuse anyway, one round trip later.

Updating from the document's own values

An update that computes from the document itself — a field built from two others, a value run through an aggregation expression — is a pipeline, not operators. Update.pipeline takes the stages the server accepts in one ($addFields, $project, $unset, $replaceRoot, $replaceWith, or a Stage.raw) and refuses any other. The values are aggregation expressions:

val exclaimed = Update.pipeline(
  Stage.addFields[User]("name" := Field.of[User, String](_.name).expr ++ Expr.lit("!"))
)

collection.updateMany(Filter.all, exclaimed)

updateOne, updateMany, findOneAndUpdate and a bulk UpdateOne/UpdateMany all take one. Two pipelines and into one; a pipeline and an operator update do not combine, because the server takes one or the other.

Pushing onto arrays

pushAll takes a PushOptions[A] carrying MongoDB's $each modifiers — $slice, $sort and $position. The classic use is a capped, ordered log kept by the server:

tagsField.pushAll(List("vip"), PushOptions.default[String].sortedAscending.withSlice(10))
// {"$push": {"tags": {"$each": ["vip"], "$slice": 10, "$sort": 1}}}

$sort has two shapes and they are mutually exclusive, so setting one clears the other: sortedAscending/ sortedDescending for arrays of scalars, which the server sorts as bare values, and sortedBy(Sort[A]) for arrays of documents — PushOptions.default[Note].sortedBy(Sort.desc(rankField)) renders "$sort": {"rank": -1}.

Updating array elements

The positional operators are methods on an array field, each giving a field of the element type, and / with another Field reaches into the element — spelled through the collection's FieldNaming like every other selector. each is $[], every element:

val orderId  = Field.of[Order, String](_.id)
val items    = Field.of[Order, List[Item]](_.items)
val quantity = Field.of[Item, Int](_.quantity)

collection.updateOne(orderId.equalTo("1"), Update.set(items.each / quantity, 0))

matched is $, the first element the query matched — so the query has to match one:

collection.updateOne(items.elemMatch(Field.of[Item, String](_.sku).equalTo("abc")), Update.inc(items.matched / quantity, 1))

where is $[identifier], the elements an array filter selects. The identifier is a Field.element, and the same value builds both the update path and the filter, so the two cannot drift apart:

val low = Field.element[Item]("low")

collection.updateOne(
  orderId.equalTo("1"),
  Update.set(items.where(low) / quantity, 100),
  UpdateOptions.default.withArrayFilters(Seq((low / quantity).lt(3))),
)
// {"$set": {"items.$[low].quantity": 100}} with arrayFilters [{"low.quantity": {"$lt": 3}}]

On an array of scalars the element is the value itself: scores.where(high) with high.gte(90) as its filter. An identifier the server would refuse — anything but a lowercase letter followed by letters and digits — is refused when it is built, and so is a where given anything but such an identifier: Field.self would render $[], which is each.

Write options

updateOne/updateMany take an UpdateOptions, replaceOne a ReplaceOptions. Both are immutable, built by chaining from default, and exist so that new options stay additive — a write method's signature never has to grow another parameter again:

UpdateOptions.default                                     // nothing set
UpdateOptions.upsert                                      // shorthand for default.withUpsert
UpdateOptions.upsert.withArrayFilters(Seq(elementQty.lt(3)))
UpdateOptions.default.withCollation(caseInsensitive).withComment("nightly reconcile")

deleteOne/deleteMany take a DeleteOptions and count a CountOptions, on the same pattern:

collection.deleteMany(filter, DeleteOptions.default.withCollation(caseInsensitive))
collection.count(filter, CountOptions.default.withSkip(20).withLimit(10))

What each type carries today: collation, hint and comment everywhere; let variables on every write, which an $expr reads as Expr.variable; bypassDocumentValidation on the two that write whole documents (UpdateOptions, ReplaceOptions); limit, skip and maxTime on CountOptions.

UpdateOptions and ReplaceOptions also take a sort, which picks the one document updateOne/replaceOne changes when the filter matches several (MongoDB 8.0+). It means nothing to updateMany, which changes every match, so updateMany refuses one rather than letting the driver drop it without a word:

collection.updateOne(Filter.all, Update.set(nameField, "oldest"), UpdateOptions.default.withSort(Sort.asc(ageField)))

Inserts take options on the same pattern — InsertOneOptions and InsertManyOptions, with comment and bypassDocumentValidation. An insertMany is ordered by default and stops at the first document that fails; InsertManyOptions.unordered has the server carry on past it, and the BulkWriteFailed it raises lists every failure, with the count of what landed:

collection.insertMany(users, InsertManyOptions.unordered)

Each operation family has its own options type on purpose rather than one shared bag: upsert means nothing on a delete, and a type that offered it there would let you write down a request the server cannot answer.

The findOneAnd* trio works the same way, except its options carry the entity type because sort and projection do — FindOneAndUpdateOptions[A], FindOneAndReplaceOptions[A], FindOneAndDeleteOptions[A]. returnUpdated defaults to true, and returningPrevious is how you ask for the document as it was:

collection.findOneAndUpdate(filter, update)                                          // the updated document
collection.findOneAndUpdate(filter, update, FindOneAndUpdateOptions.default[User].returningPrevious)
collection.findOneAndUpdate(filter, update, FindOneAndUpdateOptions.upsert[User].withSort(Sort.asc(ageField)))

They carry collation, hint, maxTime and comment as well, and the update and replace ones bypassDocumentValidation, the same as the plain writes.

An update that would produce no operators throws instead of sending {} — MongoDB rejects it, and failing at the call site beats a write that silently does nothing.

bulkWrite takes a Seq[WriteCommand[E]] — InsertOne/ReplaceOne/UpdateOne/UpdateMany/DeleteOne/DeleteMany, carrying the same Filter and Update values the single-document calls take:

collection.bulkWrite(
  Seq(
    WriteCommand.InsertOne(User("2", "Bob", 41)),
    WriteCommand.updateOne(named, setAge), // lowercase helpers default upsert = false
    WriteCommand.DeleteMany(ageField.lt(0)),
  )
)

Commands run in the order given; BulkWriteOptions.unordered lets the server carry on past one that fails instead of stopping there, and BulkWriteOptions carries comment, bypassDocumentValidation and let for the whole bulk. The delete commands take a DeleteOptions — WriteCommand.deleteOne(filter, DeleteOptions.default.withHint(byAge)). BulkWriteResult.upsertedIds is keyed by each command's position in the sequence, and so is a failure: a MongoError.BulkWriteFailed names each command that failed by its index, and its result is what the others wrote.

Writing across collections at once

MongoDB 8.0 writes to several collections — in several databases — in one command, the client-level bulkWrite. Each collection turns its own WriteCommand into a write addressed to it, with its own codec and naming, and the client sends them together:

client.bulkWrite(
  List(
    users.write(WriteCommand.InsertOne(User("3", "Cleo", 28))),
    orders.write(WriteCommand.updateOne(orderId.equalTo("1"), Update.inc(totalField, 5))),
  )
)

It takes the same BulkWriteOptions and returns a ClientBulkWriteResult of counts. A failure is a MongoError.ClientBulkWriteFailed naming each failed write by its position and carrying the partial result. A comment, let or bypassDocumentValidation belongs to the whole bulk here, so setting one on a single command's options is refused rather than quietly dropped. It is not a transaction — the writes before a failure stay — but it runs inside one, through withTransaction, when they must land together.

Queries

find returns a builder; nothing is sent until first, all, stream or attempting:

collection
  .find(adults)
  .sort(Sort.asc(nameField))
  .projection(Projection.empty[User].include(ageField).withoutId)
  .skip(20)
  .limit(10)
  .hint(indexKeys)
  .collation(Collation.builder().locale("en").build())
  .maxTime(5.seconds)
  .batchSize(100)
  .comment("adults page 3")
  .all

filter narrows what is already there — it ands with the filter the query was built from — while sort, projection, skip and limit replace. first reads one document rather than the query's limit. find also takes let variables for its $expr, allowDiskUse, noCursorTimeout, and the cursor's cursorType and maxAwaitTime. The same hint/collation/maxTime/batchSize/comment options are on aggregate, which adds allowDiskUse, let, maxAwaitTime and bypassDocumentValidation — the last for a $out or $merge into a validated collection — and distinct takes collation, hint, comment, maxTime and batchSize. estimatedCount takes an EstimatedCountOptions with maxTime and comment.

A tailable cursor on a capped collection stays open after the last document and delivers each one written after it, the way a log is tailed:

collection.find().cursorType(CursorType.TailableAwait).maxAwaitTime(1.second).stream

Its stream is bridged as a live stream, emitting each document as it arrives rather than once a batch fills — the same bridge a change stream takes. all on it never finishes, since the cursor never ends; take from the stream instead.

The numbers are checked where you write them, because MongoDB reads the edge cases as something else: a limit or batchSize below one — 0 is no limit at all, and a negative one a single batch — a negative skip, and a maxTime under a millisecond, which truncates to 0 and so to no time limit. Each throws at the builder, naming the value.

distinct over a field typed as an array answers with the elements' type, because that is what the server answers with — distinct(Field.of[User, List[String]](_.tags)) is a DistinctQuery of String, one value per distinct tag.

Projection keeps inclusion and exclusion apart, because MongoDB rejects a projection mixing them. They are separate types — Projection.empty starts neutral and the first include or exclude commits it — so chaining exclude onto an inclusion projection does not compile at all, rather than silently returning more fields than you asked for. _id is the exception: withoutId drops it from an inclusion projection, giving {"field": 1, "_id": 0}.

slice cuts an array field down server-side, so a document with ten thousand comments does not travel to fetch the last five. It is neither an inclusion nor an exclusion, so it chains onto any of the three without committing the projection either way:

Projection.empty[Post].slice(commentsField, -5)                    // the last five
Projection.empty[Post].sliceFrom(commentsField, skip = 20, count = 10)
Projection.empty[Post].include(titleField).slice(commentsField, 5) // and still an inclusion

The field has to be an array — slice on a scalar does not compile. The document still decodes through the entity's own codec, because the field is present and merely shorter.

Slicing a field the projection also includes keeps it once, sliced. A $slice alone would put the server back into returning every field, so when the sliced fields are all an inclusion holds, _id is kept explicitly to hold it in inclusion mode — and the one combination the server cannot express, only sliced fields and no _id, is refused rather than answered with the whole document.

An inclusion projection can also compute fields from aggregation expressions, in find as well as in $project — Projection.empty[User].include(nameField).compute("initial" := nameField.expr.substring(0, 1)).

Sorting by $text relevance is Sort.byTextScore(), which renders {"score": {"$meta": "textScore"}} and takes the output field's name if you want a different one. It is what makes Filter.text useful — an unranked text search returns matches in whatever order the server finds them:

collection.find(Filter.text("mongodb")).sort(Sort.byTextScore()).all

TextSearchOptions carries the rest of $text: withLanguage for the stemming and stop words, withCaseSensitive and withDiacriticSensitive to tell Café from cafe and café from cafe. Each is left out of the query unless it is set, so the index's own defaults apply.

explain answers the question a typed filter otherwise leaves open — did it use an index?

val plan: IO[BsonDocument] = collection.find(adults).sort(Sort.asc(nameField)).explain()

val measured: IO[BsonDocument] =
  collection.find(adults).explain(ExplainVerbosity.EXECUTION_STATS)

It is on find and on aggregate, runs the query the builder had already produced — options, hint and collation included — and returns the server's plan as a BsonDocument. explain on aggregate describes the whole pipeline, not the $limit-ed form first sends.

The document stays a document because an explain result is not one shape. It is a union that depends on how the server chose to optimise: the same aggregate(...).explain() call returns a top-level queryPlanner when MongoDB collapses the pipeline into a plain query, and a top-level stages array when it does not. Inside, the plan is a recursive tree over an open set of stage names, each carrying its own fields, with a layer per shard on a sharded cluster — and the server stamps explainVersion on it precisely because it reserves the right to change.

ExplainSummary types the part that is stable — the questions you actually ask of a plan:

import mongo4s.results.ExplainSummary

val summary: IO[ExplainSummary] =
  collection.find(adults).explain(ExplainVerbosity.EXECUTION_STATS).map(ExplainSummary.of)

// summary.indexes            List("age_1")
// summary.usedIndex          true
// summary.scannedCollection  false
// summary.sortedInMemory     false — a SORT stage means the order was not index-provided
// summary.execution          Some(ExecutionSummary(returned, keysExamined, docsExamined, durationMillis))

It reads by walking for field names rather than by following a fixed path, so it answers the same way for a find, a pipeline the server optimised into a find, and a pipeline it kept as stages. execution is present only when the plan was actually run — ExplainVerbosity.EXECUTION_STATS or higher. It is a summary, not a model of the output: when it cannot recognise something it says nothing rather than guessing, and the full document is still there.

Partial reads

find(...).all decodes whole entities, which means every modelled field has to be there. When you want a few fields rather than the document, selectAs names the shape you want and gives you exactly that:

val summaries = collection.find(adults).sort(Sort.asc(nameField)).selectAs[(name: String, age: Int)].all
// F[List[(name: String, age: Int)]]

summaries.map(_.map(_.name)) // fields are read by label, not by position

The named tuple is the whole specification. The projection sent to the server is built from its labels, and so is the decoder, so the two cannot drift apart — there is no second place to keep in step, and no case class and codec to declare for a shape you only wanted once. A label that is not a field of the entity, or a field asked for at the wrong type, is a compile error naming it.

selectAs closes the chain: filter, sort, skip and limit are set before it, and first, all, stream and attempting after it. It works the same on a getDirectCollection, which is where it matters most — the entity's WireCodec demands every modelled field, so before this a partial read there was not possible at all.

Labels are derived names, spelled through the collection's FieldNaming exactly as Field.of is, so a (fullName: String) on a snakeCase collection projects and reads full_name.

An Option label reads a field the document does not carry as None, which is the answer reading the whole entity gives for the same document — a projected shape and a full read never disagree about the same absence. A field the entity declares as required but the document is missing is still a decode error naming it, reported per document by attempting.

One field per label: selectAs[(city: String)] cannot reach address.city, because a nested projection comes back as a nested document that a flat result cannot represent. Project the outer field, or read the shape through getCollection with a codec of your own.

Results

Writes report what actually happened. UpdateResult carries matchedCount, modifiedCount and upsertedId, which is the only way to tell "matched but unchanged" from "nothing matched", or to recover the id an upsert generated:

val result = collection.updateOne(filter, update, UpdateOptions.upsert)
result.map(r => if r.wasUpserted then r.upsertedId else None)

DeleteResult carries deletedCount; BulkWriteResult carries all four counts plus upsertedIds, keyed by the position of the command that produced it. InsertOneResult/InsertManyResult carry the stored _ids.

upsertedId is the document's _id, not your primary key, and it is reported only when the upsert actually inserted — an upsert that matched reports matchedCount = 1 and None. Whether that _id is anything you recognise depends on where your key lives:

Read that table as three common arrangements, not as a rule — _id and K are independent. What lands in _id is decided by the entity's codec; PrimaryKey only decides what the repository filters on. Let them disagree and upsertedId is neither:

final case class Doc(_id: String, tenant: String, name: String) derives MedeiaDocumentCodec
object Doc:
  given PrimaryKey[Doc, String] = PrimaryKey.single("tenant")(_.tenant)

repo.upsert(Doc("d1", "tenant-A", "x"))  // K is "tenant-A"; upsertedId is BsonString("d1")

_id can also be an Int, or a whole subdocument, or pinned by the filter (_id equality in an upsert's filter is copied into the document the server creates). That is why the type is BsonValue and not something narrower: at the collection level there is no K to narrow it to, and at the repository level the key is not what the field holds. Repository.insertOne returns K for exactly this reason — the key comes from the entity, which is right in every arrangement above, whereas the id is only ever right about _id.

This all holds identically for collection.replaceOne/updateOne with upsert and for Repository.upsert/ upsertMany — the repository builds the same command.

Under an unacknowledged write concern (w=0) the server sends nothing back, so every one of these comes back empty — zero counts, None for the ids. That is indistinguishable from a write that matched nothing, which is the trade w=0 makes: the driver has no answer to report. Nothing throws, so the write path stays usable; if you need to tell the two apart, do not use w=0. Repository.insertOne/insertMany are the one exception: their key comes from the entity you passed, never from the server, so they still return it under w=0 — a write that failed silently there returns a key just the same, exactly as w=0 asks for.

Reads that survive bad documents

find(...).all fails the whole query if any document does not decode. A collection is rarely written by one version of one service, though, and when it isn't, a document that doesn't fit is a fact about the data rather than a reason to lose the rest of the page. attempting reports each document separately:

val readable: IO[List[User]] =
  collection.find().attempting.all.map(_.collect { case Right(user) => user })

val everything: S[DecodeResult[User]] = collection.find().attempting.stream

DecodeResult[A] is Either[BsonError, A]. It's available on find, aggregate and distinct, and as watchAttempting on a collection's watch — watchAsAttempting at the client and database level, where the element type is named at the call site. Transport errors still fail the effect — only decoding is made per-document.

A change stream is the one place the decode result sits inside the element: watchAttempting gives a ChangeEvent[DecodeResult[A]], so an event whose document does not decode still arrives whole — its operation, its namespace, its key and, above all, its resume token. A consumer can set the bad document aside and still checkpoint past it, instead of reading it again on every restart.

Errors the server raises

A failed write does not arrive as a driver exception. mongo4s translates every failure the driver reports into MongoError, so branching on one is a pattern match rather than a code comparison:

import mongo4s.MongoError

collection.insertOne(user).recoverWith {
  case MongoError.DuplicateKey(_)  => collection.replaceOne(byEmail, user).void
  case MongoError.WriteConflict(_) => retry
}

Every case carries cause, the driver's own exception, so nothing is lost — along with code, labels and hasLabel, which is how withTransaction decides what to retry. writeConcernErrors lists what the write concern reported: the write itself was applied, but not acknowledged the way it was asked to be. On a bulk write that can be the only failure, a BulkWriteFailed whose failures is empty. Errors mongo4s raises itself are untouched: BsonError.DecodingFailure for a document that does not fit, RsBridgeError for the stream bridge.

A DecodingFailure from a WireCodec names where in the document it happened, as BsonError.Nested segments — the field, then the array position or map key under it — so a forty-field entity fails with address -> lines -> 2 -> zip -> Expected string but got int, not with the last line alone. A field missing at the top level is a plain MissingField, since it already names itself. The BsonDecoder collections read the same way: a List, Vector, Seq or Set stops at the first element that fails and names its position, a Map its key.

Translation happens on the Publisher before any runtime sees it, so IO, Task, KIO, rapid.Task and every stream get the same type for the same failure. What MongoError.DuplicateKey deliberately does not carry is the name of the index that was violated: the server sends it, but the driver's WriteError.getDetails comes back empty, so the only remaining source is the human-readable message — and scraping that would be a guess that silently changes with a server upgrade. Read getMessage when you need it.

Primary keys

PrimaryKey[E, K] turns an entity into a key-based filter — a single field, a native _id (ObjectId or your own encoder), or a compound key of any width:

given PrimaryKey[User, String]   = PrimaryKey.single("id")(_.id)
given PrimaryKey[Note, ObjectId] = PrimaryKey.storedId(_.id) // keys on "_id" — see WithId

given PrimaryKey[Order, (userId: String, seq: Int)] =
  PrimaryKey.compound(o => (userId = o.userId, seq = o.seq), FieldNaming.snakeCase)

PrimaryKey.id(_.id) is single("id") spelled short, for the common case.

PrimaryKey.of names the key by the entity's own fields instead of by stored strings. Each selector is checked the way Field.of checks it, and the field is spelled through the collection's naming when the key renders — so a key on userId matches the user_id a snakeCase codec writes, with nothing to keep in step. Two or three selectors make a compound key whose type is a plain tuple:

given PrimaryKey[User, String]                       = PrimaryKey.of(_.id)
given PrimaryKey[Link, (Provider, ProviderId)]       = PrimaryKey.of(_.provider, _.providerId)
given PrimaryKey[Link, (Provider, ProviderId, Kind)] = PrimaryKey.of(_.provider, _.providerId, _.kind)

Filters, keyset paging and ensureKeyIndex all spell such a key through the naming. Given a key in scope, entity.primaryKeyFilter and entities.primaryKeysFilter (from PrimaryKey.primaryKeyFilter/primaryKeysFilter) are the filters on one entity's key and on several; PrimaryKey is a KeyRef, the part of a key that knows only names and values, which is what eqFilter, inFilter, afterFilter and keyOrder are built on.

A compound key is a named tuple, so its labels are the field names — there is no second list of strings to keep in step with the extractors, and no arity ceiling. The labels are Scala identifiers; the optional FieldNaming spells them the way the collection stores them, so userId above is written as user_id while the key still reads as ordinary Scala at every call site:

users.findOne((userId = "u1", seq = 3))

The order of the labels is part of the key's type, so writing them in a different order at a call site is a compile error rather than a key silently assembled wrong.

Field names are known without a key value — that is what lets repository.ensureKeyIndex build the unique index that makes the key a key. Without one, two concurrent upserts on the same key can both miss and both insert.

storedId keys on _id whatever the key's type; the entity's codec has to actually write it there, so an entity whose codec writes id will never match. Note that the names single, id, storedId and make take are stored names, used verbatim: the collection's FieldNaming is not applied to them, so under snakeCase write "user_id", not "userId" — or use PrimaryKey.of.

For entities that don't carry their own id field, WithId[Id, E] ships a ready-made PrimaryKey on _id — no given needed — for an ObjectId and for any id with a BsonEncoder, a String, a Long, a UUID or an opaque id of your own.

inFilter on a compound key produces an $or of $ands; on a single field it's a plain $in — and an empty key list always produces Filter.none, so findMany(Nil)/deleteMany(Nil) are safe no-ops instead of matching every document.

Sessions & transactions

client.withTransaction opens a session, runs the body in a transaction on it, and closes it — on every path:

client.withTransaction {
  users.insertOne(User("2", "Bob", 41)) // Option[ClientSession] is already given here
}

The session is given implicitly to everything inside the body, so a collection or repository call joins the transaction without being told to — no (using Some(session)) at each call site.

A helper defined elsewhere has to ask for it. The session reaches calls that are written inside the body; a method compiled somewhere else was compiled against the default — no session — and its writes commit on their own, outside the transaction, which a rollback then does not undo. Give any helper the parameter and it joins:

def register(users: MongoCollection[F, S, User], user: User)(using Option[ClientSession]): F[Unit] =
  users.insertOne(user).void          // joins the caller's transaction

def register(users: MongoCollection[F, S, User], user: User): F[Unit] =
  users.insertOne(user).void          // commits outside it, silently

The two shapes differ by one using clause and by whether a rollback takes the write with it; there is no warning, because a missing implicit is exactly what the default is there to supply. If you prefer passing a value, open the session yourself with client.startSession and use session.withTransaction.

The same holds for a call built before the transaction and only run inside it. A call takes its session when it is written, not when it runs, so val write = users.insertOne(user) outside the block commits on its own even when write runs inside it. Write the call inside the body, or build it in a function that takes the session.

A transaction does not nest. A client.withTransaction inside another one starts a session and a transaction of its own, which commits when it ends — the outer one's rollback does not take its writes back. A session.withTransaction on a session that is already in a transaction is refused by the driver with an IllegalStateException, and the outer transaction rolls back. To make a helper part of the caller's transaction, give it the using Option[ClientSession] above rather than a transaction of its own.

It commits on success and rolls back on failure and on cancellation: Effect[F] carries a guaranteeCase that sees how the action ended, so an interrupted transaction does not linger on the server until it is reaped. That holds on cats, zio and kyo, and on each of them the session is closed even when the interrupt lands as it is being started. The rapid backend builds guaranteeCase from attempt, which sees success and failure only, so a cancelled rapid transaction is left to the server's timeout rather than rolled back. A rollback that itself fails is attached as a suppressed exception rather than replacing the error that caused it.

It also retries, the way the driver's own withTransaction does. A TransientTransactionError means nothing was committed, so the whole transaction — body included — is started again; an UnknownTransactionCommitResult means the commit may already have landed, so only the commit is asked again rather than the work being redone. Both are bounded by one deadline taken before the first attempt, 120 seconds by default:

import mongo4s.operations.TransactionOptions

client.withTransaction(users.insertOne(User("2", "Bob", 41)), TransactionOptions.default.withRetryTimeout(10.seconds))

client.withTransaction(body, TransactionOptions.withoutRetries) // report the first failure, as 2.x did

TransactionOptions also carries the transaction's own readConcern, writeConcern, readPreference and maxCommitTime. Errors the server did not label are never retried — a failure in your own code fails the transaction immediately, exactly as before.

Two rules follow the driver specification for this API. Between one attempt of the whole transaction and the next there is a short random pause — up to 5 ms, growing by half each attempt to a ceiling of 500 ms — so two clients that conflicted do not retry in lockstep and conflict again. And a commit that ran out of maxCommitTime is not asked again, even though the server labels it as possibly committed: a limit you set is not one to keep retrying past.

The deadline is measured with Effect.monotonic, which has a default implementation reading System.nanoTime; the cats and zio backends override it with their runtime's own clock, so TestControl/TestClock can drive the retry window in a test without waiting on a real one.

If you're reusing one already-open session across more than one transaction, the same behaviour applies to the session itself — it commits and rolls back the same way, but leaves the session's own lifetime to you:

import mongo4s.withTransaction // the extension on ClientSession

for
  session <- client.startSession
  _       <- session.withTransaction(users.insertOne(User("2", "Bob", 41)))
  _       <- session.withTransaction(users.insertOne(User("3", "Carol", 29))) // same session, second transaction
  _       <- IO.delay(session.close())
yield ()

MongoSession.startTransaction/commitTransaction/abortTransaction are there for the fully manual path, where you pass the session explicitly with (using Some(session)) at each call site. Nothing is automatic on that path — including the rollback.

A session is useful without a transaction too — to read your own writes causally, or to read one point in time from a snapshot. client.withSession opens one, gives it to every call in the body the same way, and closes it however the body ends:

import mongo4s.operations.SessionOptions

client.withSession(
  for
    before <- users.count()
    rows   <- users.find().all // the same point in time as the count, whatever is written meanwhile
  yield (before, rows),
  SessionOptions.snapshot,
)

SessionOptions carries causallyConsistent, snapshot and the TransactionOptions a session's transactions start with; a session cannot be both a snapshot and causally consistent, so asking for both throws. startSession takes the same options when you want the session's lifetime in your own hands.

Transactions need a replica set or sharded cluster.

Aggregation pipelines

aggregate takes a Seq[Stage[A]] — a typed pipeline-stage AST, mirroring Filter/Update/Sort, instead of raw BsonDocuments — built with the same Field.of selectors:

import mongo4s.operations.{Sort, Stage}

val pipeline = Seq(
  Stage.matching(Field.of[User, Int](_.age).gte(18)),
  Stage.sortBy(Sort.asc(Field.of[User, String](_.name))),
  Stage.limit(10),
)

// aggregate lives on the collection, and decodes into any type with a BsonDocumentCodec
val adults: IO[List[User]] = collection.aggregate[User](pipeline).all

Grouping uses typed accumulators:

import mongo4s.operations.Accumulator

val byAge = Seq(
  Stage.groupBy(Field.of[User, Int](_.age))(
    "count" -> Accumulator.count[User],
    "names" -> Accumulator.push(Field.of[User, String](_.name)),
  )
)

Beyond sum/avg/min/max/first/last/push/addToSet/count, the accumulators MongoDB added in 5.x and 7.0 are typed too:

Stage.groupBy(cityField)(
  "best"   -> Accumulator.topN(Sort.desc(scoreField), nameField, 3), // also top, bottom, bottomN
  "recent" -> Accumulator.lastN(nameField, 2),                      // also firstN, maxN, minN
  "median" -> Accumulator.median(ageField),
  "p90"    -> Accumulator.percentile(ageField, 0.9),
  "spread" -> Accumulator.stdDevPop(ageField),                      // also stdDevSamp
  "traits" -> Accumulator.mergeObjects(traitsField),
)

Stage.groupAll(accumulators*) is the same stage over the whole input as one group, _id: null. top/bottom and their N forms rank by the Sort they are given and output one field; firstN/lastN follow the order the documents arrive in, so sort before the group if that order matters. firstN, lastN and the ranking forms keep a missing value as null, while maxN/minN skip it. sum, avg, median, percentile and the standard deviations compile only on a numeric field, or an Option of one — the server counts anything else as 0 or null without saying so — and median and percentile always ask for the server's only method, approximate. An n below one, an empty sort and a percentile outside 0..1 are refused where they are written.

Stage covers $match/$project/$sort/$limit/$skip/$count/$unwind/$lookup/$graphLookup/$group/ $addFields/$unset/$replaceRoot/$replaceWith/$facet/$sample/$sortByCount/$unionWith/$bucket/ $bucketAuto/$densify/$fill/$setWindowFields/$geoNear/$redact/$out/$merge, and Atlas Search's $search/$searchMeta/$vectorSearch (see Atlas Search), with Stage.raw(document) as the escape hatch for anything else. $unwind takes includeArrayIndex to number each element it unwinds, and $unionWith a pipeline over the other collection — Stage.unionWith[Order, Order]("archive", List(Stage.matching(recent))).

$addFields, a $group key, an accumulator's input, $replaceWith and $redact take aggregation expressions — computed values, typed:

Stage.addFields[Order]("total" := priceField.expr * qtyField.expr)
Stage.groupByExpr(createdAtField.expr.year)("revenue" -> Accumulator.sum(priceField.expr * qtyField.expr))
Stage.groupByKeys[Order]("city" := cityField.expr, "month" := createdAtField.expr.month)("orders" -> Accumulator.count[Order])
Stage.replaceWith(Expr.mergeObjects(Expr.root[Order], Expr.document("total" := priceField.expr * qtyField.expr)))

$unset drops fields by name — the pipeline's spelling of an excluding $project. It takes one field, and more with and, because several fields of unrelated types have no single Field type to share:

Stage.unset(ageField).and(tagsField)

$redact keeps, prunes or descends into each level of a document by an aggregation expression, and Expr.keep, Expr.prune and Expr.descend are the three answers it can give:

Stage.redact(Expr.cond(levelField.expr === Expr.lit(1), Expr.prune, Expr.descend))

A raw BsonValue still works, with Stage.Descend, Stage.Prune and Stage.Keep as the answers spelled in BSON.

$lookup has three forms. Stage.lookup is the equality join on localField/foreignField; Stage.lookupWith takes a sub-pipeline that runs against the foreign collection, with let binding values from the outer document; and Stage.lookupWithKeys does both — joins on the keys, then narrows what joined with the sub-pipeline, the concise form MongoDB 5.0 added. Its keys are typed on both sides, so joining fields of different types does not compile:

Stage.lookupWith[Order, User](
  from = "users",
  pipeline = List(Stage.matching(Field.of[User, Int](_.age).gte(18))),
  as = "adults",
  let = Some(BsonDocument("owner", BsonString("$user_id"))),
)

Stage.lookupWithKeys[Order, User, String](
  from = "users",
  localField = Field.of[Order, String](_.userId),
  foreignField = Field.of[User, String](_.id),
  pipeline = List(Stage.matching(Field.of[User, Int](_.age).gte(18))),
  as = "adultOwner",
)

Stage.graphLookup[Order, User, String](
  from = "users",
  startWith = BsonString("$user_id"),
  connectFrom = Field.of[User, String](_.managerId),
  connectTo = Field.of[User, String](_.id),
  as = "chain",
  options = GraphLookupOptions.default[User].withMaxDepth(3).withDepthField("depth"),
)

Note the second type parameter: the sub-pipeline and the connectFrom/connectTo fields belong to the foreign collection, so they are typed against it rather than against A. Given its name, they are rendered with the outer collection's FieldNaming — fine when a project uses one convention throughout. Given the foreign collection itself, each form takes the name from it and renders the foreign side through its naming, so two collections that disagree still join on the right fields, and the foreign entity type is inferred:

Stage.lookupWithKeys(users, Field.of[Order, String](_.userId), Field.of[User, String](_.id), List(Stage.matching(adults)), "adultOwner")
Stage.graphLookup[Order, User, String](users, BsonString("$user_id"), Field.of[User, String](_.managerId), Field.of[User, String](_.id), "chain")

Stage.lookup(users, localField, foreignField, as) is the typed equality join. let and GraphLookupOptions are the last argument on these forms, with an overload that leaves them out.

$out and $merge write the pipeline's result into a collection, and both take options that stay out of the way until you need them — with none set they render as the bare collection name the server also accepts:

Stage.out[Order]("archive")                                        // {"$out": "archive"}
Stage.out[Order]("archive", OutOptions.default.inDatabase("cold")) // {"$out": {"db": "cold", "coll": "archive"}}

Stage.merge[Order](
  "archive",
  MergeOptions.default
    .onFields(List("user_id", "seq"))
    .whenMatched(MergeOptions.WhenMatched.KeepExisting)
    .whenNotMatched(MergeOptions.WhenNotMatched.Insert),
)

Aggregation expressions

An Expr[E, A] is an aggregation expression that reads a document of E and computes an A. field.expr turns a Field into one — spelled through the collection's FieldNaming, and reading an Option[A] field as the A it holds — Expr.lit a value, and "name" := expr or field := expr a computed field:

val total: Expr[Order, Double] = priceField.expr * qtyField.expr        // a Double times an Int is a Double
val net                         = total - discountField.expr.orElse(Expr.lit(0.0)) // $ifNull when the field is null
val size                        = Expr.cond(qtyField.expr > Expr.lit(10), Expr.lit("bulk"), Expr.lit("single"))

Stage.addFields[Order]("total" := total, "size" := size)

The operators are typed by the capability the server needs, the same way filters are. Arithmetic compiles on numbers and combines widths the way the server does — an Int with a Long is a Long, anything with a Double a Double, anything with a BigDecimal a BigDecimal — and / is always a Double. A comparison takes two values of one kind, or two numbers. The string operators compile on text, the array ones on a collection, the date ones on an Instant:

plus and minus move a date by an Int or a Long amount of a DateUnit, until counts the units between two.

filter and map hand you the element as an expression, and get reads a field of it:

itemsField.expr.filter(item => item.get(itemQty) > Expr.lit(1))
itemsField.expr.map(item => item.get(itemPrice) * item.get(itemQty))

The element is MongoDB's $$this; nesting one filter or map inside another needs filterAs/mapAs with a name of its own, so the inner one does not hide the outer. Expr.array(a, b, c) builds an array from expressions of one type.

Three things the Scala syntax decides rather than the library. -> binds tighter than >, so a switch branch whose condition is a comparison needs parentheses — (qtyField.expr > Expr.lit(10)) -> Expr.lit("big"). ScalaTest's Matchers bring a === of their own that wins inside a spec, so write equalTo there. And a let variable is read with Expr.variable[Int]("limit"), which has no field to check the name against.

Expr.raw carries any expression the AST does not model. The fake does not evaluate expressions, so it refuses every stage, projection and filter that holds one, by name.

Nearest first

$geoNear is $near as a stage: it sorts by distance, writes each document's distance into a field you name, and — unlike $near — lets any other stage follow. The server takes it only as a pipeline's first stage, and only over a geospatial index:

Stage.geoNear(locationField, Geometry.Point(13.405, 52.52), "distance")

Stage.geoNear(
  locationField,
  Geometry.Point(13.405, 52.52),
  "km",
  GeoNearOptions.default[Place].where(openField.equalTo(true)).withMaxDistance(5000).withDistanceMultiplier(0.001),
)

The point is GeoJSON, so distances are in metres: withDistanceMultiplier rescales what is written, not what withMinDistance/withMaxDistance read, and a negative distance is refused before the server sees it. The field is always sent as key, which the server needs as soon as a collection has more than one geospatial index, and withIncludeLocs names a field to receive the location that matched. Legacy 2d coordinate pairs are not modelled — Stage.raw carries them.

Bucketing, filling gaps and windows

$bucket takes its boundaries as values of the field being grouped, so they are typed rather than raw BSON, and default names the bucket for everything outside them — without it the server rejects such a document:

Stage.bucketBy(ageField, Seq(0, 20, 40), default = Some(BsonString("other")))("count" -> Accumulator.count[Person])

$bucketAuto picks the boundaries itself, aiming for the number of buckets asked for with about as many documents in each. bucketAutoRounded also rounds them to a preferred-number series — R5 to R80, 1-2-5, E6 to E192, or powers of two — which the server only does for numbers, so it compiles only on a numeric field:

Stage.bucketAuto(nameField, 4)("count" -> Accumulator.count[Person])
Stage.bucketAutoRounded(ageField, 5, BucketGranularity.R10)()

$sortByCount is the $group-then-$sort written by hand everywhere: one {_id: value, count: n} per value, most frequent first, ties in no particular order:

Stage.sortByCount(cityField)

$densify fills the gaps in a series, so every step is present whether or not a document was written for it — useful before a window function that would otherwise skip the missing rows:

Stage.densify(ageField, DensifyRange.by(10).within(DensifyBounds.Between(BsonInt32(0), BsonInt32(40))))
Stage.densify(recordedAt, DensifyRange.every(1, DateUnit.Hour).within(DensifyBounds.Partition), Seq(sensorField.path))

$setWindowFields computes an accumulator over a span around each document rather than over the whole group. The sort is what gives a window its direction, so it is required rather than optional; an output with no window covers the whole partition:

Stage.setWindowFields(Sort.asc(ageField), partitionBy = Some(sensorField.path))(
  "runningTotal" -> WindowOutput(Accumulator.sum(ageField)).over(Window.documents(WindowBound.Unbounded, WindowBound.Current))
)

Window.documents counts in rows, Window.range in the sort field's own values — rows sharing a value fall in the same window — and Window.rangeOver does the same for a date field measured in a DateUnit.

The window functions — what only makes sense in a window, never in a $group — sit beside the accumulators in the same output list, as WindowFunctions:

Stage.setWindowFields(Sort.asc(dayField), partitionBy = Some(shopField.path))(
  "total" -> WindowOutput(Accumulator.sum(amountField)),
  "place" -> WindowFunction.rank[Sale],                                  // also denseRank, documentNumber
  "prev"  -> WindowFunction.shift(amountField, by = -1, default = Some(0)),
  "kept"  -> WindowFunction.locf(amountField),                           // also linearFill
  "slope" -> WindowFunction.derivative(amountField, Window.range(WindowBound.At(-1), WindowBound.Current), Some(DateUnit.Hour)),
  "area"  -> WindowFunction.integral(amountField, Window.documents(WindowBound.Unbounded, WindowBound.Current)),
  "trend" -> WindowFunction.expMovingAvg(amountField, 7),                 // or expMovingAvgAlpha(amountField, 0.3)
)

$derivative and $integral need a window, so it is a parameter rather than an option; the others take none. linearFill, the calculus and the moving average compile only on a numeric field, and a moving average over fewer than one document, or with an alpha outside (0, 1), is refused where it is written.

$fill fills the nulls and the missing values a $densify leaves behind — with a value, with the last value seen, or with the line between the neighbours. The last two follow the sort order, so they need a sortBy:

Stage.fill(Sort.asc(recordedAt), partitionBy = Seq(sensorField.path))(
  FillOutput.linear(temperatureField), // numeric fields only
  FillOutput.locf(statusField),
  FillOutput.value(noteField, "none"),
)

Where there is nothing to fill from — no value before a gap for locf, no neighbour on one side for linear — the server writes an explicit null rather than leaving the field missing.

These stages take their partition fields as FieldPath rather than Field, which is why the examples above write sensorField.path: a partition is a list of fields of unrelated types, and Field[E, ?] loses the type the path extension needs. Everything else about them is the usual typed selector.

on names fields of the target collection, so it is a list of stored names rather than Field values — the target's shape is not A. A single field renders as a string and several as an array, which is what the server expects. Note that aggregate(...).first does not append its usual $limit after a terminal stage, since the server rejects anything following $out/$merge.

Two things about the typing. A is the pipeline's starting document type and never changes down the pipeline — a $group or $project invents a new shape, but the stages after it are still typed against A, and the pipeline's real output type is stated once, at aggregate[B]. And a field belonging to a stage's own output — _id after a $group, a facet's counter — has no Field[A, _] to name it, so it goes through Stage.raw:

val buckets = Seq(
  Stage.groupBy(ageField)("count" -> Accumulator.count[User]),
  Stage.raw[User](BsonDocument("$sort", BsonDocument("_id", BsonInt32(1)))), // "_id" is the group's, not User's
)

aggregate asks for a BsonDocumentDecoder[B], not a full codec: a pipeline's output is read and never written, so a read model needs no encoder you would have to invent. A BsonDocumentCodec satisfies it, and aggregate[BsonDocument] works out of the box — useful for $facet and ad-hoc $projects.

aggregateDirect[B] is the AST-free counterpart: it asks for a WireCodec[B] rather than a BsonDocumentDecoder[B] and, on a collection opened with getDirectCollection, decodes the pipeline's output straight off the wire with no BsonDocument in between — the same trip find already makes there.

final case class ByAge(_id: Int, total: Int) derives WireCodec

collection.aggregateDirect[ByAge](Seq(Stage.groupBy(ageField)("total" -> Accumulator.count[User]))).all

It carries the direct path's strictness with it — every field the output type declares has to be present, exactly as for an entity read through getDirectCollection, so a model with a field the pipeline does not produce is a decode error rather than a default. attempting reports it per document.

On a collection opened with getCollection the same call still works — the WireCodec is bridged to a document codec — so the method is about the codec you have, not about which constructor you used.

It is worth reaching for when the pipeline returns a lot. Measured against a real MongoDB it is 1.42× the bridged path over ten thousand documents and indistinguishable from it over ten, where the round trip is all there is — see BENCHMARKS.md.

Indexes

Index[E] is built from the same field selectors, and carries the options MongoDB attaches to an index:

import mongo4s.operations.Index

collection.createIndex(Index.ascending(nameField).descending(ageField).named("name_age"))
collection.createIndex(Index.unique(idField))
collection.createIndex(Index.ascending(createdAtField).expiringAfter(30.days)) // TTL
collection.createIndex(Index.ascending(ageField).where(ageField.gte(18))) // partial
collection.createIndex(Index.text(bioField).withSparse)
collection.createIndex(Index.text(titleField).text(bioField).withWeight(titleField, 10).withDefaultLanguage("spanish"))
collection.createIndex(Index.hashed(idField)) // sharding
collection.createIndex(Index.geo2DSphere(Field.stored[User, Any]("location"))) // also geo2D
collection.createIndex(Index.ascending(ageField).withHidden) // built, but ignored by the planner
collection.createIndex(Index.ascending(Field.stored[User, Any]("$**"))) // wildcard: just a stored path

collection.listIndexes // F[List[BsonDocument]], as the server reports them
collection.dropIndex("name_age")

collection.createIndexes(List(Index.ascending(nameField), Index.descending(ageField))) // one command
collection.dropIndexes // every index but the one on _id

Keys are ordered — a compound index is only usable by queries that respect that order. A key named a second time replaces the first in direction and position, so it moves to the end: Index.ascending(a).ascending(b).descending(a) is {b: 1, a: -1}, which serves queries on b, not on a. A Sort reads its fields the same way. createIndex returns the name the server gave it, and is idempotent, so it's safe on every start. createIndexes builds several in one command and returns their names in the order given; the server builds them together, so if one cannot be built, none is.

A wildcard index needs no API of its own: $** is an ordinary stored path, so Field.stored builds it and the collection's FieldNaming leaves it alone. withWildcardProjection narrows which fields such an index covers, and withCollation attaches one to any index.

expireAfterSeconds is a whole number on the server, so a sub-second TTL is rejected rather than silently truncated to "expire immediately". When "immediately" is what you mean — each document carries its own expiry time and should go when that time comes — say so with expiringAtFieldTime, which is a TTL of zero asked for on purpose:

collection.createIndex(Index.ascending(expiresAtField).expiringAtFieldTime)

A text index weighs its fields with withWeight — a match in a field of weight 10 scores ten times one of weight 1 — and withDefaultLanguage and withLanguageOverride choose the stemming language, the second from a field of each document.

A partial index takes the filter operators the server accepts there: equality, the comparisons, $in, exists, hasType, $and and $or. Negations are not among them — and !f renders as $nor, so Index.where(!f) is one the server refuses. Write the positive form of the condition instead.

From a repository, ensureKeyIndex builds the unique index the PrimaryKey describes, without you restating its fields. A key stored as _id needs none — the server keeps _id unique on its own and refuses a unique flag on it — so for such a key ensureKeyIndex builds nothing and answers _id_, the index that is already there.

Atlas Search

Full-text and vector search run on Atlas Search indexes, which live beside the collection rather than in it. A collection creates, lists, redefines and drops them:

collection.createSearchIndexes(
  List(
    SearchIndex.dynamic[Movie](),                                                       // named "default", every field mapped
    SearchIndex.vector("plots", embeddingField, 1536, VectorSimilarity.Cosine).filteringOn(yearField),
  )
)
collection.listSearchIndexes // F[List[BsonDocument]], each with its status and whether it is queryable
collection.dropSearchIndex("plots")

An index builds in the background, and a search against it finds nothing until listSearchIndexes reports it queryable. SearchIndex.of(name, definition) takes a mapping you write yourself, and the vector index compiles only on a numeric array field.

$search takes a SearchOperator. The common ones are typed — the text ones on text fields, between/atLeast/ below on fields the server orders — and compound combines them; SearchOperator.raw carries the rest:

val matrix = SearchOperator.compound[Movie](
  must = List(SearchOperator.text(plotField, "hacker").orIn(titleField).fuzzy(1)),
  filter = List(SearchOperator.atLeast(yearField, 1999)),
)

collection.aggregate[Movie](
  Seq(
    Stage.search(matrix, SearchOptions.default.withCount(SearchCount.total)),
    Stage.project(Projection.empty[Movie].include(titleField).compute("score" := Expr.searchScore)),
  )
)

SearchOptions.default.withIndex("movies") names the search index when it is not default. SearchOperator also has phrase, autocomplete, wildcard, regex, equalTo, below and exists. $searchMeta takes the same operator and answers with counts instead of documents. $vectorSearch finds the nearest documents to a vector, among numCandidates approximate ones or exactly with exactVectorSearch, optionally narrowed by a Filter on the fields the index declared with filteringOn:

Stage.vectorSearch("plots", embeddingField, queryVector, limit = 10, numCandidates = 200, filter = Some(yearField.gte(2000)))

All of this runs against a local mongodb/mongodb-atlas-local container as well as against Atlas — Testcontainers ships MongoDBAtlasLocalContainer for it, which is how this library's own tests exercise it. The fake has no search index, so it refuses both the index calls and the stages by name.

Creating a collection

MongoDB creates a collection on first write, so createCollection is for the cases where the shape has to be declared up front — a capped log, a schema the server enforces, a time series, a clustered collection:

import mongo4s.operations.{ClusteredIndex, CreateCollectionOptions, TimeSeries}

database.createCollection("events", CreateCollectionOptions.capped(8 * 1024 * 1024).withMaxDocuments(10000))

database.createCollection(
  "readings",
  CreateCollectionOptions.default
    .withTimeSeries(TimeSeries.on("recordedAt").withMetaField("sensor").withGranularity(TimeSeriesGranularity.MINUTES))
    .expiringAfter(30.days),
)

database.createCollection("people", CreateCollectionOptions.default.withValidator(jsonSchema))
database.createCollection("orders", CreateCollectionOptions.default.withClusteredIndex(ClusteredIndex.onId))

withCapped takes the size because the server rejects a capped collection without one — there is no way to spell a broken request here. Two combinations the server accepts and then quietly ignores are rejected instead: a maxDocuments on a collection that is not capped, and a validation level or action with no validator to apply to. Sub-second expiringAfter is refused for the same reason it is on an index — the server stores whole seconds, and truncating to zero means "expire immediately".

ClusteredIndex.onId is the only clustered key offered because _id is the only one the server clusters on today. Anything else this API does not model is still reachable through database.runCommand.

renameCollection moves a collection to a new name in the same database. It refuses to replace an existing collection unless dropTarget says to, and the handle it was called on keeps pointing at the old name — now empty — so open the collection again under the new one:

collection.renameCollection("people_archive")
collection.renameCollection("people_archive", dropTarget = true)

A move across databases is an admin command, so it goes through runCommand on the admin database.

Change streams

watch exists at all three levels, matching the driver's own scope hierarchy — MongoClient.watch (the whole deployment), MongoDatabase.watch (one database), and MongoCollection.watch (one collection). All of them need a replica set or sharded cluster:

val events: S[ChangeEvent[User]] = collection.watch()

Every event is a ChangeEvent[A], not a bare BsonDocument:

final case class ChangeEvent[A](
  operationType: OperationType, // com.mongodb's own enum — INSERT/UPDATE/DELETE/...
  namespace: Option[Namespace],   // the database and collection it happened in
  destination: Option[Namespace], // where a rename moved the collection to
  documentKey: Option[BsonDocument],
  fullDocument: Option[A],
  fullDocumentBeforeChange: Option[A],
  updateDescription: Option[UpdateDescription], // updated/removed field paths, for UPDATE events
  resumeToken: BsonDocument,
  clusterTime: Option[BsonTimestamp],
  wallTime: Option[BsonDateTime],  // when the change was applied, in server wall-clock time
  splitEvent: Option[SplitEvent],  // set only on a fragment of an event too large for one message
)

MongoCollection.watch decodes through the collection's own codec, so fullDocument/fullDocumentBeforeChange come back as Option[A]. MongoClient.watch/MongoDatabase.watch span more than one document shape, so they default to ChangeEvent[BsonDocument] — use watchAs[A] when you know the events all decode the same way.

At those two levels namespace is how you tell which collection an event came from — Namespace(database, collection), with no collection for an event about the whole database, such as its drop.

Everything else is WatchOptions[A]:

import mongo4s.changestream.WatchOptions

collection.watch(
  WatchOptions
    .resumeAfter[User](token) // shorthand for default[User].resumingAfter(token)
    .withFullDocument(FullDocument.DEFAULT)
    .withMaxAwaitTime(2.seconds)
    .withBatchSize(64)
)

WatchOptions.default[User].startingAfter(token) // the other two starting points, same builder
WatchOptions.default[User].startingAt(timestamp)

fullDocument defaults to UPDATE_LOOKUP, not the server's own default — MongoDB fills the document in only for inserts and replaces, which leaves the most common question ("what does this document look like now?") unanswered on updates. It is always None for deletes; there is no document left to look up. fullDocumentBeforeChange needs pre-images enabled on the collection.

resumeToken is on every event so a consumer that dies mid-stream can restart from just after the last event it actually handled, rather than from now:

collection.watch().evalTap(handle).evalTap(e => saveToken(e.resumeToken))
// later, on restart:
collection.watch(WatchOptions.resumeAfter[User](savedToken))

resumingAfter and startingAfter are alternatives — each clears the other, since the server rejects a stream carrying both resumeAfter and startAfter.

withExpandedEvents is the driver's showExpandedEvents: DDL events — createIndexes, drop, rename and the rest — are reported alongside the document ones. It needs MongoDB 6.0, so the flag is sent only when you ask for it rather than as a default an older server would reject.

clusterTime and wallTime answer different questions: the first is the logical timestamp the change is ordered by, the second is the server's wall clock when it was applied — useful for lag, useless for ordering. splitEvent is set only on the fragments of an event too large for one message, and says which fragment of how many; without $changeStreamSplitLargeEvent in the pipeline you will never see it.

WatchOptions.pipeline filters the change stream itself, and matches against the change event's own shape ({operationType, fullDocument, ns, ...}), not the collection's document shape. A Field.of path is therefore the wrong tool: it would render "age" where the event needs "fullDocument.age". Use Stage.raw, or Field.stored for a path under fullDocument:

val insertsOnly = WatchOptions
  .default[User]
  .withPipeline(Seq(Stage.raw(BsonDocument("$match", BsonDocument("operationType", BsonString("insert"))))))

collection.watch(insertsOnly)

A change stream never completes on its own — take from it, or interrupt it. A document that does not decode ends it with an error; watchAttempting — watchAsAttempting at the client and database level — reports the failure and carries on instead, which is what you want for a long-lived subscription. Its elements are ChangeEvent[DecodeResult[A]]: the failure is on the document, and the event around it — resume token included — still arrives, so a consumer can checkpoint past a document it could not read.

fullDocument = UPDATE_LOOKUP is a second read per update event, done by the server while it builds the event, and the whole document on the wire. A consumer that only needs to know what changed can ask for FullDocument.DEFAULT and read updateDescription instead.