import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

// Question: Is it hacky to hang onto the old Job where the first flow is collected
// and then to cancel it to achieve the desired results?

private val globalScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)

val remoteData = listOf(
    "abc",
    "ab",
    "abb",
    "anb",
    "abcd",
    "abcdef",
    "ged",
    "maybe"
)

class SimpleCache {
    private var initial = setOf(
        "one",
        "two"
    )
    private val mutableStateFlow = MutableStateFlow(initial)

    fun resetCache() = mutableStateFlow.update { initial }

    fun search(term: String) =
        mutableStateFlow.asStateFlow().map { valuesSet ->
            valuesSet.filter { it.contains(term) }
        }

    suspend fun fetchData(term: String) {
        // Taking 3 to simulate an API where there are more matches for the
        // filter and a subsequent fetch can increase the results for a previous
        // call to `search`
        val newItems = remoteData.filter { it.contains(term) }.take(3)

        delay(500)

        mutableStateFlow.update { it + newItems }
    }

    var previousSearchJob: Job? = null
}

suspend fun searchWithCancellation(term: String, cache: SimpleCache) {
    globalScope.launch {
        cache.previousSearchJob?.cancel()

        cache.previousSearchJob = launch {
            cache.search(term).collect {
                println("For: $term")
                println("Collected: $it")
            }
        }
        launch {
            cache.fetchData(term)
        }
    }
}

suspend fun searchWithoutCancellation(term: String, cache: SimpleCache) {
    globalScope.launch {
        launch {
            cache.search(term).collect {
                println("For: $term")
                println("Collected: $it")
            }
        }
        launch {
            cache.fetchData(term)
        }
    }
}

fun main() {
    val cancellationCache = SimpleCache()
    val baseCache = SimpleCache()

    runBlocking {
        // This section is fine for the first flow but once there are
        // more than one being collected at the same time things break.
        println("Without Cancellation")
        searchWithoutCancellation(term = "a", baseCache)

        delay(1000)

        searchWithoutCancellation(term = "abc", baseCache)

        delay(3000)

        // This section prints the desired results
        println()
        println("With Cancellation")

        searchWithCancellation(term = "a", cancellationCache)

        delay(1000)

        searchWithCancellation(term = "abc", cancellationCache)

        delay(3000)
    }
    /* Output
        Without Cancellation
        For: a
        Collected: []
        For: a
        Collected: [abc, ab, abb]
        For: abc
        Collected: [abc]
        For: a <-- This collection is undesired and could come before or after the collection of "abc"
        Collected: [abc, ab, abb, abcd, abcdef]
        For: abc
        Collected: [abc, abcd, abcdef]

        With Cancellation
        For: a
        Collected: []
        For: a
        Collected: [abc, ab, abb]
        For: abc
        Collected: [abc]
        For: abc
        Collected: [abc, abcd, abcdef]
     */
}