/*
 * Copyright (c) 2006-Present, Redis Ltd.
 * All rights reserved.
 *
 * Licensed under your choice of the Redis Source Available License 2.0
 * (RSALv2); or (b) the Server Side Public License v1 (SSPLv1); or (c) the
 * GNU Affero General Public License v3 (AGPLv3).
*/

#include "internal.h"
#include "util.h"
#include "redismock.h"
#include "config.h"

#include <string>
#include <map>
#include <vector>
#include <iostream>
#include <list>
#include <set>
#include <cstdarg>
#include <cstring>
#include <cerrno>
#include <cmath>
#include <cstdlib>
#include <unistd.h>
#include <sys/wait.h>
#include <climits>
#include <cassert>
#include <mutex>

#define __ignore__(X) \
    do { \
        int rc = (X); \
        if (rc == -1) \
            ; \
    } while(0)

static std::mutex RMCK_GlobalLock;

// KeyMeta mock storage
static std::map<std::string, std::map<RedisModuleKeyMetaClassId, uint64_t>> keyMetaStorage;
static RedisModuleKeyMetaClassId nextClassId = 1;
static std::map<RedisModuleKeyMetaClassId, RedisModuleKeyMetaClassConfig> classConfigs;
static std::map<std::string, RedisModuleKeyMetaClassId> classNames;

std::string HashValue::Key::makeKey() const {
  if (flags & REDISMODULE_HASH_CFIELDS) {
    return std::string(cstr);
  } else {
    return *rstr;
  }
}

void HashValue::add(const char *key, const char *value, int mode) {
  if (mode & REDISMODULE_HASH_XX) {
    if (m_map.find(key) == m_map.end()) {
      return;
    }
  } else if (mode & REDISMODULE_HASH_NX) {
    if (m_map.find(key) != m_map.end()) {
      return;
    }
  }
  m_map[key].value = value;
}

bool HashValue::hexpire(const HashValue::Key &k, mstime_t expireAt) {
  const char *skey;
  if (k.flags & REDISMODULE_HASH_CFIELDS) {
    skey = k.cstr;
  } else {
    skey = (*k.rstr).c_str();
  }

  auto itKey = m_map.find(skey);
  if (expireAt == REDISMODULE_NO_EXPIRE || itKey == m_map.end()) {
    return false;
  }

  // if field had a different expiration point, remove it
  if (itKey->second.expirationIt != m_expiration.end()) {
    if (itKey->second.expirationIt->first == expireAt) {
      return true;
    }
    itKey->second.expirationIt->second.erase(skey);
    itKey->second.expirationIt = m_expiration.end();
  }
  // add the new expiration point, both to expiration map and to key
  // TODO: find out why try_emplace doesn't compile on some environments
  auto it = m_expiration.find(expireAt);
  if (it == m_expiration.end()) {
    it = m_expiration.emplace(expireAt, std::unordered_set<std::string>()).first;
  }
  itKey->second.expirationIt = it;
  it->second.insert(skey);
  return true;
}

Optional<mstime_t> HashValue::min_expire_time() const {
  if (m_expiration.empty()) {
    return boost::none;
  }
  return m_expiration.begin()->first;
}

Optional<mstime_t> HashValue::get_expire_time(const Key &k) const {
  const char *skey;
  if (k.flags & REDISMODULE_HASH_CFIELDS) {
    skey = k.cstr;
  } else {
    skey = (*k.rstr).c_str();
  }

  auto it = m_map.find(skey);
  if (it == m_map.end() || it->second.expirationIt == m_expiration.end()) {
    return boost::none;
  }
  return it->second.expirationIt->first;
}

void HashValue::hset(const HashValue::Key &k, const RedisModuleString *value) {
  const char *skey;
  if (k.flags & REDISMODULE_HASH_CFIELDS) {
    skey = k.cstr;
  } else {
    skey = (*k.rstr).c_str();
  }

  if (value == REDISMODULE_HASH_DELETE) {
    m_map.erase(skey);
    return;
  }

  add(skey, value->c_str(), k.flags);

  if (k.flags & REDISMODULE_HASH_XX) {
    if (m_map.find(skey) == m_map.end()) {
      return;
    }
  } else if (k.flags & REDISMODULE_HASH_NX) {
    if (m_map.find(skey) != m_map.end()) {
      return;
    }
  }
  auto& e = m_map[skey];
  e.value = *value;
  e.expirationIt = m_expiration.end();
}

const std::string *HashValue::hget(const Key &e) const {
  auto entry = m_map.find(e.makeKey());
  if (entry == m_map.end()) {
    return NULL;
  }
  return &entry->second.value;
}

RedisModuleString **HashValue::kvarray(RedisModuleCtx *allocctx) const {
  std::vector<RedisModuleString *> ll;
  for (auto it : m_map) {
    RedisModuleString *keyp = new RedisModuleString(it.first);
    RedisModuleString *valp = new RedisModuleString(it.second.value);
    ll.push_back(keyp);
    ll.push_back(valp);
    allocctx->addPointer(keyp);
    allocctx->addPointer(valp);
  }

  RedisModuleString **strs = (RedisModuleString **)calloc(ll.size(), sizeof(*strs));
  std::copy(ll.begin(), ll.end(), strs);
  return strs;
}

RedisModuleKey *RMCK_OpenKey(RedisModuleCtx *ctx, RedisModuleString *s, int mode) {
  // Look up in db:
  Value *vv = ctx->db->get(s);
  // Always return a valid key handle, matching real Redis behavior.
  // For non-existent keys, ref (vv) will be NULL, and KeyType returns EMPTY.
  return new RedisModuleKey(ctx, s, vv, mode);
}

int RMCK_DeleteKey(RedisModuleKey *k) {
  if (!k->ref) {
    return REDISMODULE_OK;
  }
  // Delete the key from the db
  k->parent->db->erase(k->key);
  k->ref->decref();
  k->ref = NULL;
  return REDISMODULE_OK;
}

void RMCK_CloseKey(RedisModuleKey *k) {
  k->parent->notifyRemoved(k);
  delete k;
}

int RMCK_KeyType(RedisModuleKey *k) {
  if (k->ref == NULL) {
    return REDISMODULE_KEYTYPE_EMPTY;
  } else {
    return k->ref->typecode();
  }
}

size_t RMCK_ValueLength(RedisModuleKey *k) {
  if (k->ref == NULL) {
    return 0;
  } else {
    return k->ref->size();
  }
}

mstime_t RMCK_HashFieldMinExpire(RedisModuleKey *k) {
  auto hv = dynamic_cast<HashValue *>(k->ref);
  if (!hv) {
    return REDISMODULE_NO_EXPIRE;
  }
  const auto minExpire = hv->min_expire_time();
  return minExpire ? *minExpire : REDISMODULE_NO_EXPIRE;
}

/** String functions */
RedisModuleString *RMCK_CreateString(RedisModuleCtx *ctx, const char *s, size_t n) {
  RedisModuleString *rs = new RedisModuleString(s, n);
  if (ctx) {
    ctx->addPointer(rs);
  }
  return rs;
}

RedisModuleString *RMCK_CreateStringFromString(RedisModuleCtx *ctx, RedisModuleString *src) {
  size_t n;
  const char *s = RedisModule_StringPtrLen(src, &n);
  return RedisModule_CreateString(ctx, s, n);
}

RedisModuleString *RMCK_CreateStringPrintf(RedisModuleCtx *ctx, const char *fmt, ...) {
  va_list ap;
  va_start(ap, fmt);
  char *outp = NULL;
  __ignore__(vasprintf(&outp, fmt, ap));
  va_end(ap);
  RedisModuleString *ret = RMCK_CreateString(ctx, outp, strlen(outp));
  free(outp);
  return ret;
}

void RMCK_FreeString(RedisModuleCtx *ctx, RedisModuleString *s) {
  s->decref();
  if (ctx) {
    ctx->notifyRemoved(s);
  }
}

void RMCK_RetainString(RedisModuleCtx *ctx, RedisModuleString *s) {
  s->incref();
}

RedisModuleString *RMCK_HoldString(RedisModuleCtx *ctx, RedisModuleString *s) {
  RMCK_RetainString(ctx, s);
  return s;
}

