fix(sync): let sync remove a channel from a group

.github/nightly-notes/2026-10-29-fix-sync-group-members.md

@@ -0,0 +1,2 @@
+## Fixed
+- Removing a channel from a subscription group now syncs to your other devices

app/src/main/java/io/github/aedev/flow/sync/apply/SyncChangeLedgerStore.kt

@@ -62,12 +62,13 @@ class SyncChangeLedgerStore
             const val LIKES = "likes"
             const val NOTES = "notes"
             const val GROUPS = "subscriptions"
+            const val GROUP_MEMBERS = "subscription_members"
             const val PLAYLISTS = "playlists"
             const val PLAYLIST_ITEMS = "playlist_items"
             const val SETTINGS = "settings"
             const val SETTINGS_RELAY = "settings_relay"
 
-            val LEDGERED = listOf(WATCH_HISTORY, LIKES, NOTES, GROUPS, PLAYLISTS, PLAYLIST_ITEMS, SETTINGS, SETTINGS_RELAY)
+            val LEDGERED = listOf(WATCH_HISTORY, LIKES, NOTES, GROUPS, GROUP_MEMBERS, PLAYLISTS, PLAYLIST_ITEMS, SETTINGS, SETTINGS_RELAY)
         }
     }
 

app/src/main/java/io/github/aedev/flow/sync/apply/SyncChangeTracker.kt

@@ -1,13 +1,15 @@
 package io.github.aedev.flow.sync.apply
 
 import io.github.aedev.flow.sync.apply.SyncChangeLedgerStore.Companion.GROUPS
+import io.github.aedev.flow.sync.apply.SyncChangeLedgerStore.Companion.GROUP_MEMBERS
 import io.github.aedev.flow.sync.apply.SyncChangeLedgerStore.Companion.LIKES
 import io.github.aedev.flow.sync.apply.SyncChangeLedgerStore.Companion.NOTES
 import io.github.aedev.flow.sync.apply.SyncChangeLedgerStore.Companion.PLAYLISTS
 import io.github.aedev.flow.sync.apply.SyncChangeLedgerStore.Companion.PLAYLIST_ITEMS
 import io.github.aedev.flow.sync.apply.SyncChangeLedgerStore.Companion.SETTINGS
 import io.github.aedev.flow.sync.apply.SyncChangeLedgerStore.Companion.SETTINGS_RELAY
 import io.github.aedev.flow.sync.apply.SyncChangeLedgerStore.Companion.WATCH_HISTORY
+import io.github.aedev.flow.sync.canonical.CanonicalGroupMember
 import io.github.aedev.flow.sync.canonical.CanonicalLike
 import io.github.aedev.flow.sync.canonical.CanonicalNote
 import io.github.aedev.flow.sync.canonical.CanonicalPlaylist
@@ -115,20 +117,45 @@ class SyncChangeTracker
             val live = data.readSubscriptions(hlc = "")
             val at = now()
             val ledger = store.update(GROUPS) { it.observe(live.associate { g -> g.name to groupHash(g) }, at) }
-            return live.map { it.copy(hlc = stamp(ledger.stampOf(it.name), node)) } +
+            val members = store.update(GROUP_MEMBERS) { it.observe(memberKeys(live), at) }
+            return live.map { g ->
+                val present = g.channelIds.toSet()
+                g.copy(
+                    hlc = stamp(ledger.stampOf(g.name), node),
+                    members =
+                        (
+                            present.map { CanonicalGroupMember(it, stamp(members.stampOf(memberKey(g.name, it)), node)) } +
+                                tombstonesUnder(members, memberKey(g.name, ""), present, CanonicalGroupMember.serializer()) { id, atMs ->
+                                    CanonicalGroupMember(id, stamp(atMs, node), deleted = true)
+                                }
+                        ).sortedBy { it.channelId },
+                )
+            } +
                 ledger.tombstones().map { (name, e) ->
                     peerRecord(e, CanonicalSubscriptionGroup.serializer())
                         ?: CanonicalSubscriptionGroup(name = name, hlc = stamp(e.atMs, node), deleted = true)
                 }
         }
 
