local redis = {
  _VERSION     = 'redis-lua 2.0.5-dev',
  _DESCRIPTION = 'A Lua client library for the redis key value storage system.',
  _COPYRIGHT   = 'Copyright (C) 2009-2012 Daniele Alessandri',
}

-- The following line is used for backwards compatibility in order to keep the `Redis`
-- global module name. Using `Redis` is now deprecated so you should explicitly assign
-- the module to a local variable when requiring it: `local redis = require('redis')`.
Redis = redis

local unpack = _G.unpack or table.unpack
local network, request, response = {}, {}, {}

local defaults = {
  host        = '127.0.0.1',
  port        = 6379,
  tcp_nodelay = true,
  path        = nil,
}

local function merge_defaults(parameters)
  if parameters == nil then
    parameters = {}
  end
  for k, v in pairs(defaults) do
    if parameters[k] == nil then
      parameters[k] = defaults[k]
    end
  end
  return parameters
end

local function parse_boolean(v)
  if v == '1' or v == 'true' or v == 'TRUE' then
    return true
  elseif v == '0' or v == 'false' or v == 'FALSE' then
    return false
  else
    return nil
  end
end

local function toboolean(value) return value == 1 end

local function sort_request(client, command, key, Params)
  --[[ Params = {
      by    = 'weight_*',
      get   = 'object_*',
      limit = { 0, 10 },
      sort  = 'desc',
      alpha = true,
  } ]]
  local query = { key }

  if Params then
    if Params.by then
      table.insert(query, 'BY')
      table.insert(query, Params.by)
    end

    if type(Params.limit) == 'table' then
      -- TODO: check for lower and upper limits
      table.insert(query, 'LIMIT')
      table.insert(query, Params.limit[1])
      table.insert(query, Params.limit[2])
    end

    if Params.get then
      if (type(Params.get) == 'table') then
        for _, getarg in pairs(Params.get) do
          table.insert(query, 'GET')
          table.insert(query, getarg)
        end
      else
        table.insert(query, 'GET')
        table.insert(query, Params.get)
      end
    end

    if Params.sort then
      table.insert(query, Params.sort)
    end

    if Params.alpha == true then
      table.insert(query, 'ALPHA')
    end

    if Params.store then
      table.insert(query, 'STORE')
      table.insert(query, Params.store)
    end
  end

  request.multibulk(client, command, query)
end

