"""
IoT Consumer — Flask app that reads IoT events from Kafka
and inserts them into ClickHouse.

Env vars (set by Terraform):
  KAFKA_BROKERS     — Kafka bootstrap servers
  KAFKA_TOPIC       — topic name
  KAFKA_USERNAME    — SASL username
  KAFKA_PASSWORD    — SASL password
  KAFKA_GROUP_ID    — consumer group id
  CH_HOST           — ClickHouse host
  CH_PORT           — ClickHouse port (default 8123)
  CH_USER           — ClickHouse user
  CH_PASSWORD       — ClickHouse password
  CH_DATABASE       — ClickHouse database name
"""

import json
import os
import threading

from flask import Flask, jsonify
from kafka import KafkaConsumer
import clickhouse_connect

app = Flask(__name__)

# --- Config ---
KAFKA_BROKERS = os.environ["KAFKA_BROKERS"].split(",")
KAFKA_TOPIC = os.environ["KAFKA_TOPIC"]
KAFKA_USERNAME = os.environ["KAFKA_USERNAME"]
KAFKA_PASSWORD = os.environ["KAFKA_PASSWORD"]
KAFKA_GROUP_ID = os.environ.get("KAFKA_GROUP_ID", "iot-consumer-group")

CH_HOST = os.environ["CH_HOST"]
CH_PORT = int(os.environ.get("CH_PORT", 8123))
CH_USER = os.environ["CH_USER"]
CH_PASSWORD = os.environ["CH_PASSWORD"]
CH_DATABASE = os.environ["CH_DATABASE"]

events_consumed = 0
errors_count = 0
ch_client = None


def get_ch_client():
    global ch_client
    if ch_client is None:
        ch_client = clickhouse_connect.get_client(
            host=CH_HOST,
            port=CH_PORT,
            username=CH_USER,
            password=CH_PASSWORD,
            database=CH_DATABASE,
        )
        # Create table if not exists
        ch_client.command("""
            CREATE TABLE IF NOT EXISTS iot_events (
                event_id String,
                sensor_id String,
                device_type String,
                location String,
                value Float64,
                unit String,
                timestamp DateTime64(3)
            ) ENGINE = MergeTree()
            ORDER BY (timestamp, sensor_id)
        """)
        print("[CONSUMER] ClickHouse table ready")
    return ch_client


def insert_batch(events):
    if not events:
        return
    rows = [[
        e["event_id"],
        e["sensor_id"],
        e["device_type"],
        e["location"],
        e["value"],
        e["unit"],
        e["timestamp"],
    ] for e in events]
    try:
        client = get_ch_client()
        client.insert("iot_events", rows,
            column_names=["event_id", "sensor_id", "device_type", "location", "value", "unit", "timestamp"])
    except Exception as e:
        print(f"[CONSUMER] Insert error: {e}")
        raise


def consume_loop():
    global events_consumed, errors_count

    consumer = KafkaConsumer(
        KAFKA_TOPIC,
        bootstrap_servers=KAFKA_BROKERS,
        security_protocol="SASL_PLAINTEXT",
        sasl_mechanism="PLAIN",
        sasl_plain_username=KAFKA_USERNAME,
        sasl_plain_password=KAFKA_PASSWORD,
        group_id=KAFKA_GROUP_ID,
        auto_offset_reset="earliest",
        value_deserializer=lambda m: json.loads(m.decode("utf-8")),
        max_poll_records=20,
    )

    print(f"[CONSUMER] Listening on topic '{KAFKA_TOPIC}'...")

    batch = []
    for message in consumer:
        try:
            event = message.value
            batch.append(event)
            events_consumed += 1

            if len(batch) >= 10:
                insert_batch(batch)
                print(f"[CONSUMER] Inserted {len(batch)} events into ClickHouse")
                batch = []
        except Exception as e:
            errors_count += 1
            print(f"[CONSUMER] Error: {e}")


@app.route("/")
def index():
    query = "SELECT count() FROM iot_events"
    try:
        client = get_ch_client()
        total = client.query(query).result_rows[0][0]
    except Exception:
        total = 0

    return jsonify({
        "service": "iot-consumer",
        "status": "running",
        "kafka_topic": KAFKA_TOPIC,
        "events_consumed": events_consumed,
        "events_in_db": total,
        "errors": errors_count,
    })


@app.route("/health")
def health():
    return jsonify({"status": "ok", "events_consumed": events_consumed})


if __name__ == "__main__":
    threading.Thread(target=consume_loop, daemon=True).start()
    port = int(os.environ.get("PORT", 5000))
    app.run(host="0.0.0.0", port=port)