Implementation of infinite query

[rayhantr]

Is it possible to use useSubscription for react query's infinite query?

[kaciakmaciak]

Is it possible to use useSubscription for react query's infinite query?

Hi Rayhan,

It is not implemented atm. I've had a look into the react query source code and technically speaking it should be pretty straightforward to add something like useInfiniteSubscription hook into this library.

What is more complicated is to decide the hook api so it makes sense. One of the things which we need to be aware off when working with real-time data is, that the cursor might have changed between fetching the last page and the page we are about to fetch. This can happen, for example, when due to real-time update there has been a new item inserted into the collection in the current query.

Could you please maybe clarify/suggest the use case, so that we can go from there.

Until implemented, you could work around this by implementing the pagination manually.

[kaciakmaciak]

Hi Rayhan, I've implemented the useInfiniteSubscription hook and has been released as a beta. I haven't had much chance to play with it, so any feedback will be very welcome. The options are the same as for useInfiniteQuery. I've been working on examples, but still not quite there yet.

const {
  data: messages,
  isLoading,
  isError,
  isFetchingNextPage,
  hasNextPage,
  fetchNextPage,
} = useInfiniteSubscription(
  ['messages', chatId],
  ({ pageParam }) => getMessages$(chatId, pageParam),
  {
    ...options,
    getNextPageParam: (lastPage) => lastPage.nextCursor,
  }
)

You can install the beta version:

npm i [email protected]

[rayhantr]

Hello @kaciakmaciak, thanks for the amazing work. I have tested it in my project and everything looked good. But there is one problem I am facing. Let me explain you.

This is my query:

subscription GET_ALL_CONVERSATIONS($offset: Int = 0) {
      payload: POC_conversations(limit: 10, order_by: {messages_aggregate: {max: {created_at: desc}}}, offset: $offset) {
        id
        created_at
        messages(limit: 1, order_by: {id: desc}) {
          id
          message
          message_type
          created_at
        }
      }
    }

It fetches 10 conversations per request and orders by desc based on last message received. Now consider I have 20 conversations row available in DB. After I have fetched all 20 of them it is stored in the cache. So when a new conversation is added the cache is updated to 21 entries (as it should be). But the problem here is that the 21st item is not the new data, instead it is added to the cache as a duplicate of the last item (meaning 20th item). Can you kindle check what can be the problem here or am I doing anything wrong?

N.B: When using normal useInfiniteQuery with invalidateQueries it works perfectly.

[kaciakmaciak]

Hi @rayhantr, it looks like there was a bug in the useInfiniteSubscription hook, when after fetching previous/next page the previously subscribed pages have unsubscribed. This has been fixed and the unit tests have been added. I'm not sure if this was related to your issue though. If the problem persist, could you please tell me how you use the useInfiniteSubscription hook? How does your getNextPageParam / getPreviousPageParam function look like?

You can install the fixed version by running:

npm i [email protected]

[kaciakmaciak]

Hi @rayhantr another thing why you might have an issue is the way how the useInfiniteSubscription is implemented at the moment. The getNextPageParam / getPreviousPageParam is only run once per each page which means that the subscription function is run only once per each page (unless mounted/unmounted). That means, that if you want to use something like limit and offset for your pagination, the pages cannot shift and there cannot be items added/removed from pages real-time. You can achieve the real-time pagination, for example, as follows:

• Subscribe to messages from Date.now() and later using useSubscription.
• Subscribe to previous messages using useInfiniteSubscription. If we can assume there will be no change in number of messages per page, you can use limit and offset. The content of the messages will be updated real-time. If you cannot make that assumption, then I recommend to paginate using before after timestamp. This will update the pages appropriately.

Another option is to combine the above two and use useInfiniteSubscription for all messages.

Let us know how you go.

[rayhantr]

Hi @rayhantr, it looks like there was a bug in the useInfiniteSubscription hook, when after fetching previous/next page the previously subscribed pages have unsubscribed. This has been fixed and the unit tests have been added. I'm not sure if this was related to your issue though. If the problem persist, could you please tell me how you use the useInfiniteSubscription hook? How does your getNextPageParam / getPreviousPageParam function look like?

You can install the fixed version by running:

npm i [email protected]

Hello @kaciakmaciak, thank you for the effort. Here's my implementation of useInfiniteSubscription hook.
Query

