#!/usr/bin/env tarantool

local io = require('io')
local os = require('os')
local ffi = require('ffi')
local fio = require('fio')
local fun = require('fun')
local log = require('log')
local uri = require('uri')
local json = require('json')
local xlog = require('xlog')
local yaml  = require('yaml')
local errno = require('errno')
local fiber = require('fiber')
local netbox = require('net.box')
local socket = require('socket')
local console = require('console')
local argparse = require('internal.argparse').parse

ffi.cdef[[
int kill(int pid, int sig);
int isatty(int fd);
int getppid(void);
]]

local TIMEOUT_INFINITY = 100 * 365 * 86400

-- name of tarantoolctl binary
local self_name = fio.basename(arg[0])
-- command that we're executing
local command_name = arg[1]
-- true if we're running in user's HOME directory
local usermode = false
-- true if tarantoolctl is a symlink and name != tarantoolctl
local linkmode = false
-- a file with system-wide settings
local default_file
-- current instance settings
local instance_name
local instance_path
local console_sock
local group_name
-- overrides for defaults files
local instance_dir
local default_cfg
local positional_arguments
local keyword_arguments
local lua_arguments = arg
local language

-- function for printing usage reference
local usage

-- shift argv to remove 'tarantoolctl' from arg[0]
local function shift_argv(arg, argno, argcount)
    for i = argno, 128 do
        arg[i] = arg[i + argcount]
        if arg[i] == nil then
            break
        end
    end
end

local function check_user_level()
    local uid = os.getenv('UID')
    if uid == 0 or os.getenv("NOTIFY_SOCKET") then
        return nil
    end
    -- local dir configuration
    local pwd = os.getenv('PWD')
    local udir = pwd and pwd .. '/.tarantoolctl'
    udir = udir and fio.stat(udir) and udir or nil
    -- or home dir configuration
    local homedir = os.getenv('HOME')
    udir = udir or homedir and homedir .. '/.config/tarantool/tarantool'
    udir = udir and fio.stat(udir) and udir or nil
    -- if one of previous is not nil
    if udir ~= nil then
        usermode = true
        return udir
    end

    return nil
end

--
-- Find if we're running under a user, and this user has a default file in their
-- home directory. If present, use it. Otherwise assume a system-wide default.
-- If it's missing, it's OK as well.
--
local function find_default_file()
    -- try to find local dir or user config
    local user_level = check_user_level()
    if user_level ~= nil then
        return user_level
    end

    -- no user-level defaults, use system-wide ones
    local cfg = '@CMAKE_INSTALL_FULL_SYSCONFDIR@/@SYSCONFIG_DEFAULT@/tarantool'
    if fio.stat(cfg) then
        return cfg
    end
    -- It's OK if there is no default file.
    -- load_default_file() will assume some defaults
    return nil
end

local function check_file(path)
    local rv, err = loadfile(path)
    if rv == nil then
        log.error("%s", debug.traceback())
        log.error("Failed to check instance file '%s'", err)
        return err
    end
    return nil
end

--
-- System-wide default file may be missing, this is OK,
-- we'll assume built-in defaults.
-- It uses sandboxing for isolation.
-- It's not completely safe, but it won't
-- allow pollution of global variables.
--
local function load_default_file(default_file)
    if default_file then
        local env = setmetatable({}, { __index = _G })
        local ufunc, msg = loadfile(default_file)
        -- if load fails - show the last 10 lines of the log file
        if not ufunc then
            log.error("Failed to load defaults file: %s", msg)
        end
        debug.setfenv(ufunc, env)
        local state, msg = pcall(ufunc)
        if not state then
            log.error('Failed to execute defaults file: %s', msg)
        end
        default_cfg = env.default_cfg
        instance_dir = env.instance_dir
    end
    local d = default_cfg or {}

    d.pid_file  = d.pid_file  or "/var/run/tarantool"
    d.wal_dir   = d.wal_dir   or "/var/lib/tarantool"
    d.memtx_dir = d.memtx_dir or d.snap_dir or "/var/lib/tarantool"
    d.snap_dir  = nil
    d.log       = d.log       or d.logger or "/var/log/tarantool"
    d.logger    = nil
    d.vinyl_dir = d.vinyl_dir or "/var/lib/tarantool"

    d.pid_file  = fio.pathjoin(d.pid_file,  instance_name .. '.pid')
    d.wal_dir   = fio.pathjoin(d.wal_dir,   instance_name)
    d.memtx_dir = fio.pathjoin(d.memtx_dir, instance_name)
    d.vinyl_dir = fio.pathjoin(d.vinyl_dir, instance_name)
    d.log       = fio.pathjoin(d.log,       instance_name .. '.log')

    language = (d.language or "lua"):lower()
    d.language  = nil

    if language ~= "lua" and language ~= "sql" then
        log.error('Unknown language: %s', language)
        os.exit(1)
    end

    default_cfg = d

    if not usermode then
        -- change user name only if not running locally
        d.username = d.username or "tarantool"
        -- instance_dir must be set in the defaults file, but don't try to set
        -- it to the  global instance dir if the user-local defaults file is in
        -- use
        instance_dir = instance_dir or '/etc/tarantool/instances.enabled'
        -- get user data
        local user_data = ffi.C.getpwnam(ffi.cast('const char*', d.username))
        if user_data == nil then
            log.error('Unknown user: %s', d.username)
            os.exit(1)
        end

        -- get group data
        local group = ffi.C.getgrgid(user_data.pw_gid)
        if group == nil then
            log.error('Group lookup by gid failed: %d', user_data.pw_gid)
            os.exit(1)
        end
        group_name = ffi.string(group.gr_name)
    end

    if instance_dir == nil then
        log.error('Instance directory (instance_dir) is not set in %s', default_file)
        os.exit(1)
    end

    if not fio.stat(instance_dir) then
        log.error('Instance directory %s does not exist', instance_dir)
        os.exit(1)
    end
