Skip to content

Commit dd9a047

Browse files
ChriztiaanChriztiaan
committed
First run through of catching up on sync priorities and progress.
1 parent 848d78c commit dd9a047

6 files changed

Lines changed: 179 additions & 25 deletions

File tree

‎PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs‎

Lines changed: 50 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -156,21 +156,48 @@ public async Task WaitForReady()
156156
await isReadyTask;
157157
}
158158

159-
public async Task WaitForFirstSync(CancellationToken? cancellationToken = null)
159+
public class PrioritySyncRequest
160160
{
161-
if (CurrentStatus.HasSynced == true)
161+
public CancellationToken? Token { get; set; }
162+
public int? Priority { get; set; }
163+
}
164+
165+
/// <summary>
166+
/// Wait for the first sync operation to complete.
167+
/// </summary>
168+
/// <param name="request">
169+
/// An object providing a cancellation token and a priority target.
170+
/// When a priority target is set, the task may complete when all buckets with the given (or higher)
171+
/// priorities have been synchronized. This can be earlier than a complete sync.
172+
/// </param>
173+
/// <returns>A task which will complete once the first full sync has completed.</returns>
174+
public async Task WaitForFirstSync(PrioritySyncRequest? request = null)
175+
{
176+
var priority = request?.Priority ?? null;
177+
var cancellationToken = request?.Token ?? null;
178+
179+
bool StatusMatches(SyncStatus status)
180+
{
181+
if (priority == null)
182+
{
183+
return status.HasSynced == true;
184+
}
185+
return status.StatusForPriority(priority.Value).HasSynced == true;
186+
}
187+
188+
if (StatusMatches(CurrentStatus))
162189
{
163190
return;
164191
}
165192

166193
var tcs = new TaskCompletionSource<bool>();
167194
var cts = new CancellationTokenSource();
168195

169-
var _ = Task.Run(() =>
196+
_ = Task.Run(() =>
170197
{
171198
foreach (var update in Listen(cts.Token))
172199
{
173-
if (update.StatusChanged?.HasSynced == true)
200+
if (update.StatusChanged != null && StatusMatches(update.StatusChanged!))
174201
{
175202
cts.Cancel();
176203
tcs.SetResult(true);
@@ -227,7 +254,22 @@ private async Task LoadVersion()
227254
}
228255
}
229256

230-
private record LastSyncedResult(int? priority, string? last_synced_at);
257+
// protected async resolveOfflineSyncStatus() {
258+
// const result = await this.database.get<{ r: string }>('SELECT powersync_offline_sync_status() as r');
259+
// const parsed = JSON.parse(result.r) as CoreSyncStatus;
260+
261+
// const updatedStatus = new SyncStatus({
262+
// ...this.currentStatus.toJSON(),
263+
// ...coreStatusToJs(parsed)
264+
// });
265+
266+
// if (!updatedStatus.isEqual(this.currentStatus)) {
267+
// this.currentStatus = updatedStatus;
268+
// this.iterateListeners((l) => l.statusChanged?.(this.currentStatus));
269+
// }
270+
// }
271+
// TODO DELETE
272+
private record LastSyncedResult(int priority, string? last_synced_at);
231273

232274
protected async Task UpdateHasSynced()
233275
{
@@ -236,6 +278,7 @@ protected async Task UpdateHasSynced()
236278
);
237279

238280
DateTime? lastCompleteSync = null;
281+
List<DB.Crud.SyncPriorityStatus> priorityStatuses = [];
239282

240283
// TODO: Will be altered/extended when reporting individual sync priority statuses is supported
241284
foreach (var result in results)
@@ -262,6 +305,8 @@ protected async Task UpdateHasSynced()
262305
}
263306
}
264307

308+
309+
265310
/// <summary>
266311
/// Replace the schema with a new version. This is for advanced use cases - typically the schema should just be specified once in the constructor.
267312
/// Cannot be used while connected - this should only be called before <see cref="Connect"/>.

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1043,10 +1043,12 @@ protected void UpdateSyncStatus(SyncStatusOptions options, UpdateSyncStatusOptio
10431043
Connected = options.Connected ?? SyncStatus.Connected,
10441044
Connecting = !options.Connected.GetValueOrDefault() && (options.Connecting ?? SyncStatus.Connecting),
10451045
LastSyncedAt = options.LastSyncedAt ?? SyncStatus.LastSyncedAt,
1046+
PriorityStatusEntries = options.PriorityStatusEntries ?? SyncStatus.PriorityStatusEntries,
10461047
DataFlow = new SyncDataFlowStatus
10471048
{
10481049
Uploading = options.DataFlow?.Uploading ?? SyncStatus.DataFlowStatus.Uploading,
10491050
Downloading = options.DataFlow?.Downloading ?? SyncStatus.DataFlowStatus.Downloading,
1051+
DownloadProgress = options.DataFlow?.DownloadProgress ?? SyncStatus.DataFlowStatus.DownloadProgress
10501052
DownloadError = updateOptions?.ClearDownloadError == true ? null : options.DataFlow?.DownloadError ?? SyncStatus.DataFlowStatus.DownloadError,
10511053
UploadError = updateOptions?.ClearUploadError == true ? null : options.DataFlow?.UploadError ?? SyncStatus.DataFlowStatus.UploadError,
10521054
}
@@ -1088,6 +1090,7 @@ private async Task DelayRetry()
10881090
}
10891091
}
10901092

1093+
10911094
enum LockType
10921095
{
10931096
CRUD,

‎PowerSync/PowerSync.Common/DB/Crud/CrudEntry.cs‎

Lines changed: 33 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,9 @@ public class CrudEntryDataJSON
3232
[JsonProperty("data")]
3333
public Dictionary<string, object>? Data { get; set; }
3434

35+
[JsonProperty("old")]
36+
public Dictionary<string, string?>? Old { get; set; }
37+
3538
[JsonProperty("op")]
3639
public UpdateType Op { get; set; }
3740

@@ -40,6 +43,9 @@ public class CrudEntryDataJSON
4043

4144
[JsonProperty("id")]
4245
public string Id { get; set; } = null!;
46+
47+
[JsonProperty("metadata")]
48+
public string? Metadata { get; set; }
4349
}
4450

4551
public class CrudEntryOutputJSON
@@ -63,7 +69,17 @@ public class CrudEntryOutputJSON
6369
public Dictionary<string, object>? Data { get; set; }
6470
}
6571

