Integration Patterns

This guide shows common integration patterns with complete, copy-paste ready examples.

Each pattern includes: - Use case description - Complete HCL configuration - Test commands

---

Table of Contents

1. GraphQL API → Database
2. REST → GraphQL Passthrough
3. GraphQL → REST Passthrough
4. RabbitMQ → Database
5. REST → RabbitMQ
6. GraphQL → RabbitMQ
7. Raw SQL Queries (JOINs)

---

1. GraphQL API → Database

Use case: Expose a GraphQL API that reads/writes to a database.

Configuration

# config.mycel
service {
  name    = "users-graphql-api"
  version = "1.0.0"
}

# connectors.mycel

# GraphQL Server
connector "api" {
  type   = "graphql"
  driver = "server"

  port       = 4000
  endpoint   = "/graphql"
  playground = true

  cors {
    origins = ["*"]
  }

  schema {
    path = "./schema.graphql"
  }
}

# Database
connector "db" {
  type   = "database"
  driver = "sqlite"
  path   = "./data/users.db"
}

# schema.graphql
type User {
  id: ID!
  email: String!
  name: String!
  createdAt: String
}

input CreateUserInput {
  email: String!
  name: String!
}

type Query {
  users: [User!]!
  user(id: ID!): User
}

type Mutation {
  createUser(input: CreateUserInput!): User!
  deleteUser(id: ID!): Boolean!
}

# flows.mycel

flow "get_users" {
  from {
    connector = "api"
    operation = "Query.users"
  }
  to {
    connector = "db"
    target    = "users"
  }
}

flow "get_user" {
  from {
    connector = "api"
    operation = "Query.user"
  }
  to {
    connector = "db"
    target    = "users"
  }
}

flow "create_user" {
  from {
    connector = "api"
    operation = "Mutation.createUser"
  }
  transform {
    email      = "lower(input.email)"
    name       = "input.name"
    created_at = "now()"
  }
  to {
    connector = "db"
    target    = "users"
  }
}

flow "delete_user" {
  from {
    connector = "api"
    operation = "Mutation.deleteUser"
  }
  to {
    connector = "db"
    target    = "users"
  }
}

Test

# Start service
mycel start --config .

# Query all users
curl -X POST http://localhost:4000/graphql \
  -H "Content-Type: application/json" \
  -d '{"query": "{ users { id email name } }"}'

# Create user
curl -X POST http://localhost:4000/graphql \
  -H "Content-Type: application/json" \
  -d '{"query": "mutation { createUser(input: {email: \"john@example.com\", name: \"John\"}) { id email } }"}'

# Open playground
open http://localhost:4000/playground

---

2. REST → GraphQL Passthrough

Use case: Receive REST requests and forward them to an external GraphQL API.

Configuration

# config.mycel
service {
  name    = "rest-to-graphql-gateway"
  version = "1.0.0"
}

# connectors.mycel

# REST Server (receives requests)
connector "api" {
  type = "rest"
  port = 3000
}

# External GraphQL API (forwards to)
connector "products_api" {
  type     = "graphql"
  driver   = "client"
  endpoint = "https://api.example.com/graphql"

  auth {
    type  = "bearer"
    token = env("PRODUCTS_API_TOKEN")
  }

  timeout     = "30s"
  retry_count = 3
}

# flows.mycel

# GET /products -> GraphQL query
flow "get_products" {
  from {
    connector = "api"
    operation = "GET /products"
  }

  enrich "products" {
    connector = "products_api"
    operation = "query { products { id name price } }"
  }

  transform {
    products = "enriched.products.products"
  }

  to {
    connector = "api"  # Returns response
  }
}

# GET /products/:id -> GraphQL query with variable
flow "get_product" {
  from {
    connector = "api"
    operation = "GET /products/:id"
  }

  enrich "product" {
    connector = "products_api"
    operation = "query GetProduct($id: ID!) { product(id: $id) { id name price description } }"
    params {
      id = "input.id"
    }
  }

  transform {
    id          = "enriched.product.product.id"
    name        = "enriched.product.product.name"
    price       = "enriched.product.product.price"
    description = "enriched.product.product.description"
  }

  to {
    connector = "api"
  }
}

# POST /products -> GraphQL mutation
flow "create_product" {
  from {
    connector = "api"
    operation = "POST /products"
  }

  enrich "created" {
    connector = "products_api"
    operation = "mutation CreateProduct($input: ProductInput!) { createProduct(input: $input) { id name } }"
    params {
      input = "input"
    }
  }

  transform {
    id   = "enriched.created.createProduct.id"
    name = "enriched.created.createProduct.name"
  }

  to {
    connector = "api"
  }
}

