Omika/tests/Omika.Tests/Observability/External/RemoteDiagnosticForwarderTests.cs

using Omika.Core.Observability.External;
using Omika.Core.Observability.Outbox;

namespace Omika.Tests.Observability.External;

/// <summary>RemoteDiagnosticForwarder (§6/§40/§41/§42) : durable, checkpoint, doublons.</summary>
public class RemoteDiagnosticForwarderTests
{
    [Fact]
    public async Task ForwardsAll_AndAdvancesCheckpoint()
    {
        var source = new FakeForwarderSource();
        source.Add(Guid.NewGuid());
        source.Add(Guid.NewGuid());
        source.Add(Guid.NewGuid());

        var sink = new RecordingSink();
        var forwarder = NewForwarder(source, sink, out _);

        bool progressed = await forwarder.FlushAsync();

        Assert.True(progressed);
        Assert.Equal(3, sink.Received.Count);
        Assert.Equal(1, source.CommitCount);
        Assert.NotNull(source.Checkpoint);
        Assert.Equal(2, source.Checkpoint!.LineIndex);
    }

    [Fact]
    public async Task DoesNotAdvanceWhenCircuitOpen()
    {
        var source = new FakeForwarderSource();
        source.Add(Guid.NewGuid());

        var sink = new RecordingSink { Behavior = _ => 0 }; // drop (infra down)
        var circuit = new ExternalCircuitBreaker(1, TimeSpan.FromSeconds(30));
        circuit.RecordFailure(); // circuit ouvert

        var forwarder = NewForwarder(source, sink, circuit);

        bool progressed = await forwarder.FlushAsync();

        Assert.False(progressed);
        Assert.Equal(0, source.CommitCount); // pas d'avancement : on réessaiera plus tard
    }

    [Fact]
    public async Task AdvancesOverPolicyDroppedBatch()
    {
        var source = new FakeForwarderSource();
        source.Add(Guid.NewGuid());

        // La donnée est "consommée" par politique (sécurité/invalide) sans panne infra :
        // le forwarder doit avancer pour ne pas rejouer indéfiniment.
        var sink = new RecordingSink { Behavior = _ => 0 };
        var forwarder = NewForwarder(source, sink, out _);

        bool progressed = await forwarder.FlushAsync();

        Assert.False(progressed);
        Assert.Equal(1, source.CommitCount);
    }

    [Fact]
    public async Task AtLeastOnce_ProducesDuplicates_WhenCheckpointLost()
    {
        var source = new FakeForwarderSource();
        Guid eventId = Guid.NewGuid();
        source.Add(eventId);
        source.Add(Guid.NewGuid());

        var sink = new RecordingSink();
        var forwarder = NewForwarder(source, sink, out _);

        await forwarder.FlushAsync();
        Assert.Equal(2, sink.Received.Count);

        // Checkpoint perdu (redémarrage/panne) : at-least-once → réexpédition (doublons).
        source.Checkpoint = null;
        await forwarder.FlushAsync();

        Assert.Equal(4, sink.Received.Count);
        Assert.Equal(2, sink.Received.Count(r => r.EventId == eventId));
    }

    private static RemoteDiagnosticForwarder NewForwarder(
        FakeForwarderSource source,
        RecordingSink sink,
        out ExternalCircuitBreaker circuit)
    {
        circuit = new ExternalCircuitBreaker(3, TimeSpan.FromSeconds(30));
        return NewForwarder(source, sink, circuit);
    }

    private static RemoteDiagnosticForwarder NewForwarder(
        FakeForwarderSource source,
        RecordingSink sink,
        ExternalCircuitBreaker circuit)
    {
        var options = new ExternalObservabilityOptions
        {
            Enabled = true,
            OtlpEndpoint = "http://localhost:4318",
            ForwarderBatchSize = 100
        };

        return new RemoteDiagnosticForwarder(
            source,
            sink,
            options,
            new ExternalObservabilityCounters(),
            circuit);
    }

    private sealed class RecordingSink : IExternalDiagnosticSink
    {
        public List<DiagnosticEventDto> Received { get; } = [];

        public Func<IReadOnlyList<DiagnosticEventDto>, int>? Behavior { get; set; }

        public Task<int> ForwardAsync(IReadOnlyList<DiagnosticEventDto> events, CancellationToken cancellationToken = default)
        {
            Received.AddRange(events);
            return Task.FromResult(Behavior is null ? events.Count : Behavior(events));
        }
    }

    private sealed class FakeForwarderSource : IRemoteDiagnosticForwarderSource
    {
        private readonly List<RemoteDiagnosticRecord> _records = [];

        public RemoteDiagnosticForwarderPosition? Checkpoint { get; set; }

        public int CommitCount { get; private set; }

        public void Add(Guid eventId) => _records.Add(new RemoteDiagnosticRecord
        {
            Event = new DiagnosticEventDto { EventId = eventId, EventCode = "OMK-MOB-SYNC-001" }
        });

        public RemoteDiagnosticForwarderPosition? LoadCheckpoint() => Checkpoint;

        public RemoteDiagnosticForwardReadResult ReadNext(RemoteDiagnosticForwarderPosition? after, int maxCount)
        {
            int start = after is null ? 0 : Math.Min(_records.Count, after.LineIndex + 1);
            int take = Math.Min(maxCount, _records.Count - start);
            var records = _records.Skip(start).Take(take).ToList();

            return new RemoteDiagnosticForwardReadResult
            {
                Records = records,
                NextPosition = take == 0
                    ? null
                    : new RemoteDiagnosticForwarderPosition { FileName = "mem", LineIndex = start + take - 1 },
                ReachedEnd = start + take >= _records.Count
            };
        }

        public void Commit(RemoteDiagnosticForwarderPosition position)
        {
            Checkpoint = position;
            CommitCount++;
        }
    }
}