[#107] Pagination for feeds

Changelog: enchancement

Author: brandon-futo

Cargo.lock

@@ -141,6 +141,7 @@ function WebFeedViewer<T>({
   );
   const items = (data as readonly T[] | null | undefined) ?? [];
   const [visibleCount, setVisibleCount] = useState(WEB_INITIAL_VISIBLE);
+  const [atEnd, setAtEnd] = useState(false);
 
   // Keep parity with native `FlashList` by calling `onLoad`.
   const hasFiredOnLoad = useRef(false);
@@ -163,20 +164,27 @@ function WebFeedViewer<T>({
     if (!node || typeof IntersectionObserver === 'undefined') return;
     const observer = new IntersectionObserver(
       (entries) => {
-        if (!entries.some((e) => e.isIntersecting)) return;
-        setVisibleCount((c) => {
-          // First grow the local window; only delegate to the
-          // consumer's onEndReached once we've exhausted the data.
-          if (c < items.length) return c + WEB_PAGE_SIZE;
-          onEndReached?.();
-          return c;
-        });
+        const intersecting = entries.some((e) => e.isIntersecting);
+        setAtEnd(intersecting);
+        if (!intersecting) return;
+        // Grow the local window; the effect below delegates to the
+        // consumer's onEndReached once we've exhausted the data.
+        setVisibleCount((c) => (c < items.length ? c + WEB_PAGE_SIZE : c));
       },
       { rootMargin: '400px' },
     );
     observer.observe(node);
     return () => observer.disconnect();
-  }, [onEndReached, items.length]);
+  }, [items.length]);
+
+  // Fetch more from the consumer once the local window is exhausted and
+  // the sentinel is in view. Kept out of the setVisibleCount updater so
+  // we never trigger a parent setState during render.
+  useEffect(() => {
+    if (atEnd && visibleCount >= items.length) {
+      onEndReached?.();
+    }
+  }, [atEnd, visibleCount, items.length, onEndReached]);
 
   const isEmpty = items.length === 0;
   const visibleItems = isEmpty ? items : items.slice(0, visibleCount);

apps/polycentric/src/common/components/List.tsx