Test

# Start service
PRODUCTS_API_TOKEN=your-token mycel start --config .

# Get all products (REST -> GraphQL)
curl http://localhost:3000/products

# Get single product
curl http://localhost:3000/products/123

# Create product
curl -X POST http://localhost:3000/products \
  -H "Content-Type: application/json" \
  -d '{"name": "Widget", "price": 29.99}'

---

3. GraphQL → REST Passthrough

Use case: Expose a GraphQL API that internally calls REST endpoints.

Configuration

# config.mycel
service {
  name    = "graphql-to-rest-gateway"
  version = "1.0.0"
}

# connectors.mycel

# GraphQL Server (receives requests)
connector "api" {
  type   = "graphql"
  driver = "server"

  port       = 4000
  endpoint   = "/graphql"
  playground = true

  schema {
    path = "./schema.graphql"
  }
}

# External REST API (forwards to)
connector "users_rest" {
  type     = "http"
  base_url = "https://api.example.com"

  auth {
    type           = "api_key"
    api_key_header = "X-API-Key"
    api_key        = env("USERS_API_KEY")
  }

  timeout = "30s"
}

# schema.graphql
type User {
  id: ID!
  email: String!
  name: String!
  avatar: String
}

type Query {
  users: [User!]!
  user(id: ID!): User
}

type Mutation {
  createUser(email: String!, name: String!): User!
}

# flows.mycel

# Query.users -> GET /users
flow "get_users" {
  from {
    connector = "api"
    operation = "Query.users"
  }

  enrich "users" {
    connector = "users_rest"
    operation = "GET /users"
  }

  transform {
    result = "enriched.users"
  }

  to {
    connector = "api"
  }
}

# Query.user(id) -> GET /users/:id
flow "get_user" {
  from {
    connector = "api"
    operation = "Query.user"
  }

  enrich "user" {
    connector = "users_rest"
    operation = "GET /users/${input.id}"
  }

  transform {
    id     = "enriched.user.id"
    email  = "enriched.user.email"
    name   = "enriched.user.name"
    avatar = "enriched.user.avatar"
  }

  to {
    connector = "api"
  }
}

# Mutation.createUser -> POST /users
flow "create_user" {
  from {
    connector = "api"
    operation = "Mutation.createUser"
  }

  enrich "created" {
    connector = "users_rest"
    operation = "POST /users"
    params {
      email = "input.email"
      name  = "input.name"
    }
  }

  transform {
    id    = "enriched.created.id"
    email = "enriched.created.email"
    name  = "enriched.created.name"
  }

  to {
    connector = "api"
  }
}

Test

# Start service
USERS_API_KEY=your-key mycel start --config .

# Query users via GraphQL (calls REST internally)
curl -X POST http://localhost:4000/graphql \
  -H "Content-Type: application/json" \
  -d '{"query": "{ users { id email name } }"}'

# Query single user
curl -X POST http://localhost:4000/graphql \
  -H "Content-Type: application/json" \
  -d '{"query": "{ user(id: \"123\") { id email name avatar } }"}'

# Create user via GraphQL mutation
curl -X POST http://localhost:4000/graphql \
  -H "Content-Type: application/json" \
  -d '{"query": "mutation { createUser(email: \"john@example.com\", name: \"John\") { id email } }"}'

---

4. RabbitMQ → Database

Use case: Consume messages from a queue and store them in a database.

Configuration

# config.mycel
service {
  name    = "order-processor"
  version = "1.0.0"
}

# connectors.mycel

# RabbitMQ Consumer
connector "orders_queue" {
  type   = "mq"
  driver = "rabbitmq"

  host     = env("RABBITMQ_HOST", "localhost")
  port     = 5672
  user     = env("RABBITMQ_USER", "guest")
  password = env("RABBITMQ_PASS", "guest")

  queue {
    name    = "orders"
    durable = true
  }

  exchange {
    name        = "orders_exchange"
    type        = "topic"
    durable     = true
    routing_key = "order.#"
  }

  consumer {
    auto_ack    = false
    concurrency = 2
    prefetch    = 10
  }
}

# Database
connector "db" {
  type   = "database"
  driver = "postgres"

  host     = env("DB_HOST", "localhost")
  port     = 5432
  database = "orders"
  user     = env("DB_USER", "postgres")
  password = env("DB_PASS", "postgres")
}

