انتقل إلى المحتوى الرئيسي

Event-Driven Architecture for Module Integration

Architectural Decision Record: Event-Driven Architecture for Module Integration​

1. Executive Summary​

This document formalizes the architectural decision to use event-driven architecture for integration between the Inventory Management Module and other modules in the ERP system, particularly the Finance Module. Integration occurs asynchronously via a message bus (RabbitMQ with MassTransit) using integration events, rather than through direct API calls or shared databases.

This approach provides:

  • Loose coupling between modules
  • Autonomous module operation
  • Reliable message delivery with automatic retry
  • Complete audit trail of integration events
  • Scalability and resilience
  • Support for eventual consistency

This decision aligns with microservices best practices and Domain-Driven Design principles for bounded context integration.

2. Context: Module Integration Requirements​

The Inventory Management Module must integrate with other ERP modules:

Primary Integration: Finance Module

  • Inventory transactions create financial journal entries
  • Purchase transactions → Accounts Payable
  • Sales transactions → Accounts Receivable and Revenue Recognition
  • Assembly transactions → Manufacturing cost transfers
  • Adjustment transactions → Inventory gains/losses

Future Integrations:

  • Sales Module: Order fulfillment and shipment notifications
  • Purchasing Module: Receipt notifications and purchase order updates
  • Manufacturing Module: Production order completion
  • Warehouse Management: Pick/pack operations

Integration Challenges:

  • Modules deployed independently
  • Different release cycles and versioning
  • Network partitions and temporary unavailability
  • Need for transactional consistency without distributed transactions
  • Audit requirements for cross-module operations

3. The Architectural Debate: Integration Approaches​

3.1. Direct API Calls (Rejected)​

Pattern:

// In Inventory Module
public async Task CreateSalesTransaction(CreateSalesCommand command)
{
    var transaction = /* create transaction */;
    await _inventoryRepo.AddAsync(transaction);
    await _unitOfWork.CommitAsync();

    // Direct HTTP call to Finance API
    var financeClient = _httpClientFactory.CreateClient("Finance");
    var journalRequest = MapToJournalRequest(transaction);
    await financeClient.PostAsJsonAsync("/api/journals", journalRequest);
}

Arguments For:

  • Immediate consistency
  • Simple request/response model
  • Direct error feedback

Arguments Against (Why Rejected):

  • Tight Coupling: Inventory module depends on Finance module availability
  • Cascading Failures: Finance downtime breaks Inventory operations
  • Transaction Complexity: Difficult to maintain consistency if Finance call fails after Inventory commit
  • Synchronous Blocking: Inventory operations wait for Finance processing
  • Scaling Issues: Finance becomes bottleneck for Inventory throughput

3.2. Shared Database (Rejected)​

Pattern:

// Both modules access same database
public async Task CreateSalesTransaction(CreateSalesCommand command)
{
    using var transaction = await _dbContext.Database.BeginTransactionAsync();

    // Write to inventory tables
    var inventoryTxn = /* create */;
    _dbContext.InventoryTransactions.Add(inventoryTxn);

    // Write to finance tables (same database)
    var journal = /* create from transaction */;
    _dbContext.LedgerJournals.Add(journal);

    await _dbContext.SaveChangesAsync();
    await transaction.CommitAsync();
}

Arguments For:

  • ACID transactions across modules
  • Immediate consistency
  • No integration infrastructure needed

Arguments Against (Why Rejected):

  • Violates Bounded Context: Modules share data model, breaking encapsulation
  • Deployment Coupling: Schema changes require coordinated deployments
  • Scaling Limitations: Cannot scale modules independently
  • Team Dependencies: Teams must coordinate database changes
  • Not Microservices-Ready: Incompatible with future service decomposition

3.3. Event-Driven with Message Bus (Accepted)​

Pattern:

// In Inventory Module - Publish integration event
public async Task CreateSalesTransaction(CreateSalesCommand command)
{
    var transaction = SalesTransaction.Create(/* params */);
    await _inventoryRepo.AddAsync(transaction);
    await _unitOfWork.CommitAsync();  // Domain events dispatched here

    // Integration event published automatically by domain event handler
}

// Domain Event Handler publishes Integration Event
public class SalesTransactionCreatedEventHandler : INotificationHandler<SalesTransactionCreatedEvent>
{
    private readonly IPublishEndpoint _publishEndpoint;

    public async Task Handle(SalesTransactionCreatedEvent notification, CancellationToken cancellationToken)
    {
        var integrationEvent = new SalesTransactionCreatedIntegrationEvent
        {
            TransactionId = notification.TransactionId,
            TransactionDate = notification.TransactionDate,
            CustomerId = notification.CustomerId,
            TotalAmount = notification.TotalAmount,
            TotalCost = notification.TotalCost,
            LineItems = notification.LineItems
        };

        await _publishEndpoint.Publish(integrationEvent, cancellationToken);
    }
}

// In Finance Module - Consume integration event
public class SalesTransactionCreatedConsumer : IConsumer<SalesTransactionCreatedIntegrationEvent>
{
    private readonly IRepository<LedgerJournal> _journalRepo;
    private readonly IUnitOfWork _unitOfWork;

    public async Task Consume(ConsumeContext<SalesTransactionCreatedIntegrationEvent> context)
    {
        var @event = context.Message;

        // Create journal entries for revenue and COGS
        var journal = LedgerJournal.CreateFromSalesTransaction(@event);

        await _journalRepo.AddAsync(journal);
        await _unitOfWork.CommitAsync();
    }
}

Arguments For (Why Accepted):

  • Loose Coupling: Modules don't depend on each other's availability
  • Autonomous Operation: Inventory continues working if Finance is down
  • Reliable Delivery: Message bus ensures events eventually delivered
  • Audit Trail: All integration events logged and traceable
  • Scalability: Modules scale independently
  • Resilience: Automatic retry on failures
  • Flexibility: New consumers can subscribe without affecting publishers
  • Eventual Consistency: Acceptable trade-off for business requirements

4. The Accepted Solution: Event-Driven Integration Architecture​

4.1. Core Components​

Domain Events (Internal)

  • Raised by aggregates during state changes
  • Handled within same bounded context
  • Dispatched after successful SaveChangesAsync
  • Not sent across module boundaries

Integration Events (External)

  • Published to message bus for cross-module communication
  • Triggered by domain event handlers
  • Contain serializable data (no domain objects)
  • Versioned for backwards compatibility

Message Bus (RabbitMQ + MassTransit)

  • Durable queues ensure message persistence
  • Automatic retry with exponential backoff
  • Dead letter queues for failed messages
  • Publish/Subscribe pattern for flexible routing

Event Consumers

  • Subscribe to integration events from other modules
  • Process events idempotently (safe to retry)
  • Update local module state reactively
  • Can publish new events in response

4.2. Architectural Diagram​

4.3. Event Flow Patterns​

Pattern 1: Inventory Transaction → Financial Entry

  1. User creates inventory transaction (Purchase, Sales, Assembly, Adjustment)
  2. Transaction saved to Inventory database
  3. Domain event raised: *TransactionCreatedEvent
  4. Domain event handler publishes integration event
  5. Finance module consumes integration event
  6. Finance creates journal entries in Finance database
  7. Finance acknowledges message

Pattern 2: Error Handling and Retry

  1. Integration event published to message bus
  2. Consumer attempts to process event
  3. Processing fails (validation error, database timeout, etc.)
  4. Message returned to queue (not acknowledged)
  5. MassTransit automatically retries with exponential backoff
  6. After max retries (e.g., 5), message moved to dead letter queue
  7. Monitoring alerts on dead letter messages
  8. Manual intervention to resolve and replay