local function zset_range_request(client, command, ...)
  local args, opts = { ... }, {}

  if #args >= 1 and type(args[#args]) == 'table' then
    local options = table.remove(args, #args)
    if options.withscores then
      table.insert(opts, 'WITHSCORES')
    end
  end

  for _, v in pairs(opts) do table.insert(args, v) end
  request.multibulk(client, command, args)
end

local function zset_range_byscore_request(client, command, ...)
  local args, opts = { ... }, {}

  if #args >= 1 and type(args[#args]) == 'table' then
    local options = table.remove(args, #args)
    if options.limit then
      table.insert(opts, 'LIMIT')
      table.insert(opts, options.limit.offset or options.limit[1])
      table.insert(opts, options.limit.count or options.limit[2])
    end
    if options.withscores then
      table.insert(opts, 'WITHSCORES')
    end
  end

  for _, v in pairs(opts) do table.insert(args, v) end
  request.multibulk(client, command, args)
end

local function zset_range_reply(reply, command, ...)
  local args = { ... }
  local opts = args[4]
  if opts and (opts.withscores or string.lower(tostring(opts)) == 'withscores') then
    local new_reply = {}
    for i = 1, #reply, 2 do
      table.insert(new_reply, { reply[i], reply[i + 1] })
    end
    return new_reply
  else
    return reply
  end
end

local function zset_store_request(client, command, ...)
  local args, opts = { ... }, {}

  if #args >= 1 and type(args[#args]) == 'table' then
    local options = table.remove(args, #args)
    if options.weights and type(options.weights) == 'table' then
      table.insert(opts, 'WEIGHTS')
      for _, weight in ipairs(options.weights) do
        table.insert(opts, weight)
      end
    end
    if options.aggregate then
      table.insert(opts, 'AGGREGATE')
      table.insert(opts, options.aggregate)
    end
  end

  for _, v in pairs(opts) do table.insert(args, v) end
  request.multibulk(client, command, args)
end

local function mset_filter_args(client, command, ...)
  local args, arguments = { ... }, {}
  if (#args == 1 and type(args[1]) == 'table') then
    for k, v in pairs(args[1]) do
      table.insert(arguments, k)
      table.insert(arguments, v)
    end
  else
    arguments = args
  end
  request.multibulk(client, command, arguments)
end

local function hash_multi_request_builder(builder_callback)
  return function(client, command, ...)
    local args, arguments = { ... }, {}
    if #args == 2 then
      table.insert(arguments, args[1])
      for k, v in pairs(args[2]) do
        builder_callback(arguments, k, v)
      end
    else
      arguments = args
    end
    request.multibulk(client, command, arguments)
  end
end

local function parse_info(response)
  local info = {}
  local current = info

  response:gsub('([^\r\n]*)\r\n', function(kv)
    if kv == '' then return end

    local section = kv:match('^# (%w+)$')
    if section then
      current = {}
      info[section:lower()] = current
      return
    end

    local k, v = kv:match(('([^:]*):([^:]*)'):rep(1))
    if k:match('db%d+') then
      current[k] = {}
      v:gsub(',', function(dbkv)
        local dbk, dbv = kv:match('([^:]*)=([^:]*)')
        current[k][dbk] = dbv
      end)
    else
      current[k] = v
    end
  end)

  return info
end

local function scan_request(client, command, ...)
  local args, req, Params = { ... }, {}, nil

  if command == 'SCAN' then
    table.insert(req, args[1])
    Params = args[2]
  else
    table.insert(req, args[1])
    table.insert(req, args[2])
    Params = args[3]
  end

  if Params and Params.match then
    table.insert(req, 'MATCH')
    table.insert(req, Params.match)
  end

  if Params and Params.count then
    table.insert(req, 'COUNT')
    table.insert(req, Params.count)
  end

  request.multibulk(client, command, req)
end

local zscan_response = function(reply, command, ...)
  local original, new = reply[2], {}
  for i = 1, #original, 2 do
    table.insert(new, { original[i], tonumber(original[i + 1]) })
  end
  reply[2] = new

  return reply
end

local hscan_response = function(reply, command, ...)
  local original, new = reply[2], {}
  for i = 1, #original, 2 do
    new[original[i]] = original[i + 1]
  end
  reply[2] = new

  return reply
end

local function load_methods(proto, commands)
  local client = setmetatable({}, getmetatable(proto))

  for cmd, fn in pairs(commands) do
    if type(fn) ~= 'function' then
      redis.error('invalid type for command ' .. cmd .. '(must be a function)')
    end
    client[cmd] = fn
  end

  for i, v in pairs(proto) do
    client[i] = v
  end

  return client
end

local function create_client(proto, client_socket, commands)
  local client = load_methods(proto, commands)
  client.error = redis.error
  client.network = {
    socket = client_socket,
    read   = network.read,
    write  = network.write,
  }
  client.requests = {
    multibulk = request.multibulk,
  }
  return client
end

-- ############################################################################

function network.write(client, buffer)
  local _, err = unix.send(client.network.socket, buffer)
  if err then client.error(err) end
end

function network.read(client, len)
  if len == nil then len = '*l' end
  local line, err = unix.recv(client.network.socket, tonumber(len))
  if not err then return line else client.error('connection error: ' .. err) end
end

-- ############################################################################

function response.read(client)
  local payload = client.network.read(client)
  local prefix, data = payload:sub(1, - #payload), payload:sub(2)
  -- status reply
  if prefix == '+' then
    if data == 'OK' then
      return true
    elseif data == 'QUEUED' then
      return { queued = true }
    else
      return data
    end

    -- error reply
  elseif prefix == '-' then
    return client.error('redis error: ' .. data)

    -- integer reply
  elseif prefix == ':' then
    local number = tonumber(data)

    if not number then
      if data == 'nil' then
        return nil
      end
      client.error('cannot parse ' .. data .. ' as a numeric response.')
    end

    return number

    -- bulk reply
  elseif prefix == '$' then
    local length = tonumber(data:match("^[^\n]*"))
    if not length then
      client.error('cannot parse ' .. length .. ' as data length')
    end

    if length == -1 then
      return nil
    end

    local nextchunk = data:gsub("^[^\n]*\n", "")-- .. client.network.read(client, length + 2)

    return nextchunk:sub(1, -3)

    -- multibulk reply
  elseif prefix == '*' then
    local count = tonumber(data)

    if count == -1 then
      return nil
    end

    local list = {}
    if count > 0 then
      local reader = response.read
      for i = 1, count do
        list[i] = reader(client)
      end
    end
    return list

    -- unknown type of reply
  else
    return client.error('unknown response prefix: ' .. prefix)
  end
end

-- ############################################################################

function request.raw(client, buffer)
  local bufferType = type(buffer)

  if bufferType == 'table' then
    client.network.write(client, table.concat(buffer))
  elseif bufferType == 'string' then
    client.network.write(client, buffer)
  else
    client.error('argument error: ' .. bufferType)
  end
end

function request.multibulk(client, command, ...)
  local args = { ... }
  local argsn = #args
  local buffer = { true, true }

  if argsn == 1 and type(args[1]) == 'table' then
    argsn, args = #args[1], args[1]
  end

  buffer[1] = '*' .. tostring(argsn + 1) .. "\r\n"
  buffer[2] = '$' .. #command .. "\r\n" .. command .. "\r\n"

  local table_insert = table.insert
  for i = 1, argsn do
    local s_argument = tostring(args[i] or '')
    table_insert(buffer, '$' .. #s_argument .. "\r\n" .. s_argument .. "\r\n")
  end

  client.network.write(client, table.concat(buffer))
end

-- ############################################################################

local function custom(command, send, parse)
  command = string.upper(command)
  return function(client, ...)
    send(client, command, ...)
    local reply = response.read(client)

    if type(reply) == 'table' and reply.queued then
      reply.parser = parse
      return reply
    else
      if parse then
        return parse(reply, command, ...)
      end
      return reply
    end
  end
end

local function command(command, opts)
  if opts == nil or type(opts) == 'function' then
    return custom(command, request.multibulk, opts)
  else
    return custom(command, opts.request or request.multibulk, opts.response)
  end
end

local define_command_impl = function(target, name, opts)
  local opts = opts or {}
  target[string.lower(name)] = custom(
    opts.command or string.upper(name),
    opts.request or request.multibulk,
    opts.response or nil
  )
end

local undefine_command_impl = function(target, name)
  target[string.lower(name)] = nil
end

-- ############################################################################

local client_prototype = {}

client_prototype.raw_cmd = function(client, buffer)
  request.raw(client, buffer .. "\r\n")
  return response.read(client)
end

-- obsolete
client_prototype.define_command = function(client, name, opts)
  define_command_impl(client, name, opts)
end

-- obsolete
client_prototype.undefine_command = function(client, name)
  undefine_command_impl(client, name)
end

client_prototype.quit = function(client)
  request.multibulk(client, 'QUIT')
  unix.close(client.network.socket)
  return true
end

client_prototype.shutdown = function(client)
  request.multibulk(client, 'SHUTDOWN')
  unix.close(client.network.socket)
end

-- Command pipelining

client_prototype.pipeline = function(client, block)
  local requests, replies, parsers = {}, {}, {}
  local table_insert = table.insert
  local socket_write, socket_read = client.network.write, client.network.read

  client.network.write = function(_, buffer)
    table_insert(requests, buffer)
  end

  -- TODO: this hack is necessary to temporarily reuse the current
  --       request -> response handling implementation of redis-lua
  --       without further changes in the code, but it will surely
  --       disappear when the new command-definition infrastructure
  --       will finally be in place.
  client.network.read = function() return '+QUEUED' end

  local pipeline = setmetatable({}, {
    __index = function(env, name)
      local cmd = client[name]
      if not cmd then
        client.error('unknown redis command: ' .. name, 2)
      end
      return function(self, ...)
        local reply = cmd(client, ...)
        table_insert(parsers, #requests, reply.parser)
        return reply
      end
    end
  })

  local success, retval = pcall(block, pipeline)

  client.network.write, client.network.read = socket_write, socket_read
  if not success then client.error(retval, 0) end

  client.network.write(client, table.concat(requests, ''))

  for i = 1, #requests do
    local reply, parser = response.read(client), parsers[i]
    if parser then
      reply = parser(reply)
    end
    table_insert(replies, i, reply)
  end

  return replies, #requests
end

-- Publish/Subscribe

do
  local channels = function(channels)
    if type(channels) == 'string' then
      channels = { channels }
    end
    return channels
  end

  local subscribe = function(client, ...)
    request.multibulk(client, 'subscribe', ...)
  end
  local psubscribe = function(client, ...)
    request.multibulk(client, 'psubscribe', ...)
  end
  local unsubscribe = function(client, ...)
    request.multibulk(client, 'unsubscribe')
  end
  local punsubscribe = function(client, ...)
    request.multibulk(client, 'punsubscribe')
  end

  local consumer_loop = function(client)
    local aborting, subscriptions = false, 0

    local abort = function()
      if not aborting then
        unsubscribe(client)
        punsubscribe(client)
        aborting = true
      end
    end

    return coroutine.wrap(function()
      while true do
        local message
        local response = response.read(client)

        if response[1] == 'pmessage' then
          message = {
            kind    = response[1],
            pattern = response[2],
            channel = response[3],
            payload = response[4],
          }
        else
          message = {
            kind    = response[1],
            channel = response[2],
            payload = response[3],
          }
        end

        if string.match(message.kind, '^p?subscribe$') then
          subscriptions = subscriptions + 1
        end
        if string.match(message.kind, '^p?unsubscribe$') then
          subscriptions = subscriptions - 1
        end

        if aborting and subscriptions == 0 then
          break
        end
        coroutine.yield(message, abort)
      end
    end)
  end

  client_prototype.pubsub = function(client, subscriptions)
    if type(subscriptions) == 'table' then
      if subscriptions.subscribe then
        subscribe(client, channels(subscriptions.subscribe))
      end
      if subscriptions.psubscribe then
        psubscribe(client, channels(subscriptions.psubscribe))
      end
    end
    return consumer_loop(client)
  end
end

-- Redis transactions (MULTI/EXEC)

do
  local function identity(...) return ... end
  local emptytable = {}

  local function initialize_transaction(client, options, block, queued_parsers)
    local table_insert = table.insert
    local coro = coroutine.create(block)

    if options.watch then
      local watch_keys = {}
      for _, key in pairs(options.watch) do
        table_insert(watch_keys, key)
      end
      if #watch_keys > 0 then
        client:watch(unpack(watch_keys))
      end
    end

    local transaction_client           = setmetatable({}, { __index = client })
    transaction_client.exec            = function(...)
      client.error('cannot use EXEC inside a transaction block')
    end
    transaction_client.multi           = function(...)
      coroutine.yield()
    end
    transaction_client.commands_queued = function()
      return #queued_parsers
    end

    assert(coroutine.resume(coro, transaction_client))

    transaction_client.multi = nil
    transaction_client.discard = function(...)
      local reply = client:discard()
      for i, v in pairs(queued_parsers) do
        queued_parsers[i] = nil
      end
      coro = initialize_transaction(client, options, block, queued_parsers)
      return reply
    end
    transaction_client.watch = function(...)
      client.error('WATCH inside MULTI is not allowed')
    end
    setmetatable(transaction_client, {
      __index = function(t, k)
        local cmd = client[k]
        if type(cmd) == "function" then
          local function queuey(self, ...)
            local reply = cmd(client, ...)
            assert((reply or emptytable).queued == true, 'a QUEUED reply was expected')
            table_insert(queued_parsers, reply.parser or identity)
            return reply
          end
          t[k] = queuey
          return queuey
        else
          return cmd
        end
      end
    })
    client:multi()
    return coro
  end

  local function transaction(client, options, coroutine_block, attempts)
    local queued_parsers, replies = {}, {}
    local retry = tonumber(attempts) or tonumber(options.retry) or 2
    local coro = initialize_transaction(client, options, coroutine_block, queued_parsers)

    local success, retval
    if coroutine.status(coro) == 'suspended' then
      success, retval = coroutine.resume(coro)
    else
      -- do not fail if the coroutine has not been resumed (missing t:multi() with CAS)
      success, retval = true, 'empty transaction'
    end
    if #queued_parsers == 0 or not success then
      client:discard()
      assert(success, retval)
      return replies, 0
    end

    local raw_replies = client:exec()
    if not raw_replies then
      if (retry or 0) <= 0 then
        client.error("MULTI/EXEC transaction aborted by the server")
      else
        --we're not quite done yet
        return transaction(client, options, coroutine_block, retry - 1)
      end
    end

    local table_insert = table.insert
    for i, parser in pairs(queued_parsers) do
      table_insert(replies, i, parser(raw_replies[i]))
    end

    return replies, #queued_parsers
  end

  client_prototype.transaction = function(client, arg1, arg2)
    local options, block
    if not arg2 then
      options, block = {}, arg1
    elseif arg1 then --and arg2, implicitly
      options, block = type(arg1) == "table" and arg1 or { arg1 }, arg2
    else
      client.error("Invalid parameters for redis transaction.")
    end

    if not options.watch then
      local watch_keys = {}
      for i, v in pairs(options) do
        if tonumber(i) then
          table.insert(watch_keys, v)
          options[i] = nil
        end
      end
      options.watch = watch_keys
    elseif not (type(options.watch) == 'table') then
      options.watch = { options.watch }
    end

    if not options.cas then
      local tx_block = block
      block = function(client, ...)
        client:multi()
        return tx_block(client, ...) --can't wrap this in pcall because we're in a coroutine.
      end
    end

    return transaction(client, options, block)
  end
end

-- MONITOR context

do
  local monitor_loop = function(client)
    local monitoring = true

    -- Tricky since the payload format changed starting from Redis 2.6.
    local pattern = '^(%d+%.%d+)( ?.- ?) ?"(%a+)" ?(.-)$'

    local abort = function()
      monitoring = false
    end

    return coroutine.wrap(function()
      client:monitor()

      while monitoring do
        local message, matched
        local response = response.read(client)

        local ok = response:gsub(pattern, function(time, info, cmd, args)
          message = {
            timestamp = tonumber(time),
            client    = info:match('%d+.%d+.%d+.%d+:%d+'),
            database  = tonumber(info:match('%d+')) or 0,
            command   = cmd,
            arguments = args:match('.+'),
          }
          matched = true
        end)

        if not matched then
          client.error('Unable to match MONITOR payload: ' .. response)
        end

        coroutine.yield(message, abort)
      end
    end)
  end

  client_prototype.monitor_messages = function(client)
    return monitor_loop(client)
  end
end

-- ############################################################################

local function connect_tcp(socket, parameters)
  local host, port = parameters.host, tonumber(parameters.port)

  if parameters.timeout then
    -- socket:settimeout(parameters.timeout, 't')
  end


  local ok, err = unix.connect(socket, ResolveIp(host), port)
  if not ok then
    redis.error('could not connect to ' .. host .. ':' .. port .. ' [' .. EncodeJson(err) .. ']')
  end
  -- socket:setoption('tcp-nodelay', parameters.tcp_nodelay)
  return socket
end

local function connect_unix(socket, parameters)
  local ok, err = socket:connect(parameters.path)
  if not ok then
    redis.error('could not connect to ' .. parameters.path .. ' [' .. err .. ']')
  end
  return socket
end

local function create_connection(parameters)
  if parameters.socket then
    return parameters.socket
  end

  local perform_connection, socket

  if parameters.scheme == 'unix' then
    perform_connection, socket = connect_unix, unix.socket()
    assert(socket, 'your build of LuaSocket does not support UNIX domain sockets')
  else
    if parameters.scheme then
      local scheme = parameters.scheme
      assert(scheme == 'redis' or scheme == 'tcp', 'invalid scheme: ' .. scheme)
    end
    perform_connection, socket = connect_tcp, unix.socket()
  end

  return perform_connection(socket, parameters)
end

-- ############################################################################

function redis.error(message, level)
  error(message, (level or 1) + 1)
end

function redis.connect(...)
  local args, parameters = { ... }, nil

  if #args == 1 then
    if type(args[1]) == 'table' then
      parameters = args[1]
    else
      local uri = require('socket.url')
      parameters = uri.parse(select(1, ...))
      if parameters.scheme then
        if parameters.query then
          for k, v in parameters.query:gmatch('([-_%w]+)=([-_%w]+)') do
            if k == 'tcp_nodelay' or k == 'tcp-nodelay' then
              parameters.tcp_nodelay = parse_boolean(v)
            elseif k == 'timeout' then
              parameters.timeout = tonumber(v)
            end
          end
        end
      else
        parameters.host = parameters.path
      end
    end
  elseif #args > 1 then
    local host, port, timeout = unpack(args)
    parameters = { host = host, port = port, timeout = tonumber(timeout) }
  end

  local commands = redis.commands or {}
  if type(commands) ~= 'table' then
    redis.error('invalid type for the commands table')
  end

  local socket = create_connection(merge_defaults(parameters))
  local client = create_client(client_prototype, socket, commands)

  return client
end

function redis.command(cmd, opts)
  return command(cmd, opts)
end

-- obsolete
function redis.define_command(name, opts)
  define_command_impl(redis.commands, name, opts)
end

-- obsolete
function redis.undefine_command(name)
  undefine_command_impl(redis.commands, name)
end


-- ############################################################################

-- Commands defined in this table do not take the precedence over
-- methods defined in the client prototype table.

redis.commands = {
  -- commands operating on the key space
  exists           = command('EXISTS', {
    response = toboolean
  }),
  del              = command('DEL'),
  type             = command('TYPE'),
  rename           = command('RENAME'),
  renamenx         = command('RENAMENX', {
    response = toboolean
  }),
  expire           = command('EXPIRE', {
    response = toboolean
  }),
  pexpire          = command('PEXPIRE', { -- >= 2.6
    response = toboolean
  }),
  expireat         = command('EXPIREAT', {
    response = toboolean
  }),
  pexpireat        = command('PEXPIREAT', { -- >= 2.6
    response = toboolean
  }),
  ttl              = command('TTL'),
  pttl             = command('PTTL'), -- >= 2.6
  move             = command('MOVE', {
    response = toboolean
  }),
  dbsize           = command('DBSIZE'),
  persist          = command('PERSIST', { -- >= 2.2
    response = toboolean
  }),
  keys             = command('KEYS', {
    response = function(response)
      if type(response) == 'string' then
        -- backwards compatibility path for Redis < 2.0
        local keys = {}
        response:gsub('[^%s]+', function(key)
          table.insert(keys, key)
        end)
        response = keys
      end
      return response
    end
  }),
  randomkey        = command('RANDOMKEY'),
  sort             = command('SORT', {
    request = sort_request,
  }),
  scan             = command('SCAN', { -- >= 2.8
    request = scan_request,
  }),

  -- commands operating on string values
  set              = command('SET'),
  setnx            = command('SETNX', {
    response = toboolean
  }),
  setex            = command('SETEX'),  -- >= 2.0
  psetex           = command('PSETEX'), -- >= 2.6
  mset             = command('MSET', {
    request = mset_filter_args
  }),
  msetnx           = command('MSETNX', {
    request  = mset_filter_args,
    response = toboolean
  }),
  get              = command('GET'),
  mget             = command('MGET'),
  getset           = command('GETSET'),
  incr             = command('INCR'),
  incrby           = command('INCRBY'),
  incrbyfloat      = command('INCRBYFLOAT', { -- >= 2.6
    response = function(reply, command, ...)
      return tonumber(reply)
    end,
  }),
  decr             = command('DECR'),
  decrby           = command('DECRBY'),
  append           = command('APPEND'),   -- >= 2.0
  substr           = command('SUBSTR'),   -- >= 2.0
  strlen           = command('STRLEN'),   -- >= 2.2
  setrange         = command('SETRANGE'), -- >= 2.2
  getrange         = command('GETRANGE'), -- >= 2.2
  setbit           = command('SETBIT'),   -- >= 2.2
  getbit           = command('GETBIT'),   -- >= 2.2
  bitop            = command('BITOP'),    -- >= 2.6
  bitcount         = command('BITCOUNT'), -- >= 2.6

  -- commands operating on lists
  rpush            = command('RPUSH'),
  lpush            = command('LPUSH'),
  llen             = command('LLEN'),
  lrange           = command('LRANGE'),
  ltrim            = command('LTRIM'),
  lindex           = command('LINDEX'),
  lset             = command('LSET'),
  lrem             = command('LREM'),
  lpop             = command('LPOP'),
  rpop             = command('RPOP'),
  rpoplpush        = command('RPOPLPUSH'),
  blpop            = command('BLPOP'),      -- >= 2.0
  brpop            = command('BRPOP'),      -- >= 2.0
  rpushx           = command('RPUSHX'),     -- >= 2.2
  lpushx           = command('LPUSHX'),     -- >= 2.2
  linsert          = command('LINSERT'),    -- >= 2.2
  brpoplpush       = command('BRPOPLPUSH'), -- >= 2.2

  -- commands operating on sets
  sadd             = command('SADD'),
  srem             = command('SREM'),
  spop             = command('SPOP'),
  smove            = command('SMOVE', {
    response = toboolean
  }),
  scard            = command('SCARD'),
  sismember        = command('SISMEMBER', {
    response = toboolean
  }),
  sinter           = command('SINTER'),
  sinterstore      = command('SINTERSTORE'),
  sunion           = command('SUNION'),
  sunionstore      = command('SUNIONSTORE'),
  sdiff            = command('SDIFF'),
  sdiffstore       = command('SDIFFSTORE'),
  smembers         = command('SMEMBERS'),
  srandmember      = command('SRANDMEMBER'),
  sscan            = command('SSCAN', { -- >= 2.8
    request = scan_request,
  }),

  -- commands operating on sorted sets
  zadd             = command('ZADD'),
  zincrby          = command('ZINCRBY', {
    response = function(reply, command, ...)
      return tonumber(reply)
    end,
  }),
  zrem             = command('ZREM'),
  zrange           = command('ZRANGE', {
    request  = zset_range_request,
    response = zset_range_reply,
  }),
  zrevrange        = command('ZREVRANGE', {
    request  = zset_range_request,
    response = zset_range_reply,
  }),
  zrangebyscore    = command('ZRANGEBYSCORE', {
    request  = zset_range_byscore_request,
    response = zset_range_reply,
  }),
  zrevrangebyscore = command('ZREVRANGEBYSCORE', { -- >= 2.2
    request  = zset_range_byscore_request,
    response = zset_range_reply,
  }),
  zunionstore      = command('ZUNIONSTORE', { -- >= 2.0
    request = zset_store_request
  }),
  zinterstore      = command('ZINTERSTORE', { -- >= 2.0
    request = zset_store_request
  }),
  zcount           = command('ZCOUNT'),
  zcard            = command('ZCARD'),
  zscore           = command('ZSCORE'),
  zremrangebyscore = command('ZREMRANGEBYSCORE'),
  zrank            = command('ZRANK'),           -- >= 2.0
  zrevrank         = command('ZREVRANK'),        -- >= 2.0
  zremrangebyrank  = command('ZREMRANGEBYRANK'), -- >= 2.0
  zscan            = command('ZSCAN', {          -- >= 2.8
    request  = scan_request,
    response = zscan_response,
  }),

  -- commands operating on hashes
  hset             = command('HSET', { -- >= 2.0
    response = toboolean
  }),
  hsetnx           = command('HSETNX', { -- >= 2.0
    response = toboolean
  }),
  hmset            = command('HMSET', { -- >= 2.0
    request = hash_multi_request_builder(function(args, k, v)
      table.insert(args, k)
      table.insert(args, v)
    end),
  }),
  hincrby          = command('HINCRBY'),       -- >= 2.0
  hincrbyfloat     = command('HINCRBYFLOAT', { -- >= 2.6
    response = function(reply, command, ...)
      return tonumber(reply)
    end,
  }),
  hget             = command('HGET'),   -- >= 2.0
  hmget            = command('HMGET', { -- >= 2.0
    request = hash_multi_request_builder(function(args, k, v)
      table.insert(args, v)
    end),
  }),
  hdel             = command('HDEL'),     -- >= 2.0
  hexists          = command('HEXISTS', { -- >= 2.0
    response = toboolean
  }),
  hlen             = command('HLEN'),     -- >= 2.0
  hkeys            = command('HKEYS'),    -- >= 2.0
  hvals            = command('HVALS'),    -- >= 2.0
  hgetall          = command('HGETALL', { -- >= 2.0
    response = function(reply, command, ...)
      local new_reply = {}
      for i = 1, #reply, 2 do new_reply[reply[i]] = reply[i + 1] end
      return new_reply
    end
  }),
  hscan            = command('HSCAN', { -- >= 2.8
    request  = scan_request,
    response = hscan_response,
  }),

  -- connection related commands
  ping             = command('PING', {
    response = function(response) return response == 'PONG' end
  }),
  echo             = command('ECHO'),
  auth             = command('AUTH'),
  select           = command('SELECT'),

  -- transactions
  multi            = command('MULTI'),   -- >= 2.0
  exec             = command('EXEC'),    -- >= 2.0
  discard          = command('DISCARD'), -- >= 2.0
  watch            = command('WATCH'),   -- >= 2.2
  unwatch          = command('UNWATCH'), -- >= 2.2

  -- publish - subscribe
  subscribe        = command('SUBSCRIBE'),    -- >= 2.0
  unsubscribe      = command('UNSUBSCRIBE'),  -- >= 2.0
  psubscribe       = command('PSUBSCRIBE'),   -- >= 2.0
  punsubscribe     = command('PUNSUBSCRIBE'), -- >= 2.0
  publish          = command('PUBLISH'),      -- >= 2.0

  -- redis scripting
  eval             = command('EVAL'),    -- >= 2.6
  evalsha          = command('EVALSHA'), -- >= 2.6
  script           = command('SCRIPT'),  -- >= 2.6

  -- remote server control commands
  bgrewriteaof     = command('BGREWRITEAOF'),
  config           = command('CONFIG', { -- >= 2.0
    response = function(reply, command, ...)
      if (type(reply) == 'table') then
        local new_reply = {}
        for i = 1, #reply, 2 do new_reply[reply[i]] = reply[i + 1] end
        return new_reply
      end

      return reply
    end
  }),
  client           = command('CLIENT'), -- >= 2.4
  slaveof          = command('SLAVEOF'),
  save             = command('SAVE'),
  bgsave           = command('BGSAVE'),
  lastsave         = command('LASTSAVE'),
  flushdb          = command('FLUSHDB'),
  flushall         = command('FLUSHALL'),
  monitor          = command('MONITOR'),
  time             = command('TIME'),     -- >= 2.6
  slowlog          = command('SLOWLOG', { -- >= 2.2.13
    response = function(reply, command, ...)
      if (type(reply) == 'table') then
        local structured = {}
        for index, entry in ipairs(reply) do
          structured[index] = {
            id = tonumber(entry[1]),
            timestamp = tonumber(entry[2]),
            duration = tonumber(entry[3]),
            command = entry[4],
          }
        end
        return structured
      end

      return reply
    end
  }),
  info             = command('INFO', {
    response = parse_info,
  }),
}

-- ############################################################################

return redis
