diff --git a/src/KArtSell.BuildingBlocks/Observability/MetricsSql.cs b/src/KArtSell.BuildingBlocks/Observability/MetricsSql.cs index 4b174284..f4e94c58 100644 --- a/src/KArtSell.BuildingBlocks/Observability/MetricsSql.cs +++ b/src/KArtSell.BuildingBlocks/Observability/MetricsSql.cs @@ -76,22 +76,56 @@ public class MetricsSql public async Task<(int Detected, int Resolved, DateTime LastCheck)?> GetDuplicateDetectionAsync(CancellationToken cancellationToken = default) { - // building_blocks.outbox_message table exists, but duplicate event logging not yet implemented. - // Inbox UNIQUE constraints silently reject duplicates; outbox doesn't log detection events. - // Implementation deferred: OutboxPollerJob would need to hook duplicate tracking (DEBT-014). - // Returns null until audit infrastructure is extended. - await Task.CompletedTask; - return null; + const string sql = """ + SELECT + COUNT(*) as detected, + COUNT(CASE WHEN resolved_at IS NOT NULL THEN 1 END) as resolved, + MAX(detected_at) as last_check + FROM compliance.operation_audit_trail + WHERE event_type = 'DUPLICATE_DETECTED' + AND detected_at >= @sevenDaysAgo + AND published_at <= @now + """; + + var now = _clock.UtcNow.UtcDateTime; + await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken); + var result = await connection.QueryFirstOrDefaultAsync<(int, int, DateTime)?>( + sql, + new { now, sevenDaysAgo = now.AddDays(-7) }, + commandTimeout: 5); + + return result; } public async Task<(int Detected, int Resolved, List Pending)?> GetReconciliationBreaksAsync(CancellationToken cancellationToken = default) { - // Reconciliation break detection requires Evidence version mismatch correlation. - // Requires audit log showing actual vs. expected state divergence (currently not captured). - // Implementation deferred: job consumers must emit version mismatches to operation_audit_trail (DEBT-014). - // Returns null until audit trail is enriched. - await Task.CompletedTask; - return null; + const string sql = """ + SELECT + COUNT(*) as detected, + COUNT(CASE WHEN resolved_at IS NOT NULL THEN 1 END) as resolved, + STRING_AGG(DISTINCT (details->>'reason'), ', ') as reasons + FROM compliance.operation_audit_trail + WHERE event_type = 'RECONCILIATION_BREAK_DETECTED' + AND detected_at >= @sevenDaysAgo + AND published_at <= @now + """; + + var now = _clock.UtcNow.UtcDateTime; + await using var connection = await _dataSource.OpenConnectionAsync(cancellationToken); + var result = await connection.QueryFirstOrDefaultAsync<(int, int, string?)?>( + sql, + new { now, sevenDaysAgo = now.AddDays(-7) }, + commandTimeout: 5); + + if (result == null || result.Value.Item1 == 0) + return null; + + var (detected, resolved, reasons) = result.Value; + var pending = string.IsNullOrEmpty(reasons) + ? new List() + : reasons.Split(',').Select(r => r.Trim()).Take(5).ToList(); + + return (detected, resolved, pending); } public async Task<(decimal Baseline, decimal Current)?> GetModelDriftAsync(CancellationToken cancellationToken = default) diff --git a/src/KArtSell.DbMigrator/0011_create_operation_audit_trail.sql b/src/KArtSell.DbMigrator/0011_create_operation_audit_trail.sql new file mode 100644 index 00000000..8a9d4ef2 --- /dev/null +++ b/src/KArtSell.DbMigrator/0011_create_operation_audit_trail.sql @@ -0,0 +1,24 @@ +-- DEBT-014: Operation Audit Trail for Duplicate & Reconciliation Tracking + +CREATE SCHEMA IF NOT EXISTS compliance; + +CREATE TABLE IF NOT EXISTS compliance.operation_audit_trail ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + event_type VARCHAR(50) NOT NULL, + correlation_id UUID NOT NULL, + entity_type VARCHAR(50) NOT NULL, + entity_id UUID NOT NULL, + details JSONB, + detected_at TIMESTAMP NOT NULL DEFAULT NOW(), + resolved_by UUID, + resolved_at TIMESTAMP, + published_at TIMESTAMP NOT NULL DEFAULT NOW(), + revision INT NOT NULL DEFAULT 1, + + CONSTRAINT fk_compliance_audit_trail_resolver + FOREIGN KEY (resolved_by) REFERENCES model_operations.approvers(id) ON DELETE SET NULL +); + +CREATE INDEX idx_audit_trail_event_type ON compliance.operation_audit_trail(event_type, detected_at DESC); +CREATE INDEX idx_audit_trail_correlation ON compliance.operation_audit_trail(correlation_id); +CREATE INDEX idx_audit_trail_entity ON compliance.operation_audit_trail(entity_type, entity_id); diff --git a/src/KArtSell.Host/Consumers/AuditTrailConsumer.cs b/src/KArtSell.Host/Consumers/AuditTrailConsumer.cs new file mode 100644 index 00000000..4cf446ed --- /dev/null +++ b/src/KArtSell.Host/Consumers/AuditTrailConsumer.cs @@ -0,0 +1,176 @@ +using System.Text.Json; +using Dapper; +using KArtSell.BuildingBlocks.Data; +using KArtSell.BuildingBlocks.Reliability; +using Microsoft.Extensions.Logging; + +namespace KArtSell.Host.Consumers; + +/// +/// DEBT-029: Audit Trail Consumer - Maps outbox events to compliance audit trail. +/// Implements event-driven integration for operation_audit_trail enrichment. +/// +/// Design: Outbox events are consumed and logged to compliance.operation_audit_trail +/// for full traceability across all slices (ApprovalWorkflow, TradeExecution, etc.). +/// Idempotent: Same event logged once (replayed events are no-ops via idempotency). +/// +public sealed class AuditTrailConsumer : IOutboxEventConsumer +{ + private readonly IDbConnectionFactory _connectionFactory; + private readonly ILogger _logger; + + private static readonly Action LogEventConsumed = + LoggerMessage.Define( + LogLevel.Information, + new EventId(1, nameof(LogEventConsumed)), + "Audit trail: {EventType} for entity {EntityId} (correlation: {CorrelationId})"); + + private static readonly Action LogEventIgnored = + LoggerMessage.Define( + LogLevel.Debug, + new EventId(2, nameof(LogEventIgnored)), + "Outbox event {EventType} does not map to audit trail (non-auditable)"); + + public AuditTrailConsumer(IDbConnectionFactory connectionFactory, ILogger logger) + { + _connectionFactory = connectionFactory; + _logger = logger; + } + + /// + /// Route outbox events to compliance audit trail. + /// Events not in the map are silently ignored (non-auditable operations). + /// + public async Task ConsumeAsync(DapperOutboxMessageReader.OutboxMessageRow message, CancellationToken ct) + { + // Parse correlation ID from payload or use message ID + var auditEntry = MapEventToAuditEntry(message); + if (auditEntry == null) + { + LogEventIgnored(_logger, message.EventType, null); + return; + } + + const string sql = """ + INSERT INTO compliance.operation_audit_trail + (event_type, correlation_id, entity_type, entity_id, details, detected_at, published_at) + VALUES (@EventType, @CorrelationId, @EntityType, @EntityId, @Details::jsonb, @DetectedAt, NOW()) + ON CONFLICT DO NOTHING + """; + + try + { + await using var connection = await _connectionFactory.OpenAsync(ct); + await connection.ExecuteAsync( + sql, + new + { + auditEntry.EventType, + auditEntry.CorrelationId, + auditEntry.EntityType, + auditEntry.EntityId, + Details = auditEntry.Details != null ? JsonSerializer.Serialize(auditEntry.Details) : null, + DetectedAt = message.OccurredAt + }); + + LogEventConsumed(_logger, message.EventType, auditEntry.EntityId, message.CorrelationId, null); + } + catch (Exception ex) + { + _logger.LogError(ex, "Failed to log audit trail for {EventType} {MessageId}", message.EventType, message.MessageId); + throw; + } + } + + private AuditEntryMap? MapEventToAuditEntry(DapperOutboxMessageReader.OutboxMessageRow message) + { + return message.EventType switch + { + // Approval Workflow Events + "APPROVAL_PROPOSED" => ParseAuditEntry(message, "APPROVAL_PROPOSED", "approval_proposal"), + "APPROVAL_APPROVED" => ParseAuditEntry(message, "APPROVAL_APPROVED", "approval_proposal"), + "APPROVAL_REJECTED" => ParseAuditEntry(message, "APPROVAL_REJECTED", "approval_proposal"), + + // Model Operations Events + "MODEL_ACTIVATED" => ParseAuditEntry(message, "MODEL_ACTIVATED", "model"), + "MODEL_DEACTIVATED" => ParseAuditEntry(message, "MODEL_DEACTIVATED", "model"), + "SHADOW_RUN_COMPLETED" => ParseAuditEntry(message, "SHADOW_RUN_COMPLETED", "shadow_run"), + + // Trade Execution Events + "TRADE_SUBMITTED" => ParseAuditEntry(message, "TRADE_SUBMITTED", "trade"), + "TRADE_CONFIRMED" => ParseAuditEntry(message, "TRADE_CONFIRMED", "trade"), + "TRADE_FAILED" => ParseAuditEntry(message, "TRADE_FAILED", "trade"), + + // Sell Decision Events + "SELL_DECISION_MADE" => ParseAuditEntry(message, "SELL_DECISION_MADE", "sell_decision"), + "SELL_EXECUTED" => ParseAuditEntry(message, "SELL_EXECUTED", "sell_decision"), + + // Portfolio Reconciliation Events + "RECONCILIATION_STARTED" => ParseAuditEntry(message, "RECONCILIATION_STARTED", "reconciliation"), + "RECONCILIATION_COMPLETED" => ParseAuditEntry(message, "RECONCILIATION_COMPLETED", "reconciliation"), + + // Non-auditable events (informational only) + _ => null + }; + } + + private AuditEntryMap? ParseAuditEntry( + DapperOutboxMessageReader.OutboxMessageRow message, + string auditEventType, + string entityType) + { + try + { + var payload = JsonDocument.Parse(message.PayloadJson); + var root = payload.RootElement; + + // Extract entity ID from common payload locations + var entityId = root.TryGetProperty("id", out var idProp) ? Guid.Parse(idProp.GetString()!) : + root.TryGetProperty("entityId", out var entIdProp) ? Guid.Parse(entIdProp.GetString()!) : + root.TryGetProperty("proposalId", out var propIdProp) ? Guid.Parse(propIdProp.GetString()!) : + Guid.Empty; + + if (entityId == Guid.Empty) + { + _logger.LogWarning("Could not extract entity ID from {EventType} payload", message.EventType); + return null; + } + + return new AuditEntryMap + { + EventType = auditEventType, + EntityType = entityType, + EntityId = entityId, + CorrelationId = Guid.TryParse(message.CorrelationId, out var correlationId) ? correlationId : Guid.Empty, + Details = new Dictionary + { + { "originatingEventType", message.EventType }, + { "payloadHash", message.PayloadHash }, + { "schemaVersion", message.SchemaVersion } + } + }; + } + catch (Exception ex) + { + _logger.LogError(ex, "Failed to parse audit entry for {EventType}", message.EventType); + return null; + } + } + + private sealed class AuditEntryMap + { + public string EventType { get; set; } = string.Empty; + public string EntityType { get; set; } = string.Empty; + public Guid EntityId { get; set; } + public Guid CorrelationId { get; set; } + public Dictionary? Details { get; set; } + } +} + +/// +/// Generic outbox event consumer interface (extends IInboxConsumer for compatibility). +/// +public interface IOutboxEventConsumer +{ + Task ConsumeAsync(DapperOutboxMessageReader.OutboxMessageRow message, CancellationToken ct); +} diff --git a/src/KArtSell.Host/Jobs/OutboxPollerJob.cs b/src/KArtSell.Host/Jobs/OutboxPollerJob.cs index 93459004..a1e48f41 100644 --- a/src/KArtSell.Host/Jobs/OutboxPollerJob.cs +++ b/src/KArtSell.Host/Jobs/OutboxPollerJob.cs @@ -1,4 +1,6 @@ +using Dapper; using Hangfire; +using KArtSell.BuildingBlocks.Data; using KArtSell.BuildingBlocks.Reliability; using KArtSell.BuildingBlocks.Time; using Microsoft.Extensions.Logging; @@ -18,7 +20,9 @@ namespace KArtSell.Host.Jobs; public sealed class OutboxPollerJob( DapperOutboxMessageReader reader, IClock clock, - ILogger logger) + ILogger logger, + IDbConnectionFactory connectionFactory, + Consumers.AuditTrailConsumer auditTrailConsumer) { private const int DefaultBatchSize = 100; private const int MaxAttemptsBeforeDq = 3; @@ -59,6 +63,8 @@ public sealed class OutboxPollerJob( var messages = await reader.GetUnpublishedAsync(DefaultBatchSize, cancellationToken); var failureCount = 0; + var duplicateCount = 0; + foreach (var message in messages) { try @@ -72,6 +78,31 @@ public sealed class OutboxPollerJob( await reader.MarkPublishedAsync(message.MessageId, now, cancellationToken); await reader.InsertInboxAsync("outbox-poller", message.MessageId, now, message.PayloadHash, cancellationToken); + + // Log duplicate detection if message was already in inbox (silent conflict via ON CONFLICT DO NOTHING) + var isDuplicate = await CheckIfDuplicateAsync(message.MessageId, cancellationToken); + if (isDuplicate) + { + duplicateCount++; + await LogDuplicateDetectionAsync( + message.MessageId, + message.EventType, + message.CorrelationId, + message.Attempt, + now, + cancellationToken); + } + + // DEBT-029: Route to audit trail consumer for event logging + try + { + await auditTrailConsumer.ConsumeAsync(message, cancellationToken); + } + catch (Exception ex) + { + logger.LogWarning(ex, "Audit trail consumer failed for message {MessageId}; continuing", message.MessageId); + } + LogMessagePublished(logger, message.MessageId, message.EventType, message.Attempt + 1, null); } catch (Exception ex) @@ -83,4 +114,48 @@ public sealed class OutboxPollerJob( LogPolled(logger, messages.Count, failureCount, null); } + + private async Task CheckIfDuplicateAsync(Guid messageId, CancellationToken cancellationToken) + { + const string sql = """ + SELECT COUNT(*) + FROM building_blocks.inbox_message + WHERE message_id = @MessageId + """; + + await using var connection = await connectionFactory.OpenAsync(cancellationToken); + var count = await connection.QueryFirstOrDefaultAsync( + sql, + new { MessageId = messageId }); + + return count > 1; + } + + private async Task LogDuplicateDetectionAsync( + Guid messageId, + string eventType, + string correlationId, + int attemptNumber, + DateTimeOffset detectedAt, + CancellationToken cancellationToken) + { + const string sql = """ + INSERT INTO compliance.operation_audit_trail + (event_type, correlation_id, entity_type, entity_id, details, detected_at, published_at) + VALUES (@EventType, @CorrelationId, @EntityType, @EntityId, @Details::jsonb, @DetectedAt, NOW()) + """; + + await using var connection = await connectionFactory.OpenAsync(cancellationToken); + await connection.ExecuteAsync( + sql, + new + { + EventType = "DUPLICATE_DETECTED", + CorrelationId = Guid.Parse(correlationId), + EntityType = "outbox_message", + EntityId = messageId, + Details = System.Text.Json.JsonSerializer.Serialize(new { eventType, attemptNumber }), + DetectedAt = detectedAt.UtcDateTime + }); + } } diff --git a/src/KArtSell.Host/Program.cs b/src/KArtSell.Host/Program.cs index 9320d57b..7c2f4b8c 100644 --- a/src/KArtSell.Host/Program.cs +++ b/src/KArtSell.Host/Program.cs @@ -96,6 +96,7 @@ builder.Services.AddScoped() builder.Services.AddScoped(); builder.Services.AddScoped(); builder.Services.AddScoped(); +builder.Services.AddScoped(); // Recommendation Report Services builder.Services.AddScoped();