> ## Documentation Index
> Fetch the complete documentation index at: https://docs.coreflux.org/llms.txt
> Use this file to discover all available pages before exploring further.

# Data Pipeline Routes

> Configure MQTT bridges for broker-to-broker communication and email notifications

## Data Pipeline Overview

Data pipeline routes enable data flow between your Coreflux broker and external systems. They connect your broker to other MQTT brokers and email services.

<Tip>
  **Like a postal forwarding service.** Mail (messages) sent to your old address (local broker) automatically gets forwarded to your new address (remote broker). You set up the forwarding rules once, and it just works.
</Tip>

The pipeline route types are:

* **MQTT Bridge** - Broker-to-broker communication for edge-to-cloud sync and multi-site connectivity
* **Email** - Send notifications and alerts triggered by MQTT messages

***

## MQTT Bridge

The `MQTT_BRIDGE` route enables seamless data transfer between two MQTT brokers. Use it for edge-to-cloud synchronization, multi-site connectivity, or backup systems.

### Basic Syntax

```lot theme={null}
DEFINE ROUTE CloudBridge WITH TYPE MQTT_BRIDGE
    ADD SOURCE_CONFIG
        WITH BROKER SELF
    ADD DESTINATION_CONFIG
        WITH BROKER_ADDRESS "cloud.example.com"
        WITH BROKER_PORT '8883'
        WITH CLIENT_ID "EdgeDevice001"
        WITH USE_TLS "true"
    ADD MAPPING sensorSync
        WITH SOURCE_TOPIC "sensors/+/data"
        WITH DESTINATION_TOPIC "edge/sensors/+/data"
        WITH DIRECTION "out"
```

***

### Connection Configuration

<AccordionGroup>
  <Accordion title="Broker Settings">
    <ParamField path="BROKER_ADDRESS" type="string" required>
      IP address or hostname of the broker. Use `SELF` for the local broker.
    </ParamField>

    <ParamField path="BROKER_PORT" type="integer" required>
      MQTT port (typically 1883 for unencrypted, 8883 for TLS).
    </ParamField>

    <ParamField path="CLIENT_ID" type="string" required>
      Unique client identifier for the bridge connection.
    </ParamField>
  </Accordion>

  <Accordion title="Authentication">
    <ParamField path="USERNAME" type="string">
      Authentication username.
    </ParamField>

    <ParamField path="PASSWORD" type="string">
      Authentication password.
    </ParamField>
  </Accordion>

  <Accordion title="TLS/Security">
    <ParamField path="USE_TLS" type="boolean">
      Enable TLS encryption. Default: false.
    </ParamField>

    <ParamField path="ALLOW_UNTRUSTED_CERTS" type="boolean">
      Allow untrusted certificates. Default: false.
    </ParamField>

    <ParamField path="SERVER_CA_CERT_PATH" type="string">
      Path to CA certificate file.
    </ParamField>

    <ParamField path="CLIENT_CERT_PATH" type="string">
      Path to client certificate for mTLS.
    </ParamField>

    <ParamField path="CLIENT_CERT_PASS" type="string">
      Client certificate password.
    </ParamField>
  </Accordion>

  <Accordion title="Connection Handling">
    <ParamField path="RECONNECTION_RETRIES" type="integer">
      Number of reconnection attempts. Default: 5.
    </ParamField>
  </Accordion>
</AccordionGroup>

### Mapping Configuration

<AccordionGroup>
  <Accordion title="Topic Mapping">
    <ParamField path="SOURCE_TOPIC" type="string" required>
      Topic pattern to subscribe to on the source broker. Supports wildcards.
    </ParamField>

    <ParamField path="DESTINATION_TOPIC" type="string" required>
      Topic to publish to on the destination broker.
    </ParamField>

    <ParamField path="DIRECTION" type="string" required>
      Data flow direction: `out` (source→dest), `in` (dest→source), or `both`.
    </ParamField>
  </Accordion>
</AccordionGroup>

### Bridge Examples

