Use long for opIds instead of string (#89)

Author: LucDeCaf

PowerSync/PowerSync.Common/CHANGELOG.md

@@ -3,6 +3,8 @@
 ## 1.0.1
 
 - Full release.
+- `PowerSyncDatabase.OnChange` now returns the underlying raw table name instead of the view name to mirror PowerSync JS.
+- Use `long` for op IDs instead of `string`. This affects the types returned by some methods used in `PowerSyncBackendConnector.UploadData`, namely `PowerSyncDatabase.GetNextCrudTransaction()` and `PowerSyncDatabase.GetCrudBatch()`.
 
 ## 1.0.0 (unlisted)
 

PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs

@@ -669,7 +669,7 @@ public async Task<UploadQueueStats> GetUploadQueueStats(bool includeSize = false
         });
     }
 
-    public Task HandleCrudCheckpoint(long lastClientId, string? writeCheckpoint = null)
+    public Task HandleCrudCheckpoint(long lastClientId, long? writeCheckpoint = null)
     {
         return BucketStorageAdapter.HandleCrudCheckpoint(lastClientId, writeCheckpoint);
     }

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

@@ -135,9 +135,9 @@ public interface IBucketStorageAdapter : ICloseable
     Task<bool> HasCrud();
     Task<CrudBatch?> GetCrudBatch(int limit = 100);
 
-    Task<bool> UpdateLocalTarget(Func<Task<string>> callback);
+    Task<bool> UpdateLocalTarget(Func<Task<long>> callback);
 
-    Task HandleCrudCheckpoint(long lastClientId, string? writeCheckpoint = null);
+    Task HandleCrudCheckpoint(long lastClientId, long? writeCheckpoint = null);
 
     /// <summary>
     /// Get a unique client ID.

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

@@ -16,7 +16,7 @@ namespace PowerSync.Common.Client.Sync.Bucket;
 
 public class SqliteBucketStorage : IBucketStorageAdapter
 {
-    public static readonly string MAX_OP_ID = "9223372036854775807";
+    public const long MAX_OP_ID = 9223372036854775807;
 
     public BucketStorageEvents Events { get; } = new();
 
@@ -72,12 +72,10 @@ public async Task<string> GetClientId()
     /// Reads the stored target checkpoint request id, or updates it when the update parameter is set.
     /// </summary>
     /// <returns>The previous checkpoint request.</returns>
-    private static Task<string?> TargetCheckpointRequestId(ILockContext tx, string? update = null)
+    private static Task<long?> TargetCheckpointRequestId(ILockContext tx, long? update = null)
     {
-        // TODO Note that we are only casting in Dart/JS because this returns a 64-bit integer we can't natively represent there.
-        // Turning MAX_OP_ID into a 64-bit integer here and comparing ints would be better.
-        return tx.Get<string?>(
-            "SELECT CAST(powersync_control(?, ?) AS TEXT) AS r",
+        return tx.Get<long?>(
+            "SELECT powersync_control(?, ?) AS r",
             [PowerSyncControlCommand.TARGET_CHECKPOINT_REQUEST_ID, update]);
     }
 
@@ -94,7 +92,7 @@ public class ResultDetail
 
     private record SequenceResult(long seq);
 
-    public async Task<bool> UpdateLocalTarget(Func<Task<string>> callback)
+    public async Task<bool> UpdateLocalTarget(Func<Task<long>> callback)
     {
         var seqBeforeResult = await db.ReadTransaction(async tx =>
         {
@@ -118,7 +116,7 @@ public async Task<bool> UpdateLocalTarget(Func<Task<string>> callback)
             return false;
         }
 
-        string opId = await callback();
+        long opId = await callback();
 
         logger.LogDebug("[updateLocalTarget] Updating target to checkpoint {message}", opId);
 
@@ -154,7 +152,7 @@ public async Task<bool> UpdateLocalTarget(Func<Task<string>> callback)
             return true;
         });
     }
-    public Task HandleCrudCheckpoint(long lastClientId, string? writeCheckpoint = null)
+    public Task HandleCrudCheckpoint(long lastClientId, long? writeCheckpoint = null)
     {
         return db.WriteTransaction(async tx =>
         {
@@ -165,7 +163,7 @@ public Task HandleCrudCheckpoint(long lastClientId, string? writeCheckpoint = nu
 
             await TargetCheckpointRequestId(
                 tx,
-                !string.IsNullOrEmpty(writeCheckpoint) && !crudRemaining ? writeCheckpoint : MAX_OP_ID);
+                writeCheckpoint is not null && !crudRemaining ? writeCheckpoint : MAX_OP_ID);
         });
     }
 

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

@@ -804,13 +804,13 @@ public void Close()
     }
 
     public record ResponseData(
-        [property: JsonProperty("write_checkpoint")] string WriteCheckpoint
+        [property: JsonProperty("write_checkpoint")] long WriteCheckpoint
     );
 
     public record ApiResponse(
         [property: JsonProperty("data")] ResponseData Data
     );
