feat(job-queue): initial implementation of functionality

src/job-queue/vieter_job_queue.c

@@ -0,0 +1,153 @@
+#include "vieter_job_queue_internal.h"
+
+vieter_job_queue *vieter_job_queue_init() {
+  vieter_job_queue *queue = malloc(sizeof(vieter_job_queue));
+
+  queue->tree = vieter_tree_init();
+
+  for (int i = 0; i < VIETER_JOB_STATES; i++) {
+    queue->heaps[i] = vieter_heap_init();
+  }
+
+  return queue;
+}
+
+void vieter_job_queue_free(vieter_job_queue **ptp) {
+  vieter_job_queue *queue = *ptp;
+
+  vieter_tree_free(queue->tree);
+
+  for (int i = 0; i < VIETER_JOB_STATES; i++) {
+    vieter_heap_free(queue->heaps[i]);
+  }
+
+  free(queue);
+  *ptp = NULL;
+}
+
+vieter_job *vieter_job_init() { return calloc(1, sizeof(vieter_job)); }
+
+void vieter_job_free(vieter_job **ptp) {
+  vieter_job *job = *ptp;
+
+  if (job->schedule != NULL) {
+    vieter_cron_expr_free(job->schedule);
+  }
+
+  if (job->failure_report != NULL) {
+    vieter_job_failure_report_free(&job->failure_report);
+  }
+
+  free(job);
+
+  *ptp = NULL;
+}
+
+vieter_job_queue_error vieter_job_queue_insert(vieter_job_queue *queue,
+                                               vieter_job *job) {
+  vieter_tree_error tree_res = vieter_tree_insert(queue->tree, job->id, job);
+
+  if (tree_res != vieter_tree_ok) {
+    return vieter_job_queue_already_present;
+  }
+
+  vieter_heap_insert(queue->heaps[VIETER_JOB_INITIAL_STATE], job->id, job);
+
+  job->current_state = VIETER_JOB_INITIAL_STATE;
+  job->dispatched = false;
+  job->state_transition_times[VIETER_JOB_INITIAL_STATE] = time(NULL);
+
+  return vieter_job_queue_ok;
+}
+
+vieter_job_queue_error vieter_job_queue_pop(vieter_job **out,
+                                            vieter_job_queue *queue,
+                                            vieter_job_state state) {
+  vieter_heap_error res = vieter_heap_pop((void **)out, queue->heaps[state]);
+
+  if (res != vieter_heap_ok) {
+    return vieter_job_queue_state_empty;
+  }
+
+  (*out)->dispatched = true;
+
+  return vieter_job_queue_ok;
+}
+
+vieter_job_queue_error vieter_job_queue_transition(vieter_job_queue *queue,
+                                                   uint64_t id,
+                                                   vieter_job_state new_state) {
+  vieter_job *job;
+
+  vieter_tree_error res = vieter_tree_search((void **)&job, queue->tree, id);
+
+  if (res != vieter_tree_ok) {
+    return vieter_job_queue_not_present;
+  }
+
+  if (!job->dispatched) {
+    return vieter_job_queue_not_dispatched;
+  }
+
+  vieter_heap_insert(queue->heaps[new_state], job->id, job);
+
+  job->current_state = new_state;
+  job->dispatched = false;
+  job->state_transition_times[new_state] = time(NULL);
+
+  return vieter_job_queue_ok;
+}
+
+vieter_job_queue_error vieter_job_queue_remove(vieter_job **out,
+                                               vieter_job_queue *queue,
+                                               uint64_t id) {
+  vieter_tree_error res = vieter_tree_search((void **)out, queue->tree, id);
+
+  if (res != vieter_tree_ok) {
+    return vieter_job_queue_not_present;
+  }
+
+  vieter_job *job = *out;
+
+  if (!job->dispatched) {
+    return vieter_job_queue_not_dispatched;
+  }
+
+  // This can't fail if the search succeeded
+  vieter_tree_remove((void **)out, queue->tree, job->id);
+
+  return vieter_job_queue_ok;
+}
+
+vieter_job_failure_report *vieter_job_failure_report_init() {
+  return calloc(1, sizeof(vieter_job_failure_report));
+}
+
+vieter_job_queue_error vieter_job_queue_fail(vieter_job_queue *queue,
+                                             uint64_t id,
+                                             char *report_message) {
+  vieter_job *job;
+
+  vieter_tree_error res = vieter_tree_search((void **)&job, queue->tree, id);
+
+  if (res != vieter_tree_ok) {
+    return vieter_job_queue_not_present;
+  }
+
+  if (!job->dispatched) {
+    return vieter_job_queue_not_dispatched;
+  }
+
+  vieter_heap_insert(queue->heaps[VIETER_JOB_FAILURE_STATE], job->id, job);
+
+  job->dispatched = false;
+  job->state_transition_times[VIETER_JOB_FAILURE_STATE] = time(NULL);
+
+  job->failure_report = vieter_job_failure_report_init();
+  job->failure_report->failed_state = job->current_state;
+  job->failure_report->msg = strdup(report_message);
+
+  job->current_state = VIETER_JOB_FAILURE_STATE;
+
+  return vieter_job_queue_ok;
+}

src/job-queue/vieter_job_queue_internal.h

@@ -1,10 +1,13 @@
 #ifndef VIETER_JOB_QUEUE_INTERNAL
 #define VIETER_JOB_QUEUE_INTERNAL
 
+#include "vieter_heap.h"
+#include "vieter_job_queue.h"
 #include "vieter_tree.h"
 
 struct vieter_job_queue {
   vieter_tree *tree;
+  vieter_heap *heaps[VIETER_JOB_STATES];
 };
 
 #endif