@beignet/react-query

Runtime: Beignet requires Node.js 22.12 or newer. Bun is optional.

TanStack Query integration for Beignet

This package provides options-first TanStack Query integration that's automatically typed based on your contracts. Generate queryOptions, mutationOptions, and infiniteQueryOptions that work with all TanStack Query primitives (useQuery, useMutation, useInfiniteQuery, prefetchQuery, fetchQuery, etc.) with full type inference.

Installation

npm install @beignet/react-query @beignet/core @tanstack/react-query react
Copy

TypeScript requirements

This package requires TypeScript 5.0 or higher for proper type inference.

Agent skills

This package ships a TanStack Intent skill for coding agents: @beignet/react-query#client. Load it when adding typed client setup, React Query options, infinite queries, feature client helpers, hooks, cache keys, cache invalidation, server prefetching, idempotent mutations, or client/server boundary fixes in a Beignet app.

Setup

1. Create your app client helpers

import { createClient } from "@beignet/core/client";
import { createReactQuery } from "@beignet/react-query";
import { QueryClient } from "@tanstack/react-query";

export const apiClient = createClient({
  baseUrl: "https://api.example.com",
  headers: async () => ({
    Authorization: `Bearer ${getToken()}`,
  }),
});

export const rq = createReactQuery(apiClient);

export function makeQueryClient() {
  return new QueryClient();
}
Copy

2. Set up QueryClientProvider

// app/providers.tsx
import { QueryClientProvider } from "@tanstack/react-query";
import { type ReactNode, useState } from "react";
import { makeQueryClient } from "@/client";

export function Providers({ children }: { children: ReactNode }) {
  const [queryClient] = useState(() => makeQueryClient());

  return (
    <QueryClientProvider client={queryClient}>
      {children}
    </QueryClientProvider>
  );
}
Copy

3. Use contracts in components

Import the app's shared rq adapter and browser-safe feature contracts. Pass options directly to TanStack Query; invalidates selects the queries to refresh after a successful mutation:

"use client";

import { useMutation, useQuery } from "@tanstack/react-query";
import { rq } from "@/client";
import { createTodo, listTodos } from "@/features/todos/contracts";

export function TodoList() {
  const todos = useQuery(rq(listTodos).queryOptions());
  const create = useMutation(
    rq(createTodo).mutationOptions({
      invalidates: () => [rq(listTodos).contractFilter()],
    }),
  );

  if (todos.isPending) return <p>Loading todos…</p>;
  if (todos.isError) return <p role="alert">{todos.error.message}</p>;

  return (
    <div>
      <button
        type="button"
        disabled={create.isPending}
        onClick={() => create.mutate({ body: { title: "New todo" } })}
      >
        Add todo
      </button>
      {create.isError && <p role="alert">{create.error.message}</p>}
      <ul>
        {todos.data.items.map((todo) => <li key={todo.id}>{todo.title}</li>)}
      </ul>
    </div>
  );
}
Copy

Pass lifecycle callbacks inside mutationOptions({ ... }). Beignet awaits onSuccess, then the selected invalidations before the mutation settles. Declarative invalidation requires TanStack React Query 5.102.0 or newer in v5. See Declarative mutation invalidation.

Keep adapter setup in root client/. Extract feature client helpers for shared query defaults, pagination, optimistic updates, or cache filters used by mutations and broadcasts.

Usage

Options-first API (recommended)

The options-first API generates options objects that can be used with any TanStack Query primitive.

Query options

import { useQuery } from "@tanstack/react-query";
import { rq } from "@/client";
import { getTodo } from "@/features/todos/contracts";

function TodoDetail({ id }: { id: string }) {
  // Generate query options
  const todoQuery = rq(getTodo).queryOptions({
    path: { id },
    staleTime: 30_000,
  });

  // Use with any TanStack Query primitive
  const { data, isLoading, error } = useQuery(todoQuery);

  if (isLoading) return <div>Loading...</div>;
  if (error) return <div>Error: {error.message}</div>;

  return (
    <div>
      <h1>{data.title}</h1>
      <p>Completed: {data.completed ? "Yes" : "No"}</p>
    </div>
  );
}
Copy