Pattern 3: Idempotent Processing

public class SalesTransactionCreatedConsumer : IConsumer<SalesTransactionCreatedIntegrationEvent>
{
    public async Task Consume(ConsumeContext<SalesTransactionCreatedIntegrationEvent> context)
    {
        var @event = context.Message;

        // Check if already processed (idempotency)
        var existingJournal = await _journalRepo.FindByExternalReferenceAsync(
            "Inventory",
            @event.TransactionId.ToString()
        );

        if (existingJournal != null)
        {
            // Already processed, acknowledge without action
            _logger.LogInformation("Journal for transaction {TransactionId} already exists", @event.TransactionId);
            return;
        }

        // Process event
        var journal = LedgerJournal.CreateFromSalesTransaction(@event);
        journal.SetExternalReference("Inventory", @event.TransactionId.ToString());

        await _journalRepo.AddAsync(journal);
        await _unitOfWork.CommitAsync();
    }
}

5. Implementation Guidance​

5.1. Integration Event Contract Design​

Best Practices:

  • Include all necessary data for processing (avoid chatty lookups)
  • Use primitive types (strings, numbers, dates) not domain objects
  • Include correlation IDs for tracing
  • Version events for backwards compatibility
  • Keep events focused and cohesive

Example Integration Event:

namespace IM.Contract.IntegrationEvents.Transactions;

public class SalesTransactionCreatedIntegrationEvent
{
    public Guid TransactionId { get; set; }
    public DateTime TransactionDate { get; set; }
    public Guid CustomerId { get; set; }
    public string? SalesOrderNumber { get; set; }
    public string? InvoiceNumber { get; set; }

    public List<LineItemDto> LineItems { get; set; } = new();

    public decimal TotalSalesValue { get; set; }
    public decimal TotalCostValue { get; set; }

    public string CurrencyCode { get; set; } = "USD";
    public DateTime CreatedAt { get; set; } = DateTime.UtcNow;
}

public class LineItemDto
{
    public Guid ItemId { get; set; }
    public string ItemNumber { get; set; }
    public decimal Quantity { get; set; }
    public string UnitOfMeasure { get; set; }
    public decimal UnitSellingPrice { get; set; }
    public decimal UnitCost { get; set; }
    public decimal ExtendedSellingPrice { get; set; }
    public decimal ExtendedCost { get; set; }
}

5.2. MassTransit Configuration​

In Inventory Module (Program.cs or DI configuration):

services.AddMassTransit(x =>
{
    // Register consumers (if any inbound events)
    x.AddConsumers(Assembly.GetExecutingAssembly());

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("rabbitmq://localhost", h =>
        {
            h.Username("guest");
            h.Password("guest");
        });

        // Configure retry policy
        cfg.UseMessageRetry(r => r.Exponential(
            retryLimit: 5,
            minInterval: TimeSpan.FromSeconds(1),
            maxInterval: TimeSpan.FromMinutes(5),
            intervalDelta: TimeSpan.FromSeconds(10)
        ));

        // Configure dead letter queue
        cfg.UseDelayedRedelivery(r => r.Intervals(
            TimeSpan.FromMinutes(5),
            TimeSpan.FromMinutes(15),
            TimeSpan.FromMinutes(30)
        ));

        cfg.ConfigureEndpoints(context);
    });
});

5.3. Domain Event to Integration Event Bridge​

Pattern: Domain Event Handler Publishes Integration Event

