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

using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Shouldly;
using Xunit;
using ZB.MOM.WW.LocalDb;
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
using ZB.MOM.WW.OtOpcUa.Host.Configuration;

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

/// <summary>
///     The one-time copy of the pre-consolidation <c>alarm-historian.db</c> queue into the
///     consolidated LocalDb.
/// </summary>
/// <remarks>
///     <para>
///         What is being protected is undelivered alarm history: rows that are in the legacy file
///         precisely because the historian could not be reached. Dropping them on upgrade would
///         discard exactly the audit trail the store-and-forward queue exists to preserve.
///     </para>
///     <para>
///         Run against a real temp-file <see cref="ILocalDb"/> built through the production
///         <see cref="LocalDbSetup.OnReady"/>, so the registration order the migrator depends on is
///         the one the host actually runs — a hand-written schema here would prove only that the
///         test agrees with itself.
///     </para>
/// </remarks>
public sealed class AlarmSfLegacyMigratorTests : IDisposable
{
    private readonly string _dbPath =
        Path.Combine(Path.GetTempPath(), $"otopcua-alarmsf-mig-{Guid.NewGuid():N}.db");

    private readonly string _legacyPath =
        Path.Combine(Path.GetTempPath(), $"otopcua-alarmsf-legacy-{Guid.NewGuid():N}.db");

    private ServiceProvider? _provider;

