feat: batch and stream hooks

[alepane21]

Summary by CodeRabbit

• New Features

  • Added support for custom event hooks during stream publishing and batch processing, enabling advanced filtering and modification of events in real time.
  • Introduced new modules and utilities for enhanced Kafka and GraphQL subscription testing, including configurable batch and publish hooks.

• Refactor

  • Unified and extended hook management for pub/sub providers, allowing more flexible and centralized event handling.
  • Improved configuration and provider management for Kafka, NATS, and Redis integrations, simplifying adapter usage and setup.
  • Changed internal provider collections from slices to maps for efficient access and management.

• Bug Fixes

  • Enhanced error handling and type safety for event publishing and subscription flows across all supported providers.
  • Added error propagation and subscription closure handling in event update flows.
  • Improved logging and error messages for pub/sub operations.

• Tests

  • Expanded and updated test coverage for batch and publish hooks, and improved test utilities for Kafka, NATS, and Redis.
  • Replaced local Kafka and Redis test helpers with centralized utility functions for consistency.
  • Updated mocks to align with new interfaces and method signatures, removing deprecated mocks.

• Chores

  • Removed unused mock implementations and streamlined mock usage across the codebase.
  • Centralized Kafka and Redis event helper functions into a shared package for reuse.

How to use the new hook?

The hooks implemented in this PR are StreamBatchEventHook and StreamPublishEventHook.
StreamBatchEventHook is defined here:

cosmo/router/core/subscriptions_modules.go

Line 188 in 3d829f8

type StreamBatchEventHook interface {

StreamPublishEventHook is defined here:

cosmo/router/core/subscriptions_modules.go

Line 201 in 3d829f8

type StreamPublishEventHook interface {

Demo StreamBatchEventHook

Subscription

On the demo graph, we will create a module that filters out messages based on the kafka header

employeeUpdatedMyKafka(employeeID: $employeeID)

Module code that implement the new hook

const myModuleID = "streamBatchModule"

type StreamBatchModule struct {}

func (m *StreamBatchModule) OnStreamEvents(ctx core.StreamBatchEventHookContext, events []datasource.StreamEvent) ([]datasource.StreamEvent, error) {
        if ctx.SubscriptionEventConfiguration().RootFieldName() != "employeeUpdatedMyKafka" {
	        return events, nil
        }
        newEvents := []datasource.StreamEvent{}
        for _, event := range events {
	        evt, ok := event.(*kafka.Event)
	        if !ok {
		        continue
	        }
	        if string(evt.Headers["destination"]) != "external-clients" {
		        continue
	        }
	        newEvents = append(newEvents, event)
        }
        
        return newEvents, nil
}

func (m *StreamBatchModule) Module() core.ModuleInfo {
	return core.ModuleInfo{
		// This is the ID of your module, it must be unique
		ID: myModuleID,
		// The priority of your module, lower numbers are executed first
		Priority: 1,
		New: func() core.Module {
			return &StreamBatchModule{}
		},
	}
}

Demo StreamPublishEventHook

Mutation

On the demo graph, we will create a module that will add the $employeeId value in a kafka header.

updateEmployeeMyKafka(employeeID: $employeeId, update: $update)

Module code that implement the new hook

const myModuleID = "publishModule"

type PublishModule struct {}

func (m *PublishModule) OnPublishEvents(ctx core.StreamPublishEventHookContext, events []datasource.StreamEvent) ([]datasource.StreamEvent, error) {
        if ctx.PublishEventConfiguration().RootFieldName() != "updateEmployeeMyKafka" {
	        return events, nil
        }
        
        employeeID := ctx.RequestContext().Operation().Variables().GetInt("employeeID")
        
        newEvents := []datasource.StreamEvent{}
        for _, event := range events {
	        evt, ok := event.(*kafka.Event)
	        if !ok {
		        continue
	        }
	        if evt.Headers == nil {
		        evt.Headers = map[string][]byte{}
	        }
	        evt.Headers["x-employee-id"] = []byte(strconv.Itoa(employeeID))
	        newEvents = append(newEvents, event)
        }
        return newEvents, nil
}

func (m *PublishModule) Module() core.ModuleInfo {
	return core.ModuleInfo{
		// This is the ID of your module, it must be unique
		ID: myModuleID,
		// The priority of your module, lower numbers are executed first
		Priority: 1,
		New: func() core.Module {
			return &PublishModule{}
		},
	}
}

Checklist

• I have discussed my proposed changes in an issue and have received approval to proceed.
• I have followed the coding standards of the project.
• Tests or benchmarks have been added or updated.
• Documentation has been updated on https://github.com/wundergraph/cosmo-docs.
• I have read the Contributors Guide.

[dkorittki]

I have reviewed all non-test code first because of the size of this PR. I think it makes sense to discuss this first, then tests might have to be adjusted second.

I have a general question regarding the current method of pushing a message batch to brokers. In this approach we abort on first problems, with parts of messages being sent to the broker while others not. Did this change in the way EDFS did it before? Generally I would think batch events should be sent transactionally but I am lacking the historical context here. Just wanted to bring this topic up for discussion.

[dkorittki]

I believe that in the ADR the StreamHookError type is not whats in the code:

In ADR

// StreamHookError is used to customize the error messages and the behavior
type StreamHookError struct {
    HttpError core.HttpError
    CloseSubscription bool
}

In Code

type StreamHookError struct {
	err        error
	message    string
	statusCode int
	code       string
}

File: router/core/subscriptions_modules.go

[dkorittki]

Also check the response body here? Like you did on the nats test