#include "seobeo/seobeo.h"

#include <assert.h>
#include <fcntl.h>
#include <stdatomic.h>
#include <stdio.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>

/* Counter incremented by the detach callback. */
static _Atomic int g_detach_count = 0;
static Seobeo_SSE_Stream *g_detach_stream = NULL;

static void test_detach_cb(Seobeo_SSE_Stream *p_stream, void *ctx)
{
  (void)ctx;
  g_detach_stream = p_stream;
  atomic_fetch_add(&g_detach_count, 1);
}

static void read_expected(int socket_fd, const char *expected)
{
  size_t expected_length = strlen(expected);
  size_t received = 0;
  char buffer[8192] = {0};
  assert(expected_length < sizeof(buffer));

  while (received < expected_length)
  {
    ssize_t amount = recv(
        socket_fd,
        buffer + received,
        expected_length - received,
        0);
    assert(amount > 0);
    received += (size_t)amount;
  }
  assert(received == expected_length);
  assert(memcmp(buffer, expected, expected_length) == 0);
}

static void route_handler(
    Seobeo_SSE_Stream *p_stream,
    Seobeo_Request_Entry *p_request,
    Dowa_Arena *p_arena)
{
  (void)p_stream;
  (void)p_request;
  (void)p_arena;
}

static void post_route_handler(
    Seobeo_SSE_Stream *p_stream,
    Seobeo_Request_Entry *p_request,
    Dowa_Arena *p_arena)
{
  (void)p_stream;
  (void)p_request;
  (void)p_arena;
}

static void initialize_handle(
    Seobeo_Handle *p_handle,
    int socket_fd,
    uint8 *write_buffer,
    uint32 write_capacity)
{
  memset(p_handle, 0, sizeof(*p_handle));
  p_handle->socket = socket_fd;
  p_handle->connected = TRUE;
  p_handle->write_buffer = write_buffer;
  p_handle->write_buffer_capacity = write_capacity;
  atomic_init(&p_handle->destroyed, FALSE);
}

