@things-factory/integration-base

@things-factory/integration-base

A comprehensive integration framework for the Things Factory ecosystem that enables seamless connectivity with external systems and services through scenario-driven automation.

Overview

The Integration Base module provides a robust, scenario-driven automation engine that enables complex integration workflows through a visual, step-based approach. It serves as the central nervous system for enterprise automation within the Things Factory framework.

Key Features

• Scenario Engine: Visual workflow designer with 50+ built-in task types
• Multi-Protocol Support: Database, HTTP/REST, GraphQL, MQTT, WebSocket integrations
• Headless Automation: Web scraping and browser automation via Puppeteer
• Multi-Tenant Architecture: Domain-based isolation with role-based access control
• Real-Time Monitoring: GraphQL subscriptions for live status updates
• Scheduling System: Cron-based automated execution with timezone support
• Edge Computing: Distributed execution through edge appliances

Installation

npm install @things-factory/integration-base

Quick Start

1. Basic Scenario Configuration

import { ScenarioEngine } from '@things-factory/integration-base'

// Create a simple HTTP integration scenario
const scenario = {
  name: 'fetch-user-data',
  description: 'Fetch user data from external API',
  steps: [
    {
      name: 'fetch-users',
      task: 'http-get',
      connection: 'api-server',
      params: {
        path: '/api/users',
        headers: { 'Authorization': 'Bearer ${token}' }
      }
    },
    {
      name: 'store-data',
      task: 'database-query',
      connection: 'main-db',
      params: {
        query: 'INSERT INTO users (id, name, email) VALUES (?, ?, ?)',
        bindings: '${fetch-users.data}'
      }
    }
  ]
}

2. Connection Setup

// Database connection
const dbConnection = {
  name: 'main-db',
  type: 'postgresql-connector',
  endpoint: 'postgresql://localhost:5432/mydb',
  params: {
    username: 'dbuser',
    password: 'dbpass',
    database: 'myapp'
  }
}

// HTTP API connection
const apiConnection = {
  name: 'api-server',
  type: 'http-connector',
  endpoint: 'https://api.example.com',
  params: {
    timeout: 30000,
    rejectUnauthorized: true
  }
}

Supported Connectors

Database Connectors

• PostgreSQL (postgresql-connector)
• MySQL (mysql-connector)
• Microsoft SQL Server (mssql-connector)
• Oracle (oracle-connector)
• SQLite (sqlite-connector)

Communication Connectors

• HTTP/REST (http-connector)
• GraphQL (graphql-connector) - Apollo Client with OAuth2 support
• MQTT (mqtt-connector) - Async message pub/sub
• WebSocket (socket-server) - Raw TCP socket communication

Automation Connectors

• Headless Browser (headless-connector) - Puppeteer-based web automation
• Python Runtime (pyrun-connector) - Python script execution
• Operato Integration (operato-connector) - Things Factory cross-instance

Utility Connectors

• Echo Service (echo-back-connector) - Testing and debugging
• Proxy Connector (proxy-connector) - Edge appliance delegation

Task Types

Data Operations

• database-query - Execute SQL queries with parameter binding
• graphql-query / graphql-mutate - GraphQL operations
• data-mapper - Transform data structures
• jsonata - JSONata transformations
• csv-readline - CSV file processing

Communication Tasks

• http-get / http-post / http-put / http-patch / http-delete
• headless-get / headless-post / headless-put / headless-patch / headless-delete
• mqtt-publish / mqtt-subscribe
• socket-listener / publish

Control Flow

• goto / end - Flow control
• switch-goto / switch-range-goto - Conditional branching
• sub-scenario - Nested scenario execution
• empty-check - Validation tasks
• variables - Variable management

Automation & Utility

• headless-scrap - Web scraping
• pyrun-execute - Python script execution
• sleep / random - Utility functions
• script - Custom JavaScript execution

Configuration

Environment Variables

# Database connection
DB_HOST=localhost
DB_PORT=5432
DB_NAME=integration_db
DB_USERNAME=user
DB_PASSWORD=password

# Headless browser pool
HEADLESS_POOL_MIN=2
HEADLESS_POOL_MAX=10

# Logging
DEBUG=things-factory:integration-base:*
LOG_LEVEL=info

Domain Configuration

// Domain-specific environment variables
const domainConfig = {
  domain: 'tenant1.example.com',
  envVars: [
    { name: 'Connection::api-server::apiKey', value: 'secret-key-123' },
    { name: 'Connection::database::password', value: 'encrypted-password' }
  ]
}

API Reference

GraphQL Schema

type Scenario {
  id: String!
  name: String!
  description: String
  active: Boolean!
  schedule: String
  timezone: String
  steps: [Step!]!
}

type Connection {
  id: String!
  name: String!
  type: String!
  endpoint: String!
  active: Boolean!
  state: ConnectionStatus!
}

type ScenarioInstance {
  id: String!
  scenario: Scenario!
  status: ScenarioStatus!
  startedAt: String
  finishedAt: String
  result: JSON
}

REST API Endpoints

GET    /api/unstable/scenarios           # List scenarios
POST   /api/unstable/scenarios           # Create scenario
PUT    /api/unstable/scenarios/:id       # Update scenario
DELETE /api/unstable/scenarios/:id       # Delete scenario

GET    /api/unstable/scenario-instances  # List instances
POST   /api/unstable/run-scenario        # Execute scenario
POST   /api/unstable/start-scenario      # Start async scenario
POST   /api/unstable/stop-scenario       # Stop running scenario

Examples

Database Integration

