path: root/src/storage/store.c

#include "storage/store.h"
#include "storage/codec.h"
#include "sys/system.h"

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "base/allocator.h"

#define RECORD_HEADER_SIZE 24U
#define RECORD_VERSION 1U
#define RECORD_PUT 1U
#define RECORD_DEL 2U
#define RECORD_MARKER 3U
#define RECORD_PAYLOAD_MAX (TINY_CONTENT_MAX + 16384U)
#define SNAPSHOT_HEADER_SIZE 52U

struct tiny_store
{
  tiny_sys_file_t lock;
  tiny_sys_file_t log;
  char *directory;
  char *log_path;
  char *snapshot_path;
  tiny_store_entry_t *entries;
  size_t entry_count;
  size_t entry_capacity;
  uint64_t last_sequence;
  uint64_t last_ulid_ms;
  unsigned char last_random[10];
  unsigned char generation[16];
  bool has_generation;
  bool recovered_tail;
  bool snapshot_valid;
  char last_error[160];
};

static void
set_error(tiny_store_t *store, const char *message)
{
  snprintf(store->last_error, sizeof(store->last_error), "%s", message);
}

static bool
write_all(tiny_sys_file_t *file, const void *data, uint32_t length)
{
  const unsigned char *position = data;
  while (length > 0)
    {
      uint32_t written = 0;
      if (!tiny_sys_file_write(file, position, length, &written) || written == 0)
        {
          return false;
        }
      position += written;
      length -= written;
    }
  return true;
}

static bool
read_all(tiny_sys_file_t *file, void *data, uint32_t length, uint32_t *actual)
{
  unsigned char *position = data;
  uint32_t total = 0;
  while (total < length)
    {
      uint32_t count = 0;
      if (!tiny_sys_file_read(file, position + total, length - total, &count))
        {
          return false;
        }
      if (count == 0)
        {
          break;
        }
      total += count;
    }
  *actual = total;
  return true;
}

static uint64_t
now_milliseconds(void)
{
  return tiny_sys_now_milliseconds();
}

static void
make_ulid(tiny_store_t *store, uint64_t timestamp, char output[27])
{
  static const char alphabet[] = "0123456789ABCDEFGHJKMNPQRSTVWXYZ";
  unsigned char bytes[16];
  unsigned index;
  if (timestamp > store->last_ulid_ms)
    {
      store->last_ulid_ms = timestamp;
      if (!tiny_sys_random_bytes(store->last_random, sizeof(store->last_random)))
        {
          uint64_t fallback = tiny_sys_monotonic_milliseconds() ^ timestamp;
          for (index = 0; index < sizeof(store->last_random); ++index)
            {
              store->last_random[index] = (unsigned char)(fallback >> ((index % 8) * 8));
            }
        }
    }
  else
    {
      timestamp = store->last_ulid_ms;
      for (index = 10; index-- > 0;)
        {
          if (++store->last_random[index] != 0)
            {
              break;
            }
        }
    }
  bytes[0] = (unsigned char)(timestamp >> 40);
  bytes[1] = (unsigned char)(timestamp >> 32);
  bytes[2] = (unsigned char)(timestamp >> 24);
  bytes[3] = (unsigned char)(timestamp >> 16);
  bytes[4] = (unsigned char)(timestamp >> 8);
  bytes[5] = (unsigned char)timestamp;
  memcpy(bytes + 6, store->last_random, 10);
  for (index = 0; index < 26; ++index)
    {
      const unsigned bit = index * 5;
      unsigned value = 0, part;
      for (part = 0; part < 5; ++part)
        {
          const int source_bit = (int)bit + (int)part - 2;
          value <<= 1;
          if (source_bit >= 0)
            {
              value |= (bytes[source_bit / 8] >> (7 - source_bit % 8)) & 1U;
            }
        }
      output[index] = alphabet[value];
    }
  output[26] = '\0';
}

static bool
ensure_entries(tiny_store_t *store)
{
  tiny_store_entry_t *resized;
  size_t capacity;
  if (store->entry_count < store->entry_capacity)
    {
      return true;
    }
  capacity = store->entry_capacity == 0 ? 32 : store->entry_capacity * 2;
  resized = realloc(store->entries, capacity * sizeof(*resized));
  if (resized == NULL)
    {
      return false;
    }
  store->entries = resized;
  store->entry_capacity = capacity;
  return true;
}