Prefetching

import { useQueryClient } from "@tanstack/react-query";
import { rq } from "@/client";
import { getTodo } from "@/features/todos/contracts";

function TodoList() {
  const queryClient = useQueryClient();

  const handleHover = (id: string) => {
    // Prefetch on hover using queryOptions
    const todoQuery = rq(getTodo).queryOptions({
      path: { id },
    });
    
    queryClient.prefetchQuery(todoQuery);
  };

  return <div>{/* ... */}</div>;
}
Copy

Server-side data fetching

import { dehydrate, HydrationBoundary } from "@tanstack/react-query";
import { makeQueryClient, rq } from "@/client";
import { getTodo } from "@/features/todos/contracts";

export async function TodoPage({ params }: { params: { id: string } }) {
  const queryClient = makeQueryClient();

  // Fetch on the server
  const todoQuery = rq(getTodo).queryOptions({
    path: { id: params.id },
  });

  await queryClient.prefetchQuery(todoQuery);

  return (
    <HydrationBoundary state={dehydrate(queryClient)}>
      <TodoDetail id={params.id} />
    </HydrationBoundary>
  );
}
Copy

That prefetch path uses the generated HTTP query function. In Next.js apps, ordinary { contract, useCase } routes can avoid an internal HTTP request by using a cached server-context helper plus a server-only prefetch helper that preserves the contract query key and replaces only the server queryFn:

import { getTodoUseCase } from "@/features/todos/use-cases";
import { getAppRequestContext } from "@/lib/server-context";
import { serverUseCaseQueryOptions } from "@/lib/server-react-query";

const ctx = await getAppRequestContext();
const queryClient = makeQueryClient();

await queryClient.prefetchQuery(
  serverUseCaseQueryOptions(
    rq(getTodo).queryOptions({ path: { id: params.id } }),
    getTodoUseCase,
    ctx,
    { id: params.id },
  ),
);
Copy

Use the use-case path only when the use case output is the contract success body. Use the HTTP rq(...) path when the route handler owns response mapping, headers, streaming, or other HTTP-layer behavior.

Mutation options

import { useMutation, useQueryClient } from "@tanstack/react-query";
import { rq } from "@/client";
import { createTodo, listTodos } from "@/features/todos/contracts";

function CreateTodoForm() {
  const queryClient = useQueryClient();

  // Generate mutation options
  const createTodoMutation = rq(createTodo).mutationOptions({
    onSuccess: (data, vars, onMutateResult, context) => {
      // Invalidate every cached listTodos call.
      rq(listTodos).invalidate(queryClient);
    },
  });

  const mutation = useMutation(createTodoMutation);

  const handleSubmit = (data: { title: string }) => {
    mutation.mutate({ body: data });
  };

  return (
    <form onSubmit={handleSubmit}>
      {/* form fields */}
      <button disabled={mutation.isPending}>
        {mutation.isPending ? "Creating..." : "Create Todo"}
      </button>
    </form>
  );
}
Copy

Infinite query options

For contracts that follow the Beignet cursor pagination convention — an optional cursor query param and a PageResult response with page.nextCursor — spread cursorPagination() into the options:

import { useInfiniteQuery } from "@tanstack/react-query";
import { cursorPagination } from "@beignet/react-query";
import { rq } from "@/client";
import { listTodos } from "@/features/todos/contracts";

function InfiniteTodoList() {
  const todosInfiniteQuery = rq(listTodos).infiniteQueryOptions({
    query: { status: "open" },
    ...cursorPagination(),
  });

  const { data, fetchNextPage, hasNextPage, isFetchingNextPage } =
    useInfiniteQuery(todosInfiniteQuery);

  return (
    <div>
      {data?.pages.map((page) =>
        page.items.map((todo) => <div key={todo.id}>{todo.title}</div>)
      )}
      
      <button
        onClick={() => fetchNextPage()}
        disabled={!hasNextPage || isFetchingNextPage}
      >
        {isFetchingNextPage ? "Loading more..." : "Load More"}
      </button>
    </div>
  );
}
Copy