66-
public class CrudEntry(int clientId, UpdateType op, string table, string id, long? transactionId = null, Dictionary<string, object>? opData = null)
72+
73+
public class CrudEntry(
74+
int clientId,
75+
UpdateType op,
76+
string table,
77+
string id,
78+
long? transactionId = null,
79+
Dictionary<string, object>? opData = null,
80+
Dictionary<String, String?>? previousValues = null,
81+
string? metadata = null
82+
)
6783
{
6884
public int ClientId { get; private set; } = clientId;
6985
public string Id { get; private set; } = id;
@@ -72,6 +88,19 @@ public class CrudEntry(int clientId, UpdateType op, string table, string id, lon
7288
public string Table { get; private set; } = table;
7389
public long? TransactionId { get; private set; } = transactionId;
7490

91+
/// <summary>
92+
/// Previous values before this change.
93+
/// </summary>
94+
public Dictionary<String, String?>? PreviousValues { get; private set; } = previousValues;
95+
96+
/// <summary>
97+
/// Client-side metadata attached with this write.
98+
///
99+
/// This field is only available when the `trackMetadata` option was set to `true` when creating a table
100+
/// and the insert or update statement set the `_metadata` column.
101+
/// </summary>
102+
public string? Metadata { get; private set; } = metadata;
103+
75104
public static CrudEntry FromRow(CrudEntryJSON dbRow)
76105
{
77106
var data = JsonConvert.DeserializeObject<CrudEntryDataJSON>(dbRow.Data)
@@ -83,7 +112,9 @@ public static CrudEntry FromRow(CrudEntryJSON dbRow)
83112
data.Type,
84113
data.Id,
85114
dbRow.TransactionId,
86-
data.Data
115+
data.Data,
116+
data.Old,
117+
data.Metadata
87118
);
88119
}
89120

‎PowerSync/PowerSync.Common/DB/Crud/SyncStatus.cs‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -100,7 +100,7 @@ public class SyncStatus(SyncStatusOptions options)
100100
/// </summary>
101101
public SyncPriorityStatus[] PriorityStatusEntries =>
102102
(Options.PriorityStatusEntries ?? [])
103-
.OrderBy(entry => entry.Priority)
103+
.OrderBy(x => x, Comparer<SyncPriorityStatus>.Create(ComparePriorities))
104104
.ToArray();
105105

106106
/// <summary>
@@ -154,7 +154,7 @@ public SyncPriorityStatus StatusForPriority(int priority)
154154

155155
private string SerializeObject()
156156
{
157-
return JsonConvert.SerializeObject(new { Options = Options, UploadErrorMessage = Options.DataFlow?.UploadError?.Message, DownloadErrorMessage = DataFlowStatus.DownloadError?.Message });
157+
return JsonConvert.SerializeObject(new { Options, UploadErrorMessage = Options.DataFlow?.UploadError?.Message, DownloadErrorMessage = DataFlowStatus.DownloadError?.Message });
158158
}
159159

160160
public bool IsEqual(SyncStatus status)
@@ -173,4 +173,9 @@ public string ToJSON()
173173
{
174174
return SerializeObject();
175175
}
176+
177+
private static int ComparePriorities(SyncPriorityStatus a, SyncPriorityStatus b)
178+
{
179+
return b.Priority - a.Priority; // Reverse because higher priorities have lower numbers
180+
}
176181
}