end

--
-- In case there is no explicit instance name, check whether arg[0] is a
-- symlink. In that case, the name of the symlink is the instance name.
--
local function find_instance_name(arg0, arg2)
    if arg2 ~= nil then
        return fio.basename(arg2, '.lua')
    end
    local istat = fio.lstat(arg0)
    if istat == nil then
        log.error("Can't stat %s: %s", arg0, errno.strerror())
        os.exit(1)
    end
    if not istat:is_link() then usage(command_name) end
    arg[2] = arg0
    linkmode = true
    return fio.basename(arg0, '.lua')
end

local function mkdir(dirname)
    log.info("mkdir %s", dirname)
    if not fio.mkdir(dirname, tonumber('0750', 8)) then
        log.error("Can't mkdir %s: %s", dirname, errno.strerror())
        os.exit(1)
    end

    if not usermode and
       not fio.chown(dirname, default_cfg.username, group_name) then
        log.error("Can't chown(%s, %s, %s): %s", default_cfg.username,
                  group_name, dirname, errno.strerror())
    end
end

local function read_file(filename)
    local file = fio.open(filename, {'O_RDONLY'})
    if file == nil then
        return nil, errno.strerror()
    end

    local buf = {}
    local i = 1
    while true do
        buf[i] = file:read(1024)
        if buf[i] == nil then
            return nil, errno.strerror()
        elseif buf[i] == '' then
            break
        end
        i = i + 1
    end
    return table.concat(buf)
end

-- Removes leading and trailing whitespaces
local function string_trim(str)
    return str:gsub("^%s*(.-)%s*$", "%1")
end

local function logger_parse(logger)
    -- syslog
    if logger:find("syslog:") then
        logger = string_trim(logger:sub(8))
        local args = {}
        logger:gsub("([^,]+)", function(keyval)
            keyval:gsub("([^=]+)=([^=]+)", function(key, val)
                args[key] = val
            end)
        end)
        return 'syslog', args
    -- pipes
    elseif logger:find("pipe:")   then
        logger = string_trim(logger:sub(6))
        return 'pipe', logger
    elseif logger:find("|")       then
        logger = string_trim(logger:sub(2))
        return 'pipe', logger
    -- files
    elseif logger:find("file:")   then
        logger = string_trim(logger:sub(6))
        return 'file', logger
    else
        logger = string_trim(logger)
        return 'file', logger
    end
end

local function mk_default_dirs(cfg)
    local init_dirs = {
        fio.dirname(cfg.pid_file),
        cfg.wal_dir,
        cfg.snap_dir,
        cfg.vinyl_dir,
    }
    local log_type, log_args = logger_parse(cfg.log)
    if log_type == 'file' then
        table.insert(init_dirs, fio.dirname(log_args))
    end
    for _, dir in ipairs(init_dirs) do
        if fio.stat(dir) == nil then
            mkdir(dir)
        end
    end
end

-- systemd detection based on http://unix.stackexchange.com/a/164092
local function under_systemd()
    if not usermode then
        local rv = os.execute("systemctl 2>/dev/null | grep '\\-\\.mount' " ..
                              "1>/dev/null 2>/dev/null")
        if rv == 0 then
            return true
        end
    end
    return false
end

local function forward_to_systemd()
    return under_systemd() and ffi.C.getppid() >= 2
end

-- -------------------------------------------------------------------------- --
--                            CAT command helpers                             --
-- -------------------------------------------------------------------------- --

