How to skip cache for some operations

[FreeRiderBysik]

Context

On some page we use notifications implemented as URQL subscription. The user can receive many notifications (up to 50RPS). This data is not persistent (cursors, peer states, etc...)
I did not expect cache to be used for these notifications, but...

Every time I receive message from subscription it goes throw cache process: writes result to cache, and then GC deletes it because nothing have reference to this data.
This is not a performance issue, but a lot of useless work for clients.

Now I am looking for a way to not use cache-exchange at all for some operations.

My try

I tried to wrap cache-exchange with my own Exchange like this:

interface WrappedCacheExchangeOpts extends CacheExchangeOpts {
  cacheSkipOperations?: string[];
}

/** filter which operation will skip cache */
function filterOperationToSkipCache(
  operation: Operation,
  options?: WrappedCacheExchangeOpts
) {
  const operationName = getDocumentFieldName(operation.query);
  if (operationName && options?.cacheSkipOperations?.includes(operationName)) {
    return true;
  }
  return false;
}

/**
 * Wraps urql cache exchange, with ability to skip cache (request and response)
 *
 * To skip cache add operation name to options.cacheSkipOperations array
 */
function buildCacheExchangeWrapper(
  options: WrappedCacheExchangeOpts
): Exchange {
  const originalCacheExchange = cacheExchange(options);

  return (input) => {
    return (operations) => {
      // just forward without changes
      const forwardedOps = pipe(
        operations,
        filter((operation) => filterOperationToSkipCache(operation, options)),
        input.forward
      );

      // use cache pipe
      const cachedOps = pipe(
        operations,
        filter((operation) => !filterOperationToSkipCache(operation, options)),
        originalCacheExchange(input)
      );

      return merge([cachedOps, forwardedOps]);
    };
  };
}

But when i start application, i get error: Error: forward() must only be called once in each Exchange. on building exchanges. I understood, that i cannot just merge two pipes with forward.

Is there any way to achieve required behavior without changing exchange-graphcache?

[amannn]

@FreeRiderBysik Were you able to find a solution? I have the same situation and running into the same error.

My use case is that a component triggers a sequence of mutations and I want to only run the last one through the cache exchange to trigger invalidation. Otherwise, the cache would be invalidated with every request which would be very wasteful for my app. Using Graphcache is not really an option for my app, we're generally happy with the simple approach of document caching.

An alternative might be to debounce the cache invalidation, but that would require forking the cache exchange too as far as I understand.

[FreeRiderBysik]

@amannn, yes. My friend helped me. See solution below.

[amannn]

Thanks for sharing! I ended up forking the document cache and have added the capability for debounced invalidations:

import { isObject, isString } from 'lodash';
import {
  Client,
  ExchangeIO,
  ExchangeInput,
  Operation,
  OperationContext,
  OperationResult,
  formatDocument,
  makeOperation,
  makeResult
} from 'urql';
import {
  buffer,
  delay,
  filter,
  forEach,
  makeSubject,
  map,
  merge,
  pipe,
  scan,
  tap
} from 'wonka';

type ResultCache = Map<number, OperationResult>;
type OperationCache = Map<string, Set<number>>;

interface EntityLike {
  // eslint-disable-next-line @typescript-eslint/no-explicit-any -- urql uses any too
  [key: string]: EntityLike | Array<EntityLike> | any;
  __typename: string | null | void;
}

function collectTypes(obj: EntityLike | Array<EntityLike>, types: Set<string>) {
  if (Array.isArray(obj)) {
    obj.forEach(item => collectTypes(item, types));
  } else if (isObject(obj) && obj !== null) {
    Object.keys(obj).forEach(key => {
      if (key === '__typename' && isString(typeof obj[key])) {
        types.add(obj[key] as string);
      } else {
        collectTypes(obj[key], types);
      }
    });
  }
  return types;
}

/**
 * Finds and returns a list of `__typename` fields found in response data.
 */
export function collectTypenames(response: object): Array<string> {
  return [...collectTypes(response as EntityLike, new Set())];
}

/** Adds additional metadata to an `Operation`'s `context.meta` property while copying it.
 * @see {@link OperationDebugMeta} for more information on the {@link OperationContext.meta} property.
 */