static tiny_store_entry_t *
find_entry(tiny_store_t *store, const char *message_id)
{
  size_t index;
  for (index = 0; index < store->entry_count; ++index)
    {
      if (strcmp(store->entries[index].message.message_id, message_id) == 0)
        {
          return &store->entries[index];
        }
    }
  return NULL;
}

static bool
serialize_put(const tiny_message_t *message, unsigned char **payload, uint32_t *length)
{
  size_t session_len = strlen(message->session_id), content_len = strlen(message->content);
  size_t tool_len = message->metadata.tool_call_id ? strlen(message->metadata.tool_call_id) : 0;
  size_t total = 8 + 26 + 2 + 16 + session_len + content_len + tool_len;
  size_t index, offset;
  unsigned char *data;
  for (index = 0; index < message->metadata.tag_count; ++index)
    {
      total += 4 + strlen(message->metadata.tags[index]);
    }
  if (total > UINT32_MAX || total > RECORD_PAYLOAD_MAX)
    {
      return false;
    }
  data = malloc(total);
  if (data == NULL)
    {
      return false;
    }
  offset = 0;
  tiny_codec_put_u64(data + offset, (uint64_t)message->timestamp_ms);
  offset += 8;
  memcpy(data + offset, message->message_id, 26);
  offset += 26;
  data[offset++] = (unsigned char)message->role;
  data[offset++] = (unsigned char)message->metadata.memory_tier;
  tiny_codec_put_u32(data + offset, (uint32_t)session_len);
  offset += 4;
  tiny_codec_put_u32(data + offset, (uint32_t)content_len);
  offset += 4;
  tiny_codec_put_u32(data + offset, (uint32_t)tool_len);
  offset += 4;
  tiny_codec_put_u32(data + offset, (uint32_t)message->metadata.tag_count);
  offset += 4;
  memcpy(data + offset, message->session_id, session_len);
  offset += session_len;
  memcpy(data + offset, message->content, content_len);
  offset += content_len;
  if (tool_len > 0)
    {
      memcpy(data + offset, message->metadata.tool_call_id, tool_len);
      offset += tool_len;
    }
  for (index = 0; index < message->metadata.tag_count; ++index)
    {
      const size_t tag_len = strlen(message->metadata.tags[index]);
      tiny_codec_put_u32(data + offset, (uint32_t)tag_len);
      offset += 4;
      memcpy(data + offset, message->metadata.tags[index], tag_len);
      offset += tag_len;
    }
  *payload = data;
  *length = (uint32_t)total;
  return true;
}

static char *
copy_field(const unsigned char *data, size_t length)
{
  char *value = malloc(length + 1);
  if (value != NULL)
    {
      memcpy(value, data, length);
      value[length] = '\0';
    }
  return value;
}

static bool
deserialize_failed(tiny_message_t *message)
{
  tiny_message_free(message);
  return false;
}