local function find_in_list(id, list)
    if type(list) == 'number' then
        return id == list
    end
    for _, v in ipairs(list) do
        if v == id then
            return true
        end
    end
    return false
end

local write_lua_table = nil

-- escaped string will be written
local function write_lua_string(string)
    io.stdout:write("'")
    local pos, byte = 1, string:byte(1)
    while byte ~= nil do
        io.stdout:write(("\\x%02x"):format(byte))
        pos = pos + 1
        byte = string:byte(pos)
    end
    io.stdout:write("'")
end

local function write_lua_value(value)
    if type(value) == 'string' then
        write_lua_string(value)
    elseif type(value) == 'table' then
        write_lua_table(value)
    else
        io.stdout:write(tostring(value))
    end
end

local function write_lua_fieldpair(key, val)
    io.stdout:write("[")
    write_lua_value(key)
    io.stdout:write("] = ")
    write_lua_value(val)
end

write_lua_table = function(tuple)
    io.stdout:write('{')
    local is_begin = true
    for key, val in pairs(tuple) do
        if is_begin == false then
            io.stdout:write(', ')
        else
            is_begin = false
        end
        write_lua_fieldpair(key, val)
    end
    io.stdout:write('}')
end

local function cat_lua_cb(record)
    -- Ignore both versions of IPROTO_NOP: the one without a
    -- body (new), and the one with empty body (old).
    if record.HEADER.type == 'NOP' or record.BODY == nil or
       record.BODY.space_id == nil then
        return
    end
    io.stdout:write(('box.space[%d]'):format(record.BODY.space_id))
    local op = record.HEADER.type:lower()
    io.stdout:write((':%s('):format(op))
    if op == 'insert' or op == 'replace' then
        write_lua_table(record.BODY.tuple)
    elseif op == 'delete' then
        write_lua_table(record.BODY.key)
    elseif op == 'update' then
        write_lua_table(record.BODY.key)
        io.stdout:write(', ')
        write_lua_table(record.BODY.tuple)
    elseif op == 'upsert' then
        write_lua_table(record.BODY.tuple)
        io.stdout:write(', ')
        write_lua_table(record.BODY.operations)
    end
    io.stdout:write(')\n')
end

local function cat_yaml_cb(record)
    print(yaml.encode(record):sub(1, -6))
end

local function cat_json_cb(record)
    print(json.encode(record))
end

local cat_formats = setmetatable({
    yaml = cat_yaml_cb,
    json = cat_json_cb,
    lua  = cat_lua_cb,
}, {
    __index = function(self, cmd)
        error(("Unknown formatter '%s'"):format(cmd))
    end
})

-- -------------------------------------------------------------------------- --
--                               Commands                                     --
-- -------------------------------------------------------------------------- --
local orig_cfg = box.cfg

local function wrapper_cfg(cfg)
    fiber.name(instance_name, {truncate=true})
    log.info('Run console at %s', console_sock)
    console.listen(console_sock)

    if not usermode then
        -- gh-2782: socket can be owned by root
        local console_sock = uri.parse(console_sock).service
        if not fio.chown(console_sock, default_cfg.username, group_name) then
            log.error("Can't chown(%s, %s, %s) [%d]: %s", console_sock,
                default_cfg.username, group_name, errno(), errno.strerror())
        end

        -- gh-1293: members of `tarantool` group should be able to do `enter`
        local mode = '0664'
        if not fio.chmod(console_sock, tonumber(mode, 8)) then
            log.error("Can't chmod(%s, %s) [%d]: %s", console_sock, mode,
                      errno(), errno.strerror())
        end
    end

    cfg = cfg or {}
    for i, v in pairs(default_cfg) do
        if cfg[i] == nil then
            cfg[i] = v
        end
    end
    -- force these startup options
    cfg.pid_file = default_cfg.pid_file
    if os.getenv('USER') ~= default_cfg.username then
        cfg.username = default_cfg.username
    else
        cfg.username = nil
    end
    if os.getenv("NOTIFY_SOCKET") then
        cfg.background = false
    elseif cfg.background == nil then
        cfg.background = true
    end

    mk_default_dirs(cfg)
    local success, data = pcall(orig_cfg, cfg)
    if not success then
        log.error("Configuration failed: %s", data)
        if type(cfg) ~= 'function' then
            local log_type, log_args = logger_parse(cfg.log)
            if log_type == 'file' and fio.stat(log_args) then
                os.execute('tail -n 10 ' .. log_args)
            end
        end
        os.exit(1)
    end

    -- Prevent overwriting pid_file in subsequent box.cfg calls.
    local box_cfg_mt = getmetatable(box.cfg)
    local orig_cfg_call = box_cfg_mt.__call
    box_cfg_mt.__call = function(old_cfg, new_cfg)
        if old_cfg.pid_file ~= nil and new_cfg ~= nil and new_cfg.pid_file ~= nil then
            new_cfg.pid_file = old_cfg.pid_file
        end
        orig_cfg_call(old_cfg, new_cfg)
    end

    return data