# flows.mycel

# Process all order events
flow "process_order" {
  from {
    connector = "orders_queue"
    operation = "order.*"  # Matches order.created, order.updated, etc.
  }

  transform {
    id           = "input.body.order_id"
    product      = "input.body.product"
    quantity     = "input.body.quantity"
    customer     = "input.body.customer.email"
    status       = "'pending'"
    received_at  = "now()"
  }

  to {
    connector = "db"
    target    = "orders"
  }
}

# Process only order.created events
flow "new_order_notification" {
  from {
    connector = "orders_queue"
    operation = "order.created"
  }

  transform {
    order_id   = "input.body.order_id"
    email      = "input.body.customer.email"
    subject    = "'New Order Received'"
    message    = "'Your order ' + input.body.order_id + ' has been received.'"
    created_at = "now()"
  }

  to {
    connector = "db"
    target    = "notifications"
  }
}

Test

# Start RabbitMQ
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

# Start service
mycel start --config .

# Publish a test message (using rabbitmqadmin or your app)
rabbitmqadmin publish exchange=orders_exchange routing_key=order.created \
  payload='{"order_id":"ORD-001","product":"Widget","quantity":5,"customer":{"email":"john@example.com"}}'

---

5. REST → RabbitMQ

Use case: Receive REST requests and publish messages to a queue for async processing.

Configuration

# config.mycel
service {
  name    = "order-api"
  version = "1.0.0"
}

# connectors.mycel

# REST API
connector "api" {
  type = "rest"
  port = 3000
}

# RabbitMQ Publisher
connector "order_events" {
  type   = "mq"
  driver = "rabbitmq"

  host     = env("RABBITMQ_HOST", "localhost")
  port     = 5672
  user     = env("RABBITMQ_USER", "guest")
  password = env("RABBITMQ_PASS", "guest")

  publisher {
    exchange    = "orders_exchange"
    routing_key = "order.created"
    persistent  = true
  }
}

# Optional: Database to store order locally
connector "db" {
  type   = "database"
  driver = "sqlite"
  path   = "./data/orders.db"
}

# flows.mycel

# POST /orders -> Publish to queue
flow "create_order" {
  from {
    connector = "api"
    operation = "POST /orders"
  }

  transform {
    order_id   = "uuid()"
    product    = "input.product"
    quantity   = "input.quantity"
    customer   = "input.customer"
    status     = "'queued'"
    created_at = "now()"
  }

  to {
    connector   = "order_events"
    routing_key = "order.created"
  }
}

# POST /orders/:id/cancel -> Publish cancel event
flow "cancel_order" {
  from {
    connector = "api"
    operation = "POST /orders/:id/cancel"
  }

  transform {
    order_id     = "input.id"
    reason       = "input.reason"
    cancelled_at = "now()"
  }

  to {
    connector   = "order_events"
    routing_key = "order.cancelled"
  }
}

Test

# Start RabbitMQ
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

# Start service
mycel start --config .

# Create order (publishes to queue)
curl -X POST http://localhost:3000/orders \
  -H "Content-Type: application/json" \
  -d '{
    "product": "Widget Pro",
    "quantity": 5,
    "customer": {
      "name": "John Doe",
      "email": "john@example.com"
    }
  }'

# Cancel order
curl -X POST http://localhost:3000/orders/ORD-001/cancel \
  -H "Content-Type: application/json" \
  -d '{"reason": "Customer request"}'

# Check RabbitMQ management UI
open http://localhost:15672  # guest/guest

---

6. GraphQL → RabbitMQ

Use case: Expose GraphQL mutations that publish messages to a queue.

Configuration

# config.mycel
service {
  name    = "graphql-order-api"
  version = "1.0.0"
}

# connectors.mycel

# GraphQL Server
connector "api" {
  type   = "graphql"
  driver = "server"

  port       = 4000
  endpoint   = "/graphql"
  playground = true

  schema {
    path = "./schema.graphql"
  }
}

# RabbitMQ Publisher
connector "events" {
  type   = "mq"
  driver = "rabbitmq"

  host     = env("RABBITMQ_HOST", "localhost")
  port     = 5672
  user     = env("RABBITMQ_USER", "guest")
  password = env("RABBITMQ_PASS", "guest")

  publisher {
    exchange   = "events_exchange"
    persistent = true
  }
}

# Database for queries
connector "db" {
  type   = "database"
  driver = "sqlite"
  path   = "./data/orders.db"
}

