MongoDB

The mongodb connector owns a MongoDB client (the connection string, credentials, and pool) and the mongodb-* blocks bind to it by name. It is the document-store counterpart to database: a message body is already a document, so storing a webhook payload is one block rather than a column mapping.

It is a service connector and provides no source. The blocks that bind to it are mongodb-find, mongodb-insert, mongodb-update, mongodb-delete and mongodb-aggregate.

mongodb connector

The connector opens the client on start, pings the deployment (see Verifying at startup), and disconnects on stop.

env:
  - name: MONGODB_PASSWORD
    required: true

connectors:
  - name: orders-db
    type: mongodb
    settings:
      uri: mongodb://app@db.internal:27017     # no secret here
      password: ${MONGODB_PASSWORD}            # the only secret
      database: orders

Keeping only the password secret

The password setting is merged into the URI at startup, so the connection string (host, user, replica set, TLS options) can live in plain config and only the credential has to be a secret. This is the same arrangement the database connector uses for its DSN.

A password already embedded in the URI is overridden by the setting, with a warning naming which value the connector used. A password with no username to attach it to is a startup error.

Verifying at startup

By default the connector pings the deployment when it starts, so a malformed URI, a wrong credential, or an unreachable server fails there rather than on the first message.

The driver's own documentation names the other side: pinging at startup costs resilience, because a process starting while the server is temporarily unavailable or failing over will error instead of coming up. A deployment that would rather ride out a failover can set verifyOnStart: false.

The client is opened either way, only the ping is skipped. Turning it off also lets a config be built with no server to reach, which is what a flow test needs: the blocks are mocked, but the connector behind them still starts.

Pool size and workers

maxPoolSize defaults to 32, the worker count a flow gets by default (max(8, GOMAXPROCS × 4), i.e. 32 on an eight-core host), so every worker of one flow can hold a checked-out connection instead of queueing on the pool. The driver's own default is 100. Raise it when several flows share one connector; lower it when many runtime replicas point at a small server. Read it alongside any flow that sets its own workers:.

A setting always wins over the same option spelled in the connection string. A default only fills a gap neither of them filled: a URI carrying ?maxPoolSize=50 keeps its 50.

BSON and JSON

BSON has types JSON does not: ObjectId, dates, Decimal128, and true 64-bit integers. Rather than flatten them, this connector renders documents as MongoDB Extended JSON v2 in both directions, so a document you read can be handed back as a filter.

In relaxed mode (the default), ordinary values stay plain JSON and only the types that need it are wrapped:

{
  "_id":     { "$oid": "66b0f1a2c3d4e5f60718293a" },
  "created": { "$date": "2026-08-04T10:00:00Z" },
  "sku":     "A1",
  "qty":     2
}

That _id goes straight back into a filter as the shape it came out as:

filter: '{"_id": {"$oid": body._id["$oid"]}}'

In canonical mode every value is typed ({"$numberLong": "1500"}, {"$numberInt": "2"}), at the cost of turning every number into an object you have to reach into from CEL. Set extendedJson: canonical when documents carry Decimal128 money or integer IDs above 2⁵³; leave it relaxed otherwise.

Input is parsed leniently in both modes, so a filter written in either form is accepted regardless of how results are rendered.

Two things to know when writing filters and documents as expressions:

• CEL numbers are doubles, so an integer literal above 2⁵³ cannot be written directly. Write it as the string {"$numberLong": "9007199254740993"}, which the driver parses exactly.
• A date written unquoted in YAML is parsed as a timestamp and reaches the block as an RFC 3339 string. Quote it, or write it as {"$date": "2026-08-04T10:00:00Z"}.

Addressing, and results

Every block names its collection as a CEL expression, and may name a database the same way, so a flow serving several tenants can route body.tenantId to a different collection per message. An unset database uses the connector's.

Every block follows one rule for its result: it becomes the message body, unless you name a resultVar, in which case it goes there and the body is left alone.

