-- TODO: get rid of 'kong.meta'; this module is king
local meta = require "kong.meta"
local PDK = require "kong.pdk"
local process = require "ngx.process"
local phase_checker = require "kong.pdk.private.phases"
local kong_cache = require "kong.cache"
local kong_cluster_events = require "kong.cluster_events"
local private_node = require "kong.pdk.private.node"
local constants = require "kong.constants"

local ngx = ngx
local type = type
local error = error
local setmetatable = setmetatable


local KONG_VERSION = tostring(meta._VERSION)
local KONG_VERSION_NUM = tonumber(string.format("%d%.2d%.2d",
                                  meta._VERSION_TABLE.major * 100,
                                  meta._VERSION_TABLE.minor * 10,
                                  meta._VERSION_TABLE.patch))

local LOCK_OPTS = {
  exptime = 10,
  timeout = 5,
}


local _ns_mt = { __mode = "v" }
local function get_namespaces(self, ctx)
  if not ctx then
    ctx = ngx.ctx
  end

  local namespaces = ctx.KONG_NAMESPACES
  if not namespaces then
    -- 4 namespaces for request, i.e. ~4 plugins
    namespaces = self.table.new(0, 4)
    ctx.KONG_NAMESPACES = setmetatable(namespaces, _ns_mt)
  end

  return namespaces
end


local function set_namespace(self, namespace, namespace_key, ctx)
  local namespaces = get_namespaces(self, ctx)

  local ns = namespaces[namespace]
  if ns and ns == namespace_key then
    return
  end

  namespaces[namespace] = namespace_key
end


local function del_namespace(self, namespace, ctx)
  if not ctx then
    ctx = ngx.ctx
  end

  local namespaces = get_namespaces(self, ctx)
  namespaces[namespace] = nil
end


-- Runloop interface


local _GLOBAL = {
  phases                 = phase_checker.phases,
}


function _GLOBAL.new()
  return {
    version = KONG_VERSION,
    version_num = KONG_VERSION_NUM,

    configuration = nil,
  }
end


function _GLOBAL.set_named_ctx(self, name, key, ctx)
  if not self then
    error("arg #1 cannot be nil", 2)
  end

  if type(name) ~= "string" then
    error("name must be a string", 2)
  end

  if #name == 0 then
    error("name cannot be an empty string", 2)
  end

  if key == nil then
    error("key cannot be nil", 2)
  end

  if not self.table then
    error("ctx PDK module not initialized", 2)
  end

  set_namespace(self, name, key, ctx)
end


function _GLOBAL.del_named_ctx(self, name, ctx)
  if not self then
    error("arg #1 cannot be nil", 2)
  end

  if type(name) ~= "string" then
    error("name must be a string", 2)
  end

  if #name == 0 then
    error("name cannot be an empty string", 2)
  end

  del_namespace(self, name, ctx)
end


do
  local log_facilities = setmetatable({}, { __index = "k" })


  function _GLOBAL.set_namespaced_log(self, namespace, ctx)
    if not self then
      error("arg #1 cannot be nil", 2)
    end

    if type(namespace) ~= "string" then
      error("namespace (arg #2) must be a string", 2)
    end

    local log = log_facilities[namespace]
    if not log then
      log = self._log.new(namespace) -- use default namespaced format
      log_facilities[namespace] = log
    end

    (ctx or ngx.ctx).KONG_LOG = log
  end


  function _GLOBAL.reset_log(self, ctx)
    if not self then
      error("arg #1 cannot be nil", 2)
    end

    (ctx or ngx.ctx).KONG_LOG = self._log
  end
end


function _GLOBAL.init_pdk(self, kong_config)
  if not self then
    error("arg #1 cannot be nil", 2)
  end

  private_node.init_node_id(kong_config)

  PDK.new(kong_config, self)
end