# schema.graphql
type Order {
  id: ID!
  product: String!
  quantity: Int!
  status: String!
  createdAt: String!
}

input CreateOrderInput {
  product: String!
  quantity: Int!
  customerEmail: String!
}

type OrderResult {
  id: ID!
  status: String!
  message: String!
}

type Query {
  orders: [Order!]!
  order(id: ID!): Order
}

type Mutation {
  createOrder(input: CreateOrderInput!): OrderResult!
  cancelOrder(id: ID!, reason: String): OrderResult!
}

# flows.mycel

# Query orders from database
flow "get_orders" {
  from {
    connector = "api"
    operation = "Query.orders"
  }
  to {
    connector = "db"
    target    = "orders"
  }
}

# Mutation.createOrder -> Publish to queue
flow "create_order" {
  from {
    connector = "api"
    operation = "Mutation.createOrder"
  }

  transform {
    order_id       = "uuid()"
    product        = "input.product"
    quantity       = "input.quantity"
    customer_email = "input.customerEmail"
    status         = "'queued'"
    created_at     = "now()"
  }

  to {
    connector   = "events"
    routing_key = "order.created"
  }

  # Return confirmation to client
  returns = "OrderResult"
}

# Mutation.cancelOrder -> Publish cancel event
flow "cancel_order" {
  from {
    connector = "api"
    operation = "Mutation.cancelOrder"
  }

  transform {
    order_id     = "input.id"
    reason       = "default(input.reason, 'No reason provided')"
    cancelled_at = "now()"
  }

  to {
    connector   = "events"
    routing_key = "order.cancelled"
  }

  returns = "OrderResult"
}

Test

# Start RabbitMQ
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

# Start service
mycel start --config .

# Create order via GraphQL (publishes to RabbitMQ)
curl -X POST http://localhost:4000/graphql \
  -H "Content-Type: application/json" \
  -d '{
    "query": "mutation { createOrder(input: {product: \"Widget\", quantity: 5, customerEmail: \"john@example.com\"}) { id status message } }"
  }'

# Cancel order via GraphQL
curl -X POST http://localhost:4000/graphql \
  -H "Content-Type: application/json" \
  -d '{
    "query": "mutation { cancelOrder(id: \"ORD-001\", reason: \"Changed mind\") { id status } }"
  }'

# Open GraphQL Playground
open http://localhost:4000/playground

---

7. Raw SQL Queries (JOINs)

Use case: Execute complex SQL queries with JOINs, subqueries, or multi-table operations.

Configuration

# config.mycel
service {
  name    = "orders-api"
  version = "1.0.0"
}

# connectors.mycel

connector "api" {
  type = "rest"
  port = 3000
}

connector "db" {
  type   = "database"
  driver = "sqlite"  # or "postgres"
  path   = "./data/orders.db"
}

# flows.mycel

# Simple JOIN: Get order with user info
flow "get_order_with_user" {
  from {
    connector = "api"
    operation = "GET /orders/:id"
  }

  to {
    connector = "db"
    query     = <<-SQL
      SELECT
        o.id,
        o.product,
        o.quantity,
        o.status,
        o.created_at,
        u.name as user_name,
        u.email as user_email
      FROM orders o
      JOIN users u ON u.id = o.user_id
      WHERE o.id = :id
    SQL
  }
}

# Multiple named parameters
flow "get_orders_filtered" {
  from {
    connector = "api"
    operation = "GET /orders"
  }

  to {
    connector = "db"
    query     = <<-SQL
      SELECT o.*, u.name as user_name
      FROM orders o
      JOIN users u ON u.id = o.user_id
      WHERE o.status = :status
        AND o.created_at >= :from_date
      ORDER BY o.created_at DESC
      LIMIT :limit
    SQL
  }
}

# Aggregation query
flow "get_user_order_stats" {
  from {
    connector = "api"
    operation = "GET /users/:user_id/stats"
  }

  to {
    connector = "db"
    query     = <<-SQL
      SELECT
        u.id,
        u.name,
        u.email,
        COUNT(o.id) as total_orders,
        SUM(o.quantity) as total_items,
        MAX(o.created_at) as last_order_date
      FROM users u
      LEFT JOIN orders o ON o.user_id = u.id
      WHERE u.id = :user_id
      GROUP BY u.id, u.name, u.email
    SQL
  }
}