@@ -1,4 +1,5 @@
-import { v2 } from '@polycentric/react-native';
+import { QueryStatus, v2 } from '@polycentric/react-native';
+import type { UseQueryResult } from '@/src/common/query/hooks/useQuery';
 
 export function toBase64(bytes: Uint8Array): string {
   return btoa(String.fromCharCode.apply(null, Array.from(bytes)));
@@ -226,6 +227,42 @@ export function decodeFeedItems(response: v2.GetFeedResponse): PostData[] {
   return items;
 }
 
+/** Extract the items and next page indicator from the feed response. */
+export function decodeFeedQueryResult(
+  data: ArrayBuffer | undefined,
+): [PostData[], boolean] {
+  if (!data) {
+    return [[], false];
+  }
+
+  const response = v2.GetFeedResponse.fromBinary(new Uint8Array(data));
+  let hasNext = !!response.pageInfo?.hasNextPage;
+  return [decodeFeedItems(response), hasNext];
+}
+
+/** Get the forward cursor token from a previous feed query result */
+export function extractFeedToken(
+  _status: QueryStatus | undefined,
+  data: ArrayBuffer | undefined,
+): string | undefined {
+  let forwardToken: string | undefined = undefined;
+
+  if (data) {
+    const response = v2.GetFeedResponse.fromBinary(new Uint8Array(data));
+    forwardToken = response.pageInfo?.endCursor;
+  }
+
+  return forwardToken;
+}
+
+/** Check whether it's reasonable to fetch more feed data from servers */
+export function shouldExtend(
+  hasNextPage: boolean,
+  query: UseQueryResult,
+): boolean {
+  return hasNextPage && query.successfulServers > 0;
+}
+
 /**
  * Dicebear identicon URL for a public key.
  */

apps/polycentric/src/common/lib/polycentric-hooks/helpers.ts

@@ -29,6 +29,9 @@ export {
   decodeBundle,
   decodePostBundle as decodeV2PostBundle,
   decodeFeedItems,
+  decodeFeedQueryResult,
+  extractFeedToken,
+  shouldExtend,
   pubkeyStr,
   identiconUrl,
   pickImageVariant,

apps/polycentric/src/common/lib/polycentric-hooks/index.ts

@@ -9,10 +9,20 @@ import {
 import { create } from 'zustand';
 import { usePolycentric } from '../../lib/polycentric-hooks';
 
+/**
+ * Either a `Query` value or a function that can produce a `Query` from a previous
+ * query's results.
+ */
+export type QuerySource =
+  | Query
+  | ((status: QueryStatus | undefined, data: ArrayBuffer | undefined) => Query);
+
 export type QueryRef = {
   data: ArrayBuffer | undefined;
   status: QueryStatus;
   error: string | null;
+  successfulServers: number;
+  pendingServers: number | undefined;
 };
 
 type QueryArgs = {
@@ -36,13 +46,16 @@ type QueryStoreState = {
   subscribe: (key: string, args: QueryArgs) => void;
   unsubscribe: (key: string) => void;
   refresh: (key: string, args?: QueryArgs) => void;
+  extend: (key: string, args: QueryArgs) => void;
   invalidate: (key: string, args?: QueryArgs) => void;
 };
 
 const EMPTY_ENTRY: QueryRef = Object.freeze({
   data: undefined,
   status: QueryStatus.Loading,
   error: null,
+  successfulServers: 0,
+  pendingServers: undefined,
 });
 
 export const useQueryStore = create<QueryStoreState>((set, get) => {
@@ -53,7 +66,9 @@ export const useQueryStore = create<QueryStoreState>((set, get) => {
       if (
         merged.data === prev.data &&
         merged.status === prev.status &&
-        merged.error === prev.error
+        merged.error === prev.error &&
+        merged.successfulServers === prev.successfulServers &&
+        merged.pendingServers === prev.pendingServers
       ) {
         return {};
       }
@@ -68,6 +83,8 @@ export const useQueryStore = create<QueryStoreState>((set, get) => {
     updateQueryRef(key, {
       status: QueryStatus.Loading,
       error: null,
+      successfulServers: 0,
+      pendingServers: undefined,
     });
 
     // Request from rs-core
@@ -79,7 +96,12 @@ export const useQueryStore = create<QueryStoreState>((set, get) => {
     // Listen for outputs from relevant servers
     const sub = observable.subscribe({
       next(result) {
-        updateQueryRef(key, { data: result.data, status: result.status });
+        updateQueryRef(key, {
+          data: result.data,
+          status: result.status,
+          successfulServers: result.successfulServers,
+          pendingServers: result.pendingServers,
+        });
       },
       error(message) {
         console.warn(`useQuery[${key}] error: ${message}`);
@@ -139,6 +161,13 @@ export const useQueryStore = create<QueryStoreState>((set, get) => {
       sub.dispose = fetch(key, next);
     },
 
+    extend(key, args) {
+      const sub = get().subscriptions.get(key);
+      if (!sub) return;
+      sub.dispose();
+      sub.dispose = fetch(key, args);
+    },
+
     invalidate(key, args) {
       const sub = get().subscriptions.get(key);
       const target = args ?? sub?.args;
@@ -151,6 +180,12 @@ export type UseQueryResult = QueryRef & {
   isLoading: boolean;
   /** Re-run the fan-out. Cached data stays visible until the new responses arrive. */
   refresh: () => void;
+  /**
+   * Re-run the fan-out, but without updating the subscription's query args.
+   * This allows doing extra queries to add more data while still having refreshes
+   * re-run the original query.
+   */
+  extend: () => void;
   /**
    * Drop the rust-side cache for this key, then re-run the fan-out.
    * Optional `opts` overrides the original `QueryOpts` for this run
@@ -195,7 +230,9 @@ export function setQueryCache(
     if (
       merged.data === prev.data &&
       merged.status === prev.status &&
-      merged.error === prev.error
+      merged.error === prev.error &&
+      merged.successfulServers === prev.successfulServers &&
+      merged.pendingServers === prev.pendingServers
     ) {
       return {};
     }
@@ -206,23 +243,33 @@ export function setQueryCache(
 }
 
 /**
- * Subscribe to a rust-core query and share its state across every
- * consumer using the same `queryKey`. The first consumer kicks off
- * the rust-side fan-out; subsequent consumers refcount onto the same
- * subscription. `refresh` / `invalidate` re-run the shared fan-out
- * for every attached consumer. Set `enabled` to `false` to skip the
- * subscription entirely (the hook still returns cached state if any).
+ * Subscribe to a rust-core query and share its state across every consumer using
+ * the same `queryKey`.
+ * The first consumer kicks off the rust-side fan-out, and subsequent consumers
+ * refcount onto the same subscription.
+ * `refresh()` / `invalidate()` re-run the shared fan-out for every attached consumer.
+ * Set `enabled` to `false` to skip the subscription entirely (the hook still
+ * returns cached state if any).
+ *
+ * Extending/infinite queries can be done by using the "merge" update mode and
+ * providing a function for the query source.
+ * Then, calling `extend()` will keep the query key and subscription the same,
+ * while triggering a new fan-out that will be added to the existing data.
  */
 export function useQuery(
   queryKey: QueryKey,
-  query: Query,
+  querySource: QuerySource,
   opts: QueryOpts = { fetchMode: FetchMode.Default },
   enabled = true,
 ): UseQueryResult {
   const client = usePolycentric();
   const cacheKey = queryKey.join('\0');
 
   const entry = useQueryStore((s) => s.queries.get(cacheKey) ?? EMPTY_ENTRY);
+  const query =
+    typeof querySource === 'function'
+      ? querySource(undefined, undefined)
+      : querySource;
 
   // Keep `argsRef` pointing at the freshest call args so the
   // imperative handlers below always see them without needing the
@@ -240,6 +287,16 @@ export function useQuery(
   return {
     ...entry,
     isLoading: enabled && entry.status === QueryStatus.Loading,
+    extend: () => {
+      const query =
+        typeof querySource === 'function'
+          ? querySource(entry.status, entry.data)
+          : querySource;
+
+      useQueryStore
+        .getState()
+        .extend(cacheKey, { client, queryKey, query, opts });
+    },
     refresh: () => useQueryStore.getState().refresh(cacheKey, argsRef.current),
     invalidate: (overrideOpts?: QueryOpts) =>
       useQueryStore

apps/polycentric/src/common/query/hooks/useQuery.ts

@@ -35,9 +35,8 @@ export function injectPostIntoFeedCache(
     if (bundleEventId(b) === newId) return;
   }
 
-  const updated = v2.GetFeedResponse.create({
-    eventBundles: [newBundle, ...response.eventBundles],
-  });
-  const bytes = v2.GetFeedResponse.toBinary(updated);
+  response.eventBundles.unshift(newBundle);
+
+  const bytes = v2.GetFeedResponse.toBinary(response);
   setQueryCache(queryKey, { data: bytes.slice().buffer });
 }

apps/polycentric/src/features/feed/hooks/feedCache.ts

@@ -1,10 +1,12 @@
 import { useMemo } from 'react';
-import { Query, QueryStatus, v2 } from '@polycentric/react-native';
+import { Query, QueryStatus, UpdateMode } from '@polycentric/react-native';
 import {
-  decodeFeedItems,
+  decodeFeedQueryResult,
+  extractFeedToken,
+  shouldExtend,
   usePolycentricContext,
 } from '@/src/common/lib/polycentric-hooks';
-import { type FeedHookResult, NOOP } from './types';
+import { type FeedHookResult } from './types';
 import { useQuery } from '@/src/common/query/hooks/useQuery';
 import { feedQueryKeys } from './feedCache';
 
@@ -18,27 +20,34 @@ export function useExploreFeed(options?: {
 
   const query = useQuery(
     feedQueryKeys.explore(identity),
-    new Query.GetExploreFeed({
-      identity: identity === '' ? undefined : identity,
-    }),
-    undefined,
+    (status, data) => {
+      const forwardToken = extractFeedToken(status, data);
+
+      return new Query.GetExploreFeed({
+        identity: identity === '' ? undefined : identity,
+        limit: options?.perServerLimit,
+        forwardToken,
+      });
+    },
+    { updateMode: UpdateMode.Merge },
     enabled,
   );
 
-  const items = useMemo(() => {
-    if (!query.data) {
-      return [];
-    }
-    const response = v2.GetFeedResponse.fromBinary(new Uint8Array(query.data));
-    return decodeFeedItems(response);
-  }, [query.data]);
+  const [items, hasNext] = useMemo(
+    () => decodeFeedQueryResult(query.data),
+    [query.data],
+  );
 
   return {
     items,
     isLoading: query.status === QueryStatus.Loading,
     error: query.error ? new Error(query.error) : null,
-    loadMore: NOOP,
-    hasMore: false,
+    loadMore: async () => {
+      if (shouldExtend(hasNext, query)) {
+        query.extend();
+      }
+    },
+    hasMore: hasNext,
     refresh: query.refresh,
   };
 }