import textwrap
import asyncio
import json
import logging
import os
import re
import time
from typing import cast
from weakref import WeakSet

import aiomqtt
import uvicorn  # v 2.0.0
from sse_starlette.sse import EventSourceResponse
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.requests import Request
from starlette_exporter import PrometheusMiddleware, handle_metrics

from convert import converters
from victoriametrics_write import MetricsWriter

logging.basicConfig(level=logging.INFO)
log = logging.getLogger()
logging.getLogger('httpx').setLevel(logging.WARNING)

debugListeners = WeakSet()


def broadcastToDebugListeners(event: dict):
    j = json.dumps(event)
    for lis in debugListeners:
        lis.put_nowait(j)


async def debugEvents(request: Request) -> EventSourceResponse:
    q = asyncio.Queue()
    debugListeners.add(q)

    async def gen():
        try:
            while True:
                yield await q.get()
        except asyncio.CancelledError:
            debugListeners.discard(q)

    resp = EventSourceResponse(gen())
    return resp


def simplifyMqttMessage(mqttMessage: aiomqtt.Message) -> dict:
    message = {
        't': round(time.time(), 3),
        'topic': mqttMessage.topic.value,
        'payload': mqttMessage.payload,
    }
    if isinstance(message['payload'], bytes):
        message['payload'] = message['payload'].decode('utf-8')

    return message


async def mqttTask(metrics: MetricsWriter):
    try:
        client = aiomqtt.Client('mqtt2.bigasterisk.com', identifier="mqtt-exporter")
        async with client:
            await client.subscribe('#')
            async for mqttMessage in client.messages:
                try:
                    onMqttMessage(metrics, mqttMessage)
                except Exception as e:
                    log.warning("mqtt message (topic %r) failed: %r", mqttMessage.topic.value, textwrap.shorten(repr(e), width=200))
    except Exception:
        log.error("mqtt task failed", exc_info=True)
        os.abort()


def onMqttMessage(metrics, mqttMessage):
    message = simplifyMqttMessage(mqttMessage)

    metricEvent = tryConverters(message)
    if metricEvent:
        metrics.write(message['t'], metricEvent)
        broadcastToDebugListeners({'message': message, 'metricEvent': metricEvent})


def tryConverters(message) -> dict | None:
    metricEvent = None
    matchedPats = []
    for converter in converters:
        if not (m := re.search(converter.topic_pattern, message['topic'])):
            continue
        if not matchedPats:
            try:
                metricEvent = converter(message, cast(tuple[str], m.groups()))
            except Exception:
                log.error("Error converting mqtt message: topic pattern = %r", converter.topic_pattern, exc_info=True)
            matchedPats.append(converter.topic_pattern)

    if len(matchedPats) > 1:
        log.warning("Multiple patterns matched: %s", ", ".join(matchedPats))

    return metricEvent


def main():
    metrics = MetricsWriter()
    loop = asyncio.new_event_loop()
    app = Starlette(on_startup=[lambda: loop.create_task(mqttTask(metrics))], debug=True, routes=[
        Route('/api/debugEvents', debugEvents),
    ])
    app.add_middleware(PrometheusMiddleware, app_name='mqtt_metrics')
    app.add_route("/metrics", handle_metrics)
    config = uvicorn.Config(app=app, host='0.0.0.0', port=8001, loop='asyncio')
    server = uvicorn.Server(config)
    loop.run_until_complete(server.serve())


main()