void RMCK_TrimStringAllocation(RedisModuleString *s) {
  s->trim();
}

void RMCK_SetModuleOptions(RedisModuleCtx *ctx, int options) {
}


const char *RMCK_StringPtrLen(RedisModuleString *s, size_t *len) {
  if (len) {
    *len = s->size();
  }
  return s->c_str();
}

int RMCK_StringToDouble(RedisModuleString *s, double *outval) {
  char *eptr = NULL;
  double value = strtod(s->c_str(), &eptr);

  if (s->empty() || isspace(s->at(0))) {
    return REDISMODULE_ERR;
  }
  if (eptr - s->c_str() != s->size()) {
    return REDISMODULE_ERR;
  }
  if ((errno == ERANGE && (value == HUGE_VAL || value == -HUGE_VAL || value == 0)) ||
      std::isnan(value)) {
    return REDISMODULE_ERR;
  }
  *outval = value;
  return REDISMODULE_OK;
}

static int string2ll(const char *s, size_t slen, long long *value) {
  const char *p = s;
  size_t plen = 0;
  int negative = 0;
  unsigned long long v;

  if (plen == slen) return 0;

  /* Special case: first and only digit is 0. */
  if (slen == 1 && p[0] == '0') {
    if (value != NULL) *value = 0;
    return 1;
  }

  if (p[0] == '-') {
    negative = 1;
    p++;
    plen++;

    /* Abort on only a negative sign. */
    if (plen == slen) return 0;
  }

  /* First digit should be 1-9, otherwise the string should just be 0. */
  if (p[0] >= '1' && p[0] <= '9') {
    v = p[0] - '0';
    p++;
    plen++;
  } else if (p[0] == '0' && slen == 1) {
    *value = 0;
    return 1;
  } else {
    return 0;
  }

  while (plen < slen && p[0] >= '0' && p[0] <= '9') {
    if (v > (ULLONG_MAX / 10)) /* Overflow. */
      return 0;
    v *= 10;

    if (v > (ULLONG_MAX - (p[0] - '0'))) /* Overflow. */
      return 0;
    v += p[0] - '0';

    p++;
    plen++;
  }

  /* Return if not all bytes were used. */
  if (plen < slen) return 0;

  if (negative) {
    if (v > ((unsigned long long)(-(LLONG_MIN + 1)) + 1)) /* Overflow. */
      return 0;
    if (value != NULL) *value = -v;
  } else {
    if (v > LLONG_MAX) /* Overflow. */
      return 0;
    if (value != NULL) *value = v;
  }
  return 1;
}

int RMCK_StringToLongLong(RedisModuleString *s, long long *l) {
  if (string2ll(s->c_str(), s->size(), l)) {
    return REDISMODULE_OK;
  }
  return REDISMODULE_ERR;
}

/** Hash functions */
#define ENTRY_OK 1
#define ENTRY_DONE 0
#define ENTRY_ERROR -1
// Retrieves the hash value key and the following argument, and stores them in the provided pointers
static int getNextEntry(va_list &ap, HashValue::Key &e, void **vpp) {
  void *kp = va_arg(ap, void *);
  if (!kp) {
    return ENTRY_DONE;
  }
  *vpp = va_arg(ap, RedisModuleString *);
  if (!vpp) {
    return ENTRY_ERROR;
  }
  e.rawkey = kp;
  return ENTRY_OK;
}

int RMCK_HashSet(RedisModuleKey *key, int flags, ...) {
  bool wasEmpty = false;
  if (!key->ref) {
    // Empty...
    wasEmpty = true;
    key->ref = new HashValue(key->key);
    key->ref->incref();
  } else if (key->ref->typecode() != REDISMODULE_KEYTYPE_HASH) {
    return REDISMODULE_ERR;
  }

  HashValue *hv = static_cast<HashValue *>(key->ref);
  va_list ap;
  va_start(ap, flags);
  HashValue::Key e(flags);

  while (true) {
    RedisModuleString *vp;
    int rc = getNextEntry(ap, e, (void **)&vp);
    if (rc == ENTRY_DONE) {
      break;
    } else if (rc == ENTRY_ERROR) {
      goto error;
    } else {
      hv->hset(e, vp);
    }
  }
  va_end(ap);

  if (wasEmpty) {
    // Assign this value to the main DB:
    key->parent->db->set(hv);
    // and delete the original reference
    hv->decref();
  }
  return REDISMODULE_OK;

error:
  if (wasEmpty) {
    delete key->ref;
    key->ref = NULL;
  }
  return REDISMODULE_ERR;
}

int RMCK_HashGet(RedisModuleKey *key, int flags, ...) {
  va_list ap;
  va_start(ap, flags);

  HashValue::Key e(flags);
  if (!key->ref || key->ref->typecode() != REDISMODULE_KEYTYPE_HASH) {
    return REDISMODULE_ERR;
  }

  if ((flags & REDISMODULE_HASH_EXISTS) && (flags & REDISMODULE_HASH_EXPIRE_TIME))
    return REDISMODULE_ERR;

  HashValue *hv = static_cast<HashValue *>(key->ref);

  while (true) {
    void *vpp = NULL;
    e.rawkey = va_arg(ap, void *);
    if (!e.rawkey) {
      break;
    }

    // Get the key
    const std::string *value = hv->hget(e);
    if (flags & REDISMODULE_HASH_EXISTS) {
      int *exists = va_arg(ap, int *);
      *exists = value != NULL;
    } else if (flags & REDISMODULE_HASH_EXPIRE_TIME) {
      mstime_t *ms = va_arg(ap, mstime_t *);
      *ms = hv->get_expire_time(e).value_or(REDISMODULE_NO_EXPIRE);
    } else {
      RedisModuleString **value_ptr = va_arg(ap, RedisModuleString* *);
      RedisModuleString *newv = NULL;
      if (value) {
        newv = new RedisModuleString(*value);
        key->parent->addPointer(newv);
      }
      *reinterpret_cast<RedisModuleString **>(value_ptr) = newv;
    }
  }
  va_end(ap);
  return REDISMODULE_OK;
}

RedisModuleString **RMCK_HashGetAll(RedisModuleKey *key) {
  if (key->ref == NULL || key->ref->typecode() != REDISMODULE_KEYTYPE_HASH) {
    return NULL;
  }
  auto *hv = static_cast<HashValue *>(key->ref);
  return hv->kvarray(key->parent);
}

typedef enum {
  LL_DEBUG = 0,  // nlb
  LL_VERBOSE,
  LL_NOTICE,
  LL_WARNING
} LogLevel;

int RMCK_LogLevel = LL_NOTICE;
static int loglevelFromString(const char *s) {
  switch (*s) {
    case 'd':
    case 'D':
      return LL_DEBUG;
    case 'v':
    case 'V':
      return LL_VERBOSE;
    case 'n':
    case 'N':
      return LL_NOTICE;
    case 'w':
    case 'W':
      return LL_WARNING;
    default:
      return LL_DEBUG;
  }
}
void RMCK_Log(RedisModuleCtx *ctx, const char *level, const char *fmt, ...) {
  int ilevel = loglevelFromString(level);
  if (ilevel < RMCK_LogLevel) {
    return;
  }
  va_list ap;
  va_start(ap, fmt);
  vfprintf(stderr, fmt, ap);
  va_end(ap);
  fputc('\n', stderr);
}

int RMCK_StringCompare(RedisModuleString *a, RedisModuleString *b) {
  return a->compare((std::string)*b);
}

/** MODULE TYPES */
RedisModuleType *RMCK_CreateDataType(RedisModuleCtx *ctx, const char *name, int encver,
                                     RedisModuleTypeMethods *meths) {
  if (Datatype::typemap.find(name) != Datatype::typemap.end()) {
    return NULL;
  }
  RedisModuleType *ret = new RedisModuleType();
  ret->name = name;
  ret->encver = encver;
  ret->typemeths = *meths;
  Datatype::typemap[name] = ret;
  return ret;
}

