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

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

/// <summary>
///     The QuestDB modification trail's applied marker and dedup key. QuestDB has no multi-table transaction, so the trail
///     row is written BEFORE its change: a failed change left an orphan row that modified reads served as a change that
///     happened, and a trail INSERT re-sent after a client timeout duplicated. Now the row is written <c>pending</c>,
///     marked applied after the change, keyed <c>(ts, tag, modified_id)</c>; a modified read drops a pending row unless the
///     sample (latest row) or the next row's recorded prior value (any earlier row, lmxopcua#778) shows the change landed.
/// </summary>
[Collection(QuestDbCollection.Name)]
public sealed class TrailMarkerTests(QuestDbFixture fx)
{
    private static readonly DateTime T0 = new(2026, 3, 1, 12, 0, 0, DateTimeKind.Utc);
    private static readonly CancellationToken Ct = CancellationToken.None;

    private static QuestDbHistorianDataSource Source(UpdateTests.Rig rig) => new(rig.Conn, rig.Tags.Bootstrap, rig.Tags);

    private static async Task<IReadOnlyList<HistoryModifiedValue>> ModifiedAsync(UpdateTests.Rig rig, string tag) =>
        (await Source(rig).ReadModifiedAsync(tag, T0.AddMinutes(-1), T0.AddMinutes(10), 0, null, Ct)).Values;

    private static async Task<long> TrailRowsAsync(UpdateTests.Rig rig, string tag)
    {
        await using var conn = await rig.Conn.OpenAsync(Ct);
        await using var cmd = rig.Conn.CreateCommand(conn, $"SELECT count() FROM {rig.Conn.SampleModifiedTable} WHERE tag = $1");
        QuestDbConnection.BindText(cmd, tag);
        return Convert.ToInt64(await cmd.ExecuteScalarAsync());
    }

    [Fact]
    public async Task A_new_trail_table_has_the_marker_format_and_an_applied_change_reads_back()
    {
        fx.SkipIfUnavailable();
        var rig = UpdateTests.MakeRig(fx);
        (await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [UpdateTests.V(T0, 1)], UpdateTests.Who, Ct))
            .ShouldBe([HistorianStatusCodes.GoodEntryInserted]);
        rig.Tags.Bootstrap.SupportsTrailMarker.ShouldBeTrue();