-    public async Task<string> GetWriteCheckpoint()
+    public async Task<long> GetWriteCheckpoint()
     {
         var clientId = await Options.Adapter.GetClientId();
         var path = $"/write-checkpoint2.json?client_id={clientId}";

PowerSync/PowerSync.Common/Client/WatchManager.cs

@@ -73,11 +73,9 @@ public IAsyncEnumerable<WatchOnChangeEvent> OnChange(SQLWatchOptions? options)
             refreshOnSchemaChange: false
         );
 
-        // TODO: powersync-js onChange returns table names in `ps_data__{table}` format.
-        //       We should make a decision on whether or not to mirror that before v1.
         return Stream(subscription, changed => Task.FromResult(new WatchOnChangeEvent
         {
-            ChangedTables = [.. changed.Select(InternalToFriendlyTableName)]
+            ChangedTables = [.. changed]
         }));
     }
 

PowerSync/PowerSync.Common/DB/Crud/CrudBatch.cs

@@ -3,13 +3,13 @@ namespace PowerSync.Common.DB.Crud;
 using System;
 using System.Threading.Tasks;
 
-public class CrudBatch(CrudEntry[] Crud, bool HaveMore, Func<string?, Task> CompleteCallback)
+public class CrudBatch(CrudEntry[] Crud, bool HaveMore, Func<long?, Task> CompleteCallback)
 {
     public CrudEntry[] Crud { get; private set; } = Crud;
 
     public bool HaveMore { get; private set; } = HaveMore;
 
-    public async Task Complete(string? checkpoint = null)
+    public async Task Complete(long? checkpoint = null)
     {
         await CompleteCallback(checkpoint);
     }

PowerSync/PowerSync.Common/DB/Crud/CrudTransaction.cs

@@ -3,7 +3,7 @@ namespace PowerSync.Common.DB.Crud;
 using System;
 using System.Threading.Tasks;
 
-public class CrudTransaction(CrudEntry[] crud, Func<string?, Task> complete, long? transactionId = null) : CrudBatch(crud, false, complete)
+public class CrudTransaction(CrudEntry[] crud, Func<long?, Task> complete, long? transactionId = null) : CrudBatch(crud, false, complete)
 {
     public long? TransactionId { get; private set; } = transactionId;
 }

Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/StreamingSyncRetryTests.cs

@@ -99,7 +99,7 @@ SemaphoreSlim signal
     public override Task<T> Get<T>(string path, Dictionary<string, string>? headers = null)
     {
         var response = new StreamingSyncImplementation.ApiResponse(
-            new StreamingSyncImplementation.ResponseData("1")
+            new StreamingSyncImplementation.ResponseData(1)
         );
         return Task.FromResult((T)(object)response);
     }

Tests/PowerSync/PowerSync.Common.Tests/Client/Sync/SyncIterationControlFlowTests.cs

@@ -255,8 +255,8 @@ public Task<string> Control(string op, object? payload)
         public Task<CrudEntry?> NextCrudItem() => Task.FromResult<CrudEntry?>(null);
         public Task<bool> HasCrud() => Task.FromResult(false);
         public Task<CrudBatch?> GetCrudBatch(int limit = 100) => Task.FromResult<CrudBatch?>(null);
-        public Task<bool> UpdateLocalTarget(Func<Task<string>> callback) => Task.FromResult(false);
-        public Task HandleCrudCheckpoint(long lastClientId, string? writeCheckpoint = null) => Task.CompletedTask;
+        public Task<bool> UpdateLocalTarget(Func<Task<long>> callback) => Task.FromResult(false);
+        public Task HandleCrudCheckpoint(long lastClientId, long? writeCheckpoint = null) => Task.CompletedTask;
         public Task<string> GetClientId() => Task.FromResult("test-client");
         public void Close() { }
     }