DataVerse Business Events Architecture Strategy¶

// Configuration model classes
public class EventMappingConfiguration
{
    public string SchemaVersion { get; set; }
    public ConfigurationMetadata Metadata { get; set; }
    public GlobalConfiguration Global { get; set; }
    public Dictionary<string, EntityConfiguration> Entities { get; set; }
    public List<ConsumerGroupConfiguration> ConsumerGroups { get; set; }
    public ValidationConfiguration Validation { get; set; }
}

public class EntityConfiguration
{
    public string DisplayName { get; set; }
    public string PrimaryKey { get; set; }
    public EntitySettings Settings { get; set; }
    public Dictionary<string, List<EventConfiguration>> Events { get; set; }
}

public class EventConfiguration
{
    public string[] TriggerFields { get; set; }
    public string EventType { get; set; }
    public string Mode { get; set; }  // slim, fat, changes_only, tombstone
    public string[] Domains { get; set; }
    public bool BusinessCritical { get; set; }
    public List<EventCondition> Conditions { get; set; }
    public PayloadConfiguration Payload { get; set; }
    public Dictionary<string, ReferenceConfiguration> References { get; set; }
}

public class EventCondition
{
    public string Field { get; set; }
    public string Operator { get; set; }  // equals, changed_from_to, percentage_change_greater_than, etc.
    public object Value { get; set; }
    public object[] FromValues { get; set; }
    public object[] ToValues { get; set; }
    public double Threshold { get; set; }
}

// Configuration registry service
public class EventConfigurationRegistry : IEventConfigurationRegistry
{
    private readonly IConfiguration _configuration;
    private readonly ILogger<EventConfigurationRegistry> _logger;
    private readonly IMemoryCache _cache;
    private EventMappingConfiguration _eventMappings;
    private FileSystemWatcher _configWatcher;

    public EventConfigurationRegistry(IConfiguration configuration, ILogger<EventConfigurationRegistry> logger, IMemoryCache cache)
    {
        _configuration = configuration;
        _logger = logger;
        _cache = cache;

        LoadConfiguration();
        SetupHotReload();
    }

    public async Task<List<EventConfiguration>> GetEventConfigurationsAsync(
        string entityName, 
        string operation,
        Dictionary<string, FieldChange> changedFields)
    {
        var cacheKey = $"events:{entityName}:{operation}";

        if (_cache.TryGetValue(cacheKey, out List<EventConfiguration> cached))
        {
            return FilterByChangedFields(cached, changedFields);
        }

        if (!_eventMappings.Entities.TryGetValue(entityName, out var entityConfig))
        {
            _logger.LogWarning("No configuration found for entity {EntityName}", entityName);
            return new List<EventConfiguration>();
        }

        if (!entityConfig.Events.TryGetValue(operation.ToLower(), out var eventConfigs))
        {
            return new List<EventConfiguration>();
        }

        // Cache the base configurations
        _cache.Set(cacheKey, eventConfigs, TimeSpan.FromMinutes(30));

        return FilterByChangedFields(eventConfigs, changedFields);
    }

    private List<EventConfiguration> FilterByChangedFields(
        List<EventConfiguration> eventConfigs, 
        Dictionary<string, FieldChange> changedFields)
    {
        var matchingConfigs = new List<EventConfiguration>();

        foreach (var config in eventConfigs)
        {
            if (DoesConfigurationMatch(config, changedFields))
            {
                matchingConfigs.Add(config);
            }
        }

        return matchingConfigs;
    }

    private bool DoesConfigurationMatch(EventConfiguration config, Dictionary<string, FieldChange> changedFields)
    {
        // Check if any trigger field matches the changed fields
        foreach (var triggerField in config.TriggerFields)
        {
            if (triggerField == "*")
                return true; // Wildcard matches any change

            if (triggerField.EndsWith("*"))
            {
                var prefix = triggerField.TrimEnd('*');
                if (changedFields.Keys.Any(field => field.StartsWith(prefix)))
                    return true;
            }

            if (changedFields.ContainsKey(triggerField))
                return true;
        }

        return false;
    }