static bool
deserialize_put(const unsigned char *data, uint32_t length, tiny_message_t *message)
{
  uint32_t session_len, content_len, tool_len, tag_count;
  size_t offset = 0, index;
  tiny_message_init(message);
  if (length < 8 + 26 + 2 + 16)
    {
      return false;
    }
  message->timestamp_ms = (int64_t)tiny_codec_get_u64(data + offset);
  offset += 8;
  memcpy(message->message_id, data + offset, 26);
  message->message_id[26] = '\0';
  offset += 26;
  message->role = (tiny_role_t)data[offset++];
  message->metadata.memory_tier = (tiny_memory_tier_t)data[offset++];
  session_len = tiny_codec_get_u32(data + offset);
  offset += 4;
  content_len = tiny_codec_get_u32(data + offset);
  offset += 4;
  tool_len = tiny_codec_get_u32(data + offset);
  offset += 4;
  tag_count = tiny_codec_get_u32(data + offset);
  offset += 4;
  if (!tiny_message_id_valid(message->message_id)
      || (message->role < TINY_ROLE_USER || message->role > TINY_ROLE_SYSTEM)
      || (message->metadata.memory_tier < TINY_TIER_SHORT
          || message->metadata.memory_tier > TINY_TIER_DISCARD)
      || session_len == 0 || session_len > TINY_SESSION_ID_MAX || content_len == 0
      || content_len > TINY_CONTENT_MAX || tool_len > TINY_TOOL_CALL_ID_MAX
      || tag_count > TINY_TAG_COUNT_MAX || offset + session_len + content_len + tool_len > length)
    {
      return false;
    }
  message->session_id = copy_field(data + offset, session_len);
  offset += session_len;
  message->content = copy_field(data + offset, content_len);
  offset += content_len;
  if (tool_len > 0)
    {
      message->metadata.tool_call_id = copy_field(data + offset, tool_len);
      offset += tool_len;
    }
  if (message->session_id == NULL || message->content == NULL
      || (tool_len && !message->metadata.tool_call_id))
    {
      return deserialize_failed(message);
    }
  if (tag_count > 0)
    {
      message->metadata.tags = calloc(tag_count, sizeof(char *));
      if (message->metadata.tags == NULL)
        {
          return deserialize_failed(message);
        }
      for (index = 0; index < tag_count; ++index)
        {
          uint32_t tag_len;
          if (offset + 4 > length)
            {
              return deserialize_failed(message);
            }
          tag_len = tiny_codec_get_u32(data + offset);
          offset += 4;
          if (tag_len == 0 || tag_len > TINY_TAG_MAX || offset + tag_len > length)
            {
              return deserialize_failed(message);
            }
          message->metadata.tags[index] = copy_field(data + offset, tag_len);
          offset += tag_len;
          if (message->metadata.tags[index] == NULL)
            {
              return deserialize_failed(message);
            }
          message->metadata.tag_count++;
        }
    }
  if (offset != length || !tiny_utf8_valid(message->session_id, session_len)
      || !tiny_utf8_valid(message->content, content_len))
    {
      return deserialize_failed(message);
    }
  return true;
}

static tiny_store_status_t
append_record_to(tiny_sys_file_t *file,
                 uint8_t type,
                 uint64_t sequence,
                 const unsigned char *payload,
                 uint32_t payload_length,
                 bool flush)
{
  unsigned char header[RECORD_HEADER_SIZE] = { 0 };
  memcpy(header, "TCMR", 4);
  header[4] = RECORD_VERSION;
  header[5] = type;
  tiny_codec_put_u32(header + 8, payload_length);
  tiny_codec_put_u64(header + 12, sequence);
  tiny_codec_put_u32(header + 20, tiny_codec_crc32(payload, payload_length));
  if (!tiny_sys_file_seek(file, 0, TINY_SYS_SEEK_END) || !write_all(file, header, sizeof(header))
      || !write_all(file, payload, payload_length) || (flush && !tiny_sys_file_flush(file)))
    {
      return TINY_STORE_IO_ERROR;
    }
  return TINY_STORE_OK;
}

static tiny_store_status_t
apply_put(tiny_store_t *store, tiny_message_t *message, uint64_t sequence)
{
  tiny_store_entry_t *entry = find_entry(store, message->message_id);
  if (entry != NULL)
    {
      return TINY_STORE_CORRUPT;
    }
  if (!ensure_entries(store))
    {
      return TINY_STORE_NO_MEMORY;
    }
  entry = &store->entries[store->entry_count++];
  memset(entry, 0, sizeof(*entry));
  entry->message = *message;
  tiny_message_init(message);
  entry->sequence = sequence;
  return TINY_STORE_OK;
}

static void
clear_entries(tiny_store_t *store)
{
  size_t index;
  for (index = 0; index < store->entry_count; ++index)
    {
      tiny_message_free(&store->entries[index].message);
    }
  store->entry_count = 0;
  store->last_sequence = 0;
}

static tiny_store_status_t
recover_torn_tail(tiny_store_t *store, uint64_t boundary)
{
  if (boundary > INT64_MAX
      || !tiny_sys_file_seek(&store->log, (int64_t)boundary, TINY_SYS_SEEK_BEGIN)
      || !tiny_sys_file_truncate(&store->log) || !tiny_sys_file_flush(&store->log))
    {
      return TINY_STORE_IO_ERROR;
    }
  store->recovered_tail = true;
  set_error(store, "an incomplete trailing record was truncated during recovery");
  return TINY_STORE_OK;
}

