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(() => 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); } } }