Merge branch 'feat/rust-sync-implementation' into feat/api-catchup-effort-2

Author: Chriztiaan

.github/workflows/test.yml

@@ -17,6 +17,9 @@ jobs:
         with:
           dotnet-version: '8.0' 
 
+      - name: Install MAUI Workloads
+        run: dotnet workload restore
+        
       - name: Download PowerSync extension
         run: dotnet run --project Tools/Setup    
 

Directory.build.props

@@ -1,5 +1,8 @@
 <?xml version="1.0" encoding="utf-8"?>
 <Project>
+    <PropertyGroup>
+        <MSBuildWarningsAsMessages>$(MSBuildWarningsAsMessages);NETSDK1202</MSBuildWarningsAsMessages>
+    </PropertyGroup>
     <ItemGroup>
         <Compile Include="$(MSBuildThisFileDirectory)IsExternalInit.cs" Visible="false" />
     </ItemGroup>

PowerSync/PowerSync.Common/CHANGELOG.md

@@ -1,5 +1,8 @@
 # PowerSync.Common Changelog
 
+## 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()`.
+
 ## 0.0.4-alpha.1
 - Fixed MAUI issues related to extension loading when installing package outside of the monorepo. 
 

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

@@ -8,6 +8,17 @@ namespace PowerSync.Common.Client.Sync.Bucket;
 using PowerSync.Common.DB.Crud;
 using PowerSync.Common.Utils;
 
