Merge synced channel records field by field

app/src/main/java/com/pombo/android/ChannelManager.kt

@@ -777,10 +777,23 @@ class ChannelManager(
     // the store on the next save.
     internal val _channels = MutableStateFlow(store.load().distinctBy { it.messageStreamId })
 
+    /**
+     * Each record as last persisted or imported: what [saveChannels] compares
+     * against to stamp the fields that changed here.
+     */
+    @Volatile private var persisted: Map<String, Channel> = _channels.value.associateBy { it.messageStreamId }
+
+    /** Writes the list without stamping it: loads, imports and derived repairs. */
+    private fun persist(channels: List<Channel>) {
+        store.save(channels)
+        persisted = channels.associateBy { it.messageStreamId }
+    }
+
     /** Re-reads the list after the storage scope changes (account switch/guest). */
     fun reloadChannels() {
         scope.launch { channelSwitchMutex.withLock { closeCurrentInternal(); clearReopenSnapshots() } }
         _channels.value = store.load().distinctBy { it.messageStreamId }
+        persisted = _channels.value.associateBy { it.messageStreamId }
         _channelOrder.value = store.loadOrder()
     }
 
@@ -1287,12 +1300,11 @@ class ChannelManager(
             com.pombo.android.core.GraphApi.getChannelInfo(channel.messageStreamId)
         } catch (e: Exception) { null } ?: return false
         if (info.name.isNullOrEmpty()) return false
-        val fixed = channel.copy(exposure = info.exposure.ifEmpty { "visible" })
-        _channels.value = _channels.value.map {
-            if (it.messageStreamId == fixed.messageStreamId) fixed else it
+        val exposure = info.exposure.ifEmpty { "visible" }
+        if (updateStored(channel.messageStreamId) { it.copy(exposure = exposure) } == null) {
+            _current.value?.takeIf { it.messageStreamId == channel.messageStreamId }
+                ?.let { _current.value = it.copy(exposure = exposure) }
         }
-        saveChannels()
-        if (_current.value?.messageStreamId == fixed.messageStreamId) _current.value = fixed
         Log.i(TAG, "rename: chain says this channel is named publicly — exposure corrected")
         return true
     }
@@ -1312,16 +1324,16 @@ class ChannelManager(
             bridge.call("updateStreamMetadata", args, 120_000)
         }
 
-        val updated = channel.copy(
-            name = name?.trim()?.ifEmpty { null } ?: channel.name,
-            description = description?.trim() ?: channel.description,
-            // Stamped so the Graph refresh below does not revert this edit while
-            // the subgraph is still catching up.
-            metaUpdatedAt = System.currentTimeMillis()
-        )
-        _channels.value = _channels.value.map { if (it.messageStreamId == updated.messageStreamId) updated else it }
-        saveChannels()
-        _current.value = updated
+        val editedAt = System.currentTimeMillis()
+        updateStored(channel.messageStreamId) {
+            it.copy(
+                name = name?.trim()?.ifEmpty { null } ?: it.name,
+                description = description?.trim() ?: it.description,
+                // Stamped so the Graph refresh below does not revert this edit while
+                // the subgraph is still catching up.
+                metaUpdatedAt = editedAt
+            )
+        }
     }
 
     /**
@@ -1332,15 +1344,20 @@ class ChannelManager(
      * falls back to.
      */
     suspend fun refreshChannelMetadataFromGraph(): Boolean {
-        var changed = false
-        val updates = mutableMapOf<String, Channel>()
+        val infos = mutableMapOf<String, com.pombo.android.core.GraphApi.ChannelInfo>()
         for (channel in _channels.value) {
             if (channel.type == "dm") continue
-            val info = try {
+            infos[channel.messageStreamId] = try {
                 com.pombo.android.core.GraphApi.getChannelInfo(channel.messageStreamId)
             } catch (e: Exception) {
                 null
             } ?: continue
+        }
+        // Compared with the records as they are after the lookups, which may
+        // have run long enough for a sync import or a local edit to land.
+        val updates = mutableMapOf<String, Channel>()
+        for (channel in _channels.value) {
+            val info = infos[channel.messageStreamId] ?: continue
             // Indexing lag: never overwrite a local admin edit with older data.
             if (channel.metaUpdatedAt != null && info.updatedAt <= channel.metaUpdatedAt) continue
 
@@ -1359,13 +1376,18 @@ class ChannelManager(
             if (info.exposure == "visible" && info.description != channel.description) {
                 next = next.copy(description = info.description)
             }
-            if (next !== channel) { updates[channel.messageStreamId] = next; changed = true }
+            if (next !== channel) updates[channel.messageStreamId] = next
         }
+        val changed = updates.isNotEmpty()
         if (changed) {
             _channels.value = _channels.value.map { updates[it.messageStreamId] ?: it }
             saveChannels()
             // Keep the open channel's header in sync with the list.
-            _current.value?.let { cur -> updates[cur.messageStreamId]?.let { _current.value = it } }
+            _current.value?.let { cur ->
+                if (cur.messageStreamId in updates) {
+                    _channels.value.find { it.messageStreamId == cur.messageStreamId }?.let { _current.value = it }
+                }
+            }
         }
         return changed
     }
@@ -1775,20 +1797,15 @@ class ChannelManager(
         // with the old retention. A stream already at `days` counts as in
         // sync too, which is the whole point of not writing to it.
         fun settled(kind: String) = out.results[kind] == "applied" || out.results[kind] == "unchanged"
-        val updated = channel.copy(
-            storageDays = if (settled("message")) days else channel.storageDays,
-            adminStorageDays = if (settled("admin")) days else channel.adminStorageDays,
-            keysStorageDays = if (settled("keys")) days else channel.keysStorageDays,
+        val latest = _channels.value.find { it.messageStreamId == channel.messageStreamId } ?: return out
+        val updated = latest.copy(
+            storageDays = if (settled("message")) days else latest.storageDays,
+            adminStorageDays = if (settled("admin")) days else latest.adminStorageDays,
+            keysStorageDays = if (settled("keys")) days else latest.keysStorageDays,
             interactionsStorageDays =
-                if (settled("interactions")) days else channel.interactionsStorageDays
+                if (settled("interactions")) days else latest.interactionsStorageDays
         )
-        if (updated != channel) {
-            _channels.value = _channels.value.map {
-                if (it.messageStreamId == updated.messageStreamId) updated else it
-            }
-            saveChannels()
-            if (_current.value?.messageStreamId == updated.messageStreamId) _current.value = updated
-        }
+        if (updated != latest) updateStored(channel.messageStreamId) { updated }
         return out
     }
 
@@ -4687,7 +4704,7 @@ class ChannelManager(
             }
             // Local only: these ids derive from the stream id, and the web never stores
             // keysStreamId, so a sync push here would follow each of its pulls.
-            store.save(_channels.value)
+            persist(_channels.value)
         }
         return healed
     }
@@ -5301,7 +5318,7 @@ class ChannelManager(
     /** Replaces the channel list wholesale after a sync merge. */
     fun replaceChannels(channels: List<Channel>) {
         _channels.value = channels
-        store.save(channels)
+        persist(channels)
         reconcileAllGateAuthority()
     }
 
@@ -7183,9 +7200,31 @@ class ChannelManager(
         saveChannels()
     }
 
-    /** Every local change goes through here; a sync import must not, or each pull is pushed back out. */
+    /**
+     * Changes the stored record as it is now, not a copy captured before a
+     * slow call: a sync import may have landed meanwhile, and the save stamps
+     * every field that differs, so a stale copy would win them all on every
+     * device. Null when the channel is not stored.
+     */
+    internal fun updateStored(messageStreamId: String, change: (Channel) -> Channel): Channel? {
+        val latest = _channels.value.find { it.messageStreamId == messageStreamId } ?: return null
+        val updated = change(latest)
+        _channels.value = _channels.value.map { if (it.messageStreamId == messageStreamId) updated else it }
+        saveChannels()
+        val stored = _channels.value.find { it.messageStreamId == messageStreamId } ?: updated
+        if (_current.value?.messageStreamId == messageStreamId) _current.value = stored
+        return stored
+    }
+
+    /**
+     * Every local change goes through here; a sync import must not, or each
+     * pull is pushed back out, and its fields would be stamped as changed here.
+     */
     internal fun saveChannels() {
-        store.save(_channels.value)
+        val now = System.currentTimeMillis()
+        val stamped = _channels.value.map { it.stampedAgainst(persisted[it.messageStreamId], now) }
+        _channels.value = stamped
+        persist(stamped)
         onLocalStateChanged()
     }
 

app/src/main/java/com/pombo/android/core/SyncMerge.kt

@@ -135,7 +135,8 @@ object SyncMerge {
      * LWW-element-set over channels. Membership is decided by comparing the
      * join timestamp against the leave tombstone; the most recent action wins
      * and a tie goes to Join. Snapshot replacement would let an older remote
-     * snapshot delete a channel this device just joined.
+     * snapshot delete a channel this device just joined. Two copies of the
+     * same channel merge field by field ([mergeChannelRecord]).
      */
     private fun mergeChannels(
         baseChannels: JSONArray?,
@@ -170,7 +171,11 @@ object SyncMerge {
                 val id = c.optStringOrNull("messageStreamId") ?: continue
                 noteJoin(id, c)
                 val existing = byId[id]
-                if (existing == null || replaces(c, existing)) byId[id] = c
+                byId[id] = when {
+                    existing == null -> c
+                    replaces(c, existing) -> mergeChannelRecord(c, existing)
+                    else -> mergeChannelRecord(existing, c)
+                }
             }
         }
 
@@ -185,14 +190,53 @@ object SyncMerge {
         return channels to leftAt
     }
 
+    private fun fieldStamp(record: JSONObject, key: String): Long =
+        (record.optJSONObject("fieldTs")?.opt(key) as? Number)?.toLong() ?: 0L
+
+    /**
+     * One channel record from two copies of it (web mergeChannelRecord). Each
+     * field comes from the copy whose `fieldTs` stamped it later; a field
+     * neither copy stamped, or both stamped at the same time, comes from
+     * [preferred], and a field only one copy carries is kept. The stamps join,
+     * the latest per field, so a client that drops them cannot take them from
+     * the others. Returns [preferred] itself when nothing comes from [other].
+     */
+    fun mergeChannelRecord(preferred: JSONObject, other: JSONObject): JSONObject {
+        val merged = JSONObject()
+        var tookOther = false
+        val keys = LinkedHashSet<String>()
+        preferred.keys().forEach { keys.add(it) }
+        other.keys().forEach { keys.add(it) }
+        for (key in keys) {
+            if (key == "fieldTs") continue
+            val fromOther = other.has(key) &&
+                (!preferred.has(key) || fieldStamp(other, key) > fieldStamp(preferred, key))
+            merged.put(key, if (fromOther) other.get(key) else preferred.get(key))
+            if (fromOther) tookOther = true
+        }
+        if (!tookOther) return preferred
+        val sources = listOfNotNull(preferred.optJSONObject("fieldTs"), other.optJSONObject("fieldTs"))
+        if (sources.isNotEmpty()) {
+            val stamps = JSONObject()
+            sources.forEach { src ->
+                src.keys().forEach { key ->
+                    val ts = (src.opt(key) as? Number)?.toLong() ?: return@forEach
+                    if (ts > stamps.optLong(key, 0L)) stamps.put(key, ts)
+                }
+            }
+            merged.put("fieldTs", stamps)
+        }
+        return merged
+    }
+
     private fun joinTs(channel: JSONObject): Long {
         val joined = channel.optLong("joinedAt", 0L)
         return if (joined > 0L) joined else channel.optLong("createdAt", 0L)
     }
 
-    /** Incoming wins ties: the newer snapshot carries fresher metadata. An
-     *  entry with no joinedAt of its own is never a newer join than one that
-     *  has it, but its time still counts as a join against a leave tombstone. */
+    /** For the fields no stamp decides: incoming wins ties. An entry with no
+     *  joinedAt of its own is never a newer join than one that has it, but its
+     *  time still counts as a join against a leave tombstone. */
     private fun replaces(incoming: JSONObject, existing: JSONObject): Boolean {
         val incomingJoined = incoming.optLong("joinedAt", 0L) > 0L
         val existingJoined = existing.optLong("joinedAt", 0L) > 0L

app/src/main/java/com/pombo/android/core/channels/Moderation.kt

@@ -99,10 +99,11 @@ internal class Moderation(private val manager: ChannelManager) {
                 val keysId = channel.keysStreamId.ifEmpty { StreamConstants.deriveKeysId(messageStreamId) }
                 epochKeys.rotateEpoch(messageStreamId, keysId)
             }
-            override suspend fun covered(messageStreamId: String, addresses: Set<String>) =
+            override suspend fun covered(messageStreamId: String, addresses: Set<String>) {
                 updateStored(messageStreamId) {
                     it.copy(rotatedForNoAccess = (it.rotatedForNoAccess + addresses).distinct())
                 }
+            }
             override fun stillOwned(messageStreamId: String): Boolean =
                 _channels.value.find { it.messageStreamId == messageStreamId }?.let { amOwner(it) } == true
             override fun loadOwed(key: String): List<String> {
@@ -122,13 +123,8 @@ internal class Moderation(private val manager: ChannelManager) {
         _channels.value.filter { it.type == "gated" && amOwner(it) }.map { it.messageStreamId })
 
     /** Change the stored record as it is now, not a copy captured before a slow call. */
-    private fun updateStored(messageStreamId: String, change: (Channel) -> Channel) {
-        val latest = _channels.value.find { it.messageStreamId == messageStreamId } ?: return
-        val updated = change(latest)
-        _channels.value = _channels.value.map { if (it.messageStreamId == messageStreamId) updated else it }
-        manager.saveChannels()
-        if (_current.value?.messageStreamId == messageStreamId) _current.value = updated
-    }
+    private fun updateStored(messageStreamId: String, change: (Channel) -> Channel): Channel? =
+        manager.updateStored(messageStreamId, change)
 
     internal val _pins = MutableStateFlow<List<Pin>>(emptyList())
 
@@ -546,10 +542,8 @@ internal class Moderation(private val manager: ChannelManager) {
             val gate = channel.gateAddress
                 ?: throw IllegalStateException("Gate address unknown (repair pending)")
             bridge.call("gateAllow", JSONObject().put("gate", gate).put("user", addr), 180_000)
-            val updated = channel.copy(members = channel.members + addr)
-            _channels.value = _channels.value.map { if (it.messageStreamId == updated.messageStreamId) updated else it }
-            manager.saveChannels()
-            _current.value = updated
+            val updated = updateStored(channel.messageStreamId) { withMember(it, addr) }
+                ?: withMember(channel, addr)
             answerWaitingRequests(updated)
             return
         }
@@ -572,12 +566,14 @@ internal class Moderation(private val manager: ChannelManager) {
         setPermissionsRetry(keysId, rw)
 
         com.pombo.android.core.GraphApi.clearCache()
-        val updated = channel.copy(members = channel.members + addr)
-        _channels.value = _channels.value.map { if (it.messageStreamId == updated.messageStreamId) updated else it }
-        manager.saveChannels()
-        _current.value = updated
+        updateStored(channel.messageStreamId) { withMember(it, addr) }
     }
 
+    /** [channel] with [addr] among its members, once. */
+    private fun withMember(channel: Channel, addr: String): Channel =
+        if (channel.members.any { it.equals(addr, ignoreCase = true) }) channel
+        else channel.copy(members = channel.members + addr)
+
     /**
      * Revokes all permissions (web: empty permission array = revoke).
      * @return false when the key rotation that follows is still owed
@@ -624,10 +620,9 @@ internal class Moderation(private val manager: ChannelManager) {
         setPermissionsRetry(keysId, revoke)
 
         com.pombo.android.core.GraphApi.clearCache()
-        val updated = channel.copy(members = channel.members.filterNot { it.equals(addr, ignoreCase = true) })
-        _channels.value = _channels.value.map { if (it.messageStreamId == updated.messageStreamId) updated else it }
-        manager.saveChannels()
-        _current.value = updated
+        updateStored(channel.messageStreamId) { latest ->
+            latest.copy(members = latest.members.filterNot { it.equals(addr, ignoreCase = true) })
+        }
 
         // Rotate the epoch so the removed member cannot read anything published
         // from here on — they keep what they already read; the rotation protects