lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/RecorderTestDoubles.cs

using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;

namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian;

/// <summary>
///     Shared doubles for the recorder tests added with the pass-flush / off-mailbox-append work: a FIFO outbox and a
///     pass-aware writer that log every call into one ordered event list, so a test can assert the ORDER of writes,
///     flushes and acks across both. Thread-safe — the recorder appends and drains off the actor thread.
/// </summary>
internal sealed class EventLog
{
    private readonly Lock _gate = new();
    private readonly List<string> _events = new();

    public void Add(string e) { lock (_gate) _events.Add(e); }

    public IReadOnlyList<string> Snapshot() { lock (_gate) return _events.ToArray(); }
}

/// <summary>FIFO outbox honouring the truncate-through contract; appends can be made to yield or block.</summary>
internal sealed class LoggingOutbox(EventLog log) : IHistorizationOutbox
{
    private readonly Lock _gate = new();
    private readonly List<HistorizationOutboxEntry> _entries = new();
    private readonly List<HistorizationOutboxEntry> _appended = new();
    private int _appendBatchCalls;

    /// <summary>When set, every AppendBatchAsync waits on it (on a pool thread) before committing — a slow fsync.</summary>
    public TaskCompletionSource? AppendGate { get; set; }

    /// <summary>Completed by an append that has entered (before waiting on <see cref="AppendGate"/>).</summary>
    public TaskCompletionSource AppendEntered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);

    /// <summary>When set, CountAsync blocks on it — a hung outbox.</summary>
    public TaskCompletionSource? CountGate { get; set; }

    public int AppendBatchCalls { get { lock (_gate) return _appendBatchCalls; } }

    public long DroppedCount => 0;

    public IReadOnlyList<HistorizationOutboxEntry> Entries { get { lock (_gate) return _entries.ToArray(); } }

    /// <summary>Every entry ever appended, in append order (acks do not remove from it).</summary>
    public IReadOnlyList<HistorizationOutboxEntry> Appended { get { lock (_gate) return _appended.ToArray(); } }

    public ValueTask AppendAsync(HistorizationOutboxEntry entry, CancellationToken ct) => AppendBatchAsync([entry], ct);

    public async ValueTask AppendBatchAsync(IReadOnlyList<HistorizationOutboxEntry> entries, CancellationToken ct)
    {
        lock (_gate) _appendBatchCalls++;
        AppendEntered.TrySetResult();
        await Task.Delay(1, ct).ConfigureAwait(false); // complete on a pool thread, as a real fsync does
        if (AppendGate is { } gate) await gate.Task.WaitAsync(ct).ConfigureAwait(false);
        lock (_gate)
        {
            _entries.AddRange(entries);
            _appended.AddRange(entries);
        }
        log.Add($"append:{entries.Count}");
    }

    public ValueTask<IReadOnlyList<HistorizationOutboxEntry>> PeekBatchAsync(int max, CancellationToken ct)
    {
        lock (_gate) return ValueTask.FromResult<IReadOnlyList<HistorizationOutboxEntry>>(_entries.Take(max).ToArray());
    }

    public ValueTask RemoveAsync(Guid id, CancellationToken ct)
    {
        lock (_gate)
        {
            var i = _entries.FindIndex(e => e.Id == id);
            if (i >= 0) _entries.RemoveRange(0, i + 1);
        }
        log.Add("ack");
        return ValueTask.CompletedTask;
    }

    public async ValueTask<int> CountAsync(CancellationToken ct)
    {
        if (CountGate is { } gate) await gate.Task.WaitAsync(ct).ConfigureAwait(false);
        lock (_gate) return _entries.Count;
    }

    public void Dispose() { }
}

/// <summary>A pass-aware writer: logs every in-pass write and flush; plain writes are logged distinctly.</summary>
internal sealed class LoggingPassWriter(EventLog log) : IHistorianPassWriter
{
    private readonly HashSet<string> _failTags = new(StringComparer.Ordinal);

    public Exception? FlushThrows { get; set; }

    public void Fail(string tag) { lock (_failTags) _failTags.Add(tag); }

    public Task<bool> WriteLiveValuesAsync(string tag, IReadOnlyList<HistorizationValue> values, CancellationToken ct)
    {
        log.Add($"plain:{tag}");
        return Task.FromResult(true);
    }

    public async Task<bool> WriteLiveValuesInPassAsync(string tag, IReadOnlyList<HistorizationValue> values, CancellationToken ct)
    {
        await Task.Delay(1, ct).ConfigureAwait(false);
        bool fail;
        lock (_failTags) fail = _failTags.Contains(tag);
        log.Add($"write:{tag}");
        return !fail;
    }

    public async Task FlushAsync(CancellationToken ct)
    {
        await Task.Delay(1, ct).ConfigureAwait(false);
        log.Add("flush");
        if (FlushThrows is { } ex) throw ex;
    }
}