Merge pull request #504 from Mashape/feat/cascade-delete

[feat/dao] implement cascade deletion for Cassandra

kong/dao/cassandra/apis.lua

@@ -26,31 +26,4 @@ function Apis:find_all()
   return apis
 end
 
--- @override
-function Apis:delete(where_t)
-  local ok, err = Apis.super.delete(self, where_t)
-  if not ok then
-    return false, err
-  end
-
-  -- delete all related plugins configurations
-  local plugins_dao = self._factory.plugins_configurations
-  local select_q, columns = query_builder.select(plugins_dao._table, {api_id = where_t.id}, plugins_dao._column_family_details)
-
-  for rows, err in plugins_dao:execute(select_q, columns, {api_id = where_t.id}, {auto_paging = true}) do
-    if err then
-      return nil, err
-    end
-
-    for _, row in ipairs(rows) do
-      local ok_del_plugin, err = plugins_dao:delete({id = row.id})
-      if not ok_del_plugin then
-        return nil, err
-      end
-    end
-  end
-
-  return ok
-end
-
 return {apis = Apis}

kong/dao/cassandra/base_dao.lua

@@ -44,6 +44,7 @@ function BaseDao:new(properties)
 
   self._properties = properties
   self._statements_cache = {}
+  self._cascade_delete_hooks = {}
 end
 
 -- Marshall an entity. Does nothing by default,
@@ -258,9 +259,7 @@ function BaseDao:execute(query, columns, args_to_bind, options)
   end
 
   -- Execute statement
-  local results, err = self:_execute(query, args, options)
-
-  return results, err
+  return self:_execute(query, args, options)
 end
 
 -- Check all fields marked with a `unique` in the schema do not already exist.
@@ -551,10 +550,50 @@ function BaseDao:find(page_size, paging_state)
   return self:find_by_keys(nil, page_size, paging_state)
 end
 
+-- Add a delete hook on a parent DAO of a foreign row.
+-- The delete hook will basically "cascade delete" all foreign rows of a parent row.
+-- @see cassandra/factory.lua ':load_daos()'
+-- @param foreign_dao_name Name (string) of the parent DAO
+-- @param foreign_column Name (string) of the foreign column
+-- @param parent_column Name (string) of the parent column identifying the parent row
+function BaseDao:add_delete_hook(foreign_dao_name, foreign_column, parent_column)
+
+  -- The actual delete hook
+  -- @param deleted_primary_key The value of the deleted row's primary key
+  -- @return boolean True if success, false otherwise
+  -- @return table A DAOError in case of error
+  local delete_hook = function(deleted_primary_key)
+    local foreign_dao = self._factory[foreign_dao_name]
+    local select_args = {
+      [foreign_column] = deleted_primary_key[parent_column]
+    }
+
+    -- Iterate over all rows with the foreign key and delete them.
+    -- Rows need to be deleted by PRIMARY KEY, and we only have the value of the foreign key, hence we need
+    -- to retrieve all rows with the foreign key, and then delete them, identifier by their own primary key.
+    local select_q, columns = query_builder.select(foreign_dao._table, select_args, foreign_dao._column_family_details )
+    for rows, err in foreign_dao:execute(select_q, columns, select_args, {auto_paging = true}) do
+      if err then
+        return false, err
+      end
+      for _, row in ipairs(rows) do
+        local ok_del_foreign_row, err = foreign_dao:delete(row)
+        if not ok_del_foreign_row then
+          return false, err
+        end
+      end
+    end
+
+    return true
+  end
+
+  table.insert(self._cascade_delete_hooks, delete_hook)
+end
+
 -- Delete the row at a given PRIMARY KEY.
 -- @param  `where_t` A table containing the PRIMARY KEY (columns/values) of the row to delete
 -- @return `success` True if deleted, false if otherwise or not found
--- @return `error`   Error if any during the query execution
+-- @return `error`   Error if any during the query execution or the cascade delete hook
 function BaseDao:delete(where_t)
   assert(self._primary_key ~= nil and type(self._primary_key) == "table" , "Entity does not have a primary_key")
   assert(where_t ~= nil and type(where_t) == "table", "where_t must be a table")
@@ -569,7 +608,21 @@ function BaseDao:delete(where_t)
 
   local t_primary_key = extract_primary_key(where_t, self._primary_key, self._clustering_key)
   local delete_q, where_columns = query_builder.delete(self._table, t_primary_key)