<Tabs>
  <Tab title="Edge to Cloud">
    Send local sensor data to a cloud broker:

    ```lot theme={null}
    DEFINE ROUTE EdgeToCloud WITH TYPE MQTT_BRIDGE
        ADD SOURCE_CONFIG
            WITH BROKER SELF
        ADD DESTINATION_CONFIG
            WITH BROKER_ADDRESS "iot.cloud-provider.com"
            WITH BROKER_PORT '8883'
            WITH CLIENT_ID "EdgeDevice001"
            WITH USERNAME "edge_user"
            WITH PASSWORD "secure_password"
            WITH USE_TLS "true"
        ADD MAPPING sensorData
            WITH SOURCE_TOPIC "sensors/+/data"
            WITH DESTINATION_TOPIC "edge-001/sensors/+/data"
            WITH DIRECTION "out"
    ```
  </Tab>

  <Tab title="Cloud to Edge">
    Receive commands from cloud to local broker:

    ```lot theme={null}
    DEFINE ROUTE CloudCommands WITH TYPE MQTT_BRIDGE
        ADD SOURCE_CONFIG
            WITH BROKER SELF
        ADD DESTINATION_CONFIG
            WITH BROKER_ADDRESS "iot.cloud-provider.com"
            WITH BROKER_PORT '8883'
            WITH CLIENT_ID "EdgeDevice001-Commands"
            WITH USE_TLS "true"
        ADD MAPPING commands
            WITH SOURCE_TOPIC "local/commands/#"
            WITH DESTINATION_TOPIC "edge-001/commands/#"
            WITH DIRECTION "in"
    ```
  </Tab>

  <Tab title="Bidirectional">
    Full two-way synchronization:

    ```lot theme={null}
    DEFINE ROUTE FullSync WITH TYPE MQTT_BRIDGE
        ADD SOURCE_CONFIG
            WITH BROKER SELF
        ADD DESTINATION_CONFIG
            WITH BROKER_ADDRESS "partner-broker.example.com"
            WITH BROKER_PORT '1883'
            WITH CLIENT_ID "SyncClient"
        ADD MAPPING bidirectionalSync
            WITH SOURCE_TOPIC "shared/data/#"
            WITH DESTINATION_TOPIC "partner/data/#"
            WITH DIRECTION "both"
    ```
  </Tab>

  <Tab title="Secure mTLS Bridge">
    TLS-encrypted bridge with mutual authentication:

    ```lot theme={null}
    DEFINE ROUTE SecureBridge WITH TYPE MQTT_BRIDGE
        ADD SOURCE_CONFIG
            WITH BROKER_ADDRESS "192.168.1.10"
            WITH BROKER_PORT '8883'
            WITH CLIENT_ID "SecureSource"
            WITH USE_TLS "true"
            WITH SERVER_CA_CERT_PATH "/certs/ca.pem"
            WITH CLIENT_CERT_PATH "/certs/client.pem"
            WITH CLIENT_CERT_PASS "cert_password"
        ADD DESTINATION_CONFIG
            WITH BROKER_ADDRESS "secure.cloud.com"
            WITH BROKER_PORT '8883'
            WITH CLIENT_ID "SecureDest"
            WITH USE_TLS "true"
            WITH ALLOW_UNTRUSTED_CERTS "false"
        ADD MAPPING secureMapping
            WITH SOURCE_TOPIC "secure/+/data"
            WITH DESTINATION_TOPIC "cloud/+/data"
            WITH DIRECTION "out"
    ```
  </Tab>
</Tabs>

***

## Email Route

The `EMAIL` route sends templated emails based on MQTT messages. Ideal for alerts, notifications, and reports.

### Basic Syntax

```lot theme={null}
DEFINE ROUTE AlertEmail WITH TYPE EMAIL
    ADD SMTP_CONFIG
        WITH HOST "smtp.gmail.com"
        WITH PORT '587'
        WITH USERNAME "alerts@company.com"
        WITH PASSWORD "app-password-here"
        WITH USE_TLS "true"
    ADD EVENT criticalAlert
        WITH SOURCE_TOPIC "alerts/critical/+"
        WITH SUBJECT "CRITICAL: {value.json.alert_type}"
        WITH RECIPIENT "ops-team@company.com"
```

***

### SMTP Configuration

<AccordionGroup>
  <Accordion title="Server Settings">
    <ParamField path="HOST" type="string" required>
      SMTP server address (e.g., `smtp.gmail.com`, `smtp.office365.com`).
    </ParamField>

    <ParamField path="PORT" type="integer" required>
      SMTP port (typically 587 for TLS, 465 for SSL).
    </ParamField>

    <ParamField path="USE_TLS" type="boolean">
      Enable TLS encryption. Recommended: true.
    </ParamField>
  </Accordion>

  <Accordion title="Authentication">
    <ParamField path="USERNAME" type="string" required>
      Email account to send from.
    </ParamField>

    <ParamField path="PASSWORD" type="string" required>
      Email password or app-specific password.
    </ParamField>
  </Accordion>
</AccordionGroup>

### Event Configuration

<AccordionGroup>
  <Accordion title="Trigger & Routing">
    <ParamField path="SOURCE_TOPIC" type="string">
      MQTT topic to trigger email sending.
    </ParamField>

    <ParamField path="DESTINATION_TOPIC" type="string">
      MQTT topic to publish send status.
    </ParamField>
  </Accordion>

  <Accordion title="Email Content">
    <ParamField path="TEMPLATE_PATH" type="string">
      Path to HTML email template file.
    </ParamField>

    <ParamField path="SUBJECT" type="string">
      Email subject line (supports placeholders).
    </ParamField>

    <ParamField path="RECIPIENT" type="string">
      Email recipient address (supports placeholders).
    </ParamField>
  </Accordion>
</AccordionGroup>

### Template Placeholders

| Placeholder                 | Description                 |
| --------------------------- | --------------------------- |
| `{value}`                   | Full JSON payload as string |
| `{value.json.field}`        | Single field from JSON      |
| `{value.json.nested.field}` | Nested field access         |

### Email Examples

