ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardReplicationTests.cs

using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.TestSupport;

namespace ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests;

/// <summary>
/// StoreAndForward-001: the active node must forward every buffer operation
/// (add / remove / park) to the standby via the ReplicationService, so a
/// failover does not lose the buffer.
/// </summary>
public class StoreAndForwardReplicationTests : IAsyncLifetime, IDisposable
{
    private readonly TestLocalDb _localDb;
    private readonly StoreAndForwardStorage _storage;
    private readonly StoreAndForwardService _service;
    private readonly List<ReplicationOperation> _replicated = new();

    public StoreAndForwardReplicationTests()
    {
        _localDb = TestLocalDb.CreateTemp("ReplTests");

        _storage = new StoreAndForwardStorage(_localDb.Db, NullLogger<StoreAndForwardStorage>.Instance);

        var options = new StoreAndForwardOptions
        {
            DefaultRetryInterval = TimeSpan.Zero,
            DefaultMaxRetries = 1,
            RetryTimerInterval = TimeSpan.FromMinutes(10),
            ReplicationEnabled = true,
        };

        var replication = new ReplicationService(options, NullLogger<ReplicationService>.Instance);
        replication.SetReplicationHandler(op =>
        {
            lock (_replicated) _replicated.Add(op);
            return Task.CompletedTask;
        });

        _service = new StoreAndForwardService(
            _storage, options, NullLogger<StoreAndForwardService>.Instance, replication);
    }

    public async Task InitializeAsync() => await _storage.InitializeAsync();
    public Task DisposeAsync() => Task.CompletedTask;
    public void Dispose()
    {
        var path = _localDb.Path;
        _localDb.Dispose();
        TestLocalDb.DeleteFiles(path);
    }

    /// <summary>Replication is fire-and-forget (Task.Run); poll until the expected ops arrive.</summary>
    private async Task<List<ReplicationOperation>> WaitForReplicationAsync(int count)
    {
        for (var i = 0; i < 100; i++)
        {
            lock (_replicated)
                if (_replicated.Count >= count) return _replicated.ToList();
            await Task.Delay(20);
        }
        lock (_replicated) return _replicated.ToList();
    }

    [Fact]
    public async Task BufferingAMessage_ReplicatesAnAddOperation()
    {
        // No handler registered → message is buffered → an Add is replicated.
        var result = await _service.EnqueueAsync(
            StoreAndForwardCategory.ExternalSystem, "api", """{}""");
        Assert.True(result.WasBuffered);

        var ops = await WaitForReplicationAsync(1);
        Assert.Contains(ops, o =>
            o.OperationType == ReplicationOperationType.Add && o.MessageId == result.MessageId);
    }

    [Fact]
    public async Task SuccessfulRetry_ReplicatesARemoveOperation()
    {
        var calls = 0;
        _service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem,
            _ => ++calls == 1
                ? throw new HttpRequestException("transient")
                : Task.FromResult(true));

        var result = await _service.EnqueueAsync(
            StoreAndForwardCategory.ExternalSystem, "api", """{}""");
        await _service.RetryPendingMessagesAsync();

