Add updated_at sync cursors for incremental server sync

modules/vescape-core/android/src/main/java/expo/modules/vescapecore/alerts/AlertEngine.kt

@@ -138,6 +138,8 @@ internal fun withLegalModeOverlay(
         soundType = "preset:tick",
         createdAt = 0L,
         source = null,
+        // In-memory overlay: no row is ever persisted, so the sync cursor is meaningless here.
+        updatedAt = 0L,
     )
 }
 

modules/vescape-core/android/src/main/java/expo/modules/vescapecore/telemetry/AppDataRepository.kt

@@ -207,7 +207,7 @@ class AppDataRepository private constructor(private val context: Context) {
 
   suspend fun setAlertRuleEnabled(boardId: String, id: String, enabled: Boolean): Unit =
     withContext(Dispatchers.IO) {
-      dao.setAlertRuleEnabled(boardId, id, enabled)
+      dao.setAlertRuleEnabled(boardId, id, enabled, System.currentTimeMillis())
     }
 
   suspend fun deleteAlertRule(boardId: String, id: String): Unit = withContext(Dispatchers.IO) {
@@ -678,6 +678,7 @@ fun BoardEntity.toMap(settings: List<BoardSettingEntity>): Map<String, Any?> {
     "alertPresetsOnboarded" to (values["alertPresetsOnboarded"] ?: false),
     "legalMode" to (values["legalMode"] ?: mapOf("enabled" to false)),
     "link" to link,
+    "updatedAt" to updatedAt,
   )
 }
 
@@ -749,6 +750,7 @@ fun AlertRuleEntity.toMap(): Map<String, Any?> = mapOf(
   "soundType" to soundType,
   "createdAt" to createdAt,
   "source" to source,
+  "updatedAt" to updatedAt,
 )
 
 fun TuneProfileEntity.toMap(): Map<String, Any?> = mapOf(
@@ -922,11 +924,17 @@ private fun Map<String, Any?>.normalizedBoardLink(): Map<String, Any?>? {
   )
 }
 
-internal fun Map<String, Any?>.toBoardEntity(): BoardEntity = BoardEntity(
+/**
+ * Native stamps [BoardEntity.updatedAt] itself rather than trusting the bridge value: it is a sync
+ * cursor, so it must come from the device clock that already writes `created_at` and must move on
+ * every upsert, including partial edits that leave `createdAt` untouched.
+ */
+internal fun Map<String, Any?>.toBoardEntity(now: Long = System.currentTimeMillis()): BoardEntity = BoardEntity(
   id = getString("id"),
   name = getString("name"),
   bleId = normalizedBoardLink()?.get("bleId") as? String,
   createdAt = getLong("createdAt"),
+  updatedAt = now,
 )
 
 internal fun Map<String, Any?>.toBoardSettingEntities(boardId: String): Pair<List<BoardSettingEntity>, List<String>> {
@@ -1073,7 +1081,10 @@ private fun parseLegacyMapString(value: String): Map<String, Any?>? {
   }.toMap()
 }
 
-private fun Map<String, Any?>.toAlertRuleEntity(): AlertRuleEntity = AlertRuleEntity(
+/** Native stamps [AlertRuleEntity.updatedAt]; see [toBoardEntity] for why the bridge value is ignored. */
+internal fun Map<String, Any?>.toAlertRuleEntity(
+  now: Long = System.currentTimeMillis(),
+): AlertRuleEntity = AlertRuleEntity(
   boardId = getString("boardId"),
   id = getString("id"),
   controlId = getString("controlId"),
@@ -1083,6 +1094,7 @@ private fun Map<String, Any?>.toAlertRuleEntity(): AlertRuleEntity = AlertRuleEn
   soundType = get("soundType") as? String ?: "default",
   createdAt = getLong("createdAt"),
   source = get("source") as? String,
+  updatedAt = now,
 )
 
 private fun Map<String, Any?>.getString(key: String): String =

modules/vescape-core/android/src/main/java/expo/modules/vescapecore/telemetry/TelemetryBucketBuilder.kt

@@ -162,7 +162,14 @@ private class MutableBucket(
     }
   }
 
-  fun toEntity(): TelemetryMinuteBucketEntity = TelemetryMinuteBucketEntity(
+  /**
+   * [now] is the incremental-sync cursor stamped on the row, so every append or rebuild that
+   * reaches the database is visible to cursor sync.
+   *
+   * @parity /modules/vescape-core/ios/telemetry/TelemetryDao.swift `upsertBucket`
+   */
+  fun toEntity(now: Long = System.currentTimeMillis()): TelemetryMinuteBucketEntity = TelemetryMinuteBucketEntity(
+    updatedAt = now,
     bucketStartMs = bucketStartMs,
     deviceId = deviceId,
     deviceName = deviceName,

modules/vescape-core/android/src/main/java/expo/modules/vescapecore/telemetry/TelemetryDao.kt

@@ -385,8 +385,15 @@ interface TelemetryDao {
   @Insert(onConflict = OnConflictStrategy.REPLACE)
   suspend fun upsertAlertRule(rule: AlertRuleEntity)
 
-  @Query("UPDATE alerts SET enabled = :enabled WHERE board_id = :boardId AND id = :id")
-  suspend fun setAlertRuleEnabled(boardId: String, id: String, enabled: Boolean)
+  /**
+   * Targeted toggle. Unlike the `@Insert` upserts it never round-trips an entity, so `updated_at`
+   * has to move here explicitly — without it, toggling a rule leaves the sync cursor stale and the
+   * change never reaches the server.
+   *
+   * @parity /modules/vescape-core/ios/telemetry/AppDataRepository.swift `setAlertRuleEnabled`
+   */
+  @Query("UPDATE alerts SET enabled = :enabled, updated_at = :updatedAt WHERE board_id = :boardId AND id = :id")
+  suspend fun setAlertRuleEnabled(boardId: String, id: String, enabled: Boolean, updatedAt: Long)
 
   @Query("DELETE FROM alerts WHERE board_id = :boardId AND id = :id")
   suspend fun deleteAlertRule(boardId: String, id: String)
@@ -588,6 +595,15 @@ private fun TelemetryMinuteBucketEntity.merge(next: TelemetryMinuteBucketEntity)
     },
     firstMovingAtMs = mergeNullableMin(firstMovingAtMs, next.firstMovingAtMs),
     lastMovingAtMs = mergeNullableMax(lastMovingAtMs, next.lastMovingAtMs),
+    // The merged row is being written now, so `next` normally carries the fresher stamp. `maxOf`
+    // clamps a backwards device-clock step: the value stays at the last real write time instead of
+    // regressing below a cursor already synced. Bounded and self-correcting — once the clock passes
+    // the old value again the stamp is truthful, unlike a ratcheting `existing + 1` counter, which
+    // would permanently inflate this row against other devices under last-write-wins.
+    //
+    // Not a completeness guarantee: a frozen stamp is only picked up because the sync query is
+    // `updated_at >= watermark`. Clock-rewind completeness is tracked separately (see #275).
+    updatedAt = maxOf(updatedAt, next.updatedAt),
   )
 }
 

modules/vescape-core/android/src/main/java/expo/modules/vescapecore/telemetry/TelemetryDatabase.kt

@@ -12,7 +12,7 @@ import java.io.File
 // @parity /modules/vescape-core/ios/VescapeCoreModule.swift
 internal const val TELEMETRY_DATABASE_NAME = "vescape.db"
 internal const val LEGACY_TELEMETRY_DATABASE_NAME = "telemetry.db"
-internal const val TELEMETRY_DATABASE_VERSION = 27
+internal const val TELEMETRY_DATABASE_VERSION = 28
 
 @Database(
   entities = [
@@ -472,6 +472,38 @@ abstract class TelemetryDatabase : RoomDatabase() {
       }
     }
 
+    /**
+     * Incremental-sync cursor on `boards`, `alerts` and `telemetry_minute_buckets`. The first two
+     * carried `created_at` only, so a board rename or an alert toggle was invisible to an
+     * "everything changed since T" query — the shape every other mutable table already supports.
+     * Buckets are append-and-merge targets with no cursor at all.
+     *
+     * Existing rows backfill to the best evidence of when they last changed — `created_at` for
+     * boards and alerts, `last_sample_at_ms` for buckets — never 0 and never null, so a first sync
+     * after upgrade reports each row at its true age instead of flooding the server with
+     * epoch-zero rows.
+     *
+     * @parity /modules/vescape-core/ios/telemetry/TelemetryDatabase.swift `v28_sync_cursors`
+     */
+    internal val MIGRATION_27_28 = object : Migration(27, 28) {
+      override fun migrate(db: SupportSQLiteDatabase) {
+        db.execSQL("ALTER TABLE boards ADD COLUMN updated_at INTEGER NOT NULL DEFAULT 0")
+        db.execSQL("UPDATE boards SET updated_at = created_at")
+        db.execSQL("CREATE INDEX IF NOT EXISTS index_boards_updated_at ON boards(updated_at)")
+
+        db.execSQL("ALTER TABLE alerts ADD COLUMN updated_at INTEGER NOT NULL DEFAULT 0")
+        db.execSQL("UPDATE alerts SET updated_at = created_at")
+        db.execSQL("CREATE INDEX IF NOT EXISTS index_alerts_updated_at ON alerts(updated_at)")
+
+        db.execSQL("ALTER TABLE telemetry_minute_buckets ADD COLUMN updated_at INTEGER NOT NULL DEFAULT 0")
+        db.execSQL("UPDATE telemetry_minute_buckets SET updated_at = last_sample_at_ms")
+        db.execSQL(
+          "CREATE INDEX IF NOT EXISTS index_telemetry_minute_buckets_updated_at " +
+            "ON telemetry_minute_buckets(updated_at)",
+        )
+      }
+    }
+
     /**
      * One-time file rename from the pre-release "telemetry.db" name. Checkpoints the legacy WAL so
      * the whole database lives in the main file, then renames it in place. Idempotent: once the new
@@ -526,6 +558,7 @@ abstract class TelemetryDatabase : RoomDatabase() {
             MIGRATION_24_25,
             MIGRATION_25_26,
             MIGRATION_26_27,
+            MIGRATION_27_28,
           )
           .fallbackToDestructiveMigration(true)
           .addCallback(object : Callback() {

modules/vescape-core/android/src/main/java/expo/modules/vescapecore/telemetry/TelemetryEntities.kt

@@ -109,7 +109,10 @@ data class TelemetryFrameEntity(
 @Entity(
   tableName = "telemetry_minute_buckets",
   primaryKeys = ["bucket_start_ms", "device_id"],
-  indices = [Index(value = ["bucket_start_ms"])],
+  indices = [
+    Index(value = ["bucket_start_ms"]),
+    Index(value = ["updated_at"]),
+  ],
 )
 data class TelemetryMinuteBucketEntity(
   @ColumnInfo(name = "bucket_start_ms")
@@ -170,6 +173,14 @@ data class TelemetryMinuteBucketEntity(
   val firstMovingAtMs: Long? = null,
   @ColumnInfo(name = "last_moving_at_ms")
   val lastMovingAtMs: Long? = null,
+  /**
+   * Incremental-sync cursor: wall-clock epoch ms of the last write to this bucket. Distinct from
+   * [lastSampleAtMs], which tracks the newest *sample* in the bucket — a merge that folds in older
+   * samples, or a bucket rebuild, changes the row without moving that. Only a write-time stamp
+   * makes an appended bucket visible to a "everything changed since T" query.
+   */
+  @ColumnInfo(name = "updated_at")
+  val updatedAt: Long,
 )
 
 @Entity(
@@ -228,6 +239,7 @@ data class DiagnosticEventEntity(
   tableName = "boards",
   indices = [
     Index(value = ["created_at"]),
+    Index(value = ["updated_at"]),
   ],
 )
 data class BoardEntity(
@@ -238,6 +250,13 @@ data class BoardEntity(
   val bleId: String?,
   @ColumnInfo(name = "created_at")
   val createdAt: Long,
+  /**
+   * Incremental-sync cursor: epoch ms of the last write to this row, from the same clock as
+   * [createdAt]. Equal to [createdAt] on insert and bumped on every mutation, so the client can ask
+   * the server for "everything changed since T". Indexed because cursor sync scans on it.
+   */
+  @ColumnInfo(name = "updated_at")
+  val updatedAt: Long,
 )
 
 @Entity(
@@ -265,6 +284,7 @@ data class BoardSettingEntity(
     Index(value = ["control_id"]),
     Index(value = ["enabled"]),
     Index(value = ["created_at"]),
+    Index(value = ["updated_at"]),
   ],
 )
 data class AlertRuleEntity(
@@ -286,6 +306,14 @@ data class AlertRuleEntity(
    * JS authors and regenerates preset rules; native only persists the string.
    */
   val source: String?,
+  /**
+   * Incremental-sync cursor: epoch ms of the last write to this row, from the same clock as
+   * [createdAt]. Equal to [createdAt] on insert and bumped on every mutation — including the
+   * targeted enable/disable update — so a toggled rule is visible to sync. Indexed because cursor
+   * sync scans on it.
+   */
+  @ColumnInfo(name = "updated_at")
+  val updatedAt: Long,
 )
 
 @Entity(

modules/vescape-core/android/src/test/java/expo/modules/vescapecore/alerts/AlertEngineTest.kt

@@ -29,6 +29,7 @@ class AlertEngineTest {
         soundType = soundType,
         createdAt = 0L,
         source = null,
+        updatedAt = 0L,
     )
 
     private fun telemetry(

modules/vescape-core/android/src/test/java/expo/modules/vescapecore/telemetry/ProfileStatsRepositoryTest.kt

@@ -136,5 +136,6 @@ class ProfileStatsRepositoryTest {
     maxGpsSpeedCentiMps = 9_999,
     firstMovingAtMs = firstMoving,
     lastMovingAtMs = lastMoving,
+    updatedAt = end,
   )
 }

modules/vescape-core/android/src/test/java/expo/modules/vescapecore/telemetry/SyncCursorMigrationTest.kt

@@ -0,0 +1,133 @@
+package expo.modules.vescapecore.telemetry
+
+import androidx.sqlite.db.SupportSQLiteDatabase
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import java.io.File
+import java.lang.reflect.Proxy
+
+/**
+ * Incremental-sync cursors: schema 27→28 adds `updated_at` to `boards`, `alerts` and
+ * `telemetry_minute_buckets`, backfills it from each table's best evidence of last change, and
+ * indexes it. Every write path then has to move it.
+ *
+ * @parity /modules/vescape-core/ios/telemetry/SyncCursorMigrationTests.swift
+ */
+class SyncCursorMigrationTest {
+  /** Table → the column its pre-28 rows backfill from. */
+  private val backfillSource = mapOf(
+    "boards" to "created_at",
+    "alerts" to "created_at",
+    "telemetry_minute_buckets" to "last_sample_at_ms",
+  )
+
+  private fun migrationSql(): List<String> {
+    val sql = mutableListOf<String>()
+    val db = Proxy.newProxyInstance(
+      SupportSQLiteDatabase::class.java.classLoader,
+      arrayOf(SupportSQLiteDatabase::class.java),
+    ) { _, method, args ->
+      if (method.name == "execSQL") {
+        sql += args?.firstOrNull() as String
+        null
+      } else {
+        throw UnsupportedOperationException(method.name)
+      }
+    } as SupportSQLiteDatabase
+    TelemetryDatabase.MIGRATION_27_28.migrate(db)
+    return sql
+  }
+
+  @Test
+  fun migrationAddsUpdatedAtColumnAndIndexToEverySyncedTable() {
+    val sql = migrationSql()
+
+    for (table in backfillSource.keys) {
+      assertTrue(
+        "missing updated_at column on $table",
+        sql.any { it == "ALTER TABLE $table ADD COLUMN updated_at INTEGER NOT NULL DEFAULT 0" },
+      )
+      assertTrue(
+        "missing updated_at index on $table",
+        sql.any {
+          it == "CREATE INDEX IF NOT EXISTS index_${table}_updated_at ON $table(updated_at)"
+        },
+      )
+    }
+  }
+
+  /**
+   * The backfill is the whole point of shipping this as a migration rather than a plain column add:
+   * a row left at the `DEFAULT 0` would report epoch zero to the server and get re-synced forever.
+   */
+  @Test
+  fun migrationBackfillsExistingRowsInsteadOfLeavingThemAtZero() {
+    val sql = migrationSql()
+
+    for ((table, source) in backfillSource) {
+      val added = sql.indexOf("ALTER TABLE $table ADD COLUMN updated_at INTEGER NOT NULL DEFAULT 0")
+      val backfilled = sql.indexOf("UPDATE $table SET updated_at = $source")
+      assertTrue("missing backfill for $table", backfilled >= 0)
+      // The backfill only works once the column exists.
+      assertTrue("backfill for $table runs before the column is added", backfilled > added)
+    }
+  }
+
+  @Test
+  fun migrationTargetsTheCurrentSchemaVersion() {
+    assertEquals(28, TELEMETRY_DATABASE_VERSION)
+    assertEquals(27, TelemetryDatabase.MIGRATION_27_28.startVersion)
+    assertEquals(28, TelemetryDatabase.MIGRATION_27_28.endVersion)
+  }
+
+  /**
+   * The regression this whole change exists to prevent. `setAlertRuleEnabled` is a targeted UPDATE
+   * rather than an entity round-trip, so it is the one write path that can silently skip the cursor
+   * — toggling an alert would then never reach the server.
+   *
+   * Asserted against the DAO source because Room's `@Query` has BINARY retention (invisible to
+   * runtime reflection) and its generated implementation keeps the SQL in a method-local string.
+   * A JVM unit test has no other handle on the statement Room will actually run.
+   */
+  @Test
+  fun setAlertRuleEnabledQueryBumpsUpdatedAt() {
+    val dao = File("src/main/java/expo/modules/vescapecore/telemetry/TelemetryDao.kt").readText()
+    val query = Regex("""@Query\("(UPDATE alerts SET[^"]*)"\)""").find(dao)?.groupValues?.get(1)
+
+    assertEquals(
+      "UPDATE alerts SET enabled = :enabled, updated_at = :updatedAt " +
+        "WHERE board_id = :boardId AND id = :id",
+      query,
+    )
+  }
+
+  @Test
+  fun boardAndAlertRuleBridgeShapesCarryTheCursor() {
+    val board = mapOf(
+      "id" to "board-1",
+      "name" to "ADV",
+      "createdAt" to 1_000L,
+      // Native ignores a bridge-supplied cursor and stamps its own.
+      "updatedAt" to 1L,
+    ).toBoardEntity(now = 2_000L)
+
+    assertEquals(1_000L, board.createdAt)
+    assertEquals(2_000L, board.updatedAt)
+    assertEquals(2_000L, board.toMap(emptyList())["updatedAt"])
+
+    val rule = mapOf(
+      "boardId" to "board-1",
+      "id" to "rule-1",
+      "controlId" to "duty",
+      "threshold" to 70.0,
+      "enabled" to true,
+      "createdAt" to 1_000L,
+      "updatedAt" to 1L,
+    ).toAlertRuleEntity(now = 2_000L)
+
+    assertEquals(1_000L, rule.createdAt)
+    assertEquals(2_000L, rule.updatedAt)
+    assertEquals(2_000L, rule.toMap()["updatedAt"])
+  }
+}