Merge branch 'main' into chore/formatting and run dotnet format

Author: LucDeCaf

PowerSync/PowerSync.Common/CHANGELOG.md

@@ -1,7 +1,31 @@
 # PowerSync.Common Changelog
 
+## 0.0.6-alpha.1
+- Updated to the latest version (0.4.10) of the core extension.
+- Dropping support for the legacy C# sync implementation.
+- Add `trackPreviousValues` option on `TableOptions` which sets `CrudEntry.PreviousValues` to previous values on updates.
+- Add `trackMetadata` option on `TableOptions` which adds a `_metadata` column that can be used for updates. The configured metadata is available through `CrudEntry.Metadata`.
+- Add `ignoreEmptyUpdates` option on `TableOptions` which skips creating CRUD entries for updates that don't change any values.
+- Reporting progress information about downloaded rows. Sync progress is available through `SyncStatus.DownloadProgress()`.
+- Support bucket priorities.
+- Report `PriorityStatusEntries` on `SyncStatus`.
+- Added ability to specify `AppMetadata` for sync/stream requests.
+
+Note: This requires a PowerSync service version `>=1.17.0` in order for logs to display metadata.
+
+```csharp
+db.Connect(connector, new PowerSync.Common.Client.Sync.Stream.PowerSyncConnectionOptions
+{
+    // This will be included in PowerSync service logs
+    AppMetadata = new Dictionary<string, string>
+    {
+        { "app_version", myAppVersion },
+    }
+});
+```
+
 ## 0.0.5-alpha.1
-- Using the latest (0.4.9) version of the core extension, it introduces support for the Rust Sync implementation and also makes it the default - users can still opt out and use the legacy C# sync implementation as option when calling `connect()`.
+- Using the latest version (0.4.9) of the core extension, it introduces support for the Rust Sync implementation and also makes it the default - users can still opt out and use the legacy C# sync implementation as option when calling `connect()`.
 
 ## 0.0.4-alpha.1
 - Fixed MAUI issues related to extension loading when installing package outside of the monorepo. 
@@ -25,4 +49,4 @@
 * linux-x64
 * osx-arm64
 * osx-x64
-* wind-x64
+* wind-x64

PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs

@@ -80,8 +80,6 @@ public interface IPowerSyncDatabase : IEventStream<PowerSyncDBEvent>
 
 public class PowerSyncDatabase : EventStream<PowerSyncDBEvent>, IPowerSyncDatabase
 {
-    private static readonly int FULL_SYNC_PRIORITY = 2147483647;
-
     public IDBAdapter Database;
     private Schema schema;
 
@@ -156,21 +154,48 @@ public async Task WaitForReady()
         await isReadyTask;
     }
 
-    public async Task WaitForFirstSync(CancellationToken? cancellationToken = null)
+    public class PrioritySyncRequest
+    {
+        public CancellationToken? Token { get; set; }
+        public int? Priority { get; set; }
+    }
+
+    /// <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;
+        var cancellationToken = request?.Token;
+
+        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);
@@ -192,7 +217,7 @@ protected async Task Initialize()
         await BucketStorageAdapter.Init();
         await LoadVersion();
         await UpdateSchema(schema);
-        await UpdateHasSynced();
+        await ResolveOfflineSyncStatus();
         await Database.Execute("PRAGMA RECURSIVE_TRIGGERS=TRUE");
         Ready = true;
         Emit(new PowerSyncDBEvent { Initialized = true });
@@ -216,48 +241,29 @@ private async Task LoadVersion()
         catch (Exception e)
         {
             throw new Exception(
-                $"Unsupported PowerSync extension version. Need >=0.2.0 <1.0.0, got: {sdkVersion}. Details: {e.Message}"
+                $"Unsupported PowerSync extension version. Need >=0.4.5 <1.0.0, got: {sdkVersion}. Details: {e.Message}"
             );
         }
 