        var ops = await WaitForReplicationAsync(2);
        Assert.Contains(ops, o => o.OperationType == ReplicationOperationType.Add);
        Assert.Contains(ops, o =>
            o.OperationType == ReplicationOperationType.Remove && o.MessageId == result.MessageId);
    }

    [Fact]
    public async Task ParkedMessage_ReplicatesAParkOperation()
    {
        _service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem,
            _ => throw new HttpRequestException("always fails"));

        var result = await _service.EnqueueAsync(
            StoreAndForwardCategory.ExternalSystem, "api", """{}""", maxRetries: 1);
        await _service.RetryPendingMessagesAsync();

        var ops = await WaitForReplicationAsync(2);
        Assert.Contains(ops, o =>
            o.OperationType == ReplicationOperationType.Park && o.MessageId == result.MessageId);
    }

    /// <summary>
    /// StoreAndForward-016: an operator discarding a parked message must replicate
    /// a Remove so the standby's copy is also deleted (otherwise the discarded
    /// message reappears in the parked list after a failover).
    /// </summary>
    [Fact]
    public async Task DiscardingAParkedMessage_ReplicatesARemoveOperation()
    {
        _service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem,
            _ => throw new HttpRequestException("always fails"));

        var result = await _service.EnqueueAsync(
            StoreAndForwardCategory.ExternalSystem, "api", """{}""", maxRetries: 1);
        await _service.RetryPendingMessagesAsync(); // -> parked
        await WaitForReplicationAsync(2);

        var discarded = await _service.DiscardParkedMessageAsync(result.MessageId);
        Assert.True(discarded);

        var ops = await WaitForReplicationAsync(3);
        Assert.Contains(ops, o =>
            o.OperationType == ReplicationOperationType.Remove && o.MessageId == result.MessageId);
    }

    /// <summary>
    /// StoreAndForward-016: an operator retrying a parked message must replicate a
    /// Requeue so the standby's copy moves back to Pending (otherwise it stays
    /// Parked on the standby and the operator's retry is lost across a failover).
    /// </summary>
    [Fact]
    public async Task RetryingAParkedMessage_ReplicatesARequeueOperation()
    {
        _service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem,
            _ => throw new HttpRequestException("always fails"));

        var result = await _service.EnqueueAsync(
            StoreAndForwardCategory.ExternalSystem, "api", """{}""", maxRetries: 1);
        await _service.RetryPendingMessagesAsync(); // -> parked
        await WaitForReplicationAsync(2);

        var retried = await _service.RetryParkedMessageAsync(result.MessageId);
        Assert.True(retried);

        var ops = await WaitForReplicationAsync(3);
        var requeue = ops.SingleOrDefault(o =>
            o.OperationType == ReplicationOperationType.Requeue && o.MessageId == result.MessageId);
        Assert.NotNull(requeue);
        Assert.NotNull(requeue!.Message);
        Assert.Equal(StoreAndForwardMessageStatus.Pending, requeue.Message!.Status);
    }

    /// <summary>
    /// StoreAndForward-016: the standby applies a Requeue by moving its row back to
    /// Pending with retry_count = 0, mirroring the active node's local state.
    /// </summary>
    [Fact]
    public async Task ApplyReplicatedOperation_Requeue_MovesStandbyRowBackToPending()
    {
        var replication = new ReplicationService(
            new StoreAndForwardOptions { ReplicationEnabled = true },
            NullLogger<ReplicationService>.Instance);

        var parked = new StoreAndForwardMessage
        {
            Id = "requeue1",
            Category = StoreAndForwardCategory.ExternalSystem,
            Target = "api",
            PayloadJson = "{}",
            RetryCount = 5,
            MaxRetries = 1,
            RetryIntervalMs = 0,
            CreatedAt = DateTimeOffset.UtcNow,
            Status = StoreAndForwardMessageStatus.Parked,
        };
        await _storage.EnqueueAsync(parked);

        var requeued = new StoreAndForwardMessage
        {
            Id = parked.Id,
            Category = parked.Category,
            Target = parked.Target,
            PayloadJson = parked.PayloadJson,
            RetryCount = 0,
            MaxRetries = parked.MaxRetries,
            RetryIntervalMs = parked.RetryIntervalMs,
            CreatedAt = parked.CreatedAt,
            Status = StoreAndForwardMessageStatus.Pending,
        };
        await replication.ApplyReplicatedOperationAsync(
            new ReplicationOperation(ReplicationOperationType.Requeue, parked.Id, requeued),
            _storage);

        var row = await _storage.GetMessageByIdAsync(parked.Id);
        Assert.NotNull(row);
        Assert.Equal(StoreAndForwardMessageStatus.Pending, row!.Status);
        Assert.Equal(0, row.RetryCount);
    }

    private static ReplicationService NewReplicationService() =>
        new(new StoreAndForwardOptions { ReplicationEnabled = true },
            NullLogger<ReplicationService>.Instance);

    private static StoreAndForwardMessage NewMessage(string id) => new()
    {
        Id = id,
        Category = StoreAndForwardCategory.ExternalSystem,
        Target = "api",
        PayloadJson = "{}",
        RetryCount = 0,
        MaxRetries = 1,
        RetryIntervalMs = 0,
        CreatedAt = DateTimeOffset.UtcNow,
        Status = StoreAndForwardMessageStatus.Pending,
    };

    /// <summary>
    /// A Park (or Requeue) whose original Add was lost — e.g. the Add's
    /// fire-and-forget replication dropped — must still materialise the row on the
    /// standby. The full message rides in the operation, so an upsert self-heals;
    /// a blind UPDATE would affect 0 rows and the row would be gone forever.
    /// </summary>
    [Fact]
    public async Task ApplyReplicatedPark_WhenAddWasLost_MaterializesTheParkedRow()
    {
        var service = NewReplicationService();
        var msg = NewMessage("lost-add-1"); // never Added on this (standby) storage
        msg.Status = StoreAndForwardMessageStatus.Parked;

        await service.ApplyReplicatedOperationAsync(
            new ReplicationOperation(ReplicationOperationType.Park, msg.Id, msg), _storage);

        var row = await _storage.GetMessageByIdAsync("lost-add-1");
        Assert.NotNull(row); // pre-fix: blind UPDATE affected 0 rows, row is gone forever
        Assert.Equal(StoreAndForwardMessageStatus.Parked, row!.Status);
    }

    /// <summary>
    /// A duplicate Add (e.g. after Task 21's anti-entropy resync re-issues an Add
    /// the standby already holds) must not violate the PK — the upsert applies
    /// newest-wins instead of throwing.
    /// </summary>
    [Fact]
    public async Task ApplyReplicatedAdd_Twice_IsIdempotent_NewestWins()
    {
        var service = NewReplicationService();
        var msg = NewMessage("dup-add");
        await service.ApplyReplicatedOperationAsync(
            new ReplicationOperation(ReplicationOperationType.Add, msg.Id, msg), _storage);
        msg.RetryCount = 3;
        await service.ApplyReplicatedOperationAsync( // pre-fix: SqliteException PK violation
            new ReplicationOperation(ReplicationOperationType.Add, msg.Id, msg), _storage);

        Assert.Equal(3, (await _storage.GetMessageByIdAsync("dup-add"))!.RetryCount);
    }
}