# Subquery example
flow "get_top_customers" {
  from {
    connector = "api"
    operation = "GET /reports/top-customers"
  }

  to {
    connector = "db"
    query     = <<-SQL
      SELECT
        u.id,
        u.name,
        u.email,
        order_stats.order_count,
        order_stats.total_spent
      FROM users u
      JOIN (
        SELECT
          user_id,
          COUNT(*) as order_count,
          SUM(total) as total_spent
        FROM orders
        WHERE status = 'completed'
        GROUP BY user_id
        HAVING COUNT(*) >= 5
      ) order_stats ON order_stats.user_id = u.id
      ORDER BY order_stats.total_spent DESC
      LIMIT 10
    SQL
  }
}

# INSERT with raw SQL
flow "create_order_with_audit" {
  from {
    connector = "api"
    operation = "POST /orders"
  }

  to {
    connector = "db"
    query     = <<-SQL
      INSERT INTO orders (id, user_id, product, quantity, status, created_at)
      VALUES (:id, :user_id, :product, :quantity, 'pending', datetime('now'))
    SQL
  }

  transform {
    id       = "uuid()"
    user_id  = "input.user_id"
    product  = "input.product"
    quantity = "input.quantity"
  }
}

# UPDATE with JOIN (PostgreSQL)
flow "update_order_status" {
  from {
    connector = "api"
    operation = "PUT /orders/:id/status"
  }

  to {
    connector = "db"
    query     = <<-SQL
      UPDATE orders
      SET status = :status,
          updated_at = NOW(),
          updated_by = :updated_by
      WHERE id = :id
      RETURNING id, status, updated_at
    SQL
  }

  transform {
    id         = "input.id"
    status     = "input.status"
    updated_by = "input.user_id"
  }
}

Named Parameters

Use :param_name syntax for parameters. Values come from: - Path parameters (:id from /orders/:id) - Query parameters (?status=active) - Request body (POST/PUT) - Transform output

# Parameter sources:
# - Path: GET /orders/:id        -> :id
# - Query: ?status=active        -> :status
# - Body: {"user_id": 123}       -> :user_id
# - Transform: id = "uuid()"     -> :id

Test

# Start service
mycel start --config .

# Get order with user info (JOIN)
curl http://localhost:3000/orders/1

# Get filtered orders with multiple params
curl "http://localhost:3000/orders?status=pending&from_date=2024-01-01&limit=10"

# Get user stats (aggregation)
curl http://localhost:3000/users/1/stats

# Get top customers (subquery)
curl http://localhost:3000/reports/top-customers

# Create order with raw SQL
curl -X POST http://localhost:3000/orders \
  -H "Content-Type: application/json" \
  -d '{"user_id": 1, "product": "Widget", "quantity": 5}'

---

Quick Reference

Connector Types

Type	Drivers	Use as Source	Use as Target
rest	-	HTTP Server	HTTP Response
graphql	server, client	GraphQL Server	GraphQL Client
database	sqlite, postgres	Query	Insert/Update
mq	rabbitmq, kafka	Consumer	Publisher
http	-	-	HTTP Client
tcp	-	TCP Server	TCP Client

Flow Structure

flow "name" {
  from {
    connector = "source_connector"
    operation = "operation"  # REST: "GET /path", GraphQL: "Query.field", MQ: "routing.key"
  }

  # Optional: validate input
  validate {
    input = "type.name"
  }

  # Optional: enrich from external services
  enrich "name" {
    connector = "other_connector"
    operation = "GET /lookup"
    params {
      id = "input.id"
    }
  }

  # Optional: transform data
  transform {
    field = "expression"
  }

  to {
    connector = "target_connector"
    target    = "table_or_topic"      # For database/mq
    query     = "raw SQL"             # For complex queries
  }

  # Optional: specify return type (GraphQL)
  returns = "Type"
}

Common CEL Functions

transform {
  # String functions
  id         = "uuid()"
  email      = "lower(trim(input.email))"
  slug       = "replace(lower(input.name), ' ', '-')"

  # Date/time
  created_at = "now()"
  timestamp  = "now_unix()"

  # Conditionals
  status     = "input.age >= 18 ? 'adult' : 'minor'"
  role       = "default(input.role, 'user')"

  # Enriched data
  price      = "enriched.pricing.price"

  # Math
  total      = "input.quantity * enriched.product.price"
}

---

More Examples

• GraphQL Full Example
• Message Queue Example
• Data Enrichment
• TCP Connector
• Transformations Guide

---

Event-Driven Integration Patterns