public class SalesTransactionCreatedEventHandler
    : INotificationHandler<SalesTransactionCreatedEvent>
{
    private readonly IPublishEndpoint _publishEndpoint;
    private readonly IRepository<InventoryTransaction> _transactionRepo;

    public async Task Handle(SalesTransactionCreatedEvent notification, CancellationToken cancellationToken)
    {
        // Load full transaction with details
        var transaction = await _transactionRepo.GetByIdAsync(notification.TransactionId);

        if (transaction == null || transaction.Type != TransactionType.Sales)
            return;

        // Map to integration event
        var integrationEvent = new SalesTransactionCreatedIntegrationEvent
        {
            TransactionId = transaction.Id,
            TransactionDate = transaction.TransactionDate,
            CustomerId = transaction.SalesDetails!.CustomerId,
            SalesOrderNumber = transaction.SalesDetails.SalesOrderNumber,
            InvoiceNumber = transaction.SalesDetails.InvoiceNumber,
            LineItems = transaction.Lines.Select(l => new LineItemDto
            {
                ItemId = l.ItemId,
                ItemNumber = l.Item.ItemNumber,
                Quantity = l.Quantity,
                UnitOfMeasure = l.UnitOfMeasure.Symbol,
                UnitSellingPrice = l.UnitSellingPrice,
                UnitCost = l.UnitCost,
                ExtendedSellingPrice = l.Quantity * l.UnitSellingPrice,
                ExtendedCost = l.Quantity * l.UnitCost
            }).ToList(),
            TotalSalesValue = transaction.Lines.Sum(l => l.Quantity * l.UnitSellingPrice),
            TotalCostValue = transaction.Lines.Sum(l => l.Quantity * l.UnitCost),
            CreatedAt = DateTime.UtcNow
        };

        // Publish to message bus
        await _publishEndpoint.Publish(integrationEvent, cancellationToken);
    }
}

5.4. Consumer Implementation​

Base Consumer with Error Handling:

public abstract class BaseEventConsumer<TEvent> : IConsumer<TEvent>
    where TEvent : class
{
    protected readonly ILogger<BaseEventConsumer<TEvent>> Logger;

    protected BaseEventConsumer(ILogger<BaseEventConsumer<TEvent>> logger)
    {
        Logger = logger;
    }

    public async Task Consume(ConsumeContext<TEvent> context)
    {
        var @event = context.Message;
        var eventType = typeof(TEvent).Name;

        try
        {
            Logger.LogInformation("Processing {EventType}: {@Event}", eventType, @event);

            await ProcessEvent(@event, context.CancellationToken);

            Logger.LogInformation("Successfully processed {EventType}", eventType);
        }
        catch (Exception ex)
        {
            Logger.LogError(ex, "Error processing {EventType}: {@Event}", eventType, @event);
            throw;  // Rethrow to trigger retry
        }
    }

    protected abstract Task ProcessEvent(TEvent @event, CancellationToken cancellationToken);
}

Concrete Consumer:

public class SalesTransactionCreatedConsumer
    : BaseEventConsumer<SalesTransactionCreatedIntegrationEvent>
{
    private readonly IRepository<LedgerJournal> _journalRepo;
    private readonly IUnitOfWork _unitOfWork;

    protected override async Task ProcessEvent(
        SalesTransactionCreatedIntegrationEvent @event,
        CancellationToken cancellationToken)
    {
        // Idempotency check
        var exists = await _journalRepo.ExistsByExternalRefAsync(
            "Inventory", @event.TransactionId.ToString());

        if (exists) return;

        // Create journal entries
        var journal = LedgerJournal.CreateSalesJournal(
            journalDate: @event.TransactionDate,
            description: $"Sales Transaction {@event.InvoiceNumber}",
            externalRef: @event.TransactionId.ToString()
        );

        // Add revenue recognition entry
        journal.AddLine(
            accountId: /* AR Account */,
            debit: @event.TotalSalesValue,
            credit: 0,
            description: "Accounts Receivable"
        );

        journal.AddLine(
            accountId: /* Revenue Account */,
            debit: 0,
            credit: @event.TotalSalesValue,
            description: "Sales Revenue"
        );

        // Add COGS entry
        journal.AddLine(
            accountId: /* COGS Account */,
            debit: @event.TotalCostValue,
            credit: 0,
            description: "Cost of Goods Sold"
        );

        journal.AddLine(
            accountId: /* Inventory Asset Account */,
            debit: 0,
            credit: @event.TotalCostValue,
            description: "Inventory Reduction"
        );

        await _journalRepo.AddAsync(journal);
        await _unitOfWork.CommitAsync();
    }
}