static tiny_store_status_t
replay_log(tiny_store_t *store, uint64_t starting_offset)
{
  uint64_t file_size, boundary = starting_offset;
  if (starting_offset > INT64_MAX
      || !tiny_sys_file_seek(&store->log, (int64_t)starting_offset, TINY_SYS_SEEK_BEGIN)
      || !tiny_sys_file_size(&store->log, &file_size))
    {
      return TINY_STORE_IO_ERROR;
    }
  while (boundary < file_size)
    {
      unsigned char header[RECORD_HEADER_SIZE], *payload = NULL;
      uint32_t actual = 0, payload_length, expected_crc;
      uint64_t sequence;
      tiny_store_status_t status = TINY_STORE_OK;
      if (!read_all(&store->log, header, sizeof(header), &actual))
        {
          return TINY_STORE_IO_ERROR;
        }
      if (actual < sizeof(header))
        {
          return recover_torn_tail(store, boundary);
        }
      if (memcmp(header, "TCMR", 4) != 0 || header[4] != RECORD_VERSION
          || (header[5] != RECORD_PUT && header[5] != RECORD_DEL && header[5] != RECORD_MARKER))
        {
          return TINY_STORE_CORRUPT;
        }
      payload_length = tiny_codec_get_u32(header + 8);
      sequence = tiny_codec_get_u64(header + 12);
      expected_crc = tiny_codec_get_u32(header + 20);
      if (payload_length > RECORD_PAYLOAD_MAX
          || (header[5] != RECORD_MARKER && sequence <= store->last_sequence))
        {
          return TINY_STORE_CORRUPT;
        }
      payload = malloc(payload_length == 0 ? 1 : payload_length);
      if (payload == NULL)
        {
          return TINY_STORE_NO_MEMORY;
        }
      if (!read_all(&store->log, payload, payload_length, &actual))
        {
          free(payload);
          return TINY_STORE_IO_ERROR;
        }
      if (actual < payload_length)
        {
          free(payload);
          return recover_torn_tail(store, boundary);
        }
      if (tiny_codec_crc32(payload, payload_length) != expected_crc)
        {
          free(payload);
          return TINY_STORE_CORRUPT;
        }
      switch (header[5])
        {
        case RECORD_MARKER:
          if (boundary != 0 || sequence != 0 || payload_length != sizeof(store->generation))
            {
              free(payload);
              return TINY_STORE_CORRUPT;
            }
          memcpy(store->generation, payload, sizeof(store->generation));
          store->has_generation = true;
          break;

        case RECORD_PUT:
          {
            tiny_message_t message;
            if (!deserialize_put(payload, payload_length, &message))
              {
                free(payload);
                return TINY_STORE_CORRUPT;
              }
            status = apply_put(store, &message, sequence);
            tiny_message_free(&message);
          }
          break;

        case RECORD_DEL:
          {
            char id[27];
            tiny_store_entry_t *entry;
            if (payload_length != 34)
              {
                free(payload);
                return TINY_STORE_CORRUPT;
              }
            memcpy(id, payload, 26);
            id[26] = '\0';
            entry = find_entry(store, id);
            if (entry == NULL || entry->deleted || !tiny_message_id_valid(id))
              {
                free(payload);
                return TINY_STORE_CORRUPT;
              }
            entry->deleted = true;
          }
          break;

        default:
          free(payload);
          return TINY_STORE_CORRUPT;
        }
      free(payload);
      if (status != TINY_STORE_OK)
        {
          return status;
        }
      if (header[5] != RECORD_MARKER)
        {
          store->last_sequence = sequence;
        }
      boundary += RECORD_HEADER_SIZE + payload_length;
    }
  return TINY_STORE_OK;
}