-        suspend fun settleGroups(merged: List<CanonicalSubscriptionGroup>) =
+        suspend fun settleGroups(merged: List<CanonicalSubscriptionGroup>) {
+            val written = data.readSubscriptions(hlc = "")
+            val (deleted, live) = merged.partition { it.deleted }
             settle(
                 GROUPS,
-                data.readSubscriptions(hlc = "").associate { it.name to groupHash(it) },
-                merged.filter { it.deleted }.associate { it.name to peer(it.hlc, it, CanonicalSubscriptionGroup.serializer()) },
-                stamps = merged.filter { !it.deleted }.associate { it.name to physical(it.hlc) },
+                written.associate { it.name to groupHash(it) },
+                deleted.associate { it.name to peer(it.hlc, it, CanonicalSubscriptionGroup.serializer()) },
+                stamps = live.associate { it.name to physical(it.hlc) },
+            )
+            val memberStates = live.flatMap { g -> g.members.map { memberKey(g.name, it.channelId) to it } }
+            settle(
+                GROUP_MEMBERS,
+                memberKeys(written),
+                memberStates.filter { it.second.deleted }.associate { (key, m) ->
+                    key to peer(m.hlc, m, CanonicalGroupMember.serializer())
+                },
+                stamps = memberStates.filter { !it.second.deleted }.associate { (key, m) -> key to physical(m.hlc) },
             )
+        }
 
         // --- playlists ---
 
@@ -148,7 +175,14 @@ class SyncChangeTracker
                         updatedHlc = stamp(playlists.stampOf(p.syncId), node),
                         items =
                             p.items.map { it.copy(hlc = stamp(items.stampOf(itemKey(p.syncId, it.videoId)), node)) } +
-                                itemTombstones(items, p.syncId, present, node),
+                                tombstonesUnder(
+                                    items,
+                                    itemKey(p.syncId, ""),
+                                    present,
+                                    CanonicalPlaylistItem.serializer(),
+                                ) { videoId, atMs ->
+                                    CanonicalPlaylistItem(videoId = videoId, deleted = true, hlc = stamp(atMs, node))
+                                },
                     )
                 }
             val deleted =
@@ -184,21 +218,21 @@ class SyncChangeTracker
             )
         }
 
-        private fun itemTombstones(
+        /** Tombstones of the entries under [prefix] that aren't [present]: a peer's record, or one stamped here. */
+        private fun <T> tombstonesUnder(
             ledger: CollectionLedger,
-            syncId: String,
+            prefix: String,
             present: Set<String>,
-            node: String,
-        ): List<CanonicalPlaylistItem> {
-            val prefix = itemKey(syncId, "")
-            return ledger
+            serializer: KSerializer<T>,
+            make: (id: String, atMs: Long) -> T,
+        ): List<T> =
+            ledger
                 .tombstones()
                 .filterKeys { it.startsWith(prefix) && it.removePrefix(prefix) !in present }
-                .map { (key, e) ->
-                    peerRecord(e, CanonicalPlaylistItem.serializer())
-                        ?: CanonicalPlaylistItem(videoId = key.removePrefix(prefix), deleted = true, hlc = stamp(e.atMs, node))
-                }
-        }
+                .map { (key, e) -> peerRecord(e, serializer) ?: make(key.removePrefix(prefix), e.atMs) }
+
+        private fun memberKeys(groups: List<CanonicalSubscriptionGroup>): Map<String, Int> =
+            groups.flatMap { g -> g.channelIds.map { memberKey(g.name, it) to PRESENT } }.toMap()
 
         private fun playlistTombstone(
             syncId: String,
@@ -319,6 +353,12 @@ class SyncChangeTracker
                 videoId: String,
             ) = "$syncId|$videoId"
 
+            // Group names are typed by the user, so the separator is one a name can't hold.
+            fun memberKey(
+                group: String,
+                channelId: String,
+            ) = "$group\u0000$channelId"
+
             fun groupHash(g: CanonicalSubscriptionGroup) = "${g.channelIds.sorted()}|${g.sortOrder}".hashCode()
 
             fun playlistHash(p: CanonicalPlaylist) = "${p.title}|${p.description}|${p.isMusic}|${p.isUserCreated}|${p.youtubeId}".hashCode()

app/src/main/java/io/github/aedev/flow/sync/canonical/Canonical.kt

@@ -1,6 +1,8 @@
 package io.github.aedev.flow.sync.canonical
 
 import io.github.aedev.flow.sync.identity.Hlc
+import kotlinx.serialization.EncodeDefault
+import kotlinx.serialization.ExperimentalSerializationApi
 import kotlinx.serialization.KSerializer
 import kotlinx.serialization.Serializable
 import kotlinx.serialization.builtins.MapSerializer
