# Connection Configuration Reference

**Official examples repo**: https://github.com/mongodb/ASP_example — check quickstarts, example processors, and Terraform examples. Start with quickstarts.

## Connection Capabilities — Source/Sink Reference

Know what each connection type can do before creating pipelines:

| Connection Type | As Source ($source) | As Sink ($merge / $emit) | Mid-Pipeline | Notes |
|-----------------|---------------------|--------------------------|--------------|-------|
| **Cluster** | ✅ Change streams | ✅ $merge to collections | ✅ $lookup | Change streams monitor insert/update/delete/replace operations |
| **Kafka** | ✅ Topic consumer | ✅ $emit to topics | ❌ | Source MUST include `topic` field |
| **Sample Stream** | ✅ Sample data | ❌ Not valid | ❌ | Testing/demo only |
| **S3** | ❌ Not valid | ✅ $emit to buckets | ❌ | Sink only - use `path`, `format`, `compression` |
| **Https** | ❌ Not valid | ✅ $https as sink | ✅ $https enrichment | Can be used mid-pipeline for enrichment OR as final sink stage |
| **AWSLambda** | ❌ Not valid | ✅ $externalFunction (async only) | ✅ $externalFunction (sync or async) | **Sink:** `execution: "async"` required. **Mid-pipeline:** `execution: "sync"` or `"async"` |
| **AWS Kinesis** | ✅ Stream consumer | ✅ $emit to streams | ❌ | Similar to Kafka pattern |
| **SchemaRegistry** | ❌ Not valid | ❌ Not valid | ✅ Schema resolution | **Metadata only** - used by Kafka connections for Avro schemas |

**Common connection usage mistakes to avoid:**
- ❌ Using HTTPS connections as `$source` → HTTPS is for enrichment or sink only
- ❌ Using `$externalFunction` as sink with `execution: "sync"` → Must use `execution: "async"` for sink stage
- ❌ Forgetting change streams exist → Atlas Cluster is a powerful source, not just a sink
- ❌ Using `$merge` with Kafka → Use `$emit` for Kafka sinks

**$externalFunction execution modes:**
- **Mid-pipeline:** Can use `execution: "sync"` (blocks until Lambda returns) or `execution: "async"` (non-blocking)
- **Final sink stage:** MUST use `execution: "async"` only

## Connection Naming Best Practices

**CRITICAL**: Connection names should clearly indicate their actual targets to avoid confusion and prevent writing data to wrong destinations.

### Good Naming Patterns

**Match the actual target name:**
- Cluster connection to "ClusterRestoreTest" → name it `cluster-restore-test` or `ClusterRestoreTest`
- Cluster connection to "AtlasCluster" → name it `atlas-cluster` or `AtlasCluster`

**Use descriptive names with context:**
- `prod-kafka-orders` (indicates environment + service + purpose)
- `dev-atlas-main` (indicates environment + service + designation)
- `staging-s3-exports` (indicates environment + service + purpose)

### Bad Naming Patterns (AVOID)

❌ **Generic names that don't match targets:**
- Connection "atlascluster" pointing to "ClusterRestoreTest" ← CONFUSING!
- Connection "kafka" pointing to multiple different topics ← NOT SPECIFIC!

❌ **Reusing names across workspaces without context:**
- "myconnection" in workspace A and workspace B with different targets

❌ **Names that don't indicate connection type:**
- "connection1", "test", "temp" ← NO CONTEXT!

### Verification Workflow

**Before creating processors**, always inspect your connections to verify they point where you expect:
```
1. atlas-streams-discover → action: "list-connections"
2. atlas-streams-discover → action: "inspect-connection" for each
3. Verify connection name matches actual target (clusterName, bootstrapServers, url, etc.)
4. If mismatch exists, consider renaming or warn the user
```

See [development-workflow.md](development-workflow.md) "Pre-Deployment Connection Validation" section for the complete validation procedure.

## Important Notes
- HTTPS connections are for `$https` enrichment ONLY — they are NOT valid as `$source` data sources
- Store API authentication in connection settings, never hardcode in processor pipelines
- AWS connections (S3, Kinesis, Lambda) require IAM role ARN registered via Atlas Cloud Provider Access first
- Supported `connectionType` values: `Kafka`, `Cluster`, `S3`, `Https`, `AWSKinesisDataStreams`, `AWSLambda`, `SchemaRegistry`, `Sample`

## AWS Cloud Provider Access Prerequisites

**For S3, Kinesis, and Lambda connections:**

AWS connections (S3, Kinesis, Lambda) require that the IAM role ARN be **registered in the Atlas project via Cloud Provider Access** before creating the connection. This is a prerequisite — the connection creation will fail without it.

**Always mention this prerequisite** in your response when the user wants to create AWS connections, even if the user says connections already exist. Confirm with language like:
- "IAM role ARNs are registered via Atlas Cloud Provider Access"
- "Ensure IAM role ARNs are registered via Atlas Cloud Provider Access before creating connections"

**Security best practice:** Use a dedicated IAM role per processor (or group of related processors) with least-privilege permissions scoped only to the specific S3 buckets, Kinesis streams, or Lambda functions that processor needs. Avoid sharing broad-access roles across unrelated processors.

## Region Mapping Reference

The `region` field for workspace creation uses Atlas-specific names that differ by cloud provider. Using the wrong format returns a cryptic `dataProcessRegion` error.

| Provider | Cloud Region | Streams `region` Value |
|----------|-------------|----------------------|
| **AWS** | us-east-1 | `VIRGINIA_USA` |
| **AWS** | us-east-2 | `OHIO_USA` |
| **AWS** | us-west-2 | `OREGON_USA` |
| **AWS** | ca-central-1 | `MONTREAL_CAN` |
| **AWS** | sa-east-1 | `SAOPAULO_BRA` |
| **AWS** | eu-west-1 | `DUBLIN_IRL` |
| **AWS** | ap-southeast-1 | `SINGAPORE_SGP` |
| **AWS** | ap-south-1 | `MUMBAI_IND` |
| **AWS** | ap-northeast-1 | `TOKYO_JPN` |
| **GCP** | us-central1 | `US_CENTRAL1` |
| **GCP** | europe-west1 | `EUROPE_WEST1` |
| **GCP** | us-east4 | `US_EAST4` |
| **Azure** | eastus | `eastus` |
| **Azure** | eastus2 | `eastus2` |
| **Azure** | westus | `westus` |
| **Azure** | westeurope | `westeurope` |

This is a partial list. If unsure, inspect an existing workspace with `atlas-streams-discover` → `inspect-workspace` and check `dataProcessRegion.region`.

## MCP Tool Behaviors for Connections

**Elicitation:** When required fields are missing, the build tool auto-prompts for them via an interactive form (MCP elicitation protocol). Do NOT manually ask the user for passwords or bootstrap servers — let the tool collect them.

**Auto-normalization:**
- `bootstrapServers` passed as array → auto-converted to comma-separated string
- `schemaRegistryUrls` passed as string → auto-wrapped in array
- Cluster `dbRoleToExecute` → auto-defaults to `{role: "readWriteAnyDatabase", type: "BUILT_IN"}` if omitted

## connectionConfig by type

### Kafka
```json
{
  "bootstrapServers": "broker1:9092,broker2:9092",
  "authentication": {
    "mechanism": "SCRAM-256",
    "username": "my-user",
    "password": "my-password"
  },
  "security": {
    "protocol": "SASL_SSL"
  }
}
```
**Important:** `bootstrapServers` is a **comma-separated string**, not an array.

All fields above are required. The tool will prompt the user for username/password via elicitation if not provided.

Authentication mechanisms: `PLAIN`, `SCRAM-256`, `SCRAM-512`, `OAUTHBEARER`
Security protocols: `SASL_SSL`, `SASL_PLAINTEXT`, `SSL`