    private void LoadConfiguration()
    {
        try
        {
            var configPath = _configuration["EventMapping:ConfigurationPath"] ?? "event-mappings.yaml";
            var yamlContent = File.ReadAllText(configPath);

            var deserializer = new DeserializerBuilder()
                .WithNamingConvention(UnderscoredNamingConvention.Instance)
                .Build();

            _eventMappings = deserializer.Deserialize<EventMappingConfiguration>(yamlContent);

            ValidateConfiguration(_eventMappings);

            _logger.LogInformation("Event mapping configuration loaded successfully from {ConfigPath}", configPath);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Failed to load event mapping configuration");
            throw;
        }
    }

    private void SetupHotReload()
    {
        var configPath = _configuration["EventMapping:ConfigurationPath"] ?? "event-mappings.yaml";
        var directory = Path.GetDirectoryName(configPath);
        var fileName = Path.GetFileName(configPath);

        _configWatcher = new FileSystemWatcher(directory, fileName)
        {
            EnableRaisingEvents = true,
            NotifyFilter = NotifyFilters.LastWrite
        };

        _configWatcher.Changed += async (sender, e) =>
        {
            _logger.LogInformation("Configuration file changed, reloading...");

            // Debounce multiple file change events
            await Task.Delay(1000);

            try
            {
                LoadConfiguration();
                _cache.Clear(); // Clear cached configurations
                _logger.LogInformation("Configuration reloaded successfully");
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Failed to reload configuration");
            }
        };
    }

    private void ValidateConfiguration(EventMappingConfiguration config)
    {
        var validator = new EventConfigurationValidator();
        var results = validator.Validate(config);

        if (results.Any(r => r.Severity == ValidationSeverity.Error))
        {
            throw new InvalidOperationException($"Configuration validation failed: {string.Join(", ", results.Select(r => r.Message))}");
        }

        if (results.Any(r => r.Severity == ValidationSeverity.Warning))
        {
            _logger.LogWarning("Configuration validation warnings: {Warnings}", 
                string.Join(", ", results.Where(r => r.Severity == ValidationSeverity.Warning).Select(r => r.Message)));
        }
    }
}

public class ConfigurationMigrationService
{
    public async Task<EventMappingConfiguration> MigrateConfigurationAsync(
        EventMappingConfiguration oldConfig, 
        string targetVersion)
    {
        var migrations = new Dictionary<string, Func<EventMappingConfiguration, EventMappingConfiguration>>
        {
            ["1.0-to-1.1"] = MigrateV1ToV11,
            ["1.1-to-1.2"] = MigrateV11ToV12
        };

        var currentVersion = oldConfig.SchemaVersion;
        var config = oldConfig;

        while (currentVersion != targetVersion)
        {
            var migrationKey = $"{currentVersion}-to-{GetNextVersion(currentVersion)}";

            if (migrations.TryGetValue(migrationKey, out var migration))
            {
                config = migration(config);
                currentVersion = config.SchemaVersion;
            }
            else
            {
                throw new InvalidOperationException($"No migration path from {currentVersion} to {targetVersion}");
            }
        }

        return config;
    }

    private EventMappingConfiguration MigrateV1ToV11(EventMappingConfiguration config)
    {
        // Example: Add new required fields introduced in v1.1
        foreach (var entity in config.Entities.Values)
        {
            foreach (var events in entity.Events.Values)
            {
                foreach (var eventConfig in events)
                {
                    // Add default domain if not specified
                    if (eventConfig.Domains == null || !eventConfig.Domains.Any())
                    {
                        eventConfig.Domains = new[] { "default" };
                    }

                    // Add default business criticality
                    if (!eventConfig.BusinessCritical)
                    {
                        eventConfig.BusinessCritical = eventConfig.EventType.Contains("Created") || 
                                                      eventConfig.EventType.Contains("Deleted");
                    }
                }
            }
        }

        config.SchemaVersion = "1.1";
        return config;
    }
}