@@ -140,6 +142,17 @@ data class CanonicalSubscriptionGroup(
     val sortOrder: Int = 0,
     val hlc: String = "",
     val deleted: Boolean = false,
+    /** Each channel's own stamp, so a removal can sync; `channelIds` stays the live list for apps without it. */
+    @OptIn(ExperimentalSerializationApi::class)
+    @EncodeDefault(EncodeDefault.Mode.NEVER)
+    val members: List<CanonicalGroupMember> = emptyList(),
+)
+
+@Serializable
+data class CanonicalGroupMember(
+    val channelId: String,
+    val hlc: String = "",
+    val deleted: Boolean = false,
 )
 
 /**

app/src/main/java/io/github/aedev/flow/sync/merge/CollectionMergers.kt

@@ -1,5 +1,6 @@
 package io.github.aedev.flow.sync.merge
 
+import io.github.aedev.flow.sync.canonical.CanonicalGroupMember
 import io.github.aedev.flow.sync.canonical.CanonicalLike
 import io.github.aedev.flow.sync.canonical.CanonicalLikeMeta
 import io.github.aedev.flow.sync.canonical.CanonicalNote
@@ -203,17 +204,39 @@ object SubscriptionsMerger {
         x: CanonicalSubscriptionGroup,
         y: CanonicalSubscriptionGroup,
     ): CanonicalSubscriptionGroup {
-        val channelIds = (x.channelIds.toSet() + y.channelIds.toSet()).sorted() // OR-Set union
         val sortWinner = Crdt.preferByHlc(x, x.hlc, y, y.hlc) { it.sortOrder.toString() }
-        return CanonicalSubscriptionGroup(
+        return withMembers(
             name = x.name,
-            channelIds = channelIds,
+            members = members(x) + members(y),
             sortOrder = sortWinner.sortOrder,
             hlc = Crdt.maxHlc(x.hlc, y.hlc),
             deleted = Crdt.resolveDeleted(x.deleted, x.hlc, y.deleted, y.hlc),
         )
     }
 
-    /** Canonical form: channelIds sorted + de-duped, so `a ⊕ a == a`. */
-    private fun normalize(g: CanonicalSubscriptionGroup) = g.copy(channelIds = g.channelIds.toSortedSet().toList())
+    /** A record from an app without per-member stamps counts each of its channels as present at the record's stamp. */
+    private fun members(g: CanonicalSubscriptionGroup): List<CanonicalGroupMember> =
+        g.members.ifEmpty { g.channelIds.map { CanonicalGroupMember(it, g.hlc) } }
+
+    /** Per channel, last writer wins and a removal is a tombstone, as for playlist items. */
+    private fun withMembers(
+        name: String,
+        members: List<CanonicalGroupMember>,
+        sortOrder: Int,
+        hlc: String,
+        deleted: Boolean,
+    ): CanonicalSubscriptionGroup {
+        val merged =
+            members
+                .groupBy { it.channelId }
+                .map { (channelId, versions) ->
+                    versions.reduce { a, b ->
+                        CanonicalGroupMember(channelId, Crdt.maxHlc(a.hlc, b.hlc), Crdt.resolveDeleted(a.deleted, a.hlc, b.deleted, b.hlc))
+                    }
+                }.sortedBy { it.channelId }
+        return CanonicalSubscriptionGroup(name, merged.filter { !it.deleted }.map { it.channelId }, sortOrder, hlc, deleted, merged)
+    }
+
+    /** Canonical form: members merged and sorted, channelIds the live ones, so `a ⊕ a == a`. */
+    private fun normalize(g: CanonicalSubscriptionGroup) = withMembers(g.name, members(g), g.sortOrder, g.hlc, g.deleted)
 }

app/src/test/java/io/github/aedev/flow/sync/MergeConvergenceTest.kt

@@ -5,6 +5,7 @@ import io.github.aedev.flow.sync.canonical.BrainLwwMaps
 import io.github.aedev.flow.sync.canonical.BrainSets
 import io.github.aedev.flow.sync.canonical.CanonicalBrain
 import io.github.aedev.flow.sync.canonical.CanonicalBrainVectors