int RMCK_ModuleTypeSetValue(RedisModuleKey *k, RedisModuleType *mt, void *value) {
  ModuleValue *mv = NULL;
  if (!k->ref) {
    mv = new ModuleValue(k->key, mt);
    k->parent->db->set(mv);
    mv->decref();
  } else if (k->ref->typecode() != REDISMODULE_KEYTYPE_MODULE) {
    return REDISMODULE_ERR;
  }
  mv->value = value;
  return REDISMODULE_OK;
}

RedisModuleType *RMCK_ModuleTypeGetType(RedisModuleKey *key) {
  if (key->ref == NULL || key->ref->typecode() != REDISMODULE_KEYTYPE_MODULE) {
    return NULL;
  }
  return static_cast<ModuleValue *>(key->ref)->mtype;
}

void *RMCK_ModuleTypeGetValue(RedisModuleKey *key) {
  if (key->ref == NULL || key->ref->typecode() != REDISMODULE_KEYTYPE_MODULE) {
    return NULL;
  }
  return static_cast<ModuleValue *>(key->ref)->value;
}

ModuleValue::~ModuleValue() {
  if (mtype->typemeths.free) {
    mtype->typemeths.free(value);
    value = NULL;
  }
}

Datatype::TypemapType Datatype::typemap;
RedisModuleCommand::CommandMap RedisModuleCommand::commands;

int RMCK_CreateCommand(RedisModuleCtx *ctx, const char *s, RedisModuleCmdFunc handler, const char *,
                       int, int, int) {
  if (RedisModuleCommand::commands.find(s) != RedisModuleCommand::commands.end()) {
    return REDISMODULE_ERR;
  }
  RedisModuleCommand *c = new RedisModuleCommand();
  c->name = s;
  c->handler = handler;
  RedisModuleCommand::commands[s] = c;
  return REDISMODULE_OK;
}

RedisModuleCommand *RMCK_GetCommand(RedisModuleCtx *ctx, const char *s) {
  auto it = RedisModuleCommand::commands.find(s);
  if (it == RedisModuleCommand::commands.end()) {
    return NULL;
  }
  return it->second;
}

int RMCK_CreateSubcommand(RedisModuleCommand *parent, const char *s, RedisModuleCmdFunc handler, const char *,
                       int, int, int) {
  if (!parent || parent->handler || parent->subcommands.find(s) != parent->subcommands.end()) {
    return REDISMODULE_ERR;
  }
  RedisModuleCommand *c = new RedisModuleCommand();
  c->name = s;
  c->handler = handler;
  parent->subcommands[s] = c;
  return REDISMODULE_OK;
}

// Internal assertion handler. We still expect to use the `RedisModule_Assert` macro.
static void RMCK__Assert(const char *estr, const char *file, int line) {
  throw std::runtime_error(std::string(estr) + " at " + file + ":" + std::to_string(line));
}

/** Allocators */
void *RMCK_Alloc(size_t n) {
  return malloc(n);
}

void RMCK_Free(void *p) {
  free(p);
}

void *RMCK_Calloc(size_t nmemb, size_t size) {
  return calloc(nmemb, size);
}

void *RMCK_Realloc(void *p, size_t n) {
  return realloc(p, n);
}

char *RMCK_Strdup(const char *s) {
  return strdup(s);
}

/** RDB Mock Operations */

void RMCK_SaveUnsigned(RedisModuleIO *io, uint64_t value) {
  if (!io) return;
  uint8_t *bytes = reinterpret_cast<uint8_t*>(&value);
  for (size_t i = 0; i < sizeof(uint64_t); i++) {
    io->buffer.push_back(bytes[i]);
  }
}

uint64_t RMCK_LoadUnsigned(RedisModuleIO *io) {
  if (!io || io->read_pos + sizeof(uint64_t) > io->buffer.size()) {
    if (io) io->error_flag = true;
    return 0;
  }
  uint64_t value = 0;
  uint8_t *bytes = reinterpret_cast<uint8_t*>(&value);
  for (size_t i = 0; i < sizeof(uint64_t); i++) {
    bytes[i] = io->buffer[io->read_pos++];
  }
  return value;
}

void RMCK_SaveSigned(RedisModuleIO *io, int64_t value) {
  if (!io) return;
  uint8_t *bytes = reinterpret_cast<uint8_t*>(&value);
  for (size_t i = 0; i < sizeof(int64_t); i++) {
    io->buffer.push_back(bytes[i]);
  }
}

int64_t RMCK_LoadSigned(RedisModuleIO *io) {
  if (!io || io->read_pos + sizeof(int64_t) > io->buffer.size()) {
    if (io) io->error_flag = true;
    return 0;
  }
  int64_t value = 0;
  uint8_t *bytes = reinterpret_cast<uint8_t*>(&value);
  for (size_t i = 0; i < sizeof(int64_t); i++) {
    bytes[i] = io->buffer[io->read_pos++];
  }
  return value;
}

void RMCK_SaveDouble(RedisModuleIO *io, double value) {
  if (!io) return;
  uint8_t *bytes = reinterpret_cast<uint8_t*>(&value);
  for (size_t i = 0; i < sizeof(double); i++) {
    io->buffer.push_back(bytes[i]);
  }
}

double RMCK_LoadDouble(RedisModuleIO *io) {
  if (!io || io->read_pos + sizeof(double) > io->buffer.size()) {
    if (io) io->error_flag = true;
    return 0.0;
  }
  double value = 0.0;
  uint8_t *bytes = reinterpret_cast<uint8_t*>(&value);
  for (size_t i = 0; i < sizeof(double); i++) {
    bytes[i] = io->buffer[io->read_pos++];
  }
  return value;
}

void RMCK_SaveStringBuffer(RedisModuleIO *io, const char *str, size_t len) {
  if (!io || !str) return;
  // Save length first
  RMCK_SaveUnsigned(io, len);
  // Save string data
  for (size_t i = 0; i < len; i++) {
    io->buffer.push_back(static_cast<uint8_t>(str[i]));
  }
}

char *RMCK_LoadStringBuffer(RedisModuleIO *io, size_t *lenptr) {
  if (!io) {
    if (lenptr) *lenptr = 0;
    return nullptr;
  }

  uint64_t len = RMCK_LoadUnsigned(io);
  if (io->error_flag || io->read_pos + len > io->buffer.size()) {
    io->error_flag = true;
    if (lenptr) *lenptr = 0;
    return nullptr;
  }

  char *str = static_cast<char*>(malloc(len + 1));
  if (!str) {
    io->error_flag = true;
    if (lenptr) *lenptr = 0;
    return nullptr;
  }

  for (size_t i = 0; i < len; i++) {
    str[i] = static_cast<char>(io->buffer[io->read_pos++]);
  }
  str[len] = '\0';

  if (lenptr) *lenptr = len;
  return str;
}

void RMCK_SaveString(RedisModuleIO *io, RedisModuleString *s) {
  if (!io || !s) return;
  RMCK_SaveStringBuffer(io, s->c_str(), s->length());
}

RedisModuleString *RMCK_LoadString(RedisModuleIO *io) {
  size_t len;
  char *str = RMCK_LoadStringBuffer(io, &len);
  if (!str) return nullptr;

  RedisModuleString *rms = new RedisModuleString(std::string(str, len));
  free(str);
  return rms;
}

int RMCK_IsIOError(RedisModuleIO *io) {
  int result = io ? (io->error_flag ? 1 : 0) : 1;
  return result;
}

void RMCK_LogIOError(RedisModuleIO *io, const char *levelstr, const char *fmt, ...) {
  (void)io;
  int ilevel = loglevelFromString(levelstr);
  if (ilevel < RMCK_LogLevel) {
    return;
  }
  va_list ap;
  va_start(ap, fmt);
  fprintf(stderr, "[%s] ", levelstr);
  vfprintf(stderr, fmt, ap);
  fprintf(stderr, "\n");
  va_end(ap);
}

int RMCK_InfoAddFieldCString(RedisModuleInfoCtx *ctx, const char *field, const char *value) {
  ctx->fields.emplace_back(field, value);
  return REDISMODULE_OK;
}

