/*
 * AIDE (Advanced Intrusion Detection Environment)
 *
 * Copyright (C) 2022-2025 Hannes von Haugwitz
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License as
 * published by the Free Software Foundation; either version 2 of the
 * License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 * General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License along
 * with this program; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 */

#include "config.h"
#include <stdlib.h>
#include <stdbool.h>

#include <pthread.h>

#include "queue.h"
#include "log.h"
#include "util.h"

typedef struct qnode_s qnode_t;

struct qnode_s {
    qnode_t *next;
    qnode_t *prev;

    void *data;
};

struct queue_s {
    qnode_t *head;
    qnode_t *tail;

    pthread_mutex_t mutex;
    pthread_cond_t cond;

    bool release;
    bool wait_for_consumers;
    int active_consumers;
};

LOG_LEVEL queue_log_level = LOG_LEVEL_TRACE;

static bool queue_enqueue(queue_ts_t * const queue, void * const data) {
    qnode_t *new;
    new = checked_malloc(sizeof(qnode_t)); /* freed in queue_dequeue */
    new->data = data;

    bool new_head_tail = false;

    if (queue->head == NULL) {
        /* new node is first element in empty queue */
        queue->head = new;
        queue->tail = new;
        new->next = NULL;
        new->prev = NULL;
        new_head_tail = true;
        log_msg(queue_log_level, "queue(%p): add node %p with payload %p as new head and new tail", (void*) queue, (void*) new, (void*) new->data);
    } else {
        /* new node is new tail */
        (queue->tail)->prev = new;
        new->next = queue->tail;
        new->prev = NULL;
        queue->tail = new;
        log_msg(queue_log_level, "queue(%p): add node %p with payload %p as new tail", (void*) queue, (void*) new, (void*) new->data);
    }
    return new_head_tail;
}

queue_ts_t *queue_ts_init(void) {
    queue_ts_t *queue = checked_malloc (sizeof(queue_ts_t));

    pthread_mutexattr_t attr;
    pthread_mutexattr_init (&attr);
    pthread_mutexattr_settype (&attr, PTHREAD_MUTEX_RECURSIVE);
    pthread_mutex_init(&queue->mutex, &attr);
    pthread_cond_init(&queue->cond, NULL);

    pthread_mutex_lock(&queue->mutex);

    queue->release = false;
    queue->wait_for_consumers = false;
    queue->active_consumers = 0;

    queue->head = NULL;
    queue->tail = NULL;

    pthread_mutex_unlock(&queue->mutex);

    log_msg(queue_log_level, "queue(%p): create new queue", (void*) queue);
    return queue;
}

void queue_ts_free(queue_ts_t *queue) {
    if (queue) {
        pthread_cond_destroy(&queue->cond);
        pthread_mutex_destroy(&queue->mutex);
        free(queue);
    }
}

bool queue_ts_enqueue(queue_ts_t * const queue, void * const data, const char *whoami) {
    pthread_mutex_lock(&queue->mutex);
    bool new_head_tail = queue_enqueue(queue,data);
    pthread_mutex_unlock(&queue->mutex);

    if (new_head_tail) {
        pthread_cond_broadcast(&queue->cond);
        log_msg(LOG_LEVEL_THREAD, "%10s: queue(%p): broadcast waiting threads for new head node in queue", whoami, (void*) queue);
    }
    return new_head_tail;
}

void *queue_ts_dequeue_wait(queue_ts_t * const queue, const char *whoami) {
    qnode_t *head;
    void *data = NULL;
    pthread_mutex_lock(&queue->mutex);

    if (queue->wait_for_consumers) {
        queue->active_consumers--;
        if (queue->active_consumers == 0 && queue->release) {
            pthread_cond_broadcast(&queue->cond);
        }
    }
    while ((head = queue->head) == NULL && (queue->release == false || queue->active_consumers > 0)) {
        log_msg(LOG_LEVEL_THREAD, "%10s: queue(%p): waiting for new node", whoami, (void*) queue);
        pthread_cond_wait(&queue->cond, &queue->mutex);
        log_msg(LOG_LEVEL_THREAD, "%10s: queue(%p): got broadcast (head: %p)", whoami, (void*) queue, (void*) queue->head);
    }
    if (head != NULL) {
        if (queue->wait_for_consumers) { queue->active_consumers++; }
        if ((queue->head = head->prev) == NULL) {
            queue->tail = NULL;
        } else {
            (queue->head)->next = NULL;
        }
        data = head->data;
        log_msg(queue_log_level, "queue(%p): return head node %p with payload %p", (void*) queue, (void*) head, (void*) head->data);
        free(head);
    } else {
        log_msg(queue_log_level, "queue(%p): return NULL from empty, released queue", (void*) queue);
    }
    pthread_mutex_unlock(&queue->mutex);
    return data;
}

void queue_ts_register(queue_ts_t * const queue, const char *whoami) {
    pthread_mutex_lock(&queue->mutex);
    queue->wait_for_consumers = true;
    queue->active_consumers++;
    pthread_mutex_unlock(&queue->mutex);
    log_msg(LOG_LEVEL_THREAD, "%10s: queue(%p): register thread", whoami, (void*) queue);
}

void queue_ts_release(queue_ts_t * const queue, const char *whoami) {
    pthread_mutex_lock(&queue->mutex);
    queue->release = true;
    pthread_mutex_unlock(&queue->mutex);
    pthread_cond_broadcast(&queue->cond);
    log_msg(LOG_LEVEL_THREAD, "%10s: queue(%p): release queue and broadcast waiting threads", whoami, (void*) queue);
}