int main(void)
{
  const char *headers =
      "HTTP/1.1 200 OK\r\n"
      "Content-Type: text/event-stream\r\n"
      "Cache-Control: no-cache\r\n"
      "Connection: keep-alive\r\n"
      "X-Accel-Buffering: no\r\n"
      "\r\n";
  int sockets[2];
  assert(socketpair(AF_UNIX, SOCK_STREAM, 0, sockets) == 0);

  uint8 write_buffer[4096] = {0};
  Seobeo_Handle handle;
  initialize_handle(
      &handle,
      sockets[0],
      write_buffer,
      (uint32)sizeof(write_buffer));
  Seobeo_SSE_Stream stream = {0};
  assert(Seobeo_SSE_Start(&stream, &handle) == 0);
  assert(Seobeo_SSE_Is_Open(&stream));
  read_expected(sockets[1], headers);

  Seobeo_SSE_Event event = {
    .event = "deployment",
    .id = "42",
    .data = "started\r\nrunning",
    .retry_ms = 1500,
  };
  assert(Seobeo_SSE_Send(&stream, &event) == 0);
  read_expected(
      sockets[1],
      "id: 42\n"
      "event: deployment\n"
      "retry: 1500\n"
      "data: started\n"
      "data: running\n"
      "\n");

  assert(Seobeo_SSE_Send_Comment(
      &stream,
      "heartbeat\nstill here") == 0);
  read_expected(
      sockets[1],
      ": heartbeat\n"
      ": still here\n"
      "\n");

  int flags = fcntl(sockets[0], F_GETFL, 0);
  assert(flags >= 0);
  assert(fcntl(sockets[0], F_SETFL, flags | O_NONBLOCK) == 0);
  char filler[4096];
  memset(filler, 'z', sizeof(filler));
  while (send(sockets[0], filler, sizeof(filler), 0) > 0)
  {
  }
  assert(errno == EAGAIN || errno == EWOULDBLOCK);
  memcpy(handle.write_buffer, "older", 5);
  handle.write_buffer_len = 5;
  assert(Seobeo_SSE_Send_Data(&stream, "queued") == 1);
  while (recv(sockets[1], filler, sizeof(filler), MSG_DONTWAIT) > 0)
  {
  }
  assert(errno == EAGAIN || errno == EWOULDBLOCK);
  assert(Seobeo_SSE_Flush(&stream) == 0);
  read_expected(sockets[1], "olderdata: queued\n\n");

  event.event = "bad\nevent";
  assert(Seobeo_SSE_Send(&stream, &event) == SEOBEO_SSE_INVALID_EVENT);
  event.event = NULL;
  event.id = "bad\rid";
  assert(Seobeo_SSE_Send(&stream, &event) == SEOBEO_SSE_INVALID_EVENT);

  char oversized[4097];
  memset(oversized, 'x', sizeof(oversized) - 1);
  oversized[sizeof(oversized) - 1] = '\0';
  assert(Seobeo_SSE_Send_Data(
      &stream,
      oversized) == SEOBEO_SSE_EVENT_TOO_LARGE);

  Seobeo_SSE_Close(&stream);
  assert(!Seobeo_SSE_Is_Open(&stream));
  assert(Seobeo_SSE_Send_Data(&stream, "closed") < 0);
  close(sockets[0]);
  close(sockets[1]);

  assert(socketpair(AF_UNIX, SOCK_STREAM, 0, sockets) == 0);
  initialize_handle(
      &handle,
      sockets[0],
      write_buffer,
      (uint32)sizeof(write_buffer));
  memset(&stream, 0, sizeof(stream));
  assert(Seobeo_SSE_Start(&stream, &handle) == 0);
  read_expected(sockets[1], headers);
  close(sockets[1]);
  assert(Seobeo_SSE_Send_Data(&stream, "disconnect") < 0);
  close(sockets[0]);

  assert(socketpair(AF_UNIX, SOCK_STREAM, 0, sockets) == 0);
  initialize_handle(
      &handle,
      sockets[0],
      write_buffer,
      (uint32)sizeof(write_buffer));
  Seobeo_SSE_Stream *p_managed = Seobeo_SSE_Server_Attach(
      &handle,
      "/events/builds");
  assert(p_managed);
  assert(handle.is_sse);
  assert(Seobeo_SSE_Retain(p_managed));
  read_expected(sockets[1], headers);
  assert(Seobeo_SSE_Send_Data(p_managed, "after-handler") == 0);
  read_expected(sockets[1], "data: after-handler\n\n");
  Seobeo_SSE_Server_Detach_Handle(&handle);
  assert(!handle.is_sse);
  assert(!Seobeo_SSE_Is_Open(p_managed));
  Seobeo_SSE_Release(p_managed);
  Seobeo_SSE_Server_Destroy();
  close(sockets[0]);
  close(sockets[1]);

  /*
   * Detach callback test: verify callback fires after handle detach and
   * that the callback is called outside g_sse_mutex (no deadlock).
   */
  {
    int cbs[2];
    assert(socketpair(AF_UNIX, SOCK_STREAM, 0, cbs) == 0);
    uint8 cb_write_buf[4096] = {0};
    Seobeo_Handle cb_handle;
    initialize_handle(&cb_handle, cbs[0], cb_write_buf, sizeof(cb_write_buf));
    Seobeo_SSE_Stream *p_cb_stream = Seobeo_SSE_Server_Attach(&cb_handle, "/cb");
    assert(p_cb_stream);
    assert(Seobeo_SSE_Retain(p_cb_stream)); /* extra reference */

    /* Register detach callback. */
    atomic_store(&g_detach_count, 0);
    g_detach_stream = NULL;
    Seobeo_SSE_Set_Detach_Callback(p_cb_stream, test_detach_cb, NULL);

    /* Detach: fires callback outside the mutex. */
    Seobeo_SSE_Server_Detach_Handle(&cb_handle);

    assert(atomic_load(&g_detach_count) == 1);
    assert(g_detach_stream == p_cb_stream);
    assert(!Seobeo_SSE_Is_Open(p_cb_stream));

    /* Second detach: no stream found, callback NOT called again. */
    Seobeo_SSE_Server_Detach_Handle(&cb_handle);
    assert(atomic_load(&g_detach_count) == 1);

    /* Clear the callback and verify it isn't called on a subsequent release. */
    Seobeo_SSE_Set_Detach_Callback(p_cb_stream, NULL, NULL);
    Seobeo_SSE_Release(p_cb_stream); /* release extra reference */
    Seobeo_SSE_Server_Destroy();
    close(cbs[0]);
    close(cbs[1]);
    printf("  detach callback fires once, outside mutex             PASS\n");
  }

  Seobeo_Router_Init();
  Seobeo_Router_Register_SSE("/events/:topic", route_handler);
  Seobeo_Router_Register_SSE("/events/:topic/fixed", route_handler);
  Seobeo_Router_Register_SSE_Method(
      "POST",
      "/events/:topic",
      post_route_handler);
  Dowa_Arena *p_arena = Dowa_Arena_Create(4096);  Seobeo_Request_Entry *p_request = NULL;
  Seobeo_SSE_Handler handler = Seobeo_Router_Find_SSE_Handler(
      "GET",
      "/events/builds",
      &p_request,
      p_arena);
  assert(handler == route_handler);
  Seobeo_Request_Entry *p_topic = Dowa_HashMap_Get_Ptr(
      p_request,
      ":topic");
  assert(p_topic && strcmp(p_topic->value, "builds") == 0);
  Seobeo_Request_Entry *p_post_request = NULL;
  Dowa_HashMap_Push_Arena(
      p_post_request,
      "Body",
      "deploy=release",
      p_arena);
  assert(Seobeo_Router_Find_SSE_Handler(
      "POST",
      "/events/builds",
      &p_post_request,
      p_arena) == post_route_handler);
  p_topic = Dowa_HashMap_Get_Ptr(p_post_request, ":topic");
  Seobeo_Request_Entry *p_body = Dowa_HashMap_Get_Ptr(
      p_post_request,
      "Body");
  assert(p_topic && strcmp(p_topic->value, "builds") == 0);
  assert(p_body && strcmp(p_body->value, "deploy=release") == 0);
  Seobeo_Request_Entry *p_wrong_method = NULL;
  assert(Seobeo_Router_Find_SSE_Handler(
      "PUT",
      "/events/builds",
      &p_wrong_method,
      p_arena) == NULL);
  assert(Dowa_HashMap_Get_Ptr(p_wrong_method, ":topic") == NULL);
  Seobeo_Request_Entry *p_unmatched = NULL;
  assert(Seobeo_Router_Find_SSE_Handler(
      "GET",
      "/events/builds/other",
      &p_unmatched,
      p_arena) == NULL);
  assert(Dowa_HashMap_Get_Ptr(p_unmatched, ":topic") == NULL);
  Dowa_Arena_Free(p_arena);
  Seobeo_Router_Destroy();

  printf("Seobeo SSE tests passed\n");
  return 0;
}