const databaseScenario = {
  name: 'sync-customer-data',
  steps: [
    {
      name: 'fetch-customers',
      task: 'database-query',
      connection: 'source-db',
      params: {
        query: 'SELECT * FROM customers WHERE updated_at > ?',
        bindings: ['${lastSync}']
      }
    },
    {
      name: 'transform-data',
      task: 'data-mapper',
      params: {
        mapping: {
          customer_id: 'id',
          full_name: 'name',
          email_address: 'email'
        }
      }
    },
    {
      name: 'insert-customers',
      task: 'database-query',
      connection: 'target-db',
      params: {
        query: 'INSERT INTO customers (id, name, email) VALUES ${fetch-customers.data}',
        batchSize: 1000
      }
    }
  ]
}

Web Scraping with Headless Browser

const scrapingScenario = {
  name: 'scrape-product-prices',
  steps: [
    {
      name: 'login-to-site',
      task: 'headless-post',
      connection: 'ecommerce-site',
      params: {
        path: '/login',
        contentType: 'application/json',
        accessor: 'credentials'
      }
    },
    {
      name: 'scrape-products',
      task: 'headless-scrap',
      connection: 'ecommerce-site',
      params: {
        path: '/products',
        selectors: [
          { text: 'productName', value: '.product-title' },
          { text: 'price', value: '.price-display' },
          { text: 'availability', value: '.stock-status' }
        ],
        waitForSelectors: '.product-list',
        waitForTimeout: 5000
      }
    },
    {
      name: 'save-results',
      task: 'database-query',
      connection: 'analytics-db',
      params: {
        query: 'INSERT INTO product_prices (name, price, availability, scraped_at) VALUES ${scrape-products.data}'
      }
    }
  ]
}

MQTT Integration

const mqttScenario = {
  name: 'iot-data-processing',
  steps: [
    {
      name: 'subscribe-sensors',
      task: 'mqtt-subscribe',
      connection: 'iot-broker',
      params: {
        topic: 'sensors/+/temperature',
        qos: 1
      }
    },
    {
      name: 'validate-data',
      task: 'empty-check',
      params: {
        accessor: 'temperature',
        throwOnEmpty: true
      }
    },
    {
      name: 'store-reading',
      task: 'database-query',
      connection: 'timeseries-db',
      params: {
        query: 'INSERT INTO sensor_readings (sensor_id, temperature, timestamp) VALUES (?, ?, NOW())',
        bindings: ['${deviceId}', '${temperature}']
      }
    },
    {
      name: 'alert-if-critical',
      task: 'switch-range-goto',
      params: {
        value: '${temperature}',
        cases: [
          { min: 80, max: 999, goto: 'send-alert' }
        ]
      }
    },
    {
      name: 'send-alert',
      task: 'http-post',
      connection: 'alert-service',
      params: {
        path: '/alerts',
        contentType: 'application/json',
        accessor: 'alertData'
      }
    }
  ]
}

Scheduling

Configure automated scenario execution using cron expressions:

const scheduledScenario = {
  name: 'daily-data-sync',
  description: 'Sync data every day at 2 AM',
  schedule: '0 2 * * *',  // Cron expression
  timezone: 'Asia/Seoul',
  active: true,
  steps: [/* scenario steps */]
}

Security

Authentication & Authorization

• Role-based access control (RBAC) integration
• Domain-based isolation for multi-tenant environments
• OAuth2 authentication for external GraphQL services
• Encrypted parameter storage for sensitive configuration

Best Practices

• Store sensitive data in domain environment variables
• Use parameter encryption for passwords and API keys
• Implement IP-based access control where needed
• Regular security audits of scenario configurations

Monitoring & Debugging

Logging

# Enable debug logging
DEBUG=things-factory:integration-base:* npm start

# Specific component logging
DEBUG=things-factory:integration-base:scenario-engine npm start
DEBUG=things-factory:integration-base:connector:* npm start

Real-time Monitoring

# Subscribe to scenario status updates
subscription {
  scenarioInstanceStatusUpdated {
    id
    status
    progress
    result
    error
  }
}

# Subscribe to connection status
subscription {
  connectionStatusUpdated {
    name
    state
    lastConnected
  }
}

Development

Building from Source

# Clone the repository
git clone https://github.com/hatiolab/things-factory.git
cd things-factory/packages/integration-base

# Install dependencies
npm install

# Build the module
npm run build

# Run tests
npm test

Creating Custom Connectors

import { Connector } from '@things-factory/integration-base'

export class CustomConnector implements Connector {
  async connect(connection) {
    // Implementation
  }
  
  async disconnect(connection) {
    // Cleanup
  }
  
  get parameterSpec() {
    return [
      { type: 'string', name: 'apiKey', label: 'API Key' },
      { type: 'number', name: 'timeout', label: 'Timeout (ms)' }
    ]
  }
  
  get taskPrefixes() {
    return ['custom']
  }
}

Creating Custom Tasks

import { TaskRegistry } from '@things-factory/integration-base'

async function CustomTask(step, context) {
  const { params } = step
  const { logger, data, domain } = context
  
  // Task implementation
  return { data: result }
}

CustomTask.parameterSpec = [
  { type: 'string', name: 'parameter1', label: 'Parameter 1' }
]

TaskRegistry.registerTaskHandler('custom-task', CustomTask)

Contributing

• Fork the repository
• Create a feature branch
• Make your changes
• Add tests for new functionality
• Submit a pull request

License

MIT © Hatiolab

Support

• Documentation
• Issue Tracker
• Discussions

Note: This module is part of the Things Factory ecosystem. For complete setup and configuration, refer to the main Things Factory documentation.