import json
from datetime import datetime
from bson import json_util
from aiokafka import AIOKafkaConsumer
from app.utils.connections import connected_clients
from app.db import database  # ✅ required to use MongoDB

async def consume_tracking_data():
    consumer = AIOKafkaConsumer(
        "vehicle_tracking",
        bootstrap_servers='localhost:9092',
        group_id="tracking-group"
    )
    await consumer.start()

    mongo_gen = database.get_mongo_db()
    mongo = next(mongo_gen)
    collection = mongo["tracking_logs"]

    try:
        async for msg in consumer:
            data = json.loads(msg.value.decode("utf-8"))
            data["received_at"] = datetime.utcnow()

            print("📦 Received from Kafka:", data)

            # ✅ Insert into MongoDB
            collection.insert_one(data)

            # ✅ Broadcast to clients
            for ws in connected_clients.copy():
                try:
                    await ws.send_text(json.dumps(data, default=json_util.default))
                except Exception as e:
                    connected_clients.remove(ws)
                    print(f"❌ Failed to send: {e}")

    finally:
        await consumer.stop()
        try:
            next(mongo_gen)
        except StopIteration:
            pass