feat: batch and stream hooks

adr/cosmo-streams-v1.md

@@ -21,24 +21,18 @@ The following interfaces will extend the existing logic in the custom modules.
 These provide additional control over subscriptions by providing hooks, which are invoked during specific events.
 
 - `SubscriptionOnStartHandler`: Called once at subscription start.
-- `StreamBatchEventHook`: Called each time a batch of events is received from the provider.
-- `StreamPublishEventHook`: Called each time a batch of events is going to be sent to the provider.
+- `StreamReceiveEventHandler`: Triggered for each client/subscription when a batch of events is received from the provider, prior to delivery.
+- `StreamPublishEventHandler`: Called each time a batch of events is going to be sent to the provider.
 
 ```go
 // STRUCTURES TO BE ADDED TO PUBSUB PACKAGE
 type ProviderType string
 const (
-    ProviderTypeNats ProviderType = "nats"
+    ProviderTypeNats  ProviderType = "nats"
     ProviderTypeKafka ProviderType = "kafka"
     ProviderTypeRedis ProviderType = "redis"
 }
 
-// StreamHookError is used to customize the error messages and the behavior
-type StreamHookError struct {
-    HttpError core.HttpError
-    CloseSubscription bool
-}
-
 // OperationContext already exists, we just have to add the Variables() method
 type OperationContext interface {
     Name() string
@@ -48,8 +42,9 @@ type OperationContext interface {
 
 // each provider will have its own event type with custom fields
 // the StreamEvent interface is used to allow the hooks system to be provider-agnostic
-// there could be common fields in future, but for now we don't need them
-type StreamEvent interface {}
+type StreamEvent interface {
+    GetData() []byte
+}
 
 // SubscriptionEventConfiguration is the common interface for the subscription event configuration
 type SubscriptionEventConfiguration interface {
@@ -67,7 +62,7 @@ type PublishEventConfiguration interface {
     RootFieldName() string
 }
 
-type SubscriptionOnStartHookContext interface {
+type SubscriptionOnStartHandlerContext interface {
     // Request is the original request received by the router.
     Request() *http.Request
     // Logger is the logger for the request
@@ -85,34 +80,48 @@ type SubscriptionOnStartHookContext interface {
 
 type SubscriptionOnStartHandler interface {
     // OnSubscriptionOnStart is called once at subscription start
-    // Returning an error will result in a GraphQL error being returned to the client, could be customized returning a StreamHookError.
-    SubscriptionOnStart(ctx SubscriptionOnStartHookContext) error
+    // Returning an error will result in a GraphQL error being returned to the client
+    SubscriptionOnStart(ctx SubscriptionOnStartHandlerContext) error
 }
 
-type StreamBatchEventHookContext interface {
-    // the request context
-    RequestContext() RequestContext
-    // the subscription event configuration
+type StreamReceiveEventHandlerContext interface {
+    // Request is the initial client request that started the subscription
+    Request() *http.Request
+    // Logger is the logger for the request
+    Logger() *zap.Logger
+    // Operation is the GraphQL operation
+    Operation() OperationContext
+    // Authentication is the authentication for the request
+    Authentication() authentication.Authentication
+    // SubscriptionEventConfiguration is the subscription event configuration
     SubscriptionEventConfiguration() SubscriptionEventConfiguration
 }
 
-type StreamBatchEventHook interface {
-    // OnStreamEvents is called each time a batch of events is received from the provider
-    // Returning an error will result in a GraphQL error being returned to the client, could be customized returning a StreamHookError.
-    OnStreamEvents(ctx StreamBatchEventHookContext, events []StreamEvent) ([]StreamEvent, error)
+type StreamReceiveEventHandler interface {
+    // OnReceiveEvents is called each time a batch of events is received from the provider before delivering them to the client
+    // So for a single batch of events received from the provider, this hook will be called one time for each active subscription.
+    // It is important to optimize the logic inside this hook to avoid performance issues.
+    // Returning an error will result in a GraphQL error being returned to the client
+    OnReceiveEvents(ctx StreamReceiveEventHandlerContext, events []StreamEvent) ([]StreamEvent, error)
 }
 
-type StreamPublishEventHookContext interface {
-    // the request context
-    RequestContext() RequestContext
-    // the publish event configuration
+type StreamPublishEventHandlerContext interface {
+    // Request is the original request received by the router.
+    Request() *http.Request
+    // Logger is the logger for the request
+    Logger() *zap.Logger
+    // Operation is the GraphQL operation
+    Operation() OperationContext
+    // Authentication is the authentication for the request
+    Authentication() authentication.Authentication
+    // PublishEventConfiguration is the publish event configuration
     PublishEventConfiguration() PublishEventConfiguration
 }
 
-type StreamPublishEventHook interface {
+type StreamPublishEventHandler interface {
     // OnPublishEvents is called each time a batch of events is going to be sent to the provider
-    // Returning an error will result in a GraphQL error being returned to the client, could be customized returning a StreamHookError.
-    OnPublishEvents(ctx StreamPublishEventHookContext, events []StreamEvent) ([]StreamEvent, error)
+    // Returning an error will result in an error being returned and the client will see the mutation failing
+    OnPublishEvents(ctx StreamPublishEventHandlerContext, events []StreamEvent) ([]StreamEvent, error)
 }
 ```
 
