Add support for validation scopes.

app.go

@@ -8,6 +8,7 @@ import (
 	"github.com/dogmatiq/configkit/message"
 	"github.com/dogmatiq/dodeca/logging"
 	"github.com/dogmatiq/dogma"
+	"github.com/dogmatiq/enginekit/collections/sets"
 	"github.com/dogmatiq/interopspec/envelopespec"
 	"github.com/dogmatiq/verity/eventstream"
 	"github.com/dogmatiq/verity/eventstream/memorystream"
@@ -88,7 +89,7 @@ func (e *Engine) initApp(
 
 	e.apps[id.Key] = a
 
-	for mt := range x.Packer.Produced {
+	for mt := range x.Packer.Produced.All() {
 		e.executors[mt] = x
 	}
 
@@ -126,10 +127,11 @@ func (e *Engine) newEventCache(
 		// TODO: https://github.com/dogmatiq/verity/issues/226
 		// Make buffer size configurable.
 		BufferSize: 0,
-		Types: cfg.
-			MessageTypes().
-			Produced.
-			FilterByRole(message.EventRole),
+		Types: sets.NewFromKeys(
+			cfg.
+				MessageTypes().
+				Produced(message.EventKind),
+		),
 	}, nil
 }
 
@@ -147,10 +149,11 @@ func (e *Engine) newEventStream(
 		// TODO: https://github.com/dogmatiq/verity/issues/76
 		// Make pre-fetch buffer size configurable.
 		PreFetch: 10,
-		Types: cfg.
-			MessageTypes().
-			Produced.
-			FilterByRole(message.EventRole),
+		Types: sets.NewFromKeys(
+			cfg.
+				MessageTypes().
+				Produced(message.EventKind),
+		),
 	}
 }
 
@@ -169,10 +172,11 @@ func (e *Engine) newCommandExecutor(
 				Key:  cfg.Identity().Key,
 			},
 			Marshaler: e.opts.Marshaler,
-			Produced: cfg.
-				MessageTypes().
-				Consumed.
-				FilterByRole(message.CommandRole),
+			Produced: sets.NewFromKeys(
+				cfg.
+					MessageTypes().
+					Consumed(message.EventKind),
+			),
 		},
 	}
 }
