Mobile: pagination and real-time updates

// Android: Paging 3 with Room as the source of truth and a RemoteMediator filling it
@OptIn(ExperimentalPagingApi::class)
class OrdersRemoteMediator(
    private val api: OrdersApi,
    private val db: AppDatabase,
    private val filter: OrderFilter,
) : RemoteMediator<Int, OrderEntity>() {

    override suspend fun load(loadType: LoadType, state: PagingState<Int, OrderEntity>): MediatorResult {
        val cursor = when (loadType) {
            LoadType.REFRESH -> null
            LoadType.PREPEND -> return MediatorResult.Success(endOfPaginationReached = true)
            LoadType.APPEND -> db.remoteKeys().forFilter(filter)?.nextCursor
                ?: return MediatorResult.Success(endOfPaginationReached = true)
        }
        return try {
            val page = api.orders(filter, cursor, limit = state.config.pageSize)
            db.withTransaction {
                if (loadType == LoadType.REFRESH) {
                    db.orders().clear(filter)
                    db.remoteKeys().clear(filter)
                }
                db.orders().upsertAll(page.orders.map { it.toEntity(filter) })
                db.remoteKeys().upsert(RemoteKey(filter, page.nextCursor))
            }
            MediatorResult.Success(endOfPaginationReached = page.nextCursor == null)
        } catch (e: IOException) {
            MediatorResult.Error(e)
        }
    }
}

@HiltViewModel
class OrdersViewModel @Inject constructor(api: OrdersApi, db: AppDatabase) : ViewModel() {
    private val filter = MutableStateFlow(OrderFilter.OPEN)

    @OptIn(ExperimentalPagingApi::class, ExperimentalCoroutinesApi::class)
    val orders: Flow<PagingData<Order>> = filter.flatMapLatest { f ->   // a new filter cancels the old pager
        Pager(
            config = PagingConfig(pageSize = 20, prefetchDistance = 10),
            remoteMediator = OrdersRemoteMediator(api, db, f),
            pagingSourceFactory = { db.orders().pagingSource(f) },
        ).flow
    }.map { data -> data.map(OrderEntity::toDomain) }
        .cachedIn(viewModelScope)

    fun setFilter(f: OrderFilter) { filter.value = f }
}

// React: the same generation check in a hook
export function useOrders(api: OrdersApi, filter: OrderFilter) {
  const [orders, setOrders] = useState<Order[]>([]);
  const state = useRef({ cursor: null as string | null, loading: false, end: false, generation: 0 });

  useEffect(() => {                                    // filter changed: start a new generation
    state.current = { cursor: null, loading: false, end: false, generation: state.current.generation + 1 };
    setOrders([]);
    void loadMore();
  }, [filter]);

  const loadMore = useCallback(async () => {
    const s = state.current;
    if (s.loading || s.end) return;                    // one request at a time
    s.loading = true;
    const requested = s.generation;
    try {
      const page = await api.orders(filter, s.cursor, 20);
      if (requested !== state.current.generation) return;   // stale: drop it
      setOrders((old) => {
        const known = new Set(old.map((o) => o.id));
        return [...old, ...page.orders.filter((o) => !known.has(o.id))];
      });
      s.cursor = page.nextCursor;
      s.end = page.nextCursor === null;
    } finally {
      if (requested === state.current.generation) s.loading = false;
    }
  }, [api, filter]);

  return { orders, loadMore };
}

// Android: collect live events only while the screen is at least STARTED
class OrderEventsClient(private val http: OkHttpClient, private val baseUrl: HttpUrl) {
    fun events(lastEventId: () -> String?): Flow<OrderEvent> = callbackFlow {
        val request = Request.Builder()
            .url(baseUrl.newBuilder().addPathSegments("orders/stream").build())
            .apply { lastEventId()?.let { header("Last-Event-ID", it) } }    // resume, don't miss
            .build()
        val source = EventSources.createFactory(http).newEventSource(request, object : EventSourceListener() {
            override fun onEvent(es: EventSource, id: String?, type: String?, data: String) {
                trySend(OrderEvent.parse(id, type, data))
            }
            override fun onFailure(es: EventSource, t: Throwable?, response: Response?) {
                close(t ?: IOException("stream closed"))                     // retried below
            }
        })
        awaitClose { source.cancel() }
    }.retryWhen { _, attempt ->
        delay(minOf(30_000L, 1_000L shl attempt.toInt().coerceAtMost(5)) + Random.nextLong(500))  // backoff + jitter
        true
    }
}

@Composable
fun OrdersScreen(viewModel: OrdersViewModel) {
    val lifecycle = LocalLifecycleOwner.current.lifecycle
    LaunchedEffect(lifecycle) {
        lifecycle.repeatOnLifecycle(Lifecycle.State.STARTED) {    // disconnects in the background
            viewModel.liveUpdates()
        }
    }
    /* list */
}

// Web: EventSource reconnects by itself and sends Last-Event-ID automatically
const source = new EventSource("/orders/stream", { withCredentials: true });

source.addEventListener("order-updated", (e) => {
  const order = JSON.parse((e as MessageEvent).data) as Order;
  upsertById(order);                                  // idempotent: replays are harmless
});

document.addEventListener("visibilitychange", () => {
  if (document.hidden) source.close();               // reopen (with the last id) when visible again
});