ScadaBridge/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ReplicationService.cs

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

namespace ZB.MOM.WW.ScadaBridge.StoreAndForward;

/// <summary>
/// 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;

    /// <summary>Initializes a new instance of <see cref="ReplicationService"/>.</summary>
    /// <param name="options">Store-and-forward configuration options.</param>
    /// <param name="logger">Logger instance.</param>
    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>
    /// <param name="handler">The async delegate that forwards each replication operation to the standby.</param>
    public void SetReplicationHandler(Func<ReplicationOperation, Task> handler)
    {
        _replicationHandler = handler;
    }

    /// <summary>
    /// Replicates an enqueue operation to standby (fire-and-forget).
    /// </summary>
    /// <param name="message">The message that was enqueued on the active node.</param>
    public void ReplicateEnqueue(StoreAndForwardMessage message)
    {
        if (!_options.ReplicationEnabled || _replicationHandler == null) return;

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

    /// <summary>
    /// Replicates a remove operation to standby (fire-and-forget).
    /// </summary>
    /// <param name="messageId">The identifier of the message to remove from the standby buffer.</param>
    public void ReplicateRemove(string messageId)
    {
        if (!_options.ReplicationEnabled || _replicationHandler == null) return;

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

    /// <summary>
    /// Replicates a park operation to standby (fire-and-forget).
    /// </summary>
    /// <param name="message">The message that was parked on the active node.</param>
    public void ReplicatePark(StoreAndForwardMessage message)
    {
        if (!_options.ReplicationEnabled || _replicationHandler == null) return;

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

    /// <summary>
    /// 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>
    /// <param name="message">The message in its post-requeue (Pending, retry_count=0) state.</param>
    public void ReplicateRequeue(StoreAndForwardMessage message)
    {
        if (!_options.ReplicationEnabled || _replicationHandler == null) return;

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

    /// <summary>
    /// Applies a replicated operation received from the active node.
    /// Used by the standby node to keep its SQLite in sync.
    ///
    /// Add/Park/Requeue are applied as <b>upserts</b> (<see cref="StoreAndForwardStorage.UpsertMessageAsync"/>),
    /// not blind INSERT/UPDATE: the full message rides in every one of those operations,
    /// so a Park/Requeue whose original Add was lost (fire-and-forget replication is
    /// best-effort) self-heals by materialising the row, and a duplicate Add (e.g.
    /// re-issued by an anti-entropy resync) applies newest-wins instead of throwing a
    /// primary-key violation. Remove is a plain delete.
    /// </summary>
    /// <param name="operation">The replication operation to apply.</param>
    /// <param name="storage">The standby node's store-and-forward storage to update.</param>
    /// <returns>A task representing the asynchronous apply operation.</returns>
    public async Task ApplyReplicatedOperationAsync(
        ReplicationOperation operation,
        StoreAndForwardStorage storage)
    {
        switch (operation.OperationType)
        {
            case ReplicationOperationType.Add when operation.Message != null:
                await storage.UpsertMessageAsync(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.UpsertMessageAsync(operation.Message);
                break;

            case ReplicationOperationType.Requeue when operation.Message != null:
                operation.Message.Status = StoreAndForwardMessageStatus.Pending;
                operation.Message.RetryCount = 0;
                await storage.UpsertMessageAsync(operation.Message);
                break;
        }
    }

    private void FireAndForget(ReplicationOperation operation)
    {
        // Invoked inline, NOT via Task.Run: the handler is a non-blocking Akka
        // Tell, and thread-pool hand-off destroyed Add/Remove ordering for the
        // same message id (arch review 02). Inline invocation preserves issue
        // order; Akka's per-sender/receiver guarantee preserves it on the wire.
        try
        {
            var task = _replicationHandler!.Invoke(operation);
            if (!task.IsCompletedSuccessfully)
            {
                task.ContinueWith(t =>
                    {
                        ScadaBridgeTelemetry.RecordReplicationFailure();
                        _logger.LogWarning(t.Exception,
                            "Replication of {OpType} for message {MessageId} failed (best-effort); standby buffer may be diverging",
                            operation.OperationType, operation.MessageId);
                    },
                    TaskContinuationOptions.OnlyOnFaulted);
            }
        }
        catch (Exception ex)
        {
            ScadaBridgeTelemetry.RecordReplicationFailure();
            _logger.LogWarning(ex,
                "Replication of {OpType} for message {MessageId} failed (best-effort); standby buffer may be diverging",
                operation.OperationType, operation.MessageId);
        }
    }
}

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

/// <summary>
/// Types of buffer operations that are replicated.
/// </summary>
public enum ReplicationOperationType
{
    Add,
    Remove,
    Park,
    /// <summary>
    /// 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
}