@@ -154,7 +163,7 @@ type Employee @key(fields: "id", resolvable: false) {
   id: Int! @external
 }
 ```
-After publishing the schema, the developer will need to add the module to the cosmo streams engine.
+After publishing the schema, the developer will need to add the module to the cosmo router.
 
 ### 2. Write the custom module
 
@@ -177,39 +186,38 @@ func init() {
 
 type MyModule struct {}
 
-func (m *MyModule) OnStreamEvents(ctx StreamBatchEventHookContext, events []core.StreamEvent) ([]core.StreamEvent, error) {
+func (m *MyModule) OnReceiveEvents(ctx StreamReceiveEventHandlerContext, events []core.StreamEvent) ([]core.StreamEvent, error) {
     // check if the provider is nats
-    if ctx.StreamContext().ProviderType() != pubsub.ProviderTypeNats {
+    if ctx.SubscriptionEventConfiguration().ProviderType() != pubsub.ProviderTypeNats {
         return events, nil
     }
 
     // check if the provider id is the one expected by the module
-    if ctx.StreamContext().ProviderID() != "my-nats" {
+    if ctx.SubscriptionEventConfiguration().ProviderID() != "my-nats" {
         return events, nil
     }
 
-    // check if the subject is the one expected by the module
-    natsConfig := ctx.SubscriptionEventConfiguration().(*nats.SubscriptionEventConfiguration)
-    if natsConfig.Subjects[0] != "employeeUpdates" {
-        return events, nil
-    }
+	// check if the subscription is the one expected by the module
+	if ctx.SubscriptionEventConfiguration().RootFieldName() != "employeeUpdates" {
+		return events, nil
+	}
+
+	newEvents := make([]core.StreamEvent, 0, len(events))
 
     // check if the client is authenticated
-    if ctx.RequestContext().Authentication() == nil {
+    if ctx.Authentication() == nil {
         // if the client is not authenticated, return no events
-        return events, nil
+        return newEvents, nil
     }
 
     // check if the client is allowed to subscribe to the stream
-    clientAllowedEntitiesIds, found := ctx.RequestContext().Authentication().Claims()["allowedEntitiesIds"]
+    clientAllowedEntitiesIds, found := ctx.Authentication().Claims()["allowedEntitiesIds"]
     if !found {
-        return events, fmt.Errorf("client is not allowed to subscribe to the stream")
+        return newEvents, fmt.Errorf("client is not allowed to subscribe to the stream")
     }
 
-    newEvents := make([]core.StreamEvent, 0, len(events))
-
     for _, evt := range events {
-        natsEvent, ok := evt.(*nats.NatsEvent);
+        natsEvent, ok := evt.(*nats.NatsEvent)
         if !ok {
             newEvents = append(newEvents, evt)
             continue
@@ -266,7 +274,7 @@ func (m *MyModule) Module() core.ModuleInfo {
 
 // Interface guards
 var (
-    _ core.StreamBatchEventHook = (*MyModule)(nil)
+    _ core.StreamReceiveEventHandler = (*MyModule)(nil)
 )
 ```
 
@@ -321,7 +329,7 @@ func init() {
 
 type MyModule struct {}
 
-func (m *MyModule) SubscriptionOnStart(ctx SubscriptionOnStartHookContext) error {
+func (m *MyModule) SubscriptionOnStart(ctx SubscriptionOnStartHandlerContext) error {
     // check if the provider is nats
     if ctx.SubscriptionEventConfiguration().ProviderType() != pubsub.ProviderTypeNats {
         return nil
@@ -332,32 +340,27 @@ func (m *MyModule) SubscriptionOnStart(ctx SubscriptionOnStartHookContext) error
         return nil
     }
 
-    // check if the subject is the one expected by the module
-    natsConfig := ctx.SubscriptionEventConfiguration().(*nats.SubscriptionEventConfiguration)
-    if natsConfig.Subjects[0] != "employeeUpdates" {
-        return nil
-    }
+	// check if the subscription is the one expected by the module
+	if ctx.SubscriptionEventConfiguration().RootFieldName() != "employeeUpdates" {
+		return nil
+	}
 
     // check if the client is authenticated
     if ctx.Authentication() == nil {
         // if the client is not authenticated, return an error
-        return &StreamHookError{
-            HttpError: core.HttpError{
-                Code: http.StatusUnauthorized,
-                Message: "client is not authenticated",
-            },
+        return &core.HttpError{
+            Code: http.StatusUnauthorized,
+            Message: "client is not authenticated",
             CloseSubscription: true,
         }
     }
 
     // check if the client is allowed to subscribe to the stream
     clientAllowedEntitiesIds, found := ctx.Authentication().Claims()["readEmployee"]
     if !found {
-        return &StreamHookError{
-            HttpError: core.HttpError{
-                Code: http.StatusForbidden,
-                Message: "client is not allowed to read employees",
-            },
+        return &core.HttpError{
+            Code: http.StatusForbidden,
+            Message: "client is not allowed to read employees",
             CloseSubscription: true,
         }
     }
@@ -405,4 +408,4 @@ We could also generate the AsyncAPI specification from the schema and the events
 
 ## Generate hooks from AsyncAPI specifications
 
-Building on the AsyncAPI integration, we could allow the user to define their streams using AsyncAPI and generate fully typesafe hooks with all events structures generated from the AsyncAPI specification.
+Building on the AsyncAPI integration, we could allow the user to define their streams using AsyncAPI and generate fully typesafe hooks with all events structures generated from the AsyncAPI specification.

router-tests/events/events_config_test.go

@@ -1,4 +1,4 @@
-package events
+package events_test
 
 import (
 	"testing"