ScadaBridge/src/ScadaLink.StoreAndForward/ReplicationService.cs

using Microsoft.Extensions.Logging;
using ScadaLink.Commons.Types.Enums;

namespace ScadaLink.StoreAndForward;

/// <summary>
/// WP-11: Async replication of buffer operations to standby node.
///
/// - Forwards add/remove/park operations to standby via a replication handler.
/// - No ack wait (fire-and-forget per design).
/// - Standby applies operations to its own SQLite.
/// - On failover, standby resumes delivery from its replicated state.
/// </summary>
public class ReplicationService
{
    private readonly StoreAndForwardOptions _options;
    private readonly ILogger<ReplicationService> _logger;
    private Func<ReplicationOperation, Task>? _replicationHandler;

    public ReplicationService(
        StoreAndForwardOptions options,
        ILogger<ReplicationService> logger)
    {
        _options = options;
        _logger = logger;
    }

    /// <summary>
    /// Sets the handler for forwarding replication operations to the standby node.
    /// Typically wraps Akka Tell to the standby's replication actor.
    /// </summary>
    public void SetReplicationHandler(Func<ReplicationOperation, Task> handler)
    {
        _replicationHandler = handler;
    }

    /// <summary>
    /// WP-11: Replicates an enqueue operation to standby (fire-and-forget).
    /// </summary>
    public void ReplicateEnqueue(StoreAndForwardMessage message)
    {
        if (!_options.ReplicationEnabled || _replicationHandler == null) return;

        FireAndForget(new ReplicationOperation(
            ReplicationOperationType.Add,
            message.Id,
            message));
    }

    /// <summary>
    /// WP-11: Replicates a remove operation to standby (fire-and-forget).
    /// </summary>
    public void ReplicateRemove(string messageId)
    {
        if (!_options.ReplicationEnabled || _replicationHandler == null) return;

        FireAndForget(new ReplicationOperation(
            ReplicationOperationType.Remove,
            messageId,
            null));
    }

    /// <summary>
    /// WP-11: Replicates a park operation to standby (fire-and-forget).
    /// </summary>
    public void ReplicatePark(StoreAndForwardMessage message)
    {
        if (!_options.ReplicationEnabled || _replicationHandler == null) return;

        FireAndForget(new ReplicationOperation(
            ReplicationOperationType.Park,
            message.Id,
            message));
    }

    /// <summary>
    /// WP-11 / StoreAndForward-016: Replicates an operator-initiated requeue (a parked
    /// message moved back to the pending queue) to standby (fire-and-forget). The
    /// carried message reflects the active node's post-requeue state (Pending,
    /// retry_count = 0) so the standby's copy can be brought into sync.
    /// </summary>
    public void ReplicateRequeue(StoreAndForwardMessage message)
    {
        if (!_options.ReplicationEnabled || _replicationHandler == null) return;

        FireAndForget(new ReplicationOperation(
            ReplicationOperationType.Requeue,
            message.Id,
            message));
    }

    /// <summary>
    /// WP-11: Applies a replicated operation received from the active node.
    /// Used by the standby node to keep its SQLite in sync.
    /// </summary>
    public async Task ApplyReplicatedOperationAsync(
        ReplicationOperation operation,
        StoreAndForwardStorage storage)
    {
        switch (operation.OperationType)
        {
            case ReplicationOperationType.Add when operation.Message != null:
                await storage.EnqueueAsync(operation.Message);
                break;

            case ReplicationOperationType.Remove:
                await storage.RemoveMessageAsync(operation.MessageId);
                break;

            case ReplicationOperationType.Park when operation.Message != null:
                operation.Message.Status = StoreAndForwardMessageStatus.Parked;
                await storage.UpdateMessageAsync(operation.Message);
                break;

            case ReplicationOperationType.Requeue when operation.Message != null:
                // StoreAndForward-016: an operator retried a parked message on the
                // active node; mirror that on the standby by moving its row back to
                // Pending with retry_count = 0 so a failover preserves the retry.
                operation.Message.Status = StoreAndForwardMessageStatus.Pending;
                operation.Message.RetryCount = 0;
                await storage.UpdateMessageAsync(operation.Message);
                break;
        }
    }

    private void FireAndForget(ReplicationOperation operation)
    {
        Task.Run(async () =>
        {
            try
            {
                await _replicationHandler!.Invoke(operation);
            }
            catch (Exception ex)
            {
                // WP-11: No ack wait — log and move on
                _logger.LogDebug(ex,
                    "Replication of {OpType} for message {MessageId} failed (best-effort)",
                    operation.OperationType, operation.MessageId);
            }
        });
    }
}

/// <summary>
/// WP-11: Represents a buffer operation to be replicated to standby.
/// </summary>
public record ReplicationOperation(
    ReplicationOperationType OperationType,
    string MessageId,
    StoreAndForwardMessage? Message);

/// <summary>
/// WP-11: Types of buffer operations that are replicated.
/// </summary>
public enum ReplicationOperationType
{
    Add,
    Remove,
    Park,
    /// <summary>
    /// StoreAndForward-016: an operator moved a parked message back to the pending
    /// queue. The standby resets its matching row to Pending with retry_count = 0.
    /// </summary>
    Requeue
}