data.pages is typed per page: the infinite options observe InfiniteData of the contract response. Contracts without a cursor query param or without page.nextCursor in the response fail to typecheck at the spread site.

Contracts that paginate differently pass initialPageParam, getNextPageParam, and page(...) by hand:

const todosInfiniteQuery = rq(listTodos).infiniteQueryOptions({
  query: { limit: 10 },
  initialPageParam: 0,
  page: ({ pageParam }) => ({
    query: { offset: pageParam },
  }),
  getNextPageParam: (lastPage) =>
    lastPage.page.hasMore
      ? lastPage.page.offset + lastPage.items.length
      : undefined,
});
Copy

Typed cache access

Use cacheEntries(queryClient, match?) to inspect cached standard queries with contract-derived response and parameter types:

const todos = rq(listTodos);
const entries = todos.cacheEntries(queryClient);

for (const { queryKey, params, data } of entries) {
  // data is the raw listTodos response, or undefined before data is available.
  console.log(queryKey, params, data?.items);
}
Copy

match uses the same parameter-prefix matching as filter(...). Path, query, body, and header objects accept a subset of their top-level fields. For a contract with both workspaceId and id path parameters, pass { path: { workspaceId } } to match every ID in that workspace. Actual query and mutation requests still require complete parameters.

Returned parameters describe normalized cache identity, not complete request arguments: nullish path/query fields are omitted, bodies use their JSON representation, and headers contain only lowercased names explicitly included through keyHeaders. Parameter objects are copied so editing them cannot change a query key.

Update matching existing entries with updateCachedQueries:

todos.updateCachedQueries(queryClient, {
  update: ({ data }) => {
    if (!data) return undefined;
    return {
      ...data,
      items: data.items.map((todo) =>
        todo.id === updatedTodo.id ? { ...todo, ...updatedTodo } : todo,
      ),
    };
  },
});
Copy

The updater receives { queryKey, params, data } for each entry. It runs synchronously against current data, must update immutably, and leaves an entry alone when it returns undefined. The helper does not create missing entries. Applications own list membership, sorting, pagination, optimistic rollback, and coordination with in-flight requests.

Both helpers recognize standard queries using the contract's generated keys, including entries seeded with setQueryData(helper.key(...), data). Store only the contract's raw response under these keys. Data is read before an observer's select transformation. Infinite queries and custom keys are excluded.

Standard query keys keep their existing format. Generated infinite-query keys append an "infinite" segment, so they remain distinct when manually seeded or dehydrated and hydrated. Use the queryKey returned by infiniteQueryOptions for direct reads, writes, and exact matching of infinite data. Namespace, contract, and parameter-prefix filters still match both query types.

When upgrading, discard persisted infinite-query caches created with the old key format. The "beignet" prefix is reserved: passing a custom key with that prefix throws. Omit key to use a generated key, and use ordinary TanStack cache APIs for custom keys and infinite-query data.

Beignet does not add query metadata. Global defaults, query-key defaults, and explicit meta options retain TanStack's normal precedence.

Declarative mutation invalidation

Pass invalidates to mutationOptions to select affected queries from the successful response and mutation variables:

const mutation = useMutation(
  rq(updateTodo).mutationOptions({
    invalidates: (_todo, variables) => [
      rq(getTodo).filter({ path: variables.path }),
      rq(listTodos).contractFilter(),
    ],
    onSuccess: (todo) => {
      // Optional application work runs before invalidation.
      console.log("Saved", todo.id);
    },
  }),
);
Copy

The callback returns an array of ordinary TanStack invalidation filters and may be asynchronous. An empty array requests no invalidation. Put shared filter selection in a feature client function when mutations and realtime events should refresh the same queries.

