add schema errors

conduit/errors.go

@@ -0,0 +1,71 @@
+// Copyright © 2024 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package conduit
+
+import (
+	"math"
+)
+
+const (
+	// ErrorCodeStart is the smallest error code which the wasm package can send.
+	ErrorCodeStart = math.MaxUint32 - 100
+
+	ErrorCodeSubjectNotFound = math.MaxUint32 - 100 + iota
+	ErrorCodeVersionNotFound
+	ErrorCodeInvalidSchema
+)
+
+var (
+	ErrorSubjectNotFound = NewError(ErrorCodeSubjectNotFound, "schema subject not found")
+	ErrorVersionNotFound = NewError(ErrorCodeVersionNotFound, "schema version not found")
+	ErrorInvalidSchema   = NewError(ErrorCodeInvalidSchema, "invalid schema")
+)
+
+type Error struct {
+	ErrCode uint32
+	Message string
+}
+
+func (e *Error) Error() string {
+	return e.Message
+}
+
+func (e *Error) Is(target error) bool {
+	t, ok := target.(*Error)
+	if !ok {
+		return false
+	}
+	return e.ErrCode == t.ErrCode
+}
+
+func NewError(code uint32, message string) *Error {
+	return &Error{
+		ErrCode: code,
+		Message: message,
+	}
+}
+
+func NewErrorFromCode(code uint32) *Error {
+	switch code {
+	case ErrorCodeSubjectNotFound:
+		return ErrorSubjectNotFound
+	case ErrorCodeVersionNotFound:
+		return ErrorVersionNotFound
+	case ErrorCodeInvalidSchema:
+		return ErrorInvalidSchema
+	default:
+		return NewError(code, "unknown error code")
+	}
+}

wasm/caller.go

@@ -18,6 +18,8 @@ package wasm
 
 import (
 	"unsafe"
+
+	"github.com/conduitio/conduit-processor-sdk/conduit"
 )
 
 // HostFunc is the function type for the imported functions from the host.
@@ -44,7 +46,7 @@ func hostCall(fn HostFunc, buf []byte) ([]byte, uint32, error) {
 		cmdSize := fn(ptr, uint32(len(buf)))
 		switch {
 		case cmdSize >= ErrorCodeStart: // error codes
-			return nil, cmdSize, NewErrorFromCode(cmdSize)
+			return nil, cmdSize, conduit.NewErrorFromCode(cmdSize)
 		case cmdSize > uint32(len(buf)) && i == 0: // not enough memory
 			oldSize := uint32(len(buf))
 			buf = append(buf, make([]byte, cmdSize-oldSize)...)

wasm/schema.go

@@ -42,7 +42,7 @@ func (*schemaService) CreateSchema(_ context.Context, req conduit.CreateSchemaRe
 
 	buffer, cmdSize, err := hostCall(_createSchema, buffer)
 	if err != nil {
-		return conduit.CreateSchemaResponse{}, err
+		return conduit.CreateSchemaResponse{}, fmt.Errorf("error while creating schema: %w", err)
 	}
 
 	var resp conduitv1.CreateSchemaResponse
@@ -68,7 +68,7 @@ func (*schemaService) GetSchema(_ context.Context, req conduit.GetSchemaRequest)
 
 	buffer, cmdSize, err := hostCall(_getSchema, buffer)
 	if err != nil {
-		return conduit.GetSchemaResponse{}, err
+		return conduit.GetSchemaResponse{}, fmt.Errorf("error while getting schema: %w", err)
 	}
 	var resp conduitv1.GetSchemaResponse
 	err = proto.Unmarshal(buffer[:cmdSize], &resp)