    private sealed class CapturingRetryWriter : 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).ToArray());
        }
    }

    [Fact]
    public async Task A_legacy_payload_keeps_its_key_and_stable_delivery_identity_across_drains()
    {
        var payload = Payload("eq/a");
        SeedLegacy((1, "eq/a", payload, 0, dead: 0));
        var db = BuildDb();
        AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());
        var row = (await db.QueryAsync("SELECT id, payload_json FROM alarm_sf_events",
            r => (Id: r.GetString(0), Payload: r.GetString(1)), null, TestContext.Current.CancellationToken)).Single();
        row.Id.ShouldBe("B1D80B7207C3F3B6CB31105DBF1A6E0D6FE300DA1B3F877AB746B36535310A60");
        row.Payload.ShouldBe(payload);
        var writer = new CapturingRetryWriter();
        using var log = new Serilog.LoggerConfiguration().CreateLogger();
        using (var sink = new LocalDbStoreAndForwardSink(db, writer, log))
            await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
        using (var sink = new LocalDbStoreAndForwardSink(db, writer, log))
            await sink.DrainOnceAsync(TestContext.Current.CancellationToken);
        writer.Events.Count.ShouldBe(2);
        writer.Events.ShouldAllBe(evt => evt.DeliveryId == Guid.Parse("55372883-eaec-8dd7-a1a4-6f98f416a272"));
    }

    // ---- the cases ---------------------------------------------------------------------------

    [Fact]
    public async Task Every_legacy_row_lands_in_the_consolidated_table()
    {
        SeedLegacy(
            (1, "eq/a", Payload("eq/a"), 0, dead: 0),
            (2, "eq/b", Payload("eq/b"), 3, dead: 0),
            (3, "eq/c", Payload("eq/c"), 9, dead: 1));

        var db = BuildDb();
        AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());

        (await CountAsync(db)).ShouldBe(3);
        // Attempt counts and the dead-letter flag must survive: a migrated row that lost its
        // attempt count would restart its retry budget, and one that lost its dead-letter flag
        // would be re-sent to a historian that already refused it permanently.
        var rows = await db.QueryAsync(
            "SELECT alarm_id, attempt_count, dead_lettered FROM alarm_sf_events ORDER BY alarm_id",
            r => (AlarmId: r.GetString(0), Attempts: r.GetInt64(1), Dead: r.GetInt64(2)),
            parameters: null,
            TestContext.Current.CancellationToken);

        rows.ShouldBe([("eq/a", 0L, 0L), ("eq/b", 3L, 0L), ("eq/c", 9L, 1L)]);
    }

    [Fact]
    public async Task Migrated_rows_enter_the_oplog()
    {
        // THE ordering pin. Capture is trigger-based, so rows copied in before
        // RegisterReplicated would never enter the oplog and would never reach the peer —
        // silently, and permanently. A migrator invoked too early passes every other case here.
        SeedLegacy((1, "eq/a", Payload("eq/a"), 0, dead: 0));

        var db = BuildDb();
        AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());

        var oplog = await db.QueryAsync(
            "SELECT COUNT(*) FROM __localdb_oplog WHERE table_name = 'alarm_sf_events'",
            r => r.GetInt64(0),
            parameters: null,
            TestContext.Current.CancellationToken);

        oplog[0].ShouldBe(1);
    }

    [Fact]
    public async Task A_second_run_is_a_no_op()
    {
        SeedLegacy((1, "eq/a", Payload("eq/a"), 0, dead: 0));
        var db = BuildDb();
        var config = ConfigWithLegacyPath();

        AlarmSfLegacyMigrator.Migrate(db, config);
        AlarmSfLegacyMigrator.Migrate(db, config);

        (await CountAsync(db)).ShouldBe(1);
        File.Exists(_legacyPath).ShouldBeFalse("the legacy file is renamed once the copy commits");
        File.Exists(_legacyPath + ".migrated").ShouldBeTrue();
    }

    [Fact]
    public async Task Re_migrating_the_same_rows_cannot_duplicate_them()
    {
        // The sidecar rename is the normal guard, but it is not the only one that has to hold:
        // a crash between commit and rename leaves the file in place, and the next boot copies it
        // again. Ids are derived from the payload, so the re-copy collides with itself and is
        // ignored rather than doubling the queue.
        SeedLegacy((1, "eq/a", Payload("eq/a"), 0, dead: 0));
        var db = BuildDb();
        AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());

        File.Move(_legacyPath + ".migrated", _legacyPath);   // simulate the crash-before-rename
        AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());

        (await CountAsync(db)).ShouldBe(1);
    }

    [Fact]
    public async Task An_older_legacy_file_missing_a_column_still_migrates()
    {
        // A file written by an older build predates LastError. Naming a missing column in the
        // SELECT throws "no such column", which would discard every row in the table rather than
        // the one field. Intersecting the column list first means the data migrates and the
        // absent column keeps its schema default.
        SeedLegacyWithoutLastError((1, "eq/a", Payload("eq/a")));

        var db = BuildDb();
        AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());

        (await CountAsync(db)).ShouldBe(1);
    }

    [Fact]
    public async Task A_failed_copy_leaves_the_legacy_file_untouched()
    {
        // The legacy table exists but its payload column is missing, so the required-column guard
        // rejects the file. Nothing is copied and nothing is renamed: the operator still has the
        // original to recover from, which is the whole point of renaming only after commit.
        SeedUnrecognisedLegacy();

        var db = BuildDb();
        AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());

        (await CountAsync(db)).ShouldBe(0);
        File.Exists(_legacyPath).ShouldBeTrue("an un-copied file must not be renamed away");
        File.Exists(_legacyPath + ".migrated").ShouldBeFalse();
    }

    [Fact]
    public async Task A_missing_legacy_file_is_a_clean_no_op()
    {
        // The expected case on a fresh install and on any node that never enabled the historian.
        var db = BuildDb();
        AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());

        (await CountAsync(db)).ShouldBe(0);
    }

    [Theory]
    [InlineData(":memory:")]
    [InlineData("file:thing?mode=memory")]
    [InlineData("")]
    public async Task Non_file_legacy_sources_are_no_ops(string configured)
    {
        var db = BuildDb();
        AlarmSfLegacyMigrator.Migrate(db, Config(configured));

        (await CountAsync(db)).ShouldBe(0);
    }

    [Fact]
    public async Task Two_nodes_migrating_the_same_event_converge_on_one_row()
    {
        // A warm pair's two legacy files overlap: HistorianAdapterActor default-writes while the
        // redundancy role is unknown, so both nodes accepted the same transitions during every
        // boot window. Node-prefixed ids would carry that duplication forward into the merged
        // buffer permanently; payload-derived ids collapse it.
        var shared = Payload("eq/shared");
        SeedLegacy((1, "eq/shared", shared, 0, dead: 0));

        var db = BuildDb();
        AlarmSfLegacyMigrator.Migrate(db, ConfigWithLegacyPath());

        // The peer's file: a different legacy RowId for the very same event.
        var peerLegacy = _legacyPath + ".peer";
        SeedLegacyAt(peerLegacy, (77, "eq/shared", shared, 0, dead: 0));
        AlarmSfLegacyMigrator.Migrate(db, Config(peerLegacy));

        (await CountAsync(db)).ShouldBe(1);

        try { File.Delete(peerLegacy + ".migrated"); } catch { /* best effort */ }
    }

    // ---- fixture -----------------------------------------------------------------------------

    private ILocalDb BuildDb()
    {
        // The sync ApiKey is what makes OnReady register the tables for capture at all — since
        // 2026-08-18 registration follows the replication config rather than being unconditional
        // (see LocalDbSetup.ReplicationIsConfigured). Without it these databases would have no
        // capture triggers and Migrated_rows_enter_the_oplog would pin nothing. A passive node
        // carries exactly this and no PeerAddress, so it is also the realistic shape.
        var configuration = new ConfigurationBuilder()
            .AddInMemoryCollection(new Dictionary<string, string?>
            {
                ["LocalDb:Path"] = _dbPath,
                ["LocalDb:Replication:ApiKey"] = "test-sync-key",
            })
            .Build();

        _provider = new ServiceCollection()
            .AddZbLocalDb(configuration, db => LocalDbSetup.OnReady(db, configuration))
            .BuildServiceProvider();

        return _provider.GetRequiredService<ILocalDb>();
    }

    private IConfiguration ConfigWithLegacyPath() => Config(_legacyPath);

    private static IConfiguration Config(string legacyPath) =>
        new ConfigurationBuilder()
            .AddInMemoryCollection(new Dictionary<string, string?>
            {
                ["AlarmHistorian:DatabasePath"] = legacyPath,
            })
            .Build();

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

    /// <summary>A payload shaped like a serialized <c>AlarmHistorianEvent</c>.</summary>
    private static string Payload(string alarmId) =>
        $$"""{"AlarmId":"{{alarmId}}","EquipmentPath":"line/eq","AlarmName":"hi","TimestampUtc":"2026-07-20T00:00:00Z"}""";

    private void SeedLegacy(params (long RowId, string AlarmId, string Payload, long Attempts, long dead)[] rows) =>
        SeedLegacyAt(_legacyPath, rows);

    /// <summary>Creates a legacy queue file with the full pre-cutover schema and the given rows.</summary>
    private static void SeedLegacyAt(
        string path, params (long RowId, string AlarmId, string Payload, long Attempts, long dead)[] rows)
    {
        using var conn = Open(path);
        Exec(conn, """
            CREATE TABLE Queue (
                RowId INTEGER PRIMARY KEY AUTOINCREMENT,
                AlarmId TEXT NOT NULL,
                EnqueuedUtc TEXT NOT NULL,
                PayloadJson TEXT NOT NULL,
                AttemptCount INTEGER NOT NULL DEFAULT 0,
                LastAttemptUtc TEXT NULL,
                LastError TEXT NULL,
                DeadLettered INTEGER NOT NULL DEFAULT 0
            );
            """);

        foreach (var row in rows)
        {
            using var cmd = conn.CreateCommand();
            cmd.CommandText = """
                INSERT INTO Queue (RowId, AlarmId, EnqueuedUtc, PayloadJson, AttemptCount, DeadLettered)
                VALUES ($id, $alarm, '2026-07-20T00:00:00.0000000Z', $payload, $attempts, $dead)
                """;
            cmd.Parameters.AddWithValue("$id", row.RowId);
            cmd.Parameters.AddWithValue("$alarm", row.AlarmId);
            cmd.Parameters.AddWithValue("$payload", row.Payload);
            cmd.Parameters.AddWithValue("$attempts", row.Attempts);
            cmd.Parameters.AddWithValue("$dead", row.dead);
            cmd.ExecuteNonQuery();
        }
    }

    /// <summary>A legacy file from an older build, predating the <c>LastError</c> column.</summary>
    private void SeedLegacyWithoutLastError(params (long RowId, string AlarmId, string Payload)[] rows)
    {
        using var conn = Open(_legacyPath);
        Exec(conn, """
            CREATE TABLE Queue (
                RowId INTEGER PRIMARY KEY AUTOINCREMENT,
                AlarmId TEXT NOT NULL,
                EnqueuedUtc TEXT NOT NULL,
                PayloadJson TEXT NOT NULL,
                AttemptCount INTEGER NOT NULL DEFAULT 0,
                DeadLettered INTEGER NOT NULL DEFAULT 0
            );
            """);

        foreach (var row in rows)
        {
            using var cmd = conn.CreateCommand();
            cmd.CommandText = """
                INSERT INTO Queue (RowId, AlarmId, EnqueuedUtc, PayloadJson)
                VALUES ($id, $alarm, '2026-07-20T00:00:00.0000000Z', $payload)
                """;
            cmd.Parameters.AddWithValue("$id", row.RowId);
            cmd.Parameters.AddWithValue("$alarm", row.AlarmId);
            cmd.Parameters.AddWithValue("$payload", row.Payload);
            cmd.ExecuteNonQuery();
        }
    }

    /// <summary>A file with a <c>Queue</c> table this build cannot read — no payload column.</summary>
    private void SeedUnrecognisedLegacy()
    {
        using var conn = Open(_legacyPath);
        Exec(conn, "CREATE TABLE Queue (RowId INTEGER PRIMARY KEY AUTOINCREMENT, Something TEXT);");
        Exec(conn, "INSERT INTO Queue (Something) VALUES ('x');");
    }

    private static SqliteConnection Open(string path)
    {
        var conn = new SqliteConnection(new SqliteConnectionStringBuilder { DataSource = path }.ToString());
        conn.Open();
        return conn;
    }

    private static void Exec(SqliteConnection conn, string sql)
    {
        using var cmd = conn.CreateCommand();
        cmd.CommandText = sql;
        cmd.ExecuteNonQuery();
    }

    public void Dispose()
    {
        _provider?.Dispose();
        SqliteConnection.ClearAllPools();

        foreach (var baseName in new[] { _dbPath, _legacyPath, _legacyPath + ".migrated" })
        {
            foreach (var suffix in new[] { "", "-wal", "-shm" })
            {
                try { File.Delete(baseName + suffix); } catch { /* best effort */ }
            }
        }
    }
}