feat: remove schema resource and add default schema

Author: cchenggit

plugins/jsonschema_validator/jsonschema/jsonschema.go

@@ -15,7 +15,7 @@ type JSONSchema struct {
 func New(schemaDef []byte) *JSONSchema {
 	return &JSONSchema{
 		schemaDef: string(schemaDef),
-		hex:       utils.Hash256(string(schemaDef)),
+		hex:       utils.Sha256(string(schemaDef)),
 	}
 }
 

plugins/jsonschema_validator/plugin.go

@@ -6,72 +6,22 @@ import (
 	"errors"
 	"fmt"
 	"github.com/getkin/kin-openapi/openapi3"
-	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/webhookx-io/webhookx/pkg/errs"
+	"github.com/webhookx-io/webhookx/pkg/http/response"
 	"github.com/webhookx-io/webhookx/pkg/plugin"
+	"github.com/webhookx-io/webhookx/pkg/types"
 	"github.com/webhookx-io/webhookx/plugins/jsonschema_validator/jsonschema"
 	"github.com/webhookx-io/webhookx/utils"
-	"io"
-	"net/http"
-	"os"
-	"time"
 )
 
 type Config struct {
-	Schemas map[string]*SchemaResource `json:"schemas" validate:"dive,required"`
+	DraftVersion  int                `json:"draft_version" validate:"required,oneof=6"`
+	DefaultSchema string             `json:"default_schema" validate:"omitempty,json,max=1048576"`
+	Schemas       map[string]*Schema `json:"schemas" validate:"dive"`
 }
 
-type EventTypeSchema struct {
-	EventType  string `json:"event_type" validate:"required,max=100"`
-	JSONSchema string `json:"jsonschema" validate:"required,jsonschema,max=1048576"`
-}
-
-type SchemaResource struct {
-	JSONString string `json:"json" validate:"omitempty,json,max=1048576"`
-	File       string `json:"file" validate:"omitempty,file"`
-	URL        string `json:"url" validate:"omitempty,url"`
-}
-
-var cache, _ = lru.New[string, []byte](128)
-
-func (s *SchemaResource) Resource() ([]byte, string, error) {
-	// priority: json > file > url
-	if s.JSONString != "" {
-		return []byte(s.JSONString), "json", nil
-	}
-	if s.File != "" {
-		bytes, ok := cache.Get(s.File)
-		if ok {
-			return bytes, "file", nil
-		}
-		bytes, err := os.ReadFile(s.File)
-		if err != nil {
-			return nil, "file", fmt.Errorf("failed to read schema: %w", err)
-		}
-		cache.Add(s.File, bytes)
-		return bytes, "file", nil
-	}
-	if s.URL != "" {
-		bytes, ok := cache.Get(s.URL)
-		if ok {
-			return bytes, "url", nil
-		}
-		client := &http.Client{
-			Timeout: time.Second * 2,
-		}
-		resp, err := client.Get(s.URL)
-		if err != nil {
-			return nil, "url", fmt.Errorf("failed to fetch schema: %w", err)
-		}
-		defer func() { _ = resp.Body.Close() }()
-		body, err := io.ReadAll(resp.Body)
-		if err != nil {
-			return nil, "url", fmt.Errorf("failed to read schema from response: %w", err)
-		}
-		cache.Add(s.URL, body)
-		return body, "url", nil
-	}
-	return nil, "json", errors.New("no schema defined")
+type Schema struct {
+	Schema string `json:"schema" validate:"omitempty,json,max=1048576"`
 }
 
 type SchemaValidatorPlugin struct {
@@ -90,47 +40,62 @@ func New(config []byte) (plugin.Plugin, error) {
 	return p, nil
 }
 
+func unmarshalAndValidateSchema(schema string) (*openapi3.Schema, error) {
+	openapiSchema := &openapi3.Schema{}
+	err := openapiSchema.UnmarshalJSON([]byte(schema))
+	if err != nil {
+		return nil, fmt.Errorf("value must be a valid jsonschema")
+	}
+	err = openapiSchema.Validate(context.Background(), openapi3.EnableSchemaFormatValidation())
+	if err != nil {
+		return openapiSchema, err
+	}
+	return openapiSchema, nil
+}
+
 func (p *SchemaValidatorPlugin) ValidateConfig() error {
 	err := utils.Validate(p.Config)
 	if err != nil {
 		return err
 	}
 
 	e := errs.NewValidateError(errors.New("request validation"))
-	for event, schema := range p.Config.Schemas {
-		field := fmt.Sprintf("schemas[%s]", event)
-		if schema == nil {
-			e.Fields[field] = fmt.Errorf("schema is empty")
-			return e
-		}
-		schemaBytes, invalidField, err := schema.Resource()
+
+	var defaultErr error
+	if p.Config.DefaultSchema != "" {
+		_, err := unmarshalAndValidateSchema(p.Config.DefaultSchema)
 		if err != nil {
-			e.Fields[field] = map[string]string{
-				invalidField: err.Error(),
+			defaultErr = err
+			e.Fields = map[string]interface{}{
+				"default_schema": err.Error(),
 			}
-			return e
 		}
-		openapiSchema := &openapi3.Schema{}
-		err = openapiSchema.UnmarshalJSON(schemaBytes)
-		if err != nil {
-			e.Fields[field] = map[string]string{
-				invalidField: "the content must be a valid json string",
+	}
+
+	for event, schema := range p.Config.Schemas {
+		field := fmt.Sprintf("schemas[%s]", event)
+		if schema == nil || schema.Schema == "" {
+			if defaultErr != nil {
+				e.Fields[field] = map[string]string{
+					"schema": "invalid due to reusing the default_schema definition",
+				}
 			}
-			return e
-		}
-		err = openapiSchema.Validate(context.Background(), openapi3.EnableSchemaFormatValidation())
-		if err != nil {
-			e.Fields[field] = map[string]string{
-				invalidField: fmt.Sprintf("invalid jsonschema: %v", err),
+		} else {
+			_, err = unmarshalAndValidateSchema(schema.Schema)
+			if err != nil {
+				e.Fields[field] = map[string]string{
+					"schema": err.Error(),
+				}
 			}
-			return e
 		}
 	}
+	if len(e.Fields) > 0 {
+		return e
+	}
 	return nil
 }
 
 func (p *SchemaValidatorPlugin) ExecuteInbound(inbound *plugin.Inbound) (res plugin.InboundResult, err error) {
-	// parse body to get event type
 	var event map[string]any
 	body := inbound.RawBody
 	if err = json.Unmarshal(body, &event); err != nil {
@@ -149,24 +114,34 @@ func (p *SchemaValidatorPlugin) ExecuteInbound(inbound *plugin.Inbound) (res plu
 		return
 	}
 
-	schemaResource, ok := p.Config.Schemas[eventType]
-	if !ok || schemaResource == nil {
+	schema, ok := p.Config.Schemas[eventType]
+	if !ok {
 		res.Payload = body
 		return
 	}
-
-	bytes, _, err := schemaResource.Resource()
-	if err != nil {
-		return
+	if schema == nil || schema.Schema == "" {
+		if p.Config.DefaultSchema == "" {
+			res.Payload = body
+			return
+		}
+		schema = &Schema{
+			Schema: p.Config.DefaultSchema,
+		}
 	}
-	validator := jsonschema.New(bytes)
-	err = validator.Validate(&jsonschema.ValidatorContext{
+
+	validator := jsonschema.New([]byte(schema.Schema))
+	e := validator.Validate(&jsonschema.ValidatorContext{
 		HTTPRequest: &jsonschema.HTTPRequest{
 			R:    inbound.Request,
 			Data: data.(map[string]any),
 		},
 	})
-	if err != nil {
+	if e != nil {
+		response.JSON(inbound.Response, 400, types.ErrorResponse{
+			Message: "Request Validation",
+			Error:   e,
+		})
+		res.Terminated = true
 		return
 	}
 	res.Payload = body

proxy/gateway.go

@@ -14,7 +14,6 @@ import (
 	"github.com/webhookx-io/webhookx/dispatcher"
 	"github.com/webhookx-io/webhookx/eventbus"
 	"github.com/webhookx-io/webhookx/mcache"
-	"github.com/webhookx-io/webhookx/pkg/errs"
 	"github.com/webhookx-io/webhookx/pkg/http/response"
 	"github.com/webhookx-io/webhookx/pkg/loglimiter"
 	"github.com/webhookx-io/webhookx/pkg/metrics"
@@ -257,14 +256,6 @@ func (gw *Gateway) handle(w http.ResponseWriter, r *http.Request) bool {
 			RawBody:  body,
 		})
 		if err != nil {
-			if validateErr, ok := err.(*errs.ValidateError); ok {
-				response.JSON(w, 400, types.ErrorResponse{
-					Message: "Request Validation",
-					Error:   validateErr,
-				})
-				return false
-			}
-
 			gw.log.Errorf("failed to execute plugin: %v", err)
 			response.JSON(w, 500, types.ErrorResponse{Message: "internal error"})
 			return false

test/cmd/admin_test.go

@@ -101,7 +101,8 @@ var _ = Describe("admin", Ordered, func() {
 				assert.Equal(GinkgoT(), `{"function": "function handle() {}"}`, string(plugins[0].Config))
 				assert.Equal(GinkgoT(), `jsonschema-validator`, plugins[1].Name)
 				assert.Equal(GinkgoT(), true, plugins[1].Enabled)
-				assert.Equal(GinkgoT(), `{"schemas": {"charge.succeeded": {"url": "", "file": "", "json": "{\n  \"type\": \"object\",\n  \"properties\": {\n      \"id\": { \"type\": \"string\" },\n      \"amount\": { \"type\": \"integer\", \"minimum\": 1 },\n      \"currency\": { \"type\": \"string\", \"minLength\": 3, \"maxLength\": 6 }\n  },\n  \"required\": [\"id\", \"amount\", \"currency\"]\n}\n"}}}`, string(plugins[1].Config))
+
+				assert.Equal(GinkgoT(), `{"schemas": {"charge.succeeded": {"schema": "{\n  \"type\": \"object\",\n  \"properties\": {\n      \"id\": { \"type\": \"string\" },\n      \"amount\": { \"type\": \"integer\", \"minimum\": 1 },\n      \"currency\": { \"type\": \"string\", \"minLength\": 3, \"maxLength\": 6 }\n  },\n  \"required\": [\"id\", \"amount\", \"currency\"]\n}\n"}}, "draft_version": 6, "default_schema": "{\n  \"type\": \"object\",\n  \"properties\": {\n      \"id\": { \"type\": \"string\" }\n  },\n  \"required\": [\"id\"]\n}\n"}`, string(plugins[1].Config))
 			})
 
 			It("entities not defined in the declarative configuration should be deleted", func() {

test/declarative/declarative_test.go

@@ -41,42 +41,38 @@ endpoints:
       - name: foo
 `
 
-	invalidSourcePluginJSONSchemaJSONYAML = `
+	invalidSourcePluginJSONSchemaConfigYAML = `
 sources:
   - name: default-source
     path: /
     methods: ["POST"]
     plugins:
       - name: "jsonschema-validator"
         config:
+          default_schema: |
+            %s
           schemas:
             charge.succeed:
-              json: '%s'
+              schema: |
+                %s
 `
 
-	invalidSourcePluginJSONSchemaFileYAML = `
-sources:
-  - name: default-source
-    path: /
-    methods: ["POST"]
-    plugins:
-      - name: "jsonschema-validator"
-        config:
-          schemas:
-            charge.succeed:
-              file: "%s"
-`
-	invalidSourcePluginJSONSchemaURLYAML = `
+	invalidSourcePluginJSONSchemaJSONYAML = `
 sources:
   - name: default-source
     path: /
     methods: ["POST"]
     plugins:
       - name: "jsonschema-validator"
         config:
+          draft_version: 6
+          default_schema: |
+            %s
           schemas:
             charge.succeed:
-              url: "http://localhost/charge.succeed.json"
+              schema: |
+                %s
+            reuse.default_schema:
 `
 )
 
@@ -143,59 +139,30 @@ var _ = Describe("Declarative", Ordered, func() {
 				assert.Equal(GinkgoT(), `{"message":"Request Validation","error":{"message":"request validation","fields":{"name":"unknown plugin name 'foo'"}}}`, string(resp.Body()))
 			})
 
-			It("should return 400 for invalid jsonschema-validator plugin config json string", func() {
-				resp, err := adminClient.R().
-					SetBody(fmt.Sprintf(invalidSourcePluginJSONSchemaJSONYAML, "invalid jsonstring")).
-					Post("/workspaces/default/config/sync")
-				assert.Nil(GinkgoT(), err)
-				assert.Equal(GinkgoT(), 400, resp.StatusCode())
-				assert.Equal(GinkgoT(),
-					`{"message":"Request Validation","error":{"message":"request validation","fields":{"config":{"schemas[charge.succeed]":{"json":"value must be a valid json string"}}}}}`,
-					string(resp.Body()))
-			})
-
-			It("should return 400 for invalid jsonschema-validator plugin config jsonschema", func() {
-				resp, err := adminClient.R().
-					SetBody(fmt.Sprintf(invalidSourcePluginJSONSchemaJSONYAML, `{"type":"invalidObject"}`)).
-					Post("/workspaces/default/config/sync")
-				assert.Nil(GinkgoT(), err)
-				assert.Equal(GinkgoT(), 400, resp.StatusCode())
-				assert.Equal(GinkgoT(),
-					`{"message":"Request Validation","error":{"message":"request validation","fields":{"config":{"schemas[charge.succeed]":{"json":"invalid jsonschema: unsupported 'type' value \"invalidObject\""}}}}}`,
-					string(resp.Body()))
-			})
-
-			It("should return 400 for invalid jsonschema-validator plugin config file", func() {
+			It("should return 400 for invalid jsonschema-validator plugin config", func() {
 				resp, err := adminClient.R().
-					SetBody(fmt.Sprintf(invalidSourcePluginJSONSchemaFileYAML, "./notexist.json")).
+					SetBody(
+						fmt.Sprintf(invalidSourcePluginJSONSchemaConfigYAML, "invalid jsonschema", "invalid jsonschema"),
+					).
 					Post("/workspaces/default/config/sync")
 
 				assert.Nil(GinkgoT(), err)
 				assert.Equal(GinkgoT(), 400, resp.StatusCode())
 				assert.Equal(GinkgoT(),
-					`{"message":"Request Validation","error":{"message":"request validation","fields":{"config":{"schemas[charge.succeed]":{"file":"value must be a valid exist file"}}}}}`,
-					string(resp.Body()))
-			})
-
-			It("should return 400 for invalid jsonschema-validator config file content", func() {
-				resp, err := adminClient.R().
-					SetBody(fmt.Sprintf(invalidSourcePluginJSONSchemaFileYAML, "../fixtures/jsonschema/invalid.json")).
-					Post("/workspaces/default/config/sync")
-				assert.Nil(GinkgoT(), err)
-				assert.Equal(GinkgoT(), 400, resp.StatusCode())
-				assert.Equal(GinkgoT(),
-					`{"message":"Request Validation","error":{"message":"request validation","fields":{"config":{"schemas[charge.succeed]":{"file":"the content must be a valid json string"}}}}}`,
+					`{"message":"Request Validation","error":{"message":"request validation","fields":{"config":{"default_schema":"value must be a valid json string","draft_version":"required field missing","schemas[charge.succeed]":{"schema":"value must be a valid json string"}}}}}`,
 					string(resp.Body()))
 			})
 
-			It("should return 400 for invalid source plugin config url", func() {
+			It("should return 400 for invalid jsonschema-validator plugin config jsonschema string", func() {
 				resp, err := adminClient.R().
-					SetBody(invalidSourcePluginJSONSchemaURLYAML).
+					SetBody(fmt.Sprintf(invalidSourcePluginJSONSchemaJSONYAML,
+						`{"type": "invlidObject","properties": {"id": { "type": "string"}}}`,
+						`{"type": "object","properties": {"id": { "type": "number", "format":"invalid"}}}`)).
 					Post("/workspaces/default/config/sync")
 				assert.Nil(GinkgoT(), err)
 				assert.Equal(GinkgoT(), 400, resp.StatusCode())
 				assert.Equal(GinkgoT(),
-					`{"message":"Request Validation","error":{"message":"request validation","fields":{"config":{"schemas[charge.succeed]":{"url":"failed to fetch schema: Get \"http://localhost/charge.succeed.json\": dial tcp [::1]:80: connect: connection refused"}}}}}`,
+					`{"message":"Request Validation","error":{"message":"request validation","fields":{"config":{"default_schema":"unsupported 'type' value \"invlidObject\"","schemas[charge.succeed]":{"schema":"unsupported 'format' value \"invalid\""},"schemas[reuse.default_schema]":{"schema":"invalid due to reusing the default_schema definition"}}}}}`,
 					string(resp.Body()))
 			})
 		})

test/fixtures/webhookx.yml

@@ -39,11 +39,18 @@ sources:
           function: "function handle() {}"
       - name: "jsonschema-validator"
         config:
+          draft_version: 6
+          default_schema: |
+            {
+              "type": "object",
+              "properties": {
+                  "id": { "type": "string" }
+              },
+              "required": ["id"]
+            }
           schemas:
             charge.succeeded:
-              #file: "../fixtures/jsonschema/charge.succeed.json"
-              #url: "https://raw.githubusercontent.com/cchenggit/webhookx/refs/heads/feat/plugin-jsonschema-validator/test/fixtures/jsonschema/charge.succeed.json"
-              json: |
+              schema: |
                 {
                   "type": "object",
                   "properties": {