MongoDB integration

Read, write, and aggregate documents in a client's MongoDB database, and start workflows when documents, collections, or fields change.

What it does

The MongoDB integration lets your agency work with a client's MongoDB or MongoDB Atlas database directly from a workflow. Actions find, count, aggregate, insert, update, replace, and delete documents, run bulk writes, and manage Atlas Search indexes. Triggers start a workflow when a document changes. On Atlas, changes arrive instantly through an Atlas Database Trigger. On any MongoDB, TaskJuice can instead check on an interval for new documents, updated documents, new collections, new databases, or new fields.

Connection details are encrypted at rest. Filters and updates are sent to the database as queries and never run as code. Results are capped, so a large read cannot overwhelm a run.

Connect a MongoDB database

Add a MongoDB connection on any MongoDB node.
Paste a connection string, for example mongodb+srv://user:password@cluster.example.net/appdb. The database named in the string is the one actions and triggers use. Connection-string mode preserves SRV records and replica-set options.
Or fill in the host, port (27017 by default), user, password, and database separately.
Save the connection. The password and connection string are stored encrypted and are never shown again or written to run output.

Use a least-privilege database user. readWrite on the database covers every action except the Atlas Search index actions, which also need the Atlas Search index privileges.

MongoDB Atlas network access. TaskJuice reaches the database from more than one network. Workflow steps in a published run connect from a fixed set of outbound addresses. The interval triggers and the editor's Test step connect from cloud infrastructure that has no fixed IP addresses, so a step that fails its editor test with a connection error can still work in a published run. In Atlas, open Network Access and allow access from anywhere (0.0.0.0/0). Otherwise those connections time out. With the allowlist open, the database user's password is what protects the cluster, so make it long and random. Self-hosted databases must be reachable on a public host over TLS. Private and internal addresses are blocked.

Triggers

New Document (Atlas, Instant), Updated Document (Atlas, Instant), and Deleted Document (Atlas, Instant) fire the moment a document is inserted, updated or replaced, or deleted in an Atlas collection. They need no polling and no timestamp field. Setup is below.
New Document checks a collection on an interval and fires for documents inserted since the last check, in _id order. Set Sort Field to a creation timestamp such as createdAt when your _id values are not ObjectIds.
New or Updated Document checks on an interval and fires for documents whose timestamp field, for example updatedAt, moved forward since the last check. Your application must set that field on every insert and update.
New Collection fires when a collection or view is created in the connection's database.
New Database fires when a database the connection's user can see appears on the cluster.
New Field fires when a top-level field name appears in the newest documents of a collection, which helps you spot schema drift.

The interval triggers record a starting point on their first check and fire nothing, so publishing a workflow never replays existing data. Each later check delivers everything new since the previous one as a single items list. Add a Loop node to handle each item on its own. A backlog larger than Max Per Cycle carries over to the next check, and nothing is skipped.

Set up the Atlas instant triggers

Atlas Database Triggers watch a collection's change stream and call a function for every change. The function below forwards each change to TaskJuice, which starts every published workflow that uses the matching instant trigger. You set this up once per collection; each workflow only picks the trigger it needs.

Add one of the Atlas instant triggers to a workflow and copy the MongoDB ingress URL it shows. The URL is private to your workspace, so treat it like a password.
In Atlas, open your project, then Triggers, and select Add Trigger. Choose Database, then pick the cluster, database, and collection.
Under Operation Type, tick Insert, Update, Replace, and Delete, or only the ones you use.
Turn on Full Document so updates include the whole document. Optionally turn on Document Preimage to also receive the document as it was before the change.
Under Event Type, choose Function, and paste:

exports = async function (changeEvent) {
  const url = context.values.get('automationIngressUrl');
  // Flatten Extended JSON: ids and decimals become strings, numbers become
  // numbers, dates become ISO-8601 strings. Atlas emits canonical Extended
  // JSON ({ "$numberInt": "1" }), so every wrapper is unwrapped here.
  const flatten = (key, value) => {
    if (value && typeof value === 'object' && !Array.isArray(value)) {
      const keys = Object.keys(value);
      if (keys.length === 1) {
        const type = keys[0];
        const inner = value[type];
        if (type === '$oid' || type === '$numberDecimal') return String(inner);
        if (type === '$numberInt' || type === '$numberDouble') return Number(inner);
        if (type === '$numberLong') {
          const n = Number(inner);
          return Number.isSafeInteger(n) ? n : String(inner);
        }
        if (type === '$date') {
          // Inner values are flattened first, so a canonical date is a number here.
          return typeof inner === 'number' ? new Date(inner).toISOString() : inner;
        }
      }
    }
    return value;
  };
  const plain = (value) =>
    value === undefined || value === null
      ? null
      : JSON.parse(EJSON.stringify(value, { relaxed: true }), flatten);
  const update = changeEvent.updateDescription || {};
  const body = {
    operationType: changeEvent.operationType,
    database: changeEvent.ns.db,
    collection: changeEvent.ns.coll,
    documentId: plain(changeEvent.documentKey._id),
    fullDocument: plain(changeEvent.fullDocument),
    fullDocumentBeforeChange: plain(changeEvent.fullDocumentBeforeChange),
    updatedFields: plain(update.updatedFields) || {},
    removedFields: update.removedFields || [],
    changedAt: plain(changeEvent.wallTime),
    eventId: changeEvent._id._data,
  };
  const response = await context.http.post({
    url,
    headers: { 'Content-Type': ['application/json'] },
    body: JSON.stringify(body),
  });
  if (response.statusCode >= 300) {
    throw new Error('Workflow ingress returned HTTP ' + response.statusCode);
  }
};

