local server = require('luatest.server')
local t = require('luatest')
local msgpack = require('msgpack')
local uri = require('uri')
local lsocket = require('socket')
 
local type = box.iproto.type
local key = box.iproto.key
 
local timeout = 60
 
local function socket_connect(server)
    local u = uri.parse(server.net_box_uri)
    local s = lsocket.tcp_connect(u.host, u.service)
    t.assert_not_equals(s, nil)
    -- Skip the greeting.
    s:read(box.iproto.GREETING_SIZE, timeout)
    return s
end
 
local function socket_write(s, header, body)
    return s:write(box.iproto.encode_packet(header, body))
end
 
local function socket_read(s)
    local size_mp = s:read(5, timeout)
    t.assert_equals(#size_mp, 5)
    local size = msgpack.decode(size_mp)
    local response = s:read(size, timeout)
    t.assert_equals(#response, size)
    return box.iproto.decode_packet(size_mp .. response)
end
 
local function socket_restart(g)
    g.s:close()
    g.s = socket_connect(g.server)
end
 
local function encode_map(map)
    return msgpack.object(setmetatable(map, {__serialize = 'map'}))
end
 
local function setmap(map)
    return setmetatable(map, {__serialize = 'map'})
end
 
local function write_fetch_snapshot(s)
    local header = {
        [key.REQUEST_TYPE] = box.iproto.type.FETCH_SNAPSHOT,
        [key.SYNC] = 1,
    }
    local body = setmap({})
    return socket_write(s, header, body)
end
 
local function write_subscribe(s, uuid, replicaset_uuid, is_anon)
    local header = {
        [key.REQUEST_TYPE] = box.iproto.type.SUBSCRIBE,
        [key.SYNC] = 1,
    }
    local body = {
        [key.REPLICASET_UUID] = replicaset_uuid,
        [key.INSTANCE_UUID] = uuid,
        [key.VCLOCK] = encode_map({}),
        [key.REPLICA_ANON] = is_anon,
    }
    return socket_write(s, header, body)
end
 
local function write_ok(s, body)
    local header = {
        [key.REQUEST_TYPE] = box.iproto.type.OK,
        [key.SYNC] = 1,
    }
    if body == nil then
        body = {}
    end
    return socket_write(s, header, body)
end
 
local function parse_data_stream(g)
    local h, b = socket_read(g.s)
    local request_type = h[key.REQUEST_TYPE]
    while request_type == type.INSERT or request_type == type.RAFT_PROMOTE do
        h, b = socket_read(g.s)
        request_type = h[key.REQUEST_TYPE]
    end
    return h, b
end
 
local g = t.group()
 
g.before_each(function(g)
    g.server = server:new()
    g.server:start()
    g.server:exec(function()
        box.schema.space.create('test')
        box.space.test:create_index('pk')
        for i = 1, 100 do
            box.space.test:replace{i}
        end
    end)
    g.s = socket_connect(g.server)
end)
 
g.after_each(function(g)
    g.s:close()
    g.server:stop()
end)
 
-- The case sends a dummy OK before subscribe
g.test_iproto_crash_on_subscribe = function(g)
    local uuid = require('uuid').str()
    local replicaset_uuid = g.server:eval('return box.info.replicaset.uuid')
 
    -- Write OK but do not read response
    write_ok(g.s)
 
    write_subscribe(g.s, uuid, replicaset_uuid, true)
    socket_read(g.s)
end
 
-- The case sends spams dummy OKs before subscribe
g.test_iproto_crash_on_subscribe_spam_ok = function(g)
    local uuid = require('uuid').str()
    local replicaset_uuid = g.server:eval('return box.info.replicaset.uuid')
 
    -- Write OK but do not read response
    for i = 1, 100 do
        write_ok(g.s)
    end
 
    write_subscribe(g.s, uuid, replicaset_uuid, true)
    socket_read(g.s)
end
 
-- The case simulates a situtaion where the user of anonymous replication
-- simply sent IPROTO_OK back after FETCH_SNAPSHOT by mistake (did not know
-- that Tarantool doesn't expect a reply on FETCH_SNAPSHOT)
g.test_iproto_crash_fetch_snapshot_subscribe = function(g)
    local uuid = require('uuid').str()
    local replicaset_uuid = g.server:eval('return box.info.replicaset.uuid')
 
    write_fetch_snapshot(g.s)
    -- Read OK response
    socket_read(g.s)
    -- Read all data sent in response
    local h, b = parse_data_stream(g)
    t.assert_equals(h.REQUEST_TYPE, type.OK)
 
    write_ok(g.s)
 
    write_subscribe(g.s, uuid, replicaset_uuid, true)
    socket_read(g.s)
end
 
-- The same as above, but additionally forgot to pass is_anon option to
-- subscribe
g.test_iproto_crash_fetch_snapshot_subscribe_non_anon = function(g)
    local uuid = require('uuid').str()
    local replicaset_uuid = g.server:eval('return box.info.replicaset.uuid')
 
    write_fetch_snapshot(g.s)
    -- Read OK response
    socket_read(g.s)
    -- Read all data sent in response
    local h, b = parse_data_stream(g)
    t.assert_equals(h.REQUEST_TYPE, type.OK)
 
    write_ok(g.s)
 
    write_subscribe(g.s, uuid, replicaset_uuid)
    socket_read(g.s)
end