#include "MockDDS.h"
#include <map>
#include <mutex>
#include <thread>
#include <condition_variable>

namespace MockDDS {

struct TopicState {
  std::vector<std::string> queue;
  std::map<SubscriptionId, Subscriber> subs;
  SubscriptionId next_sub_id{1};
  unsigned long seq{0};
};

static std::map<std::string, TopicState> g_topics;
static std::mutex g_mtx;

// Background dispatcher for async publishes
static std::vector<std::tuple<std::string, std::string, unsigned>> g_async_queue;
static std::mutex g_async_mtx;
static std::condition_variable g_async_cv;
static bool g_async_running = false;
static std::thread g_async_thread;

static void async_worker() {
    std::unique_lock<std::mutex> lk(g_async_mtx);
    while (g_async_running) {
        if (g_async_queue.empty()) {
            g_async_cv.wait(lk, [] { return !g_async_queue.empty() || !g_async_running; });
            if (!g_async_running) break;
        }
        auto task = g_async_queue.front();
        g_async_queue.erase(g_async_queue.begin());
        lk.unlock();
        auto &topic = std::get<0>(task);
        auto &msg = std::get<1>(task);
        unsigned delay_ms = std::get<2>(task);
        if (delay_ms) std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms));
        // perform synchronous publish semantics for delivered message
        {
            std::lock_guard<std::mutex> lk2(g_mtx);
            auto &ts = g_topics[topic];
            ts.seq++;
            std::string annotated = std::to_string(ts.seq) + ":" + msg;
            ts.queue.push_back(annotated);
            for (auto &p : ts.subs) {
                auto &cb = p.second;
                if (cb) cb(annotated);
            }
        }
        lk.lock();
    }
}

static void ensure_async_thread() {
    std::lock_guard<std::mutex> lk(g_async_mtx);
    if (!g_async_running) {
        g_async_running = true;
        g_async_thread = std::thread(async_worker);
        // detach worker so test binary exit doesn't require join; worker will exit when g_async_running is false
        g_async_thread.detach();
    }
}

void publish(const std::string &topic, const std::string &msg) {
    std::lock_guard<std::mutex> lk(g_mtx);
    auto &ts = g_topics[topic];
    ts.seq++;
    // annotate payload with simple seq metadata for basic QoS checks
    std::string annotated = std::to_string(ts.seq) + ":" + msg;
    ts.queue.push_back(annotated);
    // notify subscribers synchronously
    for (auto &p : ts.subs) {
        auto &cb = p.second;
        if (cb) cb(annotated);
    }
}

void publish_async(const std::string &topic, const std::string &msg, unsigned delay_ms) {
    ensure_async_thread();
    {
        std::lock_guard<std::mutex> lk(g_async_mtx);
        g_async_queue.emplace_back(topic, msg, delay_ms);
    }
    g_async_cv.notify_one();
}

MockDDS::SubscriptionId subscribe(const std::string &topic, Subscriber cb) {
    std::lock_guard<std::mutex> lk(g_mtx);
    auto &ts = g_topics[topic];
    SubscriptionId id = ts.next_sub_id++;
    ts.subs.emplace(id, std::move(cb));
    return id;
}

void unsubscribe(const std::string &topic, SubscriptionId id) {
    std::lock_guard<std::mutex> lk(g_mtx);
    auto it = g_topics.find(topic);
    if (it == g_topics.end()) return;
    it->second.subs.erase(id);
}

std::string take_last(const std::string &topic) {
    std::lock_guard<std::mutex> lk(g_mtx);
    auto it = g_topics.find(topic);
    if (it == g_topics.end() || it->second.queue.empty()) return std::string();
    return it->second.queue.back();
}

std::vector<std::string> get_queue(const std::string &topic) {
    std::lock_guard<std::mutex> lk(g_mtx);
    auto it = g_topics.find(topic);
    if (it == g_topics.end()) return {};
    return it->second.queue;
}

void clear_all() {
    {
        std::lock_guard<std::mutex> lk(g_mtx);
        g_topics.clear();
    }
    // clear pending async queue; do not join detached worker
    {
        std::lock_guard<std::mutex> lk(g_async_mtx);
        g_async_queue.clear();
    }
    g_async_cv.notify_all();
}

} // namespace MockDDS