lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/AlarmWriterTests.cs

using Npgsql;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;

namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests;

[Collection(QuestDbCollection.Name)]
public sealed class AlarmWriterTests(QuestDbFixture fx)
{
    // µs-aligned: QuestDB stores µs, so the ts column reads back exactly.
    private static readonly DateTime T = new DateTime(2026, 10, 9, 1, 2, 3, DateTimeKind.Utc).AddTicks(1230);

    private (QuestDbAlarmHistorianWriter W, QuestDbConnection C) Make(string? connection = null, bool auto = true)
    {
        var o = fx.Options(fx.UniquePrefix(), autoCreate: auto);
        if (connection is not null) o.ConnectionString = connection;
        var c = new QuestDbConnection(o);
        return (new QuestDbAlarmHistorianWriter(c, new QuestDbSchemaBootstrapper(c, o), new QuestDbWalVisibility(c, o.VisibilityTimeout)), c);
    }

    private static AlarmHistorianEvent Ev(string kind = "Activated", string type = "LimitAlarm", string? comment = "c", DateTime? at = null) =>
        new("alarm-1", "Plant/Line/Eq", "HiTemp", type, AlarmSeverity.High, kind, "msg", "user", comment, at ?? T);

    private static async Task<long> CountAsync(QuestDbConnection c, string? kind = null)
    {
        await using var conn = await c.OpenAsync(default);
        await using var cmd = c.CreateCommand(conn, $"SELECT count() FROM {c.EventTable}" + (kind is null ? "" : " WHERE event_kind = $1"));
        if (kind is not null) QuestDbConnection.BindText(cmd, kind);
        return Convert.ToInt64(await cmd.ExecuteScalarAsync());
    }

    [Fact]
    public async Task A_batch_is_one_insert_per_chunk_and_a_redelivery_converges()
    {
        fx.SkipIfUnavailable();
        var (w, c) = Make();
        var events = Enumerable.Range(0, 1200).Select(i => Ev(kind: $"k{i}", at: T.AddSeconds(i))).ToList();
        events.Insert(3, new AlarmHistorianEvent("alarm-1", "Plant/Line/Eq", "HiTemp", "LimitAlarm", AlarmSeverity.High,
            "old", "msg", "user", null, new DateTime(1969, 12, 31, 0, 0, 0, DateTimeKind.Utc))); // pre-1970: never sent

        var r = await w.WriteBatchAsync(events, CancellationToken.None);
        r[3].ShouldBe(HistorianWriteOutcome.PermanentFail);
        r.Where((_, i) => i != 3).ShouldAllBe(o => o == HistorianWriteOutcome.Ack);
        w.Statements.ShouldBe(3, "1,200 events = 500 + 500 + 200, not 1,200 round trips");
        (await CountAsync(c)).ShouldBe(1200);

        (await w.WriteBatchAsync(events.Where((_, i) => i != 3).ToList(), CancellationToken.None))
            .ShouldAllBe(o => o == HistorianWriteOutcome.Ack);
        (await CountAsync(c)).ShouldBe(1200, "DEDUP UPSERT KEYS(ts, event_id) converges a redelivered batch");
    }

    [Fact]
    public async Task Event_round_trips_and_a_repeat_is_one_row_with_every_write_Ack()
    {
        fx.SkipIfUnavailable();
        var (w, c) = Make();
        (await w.WriteBatchAsync([Ev(), Ev()], CancellationToken.None)).ShouldBe([HistorianWriteOutcome.Ack, HistorianWriteOutcome.Ack]);
        (await w.WriteBatchAsync([Ev()], CancellationToken.None)).ShouldBe([HistorianWriteOutcome.Ack]);
        (await w.WriteBatchAsync([Ev(kind: "Cleared")], CancellationToken.None)).ShouldBe([HistorianWriteOutcome.Ack]);

        await using var conn = await c.OpenAsync(default);
        await using var cmd = c.CreateCommand(conn,
            $"SELECT event_id, source, alarm_id, alarm_name, alarm_type_name, severity, message, user_name, comment, ts, received_utc FROM {c.EventTable} WHERE event_kind = 'Activated'");
        await using var r = await cmd.ExecuteReaderAsync();
        (await r.ReadAsync()).ShouldBeTrue();
        r.GetString(0).ShouldBe(AlarmEventIds.ComputeEventId(Ev()));
        r.GetString(1).ShouldBe("Plant/Line/Eq");
        r.GetString(2).ShouldBe("alarm-1");
        r.GetString(3).ShouldBe("HiTemp");
        r.GetString(4).ShouldBe("LimitAlarm");
        r.GetInt32(5).ShouldBe(AlarmEventIds.ToNumericSeverity(AlarmSeverity.High));
        r.GetInt32(5).ShouldBe(700);
        r.GetString(6).ShouldBe("msg");
        r.GetString(7).ShouldBe("user");
        r.GetString(8).ShouldBe("c");
        QuestDbConnection.ReadUtc(r, 9).ShouldBe(T);
        (DateTime.UtcNow - QuestDbConnection.ReadUtc(r, 10)).ShouldBeLessThan(TimeSpan.FromMinutes(1));
        (await r.ReadAsync()).ShouldBeFalse();
        await r.CloseAsync();
        (await CountAsync(c)).ShouldBe(2);
    }

