Skip to content

Commit 55cf059

Browse files
DominicGBauerDominicGBauer
authored
fix: down syncs not being reflected in client app (#66)
* fix: client id mismatch leading to syncing issues * fix: client id mismatch leading to syncing issues
1 parent 85da9c6 commit 55cf059

File tree

5 files changed

+32
-27
lines changed

5 files changed

+32
-27
lines changed

‎core/src/commonMain/kotlin/com/powersync/bucket/BucketStorage.kt

Lines changed: 18 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -74,10 +74,10 @@ internal class BucketStorage(
7474

7575
logger.i { "[updateLocalTarget] Updating target to checkpoint $opId" }
7676

77-
return db.readTransaction {
77+
return db.writeTransaction {
7878
if (hasCrud()) {
7979
logger.w { "[updateLocalTarget] ps crud is not empty" }
80-
return@readTransaction false
80+
return@writeTransaction false
8181
}
8282

8383
val seqAfter =
@@ -88,15 +88,17 @@ internal class BucketStorage(
8888
throw AssertionError("Sqlite Sequence should not be empty")
8989

9090
if (seqAfter != seqBefore) {
91+
logger.d("seqAfter != seqBefore seqAfter: $seqAfter seqBefore: $seqBefore")
9192
// New crud data may have been uploaded since we got the checkpoint. Abort.
92-
return@readTransaction false
93+
return@writeTransaction false
9394
}
9495

9596
db.execute(
96-
"UPDATE ${InternalTable.BUCKETS} SET target_op = ? WHERE name='\$local'",
97+
"UPDATE ${InternalTable.BUCKETS} SET target_op = CAST(? as INTEGER) WHERE name='\$local'",
9798
listOf(opId)
9899
)
99-
return@readTransaction true
100+
101+
return@writeTransaction true
100102
}
101103
}
102104

@@ -113,7 +115,7 @@ internal class BucketStorage(
113115

114116
suspend fun getBucketStates(): List<BucketState> {
115117
return db.getAll(
116-
"SELECT name as bucket, cast(last_op as TEXT) as op_id FROM ${InternalTable.BUCKETS} WHERE pending_delete = 0",
118+
"SELECT name AS bucket, CAST(last_op AS TEXT) AS op_id FROM ${InternalTable.BUCKETS} WHERE pending_delete = 0",
117119
mapper = { cursor ->
118120
BucketState(
119121
bucket = cursor.getString(0)!!,
@@ -138,6 +140,8 @@ internal class BucketStorage(
138140
)
139141
}
140142

143+
Logger.d("[deleteBucket] Done deleting")
144+
141145
this.pendingBucketDeletes.value = true
142146
}
143147

@@ -206,7 +210,7 @@ internal class BucketStorage(
206210

207211
private suspend fun validateChecksums(checkpoint: Checkpoint): SyncLocalDatabaseResult {
208212
val res = db.getOptional(
209-
"SELECT powersync_validate_checkpoint(?) as result",
213+
"SELECT powersync_validate_checkpoint(?) AS result",
210214
parameters = listOf(JsonUtil.json.encodeToString(checkpoint)),
211215
mapper = { cursor ->
212216
cursor.getString(0)!!
@@ -227,11 +231,16 @@ internal class BucketStorage(
227231
*/
228232
private suspend fun updateObjectsFromBuckets(): Boolean {
229233
return db.writeTransaction { tx ->
230-
val res = tx.execute(
234+
235+
tx.execute(
231236
"INSERT INTO powersync_operations(op, data) VALUES(?, ?)",
232237
listOf("sync_local", "")
233238
)
234239

240+
val res = tx.get("select last_insert_rowid()") { cursor ->
241+
cursor.getLong(0)!!
242+
}
243+
235244
return@writeTransaction res == 1L
236245
}
237246
}
@@ -260,12 +269,9 @@ internal class BucketStorage(
260269

261270
db.writeTransaction { tx ->
262271
tx.execute(
263-
"DELETE FROM ps_oplog WHERE bucket IN (SELECT name FROM ps_buckets WHERE pending_delete = 1 AND last_applied_op = last_op AND last_op >= target_op)",
272+
"INSERT INTO powersync_operations(op, data) VALUES (?, ?)", listOf("delete_pending_buckets","")
264273
)
265274

266-
tx.execute(
267-
"DELETE FROM ps_buckets WHERE pending_delete = 1 AND last_applied_op = last_op AND last_op >= target_op",
268-
)
269275
// Executed once after start-up, and again when there are pending deletes.
270276
pendingBucketDeletes.value = false
271277
}

‎core/src/commonMain/kotlin/com/powersync/db/PowerSyncDatabaseImpl.kt

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -192,7 +192,8 @@ internal class PowerSyncDatabaseImpl(
192192
}
193193

194194
return@readTransaction CrudTransaction(
195-
crud = entries, transactionId = txId,
195+
crud = entries,
196+
transactionId = txId,
196197
complete = { writeCheckpoint ->
197198
logger.i { "[CrudTransaction::complete] Completing transaction with checkpoint $writeCheckpoint" }
198199
handleWriteCheckpoint(entries.last().clientId, writeCheckpoint)
@@ -260,12 +261,12 @@ internal class PowerSyncDatabaseImpl(
260261

261262
if (writeCheckpoint != null && bucketStorage.hasCrud()) {
262263
tx.execute(
263-
"UPDATE ps_buckets SET target_op = CAST(? as INTEGER) WHERE name='\$local'",
264+
"UPDATE ps_buckets SET target_op = CAST(? AS INTEGER) WHERE name='\$local'",
264265
listOf(writeCheckpoint),
265266
)
266267
} else {
267268
tx.execute(
268-
"UPDATE ps_buckets SET target_op = CAST(? as INTEGER) WHERE name='\$local'",
269+
"UPDATE ps_buckets SET target_op = CAST(? AS INTEGER) WHERE name='\$local'",
269270
listOf(bucketStorage.getMaxOpId()),
270271
)
271272
}

‎core/src/commonMain/kotlin/com/powersync/sync/SyncStream.kt

Lines changed: 2 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -162,15 +162,14 @@ internal class SyncStream(
162162
return false
163163
} else {
164164
// This isolate is the only one triggering
165-
bucketStorage.updateLocalTarget { getWriteCheckpoint() }
166-
return true
165+
return bucketStorage.updateLocalTarget { getWriteCheckpoint() }
167166
}
168167
}
169168

170169
private suspend fun getWriteCheckpoint(): String {
171170
val credentials = connector.getCredentialsCached()
172171
require(credentials != null) { "Not logged in" }
173-
val uri = credentials.endpointUri("write-checkpoint2.json?client_id=$clientId'")
172+
val uri = credentials.endpointUri("write-checkpoint2.json?client_id=$clientId")
174173

175174
val response = httpClient.get(uri) {
176175
contentType(ContentType.Application.Json)
@@ -263,9 +262,7 @@ internal class SyncStream(
263262
jsonString: String,
264263
state: SyncStreamState
265264
): SyncStreamState {
266-
logger.i { "[handleInstruction] Received Instruction: $jsonString" }
267265
val obj = JsonUtil.json.parseToJsonElement(jsonString).jsonObject
268-
269266
// TODO: Clean up
270267
when {
271268
isStreamingSyncCheckpoint(obj) -> return handleStreamingSyncCheckpoint(obj, state)
@@ -293,7 +290,6 @@ internal class SyncStream(
293290
): SyncStreamState {
294291
val checkpoint =
295292
JsonUtil.json.decodeFromJsonElement<Checkpoint>(jsonObj["checkpoint"] as JsonElement)
296-
297293
state.targetCheckpoint = checkpoint
298294
val bucketsToDelete = state.bucketSet!!.toMutableList()
299295
val newBuckets = mutableSetOf<String>()
@@ -388,8 +384,6 @@ internal class SyncStream(
388384
jsonObj: JsonObject,
389385
state: SyncStreamState
390386
): SyncStreamState {
391-
392-
393387
val syncBuckets =
394388
listOf<SyncDataBucket>(JsonUtil.json.decodeFromJsonElement(jsonObj["data"] as JsonElement))
395389

‎demos/android-supabase-todolist/app/src/main/java/com/powersync/androidexample/powersync/List.kt

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,12 @@
11
package com.powersync.demos.powersync
22

33
import androidx.lifecycle.ViewModel
4+
import androidx.lifecycle.viewModelScope
45
import com.powersync.PowerSyncDatabase
56
import kotlinx.coroutines.flow.Flow
67
import kotlinx.coroutines.flow.MutableStateFlow
78
import kotlinx.coroutines.flow.StateFlow
9+
import kotlinx.coroutines.launch
810
import kotlinx.coroutines.runBlocking
911

1012
internal class ListContent(
@@ -37,7 +39,7 @@ internal class ListContent(
3739
}
3840

3941
fun onItemDeleteClicked(item: ListItem) {
40-
runBlocking {
42+
viewModelScope.launch {
4143
db.writeTransaction { tx ->
4244
tx.execute("DELETE FROM $LISTS_TABLE WHERE id = ?", listOf(item.id))
4345
tx.execute("DELETE FROM $TODOS_TABLE WHERE list_id = ?", listOf(item.id))
@@ -48,7 +50,7 @@ internal class ListContent(
4850
fun onAddItemClicked() {
4951
if (_inputText.value.isBlank()) return
5052

51-
runBlocking {
53+
viewModelScope.launch {
5254
db.writeTransaction { tx ->
5355
tx.execute(
5456
"INSERT INTO $LISTS_TABLE (id, created_at, name, owner_id) VALUES (uuid(), datetime(), ?, ?)",

‎demos/supabase-todolist/shared/src/commonMain/kotlin/com/powersync/demos/powersync/List.kt

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,12 @@
11
package com.powersync.demos.powersync
22

33
import androidx.lifecycle.ViewModel
4+
import androidx.lifecycle.viewModelScope
45
import com.powersync.PowerSyncDatabase
56
import kotlinx.coroutines.flow.Flow
67
import kotlinx.coroutines.flow.MutableStateFlow
78
import kotlinx.coroutines.flow.StateFlow
9+
import kotlinx.coroutines.launch
810
import kotlinx.coroutines.runBlocking
911

1012
internal class ListContent(
@@ -37,7 +39,7 @@ internal class ListContent(
3739
}
3840

3941
fun onItemDeleteClicked(item: ListItem) {
40-
runBlocking {
42+
viewModelScope.launch {
4143
db.writeTransaction { tx ->
4244
tx.execute("DELETE FROM $LISTS_TABLE WHERE id = ?", listOf(item.id))
4345
tx.execute("DELETE FROM $TODOS_TABLE WHERE list_id = ?", listOf(item.id))
@@ -48,7 +50,7 @@ internal class ListContent(
4850
fun onAddItemClicked() {
4951
if (_inputText.value.isBlank()) return
5052

51-
runBlocking {
53+
viewModelScope.launch {
5254
db.writeTransaction { tx ->
5355
tx.execute(
5456
"INSERT INTO $LISTS_TABLE (id, created_at, name, owner_id) VALUES (uuid(), datetime(), ?, ?)",

0 commit comments

Comments
 (0)