feat(db): entity transformations

### Summary

Removes the need to specify `input`/`needs` tables for transformations so that
full entity transformations can be implemented.

CHANGELOG.md

@@ -70,10 +70,13 @@
 ### Additions
 
 #### Core
+
 - Allow `kong.conf` ssl properties to be stored in vaults or environment
   variables. Allow such properties to be configured directly as content
   or base64 encoded content.
   [#9253](https://github.com/Kong/kong/pull/9253)
+- Add support for full entity transformations in schemas
+  [#9431](https://github.com/Kong/kong/pull/9431)
 
 ### Fixes
 

kong/db/schema/init.lua

@@ -1364,45 +1364,49 @@ local function run_transformation_checks(schema_or_subschema, input, original_in
   if transformations then
     for i = 1, #transformations do
       local transformation = transformations[i]
-      local args = {}
-      local argc = 0
-      local none_set = true
-      for j = 1, #transformation.input do
-        local input_field_name = transformation.input[j]
-        if is_nonempty(get_field(original_input or input, input_field_name)) then
-          none_set = false
-        end
+      if transformation.input or transformation.needs then
+        local args = {}
+        local argc = 0
+        local none_set = true
+        if transformation.input then
+          for j = 1, #transformation.input do
+            local input_field_name = transformation.input[j]
+            if is_nonempty(get_field(original_input or input, input_field_name)) then
+              none_set = false
+            end
 
-        argc = argc + 1
-        args[argc] = input_field_name
-      end
+            argc = argc + 1
+            args[argc] = input_field_name
+          end
+        end
 
-      local needs_changed = false
-      if transformation.needs then
-        for j = 1, #transformation.needs do
-          local input_field_name = transformation.needs[j]
-          if rbw_entity and not needs_changed then
-            local value = get_field(original_input or input, input_field_name)
-            local rbw_value = get_field(rbw_entity, input_field_name)
-            if value ~= rbw_value then
-              needs_changed = true
+        local needs_changed = false
+        if transformation.needs then
+          for j = 1, #transformation.needs do
+            local input_field_name = transformation.needs[j]
+            if rbw_entity and not needs_changed then
+              local value = get_field(original_input or input, input_field_name)
+              local rbw_value = get_field(rbw_entity, input_field_name)
+              if value ~= rbw_value then
+                needs_changed = true
+              end
             end
-          end
 
-          argc = argc + 1
-          args[argc] = input_field_name
+            argc = argc + 1
+            args[argc] = input_field_name
+          end
         end
-      end
-
-      if needs_changed or (not none_set) then
-        local ok, err = mutually_required(needs_changed and original_input or input, args)
-        if not ok then
-          insert_entity_error(errors, validation_errors.MUTUALLY_REQUIRED:format(err))
 
-        else
-          ok, err = mutually_required(original_input or input, transformation.input)
+        if needs_changed or (not none_set) then
+          local ok, err = mutually_required(needs_changed and original_input or input, args)
           if not ok then
             insert_entity_error(errors, validation_errors.MUTUALLY_REQUIRED:format(err))
+
+          else
+            ok, err = mutually_required(original_input or input, transformation.input)
+            if not ok then
+              insert_entity_error(errors, validation_errors.MUTUALLY_REQUIRED:format(err))
+            end
           end
         end
       end
@@ -2243,17 +2247,26 @@ local function run_transformations(self, transformations, input, original_input,
     end
 
     if transform then
-      local args = get_transform_args(input, original_input, output, transformation)
-      if args then
-        local data, err = transform(unpack(args))
+      if transformation.input or transformation.needs then
+        local args = get_transform_args(input, original_input, output, transformation)
+        if args then
+          local data, err = transform(unpack(args))
+          if err then
+            return nil, validation_errors.TRANSFORMATION_ERROR:format(err)
+          end
+
+          output = self:merge_values(data, output or input)
+        end
+
+      else
+        local data, err = transform(output or input)
         if err then
           return nil, validation_errors.TRANSFORMATION_ERROR:format(err)
         end
 
         output = self:merge_values(data, output or input)
       end
     end
-
   end
 
   return output or input

kong/db/schema/metaschema.lua

@@ -131,7 +131,7 @@ local transformations_array = {
       {
         input = {
           type = "array",
-          required = true,
+          required = false,
           elements = {
             type = "string"
           },
@@ -417,13 +417,15 @@ local check_fields = function(schema, errors)
   if transformations then
     for i = 1, #transformations do
       local transformation = transformations[i]
-      for j = 1, #transformation.input do
-        local input = transformation.input[j]
-        if not has_schema_field(schema, input) then
-          errors.transformations = errors.transformations or {}
-          errors.transformations.input = errors.transformations.input or {}
-          errors.transformations.input[i] = errors.transformations.input[i] or {}
-          errors.transformations.input[i][j] = fmt("invalid field name: %s", input)
+      if transformation.input then
+        for j = 1, #transformation.input do
+          local input = transformation.input[j]
+          if not has_schema_field(schema, input) then
+            errors.transformations = errors.transformations or {}
+            errors.transformations.input = errors.transformations.input or {}
+            errors.transformations.input[i] = errors.transformations.input[i] or {}
+            errors.transformations.input[i][j] = fmt("invalid field name: %s", input)
+          end
         end
       end
 
@@ -740,22 +742,24 @@ local MetaSchema = Schema.new({
     if transformations then
       for i = 1, #transformations do
         local input = transformations[i].input
-        for j = 1, #input do
-          if not has_schema_field(schema, input[j]) then
-            if not errors.transformations then
-              errors.transformations = {}
-            end
+        if input then
+          for j = 1, #input do
+            if not has_schema_field(schema, input[j]) then
+              if not errors.transformations then
+                errors.transformations = {}
+              end
 
-            if not errors.transformations.input then
-              errors.transformations.input = {}
-            end
+              if not errors.transformations.input then
+                errors.transformations.input = {}
+              end
 
 
-            if not errors.transformations.input[i] then
-              errors.transformations.input[i] = {}
-            end
+              if not errors.transformations.input[i] then
+                errors.transformations.input[i] = {}
+              end
 
-            errors.transformations.input[i][j] = fmt("invalid field name: %s", input)
+              errors.transformations.input[i][j] = fmt("invalid field name: %s", input)
+            end
           end
         end
 

spec/01-unit/01-db/01-schema/01-schema_spec.lua

@@ -3945,6 +3945,68 @@ describe("schema", function()
 
   for i = 1, 2 do
   describe("transform (" .. SchemaKind[i].name .. ")", function()
+    it("transforms entity", function()
+      local test_schema = {
+        name = "test",
+        fields = {
+          {
+            name = {
+              type = "string"
+            },
+          },
+        },
+        transformations = {
+          {
+            on_write = function(entity)
+              return { name = entity.name:upper() }
+            end,
+          },
+        },
+      }
+      local entity = { name = "test1" }
+
+      local TestEntities = SchemaKind[i].new(test_schema)
+      local transformed_entity, _ = TestEntities:transform(entity)
+
+      assert.truthy(transformed_entity)
+      assert.equal("TEST1", transformed_entity.name)
+    end)
+
+    it("transforms entity on write and read", function()
+      local test_schema = {
+        name = "test",
+        fields = {
+          {
+            name = {
+              type = "string"
+            },
+          },
+        },
+        transformations = {
+          {
+            on_write = function(entity)
+              return { name = entity.name:upper() }
+            end,
+            on_read = function(entity)
+              return { name = entity.name:lower() }
+            end,
+          },
+        },
+      }
+      local entity = { name = "TeSt1" }
+
+      local TestEntities = SchemaKind[i].new(test_schema)
+      local transformed_entity, _ = TestEntities:transform(entity)
+
+      assert.truthy(transformed_entity)
+      assert.equal("TEST1", transformed_entity.name)
+
+      transformed_entity, _ = TestEntities:transform(transformed_entity, nil, "select")
+
+      assert.truthy(transformed_entity)
+      assert.equal("test1", transformed_entity.name)
+    end)
+
     it("transforms fields", function()
       local test_schema = {
         name = "test",
@@ -4095,6 +4157,38 @@ describe("schema", function()
       assert.equal("test1", transformed_entity.name)
     end)
 
+    it("transforms entity with multiple transformations", function()
+      local test_schema = {
+        name = "test",
+        fields = {
+          {
+            name = {
+              type = "string"
+            },
+          },
+        },
+        transformations = {
+          {
+            on_write = function(entity)
+              return { name = "How are you " .. entity.name }
+            end,
+          },
+          {
+            on_write = function(entity)
+              return { name = entity.name .. "?" }
+            end,
+          },
+        },
+      }
+
+      local entity = { name = "Bob" }
+
+      local TestEntities = SchemaKind[i].new(test_schema)
+      local transformed_entity, _ = TestEntities:transform(entity)
+
+      assert.truthy(transformed_entity)
+      assert.equal("How are you Bob?", transformed_entity.name)
+    end)
 
     it("transforms fields with multiple transformations", function()
       local test_schema = {
@@ -4164,6 +4258,33 @@ describe("schema", function()
       assert.equal(3, transformed_entity.age)
     end)
 
+    it("returns error if entity transformation returns an error", function()
+      local test_schema = {
+        name = "test",
+        fields = {
+          {
+            name = {
+              type = "string"
+            },
+          },
+        },
+        transformations = {
+          {
+            on_write = function(entity)
+              return nil, "unable to transform entity"
+            end,
+          },
+        },
+      }
+      local entity = { name = "test1" }
+
+      local TestEntities = SchemaKind[i].new(test_schema)
+      local transformed_entity, err = TestEntities:transform(entity)
+
+      assert.falsy(transformed_entity)
+      assert.equal("transformation failed: unable to transform entity", err)
+    end)
+
     it("returns error if transformation returns an error", function()
       local test_schema = {
         name = "test",
@@ -4192,6 +4313,7 @@ describe("schema", function()
       assert.equal("transformation failed: unable to transform name", err)
     end)
 
+
     it("skips transformation if needs are not fulfilled", function()
       local test_schema = {
         name = "test",

spec/01-unit/01-db/01-schema/02-metaschema_spec.lua

@@ -605,6 +605,19 @@ describe("metaschema", function()
   end)
 
   it("validates transformation has transformation function specified (positive)", function()
+    assert.truthy(MetaSchema:validate({
+      name = "test",
+      primary_key = { "test" },
+      fields = {
+        { test = { type = "string" } },
+      },
+      transformations = {
+        {
+          on_write = function() return true end,
+        },
+      },
+    }))
+
     assert.truthy(MetaSchema:validate({
       name = "test",
       primary_key = { "test" },
@@ -1121,6 +1134,18 @@ describe("metasubschema", function()
   end
 
   it("validates transformation has transformation function specified (positive)", function()
+    assert.truthy(MetaSchema.MetaSubSchema:validate({
+      name = "test",
+      fields = {
+        { test = { type = "string" } },
+      },
+      transformations = {
+        {
+          on_write = function() return true end,
+        },
+      },
+    }))
+
     assert.truthy(MetaSchema.MetaSubSchema:validate({
       name = "test",
       fields = {