WIP on JsonSyncable class

src/SIL.Harmony.Core/QueryHelpers.cs

@@ -31,29 +31,24 @@ public static async IAsyncEnumerable<TCommit> GetMissingCommits<TCommit, TChange
         if (includeChangeEntities) commits = commits.Include(c => c.ChangeEntities);
         foreach (var (clientId, localTimestamp) in localState.ClientHeads)
         {
-            //client is new to the other history
-            if (!remoteState.ClientHeads.TryGetValue(clientId, out var otherTimestamp))
+            long? remoteTimestamp = remoteState.ClientHeads.TryGetValue(clientId, out var otherTimestamp)
+                ? otherTimestamp
+                : null;
+            var clientCommits = commits.Where(c => c.ClientId == clientId);
+            if (remoteTimestamp is null)
             {
-                //todo slow, it would be better if we could query on client id and get latest changes per client
-                await foreach (var commit in commits
-                                   .DefaultOrder()
-                                   .Where(c => c.ClientId == clientId)
-                                   .AsAsyncEnumerable())
-                {
+                await foreach (var commit in clientCommits.DefaultOrder().AsAsyncEnumerable())
                     yield return commit;
-                }
             }
-            //client has newer history than the other history
-            else if (localTimestamp > otherTimestamp)
+            else if (localTimestamp > remoteTimestamp)
             {
-                var otherDt = DateTimeOffset.FromUnixTimeMilliseconds(otherTimestamp);
-                //todo even slower because we need to filter out changes that are already in the other history
-                await foreach (var commit in commits
+                var otherDt = DateTimeOffset.FromUnixTimeMilliseconds(remoteTimestamp.Value);
+                await foreach (var commit in clientCommits
+                                   .Where(c => c.HybridDateTime.DateTime > otherDt)
                                    .DefaultOrder()
-                                   .Where(c => c.ClientId == clientId && c.HybridDateTime.DateTime > otherDt)
                                    .AsAsyncEnumerable())
                 {
-                    if (commit.DateTime.ToUnixTimeMilliseconds() > otherTimestamp)
+                    if (commit.DateTime.ToUnixTimeMilliseconds() > remoteTimestamp)
                         yield return commit;
                 }
             }
@@ -75,6 +70,47 @@ public static async Task<SortedSet<T>> ToSortedSetAsync<T>(this IQueryable<T> qu
         return set;
     }
 
+    public static IEnumerable<TCommit> GetMissingCommits<TCommit, TChange>(
+        this IEnumerable<TCommit> commits,
+        SyncState localState,
+        SyncState remoteState) where TCommit : CommitBase<TChange>
+    {
+        foreach (var (clientId, localTimestamp) in localState.ClientHeads)
+        {
+            long? remoteTimestamp = remoteState.ClientHeads.TryGetValue(clientId, out var otherTimestamp)
+                ? otherTimestamp
+                : null;
+            foreach (var commit in GetMissingCommitsForClient(
+                         commits.Where(c => c.ClientId == clientId), localTimestamp, remoteTimestamp))
+            {
+                yield return commit;
+            }
+        }
+    }
+
+    private static IEnumerable<TCommit> GetMissingCommitsForClient<TCommit>(
+        IEnumerable<TCommit> clientCommits,
+        long localTimestamp,
+        long? remoteTimestamp) where TCommit : CommitBase
+    {
+        if (remoteTimestamp is null)
+        {
+            foreach (var commit in clientCommits.DefaultOrder())
+                yield return commit;
+        }
+        else if (localTimestamp > remoteTimestamp)
+        {
+            var otherDt = DateTimeOffset.FromUnixTimeMilliseconds(remoteTimestamp.Value);
+            foreach (var commit in clientCommits
+                         .Where(c => c.HybridDateTime.DateTime > otherDt)
+                         .DefaultOrder())
+            {
+                if (commit.DateTime.ToUnixTimeMilliseconds() > remoteTimestamp)
+                    yield return commit;
+            }
+        }
+    }
+
     public static IQueryable<T> DefaultOrder<T>(this IQueryable<T> queryable) where T: CommitBase
     {
         return queryable

src/SIL.Harmony.Tests/JsonSyncableTests.cs

@@ -0,0 +1,78 @@
+using System.Text.Json;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging.Abstractions;
+using SIL.Harmony.Changes;
+using SIL.Harmony.Sample;
+using SIL.Harmony.Sample.Changes;
+
+namespace SIL.Harmony.Tests;
+
+public class JsonSyncableTests
+{
+    private static JsonSerializerOptions SerializerOptions { get; } = new ServiceCollection()
+        .AddCrdtDataSample(":memory:")
+        .BuildServiceProvider()
+        .GetRequiredService<JsonSerializerOptions>();
+
+    [Fact]
+    public async Task WritesPerClientFile()
+    {
+        var dir = Directory.CreateTempSubdirectory("harmony-json-syncable-test-");
+        try
+        {
+            var syncable = new JsonSyncable(dir, SerializerOptions, NullLogger<JsonSyncable>.Instance);
+            var clientId = Guid.NewGuid();
+            var commit = new Commit
+            {
+                ClientId = clientId,
+                HybridDateTime = new HybridDateTime(DateTimeOffset.UtcNow, 0),
+                ChangeEntities =
+                {
+                    new ChangeEntity<IChange>
+                    {
+                        Change = new SetWordTextChange(Guid.NewGuid(), "hello"),
+                        Index = 0,
+                        CommitId = Guid.Empty,
+                        EntityId = Guid.Empty
+                    }
+                }
+            };
+
+            await syncable.AddRangeFromSync([commit]);
+
+            var file = Path.Combine(dir.FullName, $"{JsonSyncable.FilenamePrefix}{clientId}{JsonSyncable.FilenameExtension}");
+            File.Exists(file).Should().BeTrue();
+        }
+        finally
+        {
+            dir.Delete(recursive: true);
+        }
+    }
+
+    [Fact]
+    public async Task DoesNotDuplicateFileLines()
+    {
+        var dir = Directory.CreateTempSubdirectory("harmony-json-syncable-test-");
+        try
+        {
+            var syncable = new JsonSyncable(dir, SerializerOptions, NullLogger<JsonSyncable>.Instance);
+            var commit = new Commit
+            {
+                ClientId = Guid.NewGuid(),
+                HybridDateTime = new HybridDateTime(DateTimeOffset.UtcNow, 0)
+            };
+
+            await syncable.AddRangeFromSync([commit]);
+            await syncable.AddRangeFromSync([commit]);
+
+            var file = new FileInfo(Path.Combine(dir.FullName,
+                $"{JsonSyncable.FilenamePrefix}{commit.ClientId}{JsonSyncable.FilenameExtension}"));
+            var lines = await File.ReadAllLinesAsync(file.FullName);
+            lines.Should().HaveCount(1);
+        }
+        finally
+        {
+            dir.Delete(recursive: true);
+        }
+    }
+}

src/SIL.Harmony.Tests/Syncable/CrossSyncableTests.cs

@@ -0,0 +1,56 @@
+using SIL.Harmony.Changes;
+using SIL.Harmony.Sample.Changes;
+using SIL.Harmony.Sample.Models;
+
+namespace SIL.Harmony.Tests.Syncable;
+
+public class CrossSyncableTests
+{
+    [Fact]
+    public async Task DataModelAndJsonSyncable_CanSyncBidirectionally()
+    {
+        var dataBackend = new DataModelSyncBackend();
+        var jsonBackend = new JsonSyncableSyncBackend();
+        await using var dataModel = await dataBackend.CreateAsync();
+        await using var jsonSyncable = await jsonBackend.CreateAsync();
+
+        var entity1Id = Guid.NewGuid();
+        var entity2Id = Guid.NewGuid();
+        var dataModelCommit = CreateCommit(dataModel.ClientId, new SetWordTextChange(entity1Id, "from-datamodel"));
+        await dataModel.Syncable.AddRangeFromSync([dataModelCommit]);
+        await dataModel.MirrorToReadModelAsync([dataModelCommit]);
+        var jsonSyncableCommit = CreateCommit(jsonSyncable.ClientId, new SetWordTextChange(entity2Id, "from-json"));
+        await jsonSyncable.Syncable.AddRangeFromSync([jsonSyncableCommit]);
+        await jsonSyncable.MirrorToReadModelAsync([jsonSyncableCommit]);
+
+        var syncResults = await dataModel.Syncable.SyncWith(jsonSyncable.Syncable);
+        await SyncableTestHelpers.MirrorToReadModelsAsync(dataModel, jsonSyncable, syncResults);
+
+        (await dataModel.ReadModel.GetLatest<Word>(entity1Id))!.Text.Should().Be("from-datamodel");
+        (await dataModel.ReadModel.GetLatest<Word>(entity2Id))!.Text.Should().Be("from-json");
+        (await jsonSyncable.ReadModel.GetLatest<Word>(entity1Id))!.Text.Should().Be("from-datamodel");
+        (await jsonSyncable.ReadModel.GetLatest<Word>(entity2Id))!.Text.Should().Be("from-json");
+
+        var dmState = await dataModel.Syncable.GetSyncState();
+        var jsonState = await jsonSyncable.Syncable.GetSyncState();
+        dmState.ClientHeads.Should().BeEquivalentTo(jsonState.ClientHeads);
+    }
+
+    private static Commit CreateCommit(Guid clientId, IChange change)
+    {
+        var commitId = Guid.NewGuid();
+        return new Commit(commitId)
+        {
+            ClientId = clientId,
+            HybridDateTime = new HybridDateTime(DateTimeOffset.Now, 0),
+            ChangeEntities = [new ChangeEntity<IChange>
+                {
+                    Index = 0,
+                    CommitId = commitId,
+                    EntityId = change.EntityId,
+                    Change = change
+                }
+            ]
+        };
+    }
+}

src/SIL.Harmony.Tests/Syncable/DataModelSyncBackend.cs

@@ -0,0 +1,19 @@
+namespace SIL.Harmony.Tests.Syncable;
+
+public class DataModelSyncBackend : ISyncableTestBackend
+{
+    public string Name => "DataModel";
+    public override string ToString() => Name;
+
+    public async Task<SyncableTestContext> CreateAsync()
+    {
+        var testBase = new DataModelTestBase();
+        await testBase.InitializeAsync();
+        return new SyncableTestContext(
+            testBase.DataModel,
+            testBase.DataModel,
+            Guid.NewGuid(),
+            syncableUpdatesReadModel: true,
+            () => testBase.DisposeAsync());
+    }
+}

src/SIL.Harmony.Tests/Syncable/ISyncableTestBackend.cs

@@ -0,0 +1,23 @@
+namespace SIL.Harmony.Tests.Syncable;
+
+public sealed class SyncableTestContext(
+    ISyncable syncable,
+    DataModel readModel,
+    Guid clientId,
+    bool syncableUpdatesReadModel,
+    Func<ValueTask> disposeAsync) : IAsyncDisposable
+{
+    public ISyncable Syncable { get; } = syncable;
+    public DataModel ReadModel { get; } = readModel;
+    public Guid ClientId { get; } = clientId;
+    internal bool SyncableUpdatesReadModel { get; } = syncableUpdatesReadModel;
+
+    public ValueTask DisposeAsync() => disposeAsync();
+}
+
+/// <summary>Factory for a backend under test (DataModel, JsonSyncable, ...).</summary>
+public interface ISyncableTestBackend
+{
+    string Name { get; }
+    Task<SyncableTestContext> CreateAsync();
+}

src/SIL.Harmony.Tests/Syncable/JsonSyncableSyncBackend.cs

@@ -0,0 +1,35 @@
+using System.Text.Json;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging.Abstractions;
+using SIL.Harmony.Sample;
+
+namespace SIL.Harmony.Tests.Syncable;
+
+public class JsonSyncableSyncBackend : ISyncableTestBackend
+{
+    public string Name => "JsonSyncable";
+    public override string ToString() => Name;
+
+    private JsonSerializerOptions SerializerOptions { get; } = new ServiceCollection()
+        .AddCrdtDataSample(":memory:")
+        .BuildServiceProvider()
+        .GetRequiredService<JsonSerializerOptions>();
+
+    public async Task<SyncableTestContext> CreateAsync()
+    {
+        var testBase = new DataModelTestBase();
+        await testBase.InitializeAsync();
+        var tempDir = Directory.CreateTempSubdirectory("harmony-json-syncable-");
+        var jsonSyncable = new JsonSyncable(tempDir, SerializerOptions, NullLogger<JsonSyncable>.Instance);
+        return new SyncableTestContext(
+            jsonSyncable,
+            testBase.DataModel,
+            Guid.NewGuid(),
+            syncableUpdatesReadModel: false,
+            async () =>
+            {
+                await testBase.DisposeAsync();
+                tempDir.Delete(recursive: true);
+            });
+    }
+}

src/SIL.Harmony.Tests/Syncable/SyncableTestHelpers.cs

@@ -0,0 +1,33 @@
+namespace SIL.Harmony.Tests.Syncable;
+
+public static class SyncableTestHelpers
+{
+    public static ISyncableTestBackend[] Backends =>
+    [
+        new DataModelSyncBackend(),
+        new JsonSyncableSyncBackend()
+    ];
+
+    public static IEnumerable<object[]> BackendData => Backends.Select(backend => new object[] { backend });
+
+    public static IEnumerable<object[]> BackendPairData =>
+        Backends.SelectMany(left => Backends, (left, right) => new object[] { left, right });
+
+    public static async Task MirrorToReadModelAsync(this SyncableTestContext context, IEnumerable<Commit> commits)
+    {
+        if (context.SyncableUpdatesReadModel) return;
+
+        var array = commits.ToArray();
+        if (array.Length > 0)
+            await ((ISyncable)context.ReadModel).AddRangeFromSync(array);
+    }
+
+    public static async Task MirrorToReadModelsAsync(
+        SyncableTestContext local,
+        SyncableTestContext remote,
+        SyncResults results)
+    {
+        await local.MirrorToReadModelAsync(results.MissingFromLocal);
+        await remote.MirrorToReadModelAsync(results.MissingFromRemote);
+    }
+}