MongoDB Connector Guide

The MongoDB connector lets Tealfabric workflows read and write document data in MongoDB collections for operational automation and system synchronization. This guide covers practical configuration, common CRUD patterns, and production reliability considerations.

When to use this connector

Use this connector when workflows need to persist event data, query operational records, or synchronize document-based state across systems. It is especially useful for automation pipelines that depend on flexible schemas and high-throughput reads. Defining stable collection and key patterns early helps avoid inconsistent query behavior later.

Prerequisites

Before configuring the integration, ensure your MongoDB user has the required roles for the target database and collections. Confirm network connectivity from your runtime to MongoDB, including allowlists and TLS requirements where applicable. Validating this upfront prevents common connection and permission failures.

Configuration reference

Connector settings are stored in integration configuration and reused at runtime. Keep secrets secure and rotate credentials according to policy.

• connection_string (required): MongoDB URI, for example mongodb://... or mongodb+srv://....
• database_name (required): Target database for operations.
• timeout_seconds (optional): Connection and server-selection timeout (default 30).

connection_string, database_name, and timeout_seconds are configuration parameters — do not pass them in operation callData.

Insert, update, and delete documents with send

The send operation supports insert, update, and delete sub-operations on a collection. Use collection (alias table), operation (default insert), and explicit filter/update fields for predictable writes.

const baseUrl = "https://api.example.com/api/v1";
const tenantId = "<TENANT_ID>";
const apiKey = "<API_KEY>";

async function insertOrderDocument(integrationId: string) {
  const response = await fetch(
    `${baseUrl}/integrations/${encodeURIComponent(integrationId)}/execute`,
    {
      method: "POST",
      headers: {
        "X-API-Key": apiKey,
        "X-Tenant-ID": tenantId,
        "Content-Type": "application/json",
      },
      body: JSON.stringify({
        operation: "send",
        collection: "orders",
        documents: {
          order_id: "A-100",
          amount: 49.0,
          status: "paid",
          source: "processflow",
        },
      }),
    }
  );

  if (!response.ok) throw new Error(`Request failed: ${response.status}`);
  const payload = await response.json();
  return payload.data;
}

curl -X POST "https://api.example.com/api/v1/integrations/<ENTITY_ID>/execute" \
  -H "X-API-Key: <API_KEY>" \
  -H "X-Tenant-ID: <TENANT_ID>" \
  -H "Content-Type: application/json" \
  -d '{
    "operation": "send",
    "collection": "orders",
    "documents": {
      "order_id": "A-100",
      "amount": 49.0,
      "status": "paid"
    }
  }'

{
  "success": true,
  "data": {
    "message_count": 1,
    "affected_count": 1
  }
}

For update, provide filter (alias query) and update (fields are wrapped in $set like the legacy connector). For delete, provide a non-empty filter.

Query documents with receive

Use receive for find() reads with optional filter (alias query) and driver options such as limit, sort, and projection.

const baseUrl = "https://api.example.com/api/v1";
const tenantId = "<TENANT_ID>";
const apiKey = "<API_KEY>";

async function fetchPaidOrders(integrationId: string) {
  const response = await fetch(
    `${baseUrl}/integrations/${encodeURIComponent(integrationId)}/execute`,
    {
      method: "POST",
      headers: {
        "X-API-Key": apiKey,
        "X-Tenant-ID": tenantId,
        "Content-Type": "application/json",
      },
      body: JSON.stringify({
        operation: "receive",
        collection: "orders",
        filter: { status: "paid" },
        options: {
          limit: 20,
          sort: { order_id: -1 },
          projection: { order_id: 1, amount: 1, status: 1 },
        },
      }),
    }
  );

  if (!response.ok) throw new Error(`Request failed: ${response.status}`);
  const payload = await response.json();
  return payload.data?.data ?? [];
}

{
  "success": true,
  "data": {
    "message_count": 2,
    "data": [
      { "_id": "...", "order_id": "A-100", "amount": 49.0, "status": "paid" }
    ],
    "total_size": 2
  }
}

Bidirectional sync and sequential batch

sync runs optional outbound (send) and inbound (receive) legs. Empty outbound/inbound payloads are skipped (PHP empty() semantics). Each leg result embeds legacy-shaped fields including inner success.

batch accepts operations (alias items) with per-entry operation/type (send or receive) and nested data. Failed entries are recorded without aborting the whole batch; overall success is true when at least one entry succeeds.

Validate connectivity with test

Run test before go-live and after credential changes. Only test performs full configuration validation (Configuration validation failed when connection_string or database_name is missing). Other operations defer to MongoDB connection/driver errors.

curl -X POST "https://api.example.com/api/v1/integrations/<ENTITY_ID>/execute" \
  -H "X-API-Key: <API_KEY>" \
  -H "X-Tenant-ID: <TENANT_ID>" \
  -H "Content-Type: application/json" \
  -d '{
    "operation": "test"
  }'

{
  "success": true,
  "data": {
    "message": "MongoDB connection test successful",
    "details": {
      "database": "mydatabase",
      "server": "mongo.example.com:27017"
    }
  }
}

Additional references

For query syntax and deployment best practices, review MongoDB documentation and the MongoDB query operator reference. Keeping filter contracts and index assumptions documented alongside workflows improves long-term reliability.