#include <sys/event.h>
#include "seobeo/seobeo.h"


void *Seobeo_Web_Edge_Worker(void *vargs)
{
  WorkerArgs *args = vargs;
  struct kevent evlist[64];

  // Each thread creates its own kqueue to avoid race conditions
  int kq = kqueue();
  if (kq < 0) {
    perror("kqueue");
    return NULL;
  }

  // Add server socket to this thread's kqueue
  struct kevent kev = {
    .ident  = args->srv->socket,
    .filter = EVFILT_READ,
    .flags  = EV_ADD,
    .udata  = args->srv
  };
  kevent(kq, &kev, 1, NULL, 0, NULL);

  while (!Seobeo_Web_Server_Is_Stopping())
  {
    struct timespec timeout = {
      .tv_nsec = 250 * 1000 * 1000,
    };
    int ne = kevent(kq, NULL, 0, evlist, 64, &timeout);
    if (ne < 0) {
      if (errno == EINTR) continue;
      perror("kevent");
      continue;
    }

    for (int i = 0; i < ne; i++)
    {
      Seobeo_Handle *h = evlist[i].udata;
      if (h == args->srv)
      {
        // Accept new connections in a loop (for edge-triggered behavior)
        while (1) {
          Seobeo_Handle *cli = Seobeo_Stream_Handle_Server_Accept(args->srv);
          if (!cli) break;

          struct kevent client_kev = {
            .ident  = cli->socket,
            .filter = EVFILT_READ,
            .flags  = EV_ADD | EV_ONESHOT,
            .udata  = cli
          };
          kevent(kq, &client_kev, 1, NULL, 0, NULL);
        }
      } else {
        if (h->is_sse)
        {
          if (evlist[i].flags & EV_EOF)
          {
            Seobeo_Handle_Destroy(h);
            continue;
          }
          int32 read_result = Seobeo_Handle_Read(h);
          if (read_result < 0)
          {
            Seobeo_Handle_Destroy(h);
            continue;
          }
          if (h->read_buffer_len)
            Seobeo_Handle_Consume(h, h->read_buffer_len);
          struct kevent sse_kev = {
            .ident  = h->socket,
            .filter = EVFILT_READ,
            .flags  = EV_ADD | EV_ONESHOT,
            .udata  = h
          };
          kevent(kq, &sse_kev, 1, NULL, 0, NULL);
          continue;
        }

        // Remove from kqueue first
        struct kevent del_kev = {
          .ident  = h->socket,
          .filter = EVFILT_READ,
          .flags  = EV_DELETE,
        };
        kevent(kq, &del_kev, 1, NULL, 0, NULL);

        // macOS handles ordinary requests once, while SSE takes ownership.
        Seobeo_Web_ClientHandle_Request(h, args->cache, FALSE);
        if (h->is_sse)
        {
          struct kevent sse_kev = {
            .ident  = h->socket,
            .filter = EVFILT_READ,
            .flags  = EV_ADD | EV_ONESHOT,
            .udata  = h
          };
          kevent(kq, &sse_kev, 1, NULL, 0, NULL);
        }
        else
        {
          Seobeo_Handle_Destroy(h);
        }
      }
    }
  }

  close(kq);
  return NULL;
}

void  Seobeo_Web_Edge(
    Seobeo_Handle *p_server_handle,
    int            thread_count,
    Seobeo_Cache_Entry *p_html_cache)
{
  pthread_attr_t attr;
  pthread_attr_init(&attr);
  pthread_attr_setstacksize(&attr, 5 * 1024 * 1024); // 5 MB

  pthread_t  threads[thread_count];
  for (int i = 0; i < thread_count; i++)
  {
    WorkerArgs *args = malloc(sizeof(WorkerArgs));
    *args = (WorkerArgs){ p_server_handle, p_html_cache };

    pthread_create(&threads[i], &attr, Seobeo_Web_Edge_Worker, args);
  }
  for (int i = 0; i < thread_count; i++)
  {
    pthread_join(threads[i], NULL);
  }

  pthread_attr_destroy(&attr);
  return;
}