static tiny_store_status_t
write_snapshot(tiny_store_t *store)
{
  char *temporary = tiny_sys_path_join(store->directory, "snapshot.tmp");
  tiny_sys_file_t file;
  unsigned char data[SNAPSHOT_HEADER_SIZE] = { 0 };
  uint64_t size;
  bool ok;
  size_t index;
  if (temporary == NULL)
    {
      return TINY_STORE_NO_MEMORY;
    }
  if (tiny_sys_file_open(&file, temporary, TINY_SYS_FILE_WRITE_NEW) != TINY_SYS_OK)
    {
      free(temporary);
      return TINY_STORE_IO_ERROR;
    }
  if (!tiny_sys_file_size(&store->log, &size))
    {
      tiny_sys_file_close(&file);
      free(temporary);
      return TINY_STORE_IO_ERROR;
    }
  memcpy(data, "TCMS", 4);
  data[4] = 1;
  memcpy(data + 8, store->generation, 16);
  tiny_codec_put_u64(data + 24, store->last_sequence);
  tiny_codec_put_u64(data + 32, size);
  tiny_codec_put_u64(data + 40, (uint64_t)store->entry_count);
  tiny_codec_put_u32(data + 48, tiny_codec_crc32(data, 48));
  ok = write_all(&file, data, sizeof(data));
  for (index = 0; ok && index < store->entry_count; ++index)
    {
      unsigned char prefix[16], *payload = NULL;
      uint32_t length = 0;
      if (!serialize_put(&store->entries[index].message, &payload, &length))
        {
          ok = false;
          break;
        }
      tiny_codec_put_u64(prefix, store->entries[index].sequence);
      tiny_codec_put_u32(prefix + 8, length);
      tiny_codec_put_u32(prefix + 12, tiny_codec_crc32(payload, length));
      ok = write_all(&file, prefix, sizeof(prefix)) && write_all(&file, payload, length);
      free(payload);
    }
  if (ok)
    {
      ok = tiny_sys_file_flush(&file);
    }
  tiny_sys_file_close(&file);
  if (ok)
    {
      ok = tiny_sys_path_replace(temporary, store->snapshot_path);
    }
  free(temporary);
  store->snapshot_valid = ok;
  return ok ? TINY_STORE_OK : TINY_STORE_IO_ERROR;
}

static bool
log_generation_matches(tiny_store_t *store, const unsigned char generation[16])
{
  unsigned char header[RECORD_HEADER_SIZE], payload[16];
  uint32_t actual = 0;
  if (!tiny_sys_file_seek(&store->log, 0, TINY_SYS_SEEK_BEGIN)
      || !read_all(&store->log, header, sizeof(header), &actual) || actual != sizeof(header)
      || memcmp(header, "TCMR", 4) != 0 || header[4] != RECORD_VERSION || header[5] != RECORD_MARKER
      || tiny_codec_get_u64(header + 12) != 0 || tiny_codec_get_u32(header + 8) != 16
      || !read_all(&store->log, payload, sizeof(payload), &actual) || actual != sizeof(payload)
      || tiny_codec_get_u32(header + 20) != tiny_codec_crc32(payload, sizeof(payload)))
    {
      return false;
    }
  return memcmp(payload, generation, 16) == 0;
}

static bool
snapshot_invalid(tiny_store_t *store, tiny_sys_file_t *file, bool clear_loaded_entries)
{
  if (clear_loaded_entries)
    {
      clear_entries(store);
      store->has_generation = false;
    }
  set_error(store, "snapshot validation failed; the active log was replayed from the beginning");
  tiny_sys_file_close(file);
  return false;
}

static bool
load_snapshot(tiny_store_t *store, uint64_t *log_offset)
{
  tiny_sys_file_t file;
  unsigned char header[SNAPSHOT_HEADER_SIZE];
  uint32_t actual = 0;
  uint64_t count, index, expected_sequence, offset;
  uint64_t log_size;
  if (tiny_sys_file_open(&file, store->snapshot_path, TINY_SYS_FILE_READ) != TINY_SYS_OK)
    {
      return false;
    }
  if (!read_all(&file, header, sizeof(header), &actual) || actual != sizeof(header)
      || memcmp(header, "TCMS", 4) != 0 || header[4] != 1
      || tiny_codec_get_u32(header + 48) != tiny_codec_crc32(header, 48)
      || !tiny_sys_file_size(&store->log, &log_size))
    {
      return snapshot_invalid(store, &file, false);
    }
  expected_sequence = tiny_codec_get_u64(header + 24);
  offset = tiny_codec_get_u64(header + 32);
  count = tiny_codec_get_u64(header + 40);
  if (offset > log_size || count > SIZE_MAX || !log_generation_matches(store, header + 8))
    {
      return snapshot_invalid(store, &file, false);
    }
  memcpy(store->generation, header + 8, 16);
  store->has_generation = true;
  for (index = 0; index < count; ++index)
    {
      unsigned char prefix[16], *payload;
      uint32_t length, crc;
      tiny_message_t message;
      tiny_store_status_t status;
      if (!read_all(&file, prefix, sizeof(prefix), &actual) || actual != sizeof(prefix))
        {
          return snapshot_invalid(store, &file, true);
        }
      length = tiny_codec_get_u32(prefix + 8);
      crc = tiny_codec_get_u32(prefix + 12);
      if (length > RECORD_PAYLOAD_MAX)
        {
          return snapshot_invalid(store, &file, true);
        }
      payload = malloc(length ? length : 1);
      if (!payload)
        {
          return snapshot_invalid(store, &file, true);
        }
      if (!read_all(&file, payload, length, &actual) || actual != length
          || tiny_codec_crc32(payload, length) != crc
          || !deserialize_put(payload, length, &message))
        {
          free(payload);
          return snapshot_invalid(store, &file, true);
        }
      free(payload);
      status = apply_put(store, &message, tiny_codec_get_u64(prefix));
      tiny_message_free(&message);
      if (status != TINY_STORE_OK)
        {
          return snapshot_invalid(store, &file, true);
        }
      store->last_sequence = tiny_codec_get_u64(prefix);
    }
  if (store->last_sequence != expected_sequence)
    {
      return snapshot_invalid(store, &file, true);
    }
  tiny_sys_file_close(&file);
  *log_offset = offset;
  store->snapshot_valid = true;
  return true;
}