For Confluent Cloud, use `mechanism: "PLAIN"` with your API key as `username` and API secret as `password`.

Kafka supports both **PrivateLink** and **VPC Peering** for private networking. See the [PrivateLink Reference](#privatelink-reference-all-vendors) section below for all supported vendors and providers.

**VPC Peering:**
- Supported for outbound connections to Kafka brokers in your own VPC
- Requires `SASL_SSL` security protocol
- Use `atlas-streams-manage` with `accept-peering` action to complete the peering setup
- Requires AWS account ID, VPC ID, and region information

**Important: Networking cannot be modified after connection creation.** To add or change PrivateLink/VPC peering on an existing Kafka connection, you must delete it and recreate it with the networking config.

Use `atlas-streams-discover` → `action: "get-networking"` to list available PrivateLink endpoints and VPC peering connections.

### Cluster (Atlas)
```json
{
  "clusterName": "my-atlas-cluster",
  "dbRoleToExecute": {
    "role": "readWriteAnyDatabase",
    "type": "BUILT_IN"
  }
}
```
`clusterName` is **required** — must be a cluster in the same project (use `atlas-list-clusters` to verify).

`dbRoleToExecute` defaults to `{role: "readWriteAnyDatabase", type: "BUILT_IN"}` if not provided.

Optional: `clusterGroupId` (if cluster is in a different project — requires cross-project access to be enabled at the org level).

### S3
```json
{
  "aws": {
    "roleArn": "arn:aws:iam::123456789:role/streams-s3-role",
    "testBucket": "my-test-bucket"
  }
}
```
**Prerequisite:** The IAM role ARN must be registered in the Atlas project via Cloud Provider Access before creating the connection.

Required IAM policy permissions: `s3:ListBucket`, `s3:GetObject`, `s3:PutObject`.

### Https
```json
{
  "url": "https://api.example.com/webhook",
  "headers": {
    "Authorization": "Bearer token123"
  }
}
```
**IMPORTANT:** HTTPS connections are for `$https` enrichment stages ONLY. They are NOT valid data sources — do not use them in `$source`.

Store all API authentication in the connection config headers, not in the processor pipeline.

#### HTTPS Auth Patterns

**API Key:**
```json
{"url": "https://api.example.com", "headers": {"X-API-Key": "your-api-key"}}
```

**Bearer Token:**
```json
{"url": "https://api.example.com", "headers": {"Authorization": "Bearer your-token"}}
```

**Basic Auth:**
```json
{"url": "https://api.example.com", "headers": {"Authorization": "Basic base64-encoded-credentials"}}
```

**OAuth 2.0 (pre-obtained token):**
```json
{"url": "https://api.example.com", "headers": {"Authorization": "Bearer oauth-access-token"}}
```

### AWSKinesisDataStreams
```json
{
  "aws": {
    "roleArn": "arn:aws:iam::123456789:role/streams-kinesis-role"
  }
}
```
**Prerequisite:** The IAM role ARN must be registered in the Atlas project via Cloud Provider Access before creating the connection.

Required IAM policy permissions: `kinesis:ListShards`, `kinesis:SubscribeToShard`, `kinesis:PutRecords`, `kinesis:DescribeStreamSummary`.

### AWSLambda
```json
{
  "aws": {
    "roleArn": "arn:aws:iam::123456789:role/streams-lambda-role"
  }
}
```
**Prerequisite:** The IAM role ARN must be registered in the Atlas project via Cloud Provider Access before creating the connection.

### SchemaRegistry
```json
{
  "connectionType": "SchemaRegistry",
  "connectionConfig": {
    "schemaRegistryUrls": ["https://schema-registry.example.com"],
    "schemaRegistryAuthentication": {
      "type": "USER_INFO",
      "username": "...",
      "password": "..."
    }
  }
}
```
- `connectionType` MUST be `"SchemaRegistry"` (not `"Kafka"` or `"Https"`)
- `schemaRegistryUrls` is an **array** (not a string). The tool auto-wraps a string into an array if needed.
- `schemaRegistryAuthentication.type`: `"USER_INFO"` (explicit credentials) or `"SASL_INHERIT"` (inherit from Kafka connection)
- Tool elicitation will collect sensitive fields (password) — don't ask the user for these directly

### Sample
No connectionConfig required. Provides built-in test data. Useful for development and testing without external infrastructure.

Available sample formats: `sample_stream_solar` (default, auto-created when `includeSampleData: true` on workspace), `samplestock`, `sampleweather`, `sampleiot`, `samplelog`, `samplecommerce`.

### PrivateLink Reference (All Vendors)

PrivateLink is supported for Kafka, S3, Kinesis, and Azure EventHub connections. Create a project-level PrivateLink first, then reference it in the connection's `networking.access` config.

**Step 1: Create project-level PrivateLink** via `atlas-streams-build` resource='privatelink':

| Provider | Vendor | Required privateLinkConfig fields |
|----------|--------|----------------------------------|
| AWS | CONFLUENT | provider, vendor, dnsDomain, dnsSubDomain (array, [] if none) |
| AWS | MSK | provider, vendor, arn |
| AWS | S3 | provider, vendor, region, serviceEndpointId (`com.amazonaws.<region>.s3`) |
| AWS | KINESIS | provider, vendor, region, serviceEndpointId |
| AZURE | EVENTHUB | provider, vendor, dnsDomain, serviceEndpointId |
| AZURE | CONFLUENT | provider, vendor, dnsDomain |
| GCP | CONFLUENT | provider, vendor, gcpServiceAttachmentUris |

**Step 2: Reference in connection networking config:**
```json
{
  "networking": {
    "access": {
      "type": "PRIVATE_LINK",
      "connectionId": "<PrivateLink _id from Step 1>"
    }
  }
}
```

Use `atlas-streams-discover` action='get-networking' to find the PrivateLink `_id`.

**Note:** Networking config cannot be modified after connection creation — delete and recreate to change.

# Development Workflow Reference

## Pipeline Stage Categories

Understanding stage categories helps compose valid pipelines. Stages must appear in this order:

| Category | Stages | Rules |
|----------|--------|-------|
| **Source** (1, required) | `$source` | Must be first. One per pipeline. |
| **Stateless Processing** | `$match`, `$project`, `$addFields`, `$unset`, `$unwind`, `$replaceRoot`, `$redact` | Can appear anywhere after source. No state or memory overhead. |
| **Enrichment** | `$lookup`, `$https` | I/O-bound. Use `parallelism` setting. Place `$https` after windows to batch. |
| **Stateful/Window** | `$tumblingWindow`, `$hoppingWindow`, `$sessionWindow` | Accumulates state in memory. Monitor `memoryUsageBytes`. |
| **Validation** | `$validate` | Schema enforcement. Use `validationAction: "dlq"` (not `"error"`). Place early to catch bad data. |
| **Custom Code** | `$function` | JavaScript UDFs. Requires SP30+. |
| **Output** (1+, required for deployed) | `$merge`, `$emit` | Must be last. Required for persistent processors. Sinkless = ephemeral only. |

**Key ordering principle:** Place `$match` as early as possible (reduces volume for all downstream stages). Place `$project` after `$match` (reduces document size). Place `$https` after windows (batches API calls).

## 5-Phase Development Lifecycle

### Phase 1: Project Setup

**Goal:** Workspace and connections ready.

1. Discover existing resources:
   - `atlas-streams-discover` → `list-workspaces` — see what already exists
   - If workspace exists, `inspect-workspace` to review config

2. Create workspace (if needed):
   - `atlas-streams-build` → `resource: "workspace"`
   - Choose region close to your data sources
   - Start with `tier: "SP10"` for development
   - `includeSampleData: true` (default) gives you `sample_stream_solar` for testing

3. Verify workspace:
   - `atlas-streams-discover` → `inspect-workspace` — confirm state and region

### Phase 2: Connection Development

**Goal:** All data sources and sinks connected and verified.

1. Identify required connections:
   - Source connections (Kafka, Cluster change streams, Kinesis, Sample)
   - Sink connections (Cluster for `$merge`, Kafka for `$emit`, S3, Kinesis)
   - Enrichment connections (Https for `$https`, Cluster for `$lookup`)

2. Create each connection:
   - `atlas-streams-build` → `resource: "connection"` for each
   - Let the tool elicit missing sensitive fields (passwords, bootstrap servers)
   - See [connection-configs.md](connection-configs.md) for type-specific schemas

3. Verify connections:
   - `atlas-streams-discover` → `list-connections` — confirm all created
   - `atlas-streams-discover` → `inspect-connection` for each — verify state and config

### Phase 3: Processor Development

**Goal:** Working processor with validated pipeline.

#### Pre-Deployment Connection Validation (MANDATORY)

**BEFORE creating any processor**, you MUST validate all connections referenced in your pipeline. This prevents silent failures and confusion about data destinations.

**Step 1: List all connections in workspace**
```
atlas-streams-discover → action: "list-connections", workspaceName: "<your-workspace>"
```
Verify all required connections exist.

**Step 2: Inspect EACH connection referenced in pipeline**

For EVERY `connectionName` in your pipeline (source, sink, enrichment), inspect it:
```
atlas-streams-discover → action: "inspect-connection",
                         workspaceName: "<your-workspace>",
                         resourceName: "<connection-name>"
```

**Verify for each connection:**
- [ ] Connection exists and state is READY
- [ ] Connection type matches intended usage:
  - Cluster: valid for `$source` (change streams), `$merge`, `$lookup`
  - Kafka: valid for `$source`, `$emit`
  - S3: valid for `$emit` only
  - Https: valid for `$https` enrichment or sink
  - Lambda: valid for `$externalFunction` only
- [ ] Connection name matches actual target (avoid confusion):
  - ⚠️ BAD: connection "atlascluster" → actual target "ClusterRestoreTest"
  - ✅ GOOD: connection "cluster-restore-test" → actual target "ClusterRestoreTest"
- [ ] For Cluster connections: verify the `clusterName` field points to the intended cluster

**Step 3: Present validation summary to user**

Always show the user what connections will be used:
```
"Before creating processor '<name>', I've verified your connections:
 - ✅ sample_stream_solar → Sample data (READY)
 - ⚠️ atlascluster → ClusterRestoreTest (READY)
      Warning: Connection name 'atlascluster' doesn't match actual cluster 'ClusterRestoreTest'
 - ✅ open-meteo-api → https://api.open-meteo.com/v1/... (READY)

Proceed with processor creation?"
```

**Step 4: Wait for user confirmation if warnings exist**

If any connection name doesn't match its target, ask the user to confirm before proceeding.

**Step 5: Only then create the processor**

This validation workflow prevents:
- Creating processors with non-existent connections (fails immediately)
- Writing data to unexpected clusters (e.g., "atlascluster" → "ClusterRestoreTest" instead of "AtlasCluster")
- Confusion when verifying output data later

#### Incremental Pipeline Development

Follow incremental pipeline development — test at each step:

**Step 1: Basic connectivity**
```json
[
  {"$source": {"connectionName": "my-source"}},
  {"$merge": {"into": {"connectionName": "my-sink", "db": "test", "coll": "step1"}}}
]
```
Create with `autoStart: true`. Verify documents flow. Stop processor.

**Step 2: Add filtering**
```json
[
  {"$source": {"connectionName": "my-source"}},
  {"$match": {"status": "active"}},
  {"$merge": {"into": {"connectionName": "my-sink", "db": "test", "coll": "step2"}}}
]
```
Modify pipeline (`stop` → `modify-processor` → `start`). Verify filtered output.

**Step 3: Add transformations**
```json
[
  {"$source": {"connectionName": "my-source"}},
  {"$match": {"status": "active"}},
  {"$addFields": {"processed_at": "$$NOW_NOT_VALID"}},
  {"$project": {"userId": 1, "amount": 1, "processed_at": 1}},
  {"$merge": {"into": {"connectionName": "my-sink", "db": "test", "coll": "step3"}}}
]
```
**Remember:** `$$NOW` is NOT valid in streaming. Use a field from the source document or omit.

**Step 4: Add windowing or enrichment** (if needed)

**Step 5: Add error handling**
- Configure DLQ: `{"dlq": {"connectionName": "my-sink", "db": "streams_dlq", "coll": "failed_docs"}}`
- Add `$ifNull` for optional enrichment fields
- Set `onError: "dlq"` on `$https` stages

### Phase 4: Testing & Validation

**Goal:** Processor verified working correctly.

1. Confirm processor state:
   - `atlas-streams-discover` → `inspect-processor` — state should be STARTED

2. Run diagnostics:
   - `atlas-streams-discover` → `diagnose-processor` — full health report

3. Verify data flow:
   - Use MongoDB `count` tool on output collection — documents arriving?
   - Use MongoDB `find` tool on output collection — data looks correct?
   - Use MongoDB `count` tool on DLQ collection — any errors?
   - If DLQ has documents, use MongoDB `find` tool to inspect failure reasons

4. Classify output volume:
   - See [output-diagnostics.md](output-diagnostics.md) for the full decision framework
   - Alert processors: low output is expected
   - Transformation processors: low output is a red flag

### Phase 5: Production Deployment

**Goal:** Processor running at appropriate tier with monitoring.

1. Right-size the tier:
   - See [sizing-and-parallelism.md](sizing-and-parallelism.md) for tier selection
   - Review `memoryUsageBytes` from diagnostics
   - Consider parallelism needs for `$merge`, `$lookup`, `$https`
   - Upgrade tier: `atlas-streams-manage` → `stop-processor`, then `start-processor` with `tier` override

2. Ensure DLQ is configured (mandatory for production)

3. Use descriptive processor names (e.g., `fraud-detector`, `order-enricher`, `iot-rollup`)

## Debugging Decision Trees

### Connection Failures
1. `atlas-streams-discover` → `inspect-connection` — check state
2. If Kafka: verify `bootstrapServers` is a comma-separated string (not array)
3. If Cluster: verify cluster exists in project (`atlas-list-clusters`)
4. If AWS (S3/Kinesis/Lambda): verify IAM role ARN is registered in Cloud Provider Access
5. If Https: verify URL is reachable and auth headers are in connection config

### Processor Startup Failures
1. `atlas-streams-discover` → `diagnose-processor` — check state and errors
2. If FAILED: read the error message in diagnostics
3. Common causes:
   - Invalid pipeline syntax (missing `$source`, missing sink)
   - `$$NOW`/`$$ROOT`/`$$CURRENT` used (not valid in streaming)
   - Kafka `$source` missing `topic` field
   - **Referenced connection doesn't exist** — validate with `list-connections` first
   - **Connection name doesn't match expected target** — inspect connection to verify actual cluster/resource
   - OOM — tier too small for pipeline complexity

### Processing Errors (Running but DLQ filling up)
1. Use MongoDB `find` tool on DLQ collection — inspect error messages
2. Common causes:
   - Schema mismatches in source data
   - `$https` enrichment failures (API down, auth expired)
   - Type errors in `$addFields` or `$project` expressions
3. Fix: `stop-processor` → `modify-processor` (fix pipeline) → `start-processor`

### Performance Issues (Running but slow)
1. `atlas-streams-discover` → `diagnose-processor` — check stats
2. Check `memoryUsageBytes` — if near 80% of tier RAM, upgrade tier
3. Check if `$match` is early in pipeline (reduces downstream volume)
4. Check if `$https` has `parallelism` setting (increase for I/O-bound enrichment)
5. Check if windows have `partitionIdleTimeout` (idle Kafka partitions block windows)
6. Consider upgrading tier or increasing stage parallelism

## Operational Monitoring Cadence

### Daily
- Check processor states via `atlas-streams-discover` → `list-processors`
- Verify DLQ collections aren't growing via MongoDB `count` tool
- Confirm output collections are receiving data

### Weekly
- Run `diagnose-processor` for each production processor
- Review `memoryUsageBytes` trends — approaching 80%?
- Check connection health across all connections

### Monthly
- Evaluate tier appropriateness — over-provisioned or under-provisioned?
- Review DLQ patterns — recurring errors that need pipeline fixes?
- Consider parallelism adjustments based on throughput trends

## Troubleshooting

| Symptom | Likely cause | Action |
|---------|-------------|--------|
| Processor FAILED on start | Invalid pipeline syntax, missing connection, `$$NOW` used | `diagnose-processor` → read error → fix pipeline |
| DLQ filling up | Schema mismatch, `$https` failures, type errors | `find` on DLQ → fix pipeline or connection |
| Zero output (transformation) | Connection issue, wrong topic, filter too strict | Check source health → verify connections → check `$match` |
| Zero output (alert) | Probably normal — no anomalies detected | Verify with known test event |
| Windows not closing | Idle Kafka partitions | Add `partitionIdleTimeout` to `$source` (e.g., `{"size": 30, "unit": "second"}`) |
| OOM / processor crash | Tier too small for window state | `diagnose-processor` → check `memoryUsageBytes` → upgrade tier |
| Slow throughput | Low parallelism on I/O stages | Increase `parallelism` on `$merge`/`$lookup`/`$https` |
| 404 on workspace | Doesn't exist or misspelled | `discover` → `list-workspaces` |
| 409 on create | Name already exists | Inspect existing resource or pick new name |
| 402 error on start | No billing configured | Do NOT retry. Add payment method in Atlas → Billing. Use `sp.process()` in mongosh as free alternative |
| "processor must be stopped" | Tried to modify running processor | `manage` → `stop-processor` first |
| bootstrapServers format | Passed as array instead of string | Use comma-separated string: `"broker1:9092,broker2:9092"` |
| "must choose at least one role" | Cluster connection without `dbRoleToExecute` | Defaults to `readWriteAnyDatabase` — or specify custom role |
| "No cluster named X" | Cluster doesn't exist in project | `atlas-list-clusters` to verify |
| IAM role ARN not found | ARN not registered in project | Register via Atlas → Cloud Provider Access |
| dataProcessRegion format | Wrong region format | See region table above. If unsure, inspect an existing workspace |
| Processor PROVISIONING for minutes | Restart cycle with exponential backoff | Wait for FAILED state, or stop → restart. Check logs for repeated error |
| Parallelism exceeded | Tier too small for requested parallelism | Start with higher tier (see `sizing-and-parallelism.md`) |
| Networking change needed | Networking is immutable after creation | Delete connection and recreate with new networking config |
| 401 / 403 on API call | Invalid or expired Atlas API credentials | Verify `apiClientId`/`apiClientSecret` and project-level permissions |
| 429 rate limit | Too many API calls | Wait and retry; avoid tight loops of discover calls |

## Pre-Deploy Quality Checklist

Before creating a processor, verify:

### Connection Validation (MANDATORY - Always do this first)
- [ ] **CRITICAL**: Call `atlas-streams-discover` → `action: "list-connections"` to list all connections in workspace
- [ ] **CRITICAL**: Call `atlas-streams-discover` → `action: "inspect-connection"` for EACH connection referenced in pipeline
- [ ] **CRITICAL**: Verify connection names clearly indicate their actual targets (avoid generic names like "atlascluster" pointing to "ClusterRestoreTest")
- [ ] **CRITICAL**: Present connection summary to user: "Connection 'X' → Actual target 'Y'" for each connection
- [ ] **CRITICAL**: Warn user if connection names don't match their targets and ask for confirmation
- [ ] All connections are in READY state
- [ ] Connection types match usage (Cluster for $source/$merge, Kafka for topics, etc.)

### Pipeline Validation
- [ ] `search-knowledge` was called to validate sink/source field names
- [ ] Pipeline starts with `$source` and ends with `$merge`, `$emit`, `$https`, or `$externalFunction` (async)
- [ ] No `$$NOW`, `$$ROOT`, or `$$CURRENT` in the pipeline
- [ ] Kafka `$source` includes a `topic` field
- [ ] Kafka `$source` with windowed pipeline includes `partitionIdleTimeout` (prevents windows from stalling on idle partitions)
- [ ] HTTPS connections are only used in `$https` enrichment or sink stages, not in `$source`
- [ ] DLQ is configured (recommended for production)
- [ ] `$https` stages use `onError: "dlq"` (not `"fail"`)
- [ ] `$externalFunction` stages use `onError: "dlq"` and `execution` is explicitly set
- [ ] API auth is stored in connection settings, not hardcoded in the pipeline

## Post-Deploy Verification Workflow

After creating and starting a processor:
1. `atlas-streams-discover` → `action: "inspect-processor"` — confirm state is STARTED
2. `atlas-streams-discover` → `action: "diagnose-processor"` — check for errors in the health report
3. Use MongoDB `count` tool on the DLQ collection — verify no errors accumulating
4. Use MongoDB `find` tool on the output collection — verify documents are arriving
5. If output is low/zero, classify processor type before assuming a problem (see Debug section)

# MCP Server Troubleshooting

This skill requires the MongoDB MCP Server with Atlas Stream Processing tools enabled. If these tools are unavailable, follow the diagnostic steps below.

## Step 1: Verify MCP Server Connection

Check if the MongoDB MCP Server is connected to your environment.

**If not connected:**
- Install the MongoDB MCP Server
- Configure it with your Atlas API credentials (`apiClientId` and `apiClientSecret`)

## Step 2: Verify Tool Availability

Check that all four streams tools are available:
- `atlas-streams-discover`
- `atlas-streams-build`
- `atlas-streams-manage`
- `atlas-streams-teardown`

## Fallback Options (Limited Functionality)

If you cannot configure the MCP server immediately, you have limited alternatives:

### Option 1: Atlas CLI (Read-Only)
Use Atlas CLI API commands for exploration only:
```bash
atlas api streams listStreamWorkspaces --projectId <project-id>
atlas api streams getStreamWorkspace --workspaceName <workspace-name> --projectId <project-id>
```

**Limitations:**
- Read-only operations only
- Cannot create or modify processors
- No automated validation or diagnostics

### Option 2: mongosh with sp.process() (Prototyping Only)
Use `sp.process()` in mongosh for ephemeral pipeline testing:
```javascript
sp.process([
  { $source: { connectionName: "sample_stream_solar" } },
  { $match: { temperature: { $gt: 50 } } },
  { $limit: 10 }
])
```

**Limitations:**
- Ephemeral only (no deployed processors)
- No billing (runs locally)
- Cannot test production connections
- Limited to simple pipeline validation

## Recommended Action

**For full Atlas Stream Processing capabilities, configure the MongoDB MCP Server with streams preview features enabled.** The fallback options above provide minimal functionality and are not suitable for production workflows.

# Processor Output Diagnostics Reference

## The Problem

A user says "my processor isn't outputting anything" or "output seems low." Before assuming something is broken, you must **classify the processor type** — low output may be perfectly normal.

## Processor Type Classification

### Category 1: Alert / Anomaly Detection

**Expected output:** Low or zero most of the time. Spikes during anomalous events.

Examples:
- Fraud detection (flags suspicious transactions)
- Threshold alerting (temperature > 100, latency > 500ms)
- Error monitoring (filters for error-level events)
- Security alerting (unusual login patterns)

**Green flags (healthy):**
- Zero output during normal conditions
- Occasional bursts during genuine anomalies
- DLQ is empty or near-empty

**Red flags (problem):**
- Zero output during a *known* anomaly event
- DLQ filling up with errors
- Processor state is FAILED

### Category 2: Data Transformation / Ingestion

**Expected output:** Roughly 1:1 with input volume. Output should be proportional to source.

Examples:
- Format conversion (Kafka → Atlas)
- Data enrichment (add fields, lookup)
- Schema normalization
- Archive pipelines (collection → collection)

**Green flags (healthy):**
- Output volume roughly matches input volume
- Consistent throughput over time

**Red flags (problem):**
- Output is zero while source has data
- Output is much lower than expected source volume
- Growing backlog (source advancing but output not keeping up)
- DLQ accumulating documents

### Category 3: Filter / Quality Gate

**Expected output:** Variable — depends on match rate of filter criteria.

Examples:
- Quality filtering (`$match` for valid records)
- Data routing (priority-based splitting)
- Deduplication
- Sampling

**Green flags (healthy):**
- Output is a consistent percentage of input
- Percentage aligns with expected data quality/match rate

**Red flags (problem):**
- Output drops to zero when source has data
- Sudden change in output ratio without a data source change
- DLQ filling up (filter errors, not just filtered-out data)

## Diagnostic Workflow

### Step 1: Classify the processor

Ask the user what the processor does, or inspect the pipeline:
- `atlas-streams-discover` → `inspect-processor` — read the pipeline stages

**Classification heuristics from pipeline:**
- Has `$match` with narrow conditions (e.g., `severity > 8`) → likely **Alert**
- Pipeline is mostly `$addFields`/`$project`/`$merge` → likely **Transformation**
- `$match` filters broadly (e.g., `status: "active"`) → likely **Filter**
- Has `$tumblingWindow` with `$match` inside → likely **Alert** (windowed anomaly detection)
- Has `$tumblingWindow` with `$group` only → likely **Transformation** (aggregation)

### Step 2: Check processor state

- `atlas-streams-discover` → `diagnose-processor`
- If state is FAILED → the problem is not low output, it's a crash. See debugging trees in [development-workflow.md](development-workflow.md).

### Step 3: Check operational logs

- For detailed logs, direct the user to the Atlas UI: **Atlas → Stream Processing → Workspace → Processor → Logs tab**
- Operational logs contain runtime errors: Kafka producer/consumer failures, schema serialization issues, OOM events, connection timeouts

### Step 4: Check DLQ

- Use MongoDB `count` tool on the DLQ collection
- If DLQ has documents → use MongoDB `find` tool to inspect error messages
- Growing DLQ means documents are being *rejected*, not that nothing is flowing

### Step 5: Check output collection

- Use MongoDB `count` tool on the output collection
- Use MongoDB `find` tool with `sort: {"_id": -1}` and `limit: 5` to see most recent documents
- Check timestamps — are documents recent?

### Step 6: Interpret based on processor type

| Processor type | Zero output | Low output | Action |
|---------------|-------------|------------|--------|
| **Alert** | Probably normal | Probably normal | Verify a known test event triggers output |
| **Transformation** | Problem — check connections, DLQ | Problem — check filters, DLQ | Debug pipeline and connections |
| **Filter** | Could be normal if no data matches | Could be normal | Verify filter criteria against actual source data |

## Common Diagnostic Patterns

After running `diagnose-processor`, match the symptoms to these patterns:

| Symptom | Root Cause | Fix |
|---------|------------|-----|
| **Error 419 + "no partitions found"** | Kafka topic doesn't exist or is misspelled | Verify topic name with Kafka broker; check connection config |
| **State: FAILED + multiple restarts** | Connection-level error (bypasses DLQ) | Check operational logs for repeated error; fix connection config or pipeline |
| **State: STARTED + zero output + windowed pipeline** | Idle Kafka partitions blocking window closure | Add `partitionIdleTimeout` to Kafka `$source` (e.g., `{"size": 30, "unit": "second"}`) |
| **State: STARTED + zero output + non-windowed** | Source has no data or filter too strict | Check if source (Kafka topic, collection) has data; review `$match` filters |
| **High memoryUsageBytes approaching tier limit** | OOM risk — window state or pipeline too large | Upgrade to higher tier (see sizing-and-parallelism.md) |
| **DLQ count increasing** | Per-document processing errors | Use MongoDB `find` on DLQ collection to inspect failed documents and error messages |

**When providing fix steps:**
- Commit to a specific root cause based on the evidence
- Do NOT present a list of hypothetical scenarios
- Provide concrete, ordered steps (e.g., "stop → modify pipeline to add partitionIdleTimeout → restart with resumeFromCheckpoint: false")

## Contextual Factors

Before concluding there's a problem, consider:

- **Time of day:** Business-hours-only data sources produce nothing at night
- **Seasonality:** Holiday periods, end-of-month spikes, etc.
- **Source health:** Is the source (Kafka topic, collection) actually receiving data?
- **Window timing:** Windowed processors only emit when the window closes — a 5-minute tumbling window outputs nothing for up to 5 minutes after start
- **Idle partitions:** Kafka windows won't close if a partition has no data — check `partitionIdleTimeout`

## Best Practice: Document Expected Behavior

When creating processors, encourage users to use descriptive names that indicate the processor type:

| Name pattern | Type indication |
|-------------|-----------------|
| `fraud-detector` | Alert — low output expected |
| `order-enricher` | Transformation — 1:1 output expected |
| `quality-filter` | Filter — variable output expected |
| `iot-5min-rollup` | Transformation — output every 5 min |
| `error-monitor` | Alert — low output expected |

# Pipeline Patterns Reference

**Official examples repo**: https://github.com/mongodb/ASP_example (quickstarts, example processors, Terraform examples). Start with example_processors/README.md for the full pattern catalog.
Always consult the official repo for the latest validated patterns before creating processors.

## Stage Quick-Reference

| Stage | Purpose | Category |
|-------|---------|----------|
| `$source` | Data ingress (Kafka, Cluster, Kinesis, Sample) | Source (required, first) |
| `$match` | Filter documents | Stateless |
| `$project` | Select/reshape fields | Stateless |
| `$addFields` | Add computed fields | Stateless |
| `$unset` | Remove fields | Stateless |
| `$unwind` | Explode arrays into documents | Stateless |
| `$replaceRoot` | Promote nested document to root | Stateless |
| `$redact` | Field-level access control | Stateless |
| `$validate` | Schema enforcement (route invalid to DLQ) | Validation |
| `$lookup` | Enrich from Atlas collection | Enrichment |
| `$https` | Enrich from HTTP API | Enrichment |
| `$externalFunction` | Invoke Lambda (mid-pipeline, NOT terminal) | Enrichment |
| `$tumblingWindow` | Fixed-size non-overlapping windows | Stateful |
| `$hoppingWindow` | Fixed-size overlapping windows | Stateful |
| `$sessionWindow` | Gap-based per-key windows | Stateful |
| `$function` | JavaScript UDF (requires SP30+) | Custom Code |
| `$group` | Aggregate (inside windows) | Stateful |
| `$merge` | Write to Atlas collection | Output (required, last) |
| `$emit` | Write to Kafka, Kinesis, or S3 | Output (required, last) |

| Category | Stages | Rules |
|----------|--------|-------|
| **Source** (1, required) | `$source` | Must be first. One per pipeline. |
| **Stateless Processing** | `$match`, `$project`, `$addFields`, `$unset`, `$unwind`, `$replaceRoot`, `$redact` | No state or memory overhead. Place `$match` first to reduce volume. |
| **Enrichment** | `$lookup`, `$https`, `$externalFunction` (sync/async) | I/O-bound. Use `parallelism` for throughput. `$https` and `$externalFunction` can be mid-pipeline enrichment OR terminal sink. For sinks: `$https` sends to webhooks/APIs, `$externalFunction` requires `execution: "async"`. |
| **Validation** | `$validate` | Schema enforcement. Place early to catch bad data before expensive stages. |
| **Stateful/Window** | `$tumblingWindow`, `$hoppingWindow`, `$sessionWindow` | Accumulates state in memory. Monitor `memoryUsageBytes`. |
| **Custom Code** | `$function` | JavaScript UDFs. Requires SP30+. |
| **Output** (1+, required) | `$merge`, `$emit`, `$https`, `$externalFunction` (async only) | Must be last. Required for deployed processors. |

## Invalid Constructs

Do NOT use these in streaming pipelines:
- `$$NOW`, `$$ROOT`, `$$CURRENT` — not available in stream processing
- HTTPS connections as `$source` — HTTPS is for `$https` enrichment only
- Kafka `$source` without `topic` — topic field is required
- Pipelines without a sink — `$merge`/`$emit` required for deployed processors (sinkless only works via `sp.process()`)
- Lambda connections with `$emit` — Lambda uses `$externalFunction` (can be mid-pipeline or terminal sink with async execution), not `$emit`

## Source Patterns

### MongoDB Change Stream
```json
{"$source": {"connectionName": "my-cluster"}}
```

With full document and pushdown pipeline:
```json
{"$source": {
  "connectionName": "my-cluster",
  "db": "mydb", "coll": "mycoll",
  "fullDocument": "updateLookup",
  "fullDocumentBeforeChange": "whenAvailable",
  "pipeline": [{"$match": {"operationType": "insert"}}]
}}
```

### Kafka (topic is REQUIRED)
```json
{"$source": {
  "connectionName": "my-kafka",
  "topic": "my-topic",
  "auto_offset_reset": "earliest",
  "partitionIdleTimeout": {"size": 30, "unit": "second"}
}}
```

### Kinesis
```json
{"$source": {
  "connectionName": "my-kinesis",
  "stream": "my-stream",
  "config": {"initialPosition": "TRIM_HORIZON"},
  "shardIdleTimeout": {"size": 30, "unit": "second"},
  "consumerARN": "arn:aws:kinesis:us-east-1:123456789:stream/my-stream/consumer/my-consumer:123"
}}
```

`stream` (required): Kinesis stream name. `config.initialPosition`: `TRIM_HORIZON` (oldest, default) or `LATEST`. `shardIdleTimeout`: unblocks windows when shards go idle (like Kafka `partitionIdleTimeout`). `consumerARN` (optional): enables enhanced fan-out for dedicated throughput.

### Inline Documents (ephemeral testing only)
```json
{"$source": {"documents": [{"device_id": "sensor-1", "temp": 72.5}]}}
```

## Sink Patterns

### $merge to Atlas
```json
{"$merge": {"into": {"connectionName": "my-atlas", "db": "mydb", "coll": "mycoll"}}}
```

With match behavior and parallelism:
```json
{"$merge": {
  "into": {"connectionName": "my-atlas", "db": "mydb", "coll": "mycoll"},
  "on": "_id", "whenMatched": "replace", "whenNotMatched": "insert",
  "parallelism": 4
}}
```

`whenMatched`: `replace`, `merge`, `delete` (via `$cond`). `whenNotMatched`: `insert`.

Additive merge (append to arrays):
```json
{"$merge": {
  "into": {"connectionName": "my-atlas", "db": "mydb", "coll": "mycoll"},
  "on": "device_id",
  "whenMatched": [{"$addFields": {"readings": {"$concatArrays": ["$readings", "$$new.readings"]}}}],
  "whenNotMatched": "insert"
}}
```

Dynamic routing:
```json
{"$merge": {"into": {
  "connectionName": "my-atlas", "db": "mydb",
  "coll": {"$cond": {"if": {"$eq": ["$priority", "high"]}, "then": "alerts", "else": "events"}}
}}}
```

### $emit to Kafka
```json
{"$emit": {
  "connectionName": "my-kafka", "topic": "output-topic",
  "key": {"field": "device_id", "format": "string"}
}}
```

Key formats: `string`, `json`, `int`, `long`, `binData`. Tombstone support: `"tombstoneWhen": {"$expr": {"$eq": ["$status", "deleted"]}}`.

### $emit to Kafka with Schema Registry (Avro)
```json
{"$emit": {
  "connectionName": "my-kafka", "topic": "output-topic",
  "schemaRegistry": {
    "connectionName": "my-schema-registry",
    "valueSchema": {
      "type": "avro",
      "schema": {
        "type": "record", "name": "SensorReading",
        "fields": [
          {"name": "device_id", "type": "string"},
          {"name": "temp", "type": "double"},
          {"name": "timestamp", "type": "long"}
        ]
      },
      "options": {
        "subjectNameStrategy": "TopicNameStrategy",
        "autoRegisterSchemas": true
      }
    }
  }
}}
```
Requires a `SchemaRegistry` connection (see [connection-configs.md](connection-configs.md#schemaregistry)). `valueSchema.type` must be lowercase `avro` (case-sensitive). `valueSchema.schema` is always required, even with `autoRegisterSchemas: true`.

### $emit to Kinesis
```json
{"$emit": {"connectionName": "my-kinesis", "stream": "out", "partitionKey": "$device_id"}}
```

### $emit to S3
```json
{"$emit": {
  "connectionName": "my-s3", "bucket": "my-bucket",
  "path": {"$concat": ["data/", {"$dateToString": {"format": "%Y/%m/%d", "date": "$timestamp"}}]},
  "config": {"outputFormat": "relaxedJson"}
}}
```
Fields: `connectionName` (required), `bucket` (required), `path` (required — key prefix string or expression), `region` (optional), `config` (optional — `outputFormat`, `writeOptions`, `delimiter`, `compression`).

### $https as Sink (webhook/API)
```json
{"$https": {
  "connectionName": "my-webhook",
  "path": "/events",
  "method": "POST",
  "onError": "dlq"
}}
```

When used as a **final sink stage**, `$https` sends processed documents to an external HTTP endpoint. Unlike mid-pipeline usage (which enriches documents with API responses), sink usage doesn't expect a response to merge back into the document. Useful for:
- Sending data to webhooks
- Posting to external APIs
- Triggering external systems

### $externalFunction as Sink (Lambda async)
```json
{"$externalFunction": {
  "connectionName": "my-lambda",
  "functionName": "arn:aws:lambda:us-west-1:123456789:function:my-function",
  "execution": "async",
  "onError": "dlq"
}}
```

**Important**: When used as a **final sink stage**, `$externalFunction` MUST use `execution: "async"`. This fires off the Lambda function without waiting for a response, useful for:
- Triggering downstream AWS applications or analytics
- Notifying external systems
- Firing off alerts or billing logic
- Propagating data to external workflows

Unlike mid-pipeline usage (where `execution: "sync"` is allowed for enrichment), sink usage requires async execution only. The pipeline still needs this as the terminal stage — you cannot use `$emit` to invoke Lambda.

## Window Patterns

### Tumbling
```json
{"$tumblingWindow": {
  "interval": {"size": 5, "unit": "minute"},
  "pipeline": [{"$group": {"_id": "$deviceId", "avg": {"$avg": "$temp"}, "count": {"$sum": 1}}}]
}}
```

### Hopping (with allowedLateness)
```json
{"$hoppingWindow": {
  "interval": {"size": 5, "unit": "minute"},
  "hopSize": {"size": 1, "unit": "minute"},
  "allowedLateness": {"size": 15, "unit": "second"},
  "pipeline": [{"$group": {"_id": "$region", "total": {"$sum": "$amount"}}}]
}}
```

### Session
```json
{"$sessionWindow": {
  "gap": {"size": 5, "unit": "minute"}, "key": "$userId",
  "pipeline": [{"$group": {"_id": "$userId", "actions": {"$push": "$action"}, "count": {"$sum": 1}}}]
}}
```

### Late data
```json
{"$tumblingWindow": {
  "interval": {"size": 1, "unit": "minute"},
  "allowedLateness": {"size": 30, "unit": "second"},
  "boundaryType": "eventTime",
  "pipeline": [{"$group": {"_id": "$sensorId", "max": {"$max": "$value"}}}]
}}
```

`boundaryType`: `eventTime` (document timestamp) or `processTime` (wall clock, default).

## Windowing Rules
- Windows require `$group` inside the window pipeline
- Idle Kafka partitions block windows — use `partitionIdleTimeout`
- `allowedLateness` lets late docs update closed windows

## Enrichment Patterns

### $https
```json
{"$https": {
  "connectionName": "my-api",
  "path": {"$concat": ["/users/", "$userId"]},
  "method": "GET", "as": "userInfo", "onError": "dlq"
}}
```

`onError`: `dlq` (recommended), `discard`, `fail`. Store auth in connection settings, not pipeline. Place `$https` after windows to batch requests.

### $lookup
```json
{"$lookup": {
  "connectionName": "my-atlas",
  "from": {"db": "mydb", "coll": "users"},
  "localField": "userId", "foreignField": "_id", "as": "user",
  "parallelism": 2
}}
```

### $externalFunction (Lambda - Mid-Pipeline Enrichment)
```json
{"$externalFunction": {
  "connectionName": "my-lambda",
  "functionName": "my-function-name",
  "execution": "sync",
  "as": "lambdaResult",
  "onError": "dlq",
  "payload": [
    {"$project": {"userId": 1, "data": 1}}
  ]
}}
```

**Mid-pipeline usage:**
- `execution`: `sync` (waits for Lambda result, stores in `as` field) or `async` (non-blocking)
- `as`: Field name to store Lambda response (required for `sync`, ignored for `async`)
- `payload`: Optional inner pipeline to customize request body sent to Lambda
- Use for enriching/transforming documents before downstream stages

**Sink usage:** See the Sink Patterns section. When used as final stage, MUST use `execution: "async"` only.

### $validate (Schema Validation)
```json
{"$validate": {
  "validator": {"$jsonSchema": {
    "required": ["device_id", "timestamp", "reading"],
    "properties": {
      "device_id": {"bsonType": "string"},
      "reading": {"bsonType": "double"}
    }
  }},
  "validationAction": "dlq"
}}
```

`validationAction`: `"dlq"` (recommended), `"discard"`, `"error"` (crashes processor — avoid in production). Place early to catch bad data before expensive stages.

### $function (JavaScript UDF)
```json
{"$addFields": {
  "boostedWatts": {"$function": {
    "body": "function(watts) { return watts * 1.2; }",
    "args": ["$watts"],
    "lang": "js"
  }}
}}
```

Requires **SP30+ tier**. `body`: JavaScript function as string. `args`: array of field references. `lang`: always `"js"`.

## Common Pipeline Patterns

### Array Normalization
```json
[
  {"$source": {"connectionName": "my-kafka", "topic": "orders"}},
  {"$unwind": "$items"},
  {"$replaceRoot": {"newRoot": {"$mergeObjects": ["$items", {"orderId": "$orderId", "ts": "$timestamp"}]}}},
  {"$merge": {"into": {"connectionName": "my-atlas", "db": "mydb", "coll": "line_items"}}}
]
```

### Dynamic Kafka Topic Routing
```json
{"$emit": {
  "connectionName": "my-kafka",
  "topic": {"$switch": {
    "branches": [
      {"case": {"$eq": ["$severity", "critical"]}, "then": "alerts-critical"},
      {"case": {"$eq": ["$severity", "warning"]}, "then": "alerts-warning"}
    ],
    "default": "alerts-info"
  }}
}}
```

### Complex Event Processing (Fraud Detection)
```json
[
  {"$source": {"connectionName": "my-kafka", "topic": "transactions"}},
  {"$tumblingWindow": {
    "interval": {"size": 5, "unit": "minute"},
    "pipeline": [
      {"$group": {
        "_id": "$userId",
        "txnCount": {"$sum": 1},
        "totalAmount": {"$sum": "$amount"},
        "uniqueLocations": {"$addToSet": "$location"}
      }},
      {"$addFields": {
        "suspiciousLocations": {"$gt": [{"$size": "$uniqueLocations"}, 3]},
        "highVelocity": {"$gt": ["$txnCount", 10]}
      }},
      {"$match": {"$or": [{"suspiciousLocations": true}, {"highVelocity": true}]}}
    ]
  }},
  {"$merge": {"into": {"connectionName": "my-atlas", "db": "fraud", "coll": "alerts"}}}
]
```

### Graceful Degradation with $ifNull
```json
{"$addFields": {
  "userName": {"$ifNull": ["$userInfo.name", "unknown"]},
  "userTier": {"$ifNull": ["$userInfo.tier", "standard"]},
  "enrichmentSucceeded": {"$ne": [{"$type": "$userInfo"}, "missing"]}
}}
```

## Window Metadata

Inside window pipelines, `_stream_meta.window.start` and `_stream_meta.window.end` provide boundary timestamps:
```json
{"$group": {
  "_id": "$deviceId",
  "windowStart": {"$first": "$_stream_meta.window.start"},
  "windowEnd": {"$first": "$_stream_meta.window.end"},
  "avg": {"$avg": "$temp"}
}}
```

## Checkpoint Resume Constraints

With `resumeFromCheckpoint: true` (default), you CANNOT change: window type, interval, remove windows, or modify `$source`. Set `false` to make these changes (restarts from beginning).

## DLQ Configuration
```json
{"dlq": {"connectionName": "my-atlas", "db": "streams_dlq", "coll": "failed_documents"}}
```
DLQ documents include: original document, error message, stage info, timestamp.

## Sample Stream Formats

| Format | Data type |
|--------|-----------|
| `sample_stream_solar` | Solar panel IoT readings (default) |
| `samplestock` | Stock market tick data |
| `sampleweather` | Weather station readings |
| `sampleiot` | Generic IoT sensor data |
| `samplelog` | Application log events |
| `samplecommerce` | E-commerce transaction data |

## Chained Processors (Multi-Sink Pattern)

**CRITICAL: A single pipeline can only have ONE terminal sink** (`$merge` or `$emit`). You CANNOT have both `$merge` and `$emit` as terminal stages. When a user requests multiple output destinations (e.g., "write to Atlas AND emit to Kafka" or "archive to S3 AND send to Lambda"), you MUST:

1. **Acknowledge** the single-sink constraint explicitly in your response
2. **Propose chained processors**: Processor A reads source → enriches → writes to intermediate via `$merge` (Atlas) or `$emit` (Kafka). Processor B reads from that intermediate (change stream or Kafka topic) → emits to second destination. Kafka-as-intermediate is lower latency; Atlas-as-intermediate is simpler to inspect.
3. **Show both processor pipelines** including any `$lookup` enrichment stages with `parallelism` settings.

Note: `$externalFunction` (Lambda) can be used mid-pipeline OR as a terminal sink (with `execution: "async"`). A pipeline with mid-pipeline `$externalFunction` AND a terminal `$merge`/`$emit` is a valid single-sink pattern (Lambda enriches, then the result is written to the sink).

## Required Field Examples by Stage

### $source (Kinesis)
Use `stream` (NOT `streamName` or `topic`) for the Kinesis stream name.
```json
{"$source": {"connectionName": "my-kinesis", "stream": "my-stream"}}
```

### $source (change stream)
Include `fullDocument: "updateLookup"` to get the full document content.

### $emit (Kinesis)
MUST include `partitionKey`.
```json
{"$emit": {"connectionName": "my-kinesis", "stream": "my-stream", "partitionKey": "$fieldName"}}
```

### $emit (S3)
Use `path` (NOT `prefix`).
```json
{"$emit": {"connectionName": "my-s3", "bucket": "my-bucket", "path": "data/year={$year}", "config": {"outputFormat": {"name": "json"}}}}
```

# Sizing & Parallelism Reference

## Tier Hardware Specs

| Tier | vCPU | RAM | Bandwidth | Max Parallelism | Kafka Partitions | Use case |
|------|------|-----|-----------|-----------------|------------------|----------|
| SP2  | 0.25 | 512MB | 50 Mbps | 1 | 32 | Minimal filtering, testing |
| SP5  | 0.5 | 1GB | 125 Mbps | 2 | 64 | Simple filtering and routing |
| SP10 | 1 | 2GB | 200 Mbps | 8 | Unlimited | Moderate workloads, joins, grouping |
| SP30 | 2 | 8GB | 750 Mbps | 16 | Unlimited | Windows, JavaScript UDFs, production |
| SP50 | 8 | 32GB | 2500 Mbps | 64 | Unlimited | High throughput, large window state |

**Memory rule:** 20% is reserved for overhead. User state (window accumulation, sort buffers) must stay below 80% of tier RAM. Exceeding this causes OOM failure.

## How Parallelism Works

Every stage in a pipeline runs with default `parallelism: 1`. This base level is included in your tier at no additional cost.

When you need higher throughput for specific stages, increase their parallelism beyond 1. **Only values > 1 count toward your tier's maximum.**

Stages that commonly benefit from parallelism:
- `$merge` — concurrent writes to Atlas
- `$lookup` — concurrent reads for enrichment
- `$https` — concurrent API calls

## Parallelism Calculation

**Formula:** `Total Parallelism = sum of (parallelism - 1) for all stages where parallelism > 1`

### Tier Selection Algorithm

```
If Total Parallelism = 0:   → SP2  (max 1)
If Total Parallelism = 1:   → SP5  (max 2)
If Total Parallelism ≤ 8:   → SP10 (max 8)
If Total Parallelism ≤ 16:  → SP30 (max 16)
If Total Parallelism ≤ 64:  → SP50 (max 64)
```

### Worked Examples

**Simple pipeline (all parallelism = 1):**
```
$source:    parallelism = 1  (does not count)
$match:     parallelism = 1  (does not count)
$merge:     parallelism = 1  (does not count)

Total = 0 → SP2
```

**Medium pipeline:**
```
$source:    parallelism = 1  (does not count)
$match:     parallelism = 1  (does not count)
$lookup:    parallelism = 4  (counts as 3)
$merge:     parallelism = 4  (counts as 3)

Total = 3 + 3 = 6 → SP10 (max 8)
```

**Complex pipeline:**
```
$source:    parallelism = 1  (does not count)
$https:     parallelism = 6  (counts as 5)
$merge:     parallelism = 8  (counts as 7)

Total = 5 + 7 = 12 → SP30 (max 16)
```

### API Error for Parallelism Exceeded

If you specify a tier too small for the pipeline's parallelism, the API returns:
```
"Operator parallelism requested exceeds limit for this tier.
(Requested: X, Limit: Y). Minimum tier for this workload: SPxx or larger."
```

Solution: Use `atlas-streams-manage` → `stop-processor`, then `start-processor` with a higher `tier` value.

## Complexity-Based Tier Selection

When parallelism is all default (1), choose tier based on pipeline complexity:

| Pipeline feature | Complexity weight | Minimum tier |
|-----------------|-------------------|--------------|
| Simple `$match` + `$project` only | Low | SP2-SP5 |
| `$addFields` with expressions | Low-Medium | SP5-SP10 |
| `$lookup` or `$https` enrichment | Medium | SP10 |
| `$group` aggregation | Medium | SP10 |
| `$tumblingWindow` or `$hoppingWindow` | Medium-High | SP10-SP30 |
| `$sessionWindow` | High | SP30 |
| `$function` (JavaScript UDFs) | High | SP30+ |
| Large window state (many unique keys) | Very High | SP30-SP50 |
| Multiple windows or chained enrichment | Very High | SP50 |

### Complexity Scoring Heuristic

For automated tier recommendation, score the pipeline:

| Feature | Points |
|---------|--------|
| `$function` (JavaScript) | +40 |
| Window operations (`$tumblingWindow`, `$hoppingWindow`, `$sessionWindow`) | +30 |
| `$lookup` or `$https` enrichment | +20 |
| `$group` aggregation | +15 |
| Kafka source integration | +15 |
| `$sort` operations | +10 |
| Pipeline has 5+ stages | +5 |
| Pipeline has 8+ stages | +10 |
| Pipeline has 12+ stages | +20 |

**Score → Tier mapping:**
- 0-10: SP2
- 11-20: SP5
- 21-40: SP10
- 41-60: SP30
- 61+: SP50

**Always take the higher of complexity-driven vs parallelism-driven tier recommendations.**

## Billing

Charges are **per-hour, calculated per-second**, only while the processor is running.

- `start-processor` begins billing
- `stop-processor` stops billing
- Stopped processors retain state for 45 days at no charge

**What's included in the tier price:**
- Compute (vCPU and RAM)
- State storage
- Base parallelism (parallelism = 1 for all stages)

**Additional costs (separate from tier):**
- Data transfer egress (varies by cloud provider and transfer type: intra-region, inter-region, internet)
- VPC Peering (AWS and GCP)
- Private Link connectivity

For current pricing: https://www.mongodb.com/docs/atlas/billing/stream-processing-costs/

## Sizing Workflow with MCP Tools

### Phase 1: Pre-deployment estimate

1. Score the pipeline using the complexity heuristic above
2. Calculate parallelism needs using the formula
3. Take the higher recommendation
4. Start with that tier (or one tier lower for cost savings during testing)

### Phase 2: Validation

1. Deploy the processor: `atlas-streams-build` → `resource: "processor"` with `autoStart: true`
2. Let it run for a representative period
3. Check stats: `atlas-streams-discover` → `diagnose-processor`
4. Review `memoryUsageBytes`:
   - Below 50% of tier RAM → over-provisioned, consider downsizing
   - 50-70% → good fit
   - 70-80% → at limit, monitor closely
   - Above 80% → under-provisioned, upgrade before it OOMs

### Phase 3: Optimization

1. Stop processor: `atlas-streams-manage` → `stop-processor`
2. Restart with adjusted tier: `atlas-streams-manage` → `start-processor` with `tier` override
3. Monitor for another period
4. Repeat until right-sized

### Cost Optimization: Time-of-Day Strategy

For workloads with predictable traffic patterns, adjust tiers by time of day:

| Period | Tier | Rationale |
|--------|------|-----------|
| Peak hours (business hours) | SP30-SP50 | Handle full volume |
| Off-peak hours | SP10-SP30 | Reduced volume |
| Maintenance windows | SP2-SP10 | Minimal processing |

To change tiers: `stop-processor` → `start-processor` with new `tier` value. Note: `resumeFromCheckpoint: true` (default) preserves state across tier changes.