+import io.github.aedev.flow.sync.canonical.CanonicalGroupMember
 import io.github.aedev.flow.sync.canonical.CanonicalLike
 import io.github.aedev.flow.sync.canonical.CanonicalPlaylist
 import io.github.aedev.flow.sync.canonical.CanonicalPlaylistItem
@@ -22,6 +23,7 @@ import io.github.aedev.flow.sync.merge.PlaylistMerger
 import io.github.aedev.flow.sync.merge.SettingsMerger
 import io.github.aedev.flow.sync.merge.SubscriptionsMerger
 import io.github.aedev.flow.sync.merge.WatchHistoryMerger
+import io.github.aedev.flow.sync.protocol.SyncSerialization
 import kotlinx.serialization.json.JsonPrimitive
 import org.junit.Assert.assertEquals
 import org.junit.Test
@@ -104,6 +106,54 @@ class MergeConvergenceTest {
         assertEquals(listOf("UC1", "UC2", "UC3"), SubscriptionsMerger.merge(a, b).first { it.name == "Tech" }.channelIds)
     }
 
+    @Test
+    fun a_channel_removed_from_a_group_stays_removed() {
+        fun member(
+            id: String,
+            at: Long,
+            deleted: Boolean = false,
+        ) = CanonicalGroupMember(id, hlc(at, 0, "a"), deleted)
+        val kept =
+            listOf(
+                CanonicalSubscriptionGroup(
+                    "Tech",
+                    listOf("UC1", "UC2"),
+                    0,
+                    hlc(100, 0, "a"),
+                    members = listOf(member("UC1", 100), member("UC2", 100)),
+                ),
+            )
+        val removed =
+            listOf(
+                CanonicalSubscriptionGroup(
+                    "Tech",
+                    listOf("UC1"),
+                    0,
+                    hlc(300, 0, "b"),
+                    members = listOf(member("UC1", 100), member("UC2", 300, deleted = true)),
+                ),
+            )
+        val olderApp = listOf(CanonicalSubscriptionGroup("Tech", listOf("UC1", "UC2"), 0, hlc(200, 0, "c")))
+        laws("group members", kept, removed, olderApp, SubscriptionsMerger::merge)
+        assertEquals(listOf("UC1"), SubscriptionsMerger.merge(SubscriptionsMerger.merge(kept, removed), olderApp).single().channelIds)
+
+        val wire = SyncSerialization.encodeSubscriptions(SubscriptionsMerger.merge(kept, removed)).lines.single()
+        assertEquals(
+            "the shape Flow Desktop emits",
+            """{"channelIds":["UC1"],"deleted":false,"hlc":"${hlc(
+                300,
+                0,
+                "b",
+            )}","members":[{"channelId":"UC1","deleted":false,"hlc":"${hlc(100, 0, "a")}"},""" +
+                """{"channelId":"UC2","deleted":true,"hlc":"${hlc(300, 0, "a")}"}],"name":"Tech","sortOrder":0}""",
+            wire,
+        )
+        assertEquals(false, "members" in SyncSerialization.encodeSubscriptions(olderApp).lines.single())
+
+        val laterOlderApp = listOf(CanonicalSubscriptionGroup("Tech", listOf("UC1", "UC2"), 0, hlc(400, 0, "c")))
+        assertEquals(listOf("UC1", "UC2"), SubscriptionsMerger.merge(removed, laterOlderApp).single().channelIds)
+    }
+
     @Test
     fun playlists_converge() {
         fun item(

app/src/test/java/io/github/aedev/flow/sync/apply/SyncChangeTrackerTest.kt

@@ -161,4 +161,18 @@ class SyncChangeTrackerTest {
 
             assertTrue("a clock an hour behind still stamps after the peer", Hlc.decode(edited.hlc).physicalMs > later)
         }
+
+    @Test
+    fun a_channel_removed_from_a_group_goes_out_as_a_member_tombstone() =
+        runBlocking {
+            coEvery { data.readSubscriptions("") } returns
+                listOf(CanonicalSubscriptionGroup(name = "Tech", channelIds = listOf("c1", "c2")))
+            tracker.groups(node)
+            coEvery { data.readSubscriptions("") } returns listOf(CanonicalSubscriptionGroup(name = "Tech", channelIds = listOf("c1")))
+
+            val group = tracker.groups(node).single()
+
+            assertEquals(listOf("c1"), group.channelIds)
+            assertEquals(listOf("c1" to false, "c2" to true), group.members.map { it.channelId to it.deleted })
+        }
 }