lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/LocalDb/AlarmSfConvergenceTests.cs

using Serilog;
using Shouldly;
using Xunit;
using ZB.MOM.WW.LocalDb;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;

namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests.LocalDb;

/// <summary>
///     LocalDb Phase 2 — two driver nodes replicating the alarm store-and-forward buffer over a real
///     loopback h2c transport, through the real fail-closed interceptor.
/// </summary>
/// <remarks>
///     <para>
///         This is the test the phase exists to pass. Everything upstream — the schema, the
///         registration, the sink rewire, the drain gate — can be individually green while a
///         redundant pair still fails to share the undelivered alarm history that is the whole
///         point of moving the queue here.
///     </para>
///     <para>
///         Rows are written through the production <see cref="LocalDbStoreAndForwardSink"/> rather
///         than by hand-rolled INSERTs. The id derivation, the column set and the delete-on-ack
///         semantics under test are then the ones production actually uses.
///     </para>
/// </remarks>
[Collection("LocalDbPairConvergence")]
public sealed class AlarmSfConvergenceTests
{
    private static readonly Serilog.ILogger Log = new LoggerConfiguration().CreateLogger();

    /// <summary>A writer that acknowledges everything — the drain's happy path.</summary>
    private sealed class AckWriter : IAlarmHistorianWriter
    {
        public Task<IReadOnlyList<HistorianWriteOutcome>> WriteBatchAsync(
            IReadOnlyList<AlarmHistorianEvent> batch, CancellationToken cancellationToken) =>
            Task.FromResult<IReadOnlyList<HistorianWriteOutcome>>(
                batch.Select(_ => HistorianWriteOutcome.Ack).ToList());
    }

    /// <summary>A writer that never acknowledges — a historian outage, which is when the buffer matters.</summary>
    private sealed class OutageWriter : IAlarmHistorianWriter
    {
        public List<AlarmHistorianEvent> Events { get; } = [];
        public Task<IReadOnlyList<HistorianWriteOutcome>> WriteBatchAsync(
            IReadOnlyList<AlarmHistorianEvent> batch, CancellationToken cancellationToken)
        {
            Events.AddRange(batch);
            return Task.FromResult<IReadOnlyList<HistorianWriteOutcome>>(
                batch.Select(_ => HistorianWriteOutcome.RetryPlease).ToList());
        }
    }

    private static AlarmHistorianEvent Event(string alarmId, int second) => new(
        AlarmId: alarmId,
        EquipmentPath: "line-1/eq-1",
        AlarmName: "temp-high",
        AlarmTypeName: "LimitAlarm",
        Severity: AlarmSeverity.High,
        EventKind: "Activated",
        Message: "temperature high",
        User: "system",
        Comment: null,
        TimestampUtc: new DateTime(2026, 7, 21, 0, 0, second, DateTimeKind.Utc));

    /// <summary>The set of row ids on a node, ordered — equal sets on both nodes means converged.</summary>
    private static async Task<string> IdsAsync(ILocalDb db)
    {
        var rows = await db.QueryAsync(
            "SELECT id FROM alarm_sf_events ORDER BY id",
            static r => r.GetString(0));
        return string.Join(",", rows);
    }

    private static async Task<long> CountAsync(ILocalDb db)
    {
        var rows = await db.QueryAsync(
            "SELECT COUNT(*) FROM alarm_sf_events", static r => r.GetInt64(0));
        return rows[0];
    }

    [Fact]
    public async Task AlarmBurstOnA_ConvergesToB_AndOplogDrains()
    {
        await using var pair = new LocalDbPairHarness();
        await pair.StartAsync();

        // The Primary's sink. Its drain is left unstarted so the burst stays queued and observable;
        // an outage is exactly the state in which this buffer has to survive a node loss.
        using var sink = new LocalDbStoreAndForwardSink(pair.A, new OutageWriter(), Log);
        for (var i = 0; i < 25; i++)
            await sink.EnqueueAsync(Event($"eq/alarm-{i}", i), TestContext.Current.CancellationToken);

        await LocalDbPairHarness.WaitUntilAsync(
            async () => await CountAsync(pair.B) == 25,
            "node B to hold all 25 buffered alarm events");

        (await IdsAsync(pair.B)).ShouldBe(await IdsAsync(pair.A));

        // Oplog drains to empty once B acks: the steady state of a converged pair, and the proof
        // the rows were genuinely acknowledged rather than merely sent.
        await LocalDbPairHarness.WaitUntilAsync(
            async () => await LocalDbPairHarness.OplogDepthAsync(pair.A) == 0,
            "node A's oplog to drain after node B acknowledges the burst");
    }

