feat(lander): initial integration of lsm

include/http_loop.h

@@ -166,6 +166,13 @@ bool http_loop_step_body_to_buf(event_loop_conn *conn);
  */
 bool http_loop_step_body_to_file(event_loop_conn *conn);
 
+/**
+ * Try to parse the Content-Length header.
+ *
+ * @param conn connection to process
+ */
+bool http_loop_step_parse_content_length(event_loop_conn *conn);
+
 /**
  * Authenticate the request using the X-Api-Key header.
  *
@@ -201,7 +208,7 @@ bool http_loop_step_write_body(event_loop_conn *conn);
  * Initialize a new http loop.
  *
  * @param routes array of routes that should be served
- * @parma route_count how many elements are in `routes`
+ * @param route_count how many elements are in `routes`
  * @param custom_gctx the application's custom global context; can be NULL
  * @param custom_ctx_init function to initialize a new custom context
  * @param custom_ctx_reset function to reset a custom context

include/lander.h

@@ -2,18 +2,27 @@
 #define LANDER
 
 #include "http_loop.h"
+#include "lsm/store.h"
 
 extern http_route lander_routes[4];
 
 typedef struct lander_gctx {
   const char *data_dir;
   Trie *trie;
+  lsm_store *store;
+
 } lander_gctx;
 
 typedef struct lander_ctx {
-  char *key;
+  lsm_entry_handle *entry;
+  uint64_t remaining_data;
 } lander_ctx;
 
+typedef enum lander_entry_type {
+  lander_entry_type_redirect = 0,
+  lander_entry_type_paste = 1,
+} lander_entry_type;
+
 void *lander_gctx_init();
 
 void *lander_ctx_init();
@@ -30,4 +39,14 @@ bool lander_post_redirect(event_loop_conn *conn);
 
 bool lander_post_paste(event_loop_conn *conn);
 
+bool lander_post_paste_lsm(event_loop_conn *conn);
+
+bool lander_post_redirect_lsm(event_loop_conn *conn);
+
+bool lander_stream_body_to_entry(event_loop_conn *conn);
+
+bool lander_stream_body_to_client(event_loop_conn *conn);
+
+bool lander_get_entry_lsm(event_loop_conn *conn);
+
 #endif

lsm/include/lsm/store.h

@@ -44,6 +44,16 @@ bool lsm_entry_attr_present(lsm_entry_handle *handle, lsm_attr_type type);
 lsm_error lsm_entry_attr_get(lsm_str **out, lsm_entry_handle *handle,
                              lsm_attr_type type);
 
+/**
+ * Convenience wrapper around `lsm_entry_attr_get` that can be used if we know
+ * beforehand the attribute value is a 64-bit number.
+ *
+ * @param out where to store attribute data
+ * @param entry entry to search for
+ * @param type type of attribute to return
+ */
+lsm_error lsm_entry_attr_get_num(uint64_t *out, lsm_entry_handle *handle, lsm_attr_type type);
+
 /**
  * Add a new attribute to the entry.
  *
@@ -54,6 +64,16 @@ lsm_error lsm_entry_attr_get(lsm_str **out, lsm_entry_handle *handle,
 lsm_error lsm_entry_attr_insert(lsm_entry_handle *handle, lsm_attr_type type,
                                 lsm_str *data);
 
+/**
+ * Convenience wrapper around `lsm_entry_attr_insert` that can be used if the
+ * data to be stored is a 64-bit number.
+ *
+ * @param entry entry to modify
+ * @param type type of attribute to add
+ * @param data data of attribute
+ */
+lsm_error lsm_entry_attr_insert_num(lsm_entry_handle *handle, lsm_attr_type type, uint64_t data);
+
 /**
  * Remove an atribute from the given entry, if present.
  *
@@ -151,6 +171,16 @@ lsm_error lsm_store_insert(lsm_entry_handle **out, lsm_store *store,
 lsm_error lsm_entry_data_append(lsm_store *store, lsm_entry_handle *handle,
                                 lsm_str *data);
 
+/**
+ * Same as `lsm_entry_data_append`, except that it takes a direct char array.
+ *
+ * @param store store the entry is stored in
+ * @param entry entry to append data to
+ * @param data data to append
+ * @param len length of data array
+ */
+lsm_error lsm_entry_data_append_raw(lsm_store *store, lsm_entry_handle *handle, char *data, uint64_t len);
+
 /**
  * Read a number of bytes from the entry's data field. The position from which
  * data is read is dependent on previous read calls.
@@ -163,4 +193,12 @@ lsm_error lsm_entry_data_append(lsm_store *store, lsm_entry_handle *handle,
 lsm_error lsm_entry_data_read(uint64_t *out, char *buf,
                               lsm_entry_handle *handle, uint64_t len);
 
+/**
+ * Return the length of the entry's data.
+ *
+ * @param handle entry handle to return length for
+ * @return length of the data
+ */
+uint64_t lsm_entry_data_len(lsm_entry_handle *handle);
+
 #endif