The following patterns show complete examples for event-driven architectures with RabbitMQ as the central message broker. Each one is backed by a runnable configuration in examples/integration/.

A note that applies to all of them: queue, exchange and consumer settings live on the connector, not inside the flow's from block. A flow names a connector and an operation — for a queue consumer the operation is the routing key or queue it handles, and for a publisher it is the routing key to publish under.

A message consumed from a queue arrives as input.body (the decoded payload), plus input.routing_key, input.exchange, input.headers and input.properties. See Input and Output.

Pattern: RabbitMQ → REST

Use case: Consume messages from a queue and call external REST APIs.

Common scenarios: - Process orders and notify fulfillment service - Sync data to external CRM/ERP systems - Trigger webhooks based on events

connector "rabbit" {
  type   = "mq"
  driver = "rabbitmq"

  host     = env("RABBIT_HOST", "localhost")
  port     = env("RABBIT_PORT", 5672)
  username = env("RABBIT_USER", "guest")
  password = env("RABBIT_PASS", "guest")

  queue {
    name    = "orders.pending"
    durable = true
  }

  exchange {
    name        = "orders"
    type        = "topic"
    durable     = true
    routing_key = "order.created"
  }

  consumer {
    auto_ack = false
    prefetch = 10

    dlq {
      enabled     = true
      queue       = "orders.pending.dlq"
      max_retries = 3
    }
  }
}

connector "fulfillment_api" {
  type     = "http"
  base_url = env("FULFILLMENT_API_URL")
  timeout  = "30s"

  auth {
    type  = "bearer"
    token = env("API_TOKEN")
  }
}

flow "process_order" {
  from {
    connector = "rabbit"
    operation = "orders.pending"
  }

  transform {
    external_id = "input.body.order_id"
    customer    = "input.body.customer"
    items       = "input.body.items"
    priority    = "input.body.priority == 'express' ? 'high' : 'normal'"
  }

  to {
    connector = "fulfillment_api"
    operation = "POST /v1/shipments"
  }
}

📁 Full example: examples/integration/rabbit-to-rest/

---

Pattern: RabbitMQ → GraphQL

Use case: Consume messages from a queue and call GraphQL APIs.

Common scenarios: - Update inventory in a GraphQL-based product service - Sync user data to Hasura/Apollo backend - Trigger mutations based on domain events

The transform builds the mutation's variables; the to block names the operation on the remote schema.

connector "inventory_graphql" {
  type     = "graphql"
  mode     = "client"
  endpoint = env("INVENTORY_GRAPHQL_URL")
  timeout  = "30s"
}

flow "update_inventory" {
  from {
    connector = "rabbit"
    operation = "inventory.updates"
  }

  transform {
    sku       = "input.body.sku"
    quantity  = "input.body.new_quantity"
    warehouse = "input.body.warehouse_id"
  }

  to {
    connector = "inventory_graphql"
    operation = "updateInventory"
  }
}

📁 Full example: examples/integration/rabbit-to-graphql/

---

Pattern: RabbitMQ → Exec

Use case: Consume messages from a queue and execute local processes/scripts.

Common scenarios: - PDF generation, image processing, video transcoding - Run data processing scripts (Python, R, shell) - Execute legacy system integrations - Trigger batch jobs

The command belongs to the connector, so one connector describes one script. The semaphore caps how many run at once.

connector "image_processor" {
  type   = "exec"
  driver = "local"

  command       = "./process_image.sh"
  working_dir   = env("SCRIPTS_DIR", "/app/scripts")
  timeout       = "3m"
  shell         = "/bin/bash"
  input_format  = "json"
  output_format = "json"
}

flow "process_image" {
  # At most three images being processed at any moment.
  semaphore {
    storage {
      driver = "memory"
    }
    key         = "'image_processing'"
    max_permits = 3
    timeout     = "5m"
  }

  from {
    connector = "rabbit"
    operation = "images.processing"
  }

  transform {
    source_path = "input.body.source_path"
    dest_path   = "input.body.dest_path"
    operation   = "input.body.operation"
  }

  to {
    connector = "image_processor"
  }
}

📁 Full example: examples/integration/rabbit-to-exec/

---

Pattern: REST → RabbitMQ (API Gateway)

Use case: Receive HTTP requests and publish messages to a queue.

Common scenarios: - API Gateway that decouples request handling from processing - Webhook receivers that queue events for async processing - Command endpoints that trigger background jobs

A REST request arrives flattened onto input — path parameters, query parameters and body fields all at the top level — so a :provider in the route is read as input.provider.