end

-- It's not 100% result guaranteed function, but it's ok for most cases
-- Won't help in multiple race-conditions
-- Returns nil if Tarantool already started, PID otherwise
local function start_check()
    local pid_file = default_cfg.pid_file

    local fh = fio.open(pid_file, 'O_RDONLY')
    if fh == nil then
        return nil
    end

    local pid = tonumber(fh:read(64))
    fh:close()

    if pid == nil or (ffi.C.kill(pid, 0) < 0 and errno() == errno.ESRCH) then
        return nil
    end
    return pid
end

local function start()
    log.info("Starting instance %s...", instance_name)
    if forward_to_systemd() then
        local cmd = "systemctl start tarantool@" .. instance_name
        log.info("Forwarding to '" .. cmd .. "'")
        os.execute(cmd)
        return
    end
    local stat = check_file(instance_path)
    if stat ~= nil then
        log.error("Error while checking syntax: halting")
        os.exit(1)
    end
    local pid = start_check()
    if pid then
        log.error("The daemon is already running: PID %s", pid)
        os.exit(1)
    end
    box.cfg = wrapper_cfg
    require('title').update{
        script_name = instance_path,
        __defer_update = true
    }

    -- an env variable to tell when we the
    -- instance is run under tarantoolctl
    os.setenv('TARANTOOLCTL', 'true')

    shift_argv(arg, 0, 2)
    local success, data = pcall(dofile, instance_path)
    -- if load fails - show last 10 lines of the log file and exit
    if not success then
        log.error("Start failed: %s", data)
        if type(box.cfg) ~= 'function' then
            local log_type, log_args = logger_parse(box.cfg.log)
            if log_type == 'file' and fio.stat(log_args) then
                os.execute('tail -n 10 ' .. log_args)
            end
        end
        os.exit(1)
    end
    return 0
end

local function stop()
    log.info("Stopping instance %s...", instance_name)
    if forward_to_systemd() then
        local cmd = "systemctl stop tarantool@" .. instance_name
        log.info("Forwarding to '" .. cmd .. "'")
        os.execute(cmd)
        return
    end

    -- remove console socket
    local console_sock = uri.parse(console_sock).service
    if fio.stat(console_sock) then
        fio.unlink(console_sock)
    end

    -- kill process and remove pid file
    local pid_file = default_cfg.pid_file
    if fio.stat(pid_file) == nil then
        log.error("Process is not running (pid: %s)", pid_file)
        return 0
    end

    local f = fio.open(pid_file, 'O_RDONLY')
    if f == nil then
        log.error("Can't read pid file %s: %s", pid_file, errno.strerror())
        return 1
    end

    local pid = tonumber(f:read(64))
    f:close()

    if pid == nil or pid <= 0 then
        log.error("Broken pid file %s", pid_file)
        fio.unlink(pid_file)
        return 1
    end

    if ffi.C.kill(pid, 15) < 0 then
        log.error("Can't kill process %d: %s", pid, errno.strerror())
        fio.unlink(pid_file)
        return 1
    end

    return 0
end

local function check()
    local rv = check_file(instance_path)
    if rv ~= nil then
        return 1
    end
    log.info("File '%s' is OK", instance_path)
    return 0
end

local function restart()
    local stat = check_file(instance_path)
    if stat ~= nil then
        log.error("Error while checking syntax: halting")
        return 1
    end
    stop()
    fiber.sleep(1)
    -- an env variable to tell when
    -- the instance was restarted
    os.setenv('TARANTOOL_RESTARTED', 'true')
    start()
    return 0
end

local function logrotate()
    local console_sock = uri.parse(console_sock).service
    if fio.stat(console_sock) == nil then
        -- process is not running, do nothing
        return 0
    end

    local s = socket.tcp_connect('unix/', console_sock)
    if s == nil then
        -- socket is not opened, do nothing
        return 0
    end

    s:write[[
        require('log'):rotate()
        require('log').info("Rotate log file")
    ]]

    s:read({ '[.][.][.]' }, 2)

    return 0
end