-  return self:execute(delete_q, where_columns, where_t)
+  local results, err = self:execute(delete_q, where_columns, where_t)
+  if err then
+    return false, err
+  end
+
+  -- Delete successful, trigger cascade delete hooks if any.
+  local foreign_err
+  for _, hook in ipairs(self._cascade_delete_hooks) do
+    foreign_err = select(2, hook(t_primary_key))
+    if foreign_err then
+      return false, foreign_err
+    end
+  end
+
+  return results
 end
 
 -- Truncate the table of this DAO

kong/dao/cassandra/consumers.lua

@@ -1,5 +1,4 @@
 local BaseDao = require "kong.dao.cassandra.base_dao"
-local query_builder = require "kong.dao.cassandra.query_builder"
 local consumers_schema = require "kong.dao.schemas.consumers"
 
 local Consumers = BaseDao:extend()
@@ -11,31 +10,4 @@ function Consumers:new(properties)
   Consumers.super.new(self, properties)
 end
 
--- @override
-function Consumers:delete(where_t)
-  local ok, err = Consumers.super.delete(self, where_t)
-  if not ok then
-    return false, err
-  end
-
-  local plugins_dao = self._factory.plugins_configurations
-  local select_q, columns = query_builder.select(plugins_dao._table, {consumer_id = where_t.id}, plugins_dao._column_family_details)
-
-  -- delete all related plugins configurations
-  for rows, err in plugins_dao:execute(select_q, columns, {consumer_id = where_t.id}, {auto_paging = true}) do
-    if err then
-      return nil, err
-    end
-
-    for _, row in ipairs(rows) do
-      local ok_del_plugin, err = plugins_dao:delete({id = row.id})
-      if not ok_del_plugin then
-        return nil, err
-      end
-    end
-  end
-
-  return ok
-end
-
-return { consumers = Consumers }
+return {consumers = Consumers}

kong/dao/cassandra/factory.lua

@@ -36,24 +36,53 @@ function CassandraFactory:new(properties, plugins)
 
   -- Load plugins DAOs
   if plugins then
-    for _, v in ipairs(plugins) do
-      local loaded, plugin_daos_mod = utils.load_module_if_exists("kong.plugins."..v..".daos")
-      if loaded then
-        if ngx then
-          ngx.log(ngx.DEBUG, "Loading DAO for plugin: "..v)
-        end
-        self:load_daos(plugin_daos_mod)
-      elseif ngx then
-        ngx.log(ngx.DEBUG, "No DAO loaded for plugin: "..v)
+    self:load_plugins(plugins)
+  end
+end
+
+-- Load an array of plugins (array of plugins names). If any of those plugins have DAOs,
+-- they will be loaded into the factory.
+-- @param plugins Array of plugins names
+function CassandraFactory:load_plugins(plugins)
+  for _, v in ipairs(plugins) do
+    local loaded, plugin_daos_mod = utils.load_module_if_exists("kong.plugins."..v..".daos")
+    if loaded then
+      if ngx then
+        ngx.log(ngx.DEBUG, "Loading DAO for plugin: "..v)
       end
+      self:load_daos(plugin_daos_mod)
+    elseif ngx then
+      ngx.log(ngx.DEBUG, "No DAO loaded for plugin: "..v)
     end
   end
 end
 
+-- Load a plugin's DAOs (plugins can have more than one DAO) in the factory and create cascade delete hooks.
+-- Cascade delete hooks are triggered when a parent of a foreign row is deleted.
+-- @param plugin_daos A table with key/values representing daos names and instances.
 function CassandraFactory:load_daos(plugin_daos)
+  local dao
   for name, plugin_dao in pairs(plugin_daos) do
-    self.daos[name] = plugin_dao(self._properties)
-    self.daos[name]._factory = self
+    dao = plugin_dao(self._properties)
+    dao._factory = self
+    self.daos[name] = dao
+    if dao._schema then
+      -- Check for any foreign relations to trigger cascade deletes
+      for field_name, field in pairs(dao._schema.fields) do
+        if field.foreign ~= nil then
+          -- Foreign key columns need to be queryable, hence they need to have an index
+          assert(field.queryable, "Foreign property "..field_name.." of shema "..name.." must be queryable (have an index)")
+
+          local parent_dao_name, parent_column = unpack(stringy.split(field.foreign, ":"))
+          assert(parent_dao_name ~= nil, "Foreign property "..field_name.." of schema "..name.." must contain 'parent_dao:parent_column")
+          assert(parent_column ~= nil, "Foreign property "..field_name.." of schema "..name.." must contain 'parent_dao:parent_column")
+
+          -- Add delete hook to the parent DAO
+          local parent_dao = self[parent_dao_name]
+          parent_dao:add_delete_hook(name, field_name, parent_column)
+        end
+      end
+    end
   end
 end
 