    [Fact]
    public async Task A_null_comment_and_a_very_long_type_name_are_both_stored_and_Ack()
    {
        fx.SkipIfUnavailable();
        var (w, c) = Make();
        var r = await w.WriteBatchAsync([Ev(kind: "A"), Ev(kind: "B", type: new string('t', 500)), Ev(kind: "C", comment: null)], CancellationToken.None);
        r.ShouldBe([HistorianWriteOutcome.Ack, HistorianWriteOutcome.Ack, HistorianWriteOutcome.Ack]);
        (await CountAsync(c)).ShouldBe(3);
    }

    [Fact]
    public async Task An_event_before_1970_is_PermanentFail_and_siblings_still_Ack()
    {
        fx.SkipIfUnavailable();
        var (w, c) = Make();
        var r = await w.WriteBatchAsync([Ev(kind: "A"), Ev(kind: "B", at: new DateTime(1960, 1, 1, 0, 0, 0, DateTimeKind.Utc)), Ev(kind: "C")], CancellationToken.None);
        r.ShouldBe([HistorianWriteOutcome.Ack, HistorianWriteOutcome.PermanentFail, HistorianWriteOutcome.Ack]);
        (await CountAsync(c)).ShouldBe(2);
    }

    [Fact]
    public async Task Concurrent_writers_of_the_same_event_all_Ack_with_one_row()
    {
        fx.SkipIfUnavailable();
        var (w, c) = Make();
        await w.WriteBatchAsync([Ev(kind: "warm")], CancellationToken.None);
        var rs = await Task.WhenAll(Enumerable.Range(0, 12).Select(_ => Task.Run(() => w.WriteBatchAsync([Ev()], CancellationToken.None))));
        rs.ShouldAllBe(x => x[0] == HistorianWriteOutcome.Ack);
        (await CountAsync(c, "Activated")).ShouldBe(1);
    }

    [Fact]
    public async Task Unreachable_database_is_RetryPlease_not_an_exception_and_empty_batch_is_empty()
    {
        fx.SkipIfUnavailable();
        var (w, _) = Make("Host=localhost;Port=1;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading;Timeout=2", auto: false);
        (await w.WriteBatchAsync([Ev()], CancellationToken.None)).ShouldBe([HistorianWriteOutcome.RetryPlease]);
        (await w.WriteBatchAsync([], CancellationToken.None)).ShouldBeEmpty();
    }

    [Fact]
    public void Classify_maps_unknown_and_transport_errors_to_RetryPlease_and_data_rejections_to_PermanentFail()
    {
        QuestDbAlarmHistorianWriter.Classify(new InvalidOperationException("boom")).ShouldBe(HistorianWriteOutcome.RetryPlease);
        QuestDbAlarmHistorianWriter.Classify(new IOException("reset")).ShouldBe(HistorianWriteOutcome.RetryPlease);
        QuestDbAlarmHistorianWriter.Classify(new TimeoutException()).ShouldBe(HistorianWriteOutcome.RetryPlease);
        QuestDbAlarmHistorianWriter.Classify(new PostgresException("table is busy", "ERROR", "ERROR", "00000")).ShouldBe(HistorianWriteOutcome.RetryPlease);
        QuestDbAlarmHistorianWriter.Classify(new PostgresException("designated timestamp before 1970-01-01 is not allowed", "ERROR", "ERROR", "00000"))
            .ShouldBe(HistorianWriteOutcome.PermanentFail);
        QuestDbAlarmHistorianWriter.Classify(new PostgresException("inconvertible value: x [VARCHAR -> INT]", "ERROR", "ERROR", "00000"))
            .ShouldBe(HistorianWriteOutcome.PermanentFail);
    }
}