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();
}
}
}