In Atlas Values, add a secret named automationIngressUrlSecret holding the ingress URL, then a value named automationIngressUrl linked to that secret.
Save the trigger. The next matching change starts your workflows.

One change is one run. The payload carries operationType, database, collection, documentId, fullDocument, fullDocumentBeforeChange, updatedFields, removedFields, changedAt, and eventId. Inserts reach New Document (Atlas, Instant), updates and replaces reach Updated Document (Atlas, Instant), and deletes reach Deleted Document (Atlas, Instant). All your Atlas triggers post to the same ingress URL, so a workflow receives the changes of every collection you forward. When you forward more than one collection, add a Filter node on collection right after the trigger.

Actions

Find Documents returns documents that match a filter, with optional projection, sort, and limit.
Find Document returns the first matching document, for example by _id, or found: false when nothing matches.
Count Documents counts the documents that match a filter.
Aggregate Documents runs an aggregation pipeline ($match, $group, $lookup, $sort, …) and returns its output. Pipelines that write with $out or $merge are rejected.
Insert Document inserts one document and returns its id.
Insert Documents inserts up to 1,000 documents in one call and returns their ids in order.
Update Documents applies an update such as { "$set": { … } } to every matching document.
Update Document updates the first matching document and can insert it when it is missing.
Upsert Document updates a matching document or inserts one when none matches.
Find and Update Document updates the first matching document and returns it, either after or before the change.
Find and Replace Document replaces the first matching document (keeping its _id) and returns it, either after or before the change.
Delete Documents deletes every document that matches a filter.
Delete Document deletes the first document that matches a filter.
Bulk Write runs up to 1,000 insertOne, updateOne, updateMany, replaceOne, deleteOne, and deleteMany operations in one call.
List Collections lists the collections and views in the connection's database.
List Databases lists the databases the connection's user can access.
List Search Indexes, Create Search Index, Update Search Index, and Drop Search Index manage Atlas Search and Atlas Vector Search indexes. They work on MongoDB Atlas only.

Write filters and updates

Filters, updates, documents, and pipelines are JSON objects. Special MongoDB types use Extended JSON:

filter: { "status": "active", "createdAt": { "$gte": { "$date": "2026-01-01T00:00:00Z" } } }
update: { "$set": { "status": "archived" } }
by id:  { "_id": { "$oid": "65f0c0ffee00000000000abc" } }

Results come back as plain JSON. ObjectIds become 24-character hex strings, and dates become ISO-8601 strings. A filter on _id with a 24-character hex string matches both the ObjectId and the literal string. So an id from a previous step, such as { "_id": "65f0c0ffee00000000000abc" }, finds its document without any conversion.

The $where, $function, and $accumulator operators run server-side JavaScript, so they are rejected anywhere in a filter, update, or pipeline.

Known limitations

Reads return at most 1,000 documents and about 5 MB per call. When a read is clipped, the result sets truncated to true. Use a selective filter and a limit.
Insert Documents and Bulk Write accept at most 1,000 entries per call.
An operation that runs longer than 30 seconds is cancelled by the database.
Each action and each interval check opens its own short connection and closes it when done.
New or Updated Document relies on your application's timestamp field. A write that sets an older timestamp than one already delivered is not picked up. Use the Atlas trigger for exact change capture.
New Collection, New Database, and New Field track at most 2,000 names.
New Field inspects top-level field names only, in the newest documents by _id.
The instant triggers require MongoDB Atlas, because self-hosted MongoDB has no Database Triggers. Use the interval triggers there.
Instant triggers cannot be scoped to one collection in the trigger itself; filter on collection in the workflow.
Atlas Search index actions require MongoDB Atlas.