Beignet uses the executing mutation's QueryClient, waits for its configured onSuccess, then attempts and awaits every requested invalidation before onSettled. Client-wide and mutation-key success defaults are honored unless overridden in mutationOptions. Success defaults are captured when the generated mutation function first starts, and retained across retries; changes to defaults while that request is pending apply to subsequent invocations.

Generated options can be composed and registered with setMutationDefaults. When composition retains an earlier success wrapper, each wrapper runs its invalidation once per mutation invocation, including when defaults refer back to the composed options.

Default TanStack behavior marks matching queries stale and refetches active eligible queries. Failed requests do not run invalidates.

Invalidation still runs when the configured onSuccess throws after a successful request. Callback or invalidation failures follow TanStack's post-success error handling and do not retry the successful HTTP write; multiple failures are reported as an AggregateError. Normal TanStack refetch error behavior remains in effect.

Pass lifecycle callbacks inside mutationOptions({ ... }). Spreading the returned options and replacing onSuccess replaces the invalidation wrapper. Callbacks passed to mutate(variables, { onSuccess }) keep TanStack's normal per-call behavior and run after the configured success handling.

These helpers require TanStack React Query 5.102.0 or newer in the v5 line. The executing client comes from TanStack's mutation callback context. They do not add write locks or defer polling and realtime refreshes during mutations.

API reference

createReactQuery(client, options?)

Creates a React Query adapter factory.

const rq = createReactQuery(apiClient);
Copy

Headers are excluded from generated query keys by default because persisted caches would otherwise store credentials such as Authorization tokens. When a non-secret request-shaping header changes response data, opt that specific header into keys with keyHeaders:

const rq = createReactQuery(apiClient, {
  keyHeaders: ["X-Workspace-Preview"],
});

rq(listTodos).queryOptions({
  headers: { "X-Workspace-Preview": "workspace-1", Authorization: "Bearer ..." },
});
// queryKey: ["beignet", "todos", "listTodos", "GET /todos",
//   { headers: { "x-workspace-preview": "workspace-1" } }]
Copy

Only whitelisted header names are included. Names are matched case-insensitively and stored lowercased in the key. Never whitelist credential headers; pass a per-call key when you need a fully custom key instead.

rq(contract)

Creates a contract helper with query/mutation options methods.

const helper = rq(getTodo);
Copy

helper.queryOptions(options)

[Recommended] Generate query options that can be passed to any TanStack Query primitive.

const queryOpts = helper.queryOptions({
  path: { ... },            // Required when the contract has required path params
  query?: { ... },          // Required when the contract has required query params
  body?: { ... },           // Required when the contract has a required body
  headers?: { ... },        // Sent with the request; keyed only via keyHeaders
  key?: readonly unknown[], // Custom query key
  // ...any other TanStack Query options (staleTime, enabled, select, etc.)
});

// Use with any primitive
useQuery(queryOpts);
queryClient.prefetchQuery(queryOpts);
queryClient.fetchQuery(queryOpts);
queryClient.ensureQueryData(queryOpts);
Copy

select threads through the options, so the observed data narrows while the generated queryFn still returns and caches the raw contract response:

const { data } = useQuery(
  rq(getTodo).queryOptions({
    path: { id: "123" },
    select: (todo) => todo.title,
  }),
);
// data is typed as: string | undefined
Copy

helper.mutationOptions(options)

[Recommended] Generate mutation options that can be passed to useMutation.

const mutationOpts = helper.mutationOptions({
  // ...any TanStack Query mutation options (onSuccess, onError, retry, etc.)
});

useMutation(mutationOpts);
Copy

For contracts with idempotency metadata, the generated mutationFn derives one idempotency key per variables object and keeps it stable across TanStack retry attempts, so retried requests replay instead of re-executing. Fresh variables objects passed to separate mutate(...) calls get distinct keys. Retry state is keyed by variables-object identity, so reusing the exact same object for a later intentional mutation also reuses its key. Pass idempotencyKey in the mutation variables when multiple invocations should be one logical command:

