ulapp/tests/FieldLogger.Sync.Tests/SyncEngineTests.cs

using System.Text.Json;
using System.Text.Json.Nodes;
using FieldLogger.Sync;
using Xunit;

namespace FieldLogger.Sync.Tests;

public sealed class SyncEngineTests
{
    private static readonly DateTimeOffset Now = DateTimeOffset.Parse("2026-08-21T10:00:00Z");

    static SyncEngineTests() => SQLitePCL.Batteries_V2.Init();

    [Fact]
    public void Point_serializes_to_the_frozen_singleton_batch()
    {
        var point = MakePoint("018f1a00-0000-7000-8000-000000000001");
        using var json = JsonDocument.Parse(point.ToAppLogPayloadUtf8(jobId: "job_1"));

        var root = json.RootElement;
        Assert.Equal("1", root.GetProperty("schemaVersion").GetString());
        Assert.Equal("job_1", root.GetProperty("jobId").GetString());
        var wire = Assert.Single(root.GetProperty("points").EnumerateArray());
        Assert.Equal(point.PointId, wire.GetProperty("pointId").GetString());
        Assert.Equal("APP", wire.GetProperty("origin").GetString());
        Assert.Equal("APP_MQTT", wire.GetProperty("uploadPath").GetString());
        Assert.Equal("FIXED", wire.GetProperty("fix").GetString());
        Assert.Equal(-80.2, wire.GetProperty("lng").GetDouble());
    }

    [Fact]
    public void App_serializer_matches_the_shared_contract_fixture()
    {
        JsonNode? actual = JsonNode.Parse(MakePoint("018f1a00-0000-7000-8000-000000000001")
            .ToAppLogPayloadUtf8(jobId: "job_1"));
        string path = Path.Combine(AppContext.BaseDirectory, "ContractFixtures", "app-log-points-v1.json");
        JsonNode? expected = JsonNode.Parse(File.ReadAllText(path));

        Assert.True(JsonNode.DeepEquals(expected, actual));
    }

    [Fact]
    public void Invalid_capture_is_refused_with_a_reason_code()
    {
        var point = MakePoint("018f1a00-0000-7000-8000-000000000002") with { Position = null };
        var error = Assert.Throws<PointNotPublishableException>(() => point.ToAppLogPayloadUtf8(jobId: "job_1"));
        Assert.Equal("POSITION_MISSING", error.ReasonCode);
    }

    [Fact]
    public async Task Queue_survives_restart_and_releases_only_after_acceptance()
    {
        string db = NewDbPath();
        try
        {
            var firstStore = new SqliteOutboundStore(db);
            await firstStore.InitAsync();
            var message = OutboundMessage.FromPoint(
                MakePoint("018f1a00-0000-7000-8000-000000000003"),
                "ul/org_alpha/app/client_1/log/points", Now, jobId: "job_1");
            Assert.True(await firstStore.EnqueueAsync(message));

            var firstTransport = new FakeMqttTransport();
            var firstEngine = Engine(firstTransport, firstStore);
            Assert.Equal(1, await firstEngine.DrainOnceAsync());
            Assert.Equal(1, await firstStore.PendingCountAsync()); // broker PUBACK is not enough

            var restartedStore = new SqliteOutboundStore(db);
            await restartedStore.InitAsync(); // resets interrupted in-flight work to pending
            var restartedTransport = new FakeMqttTransport();
            var restartedEngine = Engine(restartedTransport, restartedStore);
            Assert.Equal(1, await restartedEngine.DrainOnceAsync());
            await restartedTransport.InjectAcceptAsync(message.PointId);

            Assert.Equal(0, await restartedStore.PendingCountAsync());
        }
        finally
        {
            DeleteDb(db);
        }
    }

    [Fact]
    public async Task Duplicate_ack_is_safe_to_release_and_publish_failure_is_not()
    {
        string db = NewDbPath();
        try
        {
            var store = new SqliteOutboundStore(db);
            await store.InitAsync();
            var message = OutboundMessage.FromPoint(
                MakePoint("018f1a00-0000-7000-8000-000000000004"),
                "ul/org_alpha/app/client_1/log/points", Now, jobId: "job_1");
            Assert.True(await store.EnqueueAsync(message));

            var transport = new FakeMqttTransport { FailNextPublish = true };
            var engine = Engine(transport, store);
            Assert.Equal(0, await engine.DrainOnceAsync());
            Assert.Equal(1, await store.PendingCountAsync());

            await transport.InjectDuplicateAsync(message.PointId);
            Assert.Equal(0, await store.PendingCountAsync());
        }
        finally
        {
            DeleteDb(db);
        }
    }

    [Fact]
    public async Task Terminal_rejection_is_retained_and_surfaced()
    {
        string db = NewDbPath();
        try
        {
            var store = new SqliteOutboundStore(db);
            await store.InitAsync();
            var message = OutboundMessage.FromPoint(
                MakePoint("018f1a00-0000-7000-8000-000000000005"),
                "ul/org_alpha/app/client_1/log/points", Now, jobId: "job_1");
            await store.EnqueueAsync(message);
            var transport = new FakeMqttTransport();
            var engine = Engine(transport, store);
            OutboundMessage? surfaced = null;
            engine.PointRejected += value =>
            {
                surfaced = value;
                return Task.CompletedTask;
            };

            await transport.InjectRejectAsync(message.PointId, "VALIDATION_ERROR");

            Assert.Equal(0, await store.PendingCountAsync());
            Assert.Equal(message.PointId, surfaced?.PointId);
            Assert.Equal("VALIDATION_ERROR", Assert.Single(await store.GetRejectedAsync()).LastReason);
        }
        finally
        {
            DeleteDb(db);
        }
    }

    [Fact]
    public void Ack_parser_uses_results_and_outcome()
    {
        var ack = AckBatch.FromJsonUtf8(
            """{"schemaVersion":"1","results":[{"pointId":"018f1a00-0000-7000-8000-000000000006","outcome":"DUPLICATE"}]}"""u8);
        Assert.Equal(AckOutcome.DUPLICATE, Assert.Single(ack.Results).Outcome);
    }

    private static MqttSyncEngine Engine(FakeMqttTransport transport, IOutboundStore store) =>
        new(transport, store, new SyncOptions { OrgId = "org_alpha", ClientId = "client_1" },
            () => Now, new BackoffPolicy(TimeSpan.FromSeconds(1), TimeSpan.FromMinutes(1), new Random(1)));

    private static PointRecord MakePoint(string id) => new()
    {
        PointId = id,
        CreatedAt = Now,
        Position = new PositionGroup { Lat = 40.1, Lon = -80.2, PositionEpoch = Now.AddSeconds(-1) },
        Gnss = new GnssGroup { FixType = "RTK_FIXED", SatsUsed = 18, Hrms = 0.02, Vrms = 0.04 },
        Attributes = new AttributesGroup { UtilityType = "WATER" },
    };

    private static string NewDbPath() => Path.Combine(Path.GetTempPath(), $"fieldlogger-sync-{Guid.NewGuid():N}.db3");

    private static void DeleteDb(string db)
    {
        foreach (string path in new[] { db, $"{db}-shm", $"{db}-wal" })
        {
            if (File.Exists(path)) File.Delete(path);
        }
    }
}