apiVersion: v1
kind: ConfigMap
metadata:
  name: {{ include "thermo.fullname" . }}-writer
  labels:
    {{- include "thermo.labels" . | nindent 4 }}
data:
  redpanda.yaml: |
    http:
      enabled: true
      address: 0.0.0.0:4195

    input:
      kafka:
        addresses:
          {{- toYaml .Values.writer.kafka.brokers | nindent 10 }}
        topics:
          - {{ .Values.writer.kafka.topic }}
        consumer_group: {{ .Values.writer.kafka.consumerGroup }}
        start_from_oldest: {{ .Values.writer.kafka.startFromOldest }}

    pipeline:
      processors:
        - mapping: |
            root = if this.event_id.type() != "string" ||
              this.timestamp.type() != "string" ||
              this.source.type() != "string" ||
              this.event_type.type() != "string" ||
              this.severity.type() != "string" ||
              this.data.type() != "object" ||
              this.data.measurements.type() != "object" {
              deleted()
            } else {
              this
            }

    output:
      sql_raw:
        driver: postgres
        dsn: ${THERMO_POSTGRES_DSN}
        max_in_flight: 1
        query: |
          WITH payload AS (
            SELECT $1::jsonb AS document
          ), event_data AS (
            SELECT
              (document->>'event_id')::uuid AS event_id,
              COALESCE(document->>'timestamp', document#>>'{data,observed_at}')::timestamptz AS observed_at,
              COALESCE(document->>'ingested_at', document->>'timestamp')::timestamptz AS ingested_at,
              document->>'source' AS source,
              document->>'event_type' AS event_type,
              document->>'severity' AS severity,
              document#>>'{data,collector_id}' AS collector_id,
              COALESCE(
                document#>>'{data,device_id}',
                document#>>'{data,sensor_id}',
                document#>>'{data,device_address}'
              ) AS device_id,
              document AS raw_event,
              COALESCE(document#>'{data,measurements}', '{}'::jsonb) AS measurements
            FROM payload
          ), stored_event AS (
            INSERT INTO sensor_events (
              event_id, observed_at, ingested_at, source, event_type,
              severity, collector_id, device_id, raw_event
            )
            SELECT
              event_id, observed_at, ingested_at, source, event_type,
              severity, collector_id, device_id, raw_event
            FROM event_data
            ON CONFLICT (event_id) DO NOTHING
            RETURNING event_id
          )
          INSERT INTO sensor_measurements (
            event_id, observed_at, source, collector_id, device_id,
            metric_name, numeric_value, text_value, unit, source_updated_at
          )
          SELECT
            event.event_id,
            COALESCE(
              NULLIF(measurement.value->>'updated_at', '')::timestamptz,
              event.observed_at
            ),
            event.source,
            event.collector_id,
            event.device_id,
            measurement.key,
            CASE
              WHEN jsonb_typeof(measurement.value) = 'number'
                THEN measurement.value::text::double precision
              WHEN jsonb_typeof(measurement.value) = 'boolean'
                THEN CASE WHEN measurement.value::boolean THEN 1 ELSE 0 END
              WHEN jsonb_typeof(measurement.value->'value') = 'number'
                THEN (measurement.value->>'value')::double precision
              WHEN jsonb_typeof(measurement.value->'value') = 'boolean'
                THEN CASE WHEN (measurement.value->>'value')::boolean THEN 1 ELSE 0 END
              ELSE NULL
            END,
            CASE
              WHEN jsonb_typeof(measurement.value) = 'string'
                THEN measurement.value#>>'{}'
              WHEN jsonb_typeof(measurement.value->'value') = 'string'
                THEN measurement.value->>'value'
              ELSE NULL
            END,
            measurement.value->>'unit',
            NULLIF(measurement.value->>'updated_at', '')::timestamptz
          FROM event_data AS event
          CROSS JOIN LATERAL jsonb_each(event.measurements) AS measurement
          WHERE
            jsonb_typeof(measurement.value) IN ('number', 'boolean', 'string')
            OR jsonb_typeof(measurement.value->'value') IN ('number', 'boolean', 'string')
          ON CONFLICT (event_id, metric_name) DO NOTHING;
        args_mapping: |
          root = [this.string()]