mutation.mutate({ body: { title: "New todo" }, idempotencyKey: key });
Copy

Caveat: calling mutate() with no variables skips per-invocation key derivation and the client generates a fresh key per attempt. Pass a variables object when an idempotent mutation should keep its key across retries.

helper.infiniteQueryOptions(options)

[Recommended] Generate infinite query options for pagination. The options observe InfiniteData of the contract response, so useInfiniteQuery data has typed pages, and select narrows the observed data the same way as queryOptions.

For contracts that follow the cursor convention, spread cursorPagination():

const infiniteOpts = rq(listTodos).infiniteQueryOptions({
  query: { status: "open" },
  ...cursorPagination(),
});

useInfiniteQuery(infiniteOpts);
Copy

For other pagination shapes, pass the page-param options by hand:

const infiniteOpts = helper.infiniteQueryOptions({
  path?: { ... },          // Static path params included in the cache key
  query?: { ... },         // Static query params included in the cache key
  body?: { ... },          // Static body fields included in the cache key
  page?: (ctx: { pageParam }) => ({
    path?: { ... },        // Page-specific path params
    query?: { ... },       // Page-specific query params (cursor, offset, etc.)
    body?: { ... },        // Page-specific body fields (cursor, offset, etc.)
  }),
  initialPageParam: ...,
  getNextPageParam: (lastPage) => ...,
  getPreviousPageParam: (firstPage) => ...,
  key?: readonly unknown[], // Optional custom query key
  // ...any other TanStack Query infinite query options
});

useInfiniteQuery(infiniteOpts);
Copy

For body-paginated contracts, such as a POST search endpoint, keep the stable filters in the static body and put the cursor in page(...).body. The static body is part of the cache key, and each page request sends the merged body:

const searchOpts = rq(searchTodos).infiniteQueryOptions({
  body: { term: "beignet" },
  page: ({ pageParam }) => ({
    body: { cursor: pageParam ?? null },
  }),
  initialPageParam: null as string | null,
  getNextPageParam: (lastPage) => lastPage.page.nextCursor ?? undefined,
});
Copy

If you need fully dynamic params, pass a custom key and use params(...) instead:

const infiniteOpts = helper.infiniteQueryOptions({
  key: ["todos", { status: "open" }],
  params: ({ pageParam }) => ({
    query: { status: "open", cursor: pageParam ?? null },
  }),
  initialPageParam: null,
  getNextPageParam: (lastPage) => lastPage.page.nextCursor ?? undefined,
});
Copy

cursorPagination()

Infinite-query options for contracts that follow the Beignet cursor pagination convention: an optional cursor query param on the request and a PageResult-style response with page.nextCursor (string | null). Spread the result into infiniteQueryOptions(...):

import { cursorPagination } from "@beignet/react-query";

useInfiniteQuery(
  rq(listTodos).infiniteQueryOptions({
    query: { status: "open" },
    ...cursorPagination(),
  }),
);
Copy

The first page is fetched without a cursor, each next page sends lastPage.page.nextCursor, and null tells TanStack Query there is no next page. Type safety is structural: contracts without a cursor query param or without page.nextCursor in the response fail to typecheck at the spread site.

helper.namespaceKey()

Generate a namespace-level key for broad cache operations. Contracts created from defineContractGroup().namespace("todos") use that namespace. Non-namespaced contracts use null so they cannot collide with a real namespace string.

helper.namespaceKey(); // ["beignet", "todos"]
Copy

helper.contractKey()

Generate a contract-level key without path, query, or body params. The key includes the contract route ("GET /todos/:id"), so two contracts with the same derived local name but different routes — for example /v1/todos and /v2/todos list contracts — never share a cache key.

helper.contractKey(); // ["beignet", "todos", "getTodo", "GET /todos/:id"]
Copy

The route segment is also exposed as helper.route ("GET /todos/:id").

helper.key(params?)

Generate a stable query key for cache operations.