[Test]
public async Task Configuration_Should_Generate_Correct_Events_For_RequestShortlist_StatusChange()
{
    // Arrange
    var config = LoadTestConfiguration();
    var registry = new EventConfigurationRegistry(config, Mock.Of<ILogger>(), new MemoryCache(new MemoryCacheOptions()));

    var changedFields = new Dictionary<string, FieldChange>
    {
        ["rqm_salesstatus"] = new FieldChange
        {
            FieldName = "rqm_salesstatus",
            OldValue = 112020000, // Under Review
            NewValue = 112020001, // Shortlisted
            DataType = "OptionSetValue"
        }
    };

    // Act
    var eventConfigs = await registry.GetEventConfigurationsAsync("rqm_requestshortlist", "update", changedFields);

    // Assert
    Assert.Single(eventConfigs);
    Assert.Equal("RequestShortlist.StatusChanged", eventConfigs[0].EventType);
    Assert.Equal("slim", eventConfigs[0].Mode);
    Assert.Contains("matching", eventConfigs[0].Domains);
    Assert.Contains("notifications", eventConfigs[0].Domains);
    Assert.True(eventConfigs[0].BusinessCritical);
}

[Test]
public void Configuration_Should_Validate_Required_Fields()
{
    // Arrange
    var invalidConfig = new EventMappingConfiguration
    {
        Entities = new Dictionary<string, EntityConfiguration>
        {
            ["test_entity"] = new EntityConfiguration
            {
                Events = new Dictionary<string, List<EventConfiguration>>
                {
                    ["create"] = new List<EventConfiguration>
                    {
                        new EventConfiguration
                        {
                            // Missing required EventType
                            Mode = "fat",
                            Domains = new[] { "test" }
                        }
                    }
                }
            }
        }
    };

    // Act & Assert
    var validator = new EventConfigurationValidator();
    var results = validator.Validate(invalidConfig);

    Assert.True(results.Any(r => r.Severity == ValidationSeverity.Error));
    Assert.Contains(results, r => r.Message.Contains("EventType is required"));
}

public class BusinessEvent
{
    // Standard fields (always present)
    public Guid EventId { get; set; } = Guid.NewGuid();
    public string EventType { get; set; }  // "RequestShortlist.StatusChanged"
    public Guid EntityId { get; set; }
    public string EntityType { get; set; }  // "rqm_requestshortlist"
    public DateTime Timestamp { get; set; }
    public Guid CorrelationId { get; set; }
    public Guid RequestId { get; set; }  // From DataVerse context
    public string Source { get; set; } = "DataVerse.BusinessEventShim";

    // Event routing and processing hints
    public EventMetadata Metadata { get; set; }

    // Flexible payload based on event mode
    public JObject Data { get; set; }

    // Change tracking (for update events)
    public List<FieldChange> Changes { get; set; }

    // Entity references (resolved based on configuration)  
    public Dictionary<string, EntityReference> References { get; set; }
}

public class EventMetadata
{
    public string[] Domains { get; set; }  // ["matching", "crm"] 
    public bool IsBusinessCritical { get; set; }
    public bool IncludeInMetrics { get; set; } = true;
    public string Mode { get; set; }  // "slim", "fat", "changes_only", "tombstone"
    public int SchemaVersion { get; set; } = 1;
    public DateTime CreatedAt { get; set; }
    public string CreatedBy { get; set; }  // DataVerse user ID
}

public class FieldChange
{
    public string FieldName { get; set; }
    public object OldValue { get; set; }
    public object NewValue { get; set; }
    public string DataType { get; set; }  // From DataVerse metadata
    public string DisplayName { get; set; }  // User-friendly field name
}

public class EntityReference
{
    public Guid Id { get; set; }
    public string LogicalName { get; set; }
    public string DisplayName { get; set; }  // Resolved name
    public Dictionary<string, object> Attributes { get; set; }  // Key fields only
}

// Integration service consumer pattern
public class IntegrationEventConsumer : BackgroundService
{
    private readonly EventHubConsumerClient _consumerClient;
    private readonly ILogger<IntegrationEventConsumer> _logger;

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await foreach (PartitionEvent partitionEvent in 
            _consumerClient.ReadEventsAsync(stoppingToken))
        {
            var businessEvent = JsonSerializer.Deserialize<BusinessEvent>(
                partitionEvent.Data.EventBody);

            // Only process events for this service's domains
            if (!IsRelevantEvent(businessEvent))
                continue;

            // Process immediately
            await ProcessBusinessEventAsync(businessEvent);

            // Immediate checkpoint for real-time processing
            await UpdateCheckpointAsync(partitionEvent);
        }
    }

    private bool IsRelevantEvent(BusinessEvent eventData)
    {
        var serviceDomains = new[] { "matching", "notifications" };
        return eventData.Metadata.Domains.Any(d => serviceDomains.Contains(d));
    }
}

