Updated WaitForFirstSync to support priorities. Removing old sync from remote.

Author: Chriztiaan

PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs

@@ -158,22 +158,49 @@ public async Task WaitForReady()
 
         await isReadyTask;
     }
+    
+    public class PrioritySyncRequest
+    {
+        public CancellationToken? Token { get; set; }
+        public int? Priority { get; set; }
+    }
 
-    public async Task WaitForFirstSync(CancellationToken? cancellationToken = null)
+    /// <summary>
+    /// Wait for the first sync operation to complete.
+    /// </summary>
+    /// <param name="request">
+    /// An object providing a cancellation token and a priority target.
+    /// When a priority target is set, the task may complete when all buckets with the given (or higher)
+    /// priorities have been synchronized. This can be earlier than a complete sync.
+    /// </param>
+    /// <returns>A task which will complete once the first full sync has completed.</returns>
+    public async Task WaitForFirstSync(PrioritySyncRequest? request = null)
     {
-        if (CurrentStatus.HasSynced == true)
+        var priority =  request?.Priority ?? null;
+        var cancellationToken = request?.Token ?? null;
+
+        bool StatusMatches(SyncStatus status)
+        {
+            if (priority == null)
+            {
+                return status.HasSynced == true;
+            }
+            return status.StatusForPriority(priority.Value).HasSynced == true;
+        }
+
+        if (StatusMatches(CurrentStatus))
         {
             return;
         }
 
         var tcs = new TaskCompletionSource<bool>();
         var cts = new CancellationTokenSource();
 
-        var _ = Task.Run(() =>
+        _ = Task.Run(() =>
         {
             foreach (var update in Listen(cts.Token))
             {
-                if (update.StatusChanged?.HasSynced == true)
+                if  (update.StatusChanged != null && StatusMatches(update.StatusChanged!))
                 {
                     cts.Cancel();
                     tcs.SetResult(true);
@@ -230,7 +257,7 @@ private async Task LoadVersion()
         }
     }
 
-    private record LastSyncedResult(int? priority, string? last_synced_at);
+    private record LastSyncedResult(int priority, string? last_synced_at);
 
     protected async Task UpdateHasSynced()
     {
@@ -239,6 +266,7 @@ protected async Task UpdateHasSynced()
         );
 
         DateTime? lastCompleteSync = null;
+        List<DB.Crud.SyncPriorityStatus> priorityStatuses = [];
 
         foreach (var result in results)
         {
@@ -249,17 +277,28 @@ protected async Task UpdateHasSynced()
                 // This lowest-possible priority represents a complete sync.
                 lastCompleteSync = parsedDate;
             }
+            else
+            {
+                priorityStatuses.Add(new DB.Crud.SyncPriorityStatus
+                {
+                    Priority = result.priority,
+                    HasSynced = true,
+                    LastSyncedAt = parsedDate
+                });
+            }
         }
 
         var hasSynced = lastCompleteSync != null;
-        if (hasSynced != CurrentStatus.HasSynced)
+        var updatedStatus = new SyncStatus(new SyncStatusOptions(CurrentStatus.Options)
         {
-            CurrentStatus = new SyncStatus(new SyncStatusOptions(CurrentStatus.Options)
-            {
-                HasSynced = hasSynced,
-                LastSyncedAt = lastCompleteSync,
-            });
-
+            HasSynced = hasSynced,
+            PriorityStatusEntries = priorityStatuses.ToArray(),
+            LastSyncedAt = lastCompleteSync,
+        });
+        
+        if (!updatedStatus.IsEqual(CurrentStatus))
+        {
+            CurrentStatus = updatedStatus;
             Emit(new PowerSyncDBEvent { StatusChanged = CurrentStatus });
         }
     }
@@ -536,7 +575,7 @@ await tx.Execute(
     }
 
     /// <summary>
-    /// Get an unique client id for this database.
+    /// Get a unique client id for this database.
     ///
     /// The id is not reset when the database is cleared, only when the database is deleted.
     /// </summary>

PowerSync/PowerSync.Common/Client/Sync/Stream/Remote.cs

@@ -16,18 +16,13 @@ namespace PowerSync.Common.Client.Sync.Stream;
 public class SyncStreamOptions
 {
     public string Path { get; set; } = "";
+    
     public StreamingSyncRequest Data { get; set; } = new();
     public Dictionary<string, string> Headers { get; set; } = new();
 
     public CancellationToken CancellationToken { get; set; } = CancellationToken.None;
 }
 
-public class RequestDetails
-{
-    public string Url { get; set; } = "";
-    public Dictionary<string, string> Headers { get; set; } = new();
-}
-
 public class Remote
 {
     private readonly HttpClient httpClient;
@@ -120,34 +115,6 @@ public async Task<T> Get<T>(string path, Dictionary<string, string>? headers = n
         var responseData = await response.Content.ReadAsStringAsync();
         return JsonConvert.DeserializeObject<T>(responseData)!;
     }
-
-    public async IAsyncEnumerable<StreamingSyncLine?> OldPostStream(SyncStreamOptions options)
-    {
-        using var requestMessage = await BuildRequest(HttpMethod.Post, options.Path, options.Data, options.Headers);
-        using var response = await httpClient.SendAsync(requestMessage, HttpCompletionOption.ResponseHeadersRead, options.CancellationToken);
-
-        if (response.Content == null)
-        {
-            throw new HttpRequestException($"HTTP {response.StatusCode}: No content");
-        }
-        else
-        if (!response.IsSuccessStatusCode)
-        {
-            var errorText = await response.Content.ReadAsStringAsync();
-            throw new HttpRequestException($"HTTP {response.StatusCode}: {errorText}");
-        }
-
-        var stream = await response.Content.ReadAsStreamAsync();
-
-        // Read NDJSON stream
-        using var reader = new StreamReader(stream, Encoding.UTF8);
-        string? line;
-
-        while ((line = await reader.ReadLineAsync()) != null)
-        {
-            yield return ParseStreamingSyncLine(JObject.Parse(line));
-        }
-    }
 
      /// <summary>
     /// Posts to the stream endpoint and returns a raw NDJSON stream that can be read line by line.
@@ -175,35 +142,7 @@ public async Task<Stream> PostStreamRaw(SyncStreamOptions options)
         return await response.Content.ReadAsStreamAsync();
     }
 
-    public static StreamingSyncLine? ParseStreamingSyncLine(JObject json)
-    {
-        // Determine the type based on available keys
-        if (json.ContainsKey("checkpoint"))
-        {
-            return json.ToObject<StreamingSyncCheckpoint>();
-        }
-        else if (json.ContainsKey("checkpoint_diff"))
-        {
-            return json.ToObject<StreamingSyncCheckpointDiff>();
-        }
-        else if (json.ContainsKey("checkpoint_complete"))
-        {
-            return json.ToObject<StreamingSyncCheckpointComplete>();
-        }
-        else if (json.ContainsKey("data"))
-        {
-            return json.ToObject<StreamingSyncDataJSON>();
-        }
-        else if (json.ContainsKey("token_expires_in"))
-        {
-            return json.ToObject<StreamingSyncKeepalive>();
-        }
-        else
-        {
-            return null;
-        }
-    }
-
+  
     private async Task<HttpRequestMessage> BuildRequest(HttpMethod method, string path, object? data = null, Dictionary<string, string>? additionalHeaders = null)
     {
         var credentials = await GetCredentials();

PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncImplementation.cs

@@ -332,10 +332,10 @@ protected async Task StreamingSync(CancellationToken? signal, PowerSyncConnectio
         });
     }
 
-    protected async Task<bool> StreamingSyncIteration(CancellationToken signal,
+    protected async Task StreamingSyncIteration(CancellationToken signal,
         PowerSyncConnectionOptions? options)
     {
-        return await locks.ObtainLock(new LockOptions<bool>
+        await locks.ObtainLock(new LockOptions<bool>
         {
             Type = LockType.SYNC,
             Token = signal,
@@ -473,11 +473,9 @@ async Task HandleInstruction(Instruction instruction)
                                 // DownloadProgress = info.Downloading?.Buckets
                             }
                         },
-                        // TODO handle errors later?
                         new UpdateSyncStatusOptions
                         {
-                            // ClearDownloadError = true,
-                            // ClearUploadError = true
+                            ClearDownloadError = true,
                         }
                     );
 

PowerSync/PowerSync.Common/Client/Sync/Stream/StreamingSyncTypes.cs

@@ -4,54 +4,6 @@ namespace PowerSync.Common.Client.Sync.Stream;
 using PowerSync.Common.DB.Crud;
 using Newtonsoft.Json;
 
-public class ContinueCheckpointRequest
-{
-    [JsonProperty("buckets")]
-    public List<BucketRequest> Buckets { get; set; } = new();
-
-    [JsonProperty("checkpoint_token")]
-    public string CheckpointToken { get; set; } = "";
-
-    [JsonProperty("limit")]
-    public int? Limit { get; set; }
-}
-
-public class SyncNewCheckpointRequest
-{
-    [JsonProperty("buckets")]
-    public List<BucketRequest>? Buckets { get; set; }
-
-    [JsonProperty("request_checkpoint")]
-    public RequestCheckpoint RequestCheckpoint { get; set; } = new();
-
-    [JsonProperty("limit")]
-    public int? Limit { get; set; }
-}
-
-public class RequestCheckpoint
-{
-    [JsonProperty("include_data")]
-    public bool IncludeData { get; set; }
-
-    [JsonProperty("include_checksum")]
-    public bool IncludeChecksum { get; set; }
-}
-
-public class SyncResponse
-{
-    [JsonProperty("data")]
-    public List<SyncDataBucketJSON>? Data { get; set; }
-
-    [JsonProperty("has_more")]
-    public bool HasMore { get; set; }
-
-    [JsonProperty("checkpoint_token")]
-    public string? CheckpointToken { get; set; }
-
-    [JsonProperty("checkpoint")]
-    public Checkpoint? Checkpoint { get; set; }
-}
-
 public class StreamingSyncRequest
 {
     [JsonProperty("buckets")]
@@ -80,69 +32,4 @@ public class BucketRequest
 
     [JsonProperty("after")]
     public string After { get; set; } = "";
-}
-
-public abstract class StreamingSyncLine { }
-
-public class StreamingSyncCheckpoint : StreamingSyncLine
-{
-    [JsonProperty("checkpoint")]
-    public Checkpoint Checkpoint { get; set; } = new();
-}
-
-public class StreamingSyncCheckpointDiff : StreamingSyncLine
-{
-    [JsonProperty("checkpoint_diff")]
-    public CheckpointDiff CheckpointDiff { get; set; } = new();
-}
-
-public class CheckpointDiff
-{
-    [JsonProperty("last_op_id")]
-    public string LastOpId { get; set; } = "";
-
-    [JsonProperty("updated_buckets")]
-    public List<BucketChecksum> UpdatedBuckets { get; set; } = new();
-
-    [JsonProperty("removed_buckets")]
-    public List<string> RemovedBuckets { get; set; } = new();
-
-    [JsonProperty("write_checkpoint")]
-    public string WriteCheckpoint { get; set; } = "";
-}
-
-public class StreamingSyncDataJSON : StreamingSyncLine
-{
-    [JsonProperty("data")]
-    public SyncDataBucketJSON Data { get; set; } = new();
-}
-
-public class StreamingSyncCheckpointComplete : StreamingSyncLine
-{
-    [JsonProperty("checkpoint_complete")]
-    public CheckpointComplete CheckpointComplete { get; set; } = new();
-}
-
-public class CheckpointComplete
-{
-    [JsonProperty("last_op_id")]
-    public string LastOpId { get; set; } = "";
-}
-
-public class StreamingSyncKeepalive : StreamingSyncLine
-{
-    [JsonProperty("token_expires_in")]
-    public int? TokenExpiresIn { get; set; }
-}
-
-public class CrudRequest
-{
-    [JsonProperty("data")]
-    public List<CrudEntry> Data { get; set; } = new();
-}
-
-public class CrudResponse
-{
-    [JsonProperty("checkpoint")]
-    public string? Checkpoint { get; set; }
-}
+}