Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions PowerSync/PowerSync.Common/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
## 1.0.1

- Full release.
- Use `long` for opIds instead of `string`.
Comment thread
LucDeCaf marked this conversation as resolved.
Outdated

## 1.0.0 (unlisted)

Expand Down
2 changes: 1 addition & 1 deletion PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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();

Expand Down Expand Up @@ -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]);
}

Expand All @@ -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 =>
{
Expand All @@ -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);

Expand Down Expand Up @@ -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 =>
{
Expand All @@ -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);
});
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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}";
Expand Down
4 changes: 2 additions & 2 deletions PowerSync/PowerSync.Common/DB/Crud/CrudBatch.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down
2 changes: 1 addition & 1 deletion PowerSync/PowerSync.Common/DB/Crud/CrudTransaction.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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() { }
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -187,7 +187,7 @@ public override Task<Stream> PostStreamRaw(SyncStreamOptions options)
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);
Expand Down