local workspaces = require "kong.workspaces"
local constants = require "kong.constants"
local tablepool = require "tablepool"
local req_dyn_hook = require "kong.dynamic_hook"
local wasm = require "kong.runloop.wasm"


local kong = kong
local error = error
local assert = assert
local var = ngx.var
local null = ngx.null
local pcall = pcall
local subsystem = ngx.config.subsystem
local pairs = pairs
local ipairs = ipairs
local format = string.format
local fetch_table = tablepool.fetch
local release_table = tablepool.release
local uuid = require("kong.tools.uuid").uuid
local get_updated_monotonic_ms = require("kong.tools.time").get_updated_monotonic_ms
local req_dyn_hook_disable_by_default = req_dyn_hook.disable_by_default


local TTL_ZERO = { ttl = 0 }
local GLOBAL_QUERY_OPTS = { workspace = null, show_ws_id = true }


local NON_COLLECTING_PHASES, DOWNSTREAM_PHASES, DOWNSTREAM_PHASES_COUNT, COLLECTING_PHASE, CONFIGURE_PHASE
do
  if subsystem == "stream" then
    NON_COLLECTING_PHASES = {
      "certificate",
      "log",
    }

    DOWNSTREAM_PHASES = {
      "log",
    }

    COLLECTING_PHASE = "preread"

  else
    NON_COLLECTING_PHASES = {
      "certificate",
      "rewrite",
      "response",
      "header_filter",
      "body_filter",
      "log",
    }

    DOWNSTREAM_PHASES = {
      "response",
      "header_filter",
      "body_filter",
      "log",
    }

    COLLECTING_PHASE = "access"
  end

  DOWNSTREAM_PHASES_COUNT = #DOWNSTREAM_PHASES
  CONFIGURE_PHASE = "configure"
end


local PLUGINS_NS = "plugins." .. subsystem
local ENABLED_PLUGINS
local LOADED_PLUGINS
local CONFIGURABLE_PLUGINS


local PluginsIterator = {}


---
-- Build a compound key by string formatting route_id, service_id, and consumer_id with colons as separators.
--
-- @function build_compound_key
-- @tparam string|nil route_id The route identifier. If `nil`, an empty string is used.
-- @tparam string|nil service_id The service identifier. If `nil`, an empty string is used.
-- @tparam string|nil consumer_id The consumer identifier. If `nil`, an empty string is used.
-- @treturn string The compound key, in the format `route_id:service_id:consumer_id`.
local function build_compound_key(route_id, service_id, consumer_id)
  return format("%s:%s:%s", route_id or "", service_id or "", consumer_id or "")
end


local PLUGIN_GLOBAL_KEY = build_compound_key() -- all nil


local function get_table_for_ctx(ws)
  local tbl = fetch_table(PLUGINS_NS, 0, DOWNSTREAM_PHASES_COUNT + 2)
  if not tbl.initialized then
    for i = 1, DOWNSTREAM_PHASES_COUNT do
      tbl[DOWNSTREAM_PHASES[i]] = kong.table.new(ws.plugins[0] * 2, 1)
    end
    tbl.initialized = true
  end

  for i = 1, DOWNSTREAM_PHASES_COUNT do
    tbl[DOWNSTREAM_PHASES[i]][0] = 0
  end

  tbl.ws = ws

  return tbl
end


local function release(ctx)
  local plugins = ctx.plugins
  if plugins then
    release_table(PLUGINS_NS, plugins, true)
    ctx.plugins = nil
  end
end


local function get_loaded_plugins()
  return assert(kong.db.plugins:get_handlers())
end


local function get_configurable_plugins()
  local i = 0
  local plugins_with_configure_phase = {}
  for _, plugin in ipairs(LOADED_PLUGINS) do
    if plugin.handler[CONFIGURE_PHASE] then
      i = i + 1
      local name = plugin.name
      plugins_with_configure_phase[name] = true
      plugins_with_configure_phase[i] = plugin
    end
  end
  return plugins_with_configure_phase
end


local function should_process_plugin(plugin)
  if wasm.filters_by_name[plugin.name] then
    return false
  end

  if plugin.enabled then
    local c = constants.PROTOCOLS_WITH_SUBSYSTEM
    for _, protocol in ipairs(plugin.protocols) do
      if c[protocol] == subsystem then
        return true
      end
    end
  end
end