Result size, and paging

A read materialises its whole result: documents come off the cursor into a list, that list is rendered as extended JSON, and the JSON becomes the body. Nothing streams, so a read costs memory in proportion to what the query matched. Both read blocks return everything matched when no limit is set, so set one on any collection that can grow.

limit is a ceiling, not paging, and there is no paging built into these blocks. Cursor paging has nowhere to live, since a MongoDB cursor is server-side state with an owning connection and a flow is neither a session nor pinned to one. Offset paging is only awkward: limit and skip are plain integers that configure the block, so a flow answering GET /orders?page=2 cannot express that skip through mongodb-find today.

Two shapes work now. mongodb-aggregate's pipeline is one expression, so $skip and $limit are computed per message inside it, which is why that block's limit setting is a safety ceiling rather than the paging mechanism:

  - type: mongodb-aggregate
    settings:
      connector: orders-db
      collection: '"orders"'
      pipeline: |
        [
          {"$match": {"customer": vars.customerId}},
          {"$sort": [{"_id": 1}]},
          {"$skip": ("page" in vars.query ? int(vars.query.page) : 0) * 50},
          {"$limit": 50}
        ]

Range paging works with mongodb-find as it stands, because filter already is an expression: carry the last _id of a page into the next request and the skip becomes a match.

  - type: mongodb-find
    settings:
      connector: orders-db
      collection: '"orders"'
      filter: '"after" in vars.query ? {"_id": {"$gt": {"$oid": vars.query.after}}} : {}'
      sort: '[{"_id": 1}]'
      limit: 50

Prefer range paging on a large collection: $skip walks the documents it skips, so page 500 costs what pages 1 to 500 cost together, while a range match seeks straight to the key. Both need a stable sort, or consecutive pages overlap and drop documents.

mongodb-find block

Read documents from a collection. Without single, the result is a list, [] when nothing matched and never null, so a foreach over the body gets zero iterations rather than a type error.

process:
  - type: mongodb-find
    settings:
      connector: orders-db
      collection: '"orders"'
      filter: '{"status": "open", "customer": body.customerId}'
      sort: '[{"created": -1}]'
      limit: 20

With single: true the block returns the first matching document rather than a list, and null when nothing matched, which is an ordinary answer a flow branches on rather than an error. This mirrors the sql block's single.

  - type: mongodb-find
    settings:
      connector: orders-db
      collection: '"orders"'
      filter: '{"_id": {"$oid": body.orderId}}'
      single: true

Sorting by more than one key

MongoDB applies sort keys in the order they are written, and a CEL object cannot carry that order: CEL maps are unordered, and the JSON on the way to BSON sorts keys alphabetically. A multi-key object is refused at runtime rather than silently reordered, so write a list of single-key objects instead:

sort: '[{"created": -1}, {"sku": 1}]'     # created descending, then sku ascending
sort: '{"created": -1}'                   # fine: one key has no order to lose
sort: '{"created": -1, "sku": 1}'         # error: which key sorts first is unknowable

mongodb-insert block

Write documents into a collection. document decides how many by what it evaluates to: an object writes one, a list of objects writes them all. There is no mode setting. Because a message body is already a document, storing an inbound payload is document: body.

process:
  - type: mongodb-insert
    settings:
      connector: orders-db
      collection: '"orders"'
      document: body

The result is {insertedCount, insertedIds}. insertedIds is always a list, even for one document. The ids are rendered as extended JSON, so an _id the server generated comes back in the shape a filter takes:

  - type: mongodb-insert
    settings:
      connector: orders-db
      collection: '"orders"'
      document: body
      resultVar: written
  - type: mongodb-find
    settings:
      connector: orders-db
      collection: '"orders"'
      filter: '{"_id": vars.written.insertedIds[0]}'    # no conversion needed
      single: true

An empty list inserts nothing and returns {insertedCount: 0, insertedIds: []} rather than failing.

