---
# ----------------------------------------------------------------------------
# Service Commissioning File
# ----------------------------------------------------------------------------
# Copyright: Cybus GmbH
# Contact: support@cybus.io
# ----------------------------------------------------------------------------
# ClickHouse Integration (Example)
# ----------------------------------------------------------------------------

description: >
  Service commissioning file for the integration between Connectware and
  ClickHouse (Example)

metadata:
  name: ClickHouse Integration
  provider: cybus
  homepage: https://www.cybus.io
  version: 1.0.0

parameters:
  clickhouseScheme:
    description: Transport scheme, use https for ClickHouse Cloud
    type: string
    allowedValues:
      - http
      - https
    default: http

  clickhouseHost:
    description: Hostname or IP address of the ClickHouse HTTP interface
    type: string
    default: example.com

  clickhousePort:
    description: Port of the ClickHouse HTTP interface (8123 for http, 8443 for https)
    type: integer
    default: 8123

  clickhouseUsername:
    description: ClickHouse user that Connectware uses to connect
    type: string
    default: default

  clickhousePassword:
    description: Password of the ClickHouse user
    type: string

  clickhouseDatabase:
    description: Database that contains the production_events table
    type: string
    default: factory

  pollingInterval:
    description: Interval in milliseconds between polling queries
    type: integer
    default: 10000

  topicRoot:
    description: Root of the MQTT topic hierarchy
    type: string
    default: enterprise

resources:
  # Connection to the ClickHouse HTTP interface. ClickHouse answers
  # GET /ping without touching any data, so we probe there.
  clickhouseConnection:
    type: Cybus::Connection
    properties:
      protocol: Http
      connection:
        scheme: !ref clickhouseScheme
        host: !ref clickhouseHost
        port: !ref clickhousePort
        auth:
          username: !ref clickhouseUsername
          password: !ref clickhousePassword
        probePath: /ping
        probeMethod: GET

  # Inserts one row into the production_events table per message.
  # async_insert makes ClickHouse buffer the many small inserts
  # server-side and flush them in batches.
  productionEventWrite:
    type: Cybus::Endpoint
    properties:
      protocol: Http
      connection: !ref clickhouseConnection
      write:
        path: /
        query:
          database: !ref clickhouseDatabase
          query: INSERT INTO production_events FORMAT JSONEachRow
          async_insert: '1'

  # Feeds the write endpoint from the ISA-95 topic hierarchy. The named
  # wildcards provide the topic levels in $context.vars, so the transform
  # rule can build the row. The connector sends the "body" object as the
  # request body: a single JSON object, which is one JSONEachRow row.
  productionEventMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            topic: !sub '${topicRoot}/+site/+area/+line/+cell/production-events'
          publish:
            endpoint: !ref productionEventWrite
          rules:
            - transform:
                expression: >-
                  {
                    "body": {
                      "site": $context.vars.site,
                      "area": $context.vars.area,
                      "line": $context.vars.line,
                      "cell": $context.vars.cell,
                      "event_type": event_type,
                      "event_value": value
                    }
                  }

  # Polls a five-minute aggregation per line and event type. Subscribe
  # endpoints poll with GET requests, which ClickHouse treats as read-only.
  throughputQuery:
    type: Cybus::Endpoint
    properties:
      protocol: Http
      connection: !ref clickhouseConnection
      subscribe:
        path: /
        # Be careful with low values, they can overload the database
        # (the unit is milliseconds).
        interval: !ref pollingInterval
        query:
          database: !ref clickhouseDatabase
          # Return 64-bit integers, such as the count() result, as JSON
          # numbers instead of strings
          output_format_json_quote_64bit_integers: '0'
          query: >-
            SELECT site, line, event_type, count() AS events,
            round(avg(event_value), 2) AS avg_value
            FROM production_events
            WHERE event_time > now() - INTERVAL 5 MINUTE
            GROUP BY site, line, event_type
            ORDER BY site, line, event_type
            FORMAT JSON

  throughputMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            endpoint: !ref throughputQuery
          publish:
            topic: !sub '${topicRoot}/production-events/throughput'
