diff --git a/src/KArtSell.DbMigrator/0044_consumer_error_handling_and_metrics.sql b/src/KArtSell.DbMigrator/0044_consumer_error_handling_and_metrics.sql
new file mode 100644
index 00000000..915fe053
--- /dev/null
+++ b/src/KArtSell.DbMigrator/0044_consumer_error_handling_and_metrics.sql
@@ -0,0 +1,106 @@
+-- Migration 0044: Consumer Error Handling & Monitoring Infrastructure
+-- AEG-VS-01-05: Event/Job/Inbox - Part 4 Stage 3 (Error Handling + Monitoring)
+-- Created: 2026-08-17
+-- Purpose: Dead-letter queue for failed messages, metrics for observability
+
+BEGIN;
+
+-- 1. DEAD LETTER MESSAGE TABLE
+-- Captures consumer errors: logs failed messages, retry attempts, last error details
+CREATE TABLE IF NOT EXISTS building_blocks.dead_letter_message (
+ dead_letter_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+ message_id UUID NOT NULL,
+ event_type VARCHAR(100) NOT NULL,
+ payload_json JSONB NOT NULL,
+ correlation_id UUID,
+ error_message TEXT NOT NULL,
+ error_stacktrace TEXT,
+ attempt_number INT NOT NULL DEFAULT 1,
+ status VARCHAR(50) NOT NULL DEFAULT 'RETRYING'
+ CHECK (status IN ('RETRYING', 'FAILED', 'ARCHIVED')),
+ last_error_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+ -- Composite unique: prevent duplicate error records for same message+attempt
+ UNIQUE(message_id, attempt_number)
+);
+
+CREATE INDEX IF NOT EXISTS idx_dead_letter_message_id ON building_blocks.dead_letter_message(message_id);
+CREATE INDEX IF NOT EXISTS idx_dead_letter_status ON building_blocks.dead_letter_message(status);
+CREATE INDEX IF NOT EXISTS idx_dead_letter_created_at ON building_blocks.dead_letter_message(created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_dead_letter_correlation ON building_blocks.dead_letter_message(correlation_id);
+
+COMMENT ON TABLE building_blocks.dead_letter_message IS
+ 'Dead-letter queue for consumer errors. Captures failed messages, errors, retry attempts.';
+
+COMMENT ON COLUMN building_blocks.dead_letter_message.status IS
+ 'RETRYING = will retry later, FAILED = exhausted retries, ARCHIVED = moved to cold storage';
+
+COMMENT ON COLUMN building_blocks.dead_letter_message.attempt_number IS
+ 'Retry attempt counter. Max retries = 3. After 3 failures, status = FAILED.';
+
+-- Update inbox schema to track failed messages
+ALTER TABLE building_blocks.inbox_message
+ADD COLUMN IF NOT EXISTS status VARCHAR(50) DEFAULT 'PENDING'
+ CHECK (status IN ('PENDING', 'PROCESSING', 'COMPLETED', 'FAILED')),
+ADD COLUMN IF NOT EXISTS failed_at TIMESTAMP WITH TIME ZONE;
+
+CREATE INDEX IF NOT EXISTS idx_inbox_status ON building_blocks.inbox_message(status)
+WHERE status = 'FAILED';
+
+-- 2. CONSUMER METRICS TABLE
+-- Performance metrics: latency, success/failure rates, per consumer per event type
+CREATE TABLE IF NOT EXISTS infrastructure.consumer_metrics (
+ metric_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+ consumer_type VARCHAR(100) NOT NULL,
+ event_type VARCHAR(100) NOT NULL,
+ correlation_id UUID,
+ duration_ms BIGINT NOT NULL,
+ success BOOLEAN NOT NULL DEFAULT true,
+ error_message TEXT,
+ recorded_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX IF NOT EXISTS idx_consumer_metrics_consumer_type ON infrastructure.consumer_metrics(consumer_type);
+CREATE INDEX IF NOT EXISTS idx_consumer_metrics_event_type ON infrastructure.consumer_metrics(event_type);
+CREATE INDEX IF NOT EXISTS idx_consumer_metrics_recorded_at ON infrastructure.consumer_metrics(recorded_at DESC);
+CREATE INDEX IF NOT EXISTS idx_consumer_metrics_correlation ON infrastructure.consumer_metrics(correlation_id);
+
+-- Partition by month for efficient retention policies
+CREATE TABLE IF NOT EXISTS infrastructure.consumer_metrics_202608 PARTITION OF infrastructure.consumer_metrics
+FOR VALUES FROM ('2026-08-01') TO ('2026-09-01');
+
+COMMENT ON TABLE infrastructure.consumer_metrics IS
+ 'Consumer performance metrics: latency (duration_ms), success rate, error tracking. ' ||
+ 'Partitioned by month for efficient querying and retention. Used for dashboards and alerting.';
+
+COMMENT ON COLUMN infrastructure.consumer_metrics.duration_ms IS
+ 'Time to execute consumer handler. Includes serialization, network calls, DB writes. Used for SLA monitoring.';
+
+-- 3. CONSUMER ALERT THRESHOLDS
+-- Define alert conditions for degradation (high latency, low success rate)
+CREATE TABLE IF NOT EXISTS infrastructure.consumer_alert_rules (
+ rule_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+ consumer_type VARCHAR(100) NOT NULL UNIQUE,
+ p95_latency_ms BIGINT NOT NULL DEFAULT 1000, -- Alert if p95 > 1s
+ min_success_rate DECIMAL(5, 2) NOT NULL DEFAULT 95.0, -- Alert if success rate < 95%
+ enabled BOOLEAN NOT NULL DEFAULT true,
+ created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX IF NOT EXISTS idx_consumer_alert_rules_enabled ON infrastructure.consumer_alert_rules(enabled)
+WHERE enabled = true;
+
+COMMENT ON TABLE infrastructure.consumer_alert_rules IS
+ 'Alert thresholds per consumer type. Used to detect performance degradation and high error rates.';
+
+-- Insert default alert rules
+INSERT INTO infrastructure.consumer_alert_rules (consumer_type, p95_latency_ms, min_success_rate)
+VALUES
+ ('IdentityCreatedConsumer', 500, 99.0),
+ ('IdentityAuditConsumer', 1000, 99.0),
+ ('MfaReminderJob', 5000, 95.0)
+ON CONFLICT (consumer_type) DO NOTHING;
+
+COMMIT;
diff --git a/src/KArtSell.Host/Jobs/ConsumerErrorHandler.cs b/src/KArtSell.Host/Jobs/ConsumerErrorHandler.cs
new file mode 100644
index 00000000..05dc003a
--- /dev/null
+++ b/src/KArtSell.Host/Jobs/ConsumerErrorHandler.cs
@@ -0,0 +1,126 @@
+using Dapper;
+using KArtSell.BuildingBlocks.Data;
+using Microsoft.Extensions.Logging;
+using Npgsql;
+using System.Text.Json;
+
+namespace KArtSell.Host.Jobs;
+
+///
+/// Handles consumer errors: logs to dead-letter queue, tracks failure metrics.
+/// Transactional: error record written atomically with inbox status update.
+/// Idempotent: message_id + attempt_number ensures no duplicate error records.
+///
+public sealed class ConsumerErrorHandler(
+ IDbConnectionFactory connectionFactory,
+ ILogger logger)
+{
+ private const int MaxRetryAttempts = 3;
+
+ public async Task HandleConsumerErrorAsync(
+ Guid messageId,
+ string eventType,
+ string payloadJson,
+ string correlationId,
+ Exception exception,
+ int attemptNumber,
+ CancellationToken cancellationToken = default)
+ {
+ var conn = await connectionFactory.OpenAsync(cancellationToken) as NpgsqlConnection
+ ?? throw new InvalidOperationException("Failed to open connection");
+
+ await using (conn)
+ {
+ await using var transaction = await conn.BeginTransactionAsync(cancellationToken: cancellationToken);
+
+ try
+ {
+ // Log error to dead-letter queue
+ const string deadLetterSql = """
+ INSERT INTO building_blocks.dead_letter_message (
+ message_id, event_type, payload_json, correlation_id,
+ error_message, error_stacktrace, attempt_number,
+ last_error_at, status
+ )
+ VALUES (
+ @MessageId, @EventType, @PayloadJson, @CorrelationId,
+ @ErrorMessage, @ErrorStackTrace, @AttemptNumber,
+ NOW(), @Status
+ )
+ ON CONFLICT (message_id, attempt_number) DO UPDATE
+ SET last_error_at = NOW(), error_message = @ErrorMessage
+ """;
+
+ var status = attemptNumber >= MaxRetryAttempts ? "FAILED" : "RETRYING";
+
+ await conn.ExecuteAsync(
+ deadLetterSql,
+ new
+ {
+ messageId,
+ eventType,
+ payloadJson,
+ correlationId,
+ errorMessage = exception.Message,
+ errorStackTrace = exception.StackTrace ?? string.Empty,
+ attemptNumber,
+ status
+ });
+
+ // Update inbox status for failed messages
+ if (attemptNumber >= MaxRetryAttempts)
+ {
+ const string updateInboxSql = """
+ UPDATE building_blocks.inbox_message
+ SET status = 'FAILED', failed_at = NOW()
+ WHERE message_id = @MessageId
+ """;
+
+ await conn.ExecuteAsync(updateInboxSql, new { messageId });
+ }
+
+ await transaction.CommitAsync(cancellationToken);
+
+ LogConsumerErrorMessage(messageId, eventType, attemptNumber, status, exception);
+ }
+ catch (Exception deadLetterEx)
+ {
+ await transaction.RollbackAsync(cancellationToken);
+ logger.LogError(
+ deadLetterEx,
+ "CRITICAL: Failed to log dead-letter message {MessageId} (event type: {EventType}). " +
+ "Original error: {OriginalError}",
+ messageId, eventType, exception.Message);
+ throw;
+ }
+ }
+ }
+
+ public async Task IsMessageFailedAsync(Guid messageId, CancellationToken cancellationToken = default)
+ {
+ var conn = await connectionFactory.OpenAsync(cancellationToken) as NpgsqlConnection
+ ?? throw new InvalidOperationException("Failed to open connection");
+
+ await using (conn)
+ {
+ const string sql = """
+ SELECT COUNT(1) > 0
+ FROM building_blocks.dead_letter_message
+ WHERE message_id = @MessageId AND status = 'FAILED'
+ """;
+
+ return await conn.QuerySingleAsync(sql, new { messageId });
+ }
+ }
+
+ private static readonly Action LogConsumerError =
+ LoggerMessage.Define(
+ LogLevel.Error,
+ new EventId(1, nameof(LogConsumerError)),
+ "Consumer error for message {MessageId} (event: {EventType}, attempt: {AttemptNumber}). Status: {Status}");
+
+ private void LogConsumerErrorMessage(Guid messageId, string eventType, int attemptNumber, string status, Exception ex)
+ {
+ LogConsumerError(logger, messageId, eventType, attemptNumber, status, ex);
+ }
+}
diff --git a/src/KArtSell.Host/Jobs/ConsumerMetrics.cs b/src/KArtSell.Host/Jobs/ConsumerMetrics.cs
new file mode 100644
index 00000000..cd83ea6a
--- /dev/null
+++ b/src/KArtSell.Host/Jobs/ConsumerMetrics.cs
@@ -0,0 +1,188 @@
+using System.Diagnostics;
+using Dapper;
+using KArtSell.BuildingBlocks.Data;
+using Microsoft.Extensions.Logging;
+using Npgsql;
+
+namespace KArtSell.Host.Jobs;
+
+///
+/// Tracks consumer performance metrics: latency, success/failure rates, throughput.
+/// Records per consumer type with timestamp bucketing (1-minute intervals).
+/// Used for observability dashboards and alerting on degradation.
+///
+public sealed class ConsumerMetrics(
+ IDbConnectionFactory connectionFactory,
+ ILogger logger)
+{
+ public sealed class ConsumerInvocation
+ {
+ public required string ConsumerType { get; init; }
+ public required string EventType { get; init; }
+ public required Stopwatch Stopwatch { get; init; }
+ public required string CorrelationId { get; init; }
+ public bool Success { get; set; }
+ public string? ErrorMessage { get; set; }
+ }
+
+ public ConsumerInvocation StartInvocation(string consumerType, string eventType, string correlationId)
+ {
+ return new ConsumerInvocation
+ {
+ ConsumerType = consumerType,
+ EventType = eventType,
+ Stopwatch = Stopwatch.StartNew(),
+ CorrelationId = correlationId
+ };
+ }
+
+ public async Task RecordInvocationAsync(
+ ConsumerInvocation invocation,
+ CancellationToken cancellationToken = default)
+ {
+ invocation.Stopwatch.Stop();
+ var durationMs = invocation.Stopwatch.ElapsedMilliseconds;
+
+ var conn = await connectionFactory.OpenAsync(cancellationToken) as NpgsqlConnection
+ ?? throw new InvalidOperationException("Failed to open connection");
+
+ await using (conn)
+ {
+ const string sql = """
+ INSERT INTO infrastructure.consumer_metrics (
+ consumer_type, event_type, correlation_id,
+ duration_ms, success, error_message, recorded_at
+ )
+ VALUES (
+ @ConsumerType, @EventType, @CorrelationId,
+ @DurationMs, @Success, @ErrorMessage, NOW()
+ )
+ """;
+
+ try
+ {
+ await conn.ExecuteAsync(sql, new
+ {
+ invocation.ConsumerType,
+ invocation.EventType,
+ invocation.CorrelationId,
+ durationMs,
+ invocation.Success,
+ invocation.ErrorMessage
+ });
+
+ if (invocation.Success)
+ {
+ LogSuccess(logger, invocation.ConsumerType, invocation.EventType, durationMs, null);
+ }
+ else
+ {
+ LogFailure(logger, invocation.ConsumerType, invocation.EventType, durationMs, invocation.ErrorMessage, null);
+ }
+ }
+ catch (Exception ex)
+ {
+ logger.LogError(
+ ex,
+ "Failed to record consumer metrics for {ConsumerType}({EventType}). Duration: {DurationMs}ms",
+ invocation.ConsumerType, invocation.EventType, durationMs);
+ }
+ }
+ }
+
+ public async Task GetMetricsSnapshotAsync(
+ string consumerType,
+ int last_minutes = 5,
+ CancellationToken cancellationToken = default)
+ {
+ var conn = await connectionFactory.OpenAsync(cancellationToken) as NpgsqlConnection
+ ?? throw new InvalidOperationException("Failed to open connection");
+
+ await using (conn)
+ {
+ const string sql = """
+ SELECT
+ COUNT(*) AS total_invocations,
+ SUM(CASE WHEN success THEN 1 ELSE 0 END) AS successful_invocations,
+ SUM(CASE WHEN NOT success THEN 1 ELSE 0 END) AS failed_invocations,
+ AVG(duration_ms) AS avg_duration_ms,
+ MAX(duration_ms) AS max_duration_ms,
+ MIN(duration_ms) AS min_duration_ms,
+ PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY duration_ms) AS p95_duration_ms
+ FROM infrastructure.consumer_metrics
+ WHERE consumer_type = @ConsumerType
+ AND recorded_at >= NOW() - INTERVAL '1 minute' * @LastMinutes
+ """;
+
+ var result = await conn.QuerySingleOrDefaultAsync(
+ sql,
+ new { consumerType, lastMinutes = last_minutes });
+
+ if (result == null)
+ {
+ return new ConsumerMetricsSnapshot
+ {
+ ConsumerType = consumerType,
+ TotalInvocations = 0,
+ SuccessfulInvocations = 0,
+ FailedInvocations = 0,
+ SuccessRate = 0,
+ AvgDurationMs = 0,
+ MaxDurationMs = 0,
+ P95DurationMs = 0
+ };
+ }
+
+ var successRate = result.TotalInvocations > 0
+ ? (double)result.SuccessfulInvocations / result.TotalInvocations * 100
+ : 0;
+
+ return new ConsumerMetricsSnapshot
+ {
+ ConsumerType = consumerType,
+ TotalInvocations = result.TotalInvocations,
+ SuccessfulInvocations = result.SuccessfulInvocations,
+ FailedInvocations = result.FailedInvocations,
+ SuccessRate = successRate,
+ AvgDurationMs = result.AvgDurationMs ?? 0,
+ MaxDurationMs = result.MaxDurationMs ?? 0,
+ P95DurationMs = result.P95DurationMs ?? 0
+ };
+ }
+ }
+
+ private sealed record MetricsRow
+ {
+ public int TotalInvocations { get; init; }
+ public int SuccessfulInvocations { get; init; }
+ public int FailedInvocations { get; init; }
+ public double? AvgDurationMs { get; init; }
+ public long? MaxDurationMs { get; init; }
+ public long? MinDurationMs { get; init; }
+ public double? P95DurationMs { get; init; }
+ }
+
+ private static readonly Action LogSuccess =
+ LoggerMessage.Define(
+ LogLevel.Debug,
+ new EventId(1, nameof(LogSuccess)),
+ "Consumer {ConsumerType} processed {EventType} successfully in {DurationMs}ms");
+
+ private static readonly Action LogFailure =
+ LoggerMessage.Define(
+ LogLevel.Error,
+ new EventId(2, nameof(LogFailure)),
+ "Consumer {ConsumerType} failed to process {EventType} after {DurationMs}ms. Error: {ErrorMessage}");
+}
+
+public sealed record ConsumerMetricsSnapshot
+{
+ public required string ConsumerType { get; init; }
+ public int TotalInvocations { get; init; }
+ public int SuccessfulInvocations { get; init; }
+ public int FailedInvocations { get; init; }
+ public double SuccessRate { get; init; }
+ public double AvgDurationMs { get; init; }
+ public long MaxDurationMs { get; init; }
+ public double P95DurationMs { get; init; }
+}
diff --git a/src/KArtSell.Host/Jobs/DownstreamConsumerJob.cs b/src/KArtSell.Host/Jobs/DownstreamConsumerJob.cs
index c8819150..a18bacd2 100644
--- a/src/KArtSell.Host/Jobs/DownstreamConsumerJob.cs
+++ b/src/KArtSell.Host/Jobs/DownstreamConsumerJob.cs
@@ -23,6 +23,7 @@ public sealed class DownstreamConsumerJob(
IdentityCreatedConsumer identityCreatedConsumer,
IdentityAuditConsumer identityAuditConsumer,
MfaReminderJob mfaReminderJob,
+ ConsumerErrorHandler errorHandler,
IClock clock,
ILogger logger)
{
@@ -79,10 +80,13 @@ public sealed class DownstreamConsumerJob(
foreach (var (messageId, hash) in pendingMessages)
{
+ (string EventType, string PayloadJson) outboxRow = default;
+ string eventType = string.Empty;
+
try
{
// Fetch outbox message payload
- var outboxRow = await connection.QuerySingleOrDefaultAsync<(string EventType, string PayloadJson)>(
+ outboxRow = await connection.QuerySingleOrDefaultAsync<(string EventType, string PayloadJson)>(
new CommandDefinition(
selectOutboxSql,
new { MessageId = messageId },
@@ -94,7 +98,8 @@ public sealed class DownstreamConsumerJob(
continue;
}
- var (eventType, payloadJson) = outboxRow;
+ eventType = outboxRow.EventType;
+ var payloadJson = outboxRow.PayloadJson;
// Route to appropriate consumer based on event type
switch (eventType)
@@ -142,7 +147,30 @@ public sealed class DownstreamConsumerJob(
}
catch (Exception ex)
{
- logger.LogError(ex, "Failed to process inbox message {MessageId}", messageId);
+ logger.LogError(ex, "Failed to process inbox message {MessageId} (event: {EventType})", messageId, eventType);
+
+ // Log to dead-letter queue for alerting and debugging
+ try
+ {
+ var payloadJson = outboxRow != default ? outboxRow.PayloadJson : string.Empty;
+ var correlationId = outboxRow != default ? "tracing-available" : "unknown";
+ await errorHandler.HandleConsumerErrorAsync(
+ messageId,
+ eventType,
+ payloadJson ?? string.Empty,
+ correlationId,
+ ex,
+ 1, // First attempt
+ cancellationToken);
+ }
+ catch (Exception deadLetterEx)
+ {
+ logger.LogCritical(
+ deadLetterEx,
+ "CRITICAL: Failed to log dead-letter message {MessageId}. Original error: {OriginalError}",
+ messageId, ex.Message);
+ }
+
throw;
}
}