<commit_before>--
-- irina.lua
--

--
-- Space 0: Remote IMAP Collector Accounts (Instant and Usual)
--   Tuple: { email (STR), user_id (NUM), is_instant (NUM), is_expirable (NUM), inst_from (NUM), shard_id (NUM) }
--   Index 0: HASH { email }
--   Index 1: TREE { is_instant, shard_id }
--
-- Space 1: Remote IMAP Collector Listeners
--   Tuple: { shard_id (NUM), addr (STR) }
--   Index 0: TREE { shard_id }
--

local function parse_addr(addr)
	local ind = addr:find(":")
	if ind == nil then return nil end

	return addr:sub(0, ind - 1), tonumber(addr:sub(ind + 1))
end

local function send_collector_cmd(addr, cmd)
	local s = box.socket.tcp()
	if s == nil then
		print("can not create collector socket")
		return false
	end

	local host, port = parse_addr(addr)

	if not s:connect(host, port, 0.1) then
		local _, errstr = s:error()
		print("can not connect to collector[" .. host .. ":" .. port .. "]: " .. errstr)
		s:close()
		return false
	end

	local bytes_sent, status, errno, errstr = s:send(cmd, 0.1)
	if bytes_sent ~= #cmd then
		local _, errstr = s:error()
		print("can not send data to collector[" .. host .. ":" .. port .. "]: " .. errstr)
		s:close()
		return false
	end

	s:close()
	return true
end

local function get_collector_address(shardid)
	local v = box.select_limit(1, 0, 0, 1, shardid)
	if v == nil then return nil end
	return v[1]
end

--
-- Send signals to Instant Remote IMAP Collector daemon
--
local function send_change_status(email, userid, shardid, enabled, expirable)
	local addr = get_collector_address(shardid)
	if addr == nil then return end

	local data = email .. " " .. userid .. " " .. enabled .. " " .. expirable
	send_collector_cmd(addr, data)
end
local function send_add_shard(addr, shardid)
	local data = "add_shard " .. shardid
	send_collector_cmd(addr, data)
end
local function send_del_shard(addr, shardid)
	local data = "del_shard " .. shardid
	send_collector_cmd(addr, data)
end

local function get_table_size(t)
	local n = 0
	for _, _ in pairs(t) do n = n + 1 end
	return n
end

function irina_add_collector(addr, notify_added_shard)
	notify_added_shard = tonumber(notify_added_shard)

	local addrs = {}
	for tuple in box.space[1].index[0]:iterator(box.index.ALL) do
		local curr_addr = tuple[1]
		if curr_addr == addr then
			print("already has collector with addr " .. addr)
			return
		end
		if addrs[curr_addr] == nil then addrs[curr_addr] = 1
		else addrs[curr_addr] = addrs[curr_addr] + 1 end<commit_msg>irina.lua: add command for manual addition of collector for the shard<commit_after>--
-- irina.lua
--

--
-- Space 0: Remote IMAP Collector Accounts (Instant and Usual)
--   Tuple: { email (STR), user_id (NUM), is_instant (NUM), is_expirable (NUM), inst_from (NUM), shard_id (NUM) }
--   Index 0: HASH { email }
--   Index 1: TREE { is_instant, shard_id }
--
-- Space 1: Remote IMAP Collector Listeners
--   Tuple: { shard_id (NUM), addr (STR) }
--   Index 0: TREE { shard_id }
--

local function parse_addr(addr)
	local ind = addr:find(":")
	if ind == nil then return nil end

	return addr:sub(0, ind - 1), tonumber(addr:sub(ind + 1))
end

local function send_collector_cmd(addr, cmd)
	local s = box.socket.tcp()
	if s == nil then
		print("can not create collector socket")
		return false
	end

	local host, port = parse_addr(addr)

	if not s:connect(host, port, 0.1) then
		local _, errstr = s:error()
		print("can not connect to collector[" .. host .. ":" .. port .. "]: " .. errstr)
		s:close()
		return false
	end

	local bytes_sent, status, errno, errstr = s:send(cmd, 0.1)
	if bytes_sent ~= #cmd then
		local _, errstr = s:error()
		print("can not send data to collector[" .. host .. ":" .. port .. "]: " .. errstr)
		s:close()
		return false
	end

	s:close()
	return true
end

local function get_collector_address(shardid)
	local v = box.select_limit(1, 0, 0, 1, shardid)
	if v == nil then return nil end
	return v[1]
end

--
-- Send signals to Instant Remote IMAP Collector daemon
--
local function send_change_status(email, userid, shardid, enabled, expirable)
	local addr = get_collector_address(shardid)
	if addr == nil then return end

	local data = email .. " " .. userid .. " " .. enabled .. " " .. expirable
	send_collector_cmd(addr, data)
end
local function send_add_shard(addr, shardid)
	local data = "add_shard " .. shardid
	send_collector_cmd(addr, data)
end
local function send_del_shard(addr, shardid)
	local data = "del_shard " .. shardid
	send_collector_cmd(addr, data)
end

local function get_table_size(t)
	local n = 0
	for _, _ in pairs(t) do n = n + 1 end
	return n
end

function irina_add_collector_for(addr, shardid, notify_added_shard, notify_deleted_shard)
	shardid = tonumber(shardid)
	notify_added_shard = tonumber(notify_added_shard)
	notify_deleted_shard = tonumber(notify_deleted_shard)

	if shardid < 0 or shardid > 1023 then error("shard id must be in [0,1023] range") end
	if notify_added_shard == nil then notify_added_shard = 1 end
	if notify_deleted_shard == nil then notify_deleted_shard = 1 end

	local tuple = box.select_limit(1, 0, 0, 1, shardid)

	if tuple == nil then
		box.insert(1, shardid, addr)
		if notify_added_shard ~= 0 then send_add_shard(addr, shardid) end
	else
		local prev_addr = tuple[1]
		print("rebind shard #" .. shardid .. " from " .. prev_addr .. " to " .. addr)
		box.replace(1, shardid, addr)
		if notify_deleted_shard ~= 0 then
			send_del_shard(prev_addr, shardid)
		end
		if notify_added_shard ~= 0 then
			box.fiber.sleep(0.1)
			send_add_shard(addr, shardid)
		end
	end
end

function irina_add_collector(addr, notify_added_shard)
	notify_added_shard = tonumber(notify_added_shard)

	local addrs = {}
	for tuple in box.space[1].index[0]:iterator(box.index.ALL) do
		local curr_addr = tuple[1]
		if curr_addr == addr then
			print("already has collector with addr " .. addr)
			return
		end
		if addrs[curr_addr] == nil then addrs[curr_addr] = 1
		else addrs[curr_addr] = addrs[curr_addr] + 1 end
{'fraction_non_alphanumeric': 0.12392513909964593, 'fraction_numerical': 0.012139605462822459, 'mean_word_length': 3.707936507936508, 'pattern_counts': {'":': 6, '<': 4, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '279185', 'n_tokens_mistral': 2295, 'n_tokens_neox': 2169, 'n_words': 647}