#include "tlbmc/scheduler/scheduler.h"

#include <algorithm>

#include <cstddef>

#include <memory>

#include <optional>

#include <thread>

#include <utility>

#include <vector>

#include "absl/container/flat_hash_map.h"

#include "absl/functional/any_invocable.h"

#include "absl/log/log.h"

#include "absl/synchronization/mutex.h"

#include "absl/time/time.h"

#include "boost/asio/executor.hpp"

#include "boost/asio/executor_work_guard.hpp"

#include "boost/asio/steady_timer.hpp"  // NOLINT: boost is commonly used in BMC

#include "time/clock.h"

#include "nlohmann/json.hpp"

namespace milotic_tlbmc {

nlohmann::json TimeInfo::ToJson() const {

  nlohmann::json json;

  json["last_scheduled_time"] = absl::FormatTime(last_scheduled_time);

  json["last_run_time"] = absl::FormatTime(last_run_time);

  // Get the last 10 wait times.

  auto it = wait_times.rbegin();

  auto& wait_times_recent_to_oldest = json["wait_times_recent_to_oldest"];

  wait_times_recent_to_oldest = nlohmann::json::array();

  for (int i = 0; i < 10 && it != wait_times.rend(); ++i, ++it) {

    wait_times_recent_to_oldest.push_back(absl::FormatDuration(*it));

  }

  return json;

}

nlohmann::json Task::ToJson() const {

  nlohmann::json json;

  json["task_id"] = task_id_;

  json["period"] = absl::FormatDuration(period_);

  json["execution_count"] = execution_count_.load();

  {

    absl::MutexLock lock(&time_info_mutex_);

    json["time_info"] = time_info_.ToJson();

  }

  json["average_wait_time"] = absl::FormatDuration(GetAverageWaitTime());

  json["cancelled"] = cancelled_.load();

  json["task_mode"] = task_mode_ == TaskMode::kPeriodic ? "periodic" : "once";

  return json;

}

// Overload constructor for tasks that need to be invoked with an ack callback.

Task::Task(int task_id,

           absl::AnyInvocable<void(absl::AnyInvocable<void()>)> task_fn,

           absl::Duration period, boost::asio::io_context* io,

           absl::Duration task_timeout, TaskMode task_mode,

           ecclesia::Clock* clock)

    : task_id_(task_id),

      task_mode_(task_mode),

      clock_(clock),

      period_(period),

      cancelled_(false),

      task_fn_with_ack_(std::move(task_fn)),

      timer_(std::make_unique<boost::asio::steady_timer>(*io)),