void *RMCK_LoadDataTypeFromStringEncver(const RedisModuleString *str,
                                        const RedisModuleType *mt,
                                        int encver) {
  RedisModuleIO *io = RMCK_CreateRdbIO();
  io->buffer.insert(io->buffer.end(), str->c_str(), str->c_str() + str->size());
  void *ret = mt->typemeths.rdb_load(io, encver);
  RMCK_FreeRdbIO(io);
  return ret;
}

RedisModuleString *RMCK_SaveDataTypeToString(RedisModuleCtx *ctx,
                                             void *data,
                                             const RedisModuleType *mt) {
  RedisModuleIO *io = RMCK_CreateRdbIO();
  mt->typemeths.rdb_save(io, data);
  if (io->error_flag) {
    RMCK_FreeRdbIO(io);
    return nullptr;
  }
  RedisModuleString *rms = new RedisModuleString(std::string(io->buffer.begin(), io->buffer.end()));
  if (ctx) ctx->addPointer(rms);
  RMCK_FreeRdbIO(io);
  return rms;
}

int RMCK_ClusterPropagateForSlotMigration(RedisModuleCtx *ctx, const char *cmdname, const char *fmt, ...) {
  std::vector<std::string> command;
  command.emplace_back(cmdname);
  va_list ap;
  va_start(ap, fmt);
  // Parse the format string and extract arguments
  for (const char *p = fmt; *p; p++) {
    if (*p == 's') {
      RedisModuleString *str = va_arg(ap, RedisModuleString *);
      command.emplace_back(*str);
    } else if (*p == 'l') {
      long long ll = va_arg(ap, long long);
      command.emplace_back(std::to_string(ll));
    } else if (*p == 'c') {
      char *cstr = va_arg(ap, char *);
      command.emplace_back(cstr);
    } else if (*p == 'v') {
      RedisModuleString **vec = va_arg(ap, RedisModuleString **);
      size_t len = va_arg(ap, size_t);
      for (size_t i = 0; i < len; i++) {
        command.emplace_back(*vec[i]);
      }
    } else if (*p == 'b') {
      char *buf = va_arg(ap, char *);
      size_t len = va_arg(ap, size_t);
      command.emplace_back(std::string(buf, len));
    } else {
      // Unsupported format specifier
      va_end(ap);
      return REDISMODULE_ERR;
    }
  }
  va_end(ap);
  // Propagate the command (by storing it in the context)
  ctx->propagated_commands.push_back(std::move(command));
  return REDISMODULE_OK;
}

// Function to retrieve propagated commands for testing purposes
std::vector<std::vector<std::string>> &RMCK_GetPropagatedCommands(RedisModuleCtx *ctx) {
  return ctx->propagated_commands;
}

std::string &RMCK_GetLastError(RedisModuleCtx *ctx) {
  return ctx->last_error;
}

RedisModuleSlotRangeArray *RMCK_ClusterGetLocalSlotRanges(RedisModuleCtx *ctx) {
  constexpr RedisModuleSlotRange dummy_ranges[] = {
      {0, 5460},
      {10923, 16383},
  };
  auto *array = reinterpret_cast<RedisModuleSlotRangeArray *>(RMCK_Alloc(sizeof(RedisModuleSlotRangeArray) + sizeof(dummy_ranges)));
  array->num_ranges = 2;
  std::memcpy(array->ranges, dummy_ranges, sizeof(dummy_ranges));
  if (ctx && ctx->automemory) ctx->alloc_slot_ranges.insert(array);
  return array;
}

void RMCK_ClusterFreeSlotRanges(RedisModuleCtx *ctx, RedisModuleSlotRangeArray *slots) {
  if (ctx) ctx->alloc_slot_ranges.erase(slots);
  RMCK_Free(slots);
}

// ============================================================================
// Configurable cluster topology mock
// ----------------------------------------------------------------------------
// Tests populate `mockClusterNodes` via the RMCK_ClusterMock_* helpers below,
// and the mocked cluster API entry points answer queries against that state.
// Slot range arrays returned by the mocked APIs follow the real Redis
// auto-memory contract: if `ctx->automemory` is on, they are freed by the ctx
// destructor; otherwise the caller must free them via
// RedisModule_ClusterFreeSlotRanges.
// ============================================================================
namespace {
struct MockClusterNode {
  std::string id;
  std::string ip;
  int port;
  int flags;
  std::vector<RedisModuleSlotRange> slots;
};

std::vector<MockClusterNode> mockClusterNodes;
std::mutex mockClusterMutex;

const MockClusterNode *findMockNode(const char *id) {
  for (const auto &n : mockClusterNodes) {
    if (n.id == id) return &n;
  }
  return nullptr;
}
}  // namespace

void RMCK_ClusterMock_Reset() {
  std::scoped_lock lock(mockClusterMutex);
  mockClusterNodes.clear();
}

void RMCK_ClusterMock_AddNode(const char *id, const char *ip, int port, int flags,
                              const std::vector<RedisModuleSlotRange> &slots) {
  std::scoped_lock lock(mockClusterMutex);
  MockClusterNode node;
  node.id = id ? id : "";
  node.ip = ip ? ip : "";
  node.port = port;
  node.flags = flags;
  node.slots = slots;
  mockClusterNodes.push_back(std::move(node));
}

static char **RMCK_GetClusterNodesList(RedisModuleCtx * /*ctx*/, size_t *numnodes) {
  std::scoped_lock lock(mockClusterMutex);
  *numnodes = mockClusterNodes.size();
  if (*numnodes == 0) return nullptr;
  // Real Redis null-terminates the list — mirror that so callers iterating
  // with `while (ids[i])` still work even if they ignore `numnodes`.
  auto **list = static_cast<char **>(RMCK_Alloc(sizeof(char *) * (*numnodes + 1)));
  for (size_t i = 0; i < *numnodes; i++) {
    list[i] = static_cast<char *>(RMCK_Alloc(REDISMODULE_NODE_ID_LEN + 1));
    size_t copied = mockClusterNodes[i].id.copy(list[i], REDISMODULE_NODE_ID_LEN);
    list[i][copied] = '\0';
  }
  list[*numnodes] = nullptr;
  return list;
}

static void RMCK_FreeClusterNodesList(char **ids) {
  if (!ids) return;
  for (size_t i = 0; ids[i] != nullptr; i++) RMCK_Free(ids[i]);
  RMCK_Free(ids);
}

static int RMCK_GetClusterNodeInfo(RedisModuleCtx * /*ctx*/, const char *id, char *ip,
                                   char * /*master_id*/, int *port, int *flags) {
  std::scoped_lock lock(mockClusterMutex);
  const MockClusterNode *n = findMockNode(id);
  if (!n) return REDISMODULE_ERR;
  // The real API expects the caller to pass a buffer of at least
  // NET_IP_STR_LEN (46) bytes; we copy at most that to match.
  if (ip) {
    size_t copied = n->ip.copy(ip, 46 - 1);
    ip[copied] = '\0';
  }
  if (port) *port = n->port;
  if (flags) *flags = n->flags;
  return REDISMODULE_OK;
}

static const char *RMCK_GetMyClusterID(void) {
  std::scoped_lock lock(mockClusterMutex);
  for (const auto &n : mockClusterNodes) {
    if (n.flags & REDISMODULE_NODE_MYSELF) return n.id.c_str();
  }
  return nullptr;
}

static size_t RMCK_GetClusterSize(void) {
  std::scoped_lock lock(mockClusterMutex);
  return mockClusterNodes.size();
}

static RedisModuleSlotRangeArray *RMCK_GetClusterNodeSlotRanges(RedisModuleCtx *ctx,
                                                                const char *nodeid) {
  std::scoped_lock lock(mockClusterMutex);
  // Upstream contract (RM_GetClusterNodeSlotRanges): always returns a valid
  // array — possibly empty — never NULL. An unknown nodeid yields an empty
  // array, same as a known node with no assigned slots.
  const MockClusterNode *n = findMockNode(nodeid);
  size_t num_slots = n ? n->slots.size() : 0;
  size_t buf_size = sizeof(RedisModuleSlotRangeArray) + num_slots * sizeof(RedisModuleSlotRange);
  auto *arr = static_cast<RedisModuleSlotRangeArray *>(RMCK_Alloc(buf_size));
  arr->num_ranges = static_cast<int32_t>(num_slots);
  if (n && !n->slots.empty()) {
    std::memcpy(arr->ranges, n->slots.data(), num_slots * sizeof(RedisModuleSlotRange));
  }
  if (ctx && ctx->automemory) ctx->alloc_slot_ranges.insert(arr);
  return arr;
}