service {
  name = "api-gateway"

  rate_limit {
    enabled             = true
    requests_per_second = 1000
    burst               = 200
    key_extractor       = "ip"
  }
}

connector "api" {
  type = "rest"
  mode = "server"
  port = env("API_PORT", 8080)
}

connector "rabbit" {
  type   = "mq"
  driver = "rabbitmq"

  host     = env("RABBIT_HOST", "localhost")
  username = env("RABBIT_USER", "guest")
  password = env("RABBIT_PASS", "guest")

  exchange {
    name    = "events"
    type    = "topic"
    durable = true
  }
}

flow "create_order" {
  from {
    connector = "api"
    operation = "POST /orders"
  }

  transform {
    order_id   = "input.order_id ?? uuid()"
    customer   = "input.customer"
    items      = "input.items"
    created_at = "now()"
  }

  to {
    connector  = "rabbit"
    operation  = "order.created"
    persistent = true
  }

  # The response block shapes what the caller gets back. Each field is a CEL
  # expression, and `output` here is what the destination returned.
  response {
    message  = "'Order received'"
    order_id = "output.order_id"
  }
}

flow "receive_webhook" {
  from {
    connector = "api"
    operation = "POST /webhooks/:provider"
  }

  transform {
    provider   = "input.provider"
    event_type = "input.headers['x-event-type']"
    payload    = "input"
    created_at = "now()"
  }

  to {
    connector  = "rabbit"
    operation  = "webhook.received"
    persistent = true
  }

  response {
    received = "true"
  }
}

📁 Full example: examples/integration/rest-to-rabbit/

---

Pattern: File → RabbitMQ (Scheduled Import)

Use case: Read files periodically and publish content to a queue.

Common scenarios: - Process drop folders (CSV imports, data feeds) - Watch for new files and trigger processing - Batch file processing on schedule

A flow with a when attribute runs on a cron schedule instead of waiting for an event. A file source sets input._path, input._name and input._event; a multi-row file such as a CSV also sets input.rows, while a single-row file is flattened onto input directly.

connector "files" {
  type = "file"

  base_path   = env("FILES_BASE_PATH", "/data")
  format      = "csv"
  create_dirs = true
}

connector "rabbit" {
  type   = "mq"
  driver = "rabbitmq"

  host     = env("RABBIT_HOST", "localhost")
  username = env("RABBIT_USER", "guest")
  password = env("RABBIT_PASS", "guest")

  exchange {
    name    = "imports"
    type    = "topic"
    durable = true
  }
}

# Every day at 6am.
flow "process_daily_import" {
  when = "0 6 * * *"

  from {
    connector = "files"
    operation = "imports/daily/*.csv"
  }

  transform {
    batch_id    = "uuid()"
    source      = "input._name"
    rows        = "input.rows"
    row_count   = "size(input.rows)"
    imported_at = "now()"
  }

  to {
    connector  = "rabbit"
    operation  = "import.daily.batch"
    persistent = true
  }
}

# A watcher publishes one message per file as it lands.
flow "watch_drop_folder" {
  from {
    connector = "files"
    operation = "dropbox/*.json"
  }

  transform {
    file_id      = "uuid()"
    file_name    = "input._name"
    file_path    = "input._path"
    event        = "input._event"
    processed_at = "now()"
  }

  to {
    connector  = "rabbit"
    operation  = "file.dropped"
    persistent = true
  }
}

📁 Full example: examples/integration/file-to-rabbit/

---

Complete Event-Driven Architecture

Real-world systems combine multiple patterns:

┌─────────────────────────────────────────────────────────────┐
│                        Mycel Service                         │
│                                                              │
│  ┌──────────┐     ┌──────────┐     ┌──────────────────┐     │
│  │ REST API │────▶│ RabbitMQ │────▶│ External REST API │     │
│  └──────────┘     └──────────┘     └──────────────────┘     │
│                         │                                    │
│                         ├─────────▶ GraphQL Backend          │
│                         │                                    │
│                         └─────────▶ Exec (Scripts)           │
│                                                              │
│  ┌──────────┐                                                │
│  │ Files/S3 │─────────────────────▶ RabbitMQ                 │
│  └──────────┘                                                │
│     (cron)                                                   │
└─────────────────────────────────────────────────────────────┘

Example: Order Processing Pipeline

Five flows, each consuming what the previous one published. The stages are wired by routing key, so no flow knows about the next one.