lsm/src/store/lsm_store.c

@@ -153,18 +153,32 @@ lsm_error lsm_store_insert(lsm_entry_handle **out, lsm_store *store,
                            lsm_str *key) {
   // TODO what happens when two inserts to the same key happen at the same time?
   lsm_entry_wrapper *wrapper;
-  LSM_RES(lsm_entry_wrapper_init(&wrapper));
-  pthread_rwlock_wrlock(&wrapper->lock);
 
-  lsm_error res = lsm_trie_insert(store->trie, key, wrapper);
+  // If a key was previously removed from the trie, the wrapper will already be
+  // present in the trie
+  if (lsm_trie_search((void **)&wrapper, store->trie, key) == lsm_error_not_found) {
+    LSM_RES(lsm_entry_wrapper_init(&wrapper));
+    pthread_rwlock_wrlock(&wrapper->lock);
 
-  // Check if entry isn't already present in advance
-  if (res != lsm_error_ok) {
-    lsm_entry_wrapper_free(wrapper);
+    lsm_error res = lsm_trie_insert(store->trie, key, wrapper);
 
-    return res;
+    // Check if entry isn't already present in advance
+    if (res != lsm_error_ok) {
+      lsm_entry_wrapper_free(wrapper);
+
+      return res;
+    }
+  } else {
+    pthread_rwlock_wrlock(&wrapper->lock);
+
+    if (wrapper->entry != NULL) {
+      pthread_rwlock_unlock(&wrapper->lock);
+
+      return lsm_error_already_present;
+    }
   }
 
+
   lsm_entry *entry;
   LSM_RES(lsm_entry_init(&entry));
 

lsm/src/store/lsm_store_entry.c

@@ -5,6 +5,7 @@
 
 #include "lsm.h"
 #include "lsm/store_internal.h"
+#include "lsm/str.h"
 
 lsm_error lsm_entry_init(lsm_entry **ptr) {
   lsm_entry *entry = calloc(1, sizeof(lsm_entry));
@@ -77,6 +78,22 @@ lsm_error lsm_entry_attr_get(lsm_str **out, lsm_entry_handle *handle,
   return lsm_error_ok;
 }
 
+lsm_error lsm_entry_attr_get_num(uint64_t *out, lsm_entry_handle *handle, lsm_attr_type type) {
+  lsm_str *s;
+
+  LSM_RES(lsm_entry_attr_get(&s, handle, type));
+
+  uint64_t num;
+
+  for (uint8_t i = 0; i < sizeof(uint64_t) / sizeof(char); i++) {
+    ((char *)&num)[i] = lsm_str_char(s, i);
+  }
+
+  *out = num;
+
+  return lsm_error_ok;
+}
+
 lsm_error lsm_entry_attr_remove(lsm_str **out, lsm_entry_handle *handle,
                                 lsm_attr_type type) {
   if (!lsm_entry_attr_present(handle, type)) {
@@ -149,3 +166,14 @@ lsm_error lsm_entry_attr_insert(lsm_entry_handle *handle, lsm_attr_type type,
 
   return lsm_error_ok;
 }
+
+lsm_error lsm_entry_attr_insert_num(lsm_entry_handle *handle, lsm_attr_type type, uint64_t data) {
+  lsm_str *s;
+  LSM_RES(lsm_str_init_copy_n(&s, (char *)&data, sizeof(uint64_t) / sizeof(char)));
+
+  return lsm_entry_attr_insert(handle, type, s);
+}
+
+uint64_t lsm_entry_data_len(lsm_entry_handle *handle) {
+  return handle->wrapper->entry->data.len;
+}

lsm/src/str/lsm_str.c

@@ -77,16 +77,16 @@ lsm_error lsm_str_overwrite_copy(lsm_str *str, char *s) {
 }
 
 lsm_error lsm_str_overwrite_copy_n(lsm_str *str, char *s, uint64_t len) {
-  if (str->len <= 8) {
-    memcpy(str->data.val, s, str->len);
+  if (len <= 8) {
+    memcpy(str->data.val, s, len);
   } else {
-    char *buf = malloc(str->len * sizeof(char));
+    char *buf = malloc(len * sizeof(char));
 
     if (buf == NULL) {
       return lsm_error_failed_alloc;
     }
 
-    memcpy(buf, s, str->len);
+    memcpy(buf, s, len);
     str->data.ptr = buf;
   }
 

src/http_loop/http_loop_req.c

@@ -144,5 +144,6 @@ void http_loop_process_request(event_loop_conn *conn) {
   if ((conn->state != event_loop_conn_state_req) ||
       (ctx->route->steps[ctx->current_step] == NULL)) {
     ctx->current_step = 0;
+    conn->state = event_loop_conn_state_res;
   }
 }

src/http_loop/http_loop_steps.c

@@ -1,5 +1,6 @@
 #include <math.h>
 
+#include "http_loop.h"
 #include "lander.h"
 
 /*
@@ -22,12 +23,44 @@ static bool string_to_num(size_t *res, const char *s, size_t len) {
   return true;
 }
 
+bool http_loop_step_parse_content_length(event_loop_conn *conn) {
+  http_loop_ctx *ctx = conn->ctx;
+
+  for (size_t i = 0; i < ctx->req.num_headers; i++) {
+    struct phr_header *header = &ctx->req.headers[i];
+
+    if (strncmp(header->name, "Content-Length", header->name_len) == 0) {
+      // If the content length header is present but contains an invalid
+      // number, we return a bad request error
+      if (!string_to_num(&ctx->req.body.expected_len, header->value,
+                         header->value_len)) {
+        ctx->res.status = http_bad_request;
+        conn->state = event_loop_conn_state_res;
+
+        return true;
+      }
+      // The content length was actually 0, so we can instantly return here
+      else if (ctx->req.body.expected_len == 0) {
+        return true;
+      }
+    }
+  }
+
+  // A zero here means there's no content length header
+  if (ctx->req.body.expected_len == 0) {
+    ctx->res.status = http_length_required;
+    conn->state = event_loop_conn_state_res;
+  }
+
+  return true;
+}
+
 /*
  * Try to find and parse the Content-Length header. This function returns true
  * if it was successful. If false is returned, the underlying step should
  * immediately exit.
  */
-static bool try_parse_content_length(event_loop_conn *conn) {
+bool try_parse_content_length(event_loop_conn *conn) {
   http_loop_ctx *ctx = conn->ctx;
 
   for (size_t i = 0; i < ctx->req.num_headers; i++) {

src/lander/lander.c

@@ -1,6 +1,8 @@
 #include <stdio.h>
 
+#include "http_loop.h"
 #include "lander.h"
+#include "lsm/store.h"
 
 http_route lander_routes[] = {
     {.type = http_route_literal,
@@ -13,24 +15,24 @@ http_route lander_routes[] = {
         .type = http_route_regex,
         .method = http_get,
         .path = "^/([^/]+)$",
-        .steps = {lander_get_entry, NULL},
-        .steps_res = {http_loop_step_write_header, http_loop_step_write_body,
+        .steps = {lander_get_entry_lsm, NULL},
+        .steps_res = {http_loop_step_write_header, lander_stream_body_to_client,
                       NULL},
     },
     {
         .type = http_route_regex,
         .method = http_post,
         .path = "^/s(l?)/([^/]*)$",
-        .steps = {http_loop_step_auth, http_loop_step_body_to_buf,
-                  lander_post_redirect, NULL},
+        .steps = {http_loop_step_auth, http_loop_step_parse_content_length,
+                  lander_post_redirect_lsm, lander_stream_body_to_entry, NULL},
         .steps_res = {http_loop_step_write_header, http_loop_step_write_body,
                       NULL},
     },
     {.type = http_route_regex,
      .method = http_post,
      .path = "^/p(l?)/([^/]*)$",
-     .steps = {http_loop_step_auth, lander_post_paste,
-               http_loop_step_body_to_file, http_loop_step_switch_res, NULL},
+     .steps = {http_loop_step_auth, http_loop_step_parse_content_length,
+               lander_post_paste_lsm, lander_stream_body_to_entry, NULL},
      .steps_res = {http_loop_step_write_header, http_loop_step_write_body,
                    NULL}},
 };
@@ -39,6 +41,14 @@ void *lander_gctx_init() { return calloc(1, sizeof(lander_gctx)); }
 
 void *lander_ctx_init() { return calloc(1, sizeof(lander_ctx)); }
 
-void lander_ctx_reset(lander_ctx *ctx) {}
+void lander_ctx_reset(lander_ctx *ctx) {
+  if (ctx->entry != NULL) {
+    lsm_entry_close(ctx->entry);
+
+    ctx->entry = NULL;
+  }
+
+  ctx->remaining_data = 0;
+}
 
 void lander_ctx_free(lander_ctx *ctx) { free(ctx); }

src/lander/lander_get.c

@@ -1,6 +1,9 @@
 #include <stdio.h>
 
+#include "event_loop.h"
+#include "http/types.h"
 #include "lander.h"
+#include "lsm/store.h"
 
 static const char index_page[] =
     "<!DOCTYPE html>\n"
@@ -50,3 +53,82 @@ bool lander_get_entry(event_loop_conn *conn) {
 
   return true;
 }
+
+bool lander_get_entry_lsm(event_loop_conn *conn) {
+  http_loop_ctx *ctx = conn->ctx;
+  lander_ctx *c_ctx = ctx->c;
+  http_loop_gctx *gctx = ctx->g;
+  lander_gctx *c_gctx = gctx->c;
+
+  const char *key_s = &ctx->req.path[ctx->req.regex_groups[1].rm_so];
+  int key_len = ctx->req.regex_groups[1].rm_eo - ctx->req.regex_groups[1].rm_so;
+
+  lsm_str *key;
+  lsm_str_init_copy_n(&key, (char *)key_s, key_len);
+
+  switch (lsm_store_open_read(&c_ctx->entry, c_gctx->store, key)) {
+  case lsm_error_ok:
+    break;
+  case lsm_error_not_found:
+    ctx->res.status = http_not_found;
+    return true;
+  default:
+    ctx->res.status = http_internal_server_error;
+    return true;
+  }
+
+  lander_entry_type t;
+  lsm_entry_attr_get_num((uint64_t *)&t, c_ctx->entry,
+                         lsm_attr_type_entry_type);
+
+  if (t == lander_entry_type_redirect) {
+    // Stream entire redirect data into buffer to set as header
+    uint64_t data_len = lsm_entry_data_len(c_ctx->entry);
+    char *buf = malloc(data_len + 1);
+    uint64_t read = 0;
+    uint64_t total_read = 0;
+
+    while (total_read < data_len) {
+      lsm_entry_data_read(&read, &buf[total_read], c_ctx->entry,
+                          data_len - total_read);
+      total_read += read;
+    }
+
+    buf[data_len] = '\0';
+
+    ctx->res.status = http_moved_permanently;
+    http_res_add_header(&ctx->res, http_header_location, buf, true);
+
+    // We no longer need the entry at this point, so we can unlock it early
+    // This will also signal to the response code not to read any data from
+    // the entry
+    lsm_entry_close(c_ctx->entry);
+    c_ctx->entry = NULL;
+  } else {
+    ctx->res.body.expected_len = lsm_entry_data_len(c_ctx->entry);
+  }
+
+  return true;
+}
+
+bool lander_stream_body_to_client(event_loop_conn *conn) {
+  http_loop_ctx *ctx = conn->ctx;
+  lander_ctx *c_ctx = ctx->c;
+
+  if ((c_ctx->entry == NULL) ||
+      (ctx->res.body.expected_len == ctx->res.body.len)) {
+    return true;
+  }
+
+  uint64_t to_write = MIN(EVENT_LOOP_BUFFER_SIZE - conn->wbuf_size,
+                          ctx->res.body.expected_len - ctx->res.body.len);
+
+  uint64_t read = 0;
+  lsm_entry_data_read(&read, (char *)&conn->wbuf[conn->wbuf_size], c_ctx->entry,
+                      to_write);
+
+  ctx->res.body.len += read;
+  conn->wbuf_size += read;
+
+  return false;
+}

src/lander/lander_post.c

@@ -1,6 +1,16 @@
 #include "http/res.h"
+#include "http/types.h"
 #include "lander.h"
 #include "log.h"
+#include "lsm/store.h"
+
+static void randomize_key(char *key, int len) {
+  for (int i = 0; i < len; i++) {
+    key[i] = charset[rand() % charset_len];
+  }
+
+  key[len] = '\0';
+}
 
 // TODO entry leaks if key is already present
 static bool add_entry(char **key_ptr, int *key_len_ptr, http_loop_ctx *ctx,
@@ -60,6 +70,110 @@ static bool add_entry(char **key_ptr, int *key_len_ptr, http_loop_ctx *ctx,
   return true;
 }
 
+/**
+ * Insert a new entry into the store.
+ *
+ * @return true on success, false otherwise
+ */
+bool lander_insert_entry(http_loop_ctx *ctx) {
+  http_loop_gctx *gctx = ctx->g;
+  lander_gctx *c_gctx = gctx->c;
+  lander_ctx *c_ctx = ctx->c;
+
+  lsm_str *key;
+  int key_len;
+
+  if (ctx->req.regex_groups[2].rm_eo == ctx->req.regex_groups[2].rm_so) {
+    // Generate a random key to insert
+    bool secure =
+        (ctx->req.regex_groups[1].rm_eo - ctx->req.regex_groups[1].rm_so) == 1;
+    key_len = secure ? 16 : 4;
+    char *key_s = malloc((key_len + 1) * sizeof(char));
+
+    randomize_key(key_s, key_len);
+    lsm_str_init(&key, key_s);
+  } else {
+    char *key_s = (char *)&ctx->req.path[ctx->req.regex_groups[2].rm_so];
+    key_len = ctx->req.regex_groups[2].rm_eo - ctx->req.regex_groups[2].rm_so;
+
+    lsm_str_init_copy_n(&key, key_s, key_len);
+  }
+
+  // TODO free key on error
+  switch (lsm_store_insert(&c_ctx->entry, c_gctx->store, key)) {
+  case lsm_error_already_present:
+    ctx->res.status = http_conflict;
+    return false;
+  case lsm_error_ok:
+    break;
+  default:
+    ctx->res.status = http_internal_server_error;
+    return false;
+  }
+
+  // Add location header
+  char *buf = malloc(key_len + 2);
+  memcpy(&buf[1], lsm_str_ptr(key), key_len);
+  buf[0] = '/';
+  buf[key_len + 1] = '\0';
+
+  http_res_add_header(&ctx->res, http_header_location, buf, true);
+  ctx->res.status = http_created;
+
+  return true;
+}
+
+bool lander_post_redirect_lsm(event_loop_conn *conn) {
+  http_loop_ctx *ctx = conn->ctx;
+  lander_ctx *c_ctx = ctx->c;
+
+  if (!lander_insert_entry(ctx)) {
+    conn->state = event_loop_conn_state_res;
+    return true;
+  }
+
+  lsm_entry_attr_insert_num(c_ctx->entry, lsm_attr_type_entry_type,
+                            lander_entry_type_redirect);
+
+  return true;
+}
+
+bool lander_post_paste_lsm(event_loop_conn *conn) {
+  http_loop_ctx *ctx = conn->ctx;
+  lander_ctx *c_ctx = ctx->c;
+
+  if (!lander_insert_entry(ctx)) {
+    conn->state = event_loop_conn_state_res;
+    return true;
+  }
+
+  lsm_entry_attr_insert_num(c_ctx->entry, lsm_attr_type_entry_type,
+                            lander_entry_type_paste);
+
+  return true;
+}
+
+bool lander_stream_body_to_entry(event_loop_conn *conn) {
+  http_loop_ctx *ctx = conn->ctx;
+  lander_ctx *c_ctx = ctx->c;
+  http_loop_gctx *gctx = ctx->g;
+  lander_gctx *c_gctx = gctx->c;
+
+  uint64_t to_append =
+      MIN(conn->rbuf_size - conn->rbuf_read,
+          ctx->req.body.expected_len - lsm_entry_data_len(c_ctx->entry));
+
+  lsm_str *data;
+  lsm_str_init_copy_n(&data, (char *)&conn->rbuf[conn->rbuf_read], to_append);
+  lsm_entry_data_append(c_gctx->store, c_ctx->entry, data);
+
+  conn->rbuf_read += to_append;
+
+  lsm_str_free(data);
+
+  return lsm_entry_data_len(c_ctx->entry) == ctx->req.body.expected_len;
+}
+
 bool lander_post_redirect(event_loop_conn *conn) {
   http_loop_ctx *ctx = conn->ctx;
   bool random =

src/main.c

@@ -49,6 +49,11 @@ int main() {
   c_gctx->data_dir = data_dir;
   c_gctx->trie = trie;
 
+  lsm_str *db_path, *data_dir2;
+  lsm_str_init_copy(&db_path, "data/store.db");
+  lsm_str_init_copy(&data_dir2, "data");
+  lsm_store_load(&c_gctx->store, db_path, data_dir2);
+
   http_loop *hl = http_loop_init(
       lander_routes, sizeof(lander_routes) / sizeof(lander_routes[0]), c_gctx,
       lander_ctx_init, (void (*)(void *))lander_ctx_reset,