static tiny_store_status_t
store_open_failed(tiny_store_t *store, char *lock_path, tiny_store_status_t status)
{
  free(lock_path);
  tiny_store_close(store);
  return status;
}

tiny_store_status_t
tiny_store_open(tiny_store_t **output, const char *directory)
{
  tiny_store_t *store;
  char *lock_path = NULL;
  tiny_sys_status_t sys_status;
  tiny_store_status_t status;
  if (output == NULL)
    {
      return TINY_STORE_IO_ERROR;
    }
  *output = NULL;
  if (directory == NULL || directory[0] == '\0')
    {
      return TINY_STORE_IO_ERROR;
    }
  store = calloc(1, sizeof(*store));
  if (store == NULL)
    {
      return TINY_STORE_NO_MEMORY;
    }
  tiny_sys_file_init(&store->lock);
  tiny_sys_file_init(&store->log);
  store->directory = copy_field((const unsigned char *)directory, strlen(directory));
  store->log_path = tiny_sys_path_join(directory, "active.log");
  store->snapshot_path = tiny_sys_path_join(directory, "snapshot.bin");
  lock_path = tiny_sys_path_join(directory, "store.lock");
  if (!store->directory || !store->log_path || !store->snapshot_path || !lock_path)
    {
      return store_open_failed(store, lock_path, TINY_STORE_NO_MEMORY);
    }
  if (tiny_sys_directory_create(directory) != TINY_SYS_OK)
    {
      return store_open_failed(store, lock_path, TINY_STORE_IO_ERROR);
    }
  sys_status = tiny_sys_file_open(&store->lock, lock_path, TINY_SYS_FILE_LOCK);
  free(lock_path);
  lock_path = NULL;
  if (sys_status != TINY_SYS_OK)
    {
      status = sys_status == TINY_SYS_BUSY ? TINY_STORE_BUSY : TINY_STORE_IO_ERROR;
      return store_open_failed(store, lock_path, status);
    }
  if (tiny_sys_file_open(&store->log, store->log_path, TINY_SYS_FILE_READ_WRITE_CREATE)
      != TINY_SYS_OK)
    {
      return store_open_failed(store, lock_path, TINY_STORE_IO_ERROR);
    }
  {
    uint64_t replay_offset = 0;
    (void)load_snapshot(store, &replay_offset);
    status = replay_log(store, replay_offset);
  }
  if (status != TINY_STORE_OK)
    {
      return store_open_failed(store, lock_path, status);
    }
  *output = store;
  return TINY_STORE_OK;
}

void
tiny_store_close(tiny_store_t *store)
{
  size_t index;
  if (!store)
    {
      return;
    }
  tiny_sys_file_close(&store->log);
  tiny_sys_file_close(&store->lock);
  for (index = 0; index < store->entry_count; ++index)
    {
      tiny_message_free(&store->entries[index].message);
    }
  free(store->entries);
  free(store->directory);
  free(store->log_path);
  free(store->snapshot_path);
  free(store);
}

