Custom exchange debouncing events

[neoroma]

Hi, I am trying to achieve the following: calling a mutation with optimistic update without actually sending the mutation to the server untill [ the user activity is idle ], so basically sending mutation to the server only once for multiple mutations calls in a while. To achieve this I've configured cacheExchange optimistic and added my custom exchange just before the fetchExchange. The custom mutation using wonka operators looks pretty much like this:

export const debounceExchange = ({ threshold, mutationBodyStartString }: Options): Exchange => ({ forward, client }) => {
  return ops$ => {
    return pipe(
        ops$,
        debounce((o) => {
            if(o.query.loc.source.body.startsWith(mutationBodyStartString)) return threshold
            return 0
          }),
        forward
      )
  }
}

It works ok with only one event during the threshold, so I guess the amount of started operation is the same as the amount of completed ( guessing from the debug exchange ). But if the mutation is called more than once during the threshold all looks good and the request is sent once as expected, but the operation looks like never completed so the fetching is false at the end and data is undefined. Is there a way to make it work or is there some other nice way to handle the case ?

[arcanis]

We simultaneously opened threads with very similar use cases 😄 (#1255)

I've played a bit with an adaptation of this snippet from @kitten (Server | Client), it seems to kinda work with some caveats but I'm not entirely sure how... in particular:

• Thinking about it, I don't understand why debounce works - since all operations are evaluated into their own pipe (at least that's what I understood, otherwise I'd have expected ops$ to be provided with forward and client, not as a separate callback argument), how come the debounce context is shared? Shouldn't one new context be created for each time the ops$ function is called, thus turning the debounce into a kind of delay?
• While the debounce seems to work even with concurrent queries, dropped queries are kept in the pending mode (in the playground, it translates into the "(updating)" text being shown forever). I'm not sure what are the practical effects (probably at least a memory leak somewhere?). It's probably because there's something wrong in the teardown code, but not sure what.
• It's still not clear if a single mutation can have both debounced and non-debounced fields.

[JoviDeCroock]

It will be possible to debounce on certain fields but this is a bit of logic, this will be similar to how defer queries would work. You'll need an exchange that splits up your mutation into two distinct mutations, one for the prioritized field and one for the deferred fields.

This logic in itself is very custom since it also involves splitting out variables, ... but it should be possible.

If you are not relying on optimistic updates I'd advise doing this in the FE framework of your choice, let's look at an example for React in this case:

const changePrioritisedField = function(e) { mutate(e.target.value) };
const changeNormalField = debounce(function(e) { mutate(e.target.value) };)

const onChange = function(e) {
  setState(e.target.name, e.target.value);
  if (prioFields.includes(e.target.name) {
    changePrioritisedField(e);
  } else {
    changeNormalField(e);
  }
}

[arcanis]

Optimistic updates are important (at least in my case) because I use urql as the main data store whose multiple components derive their data. In the worst case though I still have the option to extract the debounced fields into their own mutation, which should solve this problem.

By any chance, do you have an idea why the (updating) tag doesn't disappear? Is my teardown code wrong?:

const filteredMutations$ = pipe(
  bufferedMutations$,
  mergeMap((buffer) => {
    for (const operation of buffer.slice(0, -1))
      client.dispatchOperation(makeOperation("teardown", operation));

    return fromValue(buffer[buffer.length - 1]);
  })
);

[kitten]

One thing worth mentioning is that all mutations need a result, no matter whether they've been removed by the denounce or not; otherwise they'll always be pending.

So my conclusion here is still the same. In the case of denouncing mutations in terms of UI interactions and other inputs, an abstraction at the UI level makes more sense. This could even be handled via a global store or denounce since mutations will be Frontend interactions

[arcanis]

Fair enough! I was still curious though, so as an exercise I tried to implement that exchange the best I could. Implementation is there, it should now work pretty well (it doesn't mark queries as forever pending anymore, plus a few other fixes). A bit more work would be needed to make it truly generic, but it should be a good start in case someone finds it useful 🙂

import {Exchange, makeOperation, Operation} from 'urql';

import {
  Source,
  fromValue,
  delay,
  mergeMap,
  merge,
  share,
  filter,
  pipe,
  fromArray,
  take,
  onPush,
  takeWhile,
} from 'wonka';

function race<T>(streams: Array<Source<T>>) {
  let selected$: Source<T> | null = null;

  const makeRace = (stream$: Source<T>) =>
    pipe(
      stream$,
      takeWhile(() => {
        return selected$ === null || selected$ === stream$;
      }),
      onPush(() => {
        selected$ = stream$;
      })
    );

  return merge(
    streams.map(stream$ => {
      return makeRace(stream$);
    })
  );
}

export const debounceExchange: () => Exchange = () => ({client, forward}) => {
  const debounceRegistry = new Map();

  const mutationFilter = (op: Operation) => {
    return op.kind === `mutation`;
  };

  return ops$ => {
    const sharedOps$ = share(ops$);

    const mutations$ = pipe(
      sharedOps$,
      filter(op => mutationFilter(op)),
      share,
    );

    const rest$ = pipe(
      sharedOps$,
      filter(op => !mutationFilter(op)),
    );

    const debouncedMutations$ = pipe(
      mutations$,
      onPush(op => {
        debounceRegistry.set(op.key, []);
      }),
      mergeMap(op =>
        pipe(
          race([
            pipe(mutations$, take(1)),
            pipe(fromValue(null), delay(1000)),
          ]),
          mergeMap((debouncedBy: Operation | null) => {
            if (!debouncedBy)
              return fromValue(op);

            // Since this request is debounced, anything that got debounced due
            // to this request must be "reparented" to the new primary mutation
            const debounced = debounceRegistry.get(op.key);
            debounceRegistry.get(debouncedBy.key).push(...debounced, op);

            // Don't forget to tell the client that noone cares about this query
            // anymore; otherwise urql seems to keep it inside a sort of cache
            // that prevents similar mutations from properly firing.
            client.dispatchOperation(makeOperation(`teardown`, op));

            return fromArray([]);
          })
        )
      )
    );

    const results$ = pipe(
      merge([
        debouncedMutations$,
        rest$,
      ]),
      forward,
    );

    return pipe(
      results$,
      mergeMap(result => {
        const debounced = debounceRegistry.get(result.operation.key);
        if (typeof debounced === `undefined`)
          return fromValue(result);

        debounceRegistry.delete(result.operation.key);

        // For each query that got debounced, we now return the primary
        // result. This way all queries will be marked as fulfilled.
        return fromArray([
          ...debounced.map(operation => ({
            ...result,
            operation,
          })),
          result,
        ]);
      })
    );
  };
};