// Track contexts associated with IO objects
static std::map<RedisModuleIO*, RedisModuleCtx*> io_contexts;
static std::mutex io_contexts_mutex;

RedisModuleCtx *RMCK_GetContextFromIO(RedisModuleIO *io) {
  if (!io) return nullptr;

  std::lock_guard<std::mutex> lock(io_contexts_mutex);

  // Check if we already have a context for this IO
  auto it = io_contexts.find(io);
  if (it != io_contexts.end()) {
    return it->second;
  }

  // Create new context and associate it with this IO
  RedisModuleCtx *ctx = new RedisModuleCtx();
  io_contexts[io] = ctx;
  return ctx;
}

RedisModuleIO *RMCK_CreateRdbIO(void) {
  return new RedisModuleIO();
}

void RMCK_FreeRdbIO(RedisModuleIO *io) {
  if (io) {
    std::lock_guard<std::mutex> lock(io_contexts_mutex);
    // Clean up associated context
    auto it = io_contexts.find(io);
    if (it != io_contexts.end()) {
      delete it->second;
      io_contexts.erase(it);
    }
  }
  delete io;
}

void RMCK_ResetRdbIO(RedisModuleIO *io) {
  if (io) {
    io->buffer.clear();
    io->read_pos = 0;
    io->error_flag = false;
  }
}

#define REPLY_FUNC(basename, ...)                           \
  int RMCK_Reply##basename(RedisModuleCtx *, __VA_ARGS__) { \
    return REDISMODULE_OK;                                  \
  }

REPLY_FUNC(WithLongLong, long long)
REPLY_FUNC(WithSimpleString, const char *)
REPLY_FUNC(WithArray, size_t)
REPLY_FUNC(WithStringBuffer, const char *, size_t)
REPLY_FUNC(WithDouble, double)
REPLY_FUNC(WithString, RedisModuleString)

int RMCK_ReplyWithNull(RedisModuleCtx *) {
  return REDISMODULE_OK;
}

int RMCK_ReplyWithError(RedisModuleCtx *ctx, const char *err) {
  if (ctx && err) {
    ctx->last_error = err;
  }
  return REDISMODULE_OK;
}

int RMCK_ReplyWithErrorFormat(RedisModuleCtx *ctx, const char *fmt, ...) {
  va_list ap;
  va_start(ap, fmt);
  char *outp = NULL;
  __ignore__(vasprintf(&outp, fmt, ap));
  va_end(ap);
  if (ctx && outp) {
    ctx->last_error = outp;
  }
  free(outp);
  return REDISMODULE_OK;
}

int RMCK_ReplySetArrayLength(RedisModuleCtx *, size_t) {
  return REDISMODULE_OK;
}

void RMCK_SetModuleAttribs(RedisModuleCtx *ctx, const char *name, int ver, int) {
  // Nothing yet.. we're not saving anything anyway
}

RedisModuleCtx *RMCK_GetThreadSafeContext(RedisModuleBlockedClient *bc) {
  assert(bc == NULL);
  return new RedisModuleCtx();
}

RedisModuleCtx *RMCK_GetDetachedThreadSafeContext(RedisModuleCtx *ctx) {
  return RMCK_GetThreadSafeContext(NULL);
}

void RMCK_FreeThreadSafeContext(RedisModuleCtx *ctx) {
  delete ctx;
}

void RMCK_AutoMemory(RedisModuleCtx *ctx) {
  ctx->automemory = true;
}

void RMCK_ThreadSafeContextLock(RedisModuleCtx *) {
  RMCK_GlobalLock.lock();
}

void RMCK_ThreadSafeContextUnlock(RedisModuleCtx *) {
  RMCK_GlobalLock.unlock();
}

static RedisModuleCallReply *RMCK_CallSet(RedisModuleCtx *ctx, const char *cmd, const char *fmt,
                                           va_list ap) {
  if (fmt[0] != 's' || fmt[1] != 's') {
    return NULL;
  }
  RedisModuleString *key = va_arg(ap, RedisModuleString *);
  RedisModuleString *value = va_arg(ap, RedisModuleString *);
  ctx->db->erase(*key);
  StringValue* v = new StringValue(*key);
  v->m_string = *value;
  ctx->db->set(v);
  v->decref();
  return NULL;
}

static RedisModuleCallReply *RMCK_CallDel(RedisModuleCtx *ctx, const char *cmd, const char *fmt,
                                           va_list ap) {
  RedisModuleCallReply* reply = new RedisModuleCallReply(ctx);
  reply->type = REDISMODULE_REPLY_INTEGER;
  reply->ll = 0;
  if (fmt[0] != 's') {
    return reply;
  }
  RedisModuleString *key = va_arg(ap, RedisModuleString *);
  const bool erased = ctx->db->erase(*key);
  reply->ll += erased;
  return reply;
}

static RedisModuleCallReply *RMCK_CallGet(RedisModuleCtx *ctx, const char *cmd, const char *fmt,
                                           va_list ap) {
  if (fmt[0] != 's') {
    return NULL;
  }
  RedisModuleString *key = va_arg(ap, RedisModuleString *);
  Value *v = ctx->db->get(key);
  if (!dynamic_cast<StringValue *>(v)) {
    return NULL;
  }
  RedisModuleCallReply *reply = new RedisModuleCallReply(ctx);
  reply->type = REDISMODULE_REPLY_STRING;
  reply->s = static_cast<StringValue *>(v)->m_string;
  return reply;
}

static RedisModuleCallReply *RMCK_CallHset(RedisModuleCtx *ctx, const char *cmd, const char *fmt,
                                           va_list ap) {
  if (strcmp(fmt, "!v") != 0) {
    return NULL;  // we support only !v for now
  }

  RedisModuleString **args = va_arg(ap, RedisModuleString **);
  size_t argLen = va_arg(ap, size_t) - 1;
  Value *v = ctx->db->get(args[0]);
  if (!v) {
    v = new HashValue(RedisModule_StringPtrLen(args[0], NULL));
    ctx->db->set(v);
    v->decref();
  }
  HashValue *hv = static_cast<HashValue *>(v);
  for (size_t i = 1; i < argLen; i += 2) {
    RedisModuleString *field = args[i];
    RedisModuleString *val = args[i + 1];
    HashValue::Key e(0);
    e.rstr = field;
    hv->hset(e, val);
  }

  RMCK_Notify("hset", REDISMODULE_NOTIFY_HASH, RedisModule_StringPtrLen(args[0], NULL));
  return NULL;
}

static RedisModuleCallReply* HExpire(RedisModuleCtx *ctx, const char *cmd, const char *fmt,
                                    va_list ap, int scale) {
  auto get_string_arg = [&ap] (const char format) -> const char * {
    if (format == 'c') {
      return va_arg(ap, const char *);
    } else if (format == 's') {
      RedisModuleString *rid = va_arg(ap, RedisModuleString *);
      return rid->c_str();
    }
    return NULL;
  };

  RedisModuleCallReply *reply = new RedisModuleCallReply(ctx);
  const char *id = get_string_arg(*fmt);
  if (!id) {
    reply->type = REDISMODULE_REPLY_ERROR;
    reply->s = "Invalid key";
    return reply;
  }

  auto value = ctx->db->get(id);
  auto hash = dynamic_cast<HashValue *>(value);
  if (!hash) {
    reply->type = REDISMODULE_REPLY_ERROR;
    reply->s = "Could not find key";
    return reply;
  }

  const mstime_t expireAt = va_arg(ap, mstime_t) * scale;
  ++fmt;
  if (*fmt != 'v') {
    reply->type = REDISMODULE_REPLY_ERROR;
    reply->s = "Unexpected format";
  }
  ++fmt; // fmt should either be c or s - a vector of const char* or redis string
  size_t count = va_arg(ap, size_t);
  reply->type = REDISMODULE_REPLY_ARRAY;
  const mstime_t now = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count();
  for (size_t index = 0; index < count; ++index) {
    reply->arr.emplace_back(RedisModuleCallReply(ctx));
    auto& fieldReply = reply->arr.back();
    fieldReply.type = REDISMODULE_REPLY_INTEGER;
    const char *field = get_string_arg(*fmt);
    if (field == NULL) {
      fieldReply.ll = -2; // no such field exists
    } else if (expireAt == 0) {
      fieldReply.ll = 2; // invalid expiration time
    } else {
      fieldReply.ll = 1;
      HashValue::Key e(REDISMODULE_HASH_CFIELDS);
      e.cstr = field;
      hash->hexpire(e, now + expireAt);
    }
  }
  RMCK_Notify("hexpire", REDISMODULE_NOTIFY_HASH, id);
  return reply;
}