local function get_plugin_config(plugin, name, ws_id)
  if not plugin or not plugin.enabled then
    return
  end

  local cfg = plugin.config or {}

  cfg.route_id = plugin.route and plugin.route.id
  cfg.service_id = plugin.service and plugin.service.id
  cfg.consumer_id = plugin.consumer and plugin.consumer.id
  cfg.plugin_instance_name = plugin.instance_name
  cfg.__plugin_id = plugin.id
  cfg.__ws_id = ws_id

  local key = kong.db.plugins:cache_key(name,
                                        cfg.route_id,
                                        cfg.service_id,
                                        cfg.consumer_id,
                                        nil,
                                        ws_id)

  -- TODO: deprecate usage of __key__ as id of plugin
  if not cfg.__key__ then
    cfg.__key__ = key
    -- generate a unique sequence across workers
    -- with a seq 0, plugin server generates an unused random instance id
    local next_seq, err = ngx.shared.kong:incr("plugins_iterator:__seq__", 1, 0, 0)
    if err then
      next_seq = 0
    end
    cfg.__seq__ = next_seq
  end

  return cfg
end


---
-- Lookup a configuration for a given combination of route_id, service_id, consumer_id
--
-- The function checks various combinations of route_id, service_id and consumer_id to find
-- the best matching configuration in the given 'combos' table. The priority order is as follows:
--
-- 1. Route, Service, Consumer
-- 2. Route, Consumer
-- 3. Service, Consumer
-- 4. Route, Service
-- 5. Consumer
-- 6. Route
-- 7. Service
-- 8. Global
--
-- @function lookup_cfg
-- @tparam table combos A table containing configuration data indexed by compound keys.
-- @tparam string|nil route_id The route identifier.
-- @tparam string|nil service_id The service identifier.
-- @tparam string|nil consumer_id The consumer identifier.
-- @return any|nil The configuration corresponding to the best matching combination, or 'nil' if no configuration is found.
local function lookup_cfg(combos, route_id, service_id, consumer_id)
  -- Use the build_compound_key function to create an index for the 'combos' table
  if route_id and service_id and consumer_id then
    local key = build_compound_key(route_id, service_id, consumer_id)
    if combos[key] then
      return combos[key]
    end
  end

  if route_id and consumer_id then
    local key = build_compound_key(route_id, nil, consumer_id)
    if combos[key] then
      return combos[key]
    end
  end

  if service_id and consumer_id then
    local key = build_compound_key(nil, service_id, consumer_id)
    if combos[key] then
      return combos[key]
    end
  end

  if route_id and service_id then
    local key = build_compound_key(route_id, service_id, nil)
    if combos[key] then
      return combos[key]
    end
  end

  if consumer_id then
    local key = build_compound_key(nil, nil, consumer_id)
    if combos[key] then
      return combos[key]
    end
  end

  if route_id then
    local key = build_compound_key(route_id, nil, nil)
    if combos[key] then
      return combos[key]
    end
  end

  if service_id then
    local key = build_compound_key(nil, service_id, nil)
    if combos[key] then
      return combos[key]
    end
  end

  return combos[PLUGIN_GLOBAL_KEY]
end


---
-- Load the plugin configuration based on the context (route, service, and consumer) and plugin handler rules.
--
-- This function filters out route, service, and consumer information from the context based on the plugin handler rules,
-- and then calls the 'lookup_cfg' function to get the best matching plugin configuration for the given combination of
-- route_id, service_id, and consumer_id.
--
-- @function load_configuration_through_combos
-- @tparam table ctx A table containing the context information, including route, service, and authenticated_consumer.
-- @tparam table combos A table containing configuration data indexed by compound keys.
-- @tparam table plugin A table containing plugin information, including the handler with no_route, no_service, and no_consumer rules.
-- @treturn any|nil The configuration corresponding to the best matching combination, or 'nil' if no configuration is found.
local function load_configuration_through_combos(ctx, combos, plugin)
  -- Filter out route, service, and consumer based on the plugin handler rules and get their ids
  local handler = plugin.handler
  local route_id = (not handler.no_route and ctx.route) and ctx.route.id or nil
  local service_id = (not handler.no_service and ctx.service) and ctx.service.id or nil
  local consumer_id = (not handler.no_consumer and ctx.authenticated_consumer) and ctx.authenticated_consumer.id or nil

  -- Call the lookup_cfg function to get the best matching plugin configuration
  return lookup_cfg(combos, route_id, service_id, consumer_id)
end


local function get_workspace(self, ctx)
  if not ctx then
    return self.ws[kong.default_workspace]
  end

  return self.ws[workspaces.get_workspace_id(ctx) or kong.default_workspace]
end


local function get_next_init_worker(plugins, i)
  local i = i + 1
  local plugin = plugins[i]
  if not plugin then
    return nil
  end

  if plugin.handler.init_worker then
    return i, plugin
  end

  return get_next_init_worker(plugins, i)
end