export function addMetadata(
  operation: Operation,
  meta: OperationContext['meta']
) {
  return makeOperation(operation.kind, operation, {
    meta: {
      ...operation.context.meta,
      ...meta
    }
  });
}

function shouldSkip({ kind }: Operation) {
  return kind !== 'mutation' && kind !== 'query';
}

/** Adds unique typenames to query (for invalidating cache entries) */
export function mapTypeNames(operation: Operation): Operation {
  const query = formatDocument(operation.query);
  if (query !== operation.query) {
    const formattedOperation = makeOperation(operation.kind, operation);
    formattedOperation.query = query;
    return formattedOperation;
  } else {
    return operation;
  }
}

/** Reexecutes an `Operation` with the `network-only` request policy. */
function reexecuteOperation(client: Client, operation: Operation) {
  return client.reexecuteOperation(
    makeOperation(operation.kind, operation, {
      requestPolicy: 'network-only'
    })
  );
}

/**
 * A fork of the built-in document cache exchange from urql that adds the
 * capability on top that mutation responses don't invalidate queries
 * immediately, but only after all currently pending mutations have completed.
 *
 * @see https://github.com/urql-graphql/urql/blob/main/packages/core/src/exchanges/cache.ts
 */
export default function cacheExchangeDebounced({
  client,
  dispatchDebug,
  forward
}: ExchangeInput): ExchangeIO {
  // After a mutation response is received, wait for a short period
  // to see if the another mutation is triggered (asynchronously)
  const DEBOUNCE_DELAY = 50;

  const resultCache: ResultCache = new Map();
  const operationCache: OperationCache = new Map();

  function isOperationCached(operation: Operation) {
    return (
      operation.kind === 'query' &&
      operation.context.requestPolicy !== 'network-only' &&
      (operation.context.requestPolicy === 'cache-only' ||
        resultCache.has(operation.key))
    );
  }

  const invalidations = makeSubject<Operation>();
  const pendingMutationsTracker = makeSubject<'start' | 'complete'>();

  const idle$ = pipe(
    pendingMutationsTracker.source,
    scan((acc, value) => acc + (value === 'start' ? 1 : -1), 0),
    filter(value => value === 0),
    delay(DEBOUNCE_DELAY)
  );

  pipe(
    invalidations.source,
    buffer(idle$),
    forEach(value => {
      value.forEach(operation => {
        reexecuteOperation(client, operation);
      });
    })
  );

  return ops$ => {
    const cachedOps$ = pipe(
      ops$,
      filter(op => !shouldSkip(op) && isOperationCached(op)),
      map(operation => {
        const cachedResult = resultCache.get(operation.key);

        dispatchDebug({
          operation,
          ...(cachedResult
            ? {
                type: 'cacheHit',
                message: 'The result was successfully retried from the cache'
              }
            : {
                type: 'cacheMiss',
                message: 'The result could not be retrieved from the cache'
              })
        });

        let result: OperationResult =
          cachedResult ||
          makeResult(operation, {
            data: null
          });

        result = {
          ...result,
          operation: addMetadata(operation, {
            cacheOutcome: cachedResult ? 'hit' : 'miss'
          })
        };

        if (operation.context.requestPolicy === 'cache-and-network') {
          result.stale = true;
          reexecuteOperation(client, operation);
        }

        return result;
      })
    );

    const forwardedOps$ = pipe(
      merge([
        pipe(
          ops$,
          filter(op => !shouldSkip(op) && !isOperationCached(op)),
          map(mapTypeNames)
        ),
        pipe(
          ops$,
          filter(op => shouldSkip(op))
        )
      ]),
      map(op => addMetadata(op, { cacheOutcome: 'miss' })),
      filter(
        op => op.kind !== 'query' || op.context.requestPolicy !== 'cache-only'
      ),
      tap(operation => {
        if (operation.kind === 'mutation') {
          pendingMutationsTracker.next('start');
        }
      }),
      forward,
      tap(response => {
        let { operation } = response;
        if (!operation) return;

        let typenames = operation.context.additionalTypenames || [];
        // NOTE: For now, we only respect `additionalTypenames` from subscriptions to
        // avoid unexpected breaking changes
        // We'd expect live queries or other update mechanisms to be more suitable rather
        // than using subscriptions as “signals” to reexecute queries. However, if they’re
        // just used as signals, it’s intuitive to hook them up using `additionalTypenames`
        if (response.operation.kind !== 'subscription') {
          typenames = collectTypenames(response.data).concat(typenames);
        }

        // Invalidates the cache given a mutation's response
        if (
          response.operation.kind === 'mutation' ||
          response.operation.kind === 'subscription'
        ) {
          const pendingOperations = new Set<number>();

          dispatchDebug({
            type: 'cacheInvalidation',
            message: `The following typenames have been invalidated: ${typenames}`,
            operation,
            data: { typenames, response }
          });

          for (let i = 0; i < typenames.length; i++) {
            const typeName = typenames[i];
            let operations = operationCache.get(typeName);
            if (!operations)
              operationCache.set(typeName, (operations = new Set()));
            Array.from(operations.values()).forEach(key => {
              pendingOperations.add(key);
            });
            operations.clear();
          }

          Array.from(pendingOperations.values()).forEach(key => {
            if (resultCache.has(key)) {
              operation = (resultCache.get(key) as OperationResult).operation;
              resultCache.delete(key);
              invalidations.next(operation);
            }
          });
        } else if (operation.kind === 'query' && response.data) {
          resultCache.set(operation.key, response);
          for (let i = 0; i < typenames.length; i++) {
            const typeName = typenames[i];
            let operations = operationCache.get(typeName);
            if (!operations)
              operationCache.set(typeName, (operations = new Set()));
            operations.add(operation.key);
          }
        }

        if (response.operation.kind === 'mutation') {
          pendingMutationsTracker.next('complete');
        }
      })
    );

    return merge([cachedOps$, forwardedOps$]);
  };
}