+public static class PowerSyncControlCommand
+{
+    public const string PROCESS_TEXT_LINE = "line_text";
+    public const string PROCESS_BSON_LINE = "line_binary";
+    public const string STOP = "stop";
+    public const string START = "start";
+    public const string NOTIFY_TOKEN_REFRESHED = "refreshed_token";
+    public const string NOTIFY_CRUD_UPLOAD_COMPLETED = "completed_upload";
+    public const string UPDATE_SUBSCRIPTIONS = "update_subscriptions";
+}
+
 public class Checkpoint
 {
     [JsonProperty("last_op_id")]
@@ -52,7 +63,7 @@ public class BucketStorageEvent
 public interface IBucketStorageAdapter : IEventStream<BucketStorageEvent>
 {
     Task Init();
-    
+
     Task<CrudEntry?> NextCrudItem();
     Task<bool> HasCrud();
     Task<CrudBatch?> GetCrudBatch(int limit = 100);
@@ -71,7 +82,7 @@ public interface IBucketStorageAdapter : IEventStream<BucketStorageEvent>
     /// Get a unique client ID.
     /// </summary>
     Task<string> GetClientId();
-    
+
     /// <summary>
     /// Invokes the `powersync_control` function for the sync client.
     /// </summary>

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

@@ -89,7 +89,166 @@ public string GetMaxOpId()
     {
         return MAX_OP_ID;
     }
-    
+
+    public void StartSession() { }
+
+    public async Task<BucketState[]> GetBucketStates()
+    {
+        return
+            await db.GetAll<BucketState>("SELECT name as bucket, cast(last_op as TEXT) as op_id FROM ps_buckets WHERE pending_delete = 0 AND name != '$local'");
+    }
+
+    public async Task SaveSyncData(SyncDataBatch batch)
+    {
+        await db.WriteTransaction(async tx =>
+        {
+            int count = 0;
+            foreach (var b in batch.Buckets)
+            {
+                var result = await tx.Execute("INSERT INTO powersync_operations(op, data) VALUES(?, ?)",
+                    ["save", JsonConvert.SerializeObject(new { buckets = new[] { JsonConvert.DeserializeObject(b.ToJSON()) } })]);
+                logger.LogDebug("saveSyncData {message}", JsonConvert.SerializeObject(result));
+                count += b.Data.Length;
+            }
+            compactCounter += count;
+        });
+    }
+
+    public async Task RemoveBuckets(string[] buckets)
+    {
+        foreach (var bucket in buckets)
+        {
+            await DeleteBucket(bucket);
+        }
+    }
+
+    private async Task DeleteBucket(string bucket)
+    {
+        await db.WriteTransaction(async tx =>
+        {
+            await tx.Execute("INSERT INTO powersync_operations(op, data) VALUES(?, ?)",
+                ["delete_bucket", bucket]);
+        });
+
+        logger.LogDebug("Done deleting bucket");
+        pendingBucketDeletes = true;
+    }
+
+    private record LastSyncedResult(string? synced_at);
+    public async Task<bool> HasCompletedSync()
+    {
+        if (hasCompletedSync) return true;
+
+        var result = await db.Get<LastSyncedResult>("SELECT powersync_last_synced_at() as synced_at");
+
+        hasCompletedSync = result.synced_at != null;
+        return hasCompletedSync;
+    }
+
+    public async Task<SyncLocalDatabaseResult> SyncLocalDatabase(Checkpoint checkpoint)
+    {
+        var validation = await ValidateChecksums(checkpoint);
+        if (!validation.CheckpointValid)
+        {
+            logger.LogError("Checksums failed for {failures}", JsonConvert.SerializeObject(validation.CheckpointFailures));
+            foreach (var failedBucket in validation.CheckpointFailures ?? [])
+            {
+                await DeleteBucket(failedBucket);
+            }
+            return new SyncLocalDatabaseResult
+            {
+                Ready = false,
+                CheckpointValid = false,
+                CheckpointFailures = validation.CheckpointFailures
+            };
+        }
+
+        var bucketNames = checkpoint.Buckets.Select(b => b.Bucket).ToArray();
+        await db.WriteTransaction(async tx =>
+        {
+            await tx.Execute(
+                "UPDATE ps_buckets SET last_op = ? WHERE name IN (SELECT json_each.value FROM json_each(?))",
+                [checkpoint.LastOpId, JsonConvert.SerializeObject(bucketNames)]
+            );
+
+            if (checkpoint.WriteCheckpoint != null)
+            {
+                await tx.Execute(
+                    "UPDATE ps_buckets SET last_op = ? WHERE name = '$local'",
+                    [checkpoint.WriteCheckpoint]
+                );
+            }
+        });
+
+        var valid = await UpdateObjectsFromBuckets(checkpoint);
+        if (!valid)
+        {
+            logger.LogDebug("Not at a consistent checkpoint - cannot update local db");
+            return new SyncLocalDatabaseResult
+            {
+                Ready = false,
+                CheckpointValid = true
+            };
+        }
+
+        await ForceCompact();
+
+        return new SyncLocalDatabaseResult
+        {
+            Ready = true,
+            CheckpointValid = true
+        };
+    }
+
+    private async Task<bool> UpdateObjectsFromBuckets(Checkpoint checkpoint)
+    {
+        return await db.WriteTransaction(async tx =>
+        {
+            var result = await tx.Execute("INSERT INTO powersync_operations(op, data) VALUES(?, ?)",
+                                           ["sync_local", ""]);
+
+            return result.InsertId == 1;
+        });
+    }
+
+    private record ResultResult(object result);
+
+    public class ResultDetail
+    {
+        [JsonProperty("valid")]
+        public bool Valid { get; set; }
+
+        [JsonProperty("failed_buckets")]
+        public List<string>? FailedBuckets { get; set; }
+    }
+
+    public async Task<SyncLocalDatabaseResult> ValidateChecksums(
+        Checkpoint checkpoint)
+    {
+        var result = await db.Get<ResultResult>("SELECT powersync_validate_checkpoint(?) as result",
+                [JsonConvert.SerializeObject(checkpoint)]);
+
+        logger.LogDebug("validateChecksums result item {message}", JsonConvert.SerializeObject(result));
+
+        if (result == null) return new SyncLocalDatabaseResult { CheckpointValid = false, Ready = false };
+
+        var resultDetail = JsonConvert.DeserializeObject<ResultDetail>(result.result.ToString() ?? "{}");
+
+        if (resultDetail?.Valid == true)
+        {
+            return new SyncLocalDatabaseResult { Ready = true, CheckpointValid = true };
+        }
+        else
+        {
+            return new SyncLocalDatabaseResult
+            {
+                CheckpointValid = false,
+                Ready = false,
+                CheckpointFailures = resultDetail?.FailedBuckets?.ToArray() ?? []
+            };
+        }
+    }
+
     /// <summary>
     /// Force a compact operation, primarily for testing purposes.
     /// </summary>
@@ -267,16 +426,15 @@ public async Task<bool> HasCrud()
     {
         return await db.GetOptional<object>("SELECT 1 as ignore FROM ps_crud LIMIT 1") != null;
     }
-    
 
     record ControlResult(string? r);
 
     public async Task<string> Control(string op, object? payload = null)
     {
         return await db.WriteTransaction(async tx =>
         {
-            var result = await tx.Get<ControlResult>("SELECT powersync_control(?, ?) AS r", [op, payload ?? ""]);
+            var result = await tx.Get<ControlResult>("SELECT powersync_control(?, ?) AS r", [op, payload]);
             return result.r!;
         });
     }
-}
+}

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