-        // Validate version is >= 0.2.0 and < 1.0.0
-        if (versionInts[0] != 0 || versionInts[1] < 2 || versionInts[2] < 0)
+        // Validate version is >= 0.4.5 and < 1.0.0
+        if (versionInts[0] != 0 || versionInts[1] < 4 || (versionInts[1] == 4 && versionInts[2] < 5))
         {
-            throw new Exception($"Unsupported PowerSync extension version. Need >=0.2.0 <1.0.0, got: {sdkVersion}");
+            throw new Exception($"Unsupported PowerSync extension version. Need >=0.4.5 <1.0.0, got: {sdkVersion}");
         }
     }
 
-    private record LastSyncedResult(int? priority, string? last_synced_at);
-
-    protected async Task UpdateHasSynced()
+    private record OfflineSyncStatusResult(string r);
+    protected async Task ResolveOfflineSyncStatus()
     {
-        var results = await Database.GetAll<LastSyncedResult>(
-            "SELECT priority, last_synced_at FROM ps_sync_state ORDER BY priority DESC"
-        );
+        var result = await Database.Get<OfflineSyncStatusResult>("SELECT powersync_offline_sync_status() as r");
+        var parsed = JsonConvert.DeserializeObject<CoreSyncStatus>(result.r);
 
-        DateTime? lastCompleteSync = null;
+        var parsedSyncStatus = CoreInstructionHelpers.CoreStatusToSyncStatus(parsed!);
+        var updatedStatus = CurrentStatus.CreateUpdatedStatus(parsedSyncStatus);
 
-        // TODO: Will be altered/extended when reporting individual sync priority statuses is supported 
-        foreach (var result in results)
+        if (!updatedStatus.IsEqual(CurrentStatus))
         {
-            var parsedDate = DateTime.Parse(result.last_synced_at + "Z");
-
-            if (result.priority == FULL_SYNC_PRIORITY)
-            {
-                // This lowest-possible priority represents a complete sync.
-                lastCompleteSync = parsedDate;
-            }
-        }
-
-        var hasSynced = lastCompleteSync != null;
-        if (hasSynced != CurrentStatus.HasSynced)
-        {
-            CurrentStatus = new SyncStatus(new SyncStatusOptions(CurrentStatus.Options)
-            {
-                HasSynced = hasSynced,
-                LastSyncedAt = lastCompleteSync,
-            });
-
+            CurrentStatus = updatedStatus;
             Emit(new PowerSyncDBEvent { StatusChanged = CurrentStatus });
         }
     }
@@ -482,7 +488,7 @@ await Database.WriteTransaction(async tx =>
                 return null;
             }
 
-            long? txId = first.TransactionId ?? null;
+            long? txId = first.TransactionId;
             List<CrudEntry> all;
 
             if (txId == null)

PowerSync/PowerSync.Common/Client/Sync/Bucket/BucketStorageAdapter.cs

@@ -96,15 +96,6 @@ public class BucketStorageEvent
 public interface IBucketStorageAdapter : IEventStream<BucketStorageEvent>
 {
     Task Init();
-    Task SaveSyncData(SyncDataBatch batch);
-    Task RemoveBuckets(string[] buckets);
-    Task SetTargetCheckpoint(Checkpoint checkpoint);
-
-    void StartSession();
-
-    Task<BucketState[]> GetBucketStates();
-
-    Task<SyncLocalDatabaseResult> SyncLocalDatabase(Checkpoint checkpoint);
 
     Task<CrudEntry?> NextCrudItem();
     Task<bool> HasCrud();
@@ -113,12 +104,6 @@ public interface IBucketStorageAdapter : IEventStream<BucketStorageEvent>
     Task<bool> HasCompletedSync();
     Task<bool> UpdateLocalTarget(Func<Task<string>> callback);
 
-    /// <summary>
-    /// Exposed for tests only.
-    /// </summary>
-    Task AutoCompact();
-    Task ForceCompact();
-
     string GetMaxOpId();
 
     /// <summary>