(#384) added pull progress report via event OfflineDbContext.SynchronizationProgress (#387)

* (#377) Fix for possible concurrent query during pull

* (#384) Added SynchronizationProgress event

* (#384) Added test scenario for  SynchronizationProgress event (OfflineDbContext_Tests.SynchronizationProgress_Event_Works)

* (#384) Added SynchronizationEventType PullStarted and PullEnded.
Added Exception and ServiceResponse to SynchronizationEventArgs.
Fixed bug in SynchronizationProgress_Event_Works test. Initialization of eventFired was wrong.
SynchronizationProgress_Event_Works added tests for start and end.

* (#384)
Added test PullAsync_List_FailedRequest_SynchronizationEventWorks
Added Synchronization events for push operations and test for it.

* (#384) Changed DateTimeConverter on client and server side to handle 'default' DateTime correctly.

* (#384) Renamed TotalNrItems on SynchronizationEventArgs to ItemsTotal.
Made DateTimeConverter on server Except again if input is empty.
Optimised DataTimeConverter to avoid double parsing.

* Update src/CommunityToolkit.Datasync.Client/Offline/SynchronizationEventArgs.cs

Co-authored-by: Copilot <[email protected]>

* Update src/CommunityToolkit.Datasync.Client/Offline/SynchronizationEventArgs.cs

Co-authored-by: Copilot <[email protected]>

* Update src/CommunityToolkit.Datasync.Client/Offline/SynchronizationEventArgs.cs

Co-authored-by: Copilot <[email protected]>

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

Co-authored-by: Copilot <[email protected]>

* Added DateTimeConverte tests for verifying default time, local time and utc time conversion.

* Changed QueueHandler_WithThreads_Enqueue test to sleep a thread for 2 seconds instead of 1. This reduces the risk of thread reusage which could cause the distinct threads count to fail.

---------

Co-authored-by: Copilot <[email protected]>

Author: StefanCuypers

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

@@ -99,6 +99,12 @@ public abstract partial class OfflineDbContext : DbContext
     /// </summary>
     internal OperationsQueueManager QueueManager { get; }
 
+    /// <summary>
+    /// An event delegate that allows the app to monitor synchronization events.
+    /// </summary>
+    /// <remarks>This event can be called from background threads.</remarks>
+    public event EventHandler<SynchronizationEventArgs>? SynchronizationProgress;
+
     /// <summary>
     ///     Initializes a new instance of the <see cref="OfflineDbContext" /> class. The 
     ///     <see cref="OnConfiguring(DbContextOptionsBuilder)" /> method will be called to 
@@ -561,6 +567,15 @@ public async Task<int> SaveChangesAsync(bool acceptAllChangesOnSuccess, bool add
         return await base.SaveChangesAsync(acceptAllChangesOnSuccess, cancellationToken).ConfigureAwait(false);
     }
 
+    /// <summary>
+    /// Sends a synchronization event to the consumers.
+    /// </summary>
+    /// <param name="eventArgs">The event arguments.</param>
+    internal void SendSynchronizationEvent(SynchronizationEventArgs eventArgs)
+    {
+        SynchronizationProgress?.Invoke(this, eventArgs);
+    }
+
     #region IDisposable
     /// <summary>
     /// Ensure that the context has not been disposed.

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

@@ -14,6 +14,7 @@
 using System.Reflection;
 using System.Text.Json;
 using System.Text.Json.Serialization;
+using static CommunityToolkit.Datasync.Client.Offline.Operations.PullOperationManager;
 
 namespace CommunityToolkit.Datasync.Client.Offline.Operations;
 
@@ -53,61 +54,87 @@ public async Task<PullResult> ExecuteAsync(IEnumerable<PullRequest> requests, Pu
 
         QueueHandler<PullResponse> databaseUpdateQueue = new(1, async pullResponse =>
         {
-            DateTimeOffset lastSynchronization = await DeltaTokenStore.GetDeltaTokenAsync(pullResponse.QueryId, cancellationToken).ConfigureAwait(false);
-            foreach (object item in pullResponse.Items)
+            if (pullResponse.Items.Any())
             {
-                EntityMetadata metadata = EntityResolver.GetEntityMetadata(item, pullResponse.EntityType);
-                object? originalEntity = await context.FindAsync(pullResponse.EntityType, [metadata.Id], cancellationToken).ConfigureAwait(false);
-
-                if (originalEntity is null && !metadata.Deleted)
-                {
-                    _ = context.Add(item);
-                    result.IncrementAdditions();
-                }
-                else if (originalEntity is not null && metadata.Deleted)
+                DateTimeOffset lastSynchronization = await DeltaTokenStore.GetDeltaTokenAsync(pullResponse.QueryId, cancellationToken).ConfigureAwait(false);
+                foreach (object item in pullResponse.Items)
                 {
-                    _ = context.Remove(originalEntity);
-                    result.IncrementDeletions();
-                }
-                else if (originalEntity is not null && !metadata.Deleted)
-                {
-                    // Gather properties marked with [JsonIgnore]
-                    HashSet<string> ignoredProps = pullResponse.EntityType
-                        .GetProperties(BindingFlags.Public | BindingFlags.Instance)
-                        .Where(p => p.IsDefined(typeof(JsonIgnoreAttribute), inherit: true))
-                        .Select(p => p.Name)
-                        .ToHashSet();
-
-                    EntityEntry originalEntry = context.Entry(originalEntity);
-                    EntityEntry newEntry = context.Entry(item);
-
-                    // Only copy properties that are not marked with [JsonIgnore]
-                    foreach (IProperty property in originalEntry.Metadata.GetProperties())
+                    EntityMetadata metadata = EntityResolver.GetEntityMetadata(item, pullResponse.EntityType);
+                    object? originalEntity = await context.FindAsync(pullResponse.EntityType, [metadata.Id], cancellationToken).ConfigureAwait(false);
+
+                    if (originalEntity is null && !metadata.Deleted)
+                    {
+                        _ = 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)
                     {
-                        if (!ignoredProps.Contains(property.Name))
+                        // Gather properties marked with [JsonIgnore]
+                        HashSet<string> ignoredProps = pullResponse.EntityType
+                            .GetProperties(BindingFlags.Public | BindingFlags.Instance)
+                            .Where(p => p.IsDefined(typeof(JsonIgnoreAttribute), inherit: true))
+                            .Select(p => p.Name)
+                            .ToHashSet();
+
+                        EntityEntry originalEntry = context.Entry(originalEntity);
+                        EntityEntry newEntry = context.Entry(item);
+
+                        // Only copy properties that are not marked with [JsonIgnore]
+                        foreach (IProperty property in originalEntry.Metadata.GetProperties())
                         {
-                            originalEntry.Property(property.Name).CurrentValue = newEntry.Property(property.Name).CurrentValue;
+                            if (!ignoredProps.Contains(property.Name))
+                            {
+                                originalEntry.Property(property.Name).CurrentValue = newEntry.Property(property.Name).CurrentValue;
+                            }
                         }
+
+                        result.IncrementReplacements();
                     }
 
-                    result.IncrementReplacements();
+                    if (metadata.UpdatedAt > lastSynchronization)
+                    {
+                        lastSynchronization = metadata.UpdatedAt.Value;
+                        bool isAdded = await DeltaTokenStore.SetDeltaTokenAsync(pullResponse.QueryId, metadata.UpdatedAt.Value, cancellationToken).ConfigureAwait(false);
+                        if (isAdded)
+                        {
+                            // Sqlite oddity - you can't add then update; it changes the change type to UPDATE, which then fails.
+                            _ = await context.SaveChangesAsync(true, false, cancellationToken).ConfigureAwait(false);
+                        }
+                    }
                 }
 
-                if (metadata.UpdatedAt > lastSynchronization)
+                if (pullOptions.SaveAfterEveryServiceRequest)
                 {
-                    lastSynchronization = metadata.UpdatedAt.Value;
-                    bool isAdded = await DeltaTokenStore.SetDeltaTokenAsync(pullResponse.QueryId, metadata.UpdatedAt.Value, cancellationToken).ConfigureAwait(false);
-                    if (isAdded)
-                    {
-                        // Sqlite oddity - you can't add then update; it changes the change type to UPDATE, which then fails.
-                        _ = await context.SaveChangesAsync(true, false, cancellationToken).ConfigureAwait(false);
-                    }
+                    _ = await context.SaveChangesAsync(true, false, cancellationToken).ConfigureAwait(false);
                 }
+
+                context.SendSynchronizationEvent(new SynchronizationEventArgs()
+                {
+                    EventType = SynchronizationEventType.ItemsCommitted,
+                    EntityType = pullResponse.EntityType,
+                    ItemsProcessed = pullResponse.TotalItemsProcessed,
+                    ItemsTotal = pullResponse.TotalRequestItems,
+                    QueryId = pullResponse.QueryId
+                });
             }
 
-            if (pullOptions.SaveAfterEveryServiceRequest)
+            if (pullResponse.Completed)
             {
-                _ = await context.SaveChangesAsync(true, false, cancellationToken).ConfigureAwait(false);
+                context.SendSynchronizationEvent(new SynchronizationEventArgs()
+                {
+                    EventType = SynchronizationEventType.PullEnded,
+                    EntityType = pullResponse.EntityType,
+                    ItemsProcessed = pullResponse.TotalItemsProcessed,
+                    ItemsTotal = pullResponse.TotalRequestItems,
+                    QueryId = pullResponse.QueryId,
+                    Exception = pullResponse.Exception,
+                    ServiceResponse = pullResponse.Exception is DatasyncPullException ex ? ex.ServiceResponse : null
+                });
             }
         });
 
@@ -116,14 +143,34 @@ public async Task<PullResult> ExecuteAsync(IEnumerable<PullRequest> requests, Pu
             Uri endpoint = ExecutableOperation.MakeAbsoluteUri(pullRequest.HttpClient.BaseAddress, pullRequest.Endpoint);
             Uri requestUri = new UriBuilder(endpoint) { Query = pullRequest.QueryDescription.ToODataQueryString() }.Uri;
             Type pageType = typeof(Page<>).MakeGenericType(pullRequest.EntityType);
+            long itemsProcessed = 0;
+            long totalCount = 0;
 
             try
             {
                 bool completed = false;
+                // Signal we started the pull operation.
+                context.SendSynchronizationEvent(new SynchronizationEventArgs()
+                {
+                    EventType = SynchronizationEventType.PullStarted,
+                    EntityType = pullRequest.EntityType,
+                    QueryId = pullRequest.QueryId
+                });
                 do
                 {
                     Page<object> page = await GetPageAsync(pullRequest.HttpClient, requestUri, pageType, cancellationToken).ConfigureAwait(false);
-                    databaseUpdateQueue.Enqueue(new PullResponse(pullRequest.EntityType, pullRequest.QueryId, page.Items));
+                    itemsProcessed += page.Items.Count();
+                    totalCount = page.Count ?? totalCount;
+
+                    context.SendSynchronizationEvent(new SynchronizationEventArgs()
+                    {
+                        EventType = SynchronizationEventType.ItemsFetched,
+                        EntityType = pullRequest.EntityType,
+                        ItemsProcessed = itemsProcessed,
+                        ItemsTotal = page.Count ?? 0,
+                        QueryId = pullRequest.QueryId
+                    });
+
                     if (!string.IsNullOrEmpty(page.NextLink))
                     {
                         requestUri = new UriBuilder(endpoint) { Query = page.NextLink }.Uri;
@@ -132,12 +179,15 @@ public async Task<PullResult> ExecuteAsync(IEnumerable<PullRequest> requests, Pu
                     {
                         completed = true;
                     }
+
+                    databaseUpdateQueue.Enqueue(new PullResponse(pullRequest.EntityType, pullRequest.QueryId, page.Items, totalCount, itemsProcessed, completed));
                 }
                 while (!completed);
             }
             catch (DatasyncPullException ex)
             {
                 result.AddFailedRequest(requestUri, ex.ServiceResponse);
+                databaseUpdateQueue.Enqueue(new PullResponse(pullRequest.EntityType, pullRequest.QueryId, [], totalCount, itemsProcessed, true, ex));
             }
         });
 
@@ -173,6 +223,8 @@ public async Task<PullResult> ExecuteAsync(IEnumerable<PullRequest> requests, Pu
     /// <exception cref="DatasyncPullException">Thrown on error</exception>
     internal async Task<Page<object>> GetPageAsync(HttpClient client, Uri requestUri, Type pageType, CancellationToken cancellationToken = default)
     {
+        PropertyInfo countPropInfo = pageType.GetProperty("Count")
+            ?? throw new DatasyncException($"Page type '{pageType.Name}' does not have a 'Count' property");
         PropertyInfo itemsPropInfo = pageType.GetProperty("Items")
             ?? throw new DatasyncException($"Page type '{pageType.Name}' does not have an 'Items' property");
         PropertyInfo nextLinkPropInfo = pageType.GetProperty("NextLink")
@@ -193,6 +245,7 @@ internal async Task<Page<object>> GetPageAsync(HttpClient client, Uri requestUri
 
             return new Page<object>()
             {
+                Count = (long?)countPropInfo.GetValue(result),
                 Items = (IEnumerable<object>)itemsPropInfo.GetValue(result)!,
                 NextLink = (string?)nextLinkPropInfo.GetValue(result)
             };
@@ -237,6 +290,10 @@ internal static QueryDescription PrepareQueryDescription(QueryDescription source
     /// <param name="EntityType">The type of entity contained within the items.</param>
     /// <param name="QueryId">The query ID for the request.</param>
     /// <param name="Items">The list of items to process.</param>
+    /// <param name="TotalRequestItems">The total number of items in the current pull request.</param>
+    /// <param name="TotalItemsProcessed">The total number of items processed, <paramref name="Items"/> included.</param>
+    /// <param name="Completed">If <c>true</c>, indicates that the pull request is completed.</param>
+    /// <param name="Exception">Indicates an exception occured during fetching of data</param>
     [ExcludeFromCodeCoverage]
-    internal record PullResponse(Type EntityType, string QueryId, IEnumerable<object> Items);
+    internal record PullResponse(Type EntityType, string QueryId, IEnumerable<object> Items, long TotalRequestItems, long TotalItemsProcessed, bool Completed, Exception? Exception = null);
 }

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

@@ -270,16 +270,42 @@ internal async Task<PushResult> PushAsync(IEnumerable<Type> entityTypes, PushOpt
 
         // Determine the list of queued operations in scope.
         List<DatasyncOperation> queuedOperations = await GetQueuedOperationsAsync(entityTypeNames, cancellationToken).ConfigureAwait(false);
+
+        // Signal we started the push operation.
+        this._context.SendSynchronizationEvent(new SynchronizationEventArgs()
+        {
+            EventType = SynchronizationEventType.PushStarted,
+            ItemsTotal = queuedOperations.Count
+        });
+
         if (queuedOperations.Count == 0)
         {
+            // Signal we ended the push operation.
+            this._context.SendSynchronizationEvent(new SynchronizationEventArgs()
+            {
+                EventType = SynchronizationEventType.PushEnded,
+                ItemsProcessed = 0,
+                ItemsTotal = 0
+            });
             return pushResult;
         }
 
+        int nrItemsProcessed = 0;
+
         // Push things in parallel, according to the PushOptions
         QueueHandler<DatasyncOperation> queueHandler = new(pushOptions.ParallelOperations, async operation =>
         {
             ServiceResponse? response = await PushOperationAsync(operation, cancellationToken).ConfigureAwait(false);
             pushResult.AddOperationResult(operation, response);
+            // We can run on multiple threads, so use Interlocked to update the number of items processed.
+            int newItemsProcessed = Interlocked.Increment(ref nrItemsProcessed);
+            this._context.SendSynchronizationEvent(new SynchronizationEventArgs()
+            {
+                EventType = SynchronizationEventType.PushItem,
+                ItemsProcessed = newItemsProcessed,
+                ItemsTotal = queuedOperations.Count,
+                PushOperation = operation,
+            });
         });
 
         // Enqueue and process all the queued operations in scope
@@ -288,6 +314,14 @@ internal async Task<PushResult> PushAsync(IEnumerable<Type> entityTypes, PushOpt
 
         // Save the changes, this time we don't update the queue.
         _ = await this._context.SaveChangesAsync(acceptAllChangesOnSuccess: true, addToQueue: false, cancellationToken).ConfigureAwait(false);
+
+        this._context.SendSynchronizationEvent(new SynchronizationEventArgs()
+        {
+            EventType = SynchronizationEventType.PushEnded,
+            ItemsProcessed = nrItemsProcessed,
+            ItemsTotal = queuedOperations.Count,
+        });
+
         return pushResult;
     }