      task_timeout_(task_timeout) {}

absl::Duration Task::GetPeriod() const { return period_.load(); }

absl::Duration Task::GetAverageWaitTime() const {

  absl::MutexLock lock(&time_info_mutex_);

  if (time_info_.wait_times.empty()) {

    // No wait times have been recorded yet. This happens when the task is yet

    // to be scheduled after it's initial run.

    return absl::InfiniteDuration();

  }

  absl::Duration total_wait_time = absl::ZeroDuration();

  for (const auto& wait_time : time_info_.wait_times) {

    total_wait_time += wait_time;

  }

  return total_wait_time / time_info_.wait_times.size();

}

absl::Time Task::GetLastRunTime() const {

  absl::MutexLock lock(&time_info_mutex_);

  return time_info_.last_run_time;

}

void Task::ScheduleNextRun() {

  {

    absl::MutexLock lock(&time_info_mutex_);

    if (clock_ != nullptr) {

      time_info_.last_scheduled_time = clock_->Now();

    }

  }

  {

    absl::MutexLock lock(&timer_mutex_);

    if (cancelled_.load()) {

      return;

    }

    last_execution_duration_ = GetLastRunTime() == absl::InfiniteFuture()

                                   ? absl::ZeroDuration()

                                   : clock_->Now() - GetLastRunTime();

    absl::Duration wait_time = period_ - last_execution_duration_;

    if (wait_time < absl::ZeroDuration()) {

      wait_time = absl::ZeroDuration();

    }

    timer_->expires_after(absl::ToChronoNanoseconds(wait_time));

    timer_->async_wait([weak_self = std::weak_ptr<Task>(shared_from_this())](

                           boost::system::error_code error) {

      if (error == boost::asio::error::operation_aborted) {

        DLOG(ERROR) << "Failed to wait for task to complete with error: "

                    << error.message();

        return;

      }

      std::shared_ptr<Task> task = weak_self.lock();

      if (task == nullptr) {

        DLOG(WARNING) << "Task is cancelled";

        return;

      }

      task->Run();

    });

  }

}

void Task::OnDone() {

  ++execution_count_;

  ScheduleNextRun();

}

// Runs the task and updates the next run time.

void Task::Run() {

  if (cancelled_.load()) {

    DLOG(WARNING) << "Task is cancelled before running!";

    return;

  }

  {

    absl::MutexLock lock(&time_info_mutex_);

    absl::Time now = clock_->Now();

    if (time_info_.last_scheduled_time != absl::InfiniteFuture()) {

      time_info_.wait_times.push_back(now - time_info_.last_scheduled_time);

    }

    time_info_.last_run_time = now;

  }

  {

    absl::MutexLock lock(&task_mutex_);

    if (task_fn_with_ack_ != nullptr) {

      task_fn_with_ack_(

          [weak_self = std::weak_ptr<Task>(shared_from_this())]() {

            std::shared_ptr<Task> task = weak_self.lock();

            if (task == nullptr) {

              DLOG(WARNING) << "Task is destroyed while waiting for ack!";

              return;

            }

            task->OnDone();

          });

      return;

    }

  }

}

// Cancels the task.

void Task::Cancel() {

  cancelled_.store(true);

  absl::MutexLock lock(&timer_mutex_);

  timer_->cancel();

}

void Task::SetPeriod(absl::Duration period) {

  Cancel();

  period_ = period;

  cancelled_.store(false);

  ScheduleNextRun();

}

TaskScheduler::TaskScheduler(ecclesia::Clock* clock, const Options& options)

    : next_task_id_(1),

      clock_(clock),

      io_(std::make_unique<boost::asio::io_context>()),

      work_guard_(io_->get_executor()),

      async_executor_thread_([this]() { io_->run(); }),

      default_task_timeout_(options.default_task_timeout) {

  cleanup_task_id_ = RunAndScheduleAsync(

      [this](absl::AnyInvocable<void()> on_done) {

        CleanupInactiveTasks();

        on_done();

      },

      options.cleanup_tasks_period);

}

TaskScheduler::~TaskScheduler() { Stop(); }

int TaskScheduler::ScheduleAsyncInternal(

    absl::AnyInvocable<void(absl::AnyInvocable<void()>)> task,

    absl::Duration period, std::optional<absl::Duration> timeout,

    ExecutionMode execution_mode) {

  if (task == nullptr) {

    LOG(ERROR) << "Task is null";

    return -1;

  }

  if (stop_.load()) {

    LOG(ERROR) << "TaskScheduler is stopped, cannot schedule async task";

    return -1;

  }

  if (period == absl::ZeroDuration()) {

    LOG(ERROR) << "Task period is zero, cannot schedule async recurring task";

    return -1;

  }

  int task_id = next_task_id_++;

  absl::Duration task_timeout = timeout.value_or(default_task_timeout_);

  auto task_ptr =

      std::make_shared<Task>(task_id, std::move(task), period, io_.get(),

                             task_timeout, TaskMode::kPeriodic, clock_);

  if (execution_mode == ExecutionMode::kRunImmediately) {

    task_ptr->Run();

  } else if (execution_mode == ExecutionMode::kRunAfterScheduling) {

    task_ptr->ScheduleNextRun();

  }

  absl::MutexLock lock(&async_tasks_mutex_);

  id_to_async_tasks_.emplace(task_id, task_ptr);

  return task_id;

}

int TaskScheduler::ScheduleAsync(

    absl::AnyInvocable<void(absl::AnyInvocable<void()>)> task,

    absl::Duration period, std::optional<absl::Duration> timeout) {

  return ScheduleAsyncInternal(std::move(task), period, timeout,

                               ExecutionMode::kRunAfterScheduling);

}

int TaskScheduler::RunAndScheduleAsync(

    absl::AnyInvocable<void(absl::AnyInvocable<void()>)> task,

    absl::Duration period, std::optional<absl::Duration> timeout) {

  return ScheduleAsyncInternal(std::move(task), period, timeout,

                               ExecutionMode::kRunImmediately);

}

int TaskScheduler::ScheduleOneShotAsync(

    absl::AnyInvocable<void(absl::AnyInvocable<void()>)> task,

    absl::Duration period, std::optional<absl::Duration> timeout) {

  if (task == nullptr) {

    LOG(ERROR) << "Task is null";

    return -1;

  }

  if (stop_.load()) {

    LOG(ERROR) << "TaskScheduler is stopped, cannot run and schedule task";

    return -1;

  }

  if (period == absl::ZeroDuration()) {

    LOG(ERROR) << "Task period is zero, cannot schedule task";

    return -1;

  }

  int task_id = next_task_id_++;

  absl::Duration task_timeout = timeout.value_or(default_task_timeout_);

  auto oneshot_on_done = [this, task_id]() { Cancel(task_id); };

  auto task_ptr = std::make_shared<Task>(

      task_id,

      [task = std::move(task), oneshot_on_done = std::move(oneshot_on_done)](

          absl::AnyInvocable<void()> on_done) mutable {

        task(std::move(oneshot_on_done));

      },

      period, io_.get(), task_timeout, TaskMode::kOnce, clock_);

  {

    absl::MutexLock lock(&async_tasks_mutex_);

    id_to_async_tasks_.emplace(task_id, task_ptr);

  }

  task_ptr->ScheduleNextRun();

  return task_id;

}

absl::flat_hash_map<int, std::shared_ptr<Task>> TaskScheduler::GetSnapshot()

