cli.js

import readline from "readline";
import KeyValueStore from "./src/core/store.js";
import StreamStore from "./src/dataStructures/StreamStore.js";

const rl = readline.createInterface({
  input: process.stdin,
  output: process.stdout,
  prompt: "> ",
});

const store = new KeyValueStore();
const streamStore = new StreamStore();

// Function to extract TTL from arguments
function extractTTL(args) {
  let ttl = null;
  const exIndex = args.findIndex((arg) => arg.toUpperCase() === "EX");
  if (exIndex !== -1 && args[exIndex + 1]) {
    ttl = parseInt(args[exIndex + 1]);
    if (isNaN(ttl)) {
      return { ttl: null, error: "ERR invalid TTL argument" };
    }
    args.splice(exIndex, 2); // Remove EX and TTL from args
  }
  return { ttl, error: null, args };
}

const commandMap = {
  SET: (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    const [key, value] = modifiedArgs;
    return store.set(key, value, ttl);
  },
  GET: (args) => store.get(args[0]),
  TTL: (args) => {
    const key = args[0];

    // Check if it's a stream
    if (store.streamStore.ttlMap && store.streamStore.ttlMap[key]) {
      const remaining = Math.round(
        (store.streamStore.ttlMap[key].expiry - Date.now()) / 1000
      );
      return remaining >= 0 ? remaining : -2;
    }

    // Check if it's a regular key or JSON object
    if (!store.store.has(key) && !store.expirations.has(key)) {
      return -2;
    }

    if (!store.expirations.has(key)) {
      return -1;
    }

    const expirationTime = store.expirations.get(key);
    const remainingTime = Math.max(
      0,
      Math.ceil((expirationTime - Date.now()) / 1000)
    );
    return remainingTime;
  },
  DEL: (args) => {
    store.deleteKey(args[0]);
    return "OK";
  },
  EXISTS: (args) => {
    if (store.streamStore.streams[args[0]]) {
      if (!store.streamStore.isExpired(args[0])) {
        return 1;
      }
    }
    return store.existsKey(args[0]) ? 1 : 0;
  },

  // String commands
  APPEND: (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeCommand("APPEND", modifiedArgs, ttl);
  },
  STRLEN: (args) => store.executeCommand("STRLEN", args),
  INCR: (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeCommand("INCR", modifiedArgs, ttl);
  },
  DECR: (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeCommand("DECR", modifiedArgs, ttl);
  },
  INCRBY: (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeCommand("INCRBY", modifiedArgs, ttl);
  },
  DECRBY: (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeCommand("DECRBY", modifiedArgs, ttl);
  },
  GETRANGE: (args) => store.executeCommand("GETRANGE", args),
  SETRANGE: (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeCommand("SETRANGE", modifiedArgs, ttl);
  },

  // JSON commands
  "JSON.SET": (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeJSONCommand("JSON.SET", modifiedArgs, ttl);
  },
  "JSON.GET": (args) => store.executeJSONCommand("JSON.GET", args),
  "JSON.DEL": (args) => store.executeJSONCommand("JSON.DEL", args),
  "JSON.ARRAPPEND": (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeJSONCommand("JSON.ARRAPPEND", modifiedArgs, ttl);
  },

  // List commands
  LPUSH: (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeListCommand("LPUSH", modifiedArgs, ttl);
  },
  RPUSH: (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeListCommand("RPUSH", modifiedArgs, ttl);
  },
  LRANGE: (args) => store.executeListCommand("LRANGE", args),
  LPOP: (args) => store.executeListCommand("LPOP", args),
  RPOP: (args) => store.executeListCommand("RPOP", args),
  LSET: (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeListCommand("LSET", modifiedArgs, ttl);
  },

  // Set commands
  SADD: (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeSetCommand("SADD", modifiedArgs, ttl);
  },
  SREM: (args) => store.executeSetCommand("SREM", args),
  SISMEMBER: (args) => store.executeSetCommand("SISMEMBER", args),
  SMEMBERS: (args) => store.executeSetCommand("SMEMBERS", args),
  SINTER: (args) => store.executeSetCommand("SINTER", args),
  SUNION: (args) => store.executeSetCommand("SUNION", args),
  SDIFF: (args) => store.executeSetCommand("SDIFF", args),

  // Hash commands
  HSET: (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeHashCommand("HSET", modifiedArgs, ttl);
  },
  HGET: (args) => store.executeHashCommand("HGET", args),
  HMSET: (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    const fields = {};
    for (let i = 1; i < modifiedArgs.length; i += 2) {
      fields[modifiedArgs[i]] = modifiedArgs[i + 1];
    }
    return store.executeHashCommand("HMSET", [modifiedArgs[0], fields], ttl);
  },
  HGETALL: (args) => store.executeHashCommand("HGETALL", args),
  HDEL: (args) => store.executeHashCommand("HDEL", args),
  HEXISTS: (args) => store.executeHashCommand("HEXISTS", args),

  // Sorted Set commands
  ZADD: (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeSortedSetCommand("ZADD", modifiedArgs, ttl);
  },
  ZRANGE: (args) => store.executeSortedSetCommand("ZRANGE", args),
  ZRANK: (args) => store.executeSortedSetCommand("ZRANK", args),
  ZREM: (args) => store.executeSortedSetCommand("ZREM", args),
  ZRANGEBYSCORE: (args) => store.executeSortedSetCommand("ZRANGEBYSCORE", args),

  // Stream commands
  XADD: (args) => {
    let { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;

    const stream = modifiedArgs[0];
    const id = modifiedArgs[1] === "*" ? "*" : modifiedArgs[1];
    const message = {};
    for (let i = 2; i < modifiedArgs.length; i += 2) {
      message[modifiedArgs[i]] = modifiedArgs[i + 1];
    }
    return store.streamStore.XADD(stream, id, message, ttl); //pass the ttl
  },
  XREAD: (args) => {
    const count = parseInt(args[1]);
    const stream = args[3];
    const start = args[4];
    return streamStore.XREAD(stream, count, start);
  },
  XRANGE: (args) => {
    const stream = args[0];
    const start = args[1];
    const end = args[2];
    return streamStore.XRANGE(stream, start, end);
  },
  XLEN: (args) => {
    const stream = args[0];
    return streamStore.XLEN(stream);
  },
  "XGROUP CREATE": (args) => {
    const stream = args[0];
    const groupName = args[1];
    const id = args[2];
    return streamStore.XGROUP_CREATE(stream, groupName, id);
  },
  XREADGROUP: (args) => {
    const groupName = args[1];
    const consumer = args[2];
    const count = parseInt(args[4]);
    const stream = args[6];
    const messageId = args[7];
    return streamStore.XREADGROUP(
      groupName,
      consumer,
      count,
      stream,
      messageId
    );
  },
  XACK: (args) => {
    const stream = args[0];
    const groupName = args[1];
    const messageId = args[2];
    return streamStore.XACK(stream, groupName, messageId);
  },

  // Geospatial commands
  GEOADD: (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeGeoCommand("GEOADD", modifiedArgs, ttl);
  },
  GEOSEARCH: (args) => store.executeGeoCommand("GEOSEARCH", args),
  GEODIST: (args) => store.executeGeoCommand("GEODIST", args),

  // Bitmap commands
  SETBIT: (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeBitmapCommand("SETBIT", modifiedArgs, ttl);
  },
  GETBIT: (args) => store.executeBitmapCommand("GETBIT", args),
  BITCOUNT: (args) => store.executeBitmapCommand("BITCOUNT", args),
  BITOP: (args) => store.executeBitmapCommand("BITOP", args),

  // Bitfield commands
  BITFIELD: (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    const key = modifiedArgs[0];
    const operations = [];
    for (let i = 1; i < modifiedArgs.length; i++) {
      const token = modifiedArgs[i];
      if (token === "GET" || token === "SET" || token === "INCRBY") {
        const command = token;
        const type = modifiedArgs[i + 1];
        const offset = parseInt(modifiedArgs[i + 2]);
        const value =
          command !== "GET" ? parseInt(modifiedArgs[i + 3]) : undefined;
        operations.push([command, type, offset, value]);
        i += command === "GET" ? 2 : 3;
      }
    }
    return store.executeBitfieldCommand("BITFIELD", [key, ...operations], ttl);
  },

  // HyperLogLog commands
  PFADD: (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeHyperLogLogCommand("PFADD", modifiedArgs, ttl);
  },
  PFCOUNT: (args) => store.executeHyperLogLogCommand("PFCOUNT", args),
  PFMERGE: (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeHyperLogLogCommand("PFMERGE", modifiedArgs, ttl);
  },

  // Time Series commands
  "TS.CREATE": (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeTimeSeriesCommand("TS.CREATE", modifiedArgs, ttl);
  },
  "TS.ADD": (args) => {
    const { ttl, error, args: modifiedArgs } = extractTTL(args);
    if (error) return error;
    return store.executeTimeSeriesCommand("TS.ADD", modifiedArgs, ttl);
  },
  "TS.RANGE": (args) => store.executeTimeSeriesCommand("TS.RANGE", args),
  "TS.GET": (args) => store.executeTimeSeriesCommand("TS.GET", args),
};

rl.prompt();
rl.on("line", (line) => {
  const input = line.trim();
  let parts = input.split(" ");
  let command = parts[0].toUpperCase(); // Declare 'command' with let
  const args = parts.slice(1);

  if (command === "XGROUP" && args[0]) {
    command += " " + args[0].toUpperCase();
    args.splice(0, 1);
  }

  const handler = commandMap[command];
  if (handler) {
    try {
      const result = handler(args);
      if (result !== undefined) {
        console.log(result);
      }
    } catch (error) {
      console.error("Error:", error.message);
    }
  } else {
    console.log(`Unknown command: ${command}`);
  }

  rl.prompt();
}).on("close", () => {
  console.log("Exiting CLI. Goodbye!");
  process.exit(0);
});