(#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.

Author: StefanCuypers

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,70 +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);
+                    EntityMetadata metadata = EntityResolver.GetEntityMetadata(item, pullResponse.EntityType);
+                    object? originalEntity = await context.FindAsync(pullResponse.EntityType, [metadata.Id], cancellationToken).ConfigureAwait(false);
 
-                    // Only copy properties that are not marked with [JsonIgnore]
-                    foreach (IProperty property in originalEntry.Metadata.GetProperties())
+                    if (originalEntity is null && !metadata.Deleted)
                     {
-                        if (!ignoredProps.Contains(property.Name))
+                        _ = 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)
+                    {
+                        // 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,
-                TotalNrItems = pullResponse.TotalRequestItems,
-                QueryId = pullResponse.QueryId
-            });
+                context.SendSynchronizationEvent(new SynchronizationEventArgs()
+                {
+                    EventType = SynchronizationEventType.ItemsCommitted,
+                    EntityType = pullResponse.EntityType,
+                    ItemsProcessed = pullResponse.TotalItemsProcessed,
+                    TotalNrItems = 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,
+                    TotalNrItems = pullResponse.TotalRequestItems,
+                    QueryId = pullResponse.QueryId,
+                    Exception = pullResponse.Exception,
+                    ServiceResponse = pullResponse.Exception is DatasyncPullException ex ? ex.ServiceResponse : null
+                });
             }
         });
 
@@ -125,15 +143,24 @@ 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;
-                long itemsProcessed = 0;
+                // 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);
                     itemsProcessed += page.Items.Count();
+                    totalCount = page.Count ?? totalCount;
 
                     context.SendSynchronizationEvent(new SynchronizationEventArgs()
                     {
@@ -144,7 +171,6 @@ public async Task<PullResult> ExecuteAsync(IEnumerable<PullRequest> requests, Pu
                         QueryId = pullRequest.QueryId
                     });
 
-                    databaseUpdateQueue.Enqueue(new PullResponse(pullRequest.EntityType, pullRequest.QueryId, page.Items, page.Count ?? 0, itemsProcessed));
                     if (!string.IsNullOrEmpty(page.NextLink))
                     {
                         requestUri = new UriBuilder(endpoint) { Query = page.NextLink }.Uri;
@@ -153,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, Enumerable.Empty<object>(), totalCount, itemsProcessed, true, ex));
             }
         });
 
@@ -263,6 +292,8 @@ internal static QueryDescription PrepareQueryDescription(QueryDescription source
     /// <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, long TotalRequestItems, long TotalItemsProcessed);
+    internal record PullResponse(Type EntityType, string QueryId, IEnumerable<object> Items, long TotalRequestItems, long TotalItemsProcessed, bool Completed, Exception? Exception = null);
 }

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

@@ -9,6 +9,11 @@ namespace CommunityToolkit.Datasync.Client.Offline;
 /// </summary>
 public enum SynchronizationEventType
 {
+    /// <summary>
+    /// Pull for the given entity starts.
+    /// </summary>
+    /// <remarks><see cref="SynchronizationEventArgs.ItemsProcessed"/> is not yet known here</remarks>
+    PullStarted,
     /// <summary>
     /// Occurs when items have been successfully fetches from the server.
     /// </summary>
@@ -20,6 +25,11 @@ public enum SynchronizationEventType
     /// </summary>
     /// <remarks>This event is raised after a page of entities was succesfully commited to the database</remarks>
     ItemsCommitted,
+
+    /// <summary>
+    /// Pull for the given entiry ended.
+    /// </summary>
+    PullEnded,
 }
 
 /// <summary>
@@ -51,4 +61,14 @@ public class SynchronizationEventArgs
     /// The query ID that is being processed
     /// </summary>
     public required string QueryId { get; init; }
+
+    /// <summary>
+    /// If not <c>null</c> on event type <see cref="SynchronizationEventType.PullEnded"/>, indicates pull failed with this exception.
+    /// </summary>
+    public Exception? Exception { get; init; }
+
+    /// <summary>
+    /// If a <see cref="DatasyncException"/> occured in <see cref="Exception"/> during server call processing, this property has more detail on the server response.
+    /// </summary>
+    public ServiceResponse? ServiceResponse { get; init; }
 }

tests/CommunityToolkit.Datasync.Client.Test/Offline/OfflineDbContext_Tests.cs

@@ -1447,8 +1447,10 @@ public async Task SynchronizationProgress_Event_Works()
         this.context.Handler.AddResponse(HttpStatusCode.OK, page3);
         this.context.Handler.AddResponse(HttpStatusCode.OK, page4);
 
-        bool eventFiredForFetch = true;
-        bool eventFiredForCommit = true;
+        bool eventFiredForFetch = false;
+        bool eventFiredForCommit = false;
+        bool eventFiredForStart = false;
+        bool eventFiredForEnd = false;
         long currentItemsFetched = 0;
         long currentItemsCommited = 0;
 
@@ -1457,19 +1459,32 @@ public async Task SynchronizationProgress_Event_Works()
             sender.Should().Be(this.context);
             args.EntityType.Should().Be<ClientMovie>();
             args.QueryId.Should().Be("CommunityToolkit.Datasync.TestCommon.Databases.ClientMovie");
-            args.TotalNrItems.Should().Be(20);
-            switch(args.EventType)
+            args.Exception.Should().BeNull();   // We don't test exceptions here, so should always be null.
+            args.ServiceResponse.Should().BeNull();
+            switch (args.EventType)
             {
                 case SynchronizationEventType.ItemsFetched:
                     currentItemsFetched += 5;
                     args.ItemsProcessed.Should().Be(currentItemsFetched);
+                    args.TotalNrItems.Should().Be(20);
                     eventFiredForFetch = true;
                     break;
                 case SynchronizationEventType.ItemsCommitted:
                     currentItemsCommited += 5;
                     args.ItemsProcessed.Should().Be(currentItemsCommited);
+                    args.TotalNrItems.Should().Be(20);
                     eventFiredForCommit = true;
                     break;
+                case SynchronizationEventType.PullStarted:
+                    eventFiredForStart.Should().BeFalse("PullStarted event should only fire once");
+                    eventFiredForStart = true;
+                    break;
+                case SynchronizationEventType.PullEnded:
+                    eventFiredForEnd.Should().BeFalse("PullEnded event should only fire once");
+                    eventFiredForEnd = true;
+                    args.ItemsProcessed.Should().Be(20);
+                    args.TotalNrItems.Should().Be(20);
+                    break;
                 default:
                     Assert.Fail($"Invalid event type: {args.EventType}");
                     break;
@@ -1478,8 +1493,10 @@ public async Task SynchronizationProgress_Event_Works()
 
         await this.context.Movies.PullAsync();
 
+        eventFiredForStart.Should().BeTrue();
         eventFiredForFetch.Should().BeTrue();
         eventFiredForCommit.Should().BeTrue();
+        eventFiredForEnd.Should().BeTrue();
         currentItemsFetched.Should().Be(20);
         currentItemsCommited.Should().Be(20);
     }