local function enter()
    local options = keyword_arguments
    language = (options.language or language):lower()
    if language ~= "lua" and language ~= "sql" then
        log.error('Unknown language: %s', options.language)
        return 1
    end
    local console_sock_path = uri.parse(console_sock).service
    if fio.stat(console_sock_path) == nil then
        log.error("Can't connect to %s (%s)", console_sock_path, errno.strerror())
        if not usermode and errno() == errno.EACCES then
            log.error("Please add $USER to group '%s': usermod -a -G %s $USER",
                      group_name, group_name)
        end
        return 1
    end
    local status, ret
    console.on_start(function(self)
        status, ret = pcall(console.connect, console_sock,
                            {connect_timeout = TIMEOUT_INFINITY})
        if not status then
            log.error("Can't connect to %s (%s)", console_sock_path, ret)
            self.running = false
            return
        end
        self:eval(string.format("\\set language %s", language))
    end)
    console.on_client_disconnect(function(self) self.running = false end)
    console.start()
    if not status then
        return 1
    end
    return 0
end

local function stdin_isatty()
    return ffi.C.isatty(0) == 1
end

local function execute_remote(uri, code)
    local status, ret
    console.on_start(function(self)
        status, ret = pcall(console.connect, uri,
                            {connect_timeout = TIMEOUT_INFINITY})
        if status then
            status, ret = pcall(self.eval, self, code)
        end
        self.running = false
    end)
    console.on_client_disconnect(function(self) self.running = false end)

    console.start()
    return status, ret
end

local function connect()
    if not stdin_isatty() then
        local code = io.stdin:read("*a")
        if code == nil then
            usage(command_name)
            return 1
        end
        local status, full_response = execute_remote(arg[2], code)
        if not status then
            log.error("Failed to connect to Tarantool")
            return 2
        end
        local error_response = yaml.decode(full_response)[1]
        if type(error_response) == 'table' and error_response.error then
            log.error("Error while executing remote command:")
            log.error(error_response.error)
            return 3
        end
        print(full_response)
        return 0
    end
    -- Otherwise we're starting console
    console.on_start(function(self)
        local status, reason
        status, reason = pcall(function()
            require('console').connect(arg[2], {
                connect_timeout = TIMEOUT_INFINITY
            })
        end)
        if not status then
            self:print(reason)
            self.running = false
        end
    end)
    console.on_client_disconnect(function(self) self.running = false end)
    console.start()
    return 0
end

local function status()
    if forward_to_systemd() then
        local cmd = "systemctl status tarantool@" .. instance_name
        log.info("Forwarding to '" .. cmd .. "'")
        os.execute(cmd)
        return
    end

    local pid_file = default_cfg.pid_file
    local console_sock = uri.parse(console_sock).service

    if fio.stat(pid_file) == nil then
        if errno() == errno.ENOENT then
            log.info('%s is stopped (pid file does not exist)', instance_name)
            return 1
        end
        log.error("Can't access pidfile %s: %s", pid_file, errno.strerror())
    end

    if fio.stat(console_sock) == nil and errno() == errno.ENOENT then
        log.error("Pid file exists, but the control socket (%s) doesn't",
                  console_sock)
        return 2
    end

    local s = socket.tcp_connect('unix/', console_sock)
    if s == nil then
        if errno() ~= errno.EACCES then
            log.warn("Can't access control socket '%s' [%d]: %s", console_sock,
                errno(), errno.strerror())
            return 2
        end
        return 0
    end

    s:close()
    log.info('%s is running (pid: %s)', instance_name, default_cfg.pid_file)
    return 0
end

local function eval()
    local console_sock_path = uri.parse(console_sock).service
    local filename = arg[3]
    local code
    if filename == nil then
        if stdin_isatty() then
            log.error("Usage:")
            log.error(" - tarantoolctl eval instance_name file.lua")
            log.error(" - <command> | tarantoolctl eval instance_name")
            return 1
        end
        code = io.stdin:read("*a")
    else
        local err
        code, err = read_file(filename)
        if code == nil then
            log.error("%s: %s", filename, err)
            return 2
        end
    end

    assert(code ~= nil, "Check that we've successfully loaded file")

    if fio.stat(console_sock_path) == nil then
        log.warn("Pid file exists, but the control socket (%s) doesn't",
                 console_sock_path)
        return 2
    end

    local status, full_response = execute_remote(console_sock, code)
    if status == false then
        log.error("Control socket exists, but Tarantool doesn't listen on it")
        return 2
    end
    local error_response = yaml.decode(full_response)[1]
    if type(error_response) == 'table' and error_response.error then
        log.error(error_response.error)
        return 3
    end

    print(full_response)
    return 0
end

