How do I make my own Paging Source when using Remote Mediator in the Paging Library?

I am trying to make my own Paging Source, because the Paging Source provided by the Room library does not suit me. I ran into such a problem that Paging Source loads the first 1 or 2 pages of data (depending on whether Remote Mediator has time to delete data from the database for the Refresh operation), when scrolling these pages down, data is no longer loaded.

I think the problem is that Paging Source does not understand that Remote Mediator has downloaded new data from the API. How do I solve this problem?

Paging Source:

class TopFilmsLocalPagingSource(
    private val filmLocalStorage: FilmLocalStorage,
    private val type: TopFilmCategories): PagingSource<Int, Film>() {

    override fun getRefreshKey(state: PagingState<Int, Film>): Int? {
        return state.anchorPosition?.let { anchorPosition ->
            val anchorPage = state.closestPageToPosition(anchorPosition)
            anchorPage?.prevKey?.plus(1) ?: anchorPage?.nextKey?.minus(1)
        }
    }

    override suspend fun load(params: LoadParams<Int>): LoadResult<Int, Film> {
        val currentPage = params.key ?: 1
        val offset = (currentPage - 1) * params.loadSize
        val count = params.loadSize

        return try {
            val films = filmLocalStorage.getFilmsByType(offset, count, type)
            val prevKey = if (currentPage == 1) null else currentPage - 1
            val nextKey = if (films.count() < count) null else currentPage + 1

            LoadResult.Page(films, prevKey, nextKey)
        } catch (ex: Exception){
            LoadResult.Error(ex)
        }
    }
}

Remote Mediator (the last field in companion object is made for testing):

@OptIn(ExperimentalPagingApi::class)
class FilmRemoteMediator(
    private val filmLocalStorage: FilmLocalStorage,
    private val filmRemoteStorage: FilmRemoteStorage,
    private val type: TopFilmCategories): RemoteMediator<Int, Film>() {

    override suspend fun load(loadType: LoadType, state: PagingState<Int, Film>): MediatorResult {
        return try{
            val loadKey = when (loadType) {
                LoadType.REFRESH -> {
                    1
                }
                LoadType.PREPEND -> {
                    return MediatorResult.Success(endOfPaginationReached = true)
                }
                LoadType.APPEND -> {
                    last += 1
                    last
                }
            }
            val films = when(type){
                TopFilmCategories.TOP_100_POPULAR_FILMS -> filmRemoteStorage.getPopularFilms(loadKey)
                TopFilmCategories.TOP_250_BEST_FILMS -> filmRemoteStorage.getBestFilms(loadKey)
                TopFilmCategories.TOP_AWAIT_FILMS -> filmRemoteStorage.getTopAwaitFilms(loadKey)
            }
            if (loadType == LoadType.REFRESH) {
                filmLocalStorage.refreshFilmsByType(films, type)
                MediatorResult.Success(
                    endOfPaginationReached = films.isEmpty()
                )
            }
            else{
                filmLocalStorage.insertAllFilms(films, type)
                MediatorResult.Success(
                    endOfPaginationReached = films.isEmpty()
                )
            }
        } catch (e: IOException) {
            MediatorResult.Error(e)
        } catch (e: HttpException) {
            MediatorResult.Error(e)
        }
    }

    companion object{
        var last = 1
    }

}

Repository:

class FilmRepositoryImpl @Inject constructor(
    private val filmRemoteStorage: FilmRemoteStorage,
    private val filmLocalStorage: FilmLocalStorage): FilmRepository {

    @OptIn(ExperimentalPagingApi::class)
    override fun getBestFilmsPaged(): Flow<PagingData<DomainFilm>> {
        return Pager(PagingConfig(pageSize = 20, initialLoadSize = 20, prefetchDistance = 20),
        remoteMediator = FilmRemoteMediator(filmLocalStorage,
            filmRemoteStorage, TopFilmCategories.TOP_250_BEST_FILMS)){
            TopFilmsLocalPagingSource(filmLocalStorage, TopFilmCategories.TOP_250_BEST_FILMS)
        }.flow.toDomain()
    }

}

fun Flow<PagingData<com.gramzin.cinescope.data.model.Film>>.toDomain(): Flow<PagingData<DomainFilm>> {
    return transform { value ->
        emit(value.map {
            it.toDomain()
        })
    }
}

I tried to log the actions that occur:

1. paging source: load page 1

2. remote mediator: refresh operation (load page 1)

3. paging source: load page 2

4. remote mediator: load success

5. remote mediator: prepend operation

6. remote mediator: append operation (load page 2)

7. remote mediator: load success

You need to add a ThreadSafeInvalidationObserver that will track changes in the database. You also need to call the registerIfNecessary method from the observer in the load method.

class TopFilmsLocalPagingSource(
private val filmLocalStorage: FilmLocalStorage,
private val type: TopFilmCategories): PagingSource<Int, Film>() {

private val observer = ThreadSafeInvalidationObserver(
    tables = arrayOf(DBUtils.filmsTableName),
    onInvalidated = ::invalidate
)

override fun getRefreshKey(state: PagingState<Int, Film>): Int? {
    return state.anchorPosition?.let { anchorPosition ->
        val anchorPage = state.closestPageToPosition(anchorPosition)
        anchorPage?.prevKey?.plus(1) ?: anchorPage?.nextKey?.minus(1)
    }
}

override suspend fun load(params: LoadParams<Int>): LoadResult<Int, Film> {
    observer.registerIfNecessary(filmLocalStorage.db)
    val currentPage = params.key ?: 1
    val offset = (currentPage - 1) * params.loadSize
    val count = params.loadSize

    return try {
        val films = filmLocalStorage.getFilmsByType(offset, count, type)
        val prevKey = if (currentPage == 1) null else currentPage - 1
        val nextKey = if (films.count() < count) null else currentPage + 1

        LoadResult.Page(films, prevKey, nextKey)
    } catch (ex: Exception){
        LoadResult.Error(ex)
    }
}
}}

It seems that the ThreadSafeInvalidationObserver does not exists, at least in the latest version of room.
Here is an other way to add an inavalidation observer

class TopFilmsLocalPagingSource(
     private val filmLocalStorage: FilmLocalStorage,
     private val type: TopFilmCategories): PagingSource<Int, Film>() {

     init {
          val db = //you need the instance of your database here
          db.invalidationTracker.addObserver(object : InvalidationTracker.Observer(arrayOf(add here your table name)) {
             override fun onInvalidated(tables: Set<String>) {
                 this@TopFilmsLocalPagingSource.invalidate()
                 //Remove this observer because when the invalidate is invoked pager creates a new instance of the paging source
                 db.invalidationTracker.removeObserver(this)
             }
         })
     }

     override fun getRefreshKey(state: PagingState<Int, Film>): Int? {
         return state.anchorPosition?.let { anchorPosition ->
             val anchorPage = state.closestPageToPosition(anchorPosition)
             anchorPage?.prevKey?.plus(1) ?: anchorPage?.nextKey?.minus(1)
         }
     }

     override suspend fun load(params: LoadParams<Int>): LoadResult<Int, Film> {
    
         val currentPage = params.key ?: 1
         val offset = (currentPage - 1) * params.loadSize
         val count = params.loadSize

         return try {
             val films = filmLocalStorage.getFilmsByType(offset, count, type)
             val prevKey = if (currentPage == 1) null else currentPage - 1
             val nextKey = if (films.count() < count) null else currentPage + 1

             LoadResult.Page(films, prevKey, nextKey)
         } catch (ex: Exception){
             LoadResult.Error(ex)
         }
     }
}