local function get_init_worker_iterator(self)
  if #self.loaded == 0 then
    return nil
  end

  return get_next_init_worker, self.loaded
end


local function get_next_global_or_collected_plugin(plugins, i)
  i = i + 2
  if i > plugins[0] then
    return nil
  end

  return i, plugins[i - 1], plugins[i]
end


local function get_global_iterator(self, phase)
  local plugins = self.globals[phase]
  local count = plugins and plugins[0] or 0
  if count == 0 then
    return nil
  end

  -- only execute this once per request
  if phase == "certificate" or (phase == "rewrite" and var.https ~= "on") then
    local i = 2
    while i <= count do
      kong.vault.update(plugins[i])
      i = i + 2
    end
  end

  return get_next_global_or_collected_plugin, plugins
end


local function get_collected_iterator(self, phase, ctx)
  local plugins = ctx.plugins
  if plugins then
    plugins = plugins[phase]
    if not plugins or plugins[0] == 0 then
      return nil
    end

    return get_next_global_or_collected_plugin, plugins
  end

  return get_global_iterator(self, phase)
end


local function get_next_and_collect(ctx, i)
  i = i + 1
  local ws = ctx.plugins.ws
  local plugins = ws.plugins
  if i > plugins[0] then
    return nil
  end

  local plugin = plugins[i]
  local name = plugin.name
  local cfg
  -- Only pass combos for the plugin we're operating on
  local combos = ws.combos[name]
  if combos then
    cfg = load_configuration_through_combos(ctx, combos, plugin)
    if cfg then
      kong.vault.update(cfg)
      local handler = plugin.handler
      local collected = ctx.plugins
      for j = 1, DOWNSTREAM_PHASES_COUNT do
        local phase = DOWNSTREAM_PHASES[j]
        if handler[phase] then
          local n = collected[phase][0] + 2
          collected[phase][0] = n
          collected[phase][n] = cfg
          collected[phase][n - 1] = plugin
          if phase == "response" and not ctx.buffered_proxying then
            ctx.buffered_proxying = true
          end
        end
      end

      if handler[COLLECTING_PHASE] then
        return i, plugin, cfg
      end
    end
  end

  return get_next_and_collect(ctx, i)
end


local function get_collecting_iterator(self, ctx)
  local ws = get_workspace(self, ctx)
  ctx.plugins = get_table_for_ctx(ws)
  if not ws then
    return nil
  end

  local plugins = ws.plugins
  if plugins[0] == 0 then
    return nil
  end

  return get_next_and_collect, ctx
end


local function new_ws_data()
  return {
    plugins = { [0] = 0 },
    combos = {},
  }
end


local function configure(configurable, ctx)
  -- Disable hooks that are selectively enabled by plugins
  -- in their :configure handler
  req_dyn_hook_disable_by_default("observability_logs")

  ctx = ctx or ngx.ctx
  local kong_global = require "kong.global"
  for _, plugin in ipairs(CONFIGURABLE_PLUGINS) do
    local name = plugin.name

    kong_global.set_namespaced_log(kong, plugin.name, ctx)
    local start = get_updated_monotonic_ms()
    local ok, err = pcall(plugin.handler[CONFIGURE_PHASE], plugin.handler, configurable[name])
    local elapsed = get_updated_monotonic_ms() - start
    kong_global.reset_log(kong, ctx)

    if not ok then
      kong.log.err("failed to execute plugin '", name, ":", CONFIGURE_PHASE, " (", err, ")")
    else
      if elapsed > 50 then
        kong.log.notice("executing plugin '", name, ":", CONFIGURE_PHASE, " took excessively long: ", elapsed, " ms")
      end
    end
  end
end