@@ -1,9 +1,8 @@
 using Newtonsoft.Json.Linq;
+using Newtonsoft.Json;
 
 namespace PowerSync.Common.Client.Sync.Stream;
 
-using Newtonsoft.Json;
-
 /// <summary>
 /// An internal instruction emitted by the sync client in the core extension in response to the
 /// SDK passing sync data into the extension.
@@ -15,7 +14,6 @@ public static Instruction[] ParseInstructions(string rawResponse)
     {
         var jsonArray = JArray.Parse(rawResponse);
         List<Instruction> instructions = [];
-        
         foreach (JObject item in jsonArray)
         {
             var instruction = ParseInstruction(item);
@@ -25,7 +23,6 @@ public static Instruction[] ParseInstructions(string rawResponse)
             }
             instructions.Add(instruction);
         }
-        
         return instructions.ToArray();
     }
 
@@ -45,12 +42,11 @@ public static Instruction[] ParseInstructions(string rawResponse)
             return new FlushFileSystem();
         if (json.ContainsKey("DidCompleteSync"))
             return new DidCompleteSync();
-      
         throw new JsonSerializationException("Unknown Instruction type.");
     }
 }
 
-public class LogLine: Instruction
+public class LogLine : Instruction
 {
     [JsonProperty("severity")]
     public string Severity { get; set; } = null!;  // "DEBUG", "INFO", "WARNING"
@@ -59,13 +55,13 @@ public class LogLine: Instruction
     public string Line { get; set; } = null!;
 }
 
-public class EstablishSyncStream: Instruction
+public class EstablishSyncStream : Instruction
 {
     [JsonProperty("request")]
     public StreamingSyncRequest Request { get; set; } = null!;
 }
 
-public class UpdateSyncStatus: Instruction
+public class UpdateSyncStatus : Instruction
 {
     [JsonProperty("status")]
     public CoreSyncStatus Status { get; set; } = null!;
@@ -119,7 +115,7 @@ public class BucketProgress
     public int TargetCount { get; set; }
 }
 
-public class FetchCredentials: Instruction
+public class FetchCredentials : Instruction
 {
     [JsonProperty("did_expire")]
     public bool DidExpire { get; set; }

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

@@ -117,6 +117,80 @@ public async Task<T> Get<T>(string path, Dictionary<string, string>? headers = n
         return JsonConvert.DeserializeObject<T>(responseData)!;
     }
 
+    /// <summary>
+    /// Posts to the stream endpoint and returns a raw NDJSON stream that can be read line by line.
+    /// </summary>
+    public async Task<Stream> PostStreamRaw(SyncStreamOptions options)
+    {
+        var requestMessage = await BuildRequest(HttpMethod.Post, options.Path, options.Data, options.Headers);
+        var response = await httpClient.SendAsync(requestMessage, HttpCompletionOption.ResponseHeadersRead, options.CancellationToken);
+
+        if (response.Content == null)
+        {
+            throw new HttpRequestException($"HTTP {response.StatusCode}: No content");
+        }
+
+        if (response.StatusCode == System.Net.HttpStatusCode.Unauthorized)
+        {
+            InvalidateCredentials();
+        }
+
+        if (!response.IsSuccessStatusCode)
+        {
+            var errorText = await response.Content.ReadAsStringAsync();
+            throw new HttpRequestException($"HTTP {response.StatusCode}: {errorText}");
+        }
+
+        return await response.Content.ReadAsStreamAsync();
+    }
+
+
+    /// <summary>  
+    /// Originally used for the C# streaming sync implementation.
+    /// </summary>
+    public async IAsyncEnumerable<StreamingSyncLine?> PostStream(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");
+        }
+
+        if (response.StatusCode == System.Net.HttpStatusCode.Unauthorized)
+        {
+            InvalidateCredentials();
+        }
+
+        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;
+
+        using var timeoutCts = new CancellationTokenSource();
+        using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(options.CancellationToken, timeoutCts.Token);
+
+        linkedCts.Token.Register(() =>
+        {
+            stream.Close();
+        });
+
+        while ((line = await reader.ReadLineAsync()) != null)
+        {
+            timeoutCts.CancelAfter(TimeSpan.FromMilliseconds(STREAMING_POST_TIMEOUT_MS));
+            yield return ParseStreamingSyncLine(JObject.Parse(line));
+        }
+    }
+
+
     public static StreamingSyncLine? ParseStreamingSyncLine(JObject json)
     {
         // Determine the type based on available keys