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

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

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

parameters:
  timescaleHost:
    description: Hostname or IP address of the TimescaleDB server
    type: string
    default: example.com

  timescalePort:
    description: Port of the TimescaleDB server
    type: integer
    default: 5432

  timescaleUsername:
    description: Database role that Connectware uses to connect
    type: string
    default: connectware

  timescalePassword:
    description: Password of the database role
    type: string

  timescaleDatabase:
    description: Database that contains the measurements hypertable
    type: string
    default: factory

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

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

resources:
  # Connection to the TimescaleDB database. TimescaleDB is a PostgreSQL
  # extension, so the connection uses the postgres URL scheme.
  timescaleConnection:
    type: Cybus::Connection
    properties:
      protocol: Sql
      connection:
        url: !sub 'postgres://${timescaleUsername}:${timescalePassword}@${timescaleHost}:${timescalePort}/${timescaleDatabase}'

  # Inserts one row into the measurements hypertable per message,
  # including the machine timestamp
  measurementWrite:
    type: Cybus::Endpoint
    properties:
      protocol: Sql
      connection: !ref timescaleConnection
      write:
        query: >-
          INSERT INTO measurements
          (time, site, area, line, cell, sensor, value)
          VALUES ($time, $site, $area, $line, $cell, $sensor, $value)

  # 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 one flat object whose keys match the query placeholders.
  measurementMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            topic: !sub '${topicRoot}/+site/+area/+line/+cell/measurements'
          publish:
            endpoint: !ref measurementWrite
          rules:
            - transform:
                expression: >-
                  {
                    "time": timestamp,
                    "site": $context.vars.site,
                    "area": $context.vars.area,
                    "line": $context.vars.line,
                    "cell": $context.vars.cell,
                    "sensor": sensor,
                    "value": value
                  }

  # Polls one-minute aggregates of the last 15 minutes using time_bucket
  aggregatedMeasurementsQuery:
    type: Cybus::Endpoint
    properties:
      protocol: Sql
      connection: !ref timescaleConnection
      subscribe:
        query: >-
          SELECT time_bucket('1 minute', time) AS bucket,
          site, area, line, cell, sensor,
          avg(value) AS avg_value,
          min(value) AS min_value,
          max(value) AS max_value
          FROM measurements
          WHERE time > now() - interval '15 minutes'
          GROUP BY bucket, site, area, line, cell, sensor
          ORDER BY bucket DESC
        # Be careful with low values, they can overload the database
        # (the unit is milliseconds).
        interval: !ref pollingInterval

  aggregatedMeasurementsMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            endpoint: !ref aggregatedMeasurementsQuery
          publish:
            topic: !sub '${topicRoot}/measurements/aggregated'
