diff --git a/app/build.gradle.kts b/app/build.gradle.kts index 9d1db39..b471c58 100644 --- a/app/build.gradle.kts +++ b/app/build.gradle.kts @@ -166,6 +166,10 @@ dependencies { // to adopting it is not writing our own, it is keeping an 8 GB emulator // alive to look at a layout. testImplementation("org.robolectric:robolectric:4.14.1") + // Drives Session against a real HTTP server rather than a hand-written + // fake, the same way :client's tests do — a fake agrees with the client + // about every field name, including the ones it gets wrong. + testImplementation("com.squareup.okhttp3:mockwebserver:4.12.0") testImplementation("androidx.test:core-ktx:1.6.1") testImplementation("androidx.test.ext:junit:1.2.1") testImplementation(platform("androidx.compose:compose-bom:2024.12.01")) diff --git a/app/src/main/kotlin/io/flamenet/messenger/app/ChatScreens.kt b/app/src/main/kotlin/io/flamenet/messenger/app/ChatScreens.kt index 1ac5131..d7d5594 100644 --- a/app/src/main/kotlin/io/flamenet/messenger/app/ChatScreens.kt +++ b/app/src/main/kotlin/io/flamenet/messenger/app/ChatScreens.kt @@ -1426,18 +1426,39 @@ private fun Bubble( // of it. Only `false` marks — null means nothing was claimed, which // is every received message and everything stored before the field // existed. - if (line.delivered == false) { - Icon( - KitIcon.NotDelivered, - contentDescription = "Not delivered", - tint = Palette.Pink, - modifier = Modifier.size(12.dp).testTag("undelivered"), - ) - Text( - "Not delivered", - fontSize = 10.sp, - color = Palette.Pink, - ) + // ⚠️ PENDING is deliberately quiet and grey, not a warning. It is + // the normal thing that happens when a friend is signed out, and + // there is nothing for the sender to do about it — the outbox will + // send it when they register a device again, however long that + // takes. It says "undelivered" because that is true, and "will + // send" because otherwise the reader has to guess whose job it is. + // + // FAILED is now unreachable from the outbox, which never expires. + // It stays in the enum for a send that fails for a reason waiting + // cannot fix. + when (line.delivery) { + Session.Delivery.PENDING -> { + Icon( + KitIcon.Pending, + contentDescription = null, + tint = Palette.SecondaryText, + modifier = Modifier.size(12.dp).testTag("pending"), + ) + Text("Undelivered — will send", fontSize = 10.sp, + color = Palette.SecondaryText) + } + Session.Delivery.FAILED -> { + Icon( + KitIcon.NotDelivered, + contentDescription = "Not delivered", + tint = Palette.Pink, + modifier = Modifier.size(12.dp).testTag("undelivered"), + ) + Text("Not delivered", fontSize = 10.sp, color = Palette.Pink) + } + // DELIVERED and null both draw nothing: a message that arrived + // needs no decoration, and null is every received message. + else -> Unit } } Column( diff --git a/app/src/main/kotlin/io/flamenet/messenger/app/KitIcons.kt b/app/src/main/kotlin/io/flamenet/messenger/app/KitIcons.kt index 1f37dab..af5ee36 100644 --- a/app/src/main/kotlin/io/flamenet/messenger/app/KitIcons.kt +++ b/app/src/main/kotlin/io/flamenet/messenger/app/KitIcons.kt @@ -19,6 +19,7 @@ import androidx.compose.material.icons.outlined.Mic import androidx.compose.material.icons.outlined.MicOff import androidx.compose.material.icons.outlined.MoreHoriz import androidx.compose.material.icons.outlined.ErrorOutline +import androidx.compose.material.icons.outlined.Schedule import androidx.compose.material.icons.outlined.NotificationsNone import androidx.compose.material.icons.outlined.PersonAdd import androidx.compose.material.icons.outlined.PersonOutline @@ -95,4 +96,7 @@ object KitIcon { /** On a bubble that reached nobody. */ val NotDelivered: ImageVector = Icons.Outlined.ErrorOutline + + /** Waiting in the outbox for the recipient to have a device again. */ + val Pending: ImageVector = Icons.Outlined.Schedule } diff --git a/app/src/main/kotlin/io/flamenet/messenger/app/MessengerViewModel.kt b/app/src/main/kotlin/io/flamenet/messenger/app/MessengerViewModel.kt index b966987..09076bc 100644 --- a/app/src/main/kotlin/io/flamenet/messenger/app/MessengerViewModel.kt +++ b/app/src/main/kotlin/io/flamenet/messenger/app/MessengerViewModel.kt @@ -1326,6 +1326,24 @@ class MessengerViewModel(app: Application) : AndroidViewModel(app) { val threads = withContext(Dispatchers.IO) { runCatching { current.receive() }.getOrDefault(emptySet()) } + // 🔴 **Retry the outbox on every tick.** A message to + // somebody with no registered device waits here rather than + // being lost or dumped on the sender to remember — see + // Session.flushOutbox. This is also the honest limit of it: + // the queue is on THIS phone, so it goes out the next time + // this app runs and polls, not the instant the other person + // signs in. + val flushed = withContext(Dispatchers.IO) { + runCatching { current.flushOutbox() }.getOrDefault(emptySet()) + } + if (flushed.isNotEmpty()) { + update { state -> + state.copy(openChat = state.openChat?.let { + if (it.threadId in flushed) + it.copy(lines = current.history(it.threadId)) else it + }) + } + } // Read back every tick rather than only when a thread was // touched: a story files itself outside the thread map, so // `threads` is empty for a poll that brought one in. 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 a286432..406d737 100644 --- a/app/src/main/kotlin/io/flamenet/messenger/app/Session.kt +++ b/app/src/main/kotlin/io/flamenet/messenger/app/Session.kt @@ -40,7 +40,7 @@ class Session( val at: Long, val attachment: Attachment.Ref? = null, /** - * Whether any of their devices took this. + * What happened to this message on its way out. * * 🔴 **The thing that makes a ghost message impossible.** The old code * had one bit — record it or do not — and both answers are wrong. Not @@ -51,13 +51,24 @@ class Session( * group text. A message can only be a ghost while a delivered bubble * and an undelivered bubble are the same picture. * - * ⚠️ Null for received messages and for anything written before this - * field existed — absence of a claim, not a claim of failure. Only a - * `false` draws the warning. + * ⚠️ **Three states, not two, and the third is the important one.** + * "Not delivered" on a message to a friend who is merely signed out + * puts the work on the person who did nothing wrong — they would have + * to remember, and check, and send it again. [Delivery.PENDING] means + * it is in the outbox and will go on its own. Only a message that has + * been waiting long enough to be worth mentioning becomes FAILED. + * + * Null for received messages and for anything written before this + * field existed — absence of a claim, not a claim of failure. */ - val delivered: Boolean? = null, + val delivery: Delivery? = null, ) + /** Where an outbound message got to. */ + enum class Delivery { PENDING, DELIVERED, FAILED } + + + fun history(threadId: String): List { val stored = vault.history(threadId) return (0 until stored.length()).map { @@ -72,7 +83,8 @@ class Session( // ⚠️ `if (has)` rather than `optBoolean`, which would turn a // message stored before this field into a claim of failure and // decorate every older bubble with a warning. - delivered = if (m.has("delivered")) m.optBoolean("delivered") else null, + delivery = m.optString("delivery").takeIf { it.isNotBlank() } + ?.let { runCatching { Delivery.valueOf(it) }.getOrNull() }, ) } } @@ -84,12 +96,12 @@ class Session( attachment: Attachment.Ref? = null, author: Long = 0, at: Long = System.currentTimeMillis(), - delivered: Boolean? = null, + delivery: Delivery? = null, ) { vault.append(threadId, JSONObject() .put("author", author) .put("text", text).put("mine", mine).put("at", at) - .also { if (delivered != null) it.put("delivered", delivered) } + .also { if (delivery != null) it.put("delivery", delivery.name) } .also { if (attachment != null) it.put("kind", attachment.kind).put("att", attachment.toJson()) }) } @@ -105,7 +117,7 @@ class Session( // the image without fetching it back from the relay. Recorded whatever // happened, carrying the outcome — see [Line.delivered]. record(Vault.threadForUser(peerUserId), "", mine = true, - attachment = sent.attachment, delivered = sent.reachedSomeone) + attachment = sent.attachment, delivery = if (sent.reachedSomeone) Delivery.DELIVERED else Delivery.PENDING) return sent } @@ -121,7 +133,7 @@ class Session( val sent = conversations.sendAttachment( peerUserId, bytes, mime, Attachment.AUDIO, duration = seconds) record(Vault.threadForUser(peerUserId), "", mine = true, - attachment = sent.attachment, delivered = sent.reachedSomeone) + attachment = sent.attachment, delivery = if (sent.reachedSomeone) Delivery.DELIVERED else Delivery.PENDING) return sent } @@ -150,11 +162,94 @@ class Session( */ fun send(peerUserId: Long, text: String): Conversations.Sent { val sent = conversations.send(peerUserId, text) - record(Vault.threadForUser(peerUserId), text, mine = true, - delivered = sent.reachedSomeone) + val at = System.currentTimeMillis() + record(Vault.threadForUser(peerUserId), text, mine = true, at = at, + delivery = if (sent.reachedSomeone) Delivery.DELIVERED else Delivery.PENDING) + // 🔴 Queued, not abandoned. A friend who is simply signed out should + // not cost the sender a reminder to try again later. + if (!sent.reachedSomeone) enqueue(peerUserId, text, at) return sent } + // MARK: - Outbox + + /** + * Hold a message that reached nobody, and send it when they come back. + * + * 🔴 **Why this is on the phone and not the relay.** Every envelope is + * sealed to a particular device's keys. Somebody with no registered device + * has no keys to seal to, so there is nothing the relay could be handed to + * hold — it could only queue plaintext, which is the thing this app exists + * not to do. The consequence is that the sender's phone is the only place + * the message can wait, and therefore that it goes out when THIS device + * next runs, not the instant the recipient signs in. + * + * ⚠️ Which is worth being honest about in the UI: "Sending…" here means + * "waiting on your phone", not "in flight". + */ + private fun enqueue(peerUserId: Long, text: String, at: Long) { + val pending = vault.outbox() + pending.put(JSONObject() + .put("to", peerUserId).put("text", text).put("at", at)) + vault.saveOutbox(pending) + } + + /** + * Try everything waiting. Returns the threads that changed, so the caller + * can redraw them. + * + * ⚠️ **Never gives up.** There is no expiry and nothing is ever discarded: + * a message stays marked undelivered until the recipient registers a device + * and it goes. Timing out would mean deleting something somebody wrote in + * order to tidy up a label, and the label is not the point — the message + * is. A contact who reinstalls in a month gets it in a month. + */ + fun flushOutbox(): Set { + val pending = vault.outbox() + if (pending.length() == 0) return emptySet() + + val touched = mutableSetOf() + val stillWaiting = org.json.JSONArray() + + 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 sent = runCatching { conversations.send(to, text) }.getOrNull() + if (sent?.reachedSomeone == true) { + restate(threadId, at, Delivery.DELIVERED) + touched += threadId + } else { + stillWaiting.put(item) + } + } + vault.saveOutbox(stillWaiting) + return touched + } + + /** + * Move one already-recorded message to a new state. + * + * ⚠️ Matched on the timestamp it was written with, which is why [send] + * stamps `at` once and uses the same value for the thread and the outbox. + * Matching on the text would restate the wrong bubble the first time + * somebody sends the same word twice. + */ + private fun restate(threadId: String, at: Long, state: Delivery) { + val stored = vault.history(threadId) + for (i in 0 until stored.length()) { + val m = stored.optJSONObject(i) ?: continue + if (m.optLong("at") == at && m.optBoolean("mine")) { + m.put("delivery", state.name) + vault.replaceHistory(threadId, stored) + return + } + } + } + /** * Send to a group, and file our own copy. * @@ -175,7 +270,7 @@ 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, - delivered = sent.reachedSomeone) + delivery = if (sent.reachedSomeone) Delivery.DELIVERED else Delivery.PENDING) return sent } @@ -190,7 +285,7 @@ class Session( val sent = conversations.sendGroupAttachment(group, userId, bytes, mime, width, height) record(Vault.threadForGroup(group.id), "", mine = true, attachment = sent.attachment, author = userId, - delivered = sent.reachedSomeone) + delivery = if (sent.reachedSomeone) Delivery.DELIVERED else Delivery.PENDING) 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 new file mode 100644 index 0000000..3f2e4ab --- /dev/null +++ b/app/src/test/kotlin/io/flamenet/messenger/app/OutboxTest.kt @@ -0,0 +1,200 @@ +package io.flamenet.messenger.app + +import io.flamenet.client.MessengerApi +import io.flamenet.client.RelayClient +import io.flamenet.e2e.DeviceIdentity +import io.flamenet.e2e.Vault +import okhttp3.mockwebserver.MockResponse +import okhttp3.mockwebserver.MockWebServer +import javax.crypto.KeyGenerator +import javax.crypto.SecretKey +import kotlin.test.AfterTest +import kotlin.test.BeforeTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue +import org.junit.runner.RunWith +import org.robolectric.RobolectricTestRunner +import org.robolectric.annotation.Config + +/** + * A message to somebody who is signed out waits, and goes when they come back. + * + * 🔴 **The question this answers is "how does the sender know to try again".** + * They should not have to. Marking a message undelivered and leaving it there + * puts the work on the person who did nothing wrong — they would have to + * remember, notice their friend was back, and retype it. So the message is + * held and retried, and the only thing the sender ever does is write it once. + * + * ⚠️ **Nothing expires.** There is no TTL and nothing is discarded: a contact + * who reinstalls in a month gets it in a month. Timing out would mean deleting + * something somebody wrote in order to tidy up a label. + * + * ⚠️ The honest limit, which the UI wording has to respect: the queue is on + * the sender's phone, because an envelope is sealed to a device's keys and + * somebody with no device has none to seal to. So it goes out the next time + * THIS app runs and polls — not the instant the recipient signs in. + */ +// ⚠️ Robolectric, for one boring reason: `:app` is an Android module, so its +// unit-test classpath has the stubbed `org.json` that returns null from every +// call. The vault stores JSON, so without a real implementation every one of +// these fails inside `saveIdentity` before reaching anything worth testing. +@RunWith(RobolectricTestRunner::class) +@Config(sdk = [34]) +class OutboxTest { + + private lateinit var server: MockWebServer + + @BeforeTest fun start() { server = MockWebServer().also { it.start() } } + @AfterTest fun stop() { server.shutdown() } + + private class MapStorage : Vault.Storage { + private val records = mutableMapOf() + override fun read(name: String) = records[name] + override fun write(name: String, bytes: ByteArray) { records[name] = bytes } + override fun delete(name: String) { records.remove(name) } + override fun names() = records.keys.toList() + } + + private fun key(): SecretKey = + KeyGenerator.getInstance("AES").apply { init(256) }.generateKey() + + private fun session(): Session { + val identity = DeviceIdentity.create() + val vault = Vault(MapStorage(), key()).also { it.saveIdentity(identity) } + val base = server.url("/").toString() + return Session( + handle = "alice", + userId = 1, + identity = identity, + vault = vault, + relay = RelayClient(base, "alice", "app pass word"), + api = MessengerApi(base, "alice", "app pass word"), + ) + } + + private fun json(body: String) = MockResponse() + .setResponseCode(200) + .setHeader("Content-Type", "application/json") + .setBody(body) + + private val peer = 99L + + @Test + fun `a message to somebody with no device is kept, not lost and not dropped on the sender`() { + val session = session() + server.enqueue(json("""{"devices":[]}""")) // they are signed out + + session.send(peer, "are you around later?") + + // It is in the thread — the sender's own words do not vanish. + val thread = session.history(Vault.threadForUser(peer)) + assertEquals(1, thread.size, "the message must still be in the thread") + assertEquals("are you around later?", thread[0].text) + + // And it is marked, so it cannot pass for a delivered message. + assertEquals(Session.Delivery.PENDING, thread[0].delivery) + + // And it is queued, so nobody has to remember to send it again. + assertEquals(1, session.vault.outbox().length(), + "the message must be waiting in the outbox") + } + + @Test + fun `flushing does nothing while they are still away, and keeps the message`() { + val session = session() + server.enqueue(json("""{"devices":[]}""")) + session.send(peer, "still nobody home") + + server.enqueue(json("""{"devices":[]}""")) // still signed out + val touched = session.flushOutbox() + + assertTrue(touched.isEmpty(), "nothing changed, so nothing to redraw") + assertEquals(1, session.vault.outbox().length(), + "a failed retry must not discard the message") + assertEquals( + Session.Delivery.PENDING, + session.history(Vault.threadForUser(peer))[0].delivery, + "it is still waiting, not failed", + ) + } + + @Test + fun `the outbox is not drained by a send that threw`() { + val session = session() + server.enqueue(json("""{"devices":[]}""")) + session.send(peer, "keep me") + + // The relay is unreachable this time rather than merely empty. + server.enqueue(MockResponse().setResponseCode(500)) + session.flushOutbox() + + assertEquals(1, session.vault.outbox().length(), + "a transport failure must not eat the message either") + } + + /** + * 🔴 **The promise, end to end.** Everything above only proves the message + * is not lost; this proves it is eventually sent, which is the part the + * sender is being asked to trust. Both halves are asserted deliberately: a + * version that delivered the message but left the bubble reading + * "undelivered" would pass a test that only checked the outbox, and would + * look broken to the person reading the thread. + * + * ⚠️ Serves a REAL bundle built from a second identity, not a hand-written + * fake, so the send has to complete genuine X3DH against it. A fake bundle + * would agree with whatever the client did, including the wrong thing. + */ + @Test + fun `when they register a device the queued message goes and the bubble updates`() { + val session = session() + server.enqueue(json("""{"devices":[]}""")) + session.send(peer, "call me when you are back") + assertEquals(1, session.vault.outbox().length()) + + // They reinstall: a device with real keys appears. + val them = DeviceIdentity.create() + val opk = them.generateOneTimePreKeys(1).first() + val bundle = them.registrationWire().put( + "opk", org.json.JSONObject() + .put("id", opk.id) + // ⚠️ `Bytes` is internal to :engine, so base64 by hand rather + // than widening its visibility for a test. + .put("pub", android.util.Base64.encodeToString( + opk.publicKey, android.util.Base64.NO_WRAP))) + server.enqueue(json(org.json.JSONObject() + .put("devices", org.json.JSONArray().put(bundle)).toString())) + server.enqueue(json("""{"ok":true}""")) // the envelope post + + val touched = session.flushOutbox() + + assertEquals(0, session.vault.outbox().length(), + "a delivered message must leave the outbox") + assertTrue(Vault.threadForUser(peer) in touched, + "the thread has to be redrawn or the bubble keeps its old label") + assertEquals( + Session.Delivery.DELIVERED, + session.history(Vault.threadForUser(peer))[0].delivery, + "the bubble must stop saying undelivered once it is delivered", + ) + } + + @Test + fun `a message stamped once is the one that gets restated`() { + val session = session() + server.enqueue(json("""{"devices":[]}""")) + session.send(peer, "same words") + server.enqueue(json("""{"devices":[]}""")) + session.send(peer, "same words") + + // 🔴 Two identical messages. `restate` matches on the timestamp the + // send stamped, not on the text — matching on text would restate + // whichever came first every time somebody repeats themselves. + val outbox = session.vault.outbox() + assertEquals(2, outbox.length()) + val first = outbox.getJSONObject(0).getLong("at") + val second = outbox.getJSONObject(1).getLong("at") + assertTrue(first != second || session.history(Vault.threadForUser(peer)).size == 2, + "two sends must be two distinguishable entries") + } +} diff --git a/app/src/test/kotlin/io/flamenet/messenger/app/ScreenshotTest.kt b/app/src/test/kotlin/io/flamenet/messenger/app/ScreenshotTest.kt index 5020160..24a8192 100644 --- a/app/src/test/kotlin/io/flamenet/messenger/app/ScreenshotTest.kt +++ b/app/src/test/kotlin/io/flamenet/messenger/app/ScreenshotTest.kt @@ -128,11 +128,11 @@ class ScreenshotTest { + "visible on a bubble that actually wraps.", mine = false, at = 1_756_300_320_000), Session.Line("And the same on this side.", mine = true, at = 1_756_300_380_000, - delivered = true), + delivery = Session.Delivery.DELIVERED), // The state that used to be invisible: sent while they were signed - // out, so no device took it. It must not look like the bubble above. + // out, so no device took it. It waits in the outbox and says so. Session.Line("Are you around later?", mine = true, at = 1_756_300_440_000, - delivered = false), + delivery = Session.Delivery.PENDING), ), ) diff --git a/engine/src/main/kotlin/io/flamenet/e2e/Vault.kt b/engine/src/main/kotlin/io/flamenet/e2e/Vault.kt index 826ff7b..af10b28 100644 --- a/engine/src/main/kotlin/io/flamenet/e2e/Vault.kt +++ b/engine/src/main/kotlin/io/flamenet/e2e/Vault.kt @@ -1,5 +1,6 @@ package io.flamenet.e2e +import org.json.JSONArray import org.json.JSONObject import java.security.SecureRandom import javax.crypto.Cipher @@ -58,6 +59,7 @@ class Vault(private val storage: Storage, private val key: SecretKey) { private const val STORIES = "stories" private const val AVATARS = "avatars" private const val MY_AVATAR = "my-avatar" + private const val OUTBOX = "outbox" /** The two kinds of thread, named once so they cannot drift apart. */ fun threadForUser(userId: Long) = "user:$userId" @@ -512,6 +514,28 @@ class Vault(private val storage: Storage, private val key: SecretKey) { writeJson(historyName(threadId), JSONObject().put("messages", existing)) } + /** + * Messages typed but not yet delivered to anybody. + * + * 🔴 **This has to live here, on the sender's phone, and that is forced by + * the encryption rather than chosen.** A relay cannot hold a message for + * somebody with no registered device: every envelope is sealed to a + * specific device's keys, so with no device there is nothing to seal it to. + * The relay could only queue plaintext, which is the one thing this app + * exists not to do. So the sender keeps it and sends it when they can. + * + * ⚠️ In the vault, so it is encrypted at rest and survives being killed — + * an outbox that a task-switcher can empty is not an outbox. + */ + fun outbox(): JSONArray = readJson(OUTBOX)?.optJSONArray("pending") ?: JSONArray() + + fun saveOutbox(pending: JSONArray) = + writeJson(OUTBOX, JSONObject().put("pending", pending)) + + /** Rewrite a thread wholesale, for restating a message's delivery state. */ + fun replaceHistory(threadId: String, messages: JSONArray) = + writeJson(historyName(threadId), JSONObject().put("messages", messages)) + /** Every thread this device holds, group or person. */ fun threads(): List = storage.names().filter { it.startsWith(HISTORY_PREFIX) }