‎PowerSync/PowerSync.Common/DB/IDBAdapter.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -30,13 +30,13 @@ public class QueryRows
3030
public interface IDBGetUtils
3131
{
3232
// Execute a read-only query and return results.
33-
Task<T[]> GetAll<T>(string sql, params object[]? parameters);
33+
Task<T[]> GetAll<T>(string sql, object[]? parameters = null);
3434

3535
// Execute a read-only query and return the first result, or null if the ResultSet is empty.
36-
Task<T?> GetOptional<T>(string sql, params object[]? parameters);
36+
Task<T?> GetOptional<T>(string sql, object[]? parameters = null);
3737

3838
// Execute a read-only query and return the first result, error if the ResultSet is empty.
39-
Task<T> Get<T>(string sql, params object[]? parameters);
39+
Task<T> Get<T>(string sql, object[]? parameters = null);
4040
}
4141

4242
public interface ILockContext : IDBGetUtils

‎PowerSync/PowerSync.Common/DB/Schema/Table.cs‎

Lines changed: 83 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -7,15 +7,57 @@ public class TableOptions(
77
Dictionary<string, List<string>>? indexes = null,
88
bool? localOnly = null,
99
bool? insertOnly = null,
10-
string? viewName = null)
10+
string? viewName = null,
11+
bool? trackMetadata = null,
12+
TrackPreviousOptions? trackPreviousOptions = null,
13+
bool? ignoreEmptyUpdates = null
14+
)
1115
{
1216
public Dictionary<string, List<string>> Indexes { get; set; } = indexes ?? [];
1317

14-
public bool LocalOnly { get; set; } = localOnly ?? false;
18+
public bool LocalOnly { get; } = localOnly ?? false;
1519

16-
public bool InsertOnly { get; set; } = insertOnly ?? false;
20+
public bool InsertOnly { get; } = insertOnly ?? false;
1721

18-
public string? ViewName { get; set; } = viewName;
22+
public string? ViewName { get; } = viewName;
23+
24+
/// <summary>
25+
/// Whether to add a hidden `_metadata` column that will be enabled for updates to attach custom
26+
/// information about writes that will be reported through [CrudEntry.metadata].
27+
/// </summary>
28+
public bool TrackMetadata { get; } = trackMetadata ?? false;
29+
30+
/// <summary>
31+
/// When set to a non-null value, track old values of columns
32+
/// </summary>
33+
public TrackPreviousOptions? TrackPreviousOptions { get; } = trackPreviousOptions ?? null;
34+
35+
/// <summary>
36+
/// Whether an `UPDATE` statement that doesn't change any values should be ignored when creating
37+
/// CRUD entries.
38+
/// </summary>
39+
public bool IgnoreEmptyUpdates { get; } = ignoreEmptyUpdates ?? false;
40+
}
41+
42+
/// <summary>
43+
/// Whether to include previous column values when PowerSync tracks local changes.
44+
/// Including old values may be helpful for some backend connector implementations,
45+
/// which is why it can be enabled on a per-table or per-column basis.
46+
/// </summary>
47+
public class TrackPreviousOptions
48+
{
49+
/// <summary>
50+
/// When defined, a list of column names for which old values should be tracked.
51+
/// </summary>
52+
[JsonProperty("columns")]
53+
public List<string>? Columns { get; set; }
54+
55+
/// <summary>
56+
/// Whether to only include old values when they were changed by an update, instead of always
57+
/// including all old values,
58+
/// </summary>
59+
[JsonProperty("onlyWhenChanged")]
60+
public bool? OnlyWhenChanged { get; set; }
1961
}
2062