<Tabs>
  <Tab title="Critical Alerts">
    Send immediate alerts for critical events:

    ```lot theme={null}
    DEFINE ROUTE CriticalAlerts WITH TYPE EMAIL
        ADD SMTP_CONFIG
            WITH HOST "smtp.gmail.com"
            WITH PORT '587'
            WITH USERNAME "alerts@company.com"
            WITH PASSWORD "app-password-here"
            WITH USE_TLS "true"
        ADD EVENT criticalNotify
            WITH SOURCE_TOPIC "alerts/critical/+"
            WITH TEMPLATE_PATH "/templates/critical_alert.html"
            WITH SUBJECT "CRITICAL: {value.json.alert_type}"
            WITH RECIPIENT "ops-team@company.com"
    ```
  </Tab>

  <Tab title="Dynamic Recipient">
    Send to email address from payload:

    ```lot theme={null}
    DEFINE ROUTE UserNotifications WITH TYPE EMAIL
        ADD SMTP_CONFIG
            WITH HOST "smtp.office365.com"
            WITH PORT '587'
            WITH USERNAME "noreply@company.com"
            WITH PASSWORD "secure-password"
            WITH USE_TLS "true"
        ADD EVENT userNotify
            WITH SOURCE_TOPIC "users/+/notifications"
            WITH TEMPLATE_PATH "/templates/user_notification.html"
            WITH SUBJECT "Hello {value.json.username}"
            WITH RECIPIENT "{value.json.email}"
    ```
  </Tab>

  <Tab title="With Embedded Resources">
    Include images in email:

    ```lot theme={null}
    DEFINE ROUTE ReportEmail WITH TYPE EMAIL
        ADD SMTP_CONFIG
            WITH HOST "smtp.company.com"
            WITH PORT '587'
            WITH USERNAME "reports@company.com"
            WITH PASSWORD "password"
            WITH USE_TLS "true"
        ADD EVENT sendReport
            WITH SOURCE_TOPIC "reports/daily"
            WITH TEMPLATE_PATH "/templates/daily_report.html"
            WITH SUBJECT "Daily Report - {value.json.date}"
            WITH RECIPIENT "management@company.com"
            WITH EMBED_RESOURCES
                ADD "/images/logo.png"
                ADD "/images/chart.png"
    ```
  </Tab>
</Tabs>

<Note>
  For Gmail, use an [App Password](https://support.google.com/accounts/answer/185833) instead of your regular password. Enable 2FA on your Google account first.
</Note>

***

## Troubleshooting

<AccordionGroup>
  <Accordion title="Bridge Connection Timeout">
    * Verify IP address and port are correct
    * Check firewall allows outbound connections on the configured port
    * Ensure remote broker is running and accessible
    * Try increasing `RECONNECTION_RETRIES`
  </Accordion>

  <Accordion title="TLS/Certificate Errors">
    * Verify certificate paths are correct
    * Ensure certificates are valid and not expired
    * Check certificate chain is complete
    * For testing, use `ALLOW_UNTRUSTED_CERTS "true"` (not recommended for production)
  </Accordion>

  <Accordion title="Messages Not Bridging">
    * Verify SOURCE\_TOPIC and DESTINATION\_TOPIC are correct
    * Check DIRECTION setting matches your expected data flow
    * Ensure CLIENT\_ID is unique across all connections
    * Check broker logs for subscription errors
  </Accordion>

  <Accordion title="Email Not Sending">
    * Verify SMTP credentials are correct
    * For Gmail, ensure you're using an App Password
    * Check firewall allows outbound SMTP connections
    * Verify recipient email address is valid
  </Accordion>

  <Accordion title="Email Authentication Failed">
    * Double-check USERNAME and PASSWORD
    * Ensure 2FA is enabled if using app passwords
    * Some providers require enabling "less secure apps" access
    * Check for account lockouts due to failed attempts
  </Accordion>
</AccordionGroup>

***

## Best Practices

<AccordionGroup>
  <Accordion title="Use TLS for External Connections">
    Always enable TLS when connecting to external brokers:

    ```lot theme={null}
    WITH USE_TLS "true"
    ```
  </Accordion>

  <Accordion title="Unique Client IDs">
    Each bridge connection needs a unique CLIENT\_ID to avoid conflicts:

    ```lot theme={null}
    WITH CLIENT_ID "EdgeDevice001-SensorBridge"
    ```
  </Accordion>

  <Accordion title="Test with Local Broker First">
    Use `WITH BROKER SELF` during development:

    ```lot theme={null}
    ADD SOURCE_CONFIG
        WITH BROKER SELF
    ```
  </Accordion>

  <Accordion title="Secure Credentials">
    Store passwords securely and use app-specific passwords where available.
  </Accordion>

  <Accordion title="Handle Reconnection">
    Configure appropriate retry settings:

    ```lot theme={null}
    WITH RECONNECTION_RETRIES '10'
    ```
  </Accordion>
</AccordionGroup>

***

## Next Steps

<CardGroup cols={2}>
  <Card title="Data Storage Routes" icon="database" href="./data-storage/overview">
    Store MQTT data in databases.
  </Card>

  <Card title="REST API Routes" icon="globe" href="./rest-api-routes">
    Full REST API client/server configuration.
  </Card>
</CardGroup>