From fa79f351f8da6c337bcfd319bb686cb2df8bab81 Mon Sep 17 00:00:00 2001 From: Eddie Date: Mon, 24 Aug 2026 09:12:48 -0400 Subject: [PATCH] fix: refresh Android course structure when requested --- .../java/org/openedx/app/di/ScreenModule.kt | 2 +- .../course/data/repository/CoalescingCache.kt | 96 ++- .../data/repository/CourseRepository.kt | 303 ++++++- .../domain/interactor/CourseInteractor.kt | 14 +- .../data/repository/CoalescingCacheTest.kt | 117 +++ .../repository/CourseRepositoryFreshTest.kt | 775 ++++++++++++++++++ .../interactor/CourseInteractorFreshTest.kt | 102 +++ 7 files changed, 1367 insertions(+), 42 deletions(-) create mode 100644 course/src/test/java/org/openedx/course/data/repository/CoalescingCacheTest.kt create mode 100644 course/src/test/java/org/openedx/course/data/repository/CourseRepositoryFreshTest.kt create mode 100644 course/src/test/java/org/openedx/course/domain/interactor/CourseInteractorFreshTest.kt diff --git a/app/src/main/java/org/openedx/app/di/ScreenModule.kt b/app/src/main/java/org/openedx/app/di/ScreenModule.kt index 1799dafc6..a2b06ac00 100644 --- a/app/src/main/java/org/openedx/app/di/ScreenModule.kt +++ b/app/src/main/java/org/openedx/app/di/ScreenModule.kt @@ -270,7 +270,7 @@ val screenModule = module { factory { CalendarInteractor(get()) } single { CourseRepository(get(), get(), get(), get(), get()) } - factory { CourseInteractor(get()) } + factory { CourseInteractor(get(), get()) } single { get() } viewModel { (pathId: String, infoType: String) -> diff --git a/course/src/main/java/org/openedx/course/data/repository/CoalescingCache.kt b/course/src/main/java/org/openedx/course/data/repository/CoalescingCache.kt index 7597e9b50..1dad9b957 100644 --- a/course/src/main/java/org/openedx/course/data/repository/CoalescingCache.kt +++ b/course/src/main/java/org/openedx/course/data/repository/CoalescingCache.kt @@ -13,24 +13,73 @@ import java.util.concurrent.ConcurrentHashMap * @param V the type of cached values * @param fetch the suspend function to fetch data for a given key * @param persist optional callback invoked after successful fetch (e.g., to save to database) + * @param autoCache whether a successful fetch is published automatically + * @param activeGeneration optional provider that invalidates entries from an earlier generation */ class CoalescingCache( private val fetch: suspend (K) -> V, - private val persist: (suspend (K, V) -> Unit)? = null + private val persist: (suspend (K, V) -> Unit)? = null, + private val autoCache: Boolean = true, + private val activeGeneration: (() -> Long)? = null, ) { - private val cache = ConcurrentHashMap() + private data class CacheEntry( + val value: V, + val generation: Long, + ) + + private val cache = ConcurrentHashMap>() private val pending = ConcurrentHashMap>() /** * Returns cached value for the key, or null if not cached. */ - fun getCached(key: K): V? = cache[key] + fun getCached(key: K): V? { + val entry = cache[key] ?: return null + val generationProvider = activeGeneration + + return when { + generationProvider == null -> { + entry.value + } + + entry.generation == generationProvider() -> { + entry.value + } + + else -> { + cache.remove(key, entry) + null + } + } + } /** * Manually sets a cached value. */ - fun setCached(key: K, value: V) { - cache[key] = value + fun setCached( + key: K, + value: V, + writeGeneration: Long = UNSPECIFIED_GENERATION, + ) { + val generationProvider = activeGeneration + if (generationProvider == null) { + cache[key] = CacheEntry(value, DEFAULT_GENERATION) + return + } + + if (writeGeneration == UNSPECIFIED_GENERATION) { + val currentGeneration = generationProvider() + cache[key] = CacheEntry(value, currentGeneration) + return + } + + cache.compute(key) { _, currentEntry -> + if (generationProvider() == writeGeneration) { + CacheEntry(value, writeGeneration) + } else { + currentEntry + } + } } /** @@ -40,6 +89,18 @@ class CoalescingCache( cache.clear() } + /** + * Cancels pending requests without allowing an earlier fetch to remove a later request. + */ + fun cancelPending() { + val pendingSnapshot = HashMap(pending) + for ((key, deferred) in pendingSnapshot) { + if (pending.remove(key, deferred)) { + deferred.cancel() + } + } + } + /** * Gets the value from cache or fetches it. * @@ -49,22 +110,24 @@ class CoalescingCache( */ suspend fun getOrFetch(key: K, forceRefresh: Boolean = false): V { if (!forceRefresh) { - cache[key]?.let { return it } + getCached(key)?.let { return it } } - val (deferred, isOwner) = getOrCreateDeferred(key) - return if (isOwner) { + val (deferred, startsFetch) = getOrCreateDeferred(key) + return if (startsFetch) { try { - val result = fetch(key) - cache[key] = result - persist?.invoke(key, result) - deferred.complete(result) - result + val value = fetch(key) + if (autoCache) { + setCached(key, value) + } + persist?.invoke(key, value) + deferred.complete(value) + value } catch (e: Exception) { deferred.completeExceptionally(e) throw e } finally { - pending.remove(key) + pending.remove(key, deferred) } } else { deferred.await() @@ -77,4 +140,9 @@ class CoalescingCache( val existing = pending.putIfAbsent(key, deferred) return if (existing != null) existing to false else deferred to true } + + private companion object { + const val DEFAULT_GENERATION = 0L + const val UNSPECIFIED_GENERATION = Long.MIN_VALUE + } } diff --git a/course/src/main/java/org/openedx/course/data/repository/CourseRepository.kt b/course/src/main/java/org/openedx/course/data/repository/CourseRepository.kt index 229963b58..9096226e4 100644 --- a/course/src/main/java/org/openedx/course/data/repository/CourseRepository.kt +++ b/course/src/main/java/org/openedx/course/data/repository/CourseRepository.kt @@ -3,10 +3,13 @@ package org.openedx.course.data.repository import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.flow import kotlinx.coroutines.flow.map +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock import okhttp3.MultipartBody import org.openedx.core.ApiConstants import org.openedx.core.data.api.CourseApi import org.openedx.core.data.model.BlocksCompletionBody +import org.openedx.core.data.model.room.CourseStructureEntity import org.openedx.core.data.model.room.OfflineXBlockProgress import org.openedx.core.data.model.room.VideoProgressEntity import org.openedx.core.data.model.room.XBlockProgressData @@ -24,6 +27,7 @@ import org.openedx.core.system.connection.NetworkConnection import java.net.URLDecoder import java.nio.charset.StandardCharsets import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.AtomicLong /** * Repository for course data with request coalescing. @@ -39,21 +43,142 @@ class CourseRepository( private val preferencesManager: CorePreferences, private val networkConnection: NetworkConnection, ) { - // Session tracking - when entering a course, mark that data needs refresh - private val needsRefresh = ConcurrentHashMap.newKeySet() + /** + * Identifies state owned by one course session generation. Old completions must address their + * original key so they cannot alter a new session's refresh marker or completion version. + */ + private data class CourseSessionKey( + val sessionGeneration: Long, + val courseId: String, + ) - private val structureCache = CoalescingCache( - fetch = { courseId -> + /** + * Carries a cache-first response through persistence. When a fresh result wins while this + * response is in flight, [returnedCourseStructure] is replaced before the coalesced request + * completes so Flow callers cannot emit the stale fetched value after the fresh commit. + */ + private data class NonFreshFetchResult( + val capturedSessionGeneration: Long, + val capturedFreshCompletionVersion: Long, + val roomEntity: CourseStructureEntity?, + val fetchedCourseStructure: CourseStructure, + ) { + var returnedCourseStructure: CourseStructure = fetchedCourseStructure + } + + /** + * Pairs the domain object and Room entity from one fresh response. The pair stays together + * until it is saved, so a later fetch cannot combine the earlier domain object with a newer + * Room entity. + */ + private data class FreshFetchResult( + val capturedSessionGeneration: Long, + val courseStructure: CourseStructure, + val roomEntity: CourseStructureEntity, + ) + + /** + * Cache entries are valid only for the current session generation. A mutex per course + * serializes non-fresh and fresh writes, plus publishing values read from Room. Keep each + * mutex after reset because an old DAO insert may already be running. A new-session write + * must follow that insert so its value is final. + */ + private val sessionGeneration = AtomicLong(0) + private val courseWriteMutexes = ConcurrentHashMap() + private val freshCompletionVersion = ConcurrentHashMap() + + /** + * Keys include the session generation so an old completion cannot remove a new session's marker. + */ + private val needsRefresh = ConcurrentHashMap.newKeySet() + + private val structureCache: CoalescingCache = CoalescingCache( + fetch = { courseSessionKey -> + val capturedGeneration = courseSessionKey.sessionGeneration + val courseId = courseSessionKey.courseId + val capturedFreshCompletionVersion = + freshCompletionVersionFor(capturedGeneration, courseId).get() val response = api.getCourseStructure( "stale-if-error=0", "v4", preferencesManager.user?.username, courseId ) - courseDao.insertCourseStructureEntity(response.mapToRoomEntity()) - response.mapToDomain() + NonFreshFetchResult( + capturedSessionGeneration = capturedGeneration, + capturedFreshCompletionVersion = capturedFreshCompletionVersion, + roomEntity = response.mapToRoomEntity(), + fetchedCourseStructure = response.mapToDomain(), + ) + }, + persist = { courseSessionKey, fetchResult -> + val courseId = courseSessionKey.courseId + + runUnderCourseWriteGuard(courseId, fetchResult.capturedSessionGeneration) { + val currentFreshCompletionVersion = freshCompletionVersionFor( + fetchResult.capturedSessionGeneration, + courseId, + ).get() + if (currentFreshCompletionVersion != + fetchResult.capturedFreshCompletionVersion + ) { + getCachedStructure(courseId)?.let { + fetchResult.returnedCourseStructure = it + } + return@runUnderCourseWriteGuard + } + + fetchResult.roomEntity?.let { courseDao.insertCourseStructureEntity(it) } + setCachedStructure( + courseId, + fetchResult.fetchedCourseStructure, + fetchResult.capturedSessionGeneration, + ) + } + needsRefresh.remove(courseSessionKey) + }, + // Cache writes are explicit so a stale fetch cannot replace a fresh or new-session value. + autoCache = false, + activeGeneration = { sessionGeneration.get() }, + ) + + /** + * Fresh requests use a separate coalescing instance so they always send `no-cache` and never + * join a non-fresh request that uses `stale-if-error=0`. + */ + private val freshStructureCache = CoalescingCache( + fetch = { courseSessionKey -> + val courseId = courseSessionKey.courseId + val response = api.getCourseStructure( + "no-cache", + "v4", + preferencesManager.user?.username, + courseId, + ) + FreshFetchResult( + capturedSessionGeneration = courseSessionKey.sessionGeneration, + courseStructure = response.mapToDomain(), + roomEntity = response.mapToRoomEntity(), + ) }, - persist = { courseId, _ -> needsRefresh.remove(courseId) } + persist = { courseSessionKey, freshResult -> + val courseId = courseSessionKey.courseId + runUnderCourseWriteGuard(courseId, courseSessionKey.sessionGeneration) { + courseDao.insertCourseStructureEntity(freshResult.roomEntity) + freshCompletionVersionFor( + courseSessionKey.sessionGeneration, + courseId, + ).incrementAndGet() + setCachedStructure( + courseId, + freshResult.courseStructure, + courseSessionKey.sessionGeneration, + ) + } + needsRefresh.remove(courseSessionKey) + }, + autoCache = false, + activeGeneration = { sessionGeneration.get() }, ) private val statusCache = CoalescingCache( @@ -84,48 +209,97 @@ class CourseRepository( * Call when entering a course to mark that data should be refreshed. */ fun startCourseSession(courseId: String) { - needsRefresh.add(courseId) + val courseSessionKey = CourseSessionKey(sessionGeneration.get(), courseId) + needsRefresh.add(courseSessionKey) } fun endCourseSession() { + sessionGeneration.incrementAndGet() + structureCache.cancelPending() + freshStructureCache.cancelPending() structureCache.clear() + freshStructureCache.clear() statusCache.clear() datesCache.clear() progressCache.clear() enrollmentCache.clear() needsRefresh.clear() + freshCompletionVersion.clear() } fun getCourseStructureFlow( courseId: String, forceRefresh: Boolean = false ): Flow = flow { - // Always emit cached data first if available - structureCache.getCached(courseId)?.let { emit(it) } + val flowSessionGeneration = sessionGeneration.get() - if (structureCache.getCached(courseId) == null) { - courseDao.getCourseStructureById(courseId)?.mapToDomain()?.let { - structureCache.setCached(courseId, it) - emit(it) + // Always emit cached data first if available + getCachedStructure(courseId)?.let { emit(it) } + + if (getCachedStructure(courseId) == null) { + val capturedFreshCompletionVersion = + freshCompletionVersionFor(flowSessionGeneration, courseId).get() + val roomStructure = courseDao.getCourseStructureById(courseId)?.mapToDomain() + if (roomStructure != null) { + val structureToEmit = runUnderCourseWriteGuard( + courseId, + flowSessionGeneration, + ) { + resolveStructureFromRoom( + courseId = courseId, + capturedGeneration = flowSessionGeneration, + capturedFreshCompletionVersion = capturedFreshCompletionVersion, + roomStructure = roomStructure, + ) + } + structureToEmit?.let { emit(it) } } } - val shouldRefresh = forceRefresh || needsRefresh.contains(courseId) - if (networkConnection.isOnline() && (structureCache.getCached(courseId) == null || shouldRefresh)) { - emit(structureCache.getOrFetch(courseId, forceRefresh = true)) + val courseSessionKey = CourseSessionKey(flowSessionGeneration, courseId) + val shouldRefresh = forceRefresh || needsRefresh.contains(courseSessionKey) + val hasCachedStructure = getCachedStructure(courseId) != null + val shouldFetch = networkConnection.isOnline() && (!hasCachedStructure || shouldRefresh) + if (shouldFetch) { + val fetchResult = structureCache.getOrFetch(courseSessionKey, forceRefresh = true) + emit(fetchResult.returnedCourseStructure) } - if (structureCache.getCached(courseId) == null) { + if (getCachedStructure(courseId) == null) { throw NoCachedDataException() } } + suspend fun getCourseStructureFresh(courseId: String): CourseStructure { + val courseSessionKey = CourseSessionKey(sessionGeneration.get(), courseId) + return freshStructureCache.getOrFetch(courseSessionKey, forceRefresh = true).courseStructure + } + suspend fun getCourseStructureFromCache(courseId: String): CourseStructure { - return structureCache.getCached(courseId) - ?: courseDao.getCourseStructureById(courseId)?.mapToDomain()?.also { - structureCache.setCached(courseId, it) - } + val cachedStructure = getCachedStructure(courseId) + if (cachedStructure != null) { + return cachedStructure + } + + val capturedGeneration = sessionGeneration.get() + val capturedFreshCompletionVersion = + freshCompletionVersionFor(capturedGeneration, courseId).get() + val roomEntity = courseDao.getCourseStructureById(courseId) ?: throw NoCachedDataException() + val roomStructure = roomEntity.mapToDomain() + + val resolvedStructure = runUnderCourseWriteGuard(courseId, capturedGeneration) { + resolveStructureFromRoom( + courseId = courseId, + capturedGeneration = capturedGeneration, + capturedFreshCompletionVersion = capturedFreshCompletionVersion, + roomStructure = roomStructure, + ) + } + if (resolvedStructure == null) { + throw NoCachedDataException() + } + return resolvedStructure } fun getEnrollmentDetailsFlow( @@ -163,7 +337,9 @@ class CourseRepository( val cached = statusCache.getCached(courseId) emit(cached ?: CourseComponentStatus("")) - val shouldRefresh = forceRefresh || needsRefresh.contains(courseId) + val capturedGeneration = sessionGeneration.get() + val courseSessionKey = CourseSessionKey(capturedGeneration, courseId) + val shouldRefresh = forceRefresh || needsRefresh.contains(courseSessionKey) if (networkConnection.isOnline() && (cached == null || shouldRefresh)) { emit(statusCache.getOrFetch(courseId, forceRefresh = true)) } @@ -184,7 +360,9 @@ class CourseRepository( val cached = datesCache.getCached(courseId) emit(cached ?: emptyCourseDatesResult()) - val shouldRefresh = forceRefresh || needsRefresh.contains(courseId) + val capturedGeneration = sessionGeneration.get() + val courseSessionKey = CourseSessionKey(capturedGeneration, courseId) + val shouldRefresh = forceRefresh || needsRefresh.contains(courseSessionKey) if (networkConnection.isOnline() && (cached == null || shouldRefresh)) { emit(datesCache.getOrFetch(courseId, forceRefresh = true)) } @@ -225,7 +403,9 @@ class CourseRepository( } } - val shouldRefresh = isRefresh || needsRefresh.contains(courseId) + val capturedGeneration = sessionGeneration.get() + val courseSessionKey = CourseSessionKey(capturedGeneration, courseId) + val shouldRefresh = isRefresh || needsRefresh.contains(courseSessionKey) val hasCache = progressCache.getCached(courseId) != null val shouldFetch = shouldRefresh || !hasCache || !getOnlyCacheIfExist @@ -288,6 +468,79 @@ class CourseRepository( submitOfflineXBlockProgress(blockId, courseId, jsonProgressData) } + private fun freshCompletionVersionFor( + generation: Long, + courseId: String, + ): AtomicLong { + val courseSessionKey = CourseSessionKey(generation, courseId) + return freshCompletionVersion.getOrPut(courseSessionKey) { AtomicLong(0) } + } + + private fun getCachedStructure( + courseId: String, + ): CourseStructure? { + return structureCache.getCached( + CourseSessionKey(sessionGeneration.get(), courseId), + )?.returnedCourseStructure + } + + private fun setCachedStructure( + courseId: String, + courseStructure: CourseStructure, + generation: Long, + ) { + structureCache.setCached( + CourseSessionKey(generation, courseId), + NonFreshFetchResult( + capturedSessionGeneration = generation, + capturedFreshCompletionVersion = freshCompletionVersionFor(generation, courseId).get(), + roomEntity = null, + fetchedCourseStructure = courseStructure, + ), + generation, + ) + } + + private suspend fun runUnderCourseWriteGuard( + courseId: String, + capturedGeneration: Long, + block: suspend () -> R, + ): R? { + val courseWriteMutex = courseWriteMutexes.getOrPut(courseId) { Mutex() } + return courseWriteMutex.withLock { + if (sessionGeneration.get() != capturedGeneration) { + return@withLock null + } + block() + } + } + + private fun resolveStructureFromRoom( + courseId: String, + capturedGeneration: Long, + capturedFreshCompletionVersion: Long, + roomStructure: CourseStructure, + ): CourseStructure? { + val currentCachedStructure = getCachedStructure(courseId) + val currentFreshCompletionVersion = + freshCompletionVersionFor(capturedGeneration, courseId).get() + + return when { + currentFreshCompletionVersion != capturedFreshCompletionVersion -> { + currentCachedStructure + } + + currentCachedStructure != null -> { + currentCachedStructure + } + + else -> { + setCachedStructure(courseId, roomStructure, capturedGeneration) + roomStructure + } + } + } + private suspend fun submitOfflineXBlockProgress( blockId: String, courseId: String, diff --git a/course/src/main/java/org/openedx/course/domain/interactor/CourseInteractor.kt b/course/src/main/java/org/openedx/course/domain/interactor/CourseInteractor.kt index 543695689..115f02cde 100644 --- a/course/src/main/java/org/openedx/course/domain/interactor/CourseInteractor.kt +++ b/course/src/main/java/org/openedx/course/domain/interactor/CourseInteractor.kt @@ -7,11 +7,13 @@ import org.openedx.core.domain.interactor.CourseInteractor import org.openedx.core.domain.model.Block import org.openedx.core.domain.model.CourseEnrollmentDetails import org.openedx.core.domain.model.CourseStructure +import org.openedx.core.system.connection.NetworkConnection import org.openedx.course.data.repository.CourseRepository @Suppress("TooManyFunctions") class CourseInteractor( - private val repository: CourseRepository + private val repository: CourseRepository, + private val networkConnection: NetworkConnection, ) : CourseInteractor { fun startCourseSession(courseId: String) { @@ -33,7 +35,15 @@ class CourseInteractor( courseId: String, isNeedRefresh: Boolean ): CourseStructure { - return repository.getCourseStructureFlow(courseId, isNeedRefresh).first() + return when { + isNeedRefresh && networkConnection.isOnline() -> { + repository.getCourseStructureFresh(courseId) + } + + else -> { + repository.getCourseStructureFlow(courseId, isNeedRefresh).first() + } + } } override suspend fun getCourseStructureFromCache(courseId: String): CourseStructure { diff --git a/course/src/test/java/org/openedx/course/data/repository/CoalescingCacheTest.kt b/course/src/test/java/org/openedx/course/data/repository/CoalescingCacheTest.kt new file mode 100644 index 000000000..1c5b4f77b --- /dev/null +++ b/course/src/test/java/org/openedx/course/data/repository/CoalescingCacheTest.kt @@ -0,0 +1,117 @@ +package org.openedx.course.data.repository + +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.CoroutineStart +import kotlinx.coroutines.Deferred +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.async +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.test.runTest +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Test + +@OptIn(ExperimentalCoroutinesApi::class) +class CoalescingCacheTest { + + @Test + fun `old generation write does not replace the current cache entry`() { + var activeGeneration = 1L + val cache = CoalescingCache( + fetch = { "unused" }, + activeGeneration = { activeGeneration }, + ) + + cache.setCached(KEY, "generation-1", writeGeneration = 1L) + activeGeneration = 2L + assertNull(cache.getCached(KEY)) + + cache.setCached(KEY, "generation-2", writeGeneration = 2L) + cache.setCached(KEY, "late-generation-1", writeGeneration = 1L) + assertEquals("generation-2", cache.getCached(KEY)) + } + + @Test + fun `cancelled fetch cannot remove a newer pending request`() = runTest { + val fetchGates = Channel>(Channel.UNLIMITED) + var fetchCount = 0 + val cache = CoalescingCache( + fetch = { + fetchCount += 1 + val gate = CompletableDeferred() + fetchGates.send(gate) + gate.await() + }, + ) + + val oldFetch = async(start = CoroutineStart.UNDISPATCHED) { + cache.getOrFetch(KEY, forceRefresh = true) + } + val oldGate = fetchGates.receive() + cache.cancelPending() + + val newFetch = async(start = CoroutineStart.UNDISPATCHED) { + cache.getOrFetch(KEY, forceRefresh = true) + } + val newGate = fetchGates.receive() + + oldGate.complete("old") + assertEquals("old", oldFetch.await()) + + val newWaiter = async(start = CoroutineStart.UNDISPATCHED) { + cache.getOrFetch(KEY, forceRefresh = true) + } + assertFalse(newWaiter.isCompleted) + assertTrue(fetchGates.tryReceive().isFailure) + + newGate.complete("new") + assertEquals("new", newFetch.await()) + assertEquals("new", newWaiter.await()) + assertEquals(2, fetchCount) + } + + @Test + fun `cancelling pending work removes it before a cancelled waiter starts a new request`() = runTest { + val fetchGates = Channel>(Channel.UNLIMITED) + val cache = CoalescingCache( + fetch = { + val gate = CompletableDeferred() + fetchGates.send(gate) + gate.await() + }, + ) + + val oldFetch = async(start = CoroutineStart.UNDISPATCHED) { + cache.getOrFetch(KEY, forceRefresh = true) + } + val oldGate = fetchGates.receive() + val cancelledWaiter = async(start = CoroutineStart.UNDISPATCHED) { + cache.getOrFetch(KEY, forceRefresh = true) + } + val laterCall = CompletableDeferred>() + cancelledWaiter.invokeOnCompletion { + laterCall.complete( + async(start = CoroutineStart.UNDISPATCHED) { + cache.getOrFetch(KEY, forceRefresh = true) + }, + ) + } + + cache.cancelPending() + + val newFetch = laterCall.await() + assertFalse(newFetch.isCancelled) + val newGate = fetchGates.receive() + + oldGate.complete("old") + assertEquals("old", oldFetch.await()) + newGate.complete("new") + assertEquals("new", newFetch.await()) + } + + private companion object { + const val KEY = "course" + } +} diff --git a/course/src/test/java/org/openedx/course/data/repository/CourseRepositoryFreshTest.kt b/course/src/test/java/org/openedx/course/data/repository/CourseRepositoryFreshTest.kt new file mode 100644 index 000000000..1c0fa3caf --- /dev/null +++ b/course/src/test/java/org/openedx/course/data/repository/CourseRepositoryFreshTest.kt @@ -0,0 +1,775 @@ +package org.openedx.course.data.repository + +import com.google.gson.JsonSyntaxException +import io.mockk.coEvery +import io.mockk.every +import io.mockk.mockk +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.CoroutineStart +import kotlinx.coroutines.Deferred +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.async +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.take +import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.supervisorScope +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.runTest +import okhttp3.MediaType.Companion.toMediaType +import okhttp3.ResponseBody.Companion.toResponseBody +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertSame +import org.junit.Assert.assertTrue +import org.junit.Assert.fail +import org.junit.Before +import org.junit.Test +import org.openedx.core.CoreMocks +import org.openedx.core.data.api.CourseApi +import org.openedx.core.data.model.CourseComponentStatus +import org.openedx.core.data.model.CourseDates +import org.openedx.core.data.model.CourseProgressResponse +import org.openedx.core.data.model.CourseStructureModel +import org.openedx.core.data.model.room.CourseEnrollmentDetailsEntity +import org.openedx.core.data.model.room.CourseProgressEntity +import org.openedx.core.data.model.room.CourseStructureEntity +import org.openedx.core.data.model.room.VideoProgressEntity +import org.openedx.core.data.storage.CorePreferences +import org.openedx.core.data.storage.CourseDao +import org.openedx.core.domain.model.CourseStructure +import org.openedx.core.exception.NoCachedDataException +import org.openedx.core.module.db.DownloadDao +import org.openedx.core.system.connection.NetworkConnection +import retrofit2.HttpException +import retrofit2.Response +import java.net.UnknownHostException +import java.util.Collections +import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.atomic.AtomicReference + +@OptIn(ExperimentalCoroutinesApi::class) +class CourseRepositoryFreshTest { + + private data class CourseStructureFixture( + val domain: CourseStructure, + val roomEntity: CourseStructureEntity, + val response: CourseStructureModel, + ) + + private data class ApiRequest( + val cacheControl: String, + val courseId: String, + val response: CompletableDeferred, + ) + + private lateinit var api: CourseApi + private lateinit var courseDao: GatedCourseDao + private lateinit var preferencesManager: CorePreferences + private lateinit var networkConnection: NetworkConnection + private lateinit var repository: CourseRepository + private lateinit var apiRequests: Channel + + @Before + fun setUp() { + api = mockk() + courseDao = GatedCourseDao() + preferencesManager = mockk() + networkConnection = mockk() + apiRequests = Channel(Channel.UNLIMITED) + + every { preferencesManager.user } returns null + every { networkConnection.isOnline() } returns true + coEvery { + api.getCourseStructure(any(), any(), any(), any()) + } coAnswers { + val response = CompletableDeferred() + val request = ApiRequest( + cacheControl = firstArg(), + courseId = arg(3), + response = response, + ) + apiRequests.send(request) + response.await() + } + + repository = CourseRepository( + api = api, + courseDao = courseDao, + downloadDao = mockk(relaxed = true), + preferencesManager = preferencesManager, + networkConnection = networkConnection, + ) + } + + @Test + fun `fresh fetch returns the server response and updates Room and memory`() = runTest { + val cachedV1 = fixture("v1") + val originV2 = fixture("v2") + courseDao.storedCourseStructure.set(cachedV1.roomEntity) + assertSame(cachedV1.domain, repository.getCourseStructureFromCache(COURSE_ID)) + + val freshCall = startFreshRequest() + val request = apiRequests.receive() + assertEquals(COURSE_ID, request.courseId) + assertEquals("no-cache", request.cacheControl) + request.response.complete(originV2.response) + + assertSame(originV2.domain, freshCall.await()) + assertSame(originV2.roomEntity, courseDao.storedCourseStructure.get()) + assertSame(originV2.domain, repository.getCourseStructureFromCache(COURSE_ID)) + } + + @Test + fun `fresh fetch failures propagate without returning cached data`() = runTest { + val cachedV1 = fixture("v1") + courseDao.storedCourseStructure.set(cachedV1.roomEntity) + assertSame(cachedV1.domain, repository.getCourseStructureFromCache(COURSE_ID)) + + val failures = listOf( + UnknownHostException("offline"), + httpException(500), + httpException(403), + JsonSyntaxException("invalid response"), + ) + + for (failure in failures) { + val actualFailure = supervisorScope { + val freshCall = async(start = CoroutineStart.UNDISPATCHED) { + repository.getCourseStructureFresh(COURSE_ID) + } + val request = apiRequests.receive() + request.response.completeExceptionally(failure) + failureFrom(freshCall) + } + + assertEquals(failure::class, actualFailure::class) + assertEquals(failure.message, actualFailure.message) + assertSame(cachedV1.domain, repository.getCourseStructureFromCache(COURSE_ID)) + } + } + + @Test + fun `cache-first Flow emits the memory cache before the server result`() = runTest { + val cachedV1 = fixture("v1") + val serverV2 = fixture("v2") + courseDao.storedCourseStructure.set(cachedV1.roomEntity) + repository.getCourseStructureFromCache(COURSE_ID) + + val collection = async(start = CoroutineStart.UNDISPATCHED) { + repository.getCourseStructureFlow(COURSE_ID, forceRefresh = true).toList() + } + val request = apiRequests.receive() + assertEquals("stale-if-error=0", request.cacheControl) + request.response.complete(serverV2.response) + + assertEquals(listOf(cachedV1.domain, serverV2.domain), collection.await()) + assertSame(serverV2.roomEntity, courseDao.storedCourseStructure.get()) + } + + @Test + fun `concurrent fresh fetches share one request and persist once`() = runTest { + val originV2 = fixture("v2") + + val firstCall = startFreshRequest() + val secondCall = startFreshRequest() + val request = apiRequests.receive() + + assertTrue(apiRequests.tryReceive().isFailure) + request.response.complete(originV2.response) + + assertSame(originV2.domain, firstCall.await()) + assertSame(originV2.domain, secondCall.await()) + assertEquals(1, courseDao.insertCalls.count { it === originV2.roomEntity }) + } + + @Test + fun `later fresh fetch waits for the first response to persist`() = runTest { + val firstResponse = fixture("first-response") + val insertGate = courseDao.gateInsert(firstResponse.roomEntity) + + val firstCall = startFreshRequest() + apiRequests.receive().response.complete(firstResponse.response) + insertGate.started.await() + + val laterCall = startFreshRequest() + assertTrue(apiRequests.tryReceive().isFailure) + + insertGate.release.complete(Unit) + assertSame(firstResponse.domain, firstCall.await()) + assertSame(firstResponse.domain, laterCall.await()) + assertSame(firstResponse.roomEntity, courseDao.storedCourseStructure.get()) + assertEquals(1, courseDao.insertCalls.count { it === firstResponse.roomEntity }) + } + + @Test + fun `fresh and cache-first fetches send different cache headers`() = runTest { + val nonFreshValue = fixture("non-fresh") + val freshValue = fixture("fresh") + + val nonFreshCall = async(start = CoroutineStart.UNDISPATCHED) { + repository.getCourseStructureFlow(COURSE_ID, forceRefresh = true).first() + } + val nonFreshRequest = apiRequests.receive() + assertEquals("stale-if-error=0", nonFreshRequest.cacheControl) + nonFreshRequest.response.complete(nonFreshValue.response) + assertSame(nonFreshValue.domain, nonFreshCall.await()) + + val freshCall = startFreshRequest() + val freshRequest = apiRequests.receive() + assertEquals("no-cache", freshRequest.cacheControl) + freshRequest.response.complete(freshValue.response) + assertSame(freshValue.domain, freshCall.await()) + } + + @Test + fun `fresh fetch never shares a pending cache-first request`() = runTest { + val nonFreshValue = fixture("non-fresh") + val freshValue = fixture("fresh") + + val nonFreshCall = async(start = CoroutineStart.UNDISPATCHED) { + repository.getCourseStructureFlow(COURSE_ID, forceRefresh = true).first() + } + val nonFreshRequest = apiRequests.receive() + + val freshCall = startFreshRequest() + val freshRequest = apiRequests.receive() + + assertEquals("stale-if-error=0", nonFreshRequest.cacheControl) + assertEquals("no-cache", freshRequest.cacheControl) + nonFreshRequest.response.complete(nonFreshValue.response) + freshRequest.response.complete(freshValue.response) + + assertSame(nonFreshValue.domain, nonFreshCall.await()) + assertSame(freshValue.domain, freshCall.await()) + } + + @Test + fun `stale cache-first fetch returns the fresh result after fresh completion`() = runTest { + val staleV1 = fixture("v1") + val freshV2 = fixture("v2") + + val nonFreshCall = async(start = CoroutineStart.UNDISPATCHED) { + repository.getCourseStructureFlow(COURSE_ID, forceRefresh = true).first() + } + val nonFreshRequest = apiRequests.receive() + val freshCall = startFreshRequest() + val freshRequest = apiRequests.receive() + + freshRequest.response.complete(freshV2.response) + assertSame(freshV2.domain, freshCall.await()) + assertSame(freshV2.domain, repository.getCourseStructureFromCache(COURSE_ID)) + + nonFreshRequest.response.complete(staleV1.response) + assertSame(freshV2.domain, nonFreshCall.await()) + assertSame(freshV2.roomEntity, courseDao.storedCourseStructure.get()) + assertSame(freshV2.domain, repository.getCourseStructureFromCache(COURSE_ID)) + } + + @Test + fun `fresh fetch replaces an earlier cache-first result`() = runTest { + val nonFreshV1 = fixture("v1") + val freshV2 = fixture("v2") + + val nonFreshCall = async(start = CoroutineStart.UNDISPATCHED) { + repository.getCourseStructureFlow(COURSE_ID, forceRefresh = true).first() + } + val nonFreshRequest = apiRequests.receive() + val freshCall = startFreshRequest() + val freshRequest = apiRequests.receive() + + nonFreshRequest.response.complete(nonFreshV1.response) + assertSame(nonFreshV1.domain, nonFreshCall.await()) + assertSame(nonFreshV1.domain, repository.getCourseStructureFromCache(COURSE_ID)) + + freshRequest.response.complete(freshV2.response) + assertSame(freshV2.domain, freshCall.await()) + assertSame(freshV2.roomEntity, courseDao.storedCourseStructure.get()) + assertSame(freshV2.domain, repository.getCourseStructureFromCache(COURSE_ID)) + } + + @Test + fun `delayed Room read does not overwrite a completed fresh fetch`() = runTest { + val roomV1 = fixture("v1") + val freshV2 = fixture("v2") + courseDao.storedCourseStructure.set(roomV1.roomEntity) + val readGate = courseDao.gateNextRead() + + every { networkConnection.isOnline() } returns false + val flowCall = async(start = CoroutineStart.UNDISPATCHED) { + repository.getCourseStructureFlow(COURSE_ID).first() + } + readGate.started.await() + + every { networkConnection.isOnline() } returns true + val freshCall = startFreshRequest() + val freshRequest = apiRequests.receive() + freshRequest.response.complete(freshV2.response) + assertSame(freshV2.domain, freshCall.await()) + + readGate.release.complete(Unit) + assertSame(freshV2.domain, flowCall.await()) + assertSame(freshV2.domain, repository.getCourseStructureFromCache(COURSE_ID)) + } + + @Test + fun `delayed cache-only Room read returns the completed fresh result`() = runTest { + val roomV1 = fixture("v1") + val freshV2 = fixture("v2") + courseDao.storedCourseStructure.set(roomV1.roomEntity) + val readGate = courseDao.gateNextRead() + + val cacheCall = async(start = CoroutineStart.UNDISPATCHED) { + repository.getCourseStructureFromCache(COURSE_ID) + } + readGate.started.await() + + val freshCall = startFreshRequest() + val freshRequest = apiRequests.receive() + freshRequest.response.complete(freshV2.response) + assertSame(freshV2.domain, freshCall.await()) + + readGate.release.complete(Unit) + assertSame(freshV2.domain, cacheCall.await()) + assertSame(freshV2.domain, repository.getCourseStructureFromCache(COURSE_ID)) + } + + @Test + fun `old session does not write after reset when it reaches the write guard`() = runTest { + val mutexHolder = fixture("mutex-holder") + val lateOldValue = fixture("late-old") + val insertGate = courseDao.gateInsert(mutexHolder.roomEntity) + + val holderCall = startFreshRequest() + apiRequests.receive().response.complete(mutexHolder.response) + insertGate.started.await() + + val lateCall = async(start = CoroutineStart.UNDISPATCHED) { + repository.getCourseStructureFlow(COURSE_ID, forceRefresh = true).first() + } + apiRequests.receive().response.complete(lateOldValue.response) + + repository.endCourseSession() + repository.startCourseSession(COURSE_ID) + insertGate.release.complete(Unit) + + holderCall.await() + lateCall.await() + assertFalse(courseDao.insertCalls.any { it === lateOldValue.roomEntity }) + courseDao.storedCourseStructure.set(null) + assertFailureType { + repository.getCourseStructureFromCache(COURSE_ID) + } + } + + @Test + fun `new session write is final after an old Room insert resumes`() = runTest { + val oldV1 = fixture("old-v1") + val newV2 = fixture("new-v2") + val oldInsertGate = courseDao.gateInsert(oldV1.roomEntity) + + val oldCall = async(start = CoroutineStart.UNDISPATCHED) { + repository.getCourseStructureFlow(COURSE_ID, forceRefresh = true).first() + } + apiRequests.receive().response.complete(oldV1.response) + oldInsertGate.started.await() + + repository.endCourseSession() + repository.startCourseSession(COURSE_ID) + + val newCall = startFreshRequest() + apiRequests.receive().response.complete(newV2.response) + assertFalse(newCall.isCompleted) + + oldInsertGate.release.complete(Unit) + oldCall.await() + assertSame(newV2.domain, newCall.await()) + + assertTrue(courseDao.insertCalls.any { it === oldV1.roomEntity }) + assertSame(newV2.roomEntity, courseDao.storedCourseStructure.get()) + assertSame(newV2.domain, repository.getCourseStructureFromCache(COURSE_ID)) + } + + @Test + fun `fresh fetch does not read an old Room row`() = runTest { + val staleRoomV1 = fixture("stale-room-v1") + val originV2 = fixture("origin-v2") + courseDao.storedCourseStructure.set(staleRoomV1.roomEntity) + + repository.endCourseSession() + repository.startCourseSession(COURSE_ID) + val readsBeforeFreshCall = courseDao.readCount.get() + + val freshCall = startFreshRequest() + apiRequests.receive().response.complete(originV2.response) + + assertSame(originV2.domain, freshCall.await()) + assertEquals(readsBeforeFreshCall, courseDao.readCount.get()) + assertSame(originV2.roomEntity, courseDao.storedCourseStructure.get()) + assertSame(originV2.domain, repository.getCourseStructureFromCache(COURSE_ID)) + } + + @Test + fun `cache-first read may return old Room data until a fresh fetch replaces it`() = runTest { + val staleRoomV1 = fixture("stale-room-v1") + val originV2 = fixture("origin-v2") + courseDao.storedCourseStructure.set(staleRoomV1.roomEntity) + repository.endCourseSession() + repository.startCourseSession(COURSE_ID) + + every { networkConnection.isOnline() } returns false + val displayedStructure = repository.getCourseStructureFlow(COURSE_ID).first() + assertSame(staleRoomV1.domain, displayedStructure) + + every { networkConnection.isOnline() } returns true + val freshCall = startFreshRequest() + apiRequests.receive().response.complete(originV2.response) + assertSame(originV2.domain, freshCall.await()) + assertSame(originV2.roomEntity, courseDao.storedCourseStructure.get()) + assertSame(originV2.domain, repository.getCourseStructureFromCache(COURSE_ID)) + } + + @Test + fun `old session completion does not clear a new session refresh marker`() = runTest { + val oldValue = fixture("old") + val cachedV1 = fixture("cached-v1") + val refreshedV2 = fixture("refreshed-v2") + + val oldCall = async(start = CoroutineStart.UNDISPATCHED) { + repository.getCourseStructureFlow(COURSE_ID, forceRefresh = true).first() + } + val oldRequest = apiRequests.receive() + + repository.endCourseSession() + repository.startCourseSession(COURSE_ID) + courseDao.storedCourseStructure.set(cachedV1.roomEntity) + assertSame(cachedV1.domain, repository.getCourseStructureFromCache(COURSE_ID)) + + oldRequest.response.complete(oldValue.response) + assertSame(oldValue.domain, oldCall.await()) + + val newFlow = async(start = CoroutineStart.UNDISPATCHED) { + repository.getCourseStructureFlow(COURSE_ID).take(2).toList() + } + val newRequest = apiRequests.receive() + assertEquals("stale-if-error=0", newRequest.cacheControl) + newRequest.response.complete(refreshedV2.response) + + assertEquals(listOf(cachedV1.domain, refreshedV2.domain), newFlow.await()) + assertSame(refreshedV2.roomEntity, courseDao.storedCourseStructure.get()) + } + + @Test + fun `status refresh marker survives an old course structure completion`() = runTest { + val cachedStatus = CourseComponentStatus("cached") + val refreshedStatus = CourseComponentStatus("refreshed") + + completeOldCourseStructureAfterStartingNewSession() + coEvery { api.getCourseStatus(any(), COURSE_ID) } returns cachedStatus + assertEquals(cachedStatus.mapToDomain(), repository.getCourseStatus(COURSE_ID)) + + coEvery { api.getCourseStatus(any(), COURSE_ID) } returns refreshedStatus + assertEquals( + listOf(cachedStatus.mapToDomain(), refreshedStatus.mapToDomain()), + repository.getCourseStatusFlow(COURSE_ID).take(2).toList(), + ) + } + + @Test + fun `dates refresh marker survives an old course structure completion`() = runTest { + val cachedDates = courseDates(hasEnded = false) + val refreshedDates = courseDates(hasEnded = true) + + completeOldCourseStructureAfterStartingNewSession() + coEvery { api.getCourseDates(COURSE_ID, any(), any()) } returns cachedDates + assertEquals( + cachedDates.getCourseDatesResult(), + repository.getCourseDates(COURSE_ID, forceRefresh = true), + ) + + coEvery { api.getCourseDates(COURSE_ID, any(), any()) } returns refreshedDates + assertEquals( + listOf(cachedDates.getCourseDatesResult(), refreshedDates.getCourseDatesResult()), + repository.getCourseDatesFlow(COURSE_ID).take(2).toList(), + ) + } + + @Test + fun `progress refresh marker survives an old course structure completion`() = runTest { + val cachedProgress = courseProgress("cached") + val refreshedProgress = courseProgress("refreshed") + + completeOldCourseStructureAfterStartingNewSession() + coEvery { api.getCourseProgress(COURSE_ID) } returns cachedProgress + assertEquals( + cachedProgress.mapToDomain(), + repository.getCourseProgress( + courseId = COURSE_ID, + isRefresh = true, + getOnlyCacheIfExist = false, + ).first(), + ) + + coEvery { api.getCourseProgress(COURSE_ID) } returns refreshedProgress + assertEquals( + listOf(cachedProgress.mapToDomain(), refreshedProgress.mapToDomain()), + repository.getCourseProgress( + courseId = COURSE_ID, + isRefresh = false, + getOnlyCacheIfExist = true, + ).take(2).toList(), + ) + } + + @Test + fun `cache-first fetch started before reset keeps its original session`() = runTest { + val staleRoomValue = fixture("stale-room") + val oldResponse = fixture("old-response") + val refreshedValue = fixture("refreshed") + courseDao.storedCourseStructure.set(staleRoomValue.roomEntity) + val readGate = courseDao.gateNextRead() + + val oldCall = async(start = CoroutineStart.UNDISPATCHED) { + repository.getCourseStructureFlow(COURSE_ID, forceRefresh = true).first() + } + readGate.started.await() + + repository.endCourseSession() + repository.startCourseSession(COURSE_ID) + readGate.release.complete(Unit) + + val oldRequest = apiRequests.receive() + oldRequest.response.complete(oldResponse.response) + assertSame(oldResponse.domain, oldCall.await()) + assertFalse(courseDao.insertCalls.any { it === oldResponse.roomEntity }) + + val newFlow = async(start = CoroutineStart.UNDISPATCHED) { + repository.getCourseStructureFlow(COURSE_ID).take(2).toList() + } + val newRequest = apiRequests.tryReceive().getOrNull() + ?: error("Expected the new session to issue a refresh request") + newRequest.response.complete(refreshedValue.response) + + assertEquals(listOf(staleRoomValue.domain, refreshedValue.domain), newFlow.await()) + assertSame(refreshedValue.roomEntity, courseDao.storedCourseStructure.get()) + } + + @Test + fun `old fresh completion does not advance the new session completion version`() = runTest { + val oldFreshV1 = fixture("old-fresh-v1") + val newNonFreshV2 = fixture("new-non-fresh-v2") + val oldInsertGate = courseDao.gateInsert(oldFreshV1.roomEntity) + + val oldFreshCall = startFreshRequest() + apiRequests.receive().response.complete(oldFreshV1.response) + oldInsertGate.started.await() + + repository.endCourseSession() + repository.startCourseSession(COURSE_ID) + + val newNonFreshCall = async(start = CoroutineStart.UNDISPATCHED) { + repository.getCourseStructureFlow(COURSE_ID, forceRefresh = true).first() + } + val newNonFreshRequest = apiRequests.receive() + + oldInsertGate.release.complete(Unit) + oldFreshCall.await() + newNonFreshRequest.response.complete(newNonFreshV2.response) + assertSame(newNonFreshV2.domain, newNonFreshCall.await()) + + assertSame(newNonFreshV2.roomEntity, courseDao.storedCourseStructure.get()) + assertSame(newNonFreshV2.domain, repository.getCourseStructureFromCache(COURSE_ID)) + } + + @Test + fun `fresh request started before reset cannot write to the new session`() = runTest { + val oldFreshValue = fixture("old-fresh") + val newFreshValue = fixture("new-fresh") + + val oldFreshCall = startFreshRequest() + val oldFreshRequest = apiRequests.receive() + repository.endCourseSession() + repository.startCourseSession(COURSE_ID) + + oldFreshRequest.response.complete(oldFreshValue.response) + assertSame(oldFreshValue.domain, oldFreshCall.await()) + assertFalse(courseDao.insertCalls.any { it === oldFreshValue.roomEntity }) + + val newFreshCall = startFreshRequest() + val newFreshRequest = apiRequests.receive() + newFreshRequest.response.complete(newFreshValue.response) + + assertSame(newFreshValue.domain, newFreshCall.await()) + assertSame(newFreshValue.roomEntity, courseDao.storedCourseStructure.get()) + } + + private fun TestScope.startFreshRequest(): Deferred { + return async(start = CoroutineStart.UNDISPATCHED) { + repository.getCourseStructureFresh(COURSE_ID) + } + } + + private suspend fun TestScope.completeOldCourseStructureAfterStartingNewSession() { + val oldStructure = fixture("old") + val oldCall = async(start = CoroutineStart.UNDISPATCHED) { + repository.getCourseStructureFlow(COURSE_ID, forceRefresh = true).first() + } + val oldRequest = apiRequests.receive() + + repository.endCourseSession() + repository.startCourseSession(COURSE_ID) + + oldRequest.response.complete(oldStructure.response) + assertSame(oldStructure.domain, oldCall.await()) + } + + private fun courseDates(hasEnded: Boolean) = CourseDates( + courseDateBlocks = emptyList(), + datesBannerInfo = null, + hasEnded = hasEnded, + ) + + private fun courseProgress(verifiedMode: String) = CourseProgressResponse( + verifiedMode = verifiedMode, + accessExpiration = null, + certificateData = null, + completionSummary = null, + courseGrade = null, + creditCourseRequirements = null, + end = null, + enrollmentMode = null, + gradingPolicy = null, + hasScheduledContent = null, + sectionScores = null, + studioUrl = null, + username = null, + userHasPassingGrade = null, + verificationData = null, + disableProgressGraph = null, + ) + + private fun fixture(version: String): CourseStructureFixture { + val domain = CoreMocks.mockCourseStructure.copy( + id = COURSE_ID, + name = version, + ) + val roomEntity = mockk() + val response = mockk() + every { roomEntity.mapToDomain() } returns domain + every { response.mapToDomain() } returns domain + every { response.mapToRoomEntity() } returns roomEntity + return CourseStructureFixture(domain, roomEntity, response) + } + + private suspend fun failureFrom(call: Deferred<*>): Throwable { + return try { + call.await() + fail("Expected the request to fail") + error("unreachable") + } catch (throwable: Throwable) { + throwable + } + } + + private suspend inline fun assertFailureType( + crossinline block: suspend () -> Unit, + ) { + val failure = try { + block() + fail("Expected ${T::class.simpleName}") + error("unreachable") + } catch (throwable: Throwable) { + throwable + } + assertTrue(failure is T) + } + + private fun httpException(statusCode: Int): HttpException { + val body = "{}".toResponseBody("application/json".toMediaType()) + return HttpException(Response.error(statusCode, body)) + } + + private class GatedCourseDao : CourseDao { + data class Gate( + val started: CompletableDeferred = CompletableDeferred(), + val release: CompletableDeferred = CompletableDeferred(), + ) + + val storedCourseStructure = AtomicReference() + val insertCalls = Collections.synchronizedList(mutableListOf()) + val readCount = AtomicInteger(0) + + private var nextReadGate: Gate? = null + private var gatedInsertEntity: CourseStructureEntity? = null + private var insertGate: Gate? = null + + fun gateNextRead(): Gate { + return Gate().also { nextReadGate = it } + } + + fun gateInsert(roomEntity: CourseStructureEntity): Gate { + gatedInsertEntity = roomEntity + return Gate().also { insertGate = it } + } + + override suspend fun getCourseStructureById(id: String): CourseStructureEntity? { + readCount.incrementAndGet() + val structureAtReadStart = storedCourseStructure.get() + val gate = nextReadGate + nextReadGate = null + if (gate != null) { + gate.started.complete(Unit) + gate.release.await() + } + return structureAtReadStart + } + + override suspend fun insertCourseStructureEntity( + vararg courseStructureEntity: CourseStructureEntity, + ) { + for (roomEntity in courseStructureEntity) { + insertCalls.add(roomEntity) + if (roomEntity === gatedInsertEntity) { + val gate = insertGate + gate?.started?.complete(Unit) + gate?.release?.await() + } + storedCourseStructure.set(roomEntity) + } + } + + override suspend fun clearCourseStructure() { + storedCourseStructure.set(null) + } + + override suspend fun clearVideoProgress() = Unit + + override suspend fun clearEnrollmentCachedData() = Unit + + override suspend fun clearCourseProgressData() = Unit + + override suspend fun insertCourseEnrollmentDetailsEntity( + vararg courseEnrollmentDetailsEntity: CourseEnrollmentDetailsEntity, + ) = Unit + + override suspend fun getCourseEnrollmentDetailsById( + id: String, + ): CourseEnrollmentDetailsEntity? = null + + override suspend fun insertVideoProgressEntity( + vararg videoProgressEntity: VideoProgressEntity, + ) = Unit + + override suspend fun getVideoProgressByBlockId(blockId: String): VideoProgressEntity? = null + + override suspend fun insertCourseProgressEntity( + vararg courseProgressEntity: CourseProgressEntity, + ) = Unit + + override suspend fun getCourseProgressById(id: String): CourseProgressEntity? = null + } + + private companion object { + const val COURSE_ID = "course-v1:TestX+Freshness+2026" + } +} diff --git a/course/src/test/java/org/openedx/course/domain/interactor/CourseInteractorFreshTest.kt b/course/src/test/java/org/openedx/course/domain/interactor/CourseInteractorFreshTest.kt new file mode 100644 index 000000000..89b1657e1 --- /dev/null +++ b/course/src/test/java/org/openedx/course/domain/interactor/CourseInteractorFreshTest.kt @@ -0,0 +1,102 @@ +package org.openedx.course.domain.interactor + +import io.mockk.coEvery +import io.mockk.coVerify +import io.mockk.every +import io.mockk.mockk +import io.mockk.verify +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.flow.flowOf +import kotlinx.coroutines.test.runTest +import org.junit.Assert.assertSame +import org.junit.Assert.assertTrue +import org.junit.Assert.fail +import org.junit.Before +import org.junit.Test +import org.openedx.core.CoreMocks +import org.openedx.core.exception.NoCachedDataException +import org.openedx.core.system.connection.NetworkConnection +import org.openedx.course.data.repository.CourseRepository + +@OptIn(ExperimentalCoroutinesApi::class) +class CourseInteractorFreshTest { + + private lateinit var repository: CourseRepository + private lateinit var networkConnection: NetworkConnection + private lateinit var interactor: CourseInteractor + + @Before + fun setUp() { + repository = mockk() + networkConnection = mockk() + interactor = CourseInteractor(repository, networkConnection) + } + + @Test + fun `offline refresh returns cached data without using the fresh path`() = runTest { + val cachedStructure = CoreMocks.mockCourseStructure.copy(name = "cached") + every { networkConnection.isOnline() } returns false + every { + repository.getCourseStructureFlow(COURSE_ID, forceRefresh = true) + } returns flowOf(cachedStructure) + + val result = interactor.getCourseStructure(COURSE_ID, isNeedRefresh = true) + + assertSame(cachedStructure, result) + coVerify(exactly = 0) { repository.getCourseStructureFresh(any()) } + } + + @Test + fun `offline refresh without cached data throws the cache exception`() = runTest { + every { networkConnection.isOnline() } returns false + every { + repository.getCourseStructureFlow(COURSE_ID, forceRefresh = true) + } returns flow { throw NoCachedDataException() } + + val failure = try { + interactor.getCourseStructure(COURSE_ID, isNeedRefresh = true) + fail("Expected NoCachedDataException") + error("unreachable") + } catch (throwable: Throwable) { + throwable + } + + assertTrue(failure is NoCachedDataException) + coVerify(exactly = 0) { repository.getCourseStructureFresh(any()) } + } + + @Test + fun `online refresh uses the fresh repository path`() = runTest { + val freshStructure = CoreMocks.mockCourseStructure.copy(name = "fresh") + every { networkConnection.isOnline() } returns true + coEvery { repository.getCourseStructureFresh(COURSE_ID) } returns freshStructure + + val result = interactor.getCourseStructure(COURSE_ID, isNeedRefresh = true) + + assertSame(freshStructure, result) + coVerify(exactly = 1) { repository.getCourseStructureFresh(COURSE_ID) } + verify(exactly = 0) { repository.getCourseStructureFlow(any(), any()) } + } + + @Test + fun `cache-first request keeps Flow behavior`() = runTest { + val cachedStructure = CoreMocks.mockCourseStructure.copy(name = "cached") + every { + repository.getCourseStructureFlow(COURSE_ID, forceRefresh = false) + } returns flowOf(cachedStructure) + + val result = interactor.getCourseStructure(COURSE_ID, isNeedRefresh = false) + + assertSame(cachedStructure, result) + verify(exactly = 1) { + repository.getCourseStructureFlow(COURSE_ID, forceRefresh = false) + } + coVerify(exactly = 0) { repository.getCourseStructureFresh(any()) } + verify(exactly = 0) { networkConnection.isOnline() } + } + + private companion object { + const val COURSE_ID = "course-v1:TestX+Freshness+2026" + } +}