2163
public class Table
@@ -35,16 +77,21 @@ public Table(Dictionary<string, ColumnType> columns, TableOptions? options = nul
3577
{
3678
ConvertedColumns = [.. columns.Select(kv => new Column(new ColumnOptions(kv.Key, kv.Value)))];
3779

38-
ConvertedIndexes = [.. (Options?.Indexes ?? [])
80+
ConvertedIndexes =
81+
[
82+
.. (Options?.Indexes ?? [])
3983
.Select(kv =>
4084
new Index(new IndexOptions(
4185
kv.Key,
42-
[.. kv.Value.Select(name =>
43-
new IndexedColumn(new IndexColumnOptions(
44-
name.Replace("-", ""), !name.StartsWith("-")))
45-
)]
86+
[
87+
.. kv.Value.Select(name =>
88+
new IndexedColumn(new IndexColumnOptions(
89+
name.Replace("-", ""), !name.StartsWith("-")))
90+
)
91+
]
4692
))
47-
)];
93+
)
94+
];
4895

4996
Options = options ?? new TableOptions();
5097

@@ -61,7 +108,18 @@ public void Validate()
61108

62109
if (Columns.Count > Column.MAX_AMOUNT_OF_COLUMNS)
63110
{
64-
throw new Exception($"Table has too many columns. The maximum number of columns is {Column.MAX_AMOUNT_OF_COLUMNS}.");
111+
throw new Exception(
112+
$"Table has too many columns. The maximum number of columns is {Column.MAX_AMOUNT_OF_COLUMNS}.");
113+
}
114+
115+
if (Options.TrackMetadata && Options.LocalOnly)
116+
{
117+
throw new Exception("Can't include metadata for local-only tables.");
118+
}
119+
120+
if (Options.TrackPreviousOptions != null && Options.LocalOnly)
121+
{
122+
throw new Exception("Can't include old values for local-only tables.");
65123
}
66124

67125
var columnNames = new HashSet<string> { "id" };
@@ -103,15 +161,27 @@ public void Validate()
103161

104162
public string ToJSON(string Name = "")
105163
{
164+
var trackPrevious = Options.TrackPreviousOptions;
165+
106166
var jsonObject = new
107167
{
108168
view_name = Options.ViewName ?? Name,
109169
local_only = Options.LocalOnly,
110170
insert_only = Options.InsertOnly,
111171
columns = ConvertedColumns.Select(c => JsonConvert.DeserializeObject<object>(c.ToJSON())).ToList(),
112-
indexes = ConvertedIndexes.Select(e => JsonConvert.DeserializeObject<object>(e.ToJSON(this))).ToList()
172+
indexes = ConvertedIndexes.Select(e => JsonConvert.DeserializeObject<object>(e.ToJSON(this))).ToList(),
173+
174+
include_metadata = Options.TrackMetadata,
175+
ignore_empty_update = Options.IgnoreEmptyUpdates,
176+
include_old = (object)(trackPrevious switch
177+
{
178+
null => false,
179+
{ Columns: null } => true,
180+
{ Columns: var cols } => cols
181+
}),
182+
include_old_only_when_changed = trackPrevious?.OnlyWhenChanged ?? false
113183
};
114184

115185
return JsonConvert.SerializeObject(jsonObject);
116186
}
117-
}
187+
}

0 commit comments

Comments
 (0)