#include "cmCTestJobServerWindows.h"

#include <algorithm>

#include <windows.h>

void cmCTestJobServerWindows::_wait_for_semaphore(uv_idle_t* handle)
{
  cmCTestJobServerWindows* self =
    reinterpret_cast<cmCTestJobServerWindows*>(handle->data);

  // Repeatedly wait until we get a timeout
  size_t slots = 0;
  while (true) {
    DWORD r = WaitForSingleObject(self->semaphore, 0);

    if (r == WAIT_OBJECT_0) {
      slots++;
    } else if (r == WAIT_TIMEOUT) {
      break;
    } else {
      if (self->FatalCallback.has_value()) {
        self->FatalCallback.value()();
      }
      return;
    }
  }

  if (slots > 0) {
    self->acquiredTokens += slots;
    self->_process_queue();
  }
}

void cmCTestJobServerWindows::_process_queue()
{
  size_t slots = this->acquiredTokens - this->usedTokens;
  size_t started = 0;
  while (slots > 0 && !queue.empty()) {
    size_t slotsNeeded = std::min(queue.front().slots, maxSlots);
    if (slotsNeeded > slots) {
      break;
    }

    cmCTestJobServerClient::Task task = std::move(queue.front());
    queue.pop_front();
    slots -= task.slots;
    this->usedTokens += task.slots;
    task.run(task.slots);

    started++;
  }

  if (queue.empty()) {
    // Be a friend and release any tokens we didn't use
    _write(this->acquiredTokens - this->usedTokens);
  } else if (started == 0 && this->usedTokens == 0) {
    // If no jobs were started, and we're not running any jobs, then we start
    // the next job with as many tokens as we have available
    cmCTestJobServerClient::Task task = std::move(queue.front());
    queue.pop_front();
    this->usedTokens += slots;
    task.run(slots);
  }
}

void cmCTestJobServerWindows::_write(size_t slots)
{
  DWORD r = ReleaseSemaphore(this->semaphore, slots, nullptr);
  if (r == 0) {
    if (this->FatalCallback.has_value()) {
      this->FatalCallback.value()();
    }
  }

  this->acquiredTokens -= slots;
}

bool cmCTestJobServerWindows::Connect(uv_loop_t* loop, const wchar_t* name)
{
  this->Loop = loop;
  this->semaphore = OpenSemaphore(SEMAPHORE_ALL_ACCESS, FALSE, name);
  if (this->semaphore == nullptr) {
    return false;
  }

  // Start idle task to wait for semaphore
  uv_idle_init(loop, &this->idle);
  this->idle.data = this;
  uv_idle_start(&this->idle, _wait_for_semaphore);

  return true;
}

bool cmCTestJobServerWindows::Connect(uv_loop_t* loop)
{
  const cm::string_view prefix = { "--jobserver-auth=" };

  cm::optional<std::string> maxJobs = cm::nullopt;
  cm::optional<std::string> auth = cm::nullopt;

  std::string makeflags;
  if (!cmSystemTools::GetEnv("MAKEFLAGS", makeflags)) {
    return false;
  }

  std::vector<std::string> args;
  cmSystemTools::ParseUnixCommandLine(makeflags.c_str(), args);
  for (const std::string& arg : args) {
    cm::string_view arg_view(arg);

    if (cmHasLiteralPrefix(arg_view, "-j")) {
      auth = arg_view.substr(cmStrLen("-j"));
    } else if (cmHasPrefix(arg_view, prefix)) {
      auth = cmTrimWhitespace(arg_view.substr(prefix.length()));
      break;
    }
  }

  if (maxJobs) {
    try {
      this->maxSlots = std::min(this->maxSlots, std::stoul(*maxJobs));
    } catch (...) {
    }
  }

  if (!auth) {
    return false;
  }

  return Connect(loop, auth->c_str());
}

void cmCTestJobServerWindows::Enqueue(std::function<void(size_t)> task,
                                      size_t slots)
{
  this->queue.push_back({ task, slots });
}

void cmCTestJobServerWindows::Release(size_t slots)
{
  this->usedTokens -= slots;
  _write(slots);
}

void cmCTestJobServerWindows::Close(bool /*force*/)
{
  _write(this->acquiredTokens);
  CloseHandle(this->semaphore);
  this->semaphore = nullptr;
  uv_idle_stop(&this->idle);
}

void cmCTestJobServerWindows::SetFatalCallback(std::function<void()> f)
{
  this->FatalCallback = f;
}
