diff --git a/app/src/main/kotlin/io/flamenet/messenger/app/Session.kt b/app/src/main/kotlin/io/flamenet/messenger/app/Session.kt index 8c4241b..d45a3a8 100644 --- a/app/src/main/kotlin/io/flamenet/messenger/app/Session.kt +++ b/app/src/main/kotlin/io/flamenet/messenger/app/Session.kt @@ -257,13 +257,20 @@ class Session( Conversations.Sent { var delivered = 0 var failed = 0 + val at = System.currentTimeMillis() + val wire = io.flamenet.e2e.Content.sticker(pack, stickerId, groupId = group.id) for (member in group.memberIds) { if (member == userId) continue val one = conversations.sendSticker(member, pack, stickerId, groupId = group.id) delivered += one.delivered.size failed += one.failed.size + // The same rule as group text: queue exactly the members nothing + // reached, with the group thread to restate. + if (!one.reachedSomeone) { + enqueue(member, wire, at, thread = Vault.threadForGroup(group.id)) + } } - record(Vault.threadForGroup(group.id), "", mine = true, author = userId, + record(Vault.threadForGroup(group.id), "", mine = true, author = userId, at = at, sticker = Sticker(pack, stickerId), delivery = if (delivered > 0) Delivery.DELIVERED else Delivery.PENDING) return Conversations.Sent( @@ -343,10 +350,15 @@ class Session( * which transmits its argument as the plaintext verbatim — so both kinds * take the same path out. */ - private fun enqueue(peerUserId: Long, text: String, at: Long) { + private fun enqueue(peerUserId: Long, text: String, at: Long, + thread: String? = null) { val pending = vault.outbox() pending.put(JSONObject() - .put("to", peerUserId).put("text", text).put("at", at)) + .put("to", peerUserId).put("text", text).put("at", at) + // ⚠️ Present only for group entries: the bubble to restate lives in + // the GROUP thread, not in a 1:1 thread with this member — which + // may not even exist. + .also { if (thread != null) it.put("thread", thread) }) vault.saveOutbox(pending) } @@ -366,23 +378,50 @@ class Session( val touched = mutableSetOf() val stillWaiting = org.json.JSONArray() + // Group entries that delivered this pass, keyed "thread|at". The bubble + // only flips when the LAST member of its set is reached — see below. + val groupDelivered = mutableSetOf>() for (i in 0 until pending.length()) { val item = pending.optJSONObject(i) ?: continue val to = item.optLong("to") val text = item.optString("text") val at = item.optLong("at") - val threadId = Vault.threadForUser(to) + val groupThread = item.optString("thread").takeIf { it.isNotBlank() } val sent = runCatching { conversations.send(to, text) }.getOrNull() if (sent?.reachedSomeone == true) { - restate(threadId, at, Delivery.DELIVERED) - touched += threadId + if (groupThread == null) { + restate(Vault.threadForUser(to), at, Delivery.DELIVERED) + touched += Vault.threadForUser(to) + } else { + groupDelivered += groupThread to at + } } else { stillWaiting.put(item) } } vault.saveOutbox(stillWaiting) + + // 🔴 **A group bubble flips only when nobody is left waiting on it.** + // One message fans out to many members and each gets its own entry; + // restating on the first success would mark it delivered while other + // members are still owed it, and "will send" would quietly stop being + // true for them. Until the set drains, PENDING remains the honest + // label — the queue genuinely will keep sending. + for ((thread, at) in groupDelivered) { + var remaining = false + for (i in 0 until stillWaiting.length()) { + val item = stillWaiting.optJSONObject(i) ?: continue + if (item.optString("thread") == thread && item.optLong("at") == at) { + remaining = true; break + } + } + if (!remaining) { + restate(thread, at, Delivery.DELIVERED) + touched += thread + } + } return touched } @@ -446,8 +485,20 @@ class Session( fun sendGroup(group: RelayClient.Group, text: String): Conversations.Sent { val sent = conversations.sendToGroup(group, text, userId) - record(Vault.threadForGroup(group.id), text, mine = true, author = userId, + val at = System.currentTimeMillis() + record(Vault.threadForGroup(group.id), text, mine = true, author = userId, at = at, delivery = if (sent.reachedSomeone) Delivery.DELIVERED else Delivery.PENDING) + // 🔴 **The group half of the outbox promise.** Members with no + // registered device are filed in the failure map as "user:" by the + // no-device guard; each gets an entry carrying the SAME group body and + // the group thread to restate. A member whose devices took it is not + // queued — resending to them would duplicate the message. + val body = io.flamenet.e2e.Content.groupText(text, group.id) + sent.failed.keys.filter { it.startsWith("user:") } + .mapNotNull { it.removePrefix("user:").toLongOrNull() } + .forEach { member -> + enqueue(member, body, at, thread = Vault.threadForGroup(group.id)) + } return sent } diff --git a/app/src/test/kotlin/io/flamenet/messenger/app/OutboxTest.kt b/app/src/test/kotlin/io/flamenet/messenger/app/OutboxTest.kt index 51b5378..a520b9c 100644 --- a/app/src/test/kotlin/io/flamenet/messenger/app/OutboxTest.kt +++ b/app/src/test/kotlin/io/flamenet/messenger/app/OutboxTest.kt @@ -222,6 +222,64 @@ class OutboxTest { "restating the delivery must not turn the sticker into text") } + /** + * 🔴 The group half of the promise, with its completion rule made + * observable: the bubble flips only when the LAST queued member is + * reached. Restating on the first success would mark the message delivered + * while other members are still owed it — "will send" would quietly stop + * being true for exactly the people it was still true for. + */ + @Test + fun `a group message waits for every queued member before the bubble flips`() { + val session = session() + val group = io.flamenet.client.RelayClient.Group( + id = "g1", name = "test", owner = 1, memberIds = listOf(101L, 102L)) + + // Both members signed out: two device lookups, both empty. + server.enqueue(json("""{"devices":[]}""")) + server.enqueue(json("""{"devices":[]}""")) + session.sendGroup(group, "meeting moved to three") + + val thread = Vault.threadForGroup("g1") + assertEquals(Session.Delivery.PENDING, session.history(thread)[0].delivery) + assertEquals(2, session.vault.outbox().length(), + "one entry per member nothing reached") + + // Member 101 registers; 102 is still away. + val m1 = DeviceIdentity.create() + val opk1 = m1.generateOneTimePreKeys(1).first() + server.enqueue(json(org.json.JSONObject().put("devices", + org.json.JSONArray().put(m1.registrationWire().put("opk", + org.json.JSONObject().put("id", opk1.id).put("pub", + android.util.Base64.encodeToString(opk1.publicKey, + android.util.Base64.NO_WRAP))))).toString())) + server.enqueue(json("""{"ok":true}""")) + server.enqueue(json("""{"devices":[]}""")) // 102 still gone + session.flushOutbox() + + assertEquals(1, session.vault.outbox().length(), + "the reached member leaves the queue; the other stays") + assertEquals(Session.Delivery.PENDING, session.history(thread)[0].delivery, + "one member reached is not the group delivered") + + // Member 102 comes back. + val m2 = DeviceIdentity.create() + val opk2 = m2.generateOneTimePreKeys(1).first() + server.enqueue(json(org.json.JSONObject().put("devices", + org.json.JSONArray().put(m2.registrationWire().put("opk", + org.json.JSONObject().put("id", opk2.id).put("pub", + android.util.Base64.encodeToString(opk2.publicKey, + android.util.Base64.NO_WRAP))))).toString())) + server.enqueue(json("""{"ok":true}""")) + val touched = session.flushOutbox() + + assertEquals(0, session.vault.outbox().length()) + assertTrue(thread in touched, "the GROUP thread is what gets redrawn") + assertEquals(Session.Delivery.DELIVERED, session.history(thread)[0].delivery) + assertEquals("meeting moved to three", session.history(thread)[0].text, + "the group body must not be replaced by the wire form") + } + @Test fun `a message stamped once is the one that gets restated`() { val session = session()