Use better validator

.VERSION

@@ -1 +1 @@
-v0.15.13
+v0.16.0

deploy/terraform/aws/lambda/variables.tf

@@ -29,7 +29,7 @@ variable "buz_domain" {
 variable "buz_version" {
   description = "The version of Buz to run."
   type        = string
-  default     = "v0.15.13"
+  default     = "v0.16.0"
 }
 
 variable "buz_lambda_memory_limit" {

deploy/terraform/gcp/cloud_run/variables.tf

@@ -28,7 +28,7 @@ variable "buz_domain" {
 variable "buz_version" {
   description = "The version of Buz to run."
   type        = string
-  default     = "v0.15.13"
+  default     = "v0.16.0"
 }
 
 variable "buz_service_timeout_seconds" {

examples/quickstart/docker-compose.yml

@@ -20,7 +20,7 @@ x-dependency:
 services:
   buz:
     container_name: buz
-    image: ghcr.io/silverton-io/buz:latest
+    image: ghcr.io/silverton-io/buz:v0.16.0
     volumes:
       - type: bind
         source: ./buz/quickstart.conf.yml

go.mod

@@ -18,12 +18,11 @@ require (
 	github.com/gin-contrib/pprof v1.4.0
 	github.com/gin-contrib/timeout v0.0.3
 	github.com/gin-gonic/gin v1.8.1
-	github.com/go-sql-driver/mysql v1.6.0
 	github.com/google/uuid v1.3.0
 	github.com/minio/minio-go/v7 v7.0.34
 	github.com/nats-io/nats.go v1.15.0
-	github.com/qri-io/jsonschema v0.2.1
 	github.com/rs/zerolog v1.26.1
+	github.com/santhosh-tekuri/jsonschema/v5 v5.3.0
 	github.com/spf13/viper v1.10.1
 	github.com/stretchr/testify v1.8.2
 	github.com/tidwall/gjson v1.13.0
@@ -68,6 +67,7 @@ require (
 	github.com/go-playground/locales v0.14.0 // indirect
 	github.com/go-playground/universal-translator v0.18.0 // indirect
 	github.com/go-playground/validator/v10 v10.10.0 // indirect
+	github.com/go-sql-driver/mysql v1.6.0 // indirect
 	github.com/go-stack/stack v1.8.0 // indirect
 	github.com/goccy/go-json v0.9.7 // indirect
 	github.com/gofrs/uuid v4.2.0+incompatible // indirect
@@ -110,7 +110,6 @@ require (
 	github.com/pierrec/lz4/v4 v4.1.14 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/qri-io/jsonpointer v0.1.1 // indirect
 	github.com/rs/xid v1.4.0 // indirect
 	github.com/shopspring/decimal v1.3.1 // indirect
 	github.com/sirupsen/logrus v1.9.0 // indirect

go.sum

@@ -442,10 +442,6 @@ github.com/pkg/sftp v1.13.1/go.mod h1:3HaPG6Dq1ILlpPZRO0HVMrsydcdLt6HRDccSgb87qR
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
-github.com/qri-io/jsonpointer v0.1.1 h1:prVZBZLL6TW5vsSB9fFHFAMBLI4b0ri5vribQlTJiBA=
-github.com/qri-io/jsonpointer v0.1.1/go.mod h1:DnJPaYgiKu56EuDp8TU5wFLdZIcAnb/uH9v37ZaMV64=
-github.com/qri-io/jsonschema v0.2.1 h1:NNFoKms+kut6ABPf6xiKNM5214jzxAhDBrPHCJ97Wg0=
-github.com/qri-io/jsonschema v0.2.1/go.mod h1:g7DPkiOsK1xv6T/Ao5scXRkd+yTFygcANPBaaqW+VrI=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
 github.com/rogpeppe/go-internal v1.8.0 h1:FCbCCtXNOY3UtUuHUYaghJg4y7Fd14rXifAYUAtL9R8=
@@ -459,9 +455,9 @@ github.com/rs/zerolog v1.15.0/go.mod h1:xYTKnLHcpfU2225ny5qZjxnj9NvkumZYjJHlAThC
 github.com/rs/zerolog v1.26.1 h1:/ihwxqH+4z8UxyI70wM1z9yCvkWcfz/a3mj48k/Zngc=
 github.com/rs/zerolog v1.26.1/go.mod h1:/wSSJWX7lVrsOwlbyTRSOJvqRlc+WjWlfes+CiJ+tmc=
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
+github.com/santhosh-tekuri/jsonschema/v5 v5.3.0 h1:uIkTLo0AGRc8l7h5l9r+GcYi9qfVPt6lD4/bhmzfiKo=
+github.com/santhosh-tekuri/jsonschema/v5 v5.3.0/go.mod h1:FKdcjfQW6rpZSnxxUvEA5H/cDPdvJ/SZJQLWWXWGrZ0=
 github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
-github.com/sergi/go-diff v1.0.0 h1:Kpca3qRNrduNnOQeazBd0ysaKrUJiIuISHxogkT9RPQ=
-github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
 github.com/shopspring/decimal v0.0.0-20180709203117-cd690d0c9e24/go.mod h1:M+9NzErvs504Cn4c5DxATwIqPbtswREoFCre64PpcG4=
 github.com/shopspring/decimal v1.2.0/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o=
 github.com/shopspring/decimal v1.3.1 h1:2Usl1nmF/WZucqkFZhnfFYxxxu8LG21F6nPQBE5gKV8=

pkg/backend/mysqldb/sink.go

@@ -7,7 +7,6 @@ package mysqldb
 import (
 	"context"
 
-	_ "github.com/go-sql-driver/mysql"
 	"github.com/rs/zerolog/log"
 	"github.com/silverton-io/buz/pkg/backend/backendutils"
 	"github.com/silverton-io/buz/pkg/config"

pkg/registry/registry.go

@@ -9,12 +9,15 @@ import (
 
 	"github.com/coocood/freecache"
 	"github.com/rs/zerolog/log"
+	"github.com/santhosh-tekuri/jsonschema/v5"
+	_ "github.com/santhosh-tekuri/jsonschema/v5/httploader"
 	"github.com/silverton-io/buz/pkg/config"
 )
 
 type Registry struct {
 	Cache        *freecache.Cache
 	Backend      SchemaCacheBackend
+	Compiler     *jsonschema.Compiler
 	maxSizeBytes int
 	ttlSeconds   int
 }
@@ -29,6 +32,7 @@ func (r *Registry) Initialize(conf config.Registry) error {
 	r.Cache = freecache.NewCache(conf.MaxSizeBytes)
 	r.maxSizeBytes = conf.MaxSizeBytes
 	r.ttlSeconds = conf.TtlSeconds
+	r.Compiler = jsonschema.NewCompiler()
 	return nil
 }
 
@@ -55,6 +59,11 @@ func (r *Registry) Get(key string) (exists bool, data []byte) {
 			log.Error().Err(err).Msg("🔴 error when setting key " + key)
 		}
 		log.Debug().Msg("🟡 " + key + " cached successfully")
+		log.Debug().Msg("🟡 adding schema to compiler " + key)
+		err = r.Compiler.AddResource(key, strings.NewReader(string(schemaContents)))
+		if err != nil {
+			log.Error().Err(err).Msg("🔴 error when compiling schema " + key)
+		}
 		return true, schemaContents // Schema was aquired from remote backed and cached successfully
 	}
 }

pkg/validator/validator.go

@@ -6,8 +6,11 @@ package validator
 
 import (
 	"encoding/json"
+	"errors"
+	"time"
 
 	"github.com/rs/zerolog/log"
+	"github.com/santhosh-tekuri/jsonschema/v5"
 	"github.com/silverton-io/buz/pkg/constants"
 	"github.com/silverton-io/buz/pkg/envelope"
 	"github.com/silverton-io/buz/pkg/protocol"
@@ -36,25 +39,34 @@ func Validate(e envelope.Envelope, registry *registry.Registry) (isValid bool, v
 		}
 		return false, validationError, nil
 	} else {
-		var payloadToValidate []byte
-		var err error
+		schema, err := registry.Compiler.Compile(e.Schema)
+		if err != nil {
+			log.Error().Err(err).Msg("could not compile schema")
+			validationError := envelope.ValidationError{
+				ErrorType:       &InvalidSchema.Type,
+				ErrorResolution: &InvalidSchema.Resolution,
+				Errors:          nil,
+			}
+			return false, validationError, schemaContents
+		}
+		var payloadToValidate interface{}
 		// Snowplow events have to be handled separately, as `self_describing_event` is
 		// the only portion that is validated according to a jsonschema.
 		if e.Protocol == protocol.SNOWPLOW {
-			e := e.Payload["self_describing_event"].(map[string]interface{})["data"]
-			payloadToValidate, err = json.Marshal(e)
+			payloadToValidate = e.Payload["self_describing_event"].(map[string]interface{})["data"]
 		} else {
-			payloadToValidate, err = e.Payload.AsByte()
-		}
-		// If the payload cannot be marshaled it should be considered invalid.
-		if err != nil {
-			log.Error().Stack().Err(err).Msg("🔴 could not marshal payload")
-			validationError := envelope.ValidationError{
-				ErrorType:       &InvalidPayload.Type,
-				ErrorResolution: &InvalidPayload.Resolution,
-				Errors:          nil,
+			contents, _ := e.Payload.AsByte()
+			err := json.Unmarshal(contents, &payloadToValidate)
+			if err != nil {
+				log.Error().Err(err).Msg("could not unmarshal payload")
+				// If the payload cannot be unmarshaled it should be considered invalid
+				validationError := envelope.ValidationError{
+					ErrorType:       &InvalidPayload.Type,
+					ErrorResolution: &InvalidPayload.Resolution,
+					Errors:          nil,
+				}
+				return false, validationError, nil
 			}
-			return false, validationError, nil
 		}
 		// If the payload is not present at all it should be considered invalid.
 		if payloadToValidate == nil {
@@ -65,7 +77,30 @@ func Validate(e envelope.Envelope, registry *registry.Registry) (isValid bool, v
 			}
 			return false, validationError, nil
 		}
-		isValid, validationError := validatePayload(payloadToValidate, schemaContents)
-		return isValid, validationError, schemaContents
+		startTime := time.Now().UTC()
+		vErr := schema.Validate(payloadToValidate)
+		log.Debug().Msg("🟡 event validated in " + time.Now().UTC().Sub(startTime).String())
+		jsonschemaValidationErr := &jsonschema.ValidationError{}
+		if vErr != nil {
+			if errors.As(vErr, &jsonschemaValidationErr) {
+				var validationErrs = []envelope.PayloadValidationError{}
+				for _, cause := range jsonschemaValidationErr.Causes {
+					validationErr := envelope.PayloadValidationError{
+						Field:       cause.InstanceLocation,
+						Description: cause.Message,
+						ErrorType:   cause.KeywordLocation,
+					}
+					validationErrs = append(validationErrs, validationErr)
+				}
+				validationError := envelope.ValidationError{
+					ErrorType:       &InvalidPayload.Type,
+					ErrorResolution: &InvalidPayload.Resolution,
+					Errors:          validationErrs,
+				}
+				return false, validationError, schemaContents
+			}
+			return false, envelope.ValidationError{}, schemaContents
+		}
+		return true, envelope.ValidationError{}, schemaContents
 	}
 }