diff --git a/app/src/main/java/app/closer/MainActivity.kt b/app/src/main/java/app/closer/MainActivity.kt index b7525ad6..3f91d327 100644 --- a/app/src/main/java/app/closer/MainActivity.kt +++ b/app/src/main/java/app/closer/MainActivity.kt @@ -43,6 +43,7 @@ import app.closer.core.notifications.TokenRegistrar import app.closer.domain.model.AuthState import app.closer.notifications.PartnerNotificationPayload import app.closer.notifications.PartnerNotificationType +import kotlinx.coroutines.flow.collectLatest import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.flow.filterIsInstance import kotlinx.coroutines.flow.first @@ -292,11 +293,11 @@ class MainActivity : AppCompatActivity() { private fun adoptCoupleKeyRotation() { lifecycleScope.launch { authRepository.authState - .filterIsInstance() - .map { it.userId } + .map { (it as? AuthState.Authenticated)?.userId } .distinctUntilChanged() - .collect { uid -> - runCatching { coupleKeyRotationAdopter.adoptIfNeeded(uid) } + .collectLatest { uid -> + // collectLatest: sign-out (null) or a different uid cancels the previous watch. + if (uid != null) runCatching { coupleKeyRotationAdopter.watch(uid) } } } } diff --git a/app/src/main/java/app/closer/data/remote/FirestoreCoupleDataSource.kt b/app/src/main/java/app/closer/data/remote/FirestoreCoupleDataSource.kt index aca67e0b..a88b7bad 100644 --- a/app/src/main/java/app/closer/data/remote/FirestoreCoupleDataSource.kt +++ b/app/src/main/java/app/closer/data/remote/FirestoreCoupleDataSource.kt @@ -8,6 +8,9 @@ import com.google.firebase.firestore.DocumentSnapshot import com.google.firebase.firestore.FirebaseFirestore import com.google.firebase.firestore.SetOptions import com.google.firebase.functions.FirebaseFunctions +import kotlinx.coroutines.channels.awaitClose +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.callbackFlow import kotlinx.coroutines.tasks.await import javax.inject.Inject import javax.inject.Singleton @@ -112,24 +115,49 @@ class FirestoreCoupleDataSource @Inject constructor( return if (snap.exists()) snap.toCouple() else null } + /** + * The couple doc as a live stream. Snapshot listeners receive the partner's writes from the + * server within seconds, which makes this the right feed for reacting to `keyGeneration` bumps — + * unlike the one-shot reads, there is no cache-staleness trap to fall into (C-ROTATE-001). + */ + fun observeCouple(coupleId: String): Flow = callbackFlow { + val listener = coupleRef(coupleId).addSnapshotListener { snap, error -> + if (error != null) { + close(error) + return@addSnapshotListener + } + trySend(snap?.takeIf { it.exists() }?.toCouple()) + } + awaitClose { listener.remove() } + } + suspend fun updateStreak(coupleId: String) { - val snap = coupleRef(coupleId).get().await() - val lastAnsweredAt = snap.millisOrNull("lastAnsweredAt") ?: 0L - val now = System.currentTimeMillis() - val current = (snap.getLong("streakCount") ?: 0L).toInt() - // Gentle streak: once per calendar day, one forgiving grace day (see StreakCalculator). - // Replaces the old raw-48h rule that inflated the count on every answer. - val next = app.closer.domain.StreakCalculator.nextCount( - lastAnsweredAtMillis = lastAnsweredAt.takeIf { it > 0L }, - nowMillis = now, - currentStreak = current - ) - // Same-day answer → nothing to change; skip the redundant write. - if (!next.changed) return - coupleRef(coupleId).set( - mapOf("streakCount" to next.newStreak, "lastAnsweredAt" to now), - SetOptions.merge() - ).await() + // Read-modify-write on a counter, so it runs as a TRANSACTION: the old cache-first + // get()-compute-set() could derive `next` from a stale snapshot and double-count or clobber + // the streak when both partners answered near-simultaneously — the same stale-read-feeding-a- + // write class as C-ROTATE-001. A transaction reads current values from the server and retries + // on contention, so concurrent answers converge instead of racing. + db.runTransaction { txn -> + val snap = txn.get(coupleRef(coupleId)) + val lastAnsweredAt = snap.millisOrNull("lastAnsweredAt") ?: 0L + val now = System.currentTimeMillis() + val current = (snap.getLong("streakCount") ?: 0L).toInt() + // Gentle streak: once per calendar day, one forgiving grace day (see StreakCalculator). + // Replaces the old raw-48h rule that inflated the count on every answer. + val next = app.closer.domain.StreakCalculator.nextCount( + lastAnsweredAtMillis = lastAnsweredAt.takeIf { it > 0L }, + nowMillis = now, + currentStreak = current + ) + // Same-day answer → nothing to change; skip the redundant write. + if (next.changed) { + txn.set( + coupleRef(coupleId), + mapOf("streakCount" to next.newStreak, "lastAnsweredAt" to now), + SetOptions.merge() + ) + } + }.await() } suspend fun leaveCouple() { diff --git a/app/src/main/java/app/closer/data/repository/CoupleRepositoryImpl.kt b/app/src/main/java/app/closer/data/repository/CoupleRepositoryImpl.kt index ee20cdc4..d5a6d6ce 100644 --- a/app/src/main/java/app/closer/data/repository/CoupleRepositoryImpl.kt +++ b/app/src/main/java/app/closer/data/repository/CoupleRepositoryImpl.kt @@ -1,6 +1,10 @@ package app.closer.data.repository import app.closer.core.crash.CrashReporter +import kotlinx.coroutines.flow.distinctUntilChanged +import kotlinx.coroutines.flow.flatMapLatest +import kotlinx.coroutines.flow.flowOf +import kotlinx.coroutines.flow.map import app.closer.crypto.CoupleEncryptionManager import app.closer.data.remote.FirestoreCoupleDataSource import app.closer.data.remote.FirestoreUserDataSource @@ -37,6 +41,15 @@ class CoupleRepositoryImpl @Inject constructor( error("Direct couple creation from the client is no longer supported; use InviteRepository.acceptInvite.") } + @kotlinx.coroutines.ExperimentalCoroutinesApi + override fun observeCoupleForUser(userId: String): kotlinx.coroutines.flow.Flow = + userDataSource.observeUser(userId) + .map { it?.coupleId } + .distinctUntilChanged() + .flatMapLatest { coupleId -> + if (coupleId.isNullOrBlank()) flowOf(null) else coupleDataSource.observeCouple(coupleId) + } + override suspend fun updateStreak(coupleId: String): Result = runCatching { coupleDataSource.updateStreak(coupleId) } diff --git a/app/src/main/java/app/closer/domain/repository/CoupleRepository.kt b/app/src/main/java/app/closer/domain/repository/CoupleRepository.kt index cd59a0cf..c987689a 100644 --- a/app/src/main/java/app/closer/domain/repository/CoupleRepository.kt +++ b/app/src/main/java/app/closer/domain/repository/CoupleRepository.kt @@ -4,6 +4,12 @@ import app.closer.domain.model.Couple interface CoupleRepository { suspend fun getCoupleForUser(userId: String): Couple? + + /** + * The user's couple as a live stream (null while unpaired). Server-fed via snapshot listener — + * the safe feed for decisions that react to couple-doc changes (e.g. keyGeneration adoption). + */ + fun observeCoupleForUser(userId: String): kotlinx.coroutines.flow.Flow suspend fun createCouple(inviterUserId: String, acceptorUserId: String, inviteCode: String, recoveryPhrase: String): Result suspend fun updateStreak(coupleId: String): Result suspend fun leaveCouple(userId: String): Result diff --git a/app/src/main/java/app/closer/domain/usecase/CoupleKeyRotationAdopter.kt b/app/src/main/java/app/closer/domain/usecase/CoupleKeyRotationAdopter.kt index 19fa3616..8d73d790 100644 --- a/app/src/main/java/app/closer/domain/usecase/CoupleKeyRotationAdopter.kt +++ b/app/src/main/java/app/closer/domain/usecase/CoupleKeyRotationAdopter.kt @@ -3,6 +3,7 @@ package app.closer.domain.usecase import app.closer.core.crash.CrashReporter import app.closer.crypto.CoupleEncryptionManager import app.closer.domain.repository.CoupleRepository +import kotlinx.coroutines.flow.catch import javax.inject.Inject import javax.inject.Singleton @@ -26,6 +27,23 @@ class CoupleKeyRotationAdopter @Inject constructor( private val encryptionManager: CoupleEncryptionManager, private val crashReporter: CrashReporter ) { + /** + * Long-lived entry point: watches the couple doc and adopts on every emission. This is what + * covers the WARM app — the partner's rotation lands via the snapshot listener within seconds, + * whereas the one-shot hooks (process start, Home load) only help after the fact. Caught in + * review: with one-shot hooks alone, a foregrounded app never adopted at all; the 🔑 push routes + * to Security, which doesn't adopt, and a process can stay warm for days showing 🔒. + * + * Adoption is idempotent (generation compare), so repeated emissions are cheap no-ops. A stream + * failure is recorded and ends the watch quietly — the process degrades to the one-shot hooks, + * never crashes; the next launch re-establishes it. + */ + suspend fun watch(uid: String) { + coupleRepository.observeCoupleForUser(uid) + .catch { crashReporter.recordException(it) } + .collect { couple -> couple?.let { adoptForCouple(it) } } + } + /** Startup entry point: resolves the couple itself. Returns the outcome for callers that care. */ suspend fun adoptIfNeeded(uid: String): CoupleEncryptionManager.AdoptionResult { val couple = runCatching { coupleRepository.getCoupleForUser(uid) } diff --git a/app/src/test/java/app/closer/domain/usecase/CoupleKeyRotationAdopterTest.kt b/app/src/test/java/app/closer/domain/usecase/CoupleKeyRotationAdopterTest.kt new file mode 100644 index 00000000..0ad1acf7 --- /dev/null +++ b/app/src/test/java/app/closer/domain/usecase/CoupleKeyRotationAdopterTest.kt @@ -0,0 +1,97 @@ +package app.closer.domain.usecase + +import app.closer.core.crash.CrashReporter +import app.closer.crypto.CoupleEncryptionManager +import app.closer.crypto.CoupleEncryptionManager.AdoptionResult +import app.closer.domain.model.Couple +import app.closer.domain.repository.CoupleRepository +import io.mockk.coEvery +import io.mockk.coVerify +import io.mockk.every +import io.mockk.mockk +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.flow.flowOf +import kotlinx.coroutines.test.runTest +import org.junit.Assert.assertEquals +import org.junit.Test + +/** + * The watcher is what covers the WARM app: with only the one-shot hooks (process start, Home load), + * a foregrounded app never adopted a partner's rotation at all — the 🔑 push routes to Security, + * which doesn't adopt, and a process can stay warm for days rendering 🔒 for every new-key message. + * + * Its contract: adopt on every couple emission, skip nulls, keep collecting past a failed adoption, + * and never let a stream error escape (a dead watcher degrades to the one-shot hooks — it must not + * take MainActivity's collector down with it). + */ +class CoupleKeyRotationAdopterTest { + + private val coupleRepository: CoupleRepository = mockk() + private val encryptionManager: CoupleEncryptionManager = mockk() + private val crashReporter: CrashReporter = mockk(relaxUnitFun = true) + + private val adopter = CoupleKeyRotationAdopter(coupleRepository, encryptionManager, crashReporter) + + private val gen1 = Couple(id = "c1", keyGeneration = 1L) + private val gen2 = Couple(id = "c1", keyGeneration = 2L) + + @Test + fun `every emission is offered for adoption — the warm-app rotation lands`() = runTest { + every { coupleRepository.observeCoupleForUser("u1") } returns flowOf(gen1, gen2) + coEvery { encryptionManager.adoptRotationIfNeeded(gen1) } returns AdoptionResult.UpToDate + coEvery { encryptionManager.adoptRotationIfNeeded(gen2) } returns AdoptionResult.Adopted + + adopter.watch("u1") + + coVerify(exactly = 1) { encryptionManager.adoptRotationIfNeeded(gen1) } + coVerify(exactly = 1) { encryptionManager.adoptRotationIfNeeded(gen2) } + } + + @Test + fun `unpaired (null) emissions are skipped`() = runTest { + every { coupleRepository.observeCoupleForUser("u1") } returns flowOf(null, gen1) + coEvery { encryptionManager.adoptRotationIfNeeded(gen1) } returns AdoptionResult.UpToDate + + adopter.watch("u1") + + coVerify(exactly = 1) { encryptionManager.adoptRotationIfNeeded(any()) } + } + + @Test + fun `a failed adoption is recorded and the watch keeps collecting`() = runTest { + val boom = RuntimeException("unwrap failed") + every { coupleRepository.observeCoupleForUser("u1") } returns flowOf(gen1, gen2) + coEvery { encryptionManager.adoptRotationIfNeeded(gen1) } returns AdoptionResult.Failed(boom) + coEvery { encryptionManager.adoptRotationIfNeeded(gen2) } returns AdoptionResult.Adopted + + adopter.watch("u1") + + coVerify { crashReporter.recordException(boom) } + // The failure on gen1 must not stop gen2 from being adopted. + coVerify(exactly = 1) { encryptionManager.adoptRotationIfNeeded(gen2) } + } + + @Test + fun `a stream error is recorded, not thrown — the collector upstairs survives`() = runTest { + val listenerError = RuntimeException("PERMISSION_DENIED after sign-out race") + every { coupleRepository.observeCoupleForUser("u1") } returns flow { + emit(gen1) + throw listenerError + } + coEvery { encryptionManager.adoptRotationIfNeeded(gen1) } returns AdoptionResult.UpToDate + + adopter.watch("u1") // must return normally + + coVerify { crashReporter.recordException(listenerError) } + } + + @Test + fun `one-shot entry point still resolves the couple and adopts`() = runTest { + coEvery { coupleRepository.getCoupleForUser("u1") } returns gen2 + coEvery { encryptionManager.adoptRotationIfNeeded(gen2) } returns AdoptionResult.Adopted + + val result = adopter.adoptIfNeeded("u1") + + assertEquals(AdoptionResult.Adopted, result) + } +}