Remote meditor notworking when return MediatorResult.Success

I'm using paging 3 to load some data in recycler view. I fetch data from the server and then store them in my local database. so here is my DAO interface:

@Dao
interface BarberDAO {
   @Insert(onConflict = OnConflictStrategy.REPLACE)
   fun insertAll(movies: List<Barber>)

   @Query("SELECT * FROM barbershop ORDER BY id ASC")
   fun selectAll(): PagingSource<Int, Barber>

   @Query("DELETE FROM barbershop")
   fun clearBarber()
}

I used a remote mediator like this

@ExperimentalPagingApi
class BarberRxRemoteMediator @Inject constructor(
    private val main: MainApi,
    private val database: BarberDatabase
) : RxRemoteMediator<Int, Barber>() {

override fun loadSingle(
    loadType: LoadType,
    state: PagingState<Int, Barber>
): Single<MediatorResult> {
    return Single.just(loadType)
        .subscribeOn(Schedulers.io())
        .map {
            when (it) {
                LoadType.REFRESH -> {
                    val remoteKeys = getRemoteKeyClosestToCurrentPosition(state)
                    remoteKeys?.nextKey?.minus(1) ?: 1
                }

                LoadType.PREPEND -> {
                    Log.d("RemoteMediator:", "PREPEND")
                    val remoteKeys = getRemoteKeyForFistItem(state)
                        ?: throw InvalidObjectException("Result is empty")
                    remoteKeys.prevKey ?: INVALID_PAGE
                }

                LoadType.APPEND -> {
                    Log.d("RemoteMediator:", "APPEND")
                    val remoteKeys = getRemoteKeyForLastItem(state)
                        ?: throw InvalidObjectException("Result is empty")

                    remoteKeys.nextKey ?: INVALID_PAGE
                }

            }
        }
        .flatMap { page ->
            if (page == INVALID_PAGE) {
                Single.just(MediatorResult.Success(endOfPaginationReached = true))
            } else {
                main.barberShops(page)
                    .map { insertToDB(page, loadType, it) }
                    .map<MediatorResult> {
                        Log.d("RemoteMediator:", "MediatorResult.Success:"+ it.isEmpty())
                        MediatorResult.Success(endOfPaginationReached = it.isEmpty())
                    }
                    .onErrorReturn {
                        Log.d("RemoteMediator:", " MediatorResult.Error:$it")
                        MediatorResult.Error(it)
                    }
            }
        }
        .onErrorReturn {
            Log.d("RemoteMediator:", " MediatorResult.Error:$it")
            MediatorResult.Error(it)
        }
}

private fun insertToDB(
    page: Int,
    loadType: LoadType,
    data: List<Barber>
): List<Barber> {
    database.runInTransaction {
        if (loadType == LoadType.REFRESH) {
            database.remoteKeyDAO().clearRemoteKeys()
            database.barberDao().clearBarber()
        }

        val prevKey = if (page == 1) null else page - 1
        val nextKey = if (data.isEmpty()) null else page + 1

        val keys = data.map {
            RemoteKey(
                id = it.id,
                prevKey = prevKey,
                nextKey = nextKey
            )
        }

        database.remoteKeyDAO().insertAll(keys)
        database.barberDao().insertAll(data)
        Log.d("RemoteMediator:", "Request api và lưu dữ liệu vào SQLite. Data :$keys")
    }
    return data
}

private fun getRemoteKeyForFistItem(state: PagingState<Int, Barber>): RemoteKey? {
    return state.pages.firstOrNull { it.data.isNotEmpty() }
        ?.data?.firstOrNull()
        ?.let { repo ->
            database.remoteKeyDAO().remoteKeysByMovieId(repo.id)
        }
}

private fun getRemoteKeyForLastItem(state: PagingState<Int, Barber>): RemoteKey? {
    return state.pages.lastOrNull { it.data.isNotEmpty() }
        ?.data?.lastOrNull()
        ?.let { repo ->
            database.remoteKeyDAO().remoteKeysByMovieId(repo.id)
        }
}

private fun getRemoteKeyClosestToCurrentPosition(state: PagingState<Int, Barber>): RemoteKey? {
    return state.anchorPosition?.let { position ->
        state.closestItemToPosition(position)?.id?.let { id ->
            database.remoteKeyDAO().remoteKeysByMovieId(id)
        }
    }
}

companion object {
    const val INVALID_PAGE = -1
}

}

and use the data using my view model class like this:

@ExperimentalPagingApi
fun getBarberTest(): Observable<PagingData<Barber>> {
    return barberRepository.getBarber().cachedIn(viewModelScope)
}

and I've got this chunk of code in my fragment:

mDisposable.add(viewModel.getBarberTest().subscribe {
     adapter.submitData(lifecycle, it)
})

Update: D/loadState: CombinedLoadStates(refresh=NotLoading(endOfPaginationReached=false), prepend=NotLoading(endOfPaginationReached=true), append=NotLoading(endOfPaginationReached=false), source=LoadStates(refresh=NotLoading(endOfPaginationReached=false), prepend=NotLoading(endOfPaginationReached=true), append=NotLoading(endOfPaginationReached=true)), mediator=LoadStates(refresh=NotLoading(endOfPaginationReached=false), prepend=NotLoading(endOfPaginationReached=true), append=NotLoading(endOfPaginationReached=false)))

When I debug and see that when the first paging data loads successfully and returns MediatorResult.Success(endOfPaginationReached = false) . paging doesn't refresh the data hence causing LoadType.APPEND error. I really don't know what I did wrong. hope all help.Sorry for my bad English