    const {

  absl::flat_hash_map<int, std::shared_ptr<Task>> snapshot;

  {

    absl::MutexLock lock(&async_tasks_mutex_);

    for (const auto& [task_id, task] : id_to_async_tasks_) {

      snapshot.emplace(task_id, task);

    }

  }

  return snapshot;

}

size_t TaskScheduler::GetAllUserTaskCount() const {

  absl::MutexLock lock(&async_tasks_mutex_);

  // We always have one internal task scheduled for cleanup.

  return id_to_async_tasks_.size() - 1;

}

size_t TaskScheduler::GetAllTaskCount() const {

  absl::MutexLock lock(&async_tasks_mutex_);

  return id_to_async_tasks_.size();

}

void TaskScheduler::UpdateTaskPeriod(int task_id, absl::Duration period) {

  if (stop_.load()) {

    LOG(WARNING) << "TaskScheduler is stopped, cannot update task period: "

                 << task_id;

    return;

  }

  if (period == absl::ZeroDuration()) {

    LOG(ERROR) << "Cannot set async task period to zero";

    return;

  }

  absl::MutexLock lock(&async_tasks_mutex_);

  auto it = id_to_async_tasks_.find(task_id);

  if (it != id_to_async_tasks_.end()) {

    if (it->second->GetPeriod() == period) {

      return;

    }

    LOG(WARNING) << "Updating task period for task: " << task_id << " from "

                 << it->second->GetPeriod() << " to " << period;

    it->second->SetPeriod(period);

  }

}

void TaskScheduler::Cancel(int task_id) {

  if (stop_.load()) {

    LOG(WARNING) << "TaskScheduler is stopped, cannot cancel task: " << task_id;

    return;

  }

  // Cancel the async task.

  {

    absl::MutexLock lock(&async_tasks_mutex_);

    auto it = id_to_async_tasks_.find(task_id);

    if (it != id_to_async_tasks_.end()) {

      it->second->Cancel();

      // Remove the task from the tasks map.

      id_to_async_tasks_.erase(it);

    }

  }

}

void TaskScheduler::CancelAll() {

  if (stop_.load()) {

    LOG(ERROR) << "TaskScheduler is stopped, cannot cancel all tasks";

    return;

  }

  {

    absl::MutexLock lock(&async_tasks_mutex_);

    for (const auto& [task_id, task] : id_to_async_tasks_) {

      task->Cancel();

    }

    id_to_async_tasks_.clear();

  }

}

void TaskScheduler::CleanupInactiveTasks() {

  if (stop_.load()) {

    return;

  }

  // Copy id_to_async_tasks_ to a vector to avoid holding the lock while

  // cancelling tasks.

  std::vector<std::pair<int, std::shared_ptr<Task>>> id_to_async_tasks_copy;

  // Task ids of Tasks to cancel.

  std::vector<int> tasks_to_cancel;

  {

    absl::MutexLock lock(&async_tasks_mutex_);

    id_to_async_tasks_copy = std::vector<std::pair<int, std::shared_ptr<Task>>>(

        id_to_async_tasks_.begin(), id_to_async_tasks_.end());

  }

  for (const auto& [task_id, task] : id_to_async_tasks_copy) {

    absl::Time last_run_time = task->GetLastRunTime();

    if (last_run_time == absl::InfiniteFuture()) {

      continue;

    }

    absl::Duration time_since_last_run = clock_->Now() - last_run_time;

    if (time_since_last_run > task->GetTaskTimeout()) {

      tasks_to_cancel.push_back(task_id);

    }

  }

  for (int task_id : tasks_to_cancel) {

    LOG(ERROR) << "Cleanup inactive task: " << task_id

               << " as it has not run for the last " << default_task_timeout_;

    Cancel(task_id);

  }

}

void TaskScheduler::Stop() {

  if (stop_.load()) {

    // Already stopped.

    return;

  }

  CancelAll();

  stop_.store(true);

  if (async_executor_thread_.joinable()) {

    io_->stop();

    async_executor_thread_.join();

  }

}

size_t TaskScheduler::GetSchedulingAccuracyPercentage() const {

  // Copy id_to_async_tasks_ to avoid holding the lock while calculating

  // accuracy.

  absl::flat_hash_map<int, std::shared_ptr<Task>> id_to_async_tasks_copy;

  {

    absl::MutexLock lock(&async_tasks_mutex_);

    id_to_async_tasks_copy = id_to_async_tasks_;

  }

  if (id_to_async_tasks_copy.empty()) {

    // If there are no tasks, we consider the scheduler to be accurate.

    return 100;

  }

  double total_inaccuracy = 0.0;

  for (const auto& [task_id, task] : id_to_async_tasks_copy) {

    absl::Duration average_wait_time = task->GetAverageWaitTime();

    double task_inaccuracy;

    if (average_wait_time == absl::InfiniteDuration()) {

      // No wait times have been recorded yet. This happens when the task is yet

      // to be scheduled after it's initial run.

      // We don't want to penalize the scheduler for this case. So we consider

      // the inaccuracy to be 0.

      continue;

    }

    absl::Duration period = task->GetPeriod();

    if (average_wait_time <= period) {

      task_inaccuracy = 0;

    } else {

      // Inaccuracy based on how much *later* than the period it waited

      absl::Duration lateness = average_wait_time - period;

      DLOG(INFO) << "Task " << task_id << " lateness: " << lateness;

      // Define how inaccuracy scales with lateness.

      task_inaccuracy = (absl::ToDoubleNanoseconds(lateness) /

                         absl::ToDoubleNanoseconds(period)) *

                        100.0;

      // Ensure inaccuracy is not negative (shouldn't happen here but for

      // safety)

      task_inaccuracy = std::max(0.0, std::min(100.0, task_inaccuracy));

    }

    total_inaccuracy += task_inaccuracy;

    DLOG(INFO) << "Task " << task_id << " has period: " << period

               << " average wait time: " << average_wait_time

               << " with inaccuracy: " << task_inaccuracy

               << "% (based on being late)";

  }

  size_t overall_inaccuracy_percentage = 0;

  overall_inaccuracy_percentage = static_cast<size_t>(

      total_inaccuracy / static_cast<double>(id_to_async_tasks_copy.size()));

  size_t overall_accuracy_percentage = 100 - overall_inaccuracy_percentage;

  DLOG(INFO) << "Overall accuracy percentage: " << overall_accuracy_percentage;

  return overall_accuracy_percentage;

}

nlohmann::json TaskScheduler::ToJson() const {

  nlohmann::json json;

  json["scheduling_accuracy_percentage"] = GetSchedulingAccuracyPercentage();

  absl::MutexLock lock(&async_tasks_mutex_);

  for (const auto& [task_id, task] : id_to_async_tasks_) {

    if (task_id == cleanup_task_id_) {

      json["scheduler_tasks"].push_back(task->ToJson());

      continue;

    }

    json["user_tasks"].push_back(task->ToJson());

  }

  json["user_async_task_count"] = id_to_async_tasks_.size() - 1;

  json["total_async_task_count"] = id_to_async_tasks_.size();

  return json;

}

}  // namespace milotic_tlbmc