-- Call a callback @a cb for each record that matches all of the
-- following conditions:
--
-- 1. opts.from <= record.HEADER.lsn < opts.to.
-- 2. If @a opts.space and @a opts['show-system'] are not set,
--    record.BODY.space_id should be nil or >= 512.
-- 3. If @a opts.space is set, record.BODY.space_id should not be
--    nil and should be in @a opts.space list.
-- 4. If @a opts.replica is set, record.HEADER.replica_id should
--    not be nil and should be in @a opts.replica list.
--
-- If @a opts.replica is set and is a singleton list and a record
-- **from this replica** with LSN >= @a opts.to is found the loop
-- stops. Note however that this function is called once for each
-- xlog / snap file, so even when it stops on LSN >= @a opts.to on
-- a current file a next file will be processed.
local function filter_xlog(gen, param, state, opts, cb)
    local from, to, spaces = opts.from, opts.to, opts.space
    local show_system, replicas = opts['show-system'], opts.replica

    for lsn, record in gen, param, state do
        local sid = record.BODY and record.BODY.space_id
        local rid = record.HEADER.replica_id
        if replicas and #replicas == 1 and replicas[1] == rid and lsn >= to then
            -- stop, as we've finished reading tuple with lsn == to
            -- and the next lsn's will be bigger
            break
        elseif (lsn < from) or (lsn >= to) or
           (not spaces and sid and sid < 512 and not show_system) or
           (spaces and (sid == nil or not find_in_list(sid, spaces))) or
           (replicas and not find_in_list(rid, replicas)) then -- luacheck: ignore
            -- pass this tuple
        else
            cb(record)
        end
    end
end

local function cat()
    local opts = keyword_arguments
    local cat_format = opts.format
    local format_cb = cat_formats[cat_format]
    local is_printed = false
    for _, file in ipairs(positional_arguments) do
        log.error("Processing file '%s'", file)
        local gen, param, state = xlog.pairs(file)
        filter_xlog(gen, param, state, opts, function(record)
            is_printed = true
            format_cb(record)
            io.stdout:flush()
        end)
        if opts.format == 'yaml' and is_printed then
            is_printed = false
            print('...\n')
        end
    end
end

local function play()
    local opts = keyword_arguments
    local uri = table.remove(positional_arguments, 1)

    if uri == nil then
        error("Empty URI is provided")
    end
    local remote = netbox.new(uri)
    if not remote:wait_connected() then
        error(("Error while connecting to host '%s'"):format(uri))
    end
    for _, file in ipairs(positional_arguments) do
        log.info(("Processing file '%s'"):format(file))
        local gen, param, state = xlog.pairs(file)
        filter_xlog(gen, param, state, opts, function(record)
            local sid = record.BODY and record.BODY.space_id
            if sid ~= nil then
                local args, so = {}, remote.space[sid]
                if so == nil then
                    error(("No space #%s, stopping"):format(sid))
                end
                table.insert(args, so)
                table.insert(args, record.BODY.key)
                table.insert(args, record.BODY.tuple)
                table.insert(args, record.BODY.operations)
                so[record.HEADER.type:lower()](unpack(args))
            end
        end)
    end
    remote:close()
end

local function rocks()
    local cfg = require("luarocks.core.cfg")
    cfg.init()
    local util = require("luarocks.util")
    local command_line = require("luarocks.cmd")
    -- Tweak help messages
    util.see_help = function(command, program) -- luacheck: no unused args
        -- TODO: print extended help message here
        return "See Tarantool documentation for help."
    end

    -- Disabled: path, upload
    local rocks_commands = {
        build = "luarocks.cmd.build",
        config = "luarocks.cmd.config",
        doc = "luarocks.cmd.doc",
        download = "luarocks.cmd.download",
        help = "luarocks.cmd.help",
        init = "luarocks.cmd.init",
        install = "luarocks.cmd.install",
        lint = "luarocks.cmd.lint",
        list = "luarocks.cmd.list",
        make = "luarocks.cmd.make",
        make_manifest = "luarocks.admin.cmd.make_manifest",
        new_version = "luarocks.cmd.new_version",
        pack = "luarocks.cmd.pack",
        purge = "luarocks.cmd.purge",
        remove = "luarocks.cmd.remove",
        search = "luarocks.cmd.search",
        show = "luarocks.cmd.show",
        test = "luarocks.cmd.test",
        unpack = "luarocks.cmd.unpack",
        which = "luarocks.cmd.which",
        write_rockspec = "luarocks.cmd.write_rockspec",
    }

    -- Prepare arguments for luarocks.
    shift_argv(lua_arguments, 0, 1)
    -- Call LuaRocks.
    command_line.run_command('LuaRocks main command-line interface',
                             rocks_commands, nil, unpack(lua_arguments))
end

local function exit_wrapper(func)
    return function() os.exit(func()) end
end

local function process_remote(cmd_function)
    cmd_function()
end