        var trail = await ModifiedAsync(rig, "t");
        trail.ShouldHaveSingleItem().UpdateType.ShouldBe(HistoryModificationType.Insert);
        (await TrailRowsAsync(rig, "t")).ShouldBe(1, "the applied marker re-upserts the row, it does not add one");
    }

    [Fact]
    public async Task The_orphan_of_a_change_that_never_landed_is_not_served()
    {
        fx.SkipIfUnavailable();
        var rig = UpdateTests.MakeRig(fx);
        await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [UpdateTests.V(T0, 1)], UpdateTests.Who, Ct);

        rig.Updater.BeforeSampleWrite = () => throw new InvalidOperationException("the sample INSERT failed");
        await Should.ThrowAsync<InvalidOperationException>(
            () => rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 2)], UpdateTests.Who, Ct));
        (await TrailRowsAsync(rig, "t")).ShouldBe(2, "precondition: the orphan trail row is stored");

        var trail = await ModifiedAsync(rig, "t");
        trail.ShouldHaveSingleItem("only the insert that happened").UpdateType.ShouldBe(HistoryModificationType.Insert);

        // Positive control: the same replace, landing, is served.
        rig.Updater.BeforeSampleWrite = null;
        await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 2)], UpdateTests.Who, Ct);
        (await ModifiedAsync(rig, "t")).Select(v => v.UpdateType)
            .ShouldBe([HistoryModificationType.Insert, HistoryModificationType.Replace]);
    }

    [Theory]
    [InlineData(false)]
    [InlineData(true)]
    public async Task A_change_that_landed_without_its_marker_is_still_served(bool delete)
    {
        // A crash between the sample change and the marker: the row stays pending, but the sample shows the change.
        fx.SkipIfUnavailable();
        var rig = UpdateTests.MakeRig(fx);
        await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [UpdateTests.V(T0, 1)], UpdateTests.Who, Ct);
        rig.Updater.SkipAppliedMarker = true;

        if (delete)
            (await rig.Updater.DeleteAtTimeAsync("t", [T0], UpdateTests.Who, Ct)).ShouldBe([HistorianStatusCodes.Good]);
        else
            await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 2)], UpdateTests.Who, Ct);

        (await ModifiedAsync(rig, "t")).Select(v => v.UpdateType).ShouldBe(
            [HistoryModificationType.Insert, delete ? HistoryModificationType.Delete : HistoryModificationType.Replace]);
    }

    [Theory]
    [InlineData(false)]
    [InlineData(true)]
    public async Task A_change_whose_marker_was_lost_stays_served_after_a_later_change_at_its_time(bool delete)
    {
        // lmxopcua#778: V0→V1 lands but its applied marker is never written; a later V1→V2 (or a delete) at the same time
        // used to hide it — the pending row was no longer the latest at its time, so it was dropped as an orphan.
        fx.SkipIfUnavailable();
        var rig = UpdateTests.MakeRig(fx);
        await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [UpdateTests.V(T0, 0)], UpdateTests.Who, Ct);
        rig.Updater.SkipAppliedMarker = true;
        await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 1)], UpdateTests.Who, Ct);
        rig.Updater.SkipAppliedMarker = false;
        (await ModifiedAsync(rig, "t")).Count.ShouldBe(2, "precondition: served while it is the latest row");

        if (delete)
            (await rig.Updater.DeleteAtTimeAsync("t", [T0], UpdateTests.Who, Ct)).ShouldBe([HistorianStatusCodes.Good]);
        else
            await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 2)], UpdateTests.Who, Ct);

        var trail = await ModifiedAsync(rig, "t");
        trail.Select(v => v.UpdateType).ShouldBe(
            [HistoryModificationType.Insert, HistoryModificationType.Replace, delete ? HistoryModificationType.Delete : HistoryModificationType.Replace]);
        trail.Select(v => v.Value.Value).ShouldBe([null, 0.0, 1.0], "each row carries the value its change replaced");
    }

    [Fact]
    public async Task An_orphan_followed_by_a_landed_change_whose_marker_was_lost_is_still_dropped()
    {
        // The other side of lmxopcua#778's rule: V0→V1 never lands (orphan), then the retry V0→V1 lands without its marker,
        // then V1→V2. The orphan's next row recorded the SAME prior value (V0) ⇒ dropped; the retry's next row recorded V1
        // ⇒ served.
        fx.SkipIfUnavailable();
        var rig = UpdateTests.MakeRig(fx);
        await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [UpdateTests.V(T0, 0)], UpdateTests.Who, Ct);
        rig.Updater.BeforeSampleWrite = () => throw new InvalidOperationException("the sample INSERT failed");
        await Should.ThrowAsync<InvalidOperationException>(
            () => rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 1)], UpdateTests.Who, Ct));
        rig.Updater.BeforeSampleWrite = null;
        rig.Updater.SkipAppliedMarker = true;
        await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 1)], UpdateTests.Who, Ct);
        rig.Updater.SkipAppliedMarker = false;
        await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 2)], UpdateTests.Who, Ct);
        (await TrailRowsAsync(rig, "t")).ShouldBe(4, "precondition: the orphan trail row is stored");

        var trail = await ModifiedAsync(rig, "t");
        trail.Select(v => v.UpdateType)
            .ShouldBe([HistoryModificationType.Insert, HistoryModificationType.Replace, HistoryModificationType.Replace]);
        trail.Select(v => v.Value.Value).ShouldBe([null, 0.0, 1.0]);
    }

    [Fact]
    public async Task A_trail_insert_re_sent_after_a_transient_failure_does_not_duplicate()
    {
        fx.SkipIfUnavailable();
        var rig = UpdateTests.MakeRig(fx);
        await rig.Tags.Bootstrap.EnsureAsync(Ct);
        var failedOnce = false;
        // The first attempt APPLIES and then fails as a client timeout would: the re-send carries the same ids.
        rig.Updater.AfterTrailInsert = attempt =>
        {
            if (attempt == 0 && !failedOnce) { failedOnce = true; throw new TimeoutException("client timeout"); }
            return Task.CompletedTask;
        };

        (await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert,
                [UpdateTests.V(T0, 1), UpdateTests.V(T0.AddSeconds(1), 2), UpdateTests.V(T0.AddSeconds(2), 3)], UpdateTests.Who, Ct))
            .ShouldAllBe(s => s == HistorianStatusCodes.GoodEntryInserted);

        failedOnce.ShouldBeTrue();
        rig.Updater.TrailInserts.ShouldBeGreaterThanOrEqualTo(3, "pending, the re-send, then the marker");
        (await TrailRowsAsync(rig, "t")).ShouldBe(3, "the re-sent rows converge on their dedup key");
        (await ModifiedAsync(rig, "t")).Count.ShouldBe(3);
    }

    /// <summary>
    ///     The paged-vs-unpaged fuzz for the reconciled modified read: a random trail (several modifications per source
    ///     time, a random mix of applied, orphaned and landed-without-marker rows) paged at random caps, forward and
    ///     reverse, through the cursor, must concatenate to the unpaged read — the window function's partition must never be
    ///     cut by the resume position — and the unpaged read must equal an independent oracle of the marker rule.
    /// </summary>
    [Fact]
    public async Task Paged_modified_reads_equal_the_unpaged_read_over_a_random_marker_mix()
    {
        fx.SkipIfUnavailable();
        var rig = UpdateTests.MakeRig(fx);
        await rig.Tags.Bootstrap.EnsureAsync(Ct);
        await rig.Tags.EnsureAsync("t", HistorianDataTypes.FromKind(HistorianValueKind.Double), Ct);
        var rng = new Random(746);
        var expected = new List<(DateTime Ts, long Id)>();
        var trailSql = new System.Text.StringBuilder();
        var sampleSql = new System.Text.StringBuilder();
        long id = 1_000;
        var mod0 = new DateTime(2026, 5, 1, 0, 0, 0, DateTimeKind.Utc);
        for (var t = 0; t < 40; t++)
        {
            var ts = T0.AddSeconds(t);
            // Each row's recorded prior value is 1.0 or 2.0: a pending row below the latest is served exactly when the
            // next row recorded a different prior (lmxopcua#778).
            var rows = new List<(long Id, bool Pending, DateTime Mod, double Prior)>();
            for (var k = rng.Next(1, 5); k > 0; k--) rows.Add((id++, rng.Next(3) == 0, mod0.AddSeconds(id), rng.Next(2) + 1.0));
            // The sample's write time: after every modification (the latest landed), or before the latest (it did not).
            var latest = rows[^1];
            var landed = rng.Next(2) == 0;
            var sampleServer = landed ? latest.Mod.AddSeconds(1) : latest.Mod.AddSeconds(-1);
            for (var j = 0; j < rows.Count; j++)
            {
                var r = rows[j];
                var served = !r.Pending || (r.Id == latest.Id ? landed : rows[j + 1].Prior != r.Prior);
                if (served) expected.Add((ts, r.Id));
                if (trailSql.Length > 0) trailSql.Append(',');
                trailSql.Append(FormattableString.Invariant(
                    $"('{ts:O}','t',{r.Id},'{r.Mod:O}',1,'p',NULL,1,{r.Prior:0.0},NULL,NULL,0,NULL,{(r.Pending ? "true" : "false")})"));
            }
            if (rng.Next(6) > 0)
            {
                if (sampleSql.Length > 0) sampleSql.Append(',');
                sampleSql.Append($"('{ts:O}','t',1,1.0,NULL,NULL,0,false,'{sampleServer:O}')");
            }
            else
            {
                // No sample at this time at all: the latest row, if pending, is not served (earlier rows are unaffected).
                expected.RemoveAll(e => e.Ts == ts && e.Id == latest.Id && latest.Pending);
            }
        }
        await using (var conn = await rig.Conn.OpenAsync(Ct))
        {
            await using (var cmd = rig.Conn.CreateCommand(conn, $"INSERT INTO {rig.Conn.SampleModifiedTable} (ts, tag, modified_id, modification_ts, " +
                             $"update_type, principal, api_key_id, kind, num_value, int_value, str_value, status, server_ts, pending) VALUES {trailSql}"))
                await cmd.ExecuteNonQueryAsync(Ct);
            if (sampleSql.Length > 0)
                await using (var cmd = rig.Conn.CreateCommand(conn, $"INSERT INTO {rig.Conn.SampleTable} (ts, tag, kind, num_value, " +
                                 $"int_value, str_value, status, deleted, server_ts) VALUES {sampleSql}"))
                    await cmd.ExecuteNonQueryAsync(Ct);
        }
        var wal = new QuestDbWalVisibility(rig.Conn, TimeSpan.FromSeconds(10));
        await wal.WaitAsync(rig.Conn.SampleModifiedTable, Ct);
        await wal.WaitAsync(rig.Conn.SampleTable, Ct);

        var src = Source(rig);
        var (lo, hi) = (T0.AddSeconds(-1), T0.AddSeconds(41));
        foreach (var reverse in new[] { false, true })
        {
            var (start, end) = reverse ? (hi, lo) : (lo, hi);
            var truth = (await src.ReadModifiedAsync("t", start, end, 0, null, Ct)).Values.Select(v => (v.SourceTimeUtc, v.ModifiedId)).ToList();
            var oracle = reverse ? expected.AsEnumerable().Reverse().ToList() : expected;
            truth.ShouldBe(oracle, $"unpaged vs the marker oracle (reverse {reverse})");
            foreach (var cap in new uint[] { 1, 2, 3, 5, 7, 13, 50 })
            {
                var paged = new List<(DateTime, long)>();
                HistoryModifiedCursor? cursor = null;
                for (var page = 0; ; page++)
                {
                    page.ShouldBeLessThan(500, "paging must terminate");
                    var r = await src.ReadModifiedAsync("t", start, end, cap, cursor, Ct);
                    r.Values.Count.ShouldBeLessThanOrEqualTo((int)cap);
                    paged.AddRange(r.Values.Select(v => (v.SourceTimeUtc, v.ModifiedId)));
                    if (!r.HasMore) break;
                    cursor = new HistoryModifiedCursor(r.Values[^1].SourceTimeUtc, r.Values[^1].ModifiedId);
                }
                paged.ShouldBe(truth, $"paged ≠ unpaged at cap {cap} (reverse {reverse})");
            }
        }
        expected.Count.ShouldBeGreaterThan(0);
        ((long)expected.Count).ShouldBeLessThan(id - 1_000, "the mix must hide some rows, or the reconciliation is untested");
    }

    [Fact]
    public async Task A_pre_marker_trail_table_is_migrated_on_boot_and_its_old_rows_stay_visible()
    {
        fx.SkipIfUnavailable();
        var p = fx.UniquePrefix();
        var o = fx.Options(p);
        var c = new QuestDbConnection(o);
        await using (var conn = await c.OpenAsync(Ct))
        {
            foreach (var ddl in QuestDbTables.CreateAll(p, null))
            {
                // The trail as it was: no pending column, no dedup key.
                var old = ddl.Replace(", pending BOOLEAN", "", StringComparison.Ordinal)
                    .Replace(" DEDUP UPSERT KEYS(ts, tag, modified_id)", "", StringComparison.Ordinal);
                await using var cmd = c.CreateCommand(conn, old);
                await cmd.ExecuteNonQueryAsync(Ct);
            }
        }

        // An old-format update first (the bootstrap with AutoCreateSchema=false keeps the old format).
        var legacyOptions = fx.Options(p, autoCreate: false);
        var legacyConn = new QuestDbConnection(legacyOptions);
        var legacyWal = new QuestDbWalVisibility(legacyConn, legacyOptions.VisibilityTimeout);
        var legacyTags = new QuestDbTagCatalog(legacyConn, new QuestDbSchemaBootstrapper(legacyConn, legacyOptions), legacyWal);
        var legacy = new UpdateTests.Rig(legacyConn, new QuestDbHistorianDataUpdater(legacyConn, legacyTags, legacyWal), legacyTags);
        (await legacy.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [UpdateTests.V(T0, 1)], UpdateTests.Who, Ct))
            .ShouldBe([HistorianStatusCodes.GoodEntryInserted]);
        legacyTags.Bootstrap.SupportsTrailMarker.ShouldBeFalse();
        (await ModifiedAsync(legacy, "t")).Count.ShouldBe(1);

        // A boot with AutoCreateSchema=true migrates; the old row (pending reads false) is still served, a new one too.
        var wal = new QuestDbWalVisibility(c, o.VisibilityTimeout);
        var tags = new QuestDbTagCatalog(c, new QuestDbSchemaBootstrapper(c, o), wal);
        var rig = new UpdateTests.Rig(c, new QuestDbHistorianDataUpdater(c, tags, wal), tags);
        await tags.Bootstrap.EnsureAsync(Ct);
        tags.Bootstrap.SupportsTrailMarker.ShouldBeTrue();
        await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 2)], UpdateTests.Who, Ct);
        (await ModifiedAsync(rig, "t")).Select(v => v.UpdateType)
            .ShouldBe([HistoryModificationType.Insert, HistoryModificationType.Replace]);
    }
}