helper.key();
// ["beignet", "todos", "getTodo", "GET /todos/:id"]
helper.key({ path: { id: "1" } });
// ["beignet", "todos", "getTodo", "GET /todos/:id", { path: { id: "1" } }]
Copy

key(...) normalizes transport params the same way the client serializes them: null and undefined path, query, and opted-in header entries are omitted, while meaningful values like 0, false, and empty strings are preserved. JSON request bodies have different semantics: null, empty objects, and null-valued object properties stay in the key, while an undefined body and undefined object properties are omitted. This keeps distinct JSON requests from sharing a cache entry. Headers are excluded unless the adapter opted in with createReactQuery(client, { keyHeaders }).

helper.namespaceFilter(options?)

Generate TanStack Query filters for every contract in the helper's Beignet namespace.

queryClient.invalidateQueries(helper.namespaceFilter());
queryClient.invalidateQueries(helper.namespaceFilter({ stale: true }));
Copy

helper.contractFilter(options?)

Generate TanStack Query filters for every cached call to this contract.

queryClient.invalidateQueries(helper.contractFilter());
Copy

helper.filter(params?, options?)

Generate TanStack Query filters for one contract call or parameter prefix.

queryClient.invalidateQueries(
  helper.filter({ path: { id: "1" } }, { exact: true }),
);
Copy

These filters wrap plain TanStack Query keys, so prefix invalidation works as expected while TanStack Query still owns cache behavior:

queryClient.invalidateQueries(rq(getTodo).namespaceFilter());
rq(getTodo).invalidate(queryClient);
rq(getTodo).invalidate(queryClient, { path: { id: "1" } });
Copy

Use the smallest filter that matches the data you want to refresh:

helper.invalidate(queryClient, params?, options?)

Invalidate cached data for one contract using the helper's generated filters. With no params it invalidates every cached call to the contract. Pass params to target a detail key or parameter prefix.

await rq(listTodos).invalidate(queryClient);
await rq(getTodo).invalidate(queryClient, { path: { id: "1" } });
Copy

For list writes, invalidate the contract key so every filtered or paginated list result refreshes:

const createTodoMutation = useMutation(
  rq(createTodo).mutationOptions({
    onSuccess: () => {
      rq(listTodos).invalidate(queryClient);
    },
  }),
);
Copy

For detail writes, update or invalidate the exact detail key and then invalidate affected list contracts:

const updateTodoMutation = useMutation(
  rq(updateTodo).mutationOptions({
    onSuccess: (_todo, vars) => {
      rq(getTodo).invalidate(queryClient, { path: vars.path });
      rq(listTodos).invalidate(queryClient);
    },
  }),
);
Copy

helper.endpoint

Access the underlying endpoint for manual calls.

const data = await helper.endpoint.call({ path: { id: "123" } });
Copy

Type inference

All options are fully typed based on your contracts:

const { data } = useQuery(rq(getTodo).queryOptions({ path: { id: "123" } }));
// data is typed as: { id: string; title: string; completed: boolean }

const mutation = useMutation(rq(createTodo).mutationOptions());
mutation.mutate({ body: { title: "New Todo" } });
// body is typed as: { title: string; completed?: boolean }
Copy

Error handling

React Query errors are typed from the endpoint contract, including declared catalog errors and non-2xx response bodies:

const todo = rq(getTodo);
const { error } = useQuery(todo.queryOptions({ path: { id: "123" } }));

if (todo.endpoint.isError(error, { code: "TODO_NOT_FOUND" })) {
  console.log(error.details); // typed from the catalog details schema
} else if (error) {
  console.log(error.message);
}
Copy

Migration from hook-first API

If you were using the hook-first API, you can migrate to the options-first API:

Before (hook-first):

const { data } = rq(getTodo).useQuery({ path: { id } });
Copy

After (options-first):

const { data } = useQuery(rq(getTodo).queryOptions({ path: { id } }));
Copy

Hook wrappers have been removed—use the options-first API for all queries and mutations.

Related packages