local function create_configure(configurable)
  -- we only want the plugin_iterator:configure to be only available on proxying
  -- nodes (or data planes), thus we disable it if this code gets executed on control
  -- plane or on a node that does not listen any proxy ports.
  --
  -- TODO: move to PDK, e.g. kong.node.is_proxying()
  if kong.configuration.role == "control_plane"
  or ((subsystem == "http"   and #kong.configuration.proxy_listeners == 0) or
      (subsystem == "stream" and #kong.configuration.stream_listeners == 0))
  then
    return function() end
  end

  return function(self, ctx)
    configure(configurable, ctx)
    -- self destruct the function so that it cannot be called twice
    -- if it ever happens to be called twice, it should be very visible
    -- because of this.
    self.configure = nil
    configurable = nil
  end
end


function PluginsIterator.new(version)
  local is_not_dbless = kong.db.strategy ~= "off"
  if is_not_dbless then
    if not version then
      error("version must be given", 2)
    end
  end

  LOADED_PLUGINS = LOADED_PLUGINS or get_loaded_plugins()
  CONFIGURABLE_PLUGINS = CONFIGURABLE_PLUGINS or get_configurable_plugins()
  ENABLED_PLUGINS = ENABLED_PLUGINS or kong.configuration.loaded_plugins

  local ws_id = workspaces.get_workspace_id() or kong.default_workspace
  local ws = {
    [ws_id] = new_ws_data()
  }

  local counter = 0
  local globals
  do
    globals = {}
    for _, phase in ipairs(NON_COLLECTING_PHASES) do
      globals[phase] = { [0] = 0 }
    end
  end

  local configurable = {}
  local has_plugins = false
  local page_size = kong.db.plugins.pagination.max_page_size

  if kong.configuration.role == "control_plane" then
    goto done
  end

  for plugin, err in kong.db.plugins:each(page_size, GLOBAL_QUERY_OPTS) do
    if err then
      return nil, err
    end

    local name = plugin.name
    if not ENABLED_PLUGINS[name] then
      return nil, name .. " plugin is in use but not enabled"
    end

    if is_not_dbless and counter > 0 and counter % page_size == 0 and kong.core_cache then
      local new_version, err = kong.core_cache:get("plugins_iterator:version", TTL_ZERO, uuid)
      if err then
        return nil, "failed to retrieve plugins iterator version: " .. err
      end

      if new_version ~= version then
        -- the plugins iterator rebuild is being done by a different process at
        -- the same time, stop here and let the other one go for it
        kong.log.info("plugins iterator was changed while rebuilding it")
        return
      end
    end

    if should_process_plugin(plugin) then
      -- Get the plugin configuration for the specified workspace (ws_id)
      local cfg = get_plugin_config(plugin, name, plugin.ws_id)
      if cfg then
        has_plugins = true

        if CONFIGURABLE_PLUGINS[name] then
          configurable[name] = configurable[name] or {}
          configurable[name][#configurable[name] + 1] = cfg
        end

        local data = ws[plugin.ws_id]
        if not data then
          data = new_ws_data()
          ws[plugin.ws_id] = data
        end
        local plugins = data.plugins
        local combos = data.combos

        plugins[name] = true

        -- Retrieve route_id, service_id, and consumer_id from the plugin object, if they exist
        local route_id = plugin.route and plugin.route.id
        local service_id = plugin.service and plugin.service.id
        local consumer_id = plugin.consumer and plugin.consumer.id

        -- Determine if the plugin configuration is global (i.e., not tied to any route, service, consumer or group)
        if not (route_id or service_id or consumer_id) and plugin.ws_id == kong.default_workspace then
          -- Store the global configuration for the plugin in the 'globals' table
          globals[name] = cfg
        end

        -- Initialize an empty table for the plugin in the 'combos' table if it doesn't already exist
        combos[name] = combos[name] or {}

        -- Build a compound key using the route_id, service_id, and consumer_id
        local compound_key = build_compound_key(route_id, service_id, consumer_id)

        -- Store the plugin configuration in the 'combos' table using the compound key
        combos[name][compound_key] = cfg
      end
    end

    counter = counter + 1
  end

  if has_plugins then
    -- loaded_plugins contains all the plugins that we _may_ execute
    for _, plugin in ipairs(LOADED_PLUGINS) do
      local name = plugin.name
      -- ws contains all the plugins that are associated to the request via route/service/global mappings
      for _, data in pairs(ws) do
        local plugins = data.plugins
        if plugins[name] then -- is the plugin associated to the request(workspace/route/service)?
          local n = plugins[0] + 1
          plugins[n] = plugin -- next item goes into next slot
          plugins[0] = n      -- index 0 holds table size
          plugins[name] = nil -- remove the placeholder value
        end
      end

      local cfg = globals[name]
      if cfg then
        for _, phase in ipairs(NON_COLLECTING_PHASES) do
          if plugin.handler[phase] then
            local plugins = globals[phase]
            local n = plugins[0] + 2
            plugins[0] = n
            plugins[n] = cfg
            plugins[n - 1] = plugin
          end
        end
      end
    end
  end

::done::

  return {
    version = version,
    ws = ws,
    loaded = LOADED_PLUGINS,
    configure = create_configure(configurable),
    globals = globals,
    get_init_worker_iterator = get_init_worker_iterator,
    get_global_iterator = get_global_iterator,
    get_collecting_iterator = get_collecting_iterator,
    get_collected_iterator = get_collected_iterator,
    has_plugins = has_plugins,
    release = release,
  }
end


-- for testing
PluginsIterator.lookup_cfg = lookup_cfg
PluginsIterator.build_compound_key = build_compound_key


return PluginsIterator