# 1. Receive the order over HTTP and publish it.
flow "receive_order" {
  from {
    connector = "api"
    operation = "POST /orders"
  }

  transform {
    order_id   = "input.order_id ?? uuid()"
    customer   = "input.customer"
    items      = "input.items"
    received_at = "now()"
  }

  to {
    connector  = "rabbit"
    operation  = "order.received"
    persistent = true
  }
}

# 2. Reserve stock through the inventory service.
flow "validate_inventory" {
  from {
    connector = "rabbit"
    operation = "order.received"
  }

  transform {
    orderId = "input.body.order_id"
    items   = "input.body.items"
  }

  to {
    connector = "inventory_graphql"
    operation = "reserveStock"
  }
}

# 3. Charge the customer. The lock keeps a redelivered message from
#    charging twice.
flow "process_payment" {
  lock {
    storage {
      driver = "redis"
      url    = env("REDIS_URL", "redis://localhost:6379")
    }
    key     = "'payment:' + input.body.order_id"
    timeout = "5m"
  }

  from {
    connector = "rabbit"
    operation = "order.validated"
  }

  transform {
    amount   = "input.body.total"
    currency = "input.body.currency ?? 'USD'"
    customer = "input.body.customer_id"
  }

  to {
    connector = "payment_api"
    operation = "POST /v1/charges"
  }
}

# 4. Generate the invoice PDF.
flow "generate_invoice" {
  from {
    connector = "rabbit"
    operation = "order.paid"
  }

  transform {
    order_id = "input.body.order_id"
    customer = "input.body.customer"
    total    = "input.body.total"
  }

  to {
    connector = "invoice_generator"
  }
}

# 5. Tell the customer.
flow "send_notification" {
  from {
    connector = "rabbit"
    operation = "order.invoiced"
  }

  transform {
    to      = "input.body.customer.email"
    subject = "'Your order ' + input.body.order_id"
    body    = "'Thanks! Your invoice is attached.'"
  }

  to {
    connector = "email"
  }
}

---

Best Practices

1. Always Use DLQ for Critical Flows

A message that keeps failing has to go somewhere other than back onto the queue. The dead-letter queue is configured on the consumer, alongside the retry count that decides when a message has failed enough times.

connector "rabbit" {
  type   = "mq"
  driver = "rabbitmq"
  host   = env("RABBIT_HOST", "localhost")

  queue {
    name    = "critical.queue"
    durable = true
  }

  consumer {
    auto_ack = false

    dlq {
      enabled     = true
      queue       = "critical.queue.dlq"
      max_retries = 3
    }
  }
}

2. Use Semaphores for Rate-Limited APIs

flow "call_rate_limited_api" {
  semaphore {
    storage {
      driver = "redis"
      url    = env("REDIS_URL", "redis://localhost:6379")
    }
    key         = "'external_api'"
    max_permits = 5           # At most five concurrent calls
    timeout     = "30s"       # How long to wait for a permit
  }

  from {
    connector = "rabbit"
    operation = "api.calls"
  }

  to {
    connector = "external_api"
    operation = "POST /v1/items"
  }
}

key is a CEL expression, which is why the constant above is quoted twice — once for HCL, once for CEL. Make it depend on the message to get one permit pool per tenant, account or resource.

3. Use Locks for Non-Idempotent Operations

flow "process_payment" {
  lock {
    storage {
      driver = "redis"
      url    = env("REDIS_URL", "redis://localhost:6379")
    }
    key     = "'payment:' + input.body.payment_id"
    timeout = "5m"
    wait    = false           # Fail fast rather than queue behind the holder
  }

  from {
    connector = "rabbit"
    operation = "payments.pending"
  }

  to {
    connector = "payment_api"
    operation = "POST /v1/charges"
  }
}

4. Configure Circuit Breakers for External Services

A circuit breaker is an aspect, so one declaration covers every flow whose name matches — you do not repeat it per flow.

aspect "protect_external_calls" {
  when = "around"
  on   = ["call_*", "sync_*"]

  circuit_breaker {
    failure_threshold = 5      # Consecutive failures that open the circuit
    success_threshold = 2      # Successes needed to close it again
    timeout           = "30s"  # How long it stays open before a trial call
  }
}

5. Use Appropriate Message Persistence

# Critical messages — survive a broker restart.
to {
  connector  = "rabbit"
  operation  = "order.created"
  persistent = true
}

# Ephemeral messages — faster, lost on restart.
to {
  connector  = "rabbit"
  operation  = "metrics.sample"
  persistent = false
}