@@ -188,7 +192,7 @@ func (e *Engine) newEntryPoint(
 	hf := &handlerFactory{
 		opts:        e.opts,
 		queue:       q,
-		queueEvents: message.TypeSet{},
+		queueEvents: sets.New[message.Type](),
 		aggregateLoader: &aggregate.Loader{
 			AggregateRepository: ds,
 			EventRepository:     ds,
@@ -221,7 +225,7 @@ func (e *Engine) newEntryPoint(
 type handlerFactory struct {
 	opts            *engineOptions
 	queue           *queue.Queue
-	queueEvents     message.TypeSet
+	queueEvents     *sets.Set[message.Type]
 	aggregateLoader *aggregate.Loader
 	processLoader   *process.Loader
 	engineLogger    logging.Logger
@@ -278,8 +282,8 @@ func (f *handlerFactory) VisitRichAggregate(_ context.Context, cfg configkit.Ric
 		Packer: &parcel.Packer{
 			Application: f.app,
 			Marshaler:   f.opts.Marshaler,
-			Produced:    cfg.MessageTypes().Produced,
-			Consumed:    cfg.MessageTypes().Consumed,
+			Produced:    sets.NewFromKeys(cfg.MessageTypes().Produced()),
+			Consumed:    sets.NewFromKeys(cfg.MessageTypes().Consumed()),
 		},
 		LoadTimeout: f.opts.MessageTimeout,
 		Logger:      f.appLogger,
@@ -311,8 +315,8 @@ func (f *handlerFactory) VisitRichProcess(_ context.Context, cfg configkit.RichP
 		Packer: &parcel.Packer{
 			Application: f.app,
 			Marshaler:   f.opts.Marshaler,
-			Produced:    cfg.MessageTypes().Produced,
-			Consumed:    cfg.MessageTypes().Consumed,
+			Produced:    sets.NewFromKeys(cfg.MessageTypes().Produced()),
+			Consumed:    sets.NewFromKeys(cfg.MessageTypes().Consumed()),
 		},
 		LoadTimeout: f.opts.MessageTimeout,
 		Logger:      f.appLogger,
@@ -332,8 +336,8 @@ func (f *handlerFactory) VisitRichIntegration(_ context.Context, cfg configkit.R
 		Packer: &parcel.Packer{
 			Application: f.app,
 			Marshaler:   f.opts.Marshaler,
-			Produced:    cfg.MessageTypes().Produced,
-			Consumed:    cfg.MessageTypes().Consumed,
+			Produced:    sets.NewFromKeys(cfg.MessageTypes().Produced()),
+			Consumed:    sets.NewFromKeys(cfg.MessageTypes().Consumed()),
 		},
 		Logger: f.appLogger,
 	})
@@ -346,10 +350,10 @@ func (f *handlerFactory) VisitRichProjection(context.Context, configkit.RichProj
 }
 
 func (f *handlerFactory) addRoutes(cfg configkit.RichHandler, h handler.Handler) {
-	for mt, r := range cfg.MessageTypes().Consumed {
+	for mt := range cfg.MessageTypes().Consumed() {
 		f.handler[mt] = append(f.handler[mt], h)
 
-		if r == message.EventRole {
+		if mt.Kind() == message.EventKind {
 			f.queueEvents.Add(mt)
 		}
 
@@ -360,7 +364,7 @@ func (f *handlerFactory) addRoutes(cfg configkit.RichHandler, h handler.Handler)
 			cfg.Identity().Name,
 			f.app.Name,
 			mt,
-			r,
+			mt.Kind(),
 		)
 	}
 }

engineoption.go

@@ -193,7 +193,7 @@ func WithProjectionCompactTimeout(d time.Duration) EngineOption {
 func NewDefaultMarshaler(configs []configkit.RichApplication) marshaler.Marshaler {
 	var types []reflect.Type
 	for _, cfg := range configs {
-		for t := range cfg.MessageTypes().All() {
+		for t := range cfg.MessageTypes() {
 			types = append(types, t.ReflectType())
 		}
 

engineoption_test.go

@@ -3,6 +3,7 @@ package verity
 import (
 	"time"
 
+	//revive:disable:dot-imports
 	"github.com/dogmatiq/configkit"
 	"github.com/dogmatiq/dodeca/logging"
 	"github.com/dogmatiq/dogma"

eventstream.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/dogmatiq/configkit"
 	"github.com/dogmatiq/configkit/message"
+	"github.com/dogmatiq/enginekit/collections/sets"
 	"github.com/dogmatiq/interopspec/envelopespec"
 	"github.com/dogmatiq/verity/eventstream"
 	"github.com/dogmatiq/verity/handler/projection"
@@ -65,18 +66,18 @@ func (e *Engine) runStreamConsumerForQueue(
 	s eventstream.Stream,
 	a *app,
 ) error {
-	events := message.TypeSet{}
+	var events *sets.Set[message.Type]
 
 	// Find all events that are consumed by processes. Events consumed by
 	// projections are not included as each projection handler has its own
 	// consumer.
 	for _, h := range a.Config.RichHandlers().Processes() {
-		h.MessageTypes().Consumed.RangeByRole(
-			message.EventRole,
-			func(t message.Type) bool {
-				events.Add(t)
-				return true
-			},
+		events = events.Union(
+			sets.NewFromKeys(
+				h.
+					MessageTypes().
+					Consumed(message.EventKind),
+			),
 		)
 	}
 
@@ -118,8 +119,12 @@ func (e *Engine) runStreamConsumerForProjection(
 	h configkit.RichProjection,
 ) error {
 	c := &eventstream.Consumer{
-		Stream:     s,
-		EventTypes: h.MessageTypes().Consumed,
+		Stream: s,
+		EventTypes: sets.NewFromKeys(
+			h.
+				MessageTypes().
+				Consumed(),
+		),
 		Handler: &projection.StreamAdaptor{
 			Identity: &envelopespec.Identity{
 				Name: h.Identity().Name,

eventstream/consumer.go

@@ -7,6 +7,7 @@ import (
 	"github.com/dogmatiq/configkit"
 	"github.com/dogmatiq/configkit/message"
 	"github.com/dogmatiq/dodeca/logging"
+	"github.com/dogmatiq/enginekit/collections/sets"
 	"github.com/dogmatiq/linger"
 	"github.com/dogmatiq/linger/backoff"
 	"golang.org/x/sync/semaphore"
@@ -33,7 +34,7 @@ type Consumer struct {
 	Stream Stream
 
 	// EventTypes is the set of event types that the handler consumes.
-	EventTypes message.TypeCollection
+	EventTypes *sets.Set[message.Type]
 
 	// Handler is the target for the events from the stream.
 	Handler Handler
@@ -97,9 +98,9 @@ func (c *Consumer) consume(ctx context.Context) error {
 		return err
 	}
 
-	relevant := message.IntersectionT(c.EventTypes, produced)
+	relevant := c.EventTypes.Intersection(produced)
 
-	if len(relevant) == 0 {
+	if relevant.Len() == 0 {
 		logging.Debug(
 			c.Logger,
 			"stream does not produce any relevant event types",
@@ -108,7 +109,7 @@ func (c *Consumer) consume(ctx context.Context) error {
 		return nil
 	}
 
-	for t := range relevant {
+	for t := range relevant.All() {
 		logging.Debug(
 			c.Logger,
 			"consuming '%s' events",
@@ -132,7 +133,7 @@ func (c *Consumer) consume(ctx context.Context) error {
 // open opens a stream cursor based on the offset given by the handler.
 func (c *Consumer) open(
 	ctx context.Context,
-	types message.TypeSet,
+	types *sets.Set[message.Type],
 ) (Cursor, error) {
 	var err error
 	c.offset, err = c.Handler.NextOffset(ctx, c.Stream.Application())
@@ -143,7 +144,7 @@ func (c *Consumer) open(
 	logging.Debug(
 		c.Logger,
 		"consuming %d event type(s), beginning at offset %d",
-		len(types),
+		types.Len(),
 		c.offset,
 	)
 

eventstream/consumer_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/dogmatiq/configkit"
 	"github.com/dogmatiq/configkit/message"
 	"github.com/dogmatiq/dodeca/logging"
+	"github.com/dogmatiq/enginekit/collections/sets"
 	. "github.com/dogmatiq/enginekit/enginetest/stubs"
 	"github.com/dogmatiq/linger/backoff"
 	. "github.com/dogmatiq/verity/eventstream"
@@ -66,7 +67,7 @@ var _ = Describe("type Consumer", func() {
 
 		mstream = &memorystream.Stream{
 			App: configkit.MustNewIdentity("<app-name>", DefaultAppKey),
-			Types: message.NewTypeSet(
+			Types: sets.New(
 				message.TypeFor[EventStub[TypeA]](),
 				message.TypeFor[EventStub[TypeB]](),
 			),
@@ -89,7 +90,7 @@ var _ = Describe("type Consumer", func() {
 
 		consumer = &Consumer{
 			Stream: stream,
-			EventTypes: message.NewTypeSet(
+			EventTypes: sets.New(
 				message.TypeFor[EventStub[TypeA]](),
 			),
 			Handler:         eshandler,
@@ -130,8 +131,8 @@ var _ = Describe("type Consumer", func() {
 		It("returns if the stream does not produce any relevant events", func() {
 			stream.EventTypesFunc = func(
 				context.Context,
-			) (message.TypeCollection, error) {
-				return message.NewTypeSet(
+			) (*sets.Set[message.Type], error) {
+				return sets.New(
 					message.TypeFor[EventStub[TypeC]](),
 				), nil
 			}
@@ -144,7 +145,7 @@ var _ = Describe("type Consumer", func() {
 			stream.OpenFunc = func(
 				context.Context,
 				uint64,
-				message.TypeCollection,
+				*sets.Set[message.Type],
 			) (Cursor, error) {
 				stream.OpenFunc = nil
 
@@ -168,7 +169,7 @@ var _ = Describe("type Consumer", func() {
 		It("restarts the consumer if querying the stream's event types returns an error", func() {
 			stream.EventTypesFunc = func(
 				context.Context,
-			) (message.TypeCollection, error) {
+			) (*sets.Set[message.Type], error) {
 				stream.EventTypesFunc = nil
 
 				eshandler.HandleEventFunc = func(
@@ -213,7 +214,7 @@ var _ = Describe("type Consumer", func() {
 
 		It("restarts the consumer if the event offset is earlier than the consumed offset", func() {
 			// Ensure the consumer tries to consume all event types.
-			consumer.EventTypes = message.NewTypeSet(
+			consumer.EventTypes = sets.New(
 				message.TypeFor[EventStub[TypeA]](),
 				message.TypeFor[EventStub[TypeB]](),
 			)
@@ -223,7 +224,7 @@ var _ = Describe("type Consumer", func() {
 			stream.OpenFunc = func(
 				ctx context.Context,
 				offset uint64,
-				types message.TypeCollection,
+				types *sets.Set[message.Type],
 			) (Cursor, error) {
 				// Reset the stream to behave normally again on the next
 				// attempt at opening a cursor.