6. Benefits and Trade-offs​

6.1. Benefits​

Loose Coupling:

  • Modules don't know about each other's implementation
  • Can be deployed and scaled independently
  • Easier to modify or replace modules

Resilience:

  • Inventory operations succeed even if Finance is down
  • Events queued and processed when Finance recovers
  • Automatic retry handles transient failures

Audit Trail:

  • All integration events logged
  • Complete history of cross-module interactions
  • Replay capability for troubleshooting

Scalability:

  • Modules scaled independently based on load
  • Message bus handles backpressure
  • Multiple consumers can process events in parallel

Flexibility:

  • New modules can subscribe to existing events
  • Publishers don't need to know about consumers
  • Easy to add new integration points

6.2. Trade-offs​

Eventual Consistency:

  • Financial entries created after inventory transactions
  • Small delay between Inventory and Finance updates
  • Mitigation: Acceptable for business (typically milliseconds to seconds), monitoring ensures timely processing

Increased Complexity:

  • Message bus infrastructure to maintain
  • Event schema versioning needed
  • More moving parts to monitor

Debugging Challenges:

  • Asynchronous flow harder to trace than synchronous calls
  • Mitigation: Correlation IDs, distributed tracing, centralized logging

Idempotency Required:

  • Consumers must handle duplicate events
  • Mitigation: Standard pattern established, use external reference tracking

7. Monitoring and Observability​

Key Metrics to Monitor:

  • Event publish rate by type
  • Event processing lag (time from publish to consume)
  • Consumer error rates
  • Dead letter queue depth
  • Message bus resource utilization

Alerting Thresholds:

  • Dead letter queue > 0 messages (immediate alert)
  • Processing lag > 5 minutes (warning)
  • Consumer error rate > 5% (warning)
  • Message bus CPU > 80% (warning)

Logging Strategy:

  • Log all event publishes with correlation ID
  • Log all consumer processing attempts
  • Log consumer errors with full event payload
  • Centralize logs in ELK or similar stack

Distributed Tracing:

  • Use OpenTelemetry or similar
  • Trace events from publish through consumption
  • Correlate with originating API requests
  • Visualize cross-module flows

8. Future Enhancements​

Event Store:

  • Persist all integration events for event sourcing
  • Enable event replay for analytics or recovery
  • Support temporal queries

Saga Pattern:

  • Coordinate complex multi-module workflows
  • Handle compensating transactions for rollbacks
  • Example: Order fulfillment across Sales, Inventory, Shipping

Event Versioning:

  • Support multiple versions of same event
  • Graceful migration during deployments
  • Backwards compatibility guarantees

Stream Processing:

  • Real-time analytics on event streams
  • Complex event processing (CEP)
  • Machine learning on event patterns

9. Final Decision​

The Event-Driven Architecture with Message Bus provides the optimal integration strategy for the Inventory Management Module and the broader ERP system. It enables loose coupling, autonomous operation, and scalability while maintaining reliability through automatic retry and dead letter handling. The trade-off of eventual consistency is acceptable given typical business requirements and the minimal delay in event processing.

All cross-module integration must use this event-driven approach. Direct API calls or shared database access between modules are prohibited unless explicitly justified and approved by the architecture review board.

Status: Accepted Implementation Start: October 2024 Applies To: All ERP modules Review Date: April 2025 (6-month retrospective)

  • ADR-SHARED-002: Soft Delete Strategy (planned)
  • ADR-SHARED-003: Audit Trail Requirements (planned)
  • Finance Module ADRs: Journal composition model, settlement records
  • Domain-Driven Design Patterns: Bounded contexts, aggregates, domain events