static RedisModuleCallReply *RMCK_CallHexpire(RedisModuleCtx *ctx, const char *cmd, const char *fmt,
                                              va_list ap) {
  return HExpire(ctx, cmd, fmt, ap, 1000);
}

static RedisModuleCallReply *RMCK_CallHpexpire(RedisModuleCtx *ctx, const char *cmd, const char *fmt,
                                              va_list ap) {
  return HExpire(ctx, cmd, fmt, ap, 1);
}

static RedisModuleCallReply *RMCK_CallHgetall(RedisModuleCtx *ctx, const char *cmd, const char *fmt,
                                              va_list ap) {
  const char *id = NULL;
  if (*fmt == 'c') {
    id = va_arg(ap, const char *);
  } else if (*fmt == 's') {
    RedisModuleString *rid = va_arg(ap, RedisModuleString *);
    id = rid->c_str();
  }

  if (!id) {
    return NULL;
  }

  auto v = ctx->db->get(id);
  RedisModuleCallReply *r = new RedisModuleCallReply(ctx);
  r->type = REDISMODULE_REPLY_ARRAY;
  if (!v) {
    return r;
  }
  if (v->typecode() != REDISMODULE_KEYTYPE_HASH) {
    return r;
  }
  HashValue *hv = static_cast<HashValue *>(v);
  for (auto it : hv->items()) {
    r->arr.push_back(RedisModuleCallReply(ctx, it.first));
    r->arr.push_back(RedisModuleCallReply(ctx, it.second.value));
  }
  return r;
}

static RedisModuleCallReply *RMCK_CallHashFieldExpireTime(RedisModuleCtx *ctx, const char *cmd, const char *fmt,
                                              va_list ap) {
  // return an empty array of expire times
  // the bare minimum to get the code to not issue an error
  RedisModuleCallReply *r = new RedisModuleCallReply(ctx);
  r->type = REDISMODULE_REPLY_ARRAY;
  return r;
}

RedisModuleCallReply *RMCK_Call(RedisModuleCtx *ctx, const char *cmd, const char *fmt, ...) {
  va_list ap;
  RedisModuleCallReply *reply = NULL;
  va_start(ap, fmt);
  errno = 0;
  if (strcasecmp(cmd, "HGETALL") == 0) {
    reply = RMCK_CallHgetall(ctx, cmd, fmt, ap);
  } else if (strcasecmp(cmd, "HSET") == 0) {
    reply = RMCK_CallHset(ctx, cmd, fmt, ap);
  } else if (strcasecmp(cmd, "HEXPIRE") == 0) {
    reply = RMCK_CallHexpire(ctx, cmd, fmt, ap);
  } else if (strcasecmp(cmd, "HPEXPIRE") == 0) {
    reply = RMCK_CallHpexpire(ctx, cmd, fmt, ap);
  } else if (strcasecmp(cmd, "SET") == 0) {
    reply = RMCK_CallSet(ctx, cmd, fmt, ap);
  } else if (strcasecmp(cmd, "GET") == 0) {
    reply = RMCK_CallGet(ctx, cmd, fmt, ap);
  } else if (strcasecmp(cmd, "DEL") == 0) {
    reply = RMCK_CallDel(ctx, cmd, fmt, ap);
  } else if (strcasecmp(cmd, "HPEXPIRETIME") == 0) {
    reply = RMCK_CallHashFieldExpireTime(ctx, cmd, fmt, ap);
  } else {
    errno = ENOTSUP;
  }

  va_end(ap);
  return reply;
}

int RMCK_CallReplyType(RedisModuleCallReply *r) {
  return r->type;
}

void RMCK_FreeCallReply(RedisModuleCallReply *r) {
  delete r;
}

size_t RMCK_CallReplyLength(RedisModuleCallReply *r) {
  if (r->type == REDISMODULE_REPLY_ARRAY) {
    return r->arr.size();
  } else if (r->type == REDISMODULE_REPLY_STRING) {
    return r->s.size();
  } else {
    return 0;
  }
}

RedisModuleCallReply *RMCK_CallReplyArrayElement(RedisModuleCallReply *r, size_t idx) {
  assert(r->type == REDISMODULE_REPLY_ARRAY && r->arr.size() > idx);
  return &r->arr[idx];
}

RedisModuleString *RMCK_CreateStringFromCallReply(RedisModuleCallReply *r) {
  switch (r->type) {
    case REDISMODULE_REPLY_STRING:
      return RedisModule_CreateString(r->ctx, r->s.c_str(), r->s.size());
    case REDISMODULE_REPLY_INTEGER:
      return RedisModule_CreateStringPrintf(r->ctx, "%lld", r->ll);
    default:
      return NULL;
  }
}

const char *RMCK_CallReplyStringPtr(RedisModuleCallReply *r, size_t *n) {
  if (r->type != REDISMODULE_REPLY_STRING && r->type != REDISMODULE_REPLY_ERROR) {
    return NULL;
  }
  *n = r->s.size();
  return r->s.c_str();
}

long long RMCK_CallReplyInteger(RedisModuleCallReply *r) {
  if (r->type != REDISMODULE_REPLY_INTEGER) {
    return 0;
  }
  return r->ll;
}

int RMCK_StringToULongLong(const RedisModuleString *str, unsigned long long *ull) {
  if (str->empty()) {
    return REDISMODULE_ERR;
  }
  char *endptr = nullptr;
  errno = 0;
  *ull = strtoull(str->c_str(), &endptr, 10);
  if (errno == ERANGE || *endptr != '\0' || endptr == str->c_str()) {
    return REDISMODULE_ERR;
  }
  return REDISMODULE_OK;
}


Module::ModuleMap Module::modules;
std::vector<KVDB *> KVDB::dbs;
static int RMCK_GetApi(const char *s, void *pp);

/** Keyspace Events */
std::vector<KeyspaceEventFunction> KeyspaceEvents_g;

void KeyspaceEventFunction::notify(const char *action, int events, const char *key) {
  RMCK::RString rstring(key);
  for (auto ff : KeyspaceEvents_g) {
    if (ff.events & events) {
      ff.call(action, events, rstring);
    }
  }
}

static int RMCK_SubscribeToKeyspaceEvents(RedisModuleCtx *, int types,
                                          RedisModuleNotificationFunc cb) {
  KeyspaceEventFunction fn;
  fn.fn = cb;
  fn.events = types;
  KeyspaceEvents_g.push_back(fn);
  return REDISMODULE_OK;
}

static int RMCK_RegisterCommandFilter(RedisModuleCtx *ctx, RedisModuleCommandFilterFunc callback,
                                      int flags) {
  return REDISMODULE_OK;
}

static std::vector<RedisModuleEventCallback> flushCallbacks;

static int RMCK_SubscribeToServerEvent(RedisModuleCtx *ctx, RedisModuleEvent event,
                                       RedisModuleEventCallback callback) {
  // Make sure we do flush?
  if (event.id == REDISMODULE_EVENT_FLUSHDB) {
    flushCallbacks.push_back(callback);
  }
  return REDISMODULE_OK;
}

void RMCK_Yield(RedisModuleCtx *ctx, int flags, const char *busy_reply) {
  return;
}

int RMCK_GetContextFlags(RedisModuleCtx *ctx) {
  return 0;
}