Tests:

import {
  Client,
  ExchangeInput,
  GraphQLRequest,
  Operation,
  OperationContext,
  OperationResult,
  gql,
  makeOperation
} from 'urql';
import { beforeEach, expect, it, vi } from 'vitest';
import { Source, Subject, forEach, makeSubject, map, pipe } from 'wonka';
import cacheExchangeDebounced from './cache-exchange-debounced';

const context: OperationContext = {
  requestPolicy: 'network-only',
  url: 'http://localhost:3000/graphql'
};

const queryGql: GraphQLRequest = {
  key: 2,
  query: gql`
    query User {
      user {
        __typename
        id
        name
      }
    }
  `,
  variables: {}
};

const queryOperation: Operation = makeOperation(
  'query',
  {
    query: queryGql.query,
    variables: queryGql.variables,
    key: queryGql.key
  },
  context
);

const queryResponse: OperationResult = {
  operation: queryOperation,
  data: {
    user: {
      __typename: 'User',
      id: '1',
      name: 'John'
    }
  },
  stale: false,
  hasNext: false
};

export const mutationGql: GraphQLRequest = {
  key: 3,
  query: gql`
    mutation UpdateUser {
      updateUser(name: $name) {
        name
      }
    }
  `,
  variables: {
    name: 'Clara'
  }
};

export const mutationOperation: Operation = makeOperation(
  'mutation',
  {
    query: mutationGql.query,
    variables: mutationGql.variables,
    key: mutationGql.key
  },
  context
);

const mutationResponse: OperationResult = {
  operation: mutationOperation,
  data: {
    updateUser: {
      __typename: 'User',
      name: 'Clara'
    }
  },
  stale: false,
  hasNext: false
};

const responses = {
  [queryOperation.key]: queryResponse,
  [mutationOperation.key]: mutationResponse
};

function sleep(ms: number) {
  return new Promise(resolve => {
    setTimeout(resolve, ms);
  });
}

let exchangeArgs: ExchangeInput;
let forwardedOperations: Array<Operation>;
let input: Subject<Operation>;
let reexecuteOperation;
let dispatchDebug;
beforeEach(() => {
  forwardedOperations = [];
  input = makeSubject<Operation>();
  reexecuteOperation = vi.fn();
  dispatchDebug = vi.fn();

  // Collect all forwarded operations
  function forward(source: Source<Operation>) {
    return pipe(
      source,
      map(operation => {
        forwardedOperations.push(operation);
        return responses[operation.key];
      })
    );
  }

  const client = { reexecuteOperation } as unknown as Client;
  exchangeArgs = { forward, client, dispatchDebug };
});

