local legacy_key_format = ARGV[1] == "true"
local lock_owner_id = ARGV[2]

-- the max duration a lock can persist before it is removed
local max_lock_lifetime_seconds = 300
-- on every keep alive poll, existing locks are extended by this period
local default_lock_lifetime_seconds = 12

-- redis does not allow maps, therefore each array index corresponds to a log level
-- 1: debug, 2: info, 3: warn, 4: error
local logBuffer = { {}, {}, {}, {} }

local function safeFormat(...)
    -- ensure each nil argument is converted to a readable string, because string.format does not allow it
    local args = { ... }
    for i = 1, select("#", ...) do
        if args[i] == nil then
            args[i] = "nil"
        end
    end
    return unpack(args)
end

--- Adds a formatted DEBUG log statement to the log buffer using optional format arguments.
--- Logs are registered in order and printed all at once in IAM when the script terminates.
--- @param message string
local function debugLog (message, ...)
    table.insert(logBuffer[1], string.format(message, safeFormat(...)))
end

--- Adds a formatted INFO log statement to the log buffer using optional format arguments.
--- Logs are registered in order and printed all at once in IAM when the script terminates.
--- @param message string
local function infoLog (message, ...)
    table.insert(logBuffer[2], string.format(message, safeFormat(...)))
end

--- Adds a formatted WARN log statement to the log buffer using optional format arguments.
--- Logs are registered in order and printed all at once in IAM when the script terminates.
--- @param message string
local function warnLog (message, ...)
    table.insert(logBuffer[3], string.format(message, safeFormat(...)))
end

--- Adds a formatted ERROR log statement to the log buffer using optional format arguments.
--- Logs are registered in order and printed all at once in IAM when the script terminates.
--- @param message string
local function errorLog (message, ...)
    table.insert(logBuffer[4], string.format(message, safeFormat(...)))
end

local function contains_value (set, val)
    if set == nil then
        return false
    end
    for _, value in ipairs(set) do
        if value == val then
            return true
        end
    end
    return false
end

local function combine (a, b)
    local result = {}

    if a ~= nil then
        for _, value in ipairs(a) do
            if not contains_value(result, value) then
                table.insert(result, value)
            end
        end
    end

    if b ~= nil then
        for _, value in ipairs(b) do
            if not contains_value(result, value) then
                table.insert(result, value)
            end
        end
    end

    return result
end

local function parseOptionalNumber (value)
    local number = tonumber(value)
    if number == -1 then
        return nil
    end
    return number
end

local function get_current_time_milliseconds ()
    local current_time = redis.call("TIME")
    return current_time[1] * 1000 + math.floor(current_time[2] / 1000)
end

local function get_existing_entry (key)
    local entry = redis.call("GET", key)
    if not entry then
        debugLog("entry for key '%s' not found", key)
        return entry
    end
    return cjson.decode(entry)
end

local function remove_entry (key)
    redis.call("DEL", key)
end

local function lock_key (key)
    if legacy_key_format then
        return "LOCK(" .. key .. ")"
    end
    -- creates a lock key for a given key by inserting a ':locks' part before the last segment
    -- example: airlock:iam:namespace:application:objectType:1234 becomes airlock:iam:namespace:application:objectType:locks:1234
    local lastcolonpos = (key:reverse()):find("%:")
    return key:sub(1, key:len() - lastcolonpos + 1) .. "locks:" .. (key:sub(1 - lastcolonpos))
end

local function get_existing_lock (key)
    return redis.call("GET", lock_key(key))
end

local function update_expire_at (key, entry)
    local current_time_milliseconds = get_current_time_milliseconds()

    -- Compute what comes sooner: time since creation exceeding lifetime, or time since last update (now) exceeding idle timeout.
    local new_expiration_time_milliseconds = math.min(
            entry.creation_time_milliseconds + (entry.lifetime_milliseconds or math.huge),
            current_time_milliseconds + (entry.idle_timeout_milliseconds or math.huge)
    )
    redis.call("PEXPIREAT", key, new_expiration_time_milliseconds)
end

local function set_value (key, value, creation_time_milliseconds, lifetime_milliseconds, idle_timeout_milliseconds, additional_lock_keys)
    local current_time_milliseconds = get_current_time_milliseconds()
    creation_time_milliseconds = creation_time_milliseconds or current_time_milliseconds

    -- Compute what comes sooner: time since creation exceeding lifetime, or time since last update (now) exceeding idle timeout.
    local new_expiration_time_milliseconds = math.min(
            (creation_time_milliseconds or math.huge) + (lifetime_milliseconds or math.huge),
            current_time_milliseconds + (idle_timeout_milliseconds or math.huge)
    )
    local serialized_value = cjson.encode({
        value = value,
        creation_time_milliseconds = creation_time_milliseconds,
        lifetime_milliseconds = lifetime_milliseconds,
        idle_timeout_milliseconds = idle_timeout_milliseconds,
        additional_lock_keys = additional_lock_keys
    })
    redis.call("SET", key, serialized_value, "PXAT", new_expiration_time_milliseconds)
end

local function remove_from_current_lock_owner_set(member)
	local result= redis.call("SREM", lock_owner_id, member)
	if (result ~= 1) then
		errorLog("Failed to remove entry '%s' for lock owner '%s'.", member, lock_owner_id)
	end
end

-- exposed only for testing
local function store_lock_for_owner_internal (lock_key, lock_creation_time_milliseconds)
	local set_entry = cjson.encode({
			lock_key = lock_key,
			creation_time_milliseconds = lock_creation_time_milliseconds
		})
	local result= redis.call("SADD", lock_owner_id, set_entry)
	if (result ~= 1) then
		errorLog("Failed to add lock key '%s' for lock owner '%s'.", lock_key, lock_owner_id)
	end
end

local function store_lock_for_owner (lock_key)
	store_lock_for_owner_internal(lock_key, get_current_time_milliseconds())
end

local function release_lock_for_owner (lock_key)
	local locks= redis.call("SMEMBERS", lock_owner_id)
	for _, member in ipairs(locks) do
		local lock= cjson.decode(member)
		if (lock.lock_key == lock_key) then
			remove_from_current_lock_owner_set(member)
		end
		-- iterate through entire set and remove all entries for lock_key.
		-- Should exist only one member per lock_key though...
	end
end

--- Returns the result of a script in the correct format for Redis/IAM.
--- Each script execution in Redis must use this function to return.
--- @param result
--- @return table
local function scriptResult (result)
    -- since redis stops processing lua tables on a nil entry, append logBuffer that is always present at the start.
    -- Optional values are appended last
    table.insert(result, 1, logBuffer)
    return result
end