Create custom error for fulfilling a pending message without losing track of failed events

Author: exanubes

errors.go

@@ -1,6 +1,9 @@
 package appsync
 
-import "github.com/exanubes/appsync/internal/app"
+import (
+	"github.com/exanubes/appsync/internal/app"
+	"github.com/exanubes/appsync/internal/app/protocol"
+)
 
 var (
 	ErrEmptyUrl              = app.ErrEmptyUrl
@@ -11,4 +14,5 @@ var (
 	ErrSubscriptionNotFound  = app.ErrSubscriptionNotFound
 	ErrHeartbeatTimeout      = app.ErrHeartbeatTimeout
 	ErrConnectionClosed      = app.ErrConnectionClosed
+	ErrBatchPublishFailed    = protocol.BatchPublishError{}
 )

internal/app/protocol/protocol.go

@@ -1,6 +1,10 @@
 package protocol
 
-import "github.com/exanubes/appsync/internal/app"
+import (
+	"fmt"
+
+	"github.com/exanubes/appsync/internal/app"
+)
 
 type KeepAliveMessage struct{}
 
@@ -46,3 +50,16 @@ type FailedEvent struct {
 	Index      int
 	RawMessage []byte
 }
+
+type BatchPublishError struct {
+	Failures []FailedEvent
+}
+
+func (err BatchPublishError) Error() string {
+	return fmt.Sprintf("batch publish failed: %d event(s) rejected", len(err.Failures))
+}
+
+func (err BatchPublishError) Is(target error) bool {
+	_, ok := target.(BatchPublishError)
+	return ok
+}

internal/app/router/router.go

@@ -35,9 +35,10 @@ func (router *MessageHandler) Handle(ctx context.Context, msg app.Message) error
 		return router.pending.Fulfill(ctx, msg.ID, nil)
 
 	case protocol.PublishResultMessage:
-		// TODO: When publishing a batch of events, need to be able to return an error per failed event or nil
-		return router.pending.Fulfill(ctx, msg.ID, nil)
-
+		if len(msg.Failures) == 0 {
+			return router.pending.Fulfill(ctx, msg.ID, nil)
+		}
+		return router.pending.Fulfill(ctx, msg.ID, protocol.BatchPublishError{Failures: msg.Failures})
 	case protocol.DataMessage:
 		return router.receive_use_case.Execute(ctx, msg)
 	}