void RMCK_SelectDb(RedisModuleCtx *ctx, int newid) {
  ctx->dbid = newid;
}

static int RMCK_GetSelectedDb(RedisModuleCtx *ctx) {
  return ctx->dbid;
}

/** Fork */
static int RMCK_Fork(RedisModuleForkDoneHandler cb, void *user_data) {
  return fork();
}

static void RMCK_SendChildHeartbeat(double progress) {
}

// like in Redis' `exitFromChild`, we exit from children using _exit() instead of
// exit(), because the latter may interact with the same file objects used by
// the parent process (may yield errors when testing with sanitizer).
// However if we are testing the coverage normal exit() is
// used in order to obtain the right coverage information.
static int RMCK_ExitFromChild(int retcode) {
#if defined(COV) || defined(COVERAGE)
  exit(retcode);
#else
  _exit(retcode);
#endif
  return REDISMODULE_OK; // never reached, but following the API "behavior"
}

static int RMCK_KillForkChild(int child_pid) {
  return waitpid(child_pid, NULL, 0);
}

static int RMCK_AddACLCategory(RedisModuleCtx *ctx, const char *category) {
  // Nothing for the mock.
  return REDISMODULE_OK;
}

static int RMCK_SetCommandACLCategories(RedisModuleCommand *cmd, const char *categories) {
  // Nothing for the mock.
  return REDISMODULE_OK;
}

static int RMCK_SetCommandInfo(RedisModuleCommand *command, const RedisModuleCommandInfo *info) {
  // Nothing for the mock.
  return REDISMODULE_OK;
}

/** Misc */
RedisModuleCtx::~RedisModuleCtx() {
  if (automemory) {
    for (auto it : allockeys) {
      delete it;
    }
    for (auto it : allocstrs) {
      delete it;
    }
    for (auto *p : alloc_slot_ranges) {
      RMCK_Free(p);
    }
  }
}

RedisModuleCtx::RedisModuleCtx(uint32_t id) : getApi(RMCK_GetApi), dbid(id) {
  if (id >= KVDB::dbs.size()) {
    KVDB::dbs.resize(id + 1);
  }
  db = KVDB::dbs[id];
  if (!db) {
    KVDB::dbs[id] = new KVDB();
    db = KVDB::dbs[id];
    db->id = id;
  }
}

void KVDB::debugDump() const {
  std::cerr << "DB: " << id << std::endl;
  std::cerr << "Containing " << db.size() << " items" << std::endl;
  for (auto ii : db) {
    std::cerr << "Key: " << ii.first << std::endl;
    std::cerr << "  Type: " << Value::typecodeToString(ii.second->typecode()) << std::endl;
    ii.second->debugDump("  ");
  }
}

/**
 * ENTRY POINTS
 */
std::map<std::string, void *> fnregistry;
#define REGISTER_API(basename) fnregistry["RedisModule_" #basename] = (void *)RMCK_##basename

static int RMCK_ExportSharedAPI(RedisModuleCtx *, const char *name, void *funcptr) {
  if (fnregistry.find(name) != fnregistry.end()) {
    return REDISMODULE_ERR;
  }
  fnregistry[name] = funcptr;
  return REDISMODULE_OK;
}
static void *RMCK_GetSharedAPI(RedisModuleCtx *, const char *name) {
  return fnregistry[name];
}

static mstime_t RMCK_GetAbsExpire(RedisModuleKey *key) {
  return REDISMODULE_NO_EXPIRE;
}

struct ServerInfo {
};

static RedisModuleServerInfoData* RMCK_GetServerInfo(RedisModuleCtx *, const char *section) {
  return reinterpret_cast<RedisModuleServerInfoData*>(new ServerInfo());
}

static void RMCK_FreeServerInfo(RedisModuleCtx *, RedisModuleServerInfoData *si) {
  delete reinterpret_cast<ServerInfo*>(si);
}


static unsigned long long RMCK_ServerInfoGetFieldUnsigned(RedisModuleServerInfoData *data, const char* field, int *out_err) {
  return 0;
}

static unsigned long long RMCK_DbSize(RedisModuleCtx *ctx) {
  return ctx->db->size();
}

struct Cursor {
  using Iterator = decltype(std::declval<HashValue>().begin());
  Iterator it;
  Iterator end;
};

static RedisModuleScanCursor* RMCK_ScanCursorCreate() {
  return reinterpret_cast<RedisModuleScanCursor*>(new Cursor());
}

static void RMCK_ScanCursorDestroy(RedisModuleScanCursor *cursor) {
  delete reinterpret_cast<Cursor*>(cursor);
}

static int RMCK_ScanKey(RedisModuleKey *key, RedisModuleScanCursor *cursor, RedisModuleScanKeyCB fn, void *privdata) {
  HashValue* hv = dynamic_cast<HashValue*>(key->ref);
  auto cur = reinterpret_cast<Cursor*>(cursor);
  if (!hv || !cur) {
    errno = EINVAL;
    return 0;
  }
  if (cur->end != hv->end()) {
    cur->it = hv->begin();
    cur->end = hv->end();
  }

  if (cur->it != cur->end) {
    RedisModuleString* field = new RedisModuleString(cur->it->first);
    RedisModuleString* value = new RedisModuleString(cur->it->second.value);
    fn(key, field, value, privdata);
    field->decref();
    value->decref();
    cur->it++;
  }
  return cur->it != cur->end;
}

// KeyMeta mock implementations
static RedisModuleKeyMetaClassId RMCK_CreateKeyMetaClass(RedisModuleCtx *ctx,
                                                         const char *name,
                                                         int encver,
                                                         RedisModuleKeyMetaClassConfig *config) {
  REDISMODULE_NOT_USED(ctx);
  REDISMODULE_NOT_USED(encver);
  RedisModuleKeyMetaClassId classId = nextClassId++;
  classConfigs[classId] = *config;
  classNames[name] = classId;
  return classId;
}

static int RMCK_GetKeyMeta(RedisModuleKeyMetaClassId class_id,
                          RedisModuleKey *key,
                          uint64_t *meta) {
  if (!key) {
    *meta = 0;
    return REDISMODULE_ERR;
  }
  auto keyIt = keyMetaStorage.find(key->key);
  if (keyIt == keyMetaStorage.end()) {
    *meta = 0;
    return REDISMODULE_OK;
  }

  auto metaIt = keyIt->second.find(class_id);
  if (metaIt == keyIt->second.end()) {
    *meta = 0;
    return REDISMODULE_OK;
  }

  *meta = metaIt->second;
  return REDISMODULE_OK;
}

static int RMCK_SetKeyMeta(RedisModuleKeyMetaClassId class_id,
                          RedisModuleKey *key,
                          uint64_t meta) {
  if (!key) {
    return REDISMODULE_ERR;
  }
  keyMetaStorage[key->key][class_id] = meta;
  return REDISMODULE_OK;
}

static void RMCK_ClearKeyMeta() {
  // Clean up any allocated metadata using the free callback
  for (auto& keyPair : keyMetaStorage) {
    for (auto& metaPair : keyPair.second) {
      if (metaPair.second != 0) {
        auto configIt = classConfigs.find(metaPair.first);
        if (configIt != classConfigs.end() && configIt->second.free) {
          configIt->second.free("testkey", metaPair.second);
        }
      }
    }
  }

  keyMetaStorage.clear();
  classConfigs.clear();
  classNames.clear();
  nextClassId = 1;
}

// External interface for clearing KeyMeta storage
void RMCK_ClearKeyMetaStorage() {
  RMCK_ClearKeyMeta();
}

RedisModuleKeyMetaClassId RMCK_GetKeyMetaClassByName(const char *name) {
  auto it = classNames.find(name);
  return it == classNames.end() ? -1 : it->second;
}

int RMCK_KeyMetaRdbLoad(RedisModuleKeyMetaClassId classId, RedisModuleIO *io,
                        uint64_t *meta, int encver) {
  auto it = classConfigs.find(classId);
  if (it == classConfigs.end() || !it->second.rdb_load) {
    if (meta) {
      *meta = 0;
    }
    return REDISMODULE_ERR;
  }
  return it->second.rdb_load(io, meta, encver);
}