kong/plugins/basicauth/daos.lua

@@ -5,7 +5,7 @@ local SCHEMA = {
   fields = {
     id = { type = "id", dao_insert_value = true },
     created_at = { type = "timestamp", dao_insert_value = true },
-    consumer_id = { type = "id", required = true, foreign = "consumers:id" },
+    consumer_id = { type = "id", required = true, queryable = true, foreign = "consumers:id" },
     username = { type = "string", required = true, unique = true, queryable = true },
     password = { type = "string" }
   }

kong/plugins/keyauth/daos.lua

@@ -14,7 +14,7 @@ local SCHEMA = {
   fields = {
     id = { type = "id", dao_insert_value = true },
     created_at = { type = "timestamp", dao_insert_value = true },
-    consumer_id = { type = "id", required = true, foreign = "consumers:id" },
+    consumer_id = { type = "id", required = true, queryable = true, foreign = "consumers:id" },
     key = { type = "string", required = false, unique = true, queryable = true, func = generate_if_missing }
   }
 }

kong/plugins/oauth2/access.lua

@@ -213,9 +213,9 @@ local function issue_token(conf)
     response_params = {[ERROR] = "access_denied", error_description = "You must use HTTPS"}
   else
     local grant_type = parameters[GRANT_TYPE]
-    if not (grant_type == GRANT_AUTHORIZATION_CODE or 
-            grant_type == GRANT_REFRESH_TOKEN or 
-            (conf.enable_client_credentials and grant_type == GRANT_CLIENT_CREDENTIALS) or 
+    if not (grant_type == GRANT_AUTHORIZATION_CODE or
+            grant_type == GRANT_REFRESH_TOKEN or
+            (conf.enable_client_credentials and grant_type == GRANT_CLIENT_CREDENTIALS) or
             (conf.enable_password_grant and grant_type == GRANT_PASSWORD)) then
       response_params = {[ERROR] = "invalid_request", error_description = "Invalid "..GRANT_TYPE}
     end
@@ -353,7 +353,7 @@ function _M.execute(conf)
   -- Check if the API has a path and if it's being invoked with the path resolver
   local path_prefix = (ngx.ctx.api.path and stringy.startswith(ngx.var.request_uri, ngx.ctx.api.path)) and ngx.ctx.api.path or ""
   if stringy.endswith(path_prefix, "/") then
-    path_prefix = path_prefix:sub(1, path_prefix:len() - 1) 
+    path_prefix = path_prefix:sub(1, path_prefix:len() - 1)
   end
 
   if ngx.req.get_method() == "POST" then

kong/plugins/oauth2/daos.lua

@@ -20,7 +20,7 @@ local OAUTH2_CREDENTIALS_SCHEMA = {
   primary_key = {"id"},
   fields = {
     id = { type = "id", dao_insert_value = true },
-    consumer_id = { type = "id", required = true, foreign = "consumers:id" },
+    consumer_id = { type = "id", required = true, queryable = true, foreign = "consumers:id" },
     name = { type = "string", required = true },
     client_id = { type = "string", required = false, unique = true, queryable = true, func = generate_if_missing },
     client_secret = { type = "string", required = false, unique = true, func = generate_if_missing },
@@ -45,7 +45,7 @@ local OAUTH2_TOKENS_SCHEMA = {
   primary_key = {"id"},
   fields = {
     id = { type = "id", dao_insert_value = true },
-    credential_id = { type = "id", required = true, foreign = "oauth2_credentials:id" },
+    credential_id = { type = "id", required = true, queryable = true, foreign = "oauth2_credentials:id" },
     token_type = { type = "string", required = true, enum = { BEARER }, default = BEARER },
     access_token = { type = "string", required = false, unique = true, queryable = true, immutable = true, func = generate_if_missing },
     refresh_token = { type = "string", required = false, unique = true, queryable = true, immutable = true, func = generate_refresh_token },

kong/plugins/oauth2/migrations/cassandra.lua

@@ -55,6 +55,19 @@ local Migrations = {
         DROP TABLE oauth2_tokens;
       ]]
     end
+  },
+  {
+    name = "2015-08-24-215800_cascade_delete_index",
+    up = function()
+      return [[
+        CREATE INDEX IF NOT EXISTS oauth2_credential_id_idx ON oauth2_tokens(credential_id);
+      ]]
+    end,
+    down = function()
+      return [[
+        DROP INDEX oauth2_credential_id_idx;
+      ]]
+    end
   }
 }