#include <eqpalg/utility/async_db_worker.h>

AsyncDbWorker &AsyncDbWorker::instance() {
  static AsyncDbWorker inst;
  return inst;
}

AsyncDbWorker::AsyncDbWorker() {
  worker_ = std::make_unique<std::thread>(&AsyncDbWorker::loop, this);
}

AsyncDbWorker::~AsyncDbWorker() {
  if (running_) {
    drain_and_stop();
  }
}

void AsyncDbWorker::submit(const std::string &rule_id,
                           std::function<void()> task) {
  {
    std::lock_guard<std::mutex> guard(mtx_);
    pending_[rule_id] = std::move(task);  // 去重
  }
  cv_.notify_one();
}

void AsyncDbWorker::drain_and_stop() {
  running_ = false;
  cv_.notify_all();
  if (worker_ && worker_->joinable()) {
    worker_->join();
  }
}

void AsyncDbWorker::loop() {
  while (running_) {
    std::function<void()> task;
    {
      std::unique_lock<std::mutex> lock(mtx_);
      cv_.wait(lock, [this]() { return !pending_.empty() || !running_; });
      if (!running_ && pending_.empty()) break;
      if (!pending_.empty()) {
        auto it = pending_.begin();
        task = std::move(it->second);
        pending_.erase(it);
      }
    }
    if (task) {
      task();
    }
  }
  // 排空剩余任务
  while (true) {
    std::function<void()> task;
    {
      std::lock_guard<std::mutex> guard(mtx_);
      if (pending_.empty()) break;
      auto it = pending_.begin();
      task = std::move(it->second);
      pending_.erase(it);
    }
    if (task) task();
  }
}