local function process_local(cmd_function)
    instance_name = find_instance_name(arg[0], arg[2])

    default_file = find_default_file()
    load_default_file(default_file)

    if #arg < 2 then
        log.error("Not enough arguments for '%s' command", command_name)
        usage(command_name)
    end

    instance_path = fio.pathjoin(instance_dir, instance_name .. '.lua')

    if not fio.stat(instance_path) then
        log.error('Instance %s is not found in %s', instance_name, instance_dir)
        os.exit(1)
    end

    -- create a path to the control socket (admin console)
    console_sock = instance_name .. '.control'
    console_sock = fio.pathjoin(fio.dirname(default_cfg.pid_file), console_sock)
    console_sock = 'unix/:' .. console_sock

    cmd_function()
end

local commands = setmetatable({
    start = {
        func = start, process = process_local, help = {
            header = "%s start INSTANCE",
            linkmode = "%s start",
            description =
[=[
        Start a Tarantool instance.
]=],
            weight = 10,
            deprecated = false,
        }
    }, stop = {
        func = exit_wrapper(stop), process = process_local, help = {
            header = "%s stop INSTANCE",
            linkmode = "%s stop",
            description =
[=[
        Stop a Tarantool instance.
]=],
            weight = 20,
            deprecated = false,
        }
    }, logrotate = {
        func = exit_wrapper(logrotate), process = process_local, help = {
            header = "%s logrotate INSTANCE",
            linkmode = "%s logrotate",
            description =
[=[
        Rotate logs of a started Tarantool instance.
        Works only if logging-into-file is enabled in the instance file.
        Pipe/syslog make no effect.
]=],
            weight = 50,
            deprecated = false,
        }
    }, status = {
        func = exit_wrapper(status), process = process_local, help = {
            header = "%s status INSTANCE",
            linkmode = "%s status",
            description =
[=[
        Show an instance's status (started/stopped).
        If pid file exists and an alive control socket exists, the return code
        is C<0>. Otherwise, the return code is not C<0>.

        Reports typical problems to stderr (e.g. pid file exists and control
        socket doesn't).
]=],
            weight = 30,
            deprecated = false,
        }
    }, enter = {
        func = exit_wrapper(enter), process = process_local, help = {
            header = "%s enter INSTANCE [--language=language]",
            linkmode = "%s enter",
            description =
[=[
        Enter an instance's interactive Lua or SQL console.

        Supported options:
        * --language=language to set interactive console language.
          May be either Lua or SQL.
]=],
            weight = 65,
            deprecated = false,
        }
    }, restart = {
        func = restart, process = process_local, help = {
            header = "%s restart INSTANCE",
            linkmode = "%s restart",
            description =
[=[
        Stop and start a Tarantool instance.
]=],
            weight = 40,
            deprecated = false,
        }
    }, reload = {
        func = exit_wrapper(eval), process = process_local, help = {
            header = "%s reload INSTANCE FILE",
            linkmode = "%s reload FILE",
            description =
[=[
        DEPRECATED in favor of "eval"
]=],
            weight = 0,
            deprecated = true,
        }
    }, eval = {
        func = exit_wrapper(eval), process = process_local, help = {
            header = {
                "%s eval INSTANCE FILE",
                "COMMAND | %s eval INSTANCE"
            },
            linkmode = {
                "%s eval FILE",
                "COMMAND | %s eval"
            },
            description =
[=[
        Evaluate a local Lua file on a Tarantool instance (if started;
        fail otherwise).
]=],
            weight = 70,
            deprecated = false,
        }
    }, check = {
        func = exit_wrapper(check), process = process_local, help = {
            header = "%s check INSTANCE",
            linkmode = "%s check",
            description =
[=[
        Check an instance file for syntax errors.
]=],
            weight = 60,
            deprecated = false,
        }
    }, connect = {
        func = exit_wrapper(connect), process = process_remote, help = {
            header = {
                "%s connect URI",
                "COMMAND | %s connect URI"
            },
            description =
[=[
        Connect to a Tarantool instance on an admin-console port.
        Supports both TCP/Unix sockets.
]=],
            weight = 80,
            deprecated = false,
        }
    }, cat = {
        func = exit_wrapper(cat), process = process_remote, help = {
            header =
                "%s cat FILE.. [--space=space_no ..] [--show-system]" ..
                " [--from=from_lsn] [--to=to_lsn] [--replica=replica_id ..]",
            description =
[=[
        Print into stdout the contents of .snap/.xlog files.

        Supported options:
        * --space=space_no to filter the output by space number.
          May be passed more than once.
        * --show-system to show the contents of system spaces.
        * --from=from_lsn to show operations starting from the given lsn.
        * --to=to_lsn to show operations ending with the given lsn.
        * --replica=replica_id to filter the output by replica id.
          May be passed more than once.
]=],
            weight = 90,
            deprecated = false,
        }
    }, play = {
        func = exit_wrapper(play), process = process_remote, help = {
            header =
                "%s play URI FILE.. [--space=space_no ..]" ..
                " [--show-system] [--from=from_lsn] [--to=to_lsn]" ..
                " [--replica=replica_id ..]",
            description =
[=[
        Play the contents of .snap/.xlog files to another Tarantool instance.

        Supported options:
        * --space=space_no to filter the output by space number.
          May be passed more than once.
        * --show-system to show the contents of system spaces.
        * --from=from_lsn to show operations starting from the given lsn.
        * --to=to_lsn to show operations ending with the given lsn.
        * --replica=replica_id to filter the output by replica id.
          May be passed more than once.
]=],
            weight = 100,
            deprecated = false,
        }
    }, rocks = {
        func = exit_wrapper(rocks), process = process_remote, help = {
            header =
                "%s rocks [OPTIONS] COMMAND",
            description =
[=[
        Tarantool package manager. To pack/unpack/install/remove
        LuaRocks packages, and many more other actions. For more
        information see

            tarantoolctl rocks --help
]=],
            weight = 100,
            deprecated = false,
        },
        -- LuaRocks parse arguments themselves
        is_self_sufficient = true
    }
}, {
    __index = function()
        log.error("Unknown command '%s'", command_name)
        usage()
    end
})

local function usage_command(name, cmd)
    local header = cmd.help.header
    if linkmode then
        header = cmd.help.linkmode
    end
    if type(header) == 'string' then
        header = { header }
    end
    for _, line in ipairs(header) do
        log.error("    " .. line, name)
    end
end

local function usage_header()
    log.error("Tarantool client utility (%s)", _TARANTOOL)
end

local function usage_commands(commands, verbose)
    local names = fun.iter(commands):map(
        function(self_name, cmd) return {self_name, cmd.help.weight or 0} end
    ):totable()
    table.sort(names, function(left, right) return left[2] < right[2] end)
    for _, cmd_name in ipairs(names) do
        local cmd = commands[cmd_name[1]]
        if cmd.help.deprecated ~= true then
            usage_command(self_name, cmd, false)
            if verbose then
                log.error("")
                log.error(cmd.help.description)
            end
            if cmd.subcommands then
                usage_commands(cmd.subcommands, verbose)
            end
        end
    end
end

usage = function(command, verbose)
    do -- in case a command is passed and is a valid command
        local command_struct = rawget(commands, command)
        if command ~= nil and command_struct then
            log.error("Usage:\n")
            usage_command(self_name, command_struct, true)
            log.error("")
            log.error(command_struct.help.description)
            os.exit(1)
        end
    end -- do this otherwise
    usage_header()
    if default_file ~= nil then
        log.error("Config file: %s", default_file)
    end
    log.error("")
    log.error("Usage:")
    usage_commands(commands, verbose)
    os.exit(1)
end

-- parse parameters and put the result into positional/keyword_arguments
local function populate_arguments()
    local function keyword_arguments_populate(ka)
        ka                = ka                or {}
        ka.from           = ka.from           or 0
        ka.to             = ka.to             or -1ULL
        ka['show-system'] = ka['show-system'] or false
        ka.format         = ka.format         or 'yaml'
        return ka
    end

    -- returns the command name, file list and named parameters
    local function parameters_parse(parameters)
        local command_name = table.remove(parameters, 1)
        local positional_arguments, keyword_arguments = {}, {}
        for k, v in pairs(parameters) do
            if type(k) == 'number' then
                positional_arguments[k] = v
            else
                keyword_arguments[k] = v
            end
        end
        return command_name, positional_arguments, keyword_arguments
    end

    local parameters = argparse(arg, {
        { 'space',       'number+' },
        { 'show-system', 'boolean' },
        { 'from',        'number'  },
        { 'to',          'number'  },
        { 'help',        'boolean' },
        { 'format',      'string'  },
        { 'replica',     'number+' },
        { 'language',    'string'  },
    })

    local cmd_name
    cmd_name, positional_arguments, keyword_arguments = parameters_parse(parameters)
    if cmd_name == 'help' or parameters.help == true or #arg < 1 then
        usage(cmd_name, true)
    end
    keyword_arguments = keyword_arguments_populate(parameters)
end

local function main()

    local cmd_pair = commands[command_name]

    -- Don't call populate arguments() for modules
    -- that parse arguments themselves.
    if not cmd_pair.is_self_sufficient then
        populate_arguments()
    end

    if #arg < 2 then
        log.error("Not enough arguments for '%s' command\n", command_name)
        usage(command_name)
    end
    cmd_pair.process(cmd_pair.func)
end

if rawget(_G, 'TARANTOOLCTL_UNIT_TEST') then
    return {
        internal = {
            filter_xlog = filter_xlog,
        }
    }
end

main()

-- vim: syntax=lua