export const GET_ALL_CONVERSATIONS_SUB = (offset?: number) => {
  const variables = { offset };
  const query = `
    subscription GET_ALL_CONVERSATIONS($offset: Int = 0) {
      payload: POC_conversations(limit: 10, order_by: {messages_aggregate: {max: {created_at: desc}}}, offset: $offset) {
        id
        created_at
        messages(limit: 1, order_by: {id: desc}) {
          id
          message
          message_type
          created_at
        }
      }
    }
  `;

  return { query, variables };
};

Hook and getNextPageParam

const { isLoading, isError, error, fetchNextPage, isFetchingNextPage, hasNextPage, data } = useInfiniteSubscription(queryKey, ({ pageParam = 0 }) => fromWsClientSubscription(gqlData.GET_ALL_CONVERSATIONS_SUB(pageParam as number)), {
    getNextPageParam: (lastPage, allPages) => {
      let length = 0;
      allPages.forEach((page: []) => (length += page.length));
      const nextPageParam = lastPage.length > 0 ? length : undefined;
      return nextPageParam;
    },
    ...options,
  });

I have installed the beta version and now I am facing another problem with this implementation. When a new conversation gets created the new conversation is added to the top of the list. But at the end of the list the last item is also incremented. Meaning now there is 2 item added to the list. After looking into it in depth, I have noticed that my getNextPageParam implementation adds an empty array in the pages list like below (index 3).

So if the previous page (here index 2) has less than 10 items available and the page 3 has an empty array, then whenever a new item is added to the list it goes into the page 2, then the last item (here item 2) from the page 2 is pushed to the page 3 (that means total 2 items has been pushed). Now if I add another item to the list then it will be added again in the same process but this time the previous item of the last item (here item 1) from page 2 is pushed to page 3. And this continues on loop.

Hi @rayhantr another thing why you might have an issue is the way how the useInfiniteSubscription is implemented at the moment. The getNextPageParam / getPreviousPageParam is only run once per each page which means that the subscription function is run only once per each page (unless mounted/unmounted). That means, that if you want to use something like limit and offset for your pagination, the pages cannot shift and there cannot be items added/removed from pages real-time. You can achieve the real-time pagination, for example, as follows:

Subscribe to messages from Date.now() and later using useSubscription.
Subscribe to previous messages using useInfiniteSubscription. If we can assume there will be no change in number of messages per page, you can use limit and offset. The content of the messages will be updated real-time. If you cannot make that assumption, then I recommend to paginate using before after timestamp. This will update the pages appropriately.
Another option is to combine the above two and use useInfiniteSubscription for all messages.

Let us know how you go.

So by far I think by using aggregation query if I get the total item count and then calculate the pagination values using offset I can disable my next page fetching function & stop my UI user to generate an empty page. By not adding the empty page to the pages list I think the problem can be solved. Please do let me know if this approach would be good or bad.

[rayhantr]

This is my implementation of this whole process.

Get conversation query

export const GET_ALL_CONVERSATIONS_SUB = (offset?: number) => {
  const variables = { offset };
  const query = `
    subscription GET_ALL_CONVERSATIONS($offset: Int = 0) {
      payload: POC_conversations(limit: 10, order_by: {messages_aggregate: {max: {created_at: desc}}}, offset: $offset) {
        id
        created_at
        messages(limit: 1, order_by: {id: desc}) {
          id
          message
          message_type
          created_at
        }
      }
    }
  `;

  return { query, variables };
};

Get conversation count

export const GET_CONVERSATION_COUNT = () => {
  const query = `
    subscription GET_CONVERSATION_COUNT {
      payload: POC_conversations_aggregate {
        aggregate {
          count
        }
      }
    }
  `;

  return { query };
};

Finally the hook that returns subscribed data

const { data: totalConversations } = useSubscription("conversation_count", () => fromWsClientSubscription<ConversationCount>(gqlData.GET_CONVERSATION_COUNT()));

const { isLoading, isError, error, fetchNextPage, isFetchingNextPage, hasNextPage, data } = useInfiniteSubscription(queryKey, ({ pageParam = 0 }) => fromWsClientSubscription(gqlData.GET_ALL_CONVERSATIONS_SUB(pageParam as number)), {
  getNextPageParam: (lastPage, allPages) => {
    const offset = 10;
    const totalPaginationSets = Math.ceil(totalConversations.aggregate.count / offset);

    let length = 0;
    allPages.forEach((page: []) => (length += page.length));
    // const nextPageParam = lastPage.length > 0 ? length : undefined;
    const nextPageParam = totalPaginationSets !== allPages.length ? length : undefined;
    return nextPageParam;
  },
  enabled: !!totalConversations,
});