// Metrics pipeline consumer pattern  
public class MetricsEventConsumer : BackgroundService
{
    private readonly EventHubConsumerClient _consumerClient;
    private readonly List<BusinessEvent> _eventBuffer = new();
    private readonly Timer _batchTimer;

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        // Setup batch processing timer
        _batchTimer = new Timer(ProcessBatch, null, 
            TimeSpan.FromMinutes(5), TimeSpan.FromMinutes(5));

        await foreach (PartitionEvent partitionEvent in 
            _consumerClient.ReadEventsAsync(stoppingToken))
        {
            var businessEvent = JsonSerializer.Deserialize<BusinessEvent>(
                partitionEvent.Data.EventBody);

            // Add all events to buffer for batch processing
            lock (_eventBuffer)
            {
                _eventBuffer.Add(businessEvent);
            }
        }
    }

    private async void ProcessBatch(object state)
    {
        List<BusinessEvent> eventsToProcess;
        lock (_eventBuffer)
        {
            eventsToProcess = new List<BusinessEvent>(_eventBuffer);
            _eventBuffer.Clear();
        }

        if (eventsToProcess.Any())
        {
            await ProcessEventBatch(eventsToProcess);

            // Update checkpoint after successful batch processing
            await UpdateBatchCheckpoint();
        }
    }
}

public class EntityReferenceService
{
    private readonly IMemoryCache _cache;
    private readonly IDataverseClient _dataverseClient;

    public async Task<EntityReference> ResolveReferenceAsync(
        string logicalName, Guid entityId, string[] requiredFields = null)
    {
        var cacheKey = $"entity:{logicalName}:{entityId}";

        if (_cache.TryGetValue(cacheKey, out EntityReference cached))
            return cached;

        // Fetch from DataVerse with only required fields
        var entity = await _dataverseClient.GetEntityAsync(
            logicalName, entityId, requiredFields ?? new[] { "name" });

        var reference = new EntityReference
        {
            Id = entityId,
            LogicalName = logicalName,
            DisplayName = entity.GetAttributeValue<string>("name"),
            Attributes = entity.Attributes.ToDictionary(
                kvp => kvp.Key, kvp => kvp.Value)
        };

        // Cache for 30 minutes (balance freshness vs performance)
        _cache.Set(cacheKey, reference, TimeSpan.FromMinutes(30));

        return reference;
    }
}

// Usage in business event creation
var candidateRef = await _entityReferenceService.ResolveReferenceAsync(
    "contact", 
    candidateId, 
    new[] { "fullname", "emailaddress1", "telephone1" });

public class EventOrderingService
{
    private readonly Dictionary<Guid, long> _entityVersions = new();
    private readonly Dictionary<Guid, Queue<BusinessEvent>> _eventBuffer = new();

    public async Task<IEnumerable<BusinessEvent>> ProcessEventAsync(BusinessEvent incomingEvent)
    {
        var entityId = incomingEvent.EntityId;
        var eventVersion = ExtractVersionFromEvent(incomingEvent);

        if (!_entityVersions.ContainsKey(entityId))
        {
            // First event for this entity
            _entityVersions[entityId] = eventVersion;
            return new[] { incomingEvent };
        }

        var lastVersion = _entityVersions[entityId];

        if (eventVersion == lastVersion + 1)
        {
            // In order - process immediately
            _entityVersions[entityId] = eventVersion;
            var eventsToEmit = new List<BusinessEvent> { incomingEvent };

            // Check if buffered events can now be processed
            eventsToEmit.AddRange(ProcessBufferedEvents(entityId));

            return eventsToEmit;
        }
        else if (eventVersion > lastVersion + 1)
        {
            // Out of order - buffer for later
            if (!_eventBuffer.ContainsKey(entityId))
                _eventBuffer[entityId] = new Queue<BusinessEvent>();

            _eventBuffer[entityId].Enqueue(incomingEvent);
            return Enumerable.Empty<BusinessEvent>();
        }
        else
        {
            // Duplicate or very old event - ignore
            return Enumerable.Empty<BusinessEvent>();
        }
    }

    private long ExtractVersionFromEvent(BusinessEvent businessEvent)
    {
        // Use DataVerse RowVersion or create sequence number
        return businessEvent.Data.Value<long>("rowversion") ?? 
               DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
    }
}