const amqp = require("amqplib");
const express = require("express");
const { createClient } = require("redis");
const { MongoClient } = require("mongodb");

const RMQ_HOST = process.env.RMQ_HOST || "localhost";
const RMQ_PORT = parseInt(process.env.RMQ_PORT || "5672");
const RMQ_USER = process.env.RMQ_USER || "guest";
const RMQ_PASS = process.env.RMQ_PASS || "guest";
const RMQ_QUEUE = process.env.RMQ_QUEUE || "iot-events";

const REDIS_HOST = process.env.REDIS_HOST || "localhost";
const REDIS_PORT = parseInt(process.env.REDIS_PORT || "6379");
const REDIS_PASS = process.env.REDIS_PASS || undefined;

const MONGO_URI = process.env.MONGO_URI || "mongodb://localhost:27017/iot";
const MONGO_DB = process.env.MONGO_DB || "iot";

let consumed = 0, errors = 0;
let redis, mongo, mongoColl;

async function getRedis() {
  if (!redis) {
    redis = createClient({ socket: { host: REDIS_HOST, port: REDIS_PORT }, password: REDIS_PASS });
    redis.on("error", () => {});
    await redis.connect();
    console.log("Redis connected");
  }
  return redis;
}

async function getMongo() {
  if (!mongo) {
    mongo = new MongoClient(MONGO_URI);
    await mongo.connect();
    mongoColl = mongo.db(MONGO_DB).collection("iot_events");
    await mongoColl.createIndex({ timestamp: 1 }, { expireAfterSeconds: 604800 });
    console.log("MongoDB connected");
  }
  return mongoColl;
}

async function store(event) {
  const r = await getRedis();
  await r.hSet("iot:latest", event.sensor_id, JSON.stringify({
    value: event.value, unit: event.unit, location: event.location, timestamp: event.timestamp,
  }));
  await r.hIncrBy("iot:counters", event.device_type, 1);
  await r.zAdd("iot:recent", [{ score: 0, value: JSON.stringify(event) }]);
  await r.zRemRangeByRank("iot:recent", 0, -1001);
  const coll = await getMongo();
  await coll.insertOne({
    event_id: event.event_id, sensor_id: event.sensor_id,
    device_type: event.device_type, location: event.location,
    value: event.value, unit: event.unit, timestamp: new Date(event.timestamp),
  });
}

async function start() {
  const conn = await amqp.connect(`amqp://${RMQ_USER}:${RMQ_PASS}@${RMQ_HOST}:${RMQ_PORT}`);
  const ch = await conn.createChannel();
  await ch.assertQueue(RMQ_QUEUE, { durable: true });
  ch.prefetch(10);
  await ch.consume(RMQ_QUEUE, async (msg) => {
    try {
      if (msg) {
        const event = JSON.parse(msg.content.toString());
        await store(event);
        consumed++;
        ch.ack(msg);
        if (consumed % 10 === 0) console.log(`Consumed ${consumed} events`);
      }
    } catch (e) {
      errors++;
      console.error("Error:", e.message);
    }
  });
  console.log(`Consumer listening on queue: ${RMQ_QUEUE}`);
}

const app = express();
app.get("/", (req, res) => res.json({ service: "iot-consumer", status: "running", queue: RMQ_QUEUE, consumed, errors }));
app.get("/health", (req, res) => res.json({ status: "ok" }));
app.listen(process.env.PORT || 3000, () => { console.log("Consumer started"); start(); });