Add Post-Dapr-PubSub Reaction

[amansinghoriginal]

Description

Adding a PostDaprPubSub Reaction for Drasi, which is designed to forward change notifications to Dapr Pub/Sub topics.

Overview

The PostDaprPubSub reaction allows developers to integrate Drasi's real-time query outputs with other services via Dapr's publish-subscribe mechanism. It handles both data change events and control signals, offering configurable behavior per query.

Key Features

• Drasi to Dapr Pub/Sub: Forwards results from Drasi queries to specified Dapr pub/sub components and topics.
• Configurable Event Formats:
  • Packed: Sends the entire ChangeEvent or ControlEvent as a single message.
  • Unpacked (Default): Sends individual messages for each change or control signal, using the Drasi native JSON format.
• Handles Change and Control Events: Processes both data modifications and system control signals from Drasi. Control signal forwarding can be disabled per query.
• Per-Query Configuration: Each Drasi query can be independently configured with:
  • pubsubName: The Dapr pub/sub component name.
  • topicName: The Dapr topic to publish to.
  • packed: Boolean to enable packed event format.
  • maxFailureCount: Threshold for consecutive errors before a query is marked as failed.
  • skipControlSignals: Boolean to disable forwarding of control signals.
• Failure Tracking: Automatically tracks consecutive failures for each query. If maxFailureCount is exceeded, the query is marked as failed, and further processing for it is halted until reset.
• Startup Validation: Validates all query configurations upon initialization to ensure correctness.

Core Components Added

• Program.cs: Main application entry point; builds and starts the reaction.
• QueryConfig.cs: Defines the C# model for query-specific JSON configuration.
• ChangeHandler.cs: Implements IChangeEventHandler to process ChangeEvent objects from Drasi, format them, and publish to Dapr.
• ControlSignalHandler.cs: Implements IControlEventHandler to process ControlEvent objects and publish them to Dapr.
• Services/DrasiChangeFormatter.cs: Formats ChangeEvent data into the Drasi native structure for unpacked messages.
• Services/ChangeFormatterFactory.cs: Provides instances of IChangeFormatter.
• QueryFailureTracker.cs: Implements IQueryFailureTracker to monitor and manage query failures.
• Services/QueryConfigValidationService.cs: Validates all query configurations at startup.
• Services/DaprInitializationService.cs: Ensures the Dapr sidecar is available before the reaction fully starts.
• ErrorStateHandler.cs: Provides a mechanism to terminate the reaction in case of critical errors.
• Models/Unpacked.generated.cs: C# classes representing the structure of unpacked Drasi native event formats.

Testing

• Unit Tests (Drasi.Reactions.PostDaprPubSub.Tests/):
  A suite of xUnit tests covers the core logic, including:

  • ChangeHandlerTests
  • ControlSignalHandlerTests
  • QueryConfigTests (and its validation)
  • QueryFailureTrackerTests
  • DaprInitializationServiceTests
  • QueryConfigValidationServiceTests
  • ChangeFormatterTests (for Drasi native format)
  • ErrorStateHandlerTests

• End-to-End Tests:

  • Verifies that INSERT operations on the product table trigger the publication of change events.
  • Verifies that UPDATE operations on the product table trigger the publication of change events.
  • Packed Format: Confirms that events are published in the "Packed" format to the e2e-topic-packed topic, containing the relevant change data and query ID (product-updates-packed).
  • Unpacked Format: Confirms that events are published in the "Unpacked" format to the e2e-topic-unpacked topic, detailing individual changes with correct before and after states (or absence thereof for inserts) and the appropriate query ID (product-updates-unpacked).

• Manual Validation Environment (validation/):
  A self-contained validation setup is provided to test the reaction end-to-end within a local Kubernetes (k3d/Kind) environment. This includes:

  • A Makefile (validation/Makefile) to deploy PostgreSQL, Redis, Dapr components, the reaction itself, and two sample Python FastAPI subscriber applications.
  • Subscriber applications (validation/subscriber_alpha/ and validation/subscriber_beta/) that listen to Dapr topics and log received messages.
  • Kubernetes YAML manifests (validation/common/) for all necessary components, including Drasi Source, ContinuousQuery, and Reaction resources for a demo scenario.

Documentation

• README.md: Provides a detailed overview of the reaction, its features, configuration parameters, event formats, and examples.

Type of change

• This pull request fixes a bug in Drasi and has an approved issue (issue link required).
• This pull request adds or changes features of Drasi and has an approved issue (issue link required).
• This pull request is a minor refactor, code cleanup, test improvement, or other maintenance task and doesn't change the functionality of Drasi (issue link optional).

Fixes: #247

[Copilot]

Pull Request Overview

This PR introduces a new PostDaprPubSub Reaction that forwards Drasi query outputs—both change and control events—to specified Dapr Pub/Sub topics. Key changes include the implementation of core services (ChangeHandler, ControlSignalHandler, QueryConfig, QueryFailureTracker), comprehensive unit tests for these components, and accompanying Docker configurations and workflow updates.

Reviewed Changes

Copilot reviewed 48 out of 48 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
Services/ChangeHandler.cs	Implements processing of change events with support for both packed and unpacked formats.
Services/ChangeFormatterFactory.cs	Provides a factory method to retrieve the DrasiChangeFormatter instance.
QueryFailureTracker.cs	Offers thread-safe tracking and management of query failures.
QueryConfig.cs	Defines the JSON configuration model with validations for the reaction.
Program.cs	Sets up application dependency injection, startup sequence, and Dapr sidecar waiting logic.
ErrorStateHandler.cs	Provides a mechanism to terminate the reaction in critical error scenarios.
csproj file	Targets .NET 9.0 and includes necessary package references.
Tests/	Contains extensive unit tests validating service behavior, configuration validation, and Docker build scripts.
Dockerfile & Dockerfile.debug	Define multi-stage container builds for production and debugging environments.
.github/workflows/draft-release.yml	Updates the release workflow to include the new reaction.

[Copilot]

Pull Request Overview

Adds a new Post-Dapr-PubSub reaction that publishes Drasi change and control events to Dapr pub/sub topics, with full unit and end-to-end tests, Dockerfiles, and CI updates.

• Implements handlers, configuration model, validation, failure tracking, and initialization for the PostDaprPubSub reaction.
• Adds a comprehensive xUnit test suite for config validation, change/control handlers, Dapr sidecar initialization, and formatting logic.
• Introduces Dockerfiles for production/debug builds, e2e tests for Redis-backed pub/sub, and CI workflow entries.

Reviewed Changes

Copilot reviewed 54 out of 54 changed files in this pull request and generated 1 comment.

File	Description
reactions/dapr/post-pubsub/Drasi.Reactions.PostDaprPubSub.Tests/QueryConfigValidationServiceTests.cs	Tests for startup config validation service
reactions/dapr/post-pubsub/Dockerfile	Production Dockerfile for PostDaprPubSub reaction
reactions/dapr/post-pubsub/Drasi.Reactions.PostDaprPubSub.Tests/ErrorStateHandlerTests.cs	Placeholder test for error state handler
reactions/dapr/post-pubsub/Drasi.Reactions.PostDaprPubSub.Tests/DaprInitializationServiceTests.cs	Tests for Dapr sidecar initialization

Comments suppressed due to low confidence (4)

reactions/dapr/post-pubsub/Drasi.Reactions.PostDaprPubSub.Tests/QueryConfigValidationServiceTests.cs:1

• Add using System.Threading; (or qualify CancellationToken) so calls to CancellationToken.None compile correctly.

// Copyright 2024 The Drasi Authors.

reactions/dapr/post-pubsub/Drasi.Reactions.PostDaprPubSub.Tests/DaprInitializationServiceTests.cs:1

• Add using System.Threading; to enable CancellationToken references in tests.

// Copyright 2024 The Drasi Authors.

reactions/dapr/post-pubsub/Drasi.Reactions.PostDaprPubSub.Tests/ErrorStateHandlerTests.cs:24

• [nitpick] This placeholder test doesn't verify any behavior of ErrorStateHandler. Implement a real assertion or remove it to improve coverage and maintainability.

public void Terminate_CallsTerminateWithError()

reactions/dapr/post-pubsub/Drasi.Reactions.PostDaprPubSub.Tests/DaprInitializationServiceTests.cs:43

• [nitpick] Test name WaitForDaprSidecarAsync_Success_LogsInfoMessage implies log verification, but no assertion is made against the logger. Either assert ILogger.LogInformation was called or rename the test to reflect behavior.

[Fact]

[danielgerlag]

LGTM, we just need to add the config schema in the reaction provider definition.