feat(tanstack-query): live query options (#721)

Use `.liveOptions` to configure live queries for Event Iterator. Unlike
`.streamedOptions` which accumulates chunks, live queries replace the
entire result with each new chunk received

```ts
const query = useQuery(orpc.live.experimental_liveOptions({
  input: { id: 123 }, // Specify input if needed
  context: { cache: true }, // Provide client context if needed
  // additional options...
}))
```

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
* Introduced experimental support for "live queries," enabling real-time
data updates that replace the entire result with each new chunk.
* Added new utility methods for generating live query keys and options,
compatible with popular query hooks (e.g., useQuery, prefetchQuery).
* Enhanced documentation with detailed explanations and examples for
live queries.

* **Bug Fixes**
* Improved error handling for invalid or unsupported live query outputs.

* **Tests**
* Added comprehensive test suites for live query utilities across React,
Solid, Svelte, Vue, and Angular integrations, ensuring correct behavior
and type inference.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: dinwwwh

apps/content/docs/integrations/tanstack-query.md

@@ -58,6 +58,12 @@ orpc.planet.find.queryOptions({ input: { id: 123 } })
 
 //
 
+//
+
+//
+
+//
+
 //
 ```
 
@@ -91,20 +97,40 @@ const query = useQuery(orpc.planet.find.queryOptions({
 
 ## Streamed Query Options
 
-Use `.streamedOptions` to configure queries for [Event Iterator](/docs/event-iterator), which is built on top of [streamedQuery](https://tanstack.com/query/latest/docs/reference/streamedQuery). Use it with hooks like `useQuery`, `useSuspenseQuery`, or `prefetchQuery`.
+Use `.streamedOptions` to configure queries for [Event Iterator](/docs/event-iterator). This is built on [TanStack Query streamedQuery](https://tanstack.com/query/latest/docs/reference/streamedQuery) and works with hooks like `useQuery`, `useSuspenseQuery`, or `prefetchQuery`.
 
 ```ts
 const query = useQuery(orpc.streamed.experimental_streamedOptions({
   input: { id: 123 }, // Specify input if needed
   context: { cache: true }, // Provide client context if needed
-  queryFnOptions: { // Specify streamedQuery options if needed
+  queryFnOptions: { // Configure streamedQuery behavior
     refetchMode: 'reset',
     maxChunks: 3,
   }
   // additional options...
 }))
 ```
 
+::: info
+Combine with [Client Retry](/docs/plugins/client-retry#event-iterator-sse) for more reliable streaming queries.
+:::
+
+## Live Query Options
+
+Use `.liveOptions` to configure live queries for [Event Iterator](/docs/event-iterator). Unlike `.streamedOptions` which accumulates chunks, live queries replace the entire result with each new chunk received. Works with hooks like `useQuery`, `useSuspenseQuery`, or `prefetchQuery`.
+
+```ts
+const query = useQuery(orpc.live.experimental_liveOptions({
+  input: { id: 123 }, // Specify input if needed
+  context: { cache: true }, // Provide client context if needed
+  // additional options...
+}))
+```
+
+::: info
+Combine with [Client Retry](/docs/plugins/client-retry#event-iterator-sse) for more reliable live queries.
+:::
+
 ## Infinite Query Options
 
 Use `.infiniteOptions` to configure infinite queries. Use it with hooks like `useInfiniteQuery`, `useSuspenseInfiniteQuery`, or `prefetchInfiniteQuery`.
@@ -263,12 +289,14 @@ import {
 interface ClientContext extends TanstackQueryOperationContext {
 }
 
+const GET_OPERATION_TYPE = new Set(['query', 'streamed', 'live', 'infinite'])
+
 const link = new RPCLink<ClientContext>({
   url: 'http://localhost:3000/rpc',
   method: ({ context }, path) => {
     const operationType = context[TANSTACK_QUERY_OPERATION_CONTEXT_SYMBOL]?.type
 
-    if (operationType === 'query' || operationType === 'streamed' || operationType === 'infinite') {
+    if (operationType && GET_OPERATION_TYPE.has(operationType)) {
       return 'GET'
     }
 

packages/tanstack-query/src/live-query.test.ts

@@ -0,0 +1,84 @@
+import { queryClient } from '../tests/shared'
+import { experimental_liveQuery as liveQuery } from './live-query'
+
+beforeEach(() => {
+  queryClient.clear()
+})
+
+describe('liveQuery', async () => {
+  it('works', async () => {
+    const queryFn = liveQuery(async function* () {
+      yield 1
+      await new Promise(resolve => setTimeout(resolve, 50))
+      yield 2
+      await new Promise(resolve => setTimeout(resolve, 50))
+      yield 3
+    })
+
+    const controller = new AbortController()
+
+    const resultPromise = expect(queryFn({
+      queryKey: ['live-query'],
+      signal: controller.signal,
+      client: queryClient,
+    } as any)).resolves.toEqual(3)
+
+    await vi.waitFor(() => {
+      expect(queryClient.getQueryData(['live-query'])).toEqual(1)
+    })
+
+    await vi.waitFor(() => {
+      expect(queryClient.getQueryData(['live-query'])).toEqual(2)
+    })
+
+    await resultPromise
+  })
+
+  it('on abort signal', async () => {
+    let cleanupCalled = false
+
+    const queryFn = liveQuery(async function* () {
+      try {
+        yield 1
+        await new Promise(resolve => setTimeout(resolve, 50))
+        yield 2
+        await new Promise(resolve => setTimeout(resolve, 50))
+        yield 3
+      }
+      finally {
+        cleanupCalled = true
+      }
+    })
+
+    const controller = new AbortController()
+
+    const resultPromise = expect(queryFn({
+      queryKey: ['live-query'],
+      signal: controller.signal,
+      client: queryClient,
+    } as any)).resolves.toEqual(1)
+
+    await vi.waitFor(() => {
+      expect(queryClient.getQueryData(['live-query'])).toEqual(1)
+    })
+
+    controller.abort()
+
+    await resultPromise
+    expect(cleanupCalled).toBe(true)
+  })
+
+  it('throw if no data yielded', async () => {
+    const queryFn = liveQuery(async function* () {
+      // No yield
+    })
+
+    await expect(queryFn({
+      queryKey: ['live-query'],
+      signal: new AbortController().signal,
+      client: queryClient,
+    } as any)).rejects.toThrowError(
+      'Live query for ["live-query"] did not yield any data. Ensure the query function returns an AsyncIterable with at least one chunk.',
+    )
+  })
+})

packages/tanstack-query/src/live-query.ts

@@ -0,0 +1,34 @@
+import type { Promisable } from '@orpc/shared'
+import type { QueryFunction, QueryFunctionContext, QueryKey } from '@tanstack/query-core'
+import { stringifyJSON } from '@orpc/shared'
+
+export function experimental_liveQuery<
+  TQueryFnData = unknown,
+  TQueryKey extends QueryKey = QueryKey,
+>(
+  queryFn: (
+    context: QueryFunctionContext<TQueryKey>,
+  ) => Promisable<AsyncIterable<TQueryFnData>>,
+): QueryFunction<TQueryFnData, TQueryKey> {
+  return async (context) => {
+    const stream = await queryFn(context)
+    let last: { chunk: TQueryFnData } | undefined
+
+    for await (const chunk of stream) {
+      if (context.signal.aborted) {
+        break
+      }
+
+      last = { chunk }
+      context.client.setQueryData<TQueryFnData>(context.queryKey, chunk)
+    }
+
+    if (!last) {
+      throw new Error(
+        `Live query for ${stringifyJSON(context.queryKey)} did not yield any data. Ensure the query function returns an AsyncIterable with at least one chunk.`,
+      )
+    }
+
+    return last.chunk
+  }
+}

packages/tanstack-query/src/procedure-utils.angular.test-d.ts

@@ -158,6 +158,72 @@ describe('ProcedureUtils', () => {
     })
   })
 
+  describe('.liveOptions', () => {
+    describe('injectQuery', () => {
+      it('without args', () => {
+        const query = injectQuery(() => streamUtils.experimental_liveOptions())
+        expectTypeOf(query.data()).toEqualTypeOf<UtilsOutput[number] | undefined>()
+        expectTypeOf(query.error()).toEqualTypeOf<UtilsError | null>()
+      })
+
+      it('can infer errors inside options', () => {
+        const query = injectQuery(() => streamUtils.experimental_liveOptions({
+          throwOnError(error) {
+            expectTypeOf(error).toEqualTypeOf<UtilsError>()
+            return false
+          },
+        }))
+        expectTypeOf(query.data()).toEqualTypeOf<UtilsOutput[number] | undefined>()
+        expectTypeOf(query.error()).toEqualTypeOf<UtilsError | null>()
+      })
+
+      it('with initial data & select', () => {
+        const query = injectQuery(() => streamUtils.experimental_liveOptions({
+          select: data => ({ mapped: data }),
+          initialData: { title: 'title' },
+        }))
+
+        expectTypeOf(query.data()).toEqualTypeOf<{ mapped: UtilsOutput[number] }>()
+        expectTypeOf(query.error()).toEqualTypeOf<UtilsError | null>()
+      })
+    })
+
+    it('injectQueries', () => {
+      const queries = injectQueries({
+        queries: signal([
+          streamUtils.experimental_liveOptions(),
+          streamUtils.experimental_liveOptions({
+            input: { search: 'search' },
+            context: { batch: true },
+          }),
+          streamUtils.experimental_liveOptions({
+            select: data => ({ mapped: data }),
+          }),
+        ]),
+      })
+
+      // @ts-expect-error - TODO: fix this, injectQueries not working
+      expectTypeOf(queries()[0].data).toEqualTypeOf<UtilsOutput[number] | undefined>()
+      // @ts-expect-error - TODO: fix this, injectQueries not work at all
+      expectTypeOf(queries()[1].data).toEqualTypeOf<UtilsOutput[number] | undefined>()
+      expectTypeOf(queries()[2].data).toEqualTypeOf<{ mapped: UtilsOutput[number] } | undefined>()
+
+      // @ts-expect-error - TODO: fix this, injectQueries not work at all
+      expectTypeOf(queries()[0].error).toEqualTypeOf<null | UtilsError>()
+      // @ts-expect-error - TODO: fix this, injectQueries not work at all
+      expectTypeOf(queries()[1].error).toEqualTypeOf<null | UtilsError>()
+      expectTypeOf(queries()[2].error).toEqualTypeOf<null | UtilsError>()
+    })
+
+    it('fetchQuery', () => {
+      expectTypeOf(
+        queryClient.fetchQuery(streamUtils.experimental_liveOptions()),
+      ).toEqualTypeOf<
+        Promise<UtilsOutput[number]>
+      >()
+    })
+  })
+
   describe('.infiniteOptions', () => {
     const getNextPageParam: GetNextPageParamFunction<number, UtilsOutput> = () => 1
     const initialPageParam = 1

packages/tanstack-query/src/procedure-utils.react.test-d.ts

@@ -253,6 +253,120 @@ describe('ProcedureUtils', () => {
     })
   })
 
+  describe('.liveOptions', () => {
+    describe('useQuery', () => {
+      it('without args', () => {
+        const query = useQuery(streamUtils.experimental_liveOptions())
+        expectTypeOf(query.data).toEqualTypeOf<UtilsOutput[number] | undefined>()
+        expectTypeOf(query.error).toEqualTypeOf<UtilsError | null>()
+      })
+
+      it('can infer errors inside options', () => {
+        const query = useQuery(streamUtils.experimental_liveOptions({
+          throwOnError(error) {
+            expectTypeOf(error).toEqualTypeOf<UtilsError>()
+            return false
+          },
+        }))
+        expectTypeOf(query.data).toEqualTypeOf<UtilsOutput[number] | undefined>()
+        expectTypeOf(query.error).toEqualTypeOf<UtilsError | null>()
+      })
+
+      it('with initial data & select', () => {
+        const query = useQuery(streamUtils.experimental_liveOptions({
+          select: data => ({ mapped: data }),
+          initialData: { title: 'title' },
+        }))
+
+        expectTypeOf(query.data).toEqualTypeOf<{ mapped: UtilsOutput[number] }>()
+        expectTypeOf(query.error).toEqualTypeOf<UtilsError | null>()
+      })
+    })
+
+    describe('useSuspenseQuery', () => {
+      it('without args', () => {
+        const query = useSuspenseQuery(streamUtils.experimental_liveOptions())
+        expectTypeOf(query.data).toEqualTypeOf<UtilsOutput[number]>()
+        expectTypeOf(query.error).toEqualTypeOf<UtilsError | null>()
+      })
+
+      it('can infer errors inside options', () => {
+        const query = useSuspenseQuery(streamUtils.experimental_liveOptions({
+          throwOnError(error) {
+            expectTypeOf(error).toEqualTypeOf<UtilsError>()
+            return false
+          },
+        }))
+
+        expectTypeOf(query.data).toEqualTypeOf<UtilsOutput[number]>()
+        expectTypeOf(query.error).toEqualTypeOf<UtilsError | null>()
+      })
+
+      it('with select', () => {
+        const query = useSuspenseQuery(streamUtils.experimental_liveOptions({
+          select: data => ({ mapped: data }),
+        }))
+
+        expectTypeOf(query.data).toEqualTypeOf<{ mapped: UtilsOutput[number] }>()
+        expectTypeOf(query.error).toEqualTypeOf<UtilsError | null>()
+      })
+    })
+
+    it('useQueries', () => {
+      const queries = useQueries({
+        queries: [
+          streamUtils.experimental_liveOptions(),
+          streamUtils.experimental_liveOptions({
+            input: { search: 'search' },
+            context: { batch: true },
+          }),
+          streamUtils.experimental_liveOptions({
+            select: data => ({ mapped: data }),
+          }),
+        ],
+      })
+
+      expectTypeOf(queries[0].data).toEqualTypeOf<UtilsOutput[number] | undefined>()
+      expectTypeOf(queries[1].data).toEqualTypeOf<UtilsOutput[number] | undefined>()
+      expectTypeOf(queries[2].data).toEqualTypeOf<{ mapped: UtilsOutput[number] } | undefined>()
+
+      expectTypeOf(queries[0].error).toEqualTypeOf<null | UtilsError>()
+      expectTypeOf(queries[1].error).toEqualTypeOf<null | UtilsError>()
+      expectTypeOf(queries[2].error).toEqualTypeOf<null | UtilsError>()
+    })
+
+    it('useSuspenseQueries', () => {
+      const queries = useSuspenseQueries({
+        queries: [
+          streamUtils.experimental_liveOptions(),
+          streamUtils.experimental_liveOptions({
+            input: { search: 'search' },
+            context: { batch: true },
+          }),
+          streamUtils.experimental_liveOptions({
+            select: data => ({ mapped: data }),
+          }),
+        ],
+      })
+
+      expectTypeOf(queries[0].data).toEqualTypeOf<UtilsOutput[number]>()
+      expectTypeOf(queries[1].data).toEqualTypeOf<UtilsOutput[number]>()
+      expectTypeOf(queries[2].data).toEqualTypeOf<{ mapped: UtilsOutput[number] }>()
+
+      expectTypeOf(queries[0].error).toEqualTypeOf<null | UtilsError>()
+      expectTypeOf(queries[1].error).toEqualTypeOf<null | UtilsError>()
+      expectTypeOf(queries[2].error).toEqualTypeOf<null | UtilsError>()
+    })
+
+    it('fetchQuery', () => {
+      expectTypeOf(
+        queryClient.fetchQuery(streamUtils.experimental_liveOptions()),
+      ).toEqualTypeOf<
+        Promise<UtilsOutput[number]>
+      >()
+    })
+  })
+
   describe('.infiniteOptions', () => {
     const getNextPageParam: GetNextPageParamFunction<number, UtilsOutput> = () => 1
     const initialPageParam = 1