using System.Text; using Omika.Core.Observability; using Omika.Core.Observability.Outbox; namespace Omika.Tests.Observability.Outbox; /// Fabrique de test partagée pour le diagnostic distant (Phase 4). internal static class OutboxTestHarness { public static TelemetryEvent NewEvent( TelemetryPriority priority = TelemetryPriority.P3, TelemetryLevel level = TelemetryLevel.Info, EventCode? code = null, Guid? eventId = null, TelemetryMetadata? metadata = null) => new() { EventId = eventId ?? Guid.NewGuid(), EventCode = code ?? TelemetryEventCodes.MobileAppStarted, Level = level, Priority = priority, Metadata = metadata }; /// /// Lit un fichier en cohabitation avec un writer encore actif. Les exporteurs /// (LocalFileTelemetryExporter / RemoteDiagnosticIngestionStore) ouvrent en /// FileAccess.Write, FileShare.Read (un writer, lectures autorisées) : un /// lecteur doit donc ouvrir avec FileShare.ReadWrite — jamais /// / dont le /// FileShare.Read rejette l'écriture concurrente. /// public static string ReadAllTextShared(string path) { using var stream = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.ReadWrite); using var reader = new StreamReader(stream, Encoding.UTF8); return reader.ReadToEnd(); } /// Équivalent lignes de (UTF-8, partage d'écriture). public static string[] ReadAllLinesShared(string path) { var lines = new List(); using var stream = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.ReadWrite); using var reader = new StreamReader(stream, Encoding.UTF8); while (reader.ReadLine() is { } line) { lines.Add(line); } return lines.ToArray(); } } /// /// Store en mémoire pour les tests Core (aucune dépendance SQLite). Permet /// d'injecter une panne (ThrowOn) pour vérifier le fail-open de l'outbox. /// internal sealed class InMemoryDiagnosticOutboxStore : IDiagnosticOutboxStore { private readonly List _rows = []; public bool ThrowOnStore { get; set; } public bool ThrowOnCount { get; set; } public Task StoreAsync(IReadOnlyList rows, CancellationToken cancellationToken = default) { if (ThrowOnStore) { throw new InvalidOperationException("store down"); } lock (_rows) { foreach (DiagnosticOutboxRow row in rows) { _rows.RemoveAll(r => r.EventId == row.EventId); _rows.Add(row); } } return Task.CompletedTask; } public Task> GetNextBatchAsync( int maxCount, long maxBytes, DateTimeOffset now, CancellationToken cancellationToken = default) { lock (_rows) { var result = new List(); long bytes = 0; foreach (DiagnosticOutboxRow row in _rows .Where(r => r.NextAttemptUtc <= now) .OrderBy(r => r.Priority) .ThenBy(r => r.CreatedAtUtc)) { long rowBytes = Encoding.UTF8.GetByteCount(row.PayloadJson); if (result.Count > 0 && bytes + rowBytes > maxBytes) { break; } result.Add(row); bytes += rowBytes; if (result.Count >= maxCount) { break; } } return Task.FromResult>(result); } } public Task DeleteAsync(IReadOnlyList eventIds, CancellationToken cancellationToken = default) { lock (_rows) { HashSet ids = eventIds.ToHashSet(); _rows.RemoveAll(r => ids.Contains(r.EventId)); } return Task.CompletedTask; } public Task UpdateAttemptAsync( IReadOnlyList eventIds, int attemptCount, DateTimeOffset nextAttemptUtc, string? lastErrorCode, CancellationToken cancellationToken = default) { lock (_rows) { HashSet ids = eventIds.ToHashSet(); foreach (DiagnosticOutboxRow row in _rows.Where(r => ids.Contains(r.EventId)).ToList()) { int index = _rows.IndexOf(row); _rows[index] = new DiagnosticOutboxRow { EventId = row.EventId, Priority = row.Priority, CreatedAtUtc = row.CreatedAtUtc, ExpiresAtUtc = row.ExpiresAtUtc, PayloadVersion = row.PayloadVersion, PayloadJson = row.PayloadJson, AttemptCount = attemptCount, NextAttemptUtc = nextAttemptUtc, LastAttemptUtc = DateTimeOffset.UtcNow, LastErrorCode = lastErrorCode }; } } return Task.CompletedTask; } public Task CountAsync(CancellationToken cancellationToken = default) { if (ThrowOnCount) { throw new InvalidOperationException("count down"); } lock (_rows) { return Task.FromResult(_rows.Count); } } public Task DeleteExpiredAsync(DateTimeOffset now, CancellationToken cancellationToken = default) { lock (_rows) { return Task.FromResult(_rows.RemoveAll(r => r.ExpiresAtUtc <= now)); } } public Task DeleteOldestAsync(TelemetryPriority priority, int count, CancellationToken cancellationToken = default) { lock (_rows) { List toDelete = _rows .Where(r => r.Priority == priority) .OrderBy(r => r.CreatedAtUtc) .Take(count) .Select(r => r.EventId) .ToList(); _rows.RemoveAll(r => toDelete.Contains(r.EventId)); return Task.FromResult(toDelete.Count); } } public Task GetDatabaseSizeBytesAsync(CancellationToken cancellationToken = default) { lock (_rows) { return Task.FromResult((long)_rows.Count * 100); } } public IReadOnlyList Snapshot() { lock (_rows) { return _rows.ToList(); } } }