    [Fact]
    public async Task DeliveredRowsAreRemovedFromBothNodes()
    {
        // The no-redeliver-after-failover property, asserted at the data layer. An acknowledged row
        // is DELETEd, and the delete has to reach the peer as a tombstone — otherwise a promoted
        // standby would find the Primary's already-delivered events still queued and send them again.
        await using var pair = new LocalDbPairHarness();
        await pair.StartAsync();

        using var buffering = new LocalDbStoreAndForwardSink(pair.A, new OutageWriter(), Log);
        for (var i = 0; i < 5; i++)
            await buffering.EnqueueAsync(Event($"eq/alarm-{i}", i), TestContext.Current.CancellationToken);

        await LocalDbPairHarness.WaitUntilAsync(
            async () => await CountAsync(pair.B) == 5,
            "node B to receive the buffered events");

        // The historian comes back and the Primary drains.
        using var draining = new LocalDbStoreAndForwardSink(pair.A, new AckWriter(), Log);
        await draining.DrainOnceAsync(TestContext.Current.CancellationToken);
        (await CountAsync(pair.A)).ShouldBe(0);

        await LocalDbPairHarness.WaitUntilAsync(
            async () => await CountAsync(pair.B) == 0,
            "node B to drop its copies once node A delivered them");

        (await LocalDbPairHarness.TombstoneCountAsync(pair.B, "alarm_sf_events"))
            .ShouldBe(5, "the deletes must arrive as tombstones, not vanish silently");
    }

    [Fact]
    public async Task WritesWhileTransportDown_SurviveRejoin()
    {
        // The realistic outage shape: alarms keep arriving on the Primary while the pair link is
        // down. Nothing may be lost when it comes back.
        await using var pair = new LocalDbPairHarness();
        await pair.StartAsync();

        using var sink = new LocalDbStoreAndForwardSink(pair.A, new OutageWriter(), Log);
        await sink.EnqueueAsync(Event("eq/before", 1), TestContext.Current.CancellationToken);

        await LocalDbPairHarness.WaitUntilAsync(
            async () => await CountAsync(pair.B) == 1,
            "the pre-outage event to reach node B");

        await pair.StopPassiveAsync();

        for (var i = 0; i < 10; i++)
            await sink.EnqueueAsync(Event($"eq/during-{i}", 10 + i), TestContext.Current.CancellationToken);

        await pair.RestartPairAsync();

        await LocalDbPairHarness.WaitUntilAsync(
            async () => await CountAsync(pair.B) == 11,
            "every event buffered during the outage to reach node B after the rejoin");

        (await IdsAsync(pair.B)).ShouldBe(await IdsAsync(pair.A));
    }

    [Fact]
    public async Task TheSameEventOnBothNodes_ConvergesToOneRow()
    {
        // Both adapters default-write while their redundancy role is unknown, so in the boot window
        // the pair genuinely double-accepts the same fanned transition. Payload-derived ids are what
        // turn that into one row instead of two; random GUIDs would leave the duplicate in the
        // buffer for the eventual Primary to deliver twice.
        await using var pair = new LocalDbPairHarness();
        await pair.StartAsync();

        var shared = Event("eq/shared", 1);
        var sharedOnB = Event("eq/shared", 1);
        shared.DeliveryId.ShouldNotBe(sharedOnB.DeliveryId);
        var writerA = new OutageWriter();
        var writerB = new OutageWriter();
        using var sinkA = new LocalDbStoreAndForwardSink(pair.A, writerA, Log);
        using var sinkB = new LocalDbStoreAndForwardSink(pair.B, writerB, Log);

        await sinkA.EnqueueAsync(shared, TestContext.Current.CancellationToken);
        await sinkB.EnqueueAsync(sharedOnB, TestContext.Current.CancellationToken);

        // A second, DISTINCT event on B. Without it the scenario would pass with replication
        // switched off entirely — each node would hold its own single copy and "count == 1 on
        // both" would be satisfied by two unconverged databases.
        await sinkB.EnqueueAsync(Event("eq/only-on-b", 2), TestContext.Current.CancellationToken);

        await LocalDbPairHarness.WaitUntilAsync(
            async () => await CountAsync(pair.A) == 2 && await CountAsync(pair.B) == 2,
            "both nodes to hold two rows: the doubly-accepted event collapsed, plus B's own");

        (await IdsAsync(pair.A)).ShouldBe(await IdsAsync(pair.B));

        // Held, not merely reached: a late-arriving replica of the peer's copy of the shared event
        // must not add a third row a moment later.
        (await LocalDbPairHarness.HeldWithinAsync(
                async () => await CountAsync(pair.A) > 2 || await CountAsync(pair.B) > 2,
                TimeSpan.FromSeconds(3)))
            .ShouldBeFalse("the duplicate must stay collapsed");

        // Either node can retry after promotion, with the same GUID despite independent input identities.
        await sinkA.DrainOnceAsync(TestContext.Current.CancellationToken);
        await sinkB.DrainOnceAsync(TestContext.Current.CancellationToken);
        var deliveryA = writerA.Events.Single(e => e.AlarmId == "eq/shared").DeliveryId;
        var deliveryB = writerB.Events.Single(e => e.AlarmId == "eq/shared").DeliveryId;
        deliveryA.ShouldNotBe(Guid.Empty);
        deliveryB.ShouldBe(deliveryA);
    }
}