using System.Diagnostics.Metrics; using Npgsql; using Tiku.Infrastructure.Messaging; namespace Tiku.Api.Observability; public sealed class OutboxBacklogSnapshot { private static readonly Meter Meter = new("Tiku.Messaging", "1.0.0"); private long pending; private double oldestAgeSeconds; public OutboxBacklogSnapshot() { Meter.CreateObservableGauge("tiku.outbox.pending", () => Interlocked.Read(ref pending)); Meter.CreateObservableGauge("tiku.outbox.oldest_age", () => Volatile.Read(ref oldestAgeSeconds), "s"); } internal long Pending => Interlocked.Read(ref pending); internal double OldestAgeSeconds => Volatile.Read(ref oldestAgeSeconds); internal void Update(long count, DateTime? oldestSentTime) { Interlocked.Exchange(ref pending, count); Volatile.Write(ref oldestAgeSeconds, oldestSentTime is null ? 0 : Math.Max(0, (DateTimeOffset.UtcNow - new DateTimeOffset(oldestSentTime.Value)).TotalSeconds)); } } internal sealed class OutboxBacklogMonitor( NpgsqlDataSource dataSource, OutboxBacklogSnapshot snapshot, MessagingOptions messagingOptions, ILogger logger) : BackgroundService { protected override async Task ExecuteAsync(CancellationToken stoppingToken) { while (!stoppingToken.IsCancellationRequested) { try { if (messagingOptions.IsConfigured) { await using var command = dataSource.CreateCommand( "SELECT count(*), min(sent_time) FROM outbox_message;"); await using var reader = await command.ExecuteReaderAsync(stoppingToken); if (await reader.ReadAsync(stoppingToken)) { snapshot.Update(reader.GetInt64(0), reader.IsDBNull(1) ? null : reader.GetDateTime(1)); } } } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { break; } catch (Exception exception) { logger.LogWarning(exception, "Outbox backlog metric collection failed."); } await Task.Delay(TimeSpan.FromSeconds(30), stoppingToken); } } }