(#298) Conflict Resolver implementation. (#325)

* (#298) Conflict resolver.

Author: adrianhall

src/CommunityToolkit.Datasync.Client/Offline/ConflictResolvers.cs

@@ -0,0 +1,55 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+namespace CommunityToolkit.Datasync.Client.Offline;
+
+/// <summary>
+/// An abstract class that provides a mechanism for resolving conflicts between client and server objects of a specified
+/// type asynchronously.  The object edition of the conflict resolver just calls the typed version.
+/// </summary>
+/// <typeparam name="TEntity">The type of entity being resolved.</typeparam>
+public abstract class AbstractConflictResolver<TEntity> : IConflictResolver<TEntity>
+{
+    /// <inheritdoc />
+    public abstract Task<ConflictResolution> ResolveConflictAsync(TEntity? clientObject, TEntity? serverObject, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// The object version of the resolver calls the typed version.
+    /// </summary>
+    /// <param name="clientObject"></param>
+    /// <param name="serverObject"></param>
+    /// <param name="cancellationToken"></param>
+    /// <returns></returns>
+    public virtual async Task<ConflictResolution> ResolveConflictAsync(object? clientObject, object? serverObject, CancellationToken cancellationToken = default)
+        => await ResolveConflictAsync((TEntity?)clientObject, (TEntity?)serverObject, cancellationToken);
+}
+
+/// <summary>
+/// A conflict resolver where the client object always wins.
+/// </summary>
+public class ClientWinsConflictResolver : IConflictResolver
+{
+    /// <inheritdoc />
+    public Task<ConflictResolution> ResolveConflictAsync(object? clientObject, object? serverObject, CancellationToken cancellationToken = default)
+        => Task.FromResult(new ConflictResolution
+        {
+            Result = ConflictResolutionResult.Client,
+            Entity = clientObject
+        });
+}
+
+/// <summary>
+/// A conflict resolver where the server object always wins.
+/// </summary>
+public class ServerWinsConflictResolver : IConflictResolver
+{
+    /// <inheritdoc />
+    public Task<ConflictResolution> ResolveConflictAsync(object? clientObject, object? serverObject, CancellationToken cancellationToken = default)
+        => Task.FromResult(new ConflictResolution
+        {
+            Result = ConflictResolutionResult.Server,
+            Entity = serverObject
+        });
+}
+

src/CommunityToolkit.Datasync.Client/Offline/DatasyncOfflineOptionsBuilder.cs

@@ -95,6 +95,7 @@ public DatasyncOfflineOptionsBuilder Entity<TEntity>(Action<EntityOfflineOptions
         EntityOfflineOptions<TEntity> entity = new();
         configure(entity);
         options.ClientName = entity.ClientName;
+        options.ConflictResolver = entity.ConflictResolver;
         options.Endpoint = entity.Endpoint;
         options.QueryDescription = new QueryTranslator<TEntity>(entity.Query).Translate();
         return this;
@@ -137,7 +138,7 @@ internal OfflineOptions Build()
 
         foreach (EntityOfflineOptions entity in this._entities.Values)
         {
-            result.AddEntity(entity.EntityType, entity.ClientName, entity.Endpoint, entity.QueryDescription);
+            result.AddEntity(entity.EntityType, entity.ClientName, entity.ConflictResolver, entity.Endpoint, entity.QueryDescription);
         }
 
         return result;
@@ -164,6 +165,11 @@ public class EntityOfflineOptions(Type entityType)
         /// </summary>
         public Uri Endpoint { get; set; } = new Uri($"/tables/{entityType.Name.ToLowerInvariant()}", UriKind.Relative);
 
+        /// <summary>
+        /// The conflict resolver for this entity.
+        /// </summary>
+        public IConflictResolver? ConflictResolver { get; set; }
+
         /// <summary>
         /// The query description for the entity type - may be null (to mean "pull everything").
         /// </summary>
@@ -186,6 +192,11 @@ public class EntityOfflineOptions<TEntity>() where TEntity : class
         /// </summary>
         public string ClientName { get; set; } = string.Empty;
 
+        /// <summary>
+        /// The conflict resolver for this entity.
+        /// </summary>
+        public IConflictResolver? ConflictResolver { get; set; }
+
         /// <summary>
         /// The endpoint for the entity type.
         /// </summary>

src/CommunityToolkit.Datasync.Client/Offline/IConflictResolver.cs

@@ -0,0 +1,40 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+namespace CommunityToolkit.Datasync.Client.Offline;
+
+/// <summary>
+/// Definition of a conflict resolver.  This is used in push situations where
+/// the server returns a 409 or 412 status code indicating that the client is
+/// out of step with the server.
+/// </summary>
+public interface IConflictResolver
+{
+    /// <summary>
+    /// Resolves the conflict between two objects - client side and server side.
+    /// </summary>
+    /// <param name="clientObject">The client object.</param>
+    /// <param name="serverObject">The server object.</param>
+    /// <param name="cancellationToken">A <see cref="CancellationToken"/> to observe.</param>
+    /// <returns>The conflict resolution.</returns>
+    Task<ConflictResolution> ResolveConflictAsync(object? clientObject, object? serverObject, CancellationToken cancellationToken = default);
+}
+
+/// <summary>
+/// Definition of a conflict resolver.  This is used in push situations where
+/// the server returns a 409 or 412 status code indicating that the client is
+/// out of step with the server.
+/// </summary>
+/// <typeparam name="TEntity">The type of the entity.</typeparam>
+public interface IConflictResolver<TEntity> : IConflictResolver
+{
+    /// <summary>
+    /// Resolves the conflict between two objects - client side and server side.
+    /// </summary>
+    /// <param name="clientObject">The client object.</param>
+    /// <param name="serverObject">The server object.</param>
+    /// <param name="cancellationToken">A <see cref="CancellationToken"/> to observe.</param>
+    /// <returns>The conflict resolution.</returns>
+    Task<ConflictResolution> ResolveConflictAsync(TEntity? clientObject, TEntity? serverObject, CancellationToken cancellationToken = default);
+}

src/CommunityToolkit.Datasync.Client/Offline/Models/ConflictResolution.cs

@@ -0,0 +1,44 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+namespace CommunityToolkit.Datasync.Client.Offline;
+
+/// <summary>
+/// The possible results of a conflict resolution.
+/// </summary>
+public enum ConflictResolutionResult
+{
+    /// <summary>
+    /// The default resolution, which is to do nothing and re-queue the operation.
+    /// </summary>
+    Default,
+
+    /// <summary>
+    /// The provided client object should be used.  This results in a new "force" submission
+    /// to the server to over-write the server entity.
+    /// </summary>
+    Client,
+
+    /// <summary>
+    /// The server object should be used.  This results in the client object being updated
+    /// with whatever the server object was provided.
+    /// </summary>
+    Server
+}
+
+/// <summary>
+/// The model class returned by a conflict resolver to indicate the resolution of the conflict.
+/// </summary>
+public class ConflictResolution
+{
+    /// <summary>
+    /// The conflict resolution result.
+    /// </summary>
+    public ConflictResolutionResult Result { get; set; } = ConflictResolutionResult.Default;
+
+    /// <summary>
+    /// The entity, if required.
+    /// </summary>
+    public object? Entity { get; set; }
+}

src/CommunityToolkit.Datasync.Client/Offline/Models/EntityDatasyncOptions.cs

@@ -11,6 +11,11 @@ namespace CommunityToolkit.Datasync.Client.Offline.Models;
 /// </summary>
 internal class EntityDatasyncOptions
 {
+    /// <summary>
+    /// The conflict resolver for the entity.
+    /// </summary>
+    internal IConflictResolver? ConflictResolver { get; init; }
+
     /// <summary>
     /// The endpoint for the entity type.
     /// </summary>

src/CommunityToolkit.Datasync.Client/Offline/Models/OfflineOptions.cs

@@ -25,11 +25,18 @@ internal class OfflineOptions()
     /// </summary>
     /// <param name="entityType">The type of the entity being stored.</param>
     /// <param name="clientName">The name of the client.</param>
+    /// <param name="conflictResolver">The conflict resolver to use.</param>
     /// <param name="endpoint">The endpoint serving the datasync services.</param>
     /// <param name="queryDescription">The optional query description to describe what entities need to be pulled.</param>
-    public void AddEntity(Type entityType, string clientName, Uri endpoint, QueryDescription? queryDescription = null)
+    public void AddEntity(Type entityType, string clientName, IConflictResolver? conflictResolver, Uri endpoint, QueryDescription? queryDescription = null)
     {
-        this._cache.Add(entityType, new EntityOptions { ClientName = clientName, Endpoint = endpoint, QueryDescription = queryDescription });
+        this._cache.Add(entityType, new EntityOptions 
+        { 
+            ClientName = clientName, 
+            ConflictResolver = conflictResolver,
+            Endpoint = endpoint, 
+            QueryDescription = queryDescription
+        });
     }
 
     /// <summary>
@@ -43,6 +50,7 @@ public EntityDatasyncOptions GetOptions(Type entityType)
         {
             return new()
             {
+                ConflictResolver = options.ConflictResolver,
                 Endpoint = options.Endpoint,
                 HttpClient = HttpClientFactory.CreateClient(options.ClientName),
                 QueryDescription = options.QueryDescription ?? new QueryDescription()
@@ -52,6 +60,7 @@ public EntityDatasyncOptions GetOptions(Type entityType)
         {
             return new()
             {
+                ConflictResolver = null,
                 Endpoint = new Uri($"tables/{entityType.Name.ToLowerInvariant()}", UriKind.Relative),
                 HttpClient = HttpClientFactory.CreateClient(),
                 QueryDescription = new QueryDescription()
@@ -69,6 +78,11 @@ internal class EntityOptions
         /// </summary>
         public required string ClientName { get; set; }
 
+        /// <summary>
+        /// The conflict resolver for the entity options.
+        /// </summary>
+        internal IConflictResolver? ConflictResolver { get; set; }
+
         /// <summary>
         /// The endpoint for the entity type.
         /// </summary>

src/CommunityToolkit.Datasync.Client/Offline/Operations/PullOperationManager.cs

@@ -59,19 +59,19 @@ public async Task<PullResult> ExecuteAsync(IEnumerable<PullRequest> requests, Pu
                 {
                     _ = context.Add(item);
                     result.IncrementAdditions();
-                } 
+                }
                 else if (originalEntity is not null && metadata.Deleted)
                 {
                     _ = context.Remove(originalEntity);
                     result.IncrementDeletions();
-                } 
+                }
                 else if (originalEntity is not null && !metadata.Deleted)
                 {
                     context.Entry(originalEntity).CurrentValues.SetValues(item);
                     result.IncrementReplacements();
                 }
 
-                if (metadata.UpdatedAt.HasValue && metadata.UpdatedAt.Value > lastSynchronization)
+                if (metadata.UpdatedAt > lastSynchronization)
                 {
                     lastSynchronization = metadata.UpdatedAt.Value;
                     bool isAdded = await DeltaTokenStore.SetDeltaTokenAsync(pullResponse.QueryId, metadata.UpdatedAt.Value, cancellationToken).ConfigureAwait(false);

src/CommunityToolkit.Datasync.Client/Offline/OperationsQueue/OperationsQueueManager.cs

@@ -8,6 +8,7 @@
 using CommunityToolkit.Datasync.Client.Threading;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.EntityFrameworkCore.ChangeTracking;
+using System.Net;
 using System.Reflection;
 using System.Text.Json;
 
@@ -77,7 +78,7 @@ internal List<EntityEntry> GetChangedEntitiesInScope()
     /// </summary>
     /// <remarks>
     /// An entity is "synchronization ready" if:
-    /// 
+    ///
     /// * It is a property on this context
     /// * The property is public and a <see cref="DbSet{TEntity}"/>.
     /// * The property does not have a <see cref="DoNotSynchronizeAttribute"/> specified.
@@ -215,7 +216,7 @@ internal IEnumerable<Type> GetSynchronizableEntityTypes(IEnumerable<Type> allowe
     /// </summary>
     /// <remarks>
     /// An entity is "synchronization ready" if:
-    /// 
+    ///
     /// * It is a property on this context
     /// * The property is public and a <see cref="DbSet{TEntity}"/>.
     /// * The property does not have a <see cref="DoNotSynchronizeAttribute"/> specified.
@@ -299,7 +300,40 @@ internal async Task<PushResult> PushAsync(IEnumerable<Type> entityTypes, PushOpt
         ExecutableOperation op = await ExecutableOperation.CreateAsync(operation, cancellationToken).ConfigureAwait(false);
         ServiceResponse response = await op.ExecuteAsync(options, cancellationToken).ConfigureAwait(false);
 
-        if (!response.IsSuccessful)
+        bool isSuccessful = response.IsSuccessful;
+        if (response.IsConflictStatusCode && options.ConflictResolver is not null)
+        {
+            object? serverEntity = JsonSerializer.Deserialize(response.ContentStream, entityType, DatasyncSerializer.JsonSerializerOptions);
+            object? clientEntity = JsonSerializer.Deserialize(operation.Item, entityType, DatasyncSerializer.JsonSerializerOptions);
+            ConflictResolution resolution = await options.ConflictResolver.ResolveConflictAsync(clientEntity, serverEntity, cancellationToken).ConfigureAwait(false);
+
+            if (resolution.Result is ConflictResolutionResult.Client)
+            {
+                operation.Item = JsonSerializer.Serialize(resolution.Entity, entityType, DatasyncSerializer.JsonSerializerOptions);
+                operation.State = OperationState.Pending;
+                operation.LastAttempt = DateTimeOffset.UtcNow;
+                operation.HttpStatusCode = response.StatusCode;
+                operation.EntityVersion = string.Empty; // Force the push
+                operation.Version++;
+                _ = this._context.Update(operation);
+                ExecutableOperation resolvedOp = await ExecutableOperation.CreateAsync(operation, cancellationToken).ConfigureAwait(false);
+                response = await resolvedOp.ExecuteAsync(options, cancellationToken).ConfigureAwait(false);
+                isSuccessful = response.IsSuccessful;
+            }
+            else if (resolution.Result is ConflictResolutionResult.Server)
+            {
+                lock (this.pushlock)
+                {
+                    operation.State = OperationState.Completed; // Make it successful
+                    operation.LastAttempt = DateTimeOffset.UtcNow;
+                    operation.HttpStatusCode = 200;
+                    isSuccessful = true;
+                    _ = this._context.Update(operation);
+                }
+            }
+        }
+
+        if (!isSuccessful)
         {
             lock (this.pushlock)
             {
@@ -315,6 +349,7 @@ internal async Task<PushResult> PushAsync(IEnumerable<Type> entityTypes, PushOpt
         // If the operation is a success, then the content may need to be updated.
         if (operation.Kind != OperationKind.Delete)
         {
+            _ = response.ContentStream.Seek(0L, SeekOrigin.Begin); // Reset the memory stream to the beginning.
             object? newValue = JsonSerializer.Deserialize(response.ContentStream, entityType, DatasyncSerializer.JsonSerializerOptions);
             object? oldValue = await this._context.FindAsync(entityType, [operation.ItemId], cancellationToken).ConfigureAwait(false);
             ReplaceDatabaseValue(oldValue, newValue);