it('invalidates after a single mutation', async () => {
  const { complete, next, source: ops$ } = input;
  const result = vi.fn();
  const exchange = cacheExchangeDebounced(exchangeArgs)(ops$);

  pipe(exchange, forEach(result));

  next(queryOperation);
  expect(forwardedOperations.length).toBe(1);
  expect(result).toHaveBeenCalledTimes(1);

  next(mutationOperation);

  expect(forwardedOperations.length).toBe(2);
  expect(result).toHaveBeenCalledTimes(2);

  await vi.waitFor(() => {
    expect(reexecuteOperation).toHaveBeenCalledTimes(1);
  });

  complete();
});

it('invalidates once after multiple mutations', async () => {
  const { complete, next, source: ops$ } = input;
  const result = vi.fn();
  const exchange = cacheExchangeDebounced(exchangeArgs)(ops$);

  pipe(exchange, forEach(result));

  next(queryOperation);
  expect(forwardedOperations.length).toBe(1);
  expect(result).toHaveBeenCalledTimes(1);

  next(mutationOperation);

  expect(forwardedOperations.length).toBe(2);
  expect(result).toHaveBeenCalledTimes(2);

  // In case a mutation response triggers another mutation, we should wait
  // for a brief moment since the next mutation might be triggered asynchronously
  await sleep(10);

  next(mutationOperation);
  expect(reexecuteOperation).toHaveBeenCalledTimes(0);

  expect(forwardedOperations.length).toBe(3);
  expect(result).toHaveBeenCalledTimes(3);

  await vi.waitFor(() => {
    expect(reexecuteOperation).toHaveBeenCalledTimes(1);
  });

  complete();
});

[FreeRiderBysik]

Here is a solution:

import { cacheExchange, CacheExchangeOpts } from '@urql/exchange-graphcache';
import {
  Exchange,
  ExchangeInput,
  ExchangeIO,
  Operation,
  OperationResult,
} from 'urql';
import { empty, filter, merge, pipe, Source } from 'wonka';
import { getDocumentFieldName } from './utils';

interface WrappedCacheExchangeOpts extends CacheExchangeOpts {
  cacheSkipOperations?: string[];
}

/** filter which operation will skip cache */
function filterOperationToSkipCache(
  operation: Operation,
  options?: WrappedCacheExchangeOpts
) {
  const operationName = getDocumentFieldName(operation.query);
  if (operationName && options?.cacheSkipOperations?.includes(operationName)) {
    return true;
  }
  return false;
}

interface ResultsRef {
  results: Source<OperationResult>;
}

function buildCacheExchangeWrapper(options: WrappedCacheExchangeOpts): Exchange {
  const originalCacheExchange = cacheExchange(options);
  return (input: ExchangeInput): ExchangeIO => {
    return (operations) => {
      const cacheSkippedOps = pipe(
        operations,
        filter((operation) => filterOperationToSkipCache(operation, options))
      );
      const cacheTargetOps = pipe(
        operations,
        filter((operation) => !filterOperationToSkipCache(operation, options))
      );
      const cacheSkipResults: ResultsRef = { results: empty };
      return merge([
        originalCacheExchange({
          ...input,
          forward: (forwardedFromCacheExchangeOps) => {
            const results = input.forward(
              merge([forwardedFromCacheExchangeOps, cacheSkippedOps])
            );
            cacheSkipResults.results = pipe(
              results,
              filter((result) =>
                filterOperationToSkipCache(result.operation, options)
              )
            );
            return pipe(
              results,
              filter(
                (result) =>
                  !filterOperationToSkipCache(result.operation, options)
              )
            );
          },
        })(cacheTargetOps),
        cacheSkipResults.results,
      ]);
    };
  };
}
export default buildCacheExchangeWrapper;

It can be really complicated to read, but it works fine.

[drc37]

How would you use this in a real-life example?