void RMCK_KeyMetaRdbSave(RedisModuleKeyMetaClassId classId, RedisModuleIO *io,
                         uint64_t *meta) {
  auto it = classConfigs.find(classId);
  if (it == classConfigs.end() || !it->second.rdb_save) {
    if (io) {
      io->error_flag = true;
    }
    return;
  }
  it->second.rdb_save(io, nullptr, meta);
}

void RMCK_KeyMetaUnlink(RedisModuleKeyMetaClassId classId, uint64_t *meta) {
  auto it = classConfigs.find(classId);
  if (it == classConfigs.end() || !it->second.unlink) {
    return;
  }
  it->second.unlink(nullptr, meta);
}

int RMCK_ConfigGetBool(RedisModuleCtx *ctx, const char *name, int *res) {
  if (!strcmp(name, "tls-cluster")) {
    *res = 0; // Simulate that tls-cluster is disabled
    return REDISMODULE_OK;
  }
  return REDISMODULE_ERR; // Unknown config
}

int RMCK_ConfigGetNumeric(RedisModuleCtx *ctx, const char *name, long long *res) {
  if (!strcmp(name, "tls-port")) {
    *res = 0; // Simulate that tls-port is not set
    return REDISMODULE_OK;
  }
  return REDISMODULE_ERR; // Unknown config
}

static void registerApis() {
  REGISTER_API(GetApi);
  REGISTER_API(Alloc);
  REGISTER_API(Calloc);
  REGISTER_API(Realloc);
  REGISTER_API(Strdup);
  REGISTER_API(Free);

  REGISTER_API(OpenKey);
  REGISTER_API(CloseKey);
  REGISTER_API(KeyType);
  REGISTER_API(DeleteKey);
  REGISTER_API(ValueLength);
  REGISTER_API(GetAbsExpire);

  REGISTER_API(HashSet);
  REGISTER_API(HashGet);
  REGISTER_API(HashGetAll);

  REGISTER_API(_Assert);

  REGISTER_API(HashFieldMinExpire);
  REGISTER_API(CreateString);
  REGISTER_API(CreateStringPrintf);
  REGISTER_API(CreateStringFromString);
  REGISTER_API(FreeString);
  REGISTER_API(RetainString);
  REGISTER_API(HoldString);
  REGISTER_API(TrimStringAllocation);
  REGISTER_API(StringPtrLen);
  REGISTER_API(StringToDouble);
  REGISTER_API(StringToLongLong);

  REGISTER_API(CreateCommand);
  REGISTER_API(GetCommand);
  REGISTER_API(CreateSubcommand);
  REGISTER_API(CreateDataType);
  REGISTER_API(ModuleTypeSetValue);
  REGISTER_API(ModuleTypeGetValue);
  REGISTER_API(ModuleTypeGetType);

  REGISTER_API(SetModuleAttribs);
  REGISTER_API(Log);
  REGISTER_API(Call);

  // REGISTER_API(ReplyWithLongLong);
  // REGISTER_API(ReplyWithSimpleString);
  // REGISTER_API(ReplyWithArray);
  // REGISTER_API(ReplyWithStringBuffer);
  // REGISTER_API(ReplyWithDouble);
  // REGISTER_API(ReplyWithString);
  // REGISTER_API(ReplyWithNull);
  REGISTER_API(ReplyWithError);
  REGISTER_API(ReplyWithErrorFormat);

  REGISTER_API(FreeCallReply);
  REGISTER_API(CallReplyLength);
  REGISTER_API(CallReplyType);
  REGISTER_API(CreateStringFromCallReply);
  REGISTER_API(CallReplyArrayElement);
  REGISTER_API(CallReplyStringPtr);
  REGISTER_API(CallReplyInteger);
  REGISTER_API(StringToULongLong);

  REGISTER_API(GetThreadSafeContext);
  REGISTER_API(GetDetachedThreadSafeContext);
  REGISTER_API(FreeThreadSafeContext);
  REGISTER_API(ThreadSafeContextLock);
  REGISTER_API(ThreadSafeContextUnlock);
  REGISTER_API(StringCompare);
  REGISTER_API(AutoMemory);
  REGISTER_API(ExportSharedAPI);
  REGISTER_API(GetSharedAPI);

  REGISTER_API(DbSize);
  REGISTER_API(GetServerInfo);
  REGISTER_API(FreeServerInfo);
  REGISTER_API(ServerInfoGetFieldUnsigned);
  REGISTER_API(ScanCursorCreate);
  REGISTER_API(ScanCursorDestroy);
  REGISTER_API(ScanKey);

  REGISTER_API(SubscribeToKeyspaceEvents);
  REGISTER_API(SubscribeToServerEvent);
  REGISTER_API(RegisterCommandFilter);

  REGISTER_API(SetModuleOptions);

  REGISTER_API(KillForkChild);
  REGISTER_API(SendChildHeartbeat);
  REGISTER_API(ExitFromChild);
  REGISTER_API(Fork);
  REGISTER_API(AddACLCategory);
  REGISTER_API(SetCommandACLCategories);
  REGISTER_API(SetCommandInfo);
  REGISTER_API(Yield);
  REGISTER_API(GetContextFlags);
  REGISTER_API(GetSelectedDb);
  REGISTER_API(SelectDb);

  // RDB operations
  REGISTER_API(SaveUnsigned);
  REGISTER_API(LoadUnsigned);
  REGISTER_API(SaveSigned);
  REGISTER_API(LoadSigned);
  REGISTER_API(SaveDouble);
  REGISTER_API(LoadDouble);
  REGISTER_API(SaveString);
  REGISTER_API(SaveStringBuffer);
  REGISTER_API(LoadString);
  REGISTER_API(LoadStringBuffer);
  REGISTER_API(IsIOError);
  REGISTER_API(LogIOError);
  REGISTER_API(GetContextFromIO);

  // Info
  REGISTER_API(InfoAddFieldCString);
  // Serialization
  REGISTER_API(LoadDataTypeFromStringEncver);
  REGISTER_API(SaveDataTypeToString);

  // Cluster
  REGISTER_API(ClusterPropagateForSlotMigration);
  REGISTER_API(ClusterGetLocalSlotRanges);
  REGISTER_API(GetClusterNodeSlotRanges);
  REGISTER_API(ClusterFreeSlotRanges);
  REGISTER_API(GetClusterNodesList);
  REGISTER_API(FreeClusterNodesList);
  REGISTER_API(GetClusterNodeInfo);
  REGISTER_API(GetMyClusterID);
  REGISTER_API(GetClusterSize);

  // KeyMeta
  REGISTER_API(CreateKeyMetaClass);
  REGISTER_API(GetKeyMeta);
  REGISTER_API(SetKeyMeta);
  REGISTER_API(ClearKeyMeta);

  // Config
  REGISTER_API(ConfigGetBool);
  REGISTER_API(ConfigGetNumeric);
}

static int RMCK_GetApi(const char *s, void *pp) {
  if (fnregistry.empty()) {
    registerApis();
  }
  *(void **)pp = fnregistry[s];
  return *(void **)pp ? REDISMODULE_OK : REDISMODULE_ERR;
}

extern "C" {

void RMCK_Notify(const char *action, int events, const char *key) {
  KeyspaceEventFunction::notify("hset", REDISMODULE_NOTIFY_HASH, key);
}

void RMCK_Bootstrap(RMCKModuleLoadFunction fn, const char **s, size_t n) {
  // Create the context:
  RedisModuleCtx ctxTmp;
  RMCK::ArgvList args(&ctxTmp, s, n);
  fn(&ctxTmp, args.data(), args.size());
}

void RMCK_Shutdown(void) {
  for (auto db : KVDB::dbs) {
    delete db;
  }
  KVDB::dbs.clear();

  for (auto c : RedisModuleCommand::commands) {
    delete c.second;
  }

  for (auto c : Datatype::typemap) {
    delete c.second;
  }
  Datatype::typemap.clear();

  RedisModuleCommand::commands.clear();
  rm_free((void *)RSGlobalConfig.defaultScorer);
  RSGlobalConfig.defaultScorer = NULL;
}
}