const char *
tiny_store_status_name(tiny_store_status_t status)
{
  switch (status)
    {
    case TINY_STORE_OK:
      return "OK";
    case TINY_STORE_NOT_FOUND:
      return "NOT_FOUND";
    case TINY_STORE_BUSY:
      return "BUSY";
    case TINY_STORE_IO_ERROR:
      return "IO_ERROR";
    case TINY_STORE_CORRUPT:
      return "CORRUPT_STORE";
    default:
      return "INTERNAL_ERROR";
    }
}

tiny_store_status_t
tiny_store_write(tiny_store_t *store, const tiny_message_t *input, tiny_message_t *written)
{
  tiny_message_t message, indexed;
  unsigned char *payload = NULL;
  uint32_t length = 0;
  tiny_store_status_t status;
  if (written != NULL)
    {
      tiny_message_init(written);
    }
  if (store == NULL || input == NULL || written == NULL || input->session_id == NULL
      || input->content == NULL)
    {
      return TINY_STORE_IO_ERROR;
    }
  tiny_message_init(&message);
  tiny_message_init(&indexed);
  if (tiny_message_copy(&message, input) != 0)
    {
      return TINY_STORE_NO_MEMORY;
    }
  message.timestamp_ms = (int64_t)now_milliseconds();
  make_ulid(store, (uint64_t)message.timestamp_ms, message.message_id);
  if (!serialize_put(&message, &payload, &length) || !ensure_entries(store)
      || tiny_message_copy(&indexed, &message) != 0)
    {
      free(payload);
      tiny_message_free(&message);
      tiny_message_free(&indexed);
      return TINY_STORE_NO_MEMORY;
    }
  status
      = append_record_to(&store->log, RECORD_PUT, store->last_sequence + 1, payload, length, true);
  free(payload);
  if (status != TINY_STORE_OK)
    {
      tiny_message_free(&message);
      tiny_message_free(&indexed);
      set_error(store, "failed to append and flush a PUT record");
      return status;
    }
  store->last_sequence++;
  memset(&store->entries[store->entry_count], 0, sizeof(store->entries[store->entry_count]));
  store->entries[store->entry_count].message = indexed;
  store->entries[store->entry_count].sequence = store->last_sequence;
  store->entry_count++;
  *written = message;
  return TINY_STORE_OK;
}

tiny_store_status_t
tiny_store_delete(tiny_store_t *store, const char *message_id, const char *session_id)
{
  tiny_store_entry_t *entry;
  unsigned char payload[34];
  tiny_store_status_t status;
  if (store == NULL || message_id == NULL || !tiny_message_id_valid(message_id))
    {
      return TINY_STORE_NOT_FOUND;
    }
  entry = find_entry(store, message_id);
  if (!entry || entry->deleted
      || (session_id && strcmp(entry->message.session_id, session_id) != 0))
    {
      return TINY_STORE_NOT_FOUND;
    }
  memcpy(payload, message_id, 26);
  tiny_codec_put_u64(payload + 26, now_milliseconds());
  status = append_record_to(
      &store->log, RECORD_DEL, store->last_sequence + 1, payload, sizeof(payload), true);
  if (status != TINY_STORE_OK)
    {
      set_error(store, "failed to append and flush a DEL record");
      return status;
    }
  store->last_sequence++;
  entry->deleted = true;
  return TINY_STORE_OK;
}

