Real-Time Infinite Queries

import { useWSInfiniteQuery } from "@explita/actyx-rpc-react";
import { getFeedPosts } from "@/backend/procedures";
 
function LiveFeed() {
  const {
    data: posts,
    fetchNext,
    hasNext,
    isFetching,
    isConnected,
    error,
  } = useWSInfiniteQuery(getFeedPosts, {
    // Infinite query options
    queryOpts: {
      input: { limit: 10 },
      getNextPageParam: (lastPage) => lastPage.nextCursor,
      queryKey: ["feed"],
    },
    // WebSocket options
    url: "/api/ws/feed",
    // Called on each incoming WS message
    onData({ data: newPost, prepend }) {
      // Prepend incoming live posts to the cache
      prepend(newPost);
    },
  });
 
  return (
    <div>
      <p>Status: {isConnected ? "🟢 Live" : "🔴 Disconnected"}</p>
      <ul>
        {posts.map((post) => (
          <li key={post.id}>{post.title}</li>
        ))}
      </ul>
      {hasNext && (
        <button onClick={() => fetchNext()} disabled={isFetching}>
          {isFetching ? "Loading..." : "Load More"}
        </button>
      )}
    </div>
  );
}

import { useSSEInfiniteQuery } from "@explita/actyx-rpc-react";
import { getNotifications } from "@/backend/procedures";
 
function NotificationFeed() {
  const {
    data: notifications,
    fetchNext,
    hasNext,
    isConnected,
    lastData: latestNotification,
    event: lastEvent,
    error,
  } = useSSEInfiniteQuery(getNotifications, {
    // Infinite query options
    queryOpts: {
      input: { limit: 20 },
      getNextPageParam: (lastPage) => lastPage.nextCursor,
      queryKey: ["notifications"],
    },
    // SSE options
    url: "/api/sse/notifications",
    maxHistory: 50,
    // Called on each SSE event
    onData({ data: notification, prepend, event }) {
      console.log(`Received event: ${event}`);
      prepend(notification);
    },
  });
 
  return (
    <div>
      <p>Status: {isConnected ? "🟢 Connected" : "🔴 Disconnected"}</p>
      {latestNotification && (
        <p className="latest">Latest: {latestNotification.title}</p>
      )}
      <ul>
        {notifications.map((n) => (
          <li key={n.id}>{n.title}</li>
        ))}
      </ul>
      {hasNext && (
        <button onClick={() => fetchNext()} disabled={isFetching}>
          Load Older
        </button>
      )}
    </div>
  );
}

onData({ data, action, allData, append, prepend, update, event }) {
  // data       - The incoming event payload
  // action     - `"added"` or `"updated"` (dedup key matched an existing item)
  // allData    - The current flattened data array from all pages
  // append     - Append item to the last page
  // prepend    - Prepend item to the first page
  // update     - Update a matching item by index or predicate
  // event      - (SSE only) The SSE event name string
}

import { rpc } from "@/lib/rpc/client";
 
function LiveTodoList() {
  const { data: todos, fetchNext, hasNext } = rpc.todos.list.useSSEInfiniteQuery({
    // Pass another proxy procedure directly (type-safe)
    stream: rpc.notif.sse,
 
    // Or pass an explicit URL string:
    // stream: "/api/rpc/notif.sse",
 
    queryOpts: {
      getNextPageParam: (lastPage) => lastPage.nextCursor,
    },
  });
 
  return (
    <ul>
      {todos.map((todo) => (
        <li key={todo.id}>{todo.text}</li>
      ))}
    </ul>
  );
}