input:
  http_server:
    path: /windfree
    allowed_verbs: [POST]
    timeout: 10s
    sync_response:
      status: "202"
      headers:
        Content-Type: application/json

pipeline:
  processors:
    - mapping: |
        root = if this.type() != "object" {
          throw("request body must be a JSON object")
        } else if this.schema_version != 1 || this.event_type != "smartthings.device_status" {
          throw("unsupported event schema")
        } else if !["info", "warning", "error"].contains(this.level) {
          throw("invalid level")
        } else if !["ok", "no_data", "error"].contains(this.status) {
          throw("invalid status")
        } else if this.device_id.type() != "string" || this.observed_at.type() != "string" {
          throw("device_id and observed_at are required")
        } else {
          this
        }
    - mutation: |
        root.ingested_at = now()
        root.ingest_source = "redpanda-connect"
        meta kafka_key = this.device_id

output:
  redpanda:
    seed_brokers: [ "${KAFKA_BROKERS:127.0.0.1:9092}" ]
    topic: "${KAFKA_TOPIC:smartthings.device-status.v1}"
    key: '${! meta("kafka_key") }'
    tls:
      enabled: ${KAFKA_TLS_ENABLED:false}