function _GLOBAL.init_worker_events(kong_config)
  -- Note: worker_events will not work correctly if required at the top of the file.
  --       It must be required right here, inside the init function
  local worker_events
  local opts

  local socket_path = kong_config.socket_path
  local sock = ngx.config.subsystem == "stream" and
               constants.SOCKETS.STREAM_WORKER_EVENTS or
               constants.SOCKETS.WORKER_EVENTS

  local listening = "unix:" .. socket_path .. "/" .. sock

  local max_payload_len = kong_config.worker_events_max_payload

  if max_payload_len and max_payload_len > 65535 then   -- default is 64KB
    ngx.log(ngx.WARN,
            "Increasing 'worker_events_max_payload' value has potential " ..
            "negative impact on Kong's response latency and memory usage")
  end

  local enable_privileged_agent = false
  if kong_config.dedicated_config_processing and
     kong_config.role == "data_plane" and
     not kong.sync  -- for rpc sync there is no privileged_agent
  then
    enable_privileged_agent = true
  end

  -- for debug and test
  ngx.log(ngx.DEBUG,
          "lua-resty-events enable_privileged_agent is ",
          enable_privileged_agent)

  opts = {
    unique_timeout = 5,     -- life time of unique event data in lrucache
    broker_id = 0,          -- broker server runs in nginx worker #0
    listening = listening,  -- unix socket for broker listening
    max_queue_len = 1024 * 50,  -- max queue len for events buffering
    max_payload_len = max_payload_len,  -- max payload size in bytes
    enable_privileged_agent = enable_privileged_agent,
  }

  worker_events = require "resty.events.compat"

  local ok, err = worker_events.configure(opts)
  if not ok then
    return nil, err
  end

  return worker_events
end


function _GLOBAL.init_cluster_events(kong_config, db)
  return kong_cluster_events.new({
    db            = db,
    poll_interval = kong_config.db_update_frequency,
    poll_offset   = kong_config.db_update_propagation,
    poll_delay    = kong_config.db_update_propagation,
  })
end


local function get_lru_size(kong_config)
  if (process.type() == "privileged agent")
  or (kong_config.role == "control_plane")
  or (kong_config.role == "traditional" and #kong_config.proxy_listeners  == 0
                                        and #kong_config.stream_listeners == 0)
  then
    return 1000
  end
end


function _GLOBAL.init_cache(kong_config, cluster_events, worker_events)
  local db_cache_ttl = kong_config.db_cache_ttl
  local db_cache_neg_ttl = kong_config.db_cache_neg_ttl
  local page = 1
  local cache_pages = 1

  if kong_config.database == "off" then
    db_cache_ttl = 0
    db_cache_neg_ttl = 0
   end

  return kong_cache.new({
    shm_name             = "kong_db_cache",
    cluster_events       = cluster_events,
    worker_events        = worker_events,
    ttl                  = db_cache_ttl,
    neg_ttl              = db_cache_neg_ttl or db_cache_ttl,
    resurrect_ttl        = kong_config.db_resurrect_ttl,
    page                 = page,
    cache_pages          = cache_pages,
    resty_lock_opts      = LOCK_OPTS,
    lru_size             = get_lru_size(kong_config),
    invalidation_channel = "invalidations",
  })
end


function _GLOBAL.init_core_cache(kong_config, cluster_events, worker_events)
  local db_cache_ttl = kong_config.db_cache_ttl
  local db_cache_neg_ttl = kong_config.db_cache_neg_ttl
  local page = 1
  local cache_pages = 1

  if kong_config.database == "off" then
    db_cache_ttl = 0
    db_cache_neg_ttl = 0
  end

  return kong_cache.new({
    shm_name        = "kong_core_db_cache",
    cluster_events  = cluster_events,
    worker_events   = worker_events,
    ttl             = db_cache_ttl,
    neg_ttl         = db_cache_neg_ttl or db_cache_ttl,
    resurrect_ttl   = kong_config.db_resurrect_ttl,
    page            = page,
    cache_pages     = cache_pages,
    resty_lock_opts = LOCK_OPTS,
    lru_size        = get_lru_size(kong_config),
  })
end


function _GLOBAL.init_timing()
  return require("kong.timing")
end


return _GLOBAL