A batch that fails may still have written part of itself: always with unordered: true, and up to the failing document without it. The flow fails either way, and the error says how many landed.

mongodb-update block

Modify documents in a collection, and optionally insert one when the filter matches nothing (upsert).

Set either update or replacement, never both. An update applies operators to the document that is there; a replacement discards it and puts a new one in its place, keeping only the _id. Configuring both, or neither, is a startup error.

process:
  - type: mongodb-update
    settings:
      connector: orders-db
      collection: '"orders"'
      filter: '{"sku": body.sku}'
      update: '{"$set": {"status": "shipped", "shippedAt": now}}'
      upsert: true

An update must use operators. A plain object is refused, naming the fix, where MongoDB's own error for this only mentions the first field name:

update: '{"$set": {"qty": body.qty}}'   # correct
update: '{"qty": body.qty}'             # error: wrap it in $set, or use replacement

An update may also be an aggregation pipeline, a list of stages, which is how one field is computed from others in the same document:

update: '[{"$set": {"total": {"$multiply": ["$qty", "$price"]}}}]'

many: true is not available with replacement, matching the driver.

The result is {matchedCount, modifiedCount, upsertedCount, upsertedId}. matchedCount and modifiedCount differ when an update sets a field to the value it already held, so both are reported. upsertedId is present and null when no upsert happened, rather than absent, and carries the same {"$oid": ...} shape a filter takes.

mongodb-delete block

Remove documents from a collection. The result is {deletedCount}.

process:
  - type: mongodb-delete
    settings:
      connector: orders-db
      collection: '"orders"'
      filter: '{"_id": {"$oid": body.orderId}}'

Deleting everything is opt-in

A filter that evaluates to {} matches every document, so many: true with an empty filter would empty the collection. That is refused unless deleteAll: true says it was meant.

The guard exists because the filter is an expression, and an empty one is rarely written on purpose: it is what {"customer": body.customerId} collapses to when the message doesn't carry customerId.

# refused: nothing here says emptying the collection was the intent
filter: '{}'
many: true

# allowed: the flow said so
filter: '{}'
many: true
deleteAll: true

Deleting a single document is not guarded: the blast radius is one document. filter is required even with deleteAll.

mongodb-aggregate block

Run an aggregation pipeline over a collection. The resulting documents become the body, the same way a find's do. The stages you would type into mongosh are the stages you write here, with body and vars reachable inside them.

process:
  - type: mongodb-aggregate
    settings:
      connector: orders-db
      collection: '"orders"'
      pipeline: |
        [
          {"$match": {"customer": body.customerId}},
          {"$group": {"_id": "$sku", "total": {"$sum": "$qty"}}},
          {"$sort": {"total": -1}}
        ]

pipeline must evaluate to a list, even for one stage. A bare stage object is refused against the field, rather than becoming a driver error further down.

limit is appended to the pipeline as a final $limit, so the server stops producing documents rather than the block reading and discarding them. A pipeline that already ends in its own $limit is unaffected: the smaller of the two wins. A pipeline ending in $out or $merge cannot take one, since those stages must themselves be last and return no documents to bound, so the combination is refused.

$sort inside a pipeline

A $sort stage has the same key-order problem as the mongodb-find sort setting, and the same answer: sorting by more than one key needs the list form, and a multi-key object is refused.

pipeline: '[{"$sort": [{"total": -1}, {"sku": 1}]}]'    # order preserved
pipeline: '[{"$sort": {"total": -1}}]'                  # fine: one key
pipeline: '[{"$sort": {"total": -1, "sku": 1}}]'        # error: order unknowable

The list form is an Octo spelling (MongoDB's own $sort takes only an object) and it is rewritten into an ordered document before the pipeline is sent. It applies to top-level $sort stages; a $sort nested inside $facet or a $lookup sub-pipeline is passed through untouched and carries the caveat.