tiny_store_status_t
tiny_store_rebuild(tiny_store_t *store, tiny_rebuild_report_t *report)
{
  char *temporary;
  tiny_sys_file_t file;
  uint64_t before = 0, after = 0;
  size_t read_index, write_index = 0;
  uint64_t removed = 0, max_sequence = 0;
  tiny_store_status_t status = TINY_STORE_OK;
  unsigned char previous_generation[sizeof(store->generation)];
  bool previous_has_generation;
  bool log_replaced = false;
  if (report != NULL)
    {
      memset(report, 0, sizeof(*report));
    }
  if (store == NULL || report == NULL)
    {
      return TINY_STORE_IO_ERROR;
    }
  memcpy(previous_generation, store->generation, sizeof(previous_generation));
  previous_has_generation = store->has_generation;
  temporary = tiny_sys_path_join(store->directory, "active.tmp");
  if (!temporary)
    {
      return TINY_STORE_NO_MEMORY;
    }
  if (!tiny_sys_file_size(&store->log, &before)
      || tiny_sys_file_open(&file, temporary, TINY_SYS_FILE_WRITE_NEW) != TINY_SYS_OK)
    {
      free(temporary);
      return TINY_STORE_IO_ERROR;
    }
  if (!tiny_sys_random_bytes(store->generation, sizeof(store->generation)))
    {
      tiny_sys_file_close(&file);
      free(temporary);
      return TINY_STORE_IO_ERROR;
    }
  store->has_generation = true;
  status = append_record_to(
      &file, RECORD_MARKER, 0, store->generation, sizeof(store->generation), false);
  for (read_index = 0; read_index < store->entry_count; ++read_index)
    {
      if (status == TINY_STORE_OK && !store->entries[read_index].deleted)
        {
          unsigned char *payload = NULL;
          uint32_t length = 0;
          if (!serialize_put(&store->entries[read_index].message, &payload, &length))
            {
              status = TINY_STORE_NO_MEMORY;
              break;
            }
          status = append_record_to(
              &file, RECORD_PUT, store->entries[read_index].sequence, payload, length, false);
          free(payload);
          if (status != TINY_STORE_OK)
            {
              break;
            }
          if (store->entries[read_index].sequence > max_sequence)
            {
              max_sequence = store->entries[read_index].sequence;
            }
        }
      else
        {
          removed++;
        }
    }
  if (status == TINY_STORE_OK && !tiny_sys_file_flush(&file))
    {
      status = TINY_STORE_IO_ERROR;
    }
  tiny_sys_file_close(&file);
  if (status == TINY_STORE_OK)
    {
      tiny_sys_file_close(&store->log);
      if (!tiny_sys_path_replace(temporary, store->log_path))
        {
          status = TINY_STORE_IO_ERROR;
        }
      else
        {
          log_replaced = true;
        }
      if (tiny_sys_file_open(&store->log, store->log_path, TINY_SYS_FILE_READ_WRITE_CREATE)
          != TINY_SYS_OK)
        {
          status = TINY_STORE_IO_ERROR;
        }
    }
  if (status == TINY_STORE_OK)
    {
      for (read_index = 0; read_index < store->entry_count; ++read_index)
        {
          if (store->entries[read_index].deleted)
            {
              tiny_message_free(&store->entries[read_index].message);
            }
          else
            {
              if (write_index != read_index)
                {
                  store->entries[write_index] = store->entries[read_index];
                }
              write_index++;
            }
        }
      store->entry_count = write_index;
      store->last_sequence = max_sequence;
      (void)tiny_sys_file_size(&store->log, &after);
      report->messages_indexed = write_index;
      report->records_removed = removed;
      report->bytes_before = before;
      report->bytes_after = after;
      status = write_snapshot(store);
    }
  if (status != TINY_STORE_OK)
    {
      if (!log_replaced)
        {
          memcpy(store->generation, previous_generation, sizeof(store->generation));
          store->has_generation = previous_has_generation;
        }
      set_error(store,
                "compaction did not complete; the previous valid log is retained when replacement "
                "had not begun");
    }
  free(temporary);
  return status;
}

void
tiny_store_health(const tiny_store_t *store, tiny_store_health_t *health)
{
  uint64_t size;
  size_t index;
  if (health == NULL)
    {
      return;
    }
  memset(health, 0, sizeof(*health));
  if (store == NULL)
    {
      return;
    }
  if (tiny_sys_file_size(&store->log, &size))
    {
      health->log_size = size;
    }
  {
    tiny_sys_file_t file;
    if (tiny_sys_file_open(&file, store->snapshot_path, TINY_SYS_FILE_READ) == TINY_SYS_OK)
      {
        if (tiny_sys_file_size(&file, &size))
          {
            health->snapshot_size = size;
          }
        tiny_sys_file_close(&file);
      }
  }
  for (index = 0; index < store->entry_count; ++index)
    {
      if (store->entries[index].deleted)
        {
          health->deleted_messages++;
        }
      else
        {
          health->active_messages++;
        }
    }
  health->last_sequence = store->last_sequence;
  health->recovered_tail = store->recovered_tail;
  health->snapshot_valid = store->snapshot_valid;
  snprintf(health->last_error, sizeof(health->last_error), "%s", store->last_error);
}

size_t
tiny_store_entry_count(const tiny_store_t *store)
{
  return store == NULL ? 0 : store->entry_count;
}
const tiny_store_entry_t *
tiny_store_entry(const tiny_store_t *store, size_t index)
{
  return store != NULL && index < store->entry_count ? &store->entries[index] : NULL;
}