• @beignet/core/contracts - Core contract definitions
• @beignet/core/client - HTTP client
• @beignet/nuqs - URL query state integration
• @beignet/react-hook-form - React Hook Form integration
• @beignet/react-form - TanStack Form integration

Broadcast query subscriptions

createBroadcastQuerySubscription({ client, channel, params, queryClient, invalidates, reconciles, onError?, onStatusChange? }) maps typed broadcasts to TanStack Query filters. invalidates(event) returns the filters affected by one hint; reconciles lists every query covered by the channel and runs after initial readiness and every reconnect. Use contract-derived contractFilter(), filter(params), or namespaceFilter() rather than hand-built query keys.

The helper marks matching queries stale, coalesces hints, and performs a later refetch if a hint arrives during an existing fetch. Active queries refetch; inactive queries remain stale even when an earlier fetch completes after the hint. Disabled and static queries retain TanStack's normal behavior. It returns unsubscribe() and getStatus(). Unsubscribe removes its cache listener and queued work without cancelling queries owned by other observers.

Share one createBroadcastClient from @beignet/core/broadcasting/client across features, then close and recreate it on user/workspace changes. Reconcile optimistic mutation state separately so a remote refetch does not overwrite an in-flight edit. See the broadcasting guide for complete channel and React Query examples.

Delay broadcast refreshes during saves

createBroadcastQuerySubscription already coordinates queries that are fetching. Its optional refreshGate also coordinates pending writes. The new createBroadcastMutationRefreshGate helper covers ordinary TanStack mutations; the new BroadcastRefreshGate interface supports application write queues. Existing subscriptions need no changes when this coordination is unnecessary.

For task queries, put the following in features/tasks/client/broadcasts.ts. This example assumes browser-safe taskChanges in features/tasks/channels.ts with a workspaceId parameter, and existing listTasks and saveTask contracts. listTasks accepts a workspaceId query parameter. Use the returned mutation options with useMutation, retaining your optimistic update/rollback callbacks:

import type { BroadcastClient } from "@beignet/core/broadcasting/client";
import {
  createBroadcastMutationRefreshGate,
  createBroadcastQuerySubscription,
} from "@beignet/react-query";
import type { QueryClient } from "@tanstack/react-query";
import { rq } from "@/client";
import { taskChanges } from "@/features/tasks/channels";
import { listTasks, saveTask } from "@/features/tasks/contracts";

export function saveTaskMutationOptions(workspaceId: string) {
  return rq(saveTask).mutationOptions({
    mutationKey: ["tasks", workspaceId, "save"],
  });
}

export function subscribeToTaskChanges(
  client: BroadcastClient,
  queryClient: QueryClient,
  workspaceId: string,
) {
  const taskQueries = rq(listTasks).filter({ query: { workspaceId } });
  return createBroadcastQuerySubscription({
    client, queryClient, channel: taskChanges, params: { workspaceId },
    invalidates: () => [taskQueries],
    reconciles: [taskQueries],
    refreshGate: createBroadcastMutationRefreshGate({
      queryClient,
      mutations: { mutationKey: ["tasks", workspaceId] },
    }),
  });
}
Copy

mutations accepts TanStack's mutationKey, exact, and predicate filters. Key matching is inclusive unless exact: true. Give every related write the same key prefix, including creates and deletes that affect lists or counters. Include workspace or resource scope when appropriate. Unrelated mutations do not block. Omitting a key/predicate deliberately selects all mutations.

The helper always checks pending mutations; status is not configurable. Paused pending mutations (offline or waiting for a mutation scope) also block, as do overlapping matching mutations until all settle. A custom predicate can exclude mutations, but excluding paused writes can refresh over unsaved edits. The gate has no timeout that overrides your blocking policy.

Combine mutations with an external write lock

Put this application helper beside the feature's broadcast query mappings. Pass createTaskWriteRefreshGate({ queryClient, workspaceId, writeLock }) as refreshGate in the subscription above. The external lock owns its state and notifies after changes; it may inspect the query key to block only related queries. Beignet owns listening, queuing, and cleanup:

import {
  type BroadcastRefreshGate,
  createBroadcastMutationRefreshGate,
} from "@beignet/react-query";
import type { Query, QueryClient } from "@tanstack/react-query";

// Application-owned write queue/lock. It can block different queries independently.
export interface TaskWriteLock {
  isBlocked(query: Query): boolean;
  subscribe(onChange: () => void): () => void;
}

export function createTaskWriteRefreshGate({
  queryClient, workspaceId, writeLock,
}: { queryClient: QueryClient; workspaceId: string; writeLock: TaskWriteLock }): BroadcastRefreshGate {
  const mutations = createBroadcastMutationRefreshGate({
    queryClient,
    mutations: { mutationKey: ["tasks", workspaceId] },
  });
  return {
    isBlocked: (query) => mutations.isBlocked(query) || writeLock.isBlocked(query),
    subscribe(onChange) {
      const stopMutations = mutations.subscribe(onChange);
      try {
        const stopLock = writeLock.subscribe(onChange);
        return () => { try { stopLock(); } finally { stopMutations(); } };
      } catch (error) {
        stopMutations();
        throw error;
      }
    },
  };
}
Copy

Both event invalidations and readiness/reconnect reconciliation queue matching existing queries, deduplicated per query. A blocked query is neither invalidated nor refetched by the subscription. The gate is subscribed before checking state to avoid a missed wakeup and checked again immediately before queued work begins. Opening it schedules active queries for refresh and marks inactive queries stale. TanStack's enabled/static rules still control fetching. If it closes again, unprocessed work stays queued. A hint during an existing fetch remains pending for a subsequent refresh after that fetch settles.

The gate controls broadcast-driven refreshes only. Polling, manual refresh, mutation callbacks that invalidate queries, already-running requests, and optimistic cache updates remain application-owned. If all refresh sources must respect a write lock, apply that policy to those sources too.

Unsubscribe removes transport, query-cache, and gate listeners, discards queued work, and prevents deferred callbacks from starting refreshes. It cannot undo an already-started request. Gate checks that throw report through onError, keep the query pending, and stay blocked until a later gate notification or broadcast hint retries the check. They do not poll or retry themselves. If subscribe throws, the subscription stays blocked: fix the cause, unsubscribe, and create a new subscription. Custom gates must release partially acquired listeners before throwing. Unsubscribe failures are reported while the other listeners are still released.

Diagnose connection changes

onSync(info) and onStatusChange(status, info) receive BroadcastConnectionInfo from @beignet/core/broadcasting/client. Callbacks may omit the metadata argument when they do not need it. createBroadcastQuerySubscription forwards status metadata and accepts an optional onSync(info) observer; its reconciliation is still scheduled even when that observer throws. This notification describes readiness, not completion of queued or gated query refreshes. Do not refetch from that observer to bypass the gate accidentally.

Readiness reports the reason for that physical connection; another observer joining an already-ready channel sees the same information. Reconnecting status reports the known reason for restarting. An elapsed lifetime alone never proves planned renewal. Servers send an explicit, connection-wide renewal control frame before closing normally. An unexplained EOF stays unknown. A renewal frame followed by a stall produces interruption when a watchdog fires. Further traffic after the frame makes a later EOF unknown, including an incomplete frame. Reconciliation continues after every renewal because events have no replay. Planned renewal is a lifecycle change, not an onError failure.

Routes use the existing broadcast instrumentation watcher for admission, connection, and cleanup events. Server closure details report planned-renewal, provider-disconnect, buffer-overflow, setup-failure, or unknown as known locally; the server cannot infer why a browser cancelled. An optional instrumentation sink on createBroadcastClient uses the existing ProviderInstrumentationPort to record broadcast.connecting, broadcast.ready, and broadcast.closed with browser reasons. These events contain safe counts, durations, and reasons, never headers, credentials, payloads, or channel params. Keep custom logging equally restrained.

License

MIT