Event-Driven Data Sync Pipeline

This project is a distributed system that demonstrates event-driven architecture using RabbitMQ, MassTransit, and .NET 10. Three independent microservices communicate exclusively through asynchronous messaging with no direct service-to-service calls anywhere in the system.

---

The Problem This Solves

In tightly coupled systems, services call each other directly. When one service is slow or unavailable it cascades and the entire system degrades. Adding new consumers requires modifying existing services, which means more risk and more coordination between teams.

This project demonstrates the alternative approach. Services publish facts about what happened. Any number of consumers react independently, at their own pace, without the publisher knowing or caring who is listening. The result is a system that is resilient, scalable, and easy to extend.

---

Architecture

┌─────────────────────┐
│   ProductService    │  REST API that creates and updates products.
│   Port: 5065        │  Publishes events via the transactional outbox.
│   DB: productdb     │
└────────┬────────────┘
         │
         │  ProductCreated / ProductUpdated
         │  via RabbitMQ fanout exchange
         │
         ├─────────────────────────────────────┐
         │                                     │
         ▼                                     ▼
┌─────────────────────┐             ┌─────────────────────┐
│   SearchService     │             │  ReportingService   │
│   Port: 5255        │             │  Port: 5094         │
│   DB: searchdb      │             │  DB: reportingdb    │
│                     │             │                     │
│  Maintains a search │             │  Records a full     │
│  index of products  │             │  audit trail of all │
│  and exposes a      │             │  product changes    │
│  search endpoint.   │             │  and exposes a      │
│                     │             │  history endpoint.  │
└─────────────────────┘             └─────────────────────┘

---

Key Technical Concepts Demonstrated

Transactional Outbox Pattern

ProductService saves messages to an outbox table in the same database transaction as the business data. A background process then publishes those messages to RabbitMQ. This guarantees that no messages are ever lost even if the application crashes between saving to the database and publishing to the broker.

At-Least-Once Delivery and Idempotency

RabbitMQ guarantees at-least-once delivery which means messages may arrive more than once. Each consumer handles duplicates safely. SearchService checks whether a product is already indexed before inserting. ReportingService uses the MassTransit MessageId as a deduplication key backed by a unique database constraint so that the same event can never be recorded twice.

Result Pattern

Handlers return a typed Result object instead of throwing exceptions for expected business failures such as not found or validation errors. Controllers read the result and translate it to the appropriate HTTP status code. Unexpected infrastructure failures bubble up to a global exception handler that logs the full error internally and returns a clean 500 to the client with no stack trace exposed.

Vertical Slice Architecture

Each feature is self-contained with its own request, response, validator, and handler living in the same folder. Controllers are deliberately thin and only handle HTTP concerns. All business logic lives in the handler where it can be tested in complete isolation without spinning up an HTTP layer.

---

Tech Stack

Technology	Purpose
.NET 10	Runtime
ASP.NET Core	REST APIs
RabbitMQ	Message broker
MassTransit 8	Messaging abstraction
PostgreSQL	Persistent storage
Entity Framework Core	ORM and migrations
FluentValidation	Input validation
xUnit	Test framework
NSubstitute	Mocking library
FluentAssertions	Test assertions

---

Prerequisites

• .NET 10 SDK
• Docker Desktop

---

Running Locally

Start the infrastructure first. RabbitMQ and PostgreSQL both run in Docker so there is nothing to install on your machine.

docker run -d --name rabbitmq \
  -p 5672:5672 -p 15672:15672 \
  rabbitmq:3-management

docker run -d --name postgres \
  -e POSTGRES_USER=admin \
  -e POSTGRES_PASSWORD=admin \
  -e POSTGRES_DB=productdb \
  -p 5435:5432 \
  postgres:latest

Then run each service in a separate terminal. Each service will automatically apply any pending database migrations on startup.

cd src/Services/ProductService && dotnet run
cd src/Services/SearchService && dotnet run
cd src/Services/ReportingService && dotnet run

To run the full test suite:

dotnet test

---

API Reference

ProductService runs on http://localhost:5065

Method	Endpoint	Description
POST	/products	Create a new product
PUT	/products/{id}	Update an existing product

To create a product send a POST request with this body:

{
    "name": "Wireless Headphones",
    "description": "Noise cancelling over-ear headphones",
    "price": 149.99,
    "category": "Electronics"
}

SearchService runs on http://localhost:5255

Method	Endpoint	Description
GET	/search	Search products

The search endpoint accepts these query parameters: query to search by name or description, category to filter by category, minPrice and maxPrice to filter by price range. All parameters are optional and can be combined freely.

GET /search?query=headphones&category=Electronics&maxPrice=200

ReportingService runs on http://localhost:5094

Method	Endpoint	Description
GET	/reports/products/{id}/history	Get the full change history for a product

---

RabbitMQ Dashboard

The RabbitMQ management dashboard is available at http://localhost:15672 using the username guest and password guest. This is where you can observe exchanges, queues, and message rates in real time.

---

Project Structure

EventDrivenSync/
├── src/
│   ├── Contracts/              shared message types consumed by all services
│   └── Services/
│       ├── ProductService/     REST API and outbox publisher
│       ├── SearchService/      event consumer and search index
│       └── ReportingService/   event consumer and audit trail
└── tests/
    ├── ProductService.Tests/
    ├── SearchService.Tests/
    └── ReportingService.Tests/

---

Test Coverage

ProductService   — 14 tests covering handler validation, persistence, and event publishing
SearchService    —  6 tests covering consumer idempotency and index updates
ReportingService —  6 tests covering consumer idempotency and changelog recording
Total            — 26 tests