chore: generic data provider improvments (#1772)

libs/fills/src/lib/fills-data-provider.ts

@@ -4,10 +4,12 @@ import {
   makeDataProvider,
   makeDerivedDataProvider,
   defaultAppend as append,
+  paginatedCombineDelta as combineDelta,
+  paginatedCombineInsertionData as combineInsertionData,
 } from '@vegaprotocol/react-helpers';
 import type { Market } from '@vegaprotocol/market-list';
 import { marketsProvider } from '@vegaprotocol/market-list';
-import type { PageInfo } from '@vegaprotocol/react-helpers';
+import type { PageInfo, Edge } from '@vegaprotocol/react-helpers';
 import { FillsDocument, FillsEventDocument } from './__generated___/Fills';
 import type {
   FillsQuery,
@@ -55,10 +57,7 @@ const update = (
 };
 
 export type Trade = Omit<FillFieldsFragment, 'market'> & { market?: Market };
-export type TradeEdge = {
+export type TradeEdge = Edge<Trade>;
-  cursor: FillEdgeFragment['cursor'];
-  node: Trade;
-};
 
 const getData = (responseData: FillsQuery): FillEdgeFragment[] =>
   responseData.party?.tradesConnection?.edges || [];
@@ -100,20 +99,6 @@ export const fillsWithMarketProvider = makeDerivedDataProvider<
           },
         }
     ) || null,
-  (parts): Trade[] | undefined => {
+  combineDelta<Trade, ReturnType<typeof getDelta>['0']>,
-    if (!parts[0].isUpdate) {
+  combineInsertionData<Trade>
-      return;
-    }
-    // map FillsSub_trades[] from subscription to updated Trade[]
-    return (parts[0].delta as ReturnType<typeof getDelta>).map(
-      (deltaTrade) => ({
-        ...((parts[0].data as ReturnType<typeof getData>)?.find(
-          (trade) => trade.node.id === deltaTrade.id
-        )?.node as FillFieldsFragment),
-        market: (parts[1].data as Market[]).find(
-          (market) => market.id === deltaTrade.marketId
-        ),
-      })
-    );
-  }
 );

libs/orders/src/lib/components/order-data-provider/order-data-provider.ts

@@ -6,10 +6,12 @@ import {
   makeDataProvider,
   makeDerivedDataProvider,
   defaultAppend as append,
+  paginatedCombineDelta as combineDelta,
+  paginatedCombineInsertionData as combineInsertionData,
 } from '@vegaprotocol/react-helpers';
 import type { Market } from '@vegaprotocol/market-list';
 import { marketsProvider } from '@vegaprotocol/market-list';
-import type { PageInfo } from '@vegaprotocol/react-helpers';
+import type { PageInfo, Edge } from '@vegaprotocol/react-helpers';
 import type {
   Orders,
   Orders_party_ordersConnection_edges,
@@ -140,10 +142,7 @@ export const update = (
 export type Order = Omit<Orders_party_ordersConnection_edges_node, 'market'> & {
   market?: Market;
 };
-export type OrderEdge = {
+export type OrderEdge = Edge<Order>;
-  node: Order;
-  cursor: Orders_party_ordersConnection_edges['cursor'];
-};
 
 const getData = (
   responseData: Orders
@@ -169,7 +168,7 @@ export const ordersProvider = makeDataProvider({
 });
 
 export const ordersWithMarketProvider = makeDerivedDataProvider<
-  OrderEdge[],
+  (OrderEdge | null)[],
   Order[]
 >(
   [ordersProvider, marketsProvider],
@@ -183,20 +182,6 @@ export const ordersWithMarketProvider = makeDerivedDataProvider<
         ),
       },
     })),
-  (parts): Order[] | undefined => {
+  combineDelta<Order, ReturnType<typeof getDelta>['0']>,
-    if (!parts[0].isUpdate) {
+  combineInsertionData<Order>
-      return;
-    }
-    // map OrderSub_orders[] from subscription to updated Order[]
-    return (parts[0].delta as ReturnType<typeof getDelta>).map(
-      (deltaOrder) => ({
-        ...((parts[0].data as ReturnType<typeof getData>)?.find(
-          (order) => order.node.id === deltaOrder.id
-        )?.node as Orders_party_ordersConnection_edges_node),
-        market: (parts[1].data as Market[]).find(
-          (market) => market.id === deltaOrder.marketId
-        ),
-      })
-    );
-  }
 );

libs/react-helpers/src/lib/generic-data-provider.spec.ts

@@ -6,6 +6,7 @@ import {
 import type {
   CombineDerivedData,
   CombineDerivedDelta,
+  CombineInsertionData,
   Query,
   UpdateCallback,
   Update,
@@ -65,34 +66,20 @@ const subscribe = makeDataProvider<QueryData, Data, SubscriptionData, Delta>({
   getDelta: (r) => r.data,
 });
 
-const secondSubscribe = makeDataProvider<
-  QueryData,
-  Data,
-  SubscriptionData,
-  Delta
->({
-  query,
-  subscriptionQuery,
-  update,
-  getData: (r) => r.data,
-  getDelta: (r) => r.data,
-});
-
 const combineData = jest.fn<
   ReturnType<CombineDerivedData<CombinedData>>,
   Parameters<CombineDerivedData<CombinedData>>
 >();
 
 const combineDelta = jest.fn<
-  ReturnType<CombineDerivedDelta<never>>,
+  ReturnType<CombineDerivedDelta<CombinedData, CombinedData>>,
-  Parameters<CombineDerivedDelta<never>>
+  Parameters<CombineDerivedDelta<CombinedData, CombinedData>>
 >();
 
-const derivedSubscribe = makeDerivedDataProvider(
+const combineInsertionData = jest.fn<
-  [subscribe, secondSubscribe],
+  ReturnType<CombineInsertionData<CombinedData>>,
-  combineData,
+  Parameters<CombineInsertionData<CombinedData>>
-  combineDelta
+>();
-);
 
 const first = 100;
 const paginatedSubscribe = makeDataProvider<
@@ -114,6 +101,13 @@ const paginatedSubscribe = makeDataProvider<
   },
 });
 
+const derivedSubscribe = makeDerivedDataProvider(
+  [paginatedSubscribe, subscribe],
+  combineData,
+  combineDelta,
+  combineInsertionData
+);
+
 const generateData = (start = 0, size = first) => {
   return new Array(size).fill(null).map((v, i) => ({
     cursor: (i + start + 1).toString(),
@@ -514,22 +508,30 @@ describe('derived data provider', () => {
 
   it('calls callback on each meaningful update, uses combineData function', async () => {
     clearPendingQueries();
-    const part1: Item[] = [];
+    const totalCount = 1000;
+    const part1 = {
+      data: generateData(),
+      totalCount,
+      pageInfo: {
+        hasNextPage: true,
+        endCursor: '100',
+      },
+    };
     const part2: Item[] = [];
     const callback = jest.fn<
-      ReturnType<UpdateCallback<CombinedData, never>>,
+      ReturnType<UpdateCallback<CombinedData, CombinedData>>,
-      Parameters<UpdateCallback<CombinedData, never>>
+      Parameters<UpdateCallback<CombinedData, CombinedData>>
     >();
     const subscription = derivedSubscribe(callback, client);
     const data = { totalCount: 0 };
     combineData.mockReturnValueOnce(data);
     expect(callback.mock.calls.length).toBe(0);
-    await resolveQuery({ data: part1 });
+    await resolveQuery(part1);
     expect(combineData.mock.calls.length).toBe(0);
     expect(callback.mock.calls.length).toBe(0);
     await resolveQuery({ data: part2 });
     expect(combineData.mock.calls.length).toBe(1);
-    expect(combineData.mock.calls[0][0][0]).toBe(part1);
+    expect(combineData.mock.calls[0][0][0]).toBe(part1.data);
     expect(combineData.mock.calls[0][0][1]).toBe(part2);
     expect(callback.mock.calls.length).toBe(1);
     expect(callback.mock.calls[0][0].data).toBe(data);
@@ -538,13 +540,15 @@ describe('derived data provider', () => {
   });
 
   it('callback with error if any dependency has error, reloads all dependencies on reload', async () => {
+    clearPendingQueries();
     combineData.mockClear();
     const part1: Item[] = [];
     const part2: Item[] = [];
     const callback = jest.fn<
-      ReturnType<UpdateCallback<CombinedData, never>>,
+      ReturnType<UpdateCallback<CombinedData, CombinedData>>,
-      Parameters<UpdateCallback<CombinedData, never>>
+      Parameters<UpdateCallback<CombinedData, CombinedData>>
     >();
+    expect(callback.mock.calls.length).toBe(0);
     const subscription = derivedSubscribe(callback, client);
     const data = { totalCount: 0 };
     combineData.mockReturnValueOnce(data);
@@ -570,4 +574,84 @@ describe('derived data provider', () => {
     expect(callback.mock.calls[2][0].error).toBeUndefined();
     subscription.unsubscribe();
   });
+
+  it('pass isUpdate on any dependency isUpdate, uses result of combineDelta as delta in next callback', async () => {
+    clearPendingQueries();
+    combineData.mockClear();
+    const part1: Item[] = [];
+    const part2: Item[] = [];
+    const callback = jest.fn<
+      ReturnType<UpdateCallback<CombinedData, CombinedData>>,
+      Parameters<UpdateCallback<CombinedData, CombinedData>>
+    >();
+    const subscription = derivedSubscribe(callback, client);
+    const data = { totalCount: 0 };
+    combineData.mockReturnValueOnce(data);
+    await resolveQuery({ data: part1 });
+    await resolveQuery({ data: part2 });
+    expect(combineData).toBeCalledTimes(1);
+    expect(callback).toBeCalledTimes(1);
+    update.mockImplementation((data, delta) => [...data, ...delta]);
+    combineData.mockReturnValueOnce({ ...data });
+    const combinedDelta = {};
+    combineDelta.mockReturnValueOnce(combinedDelta);
+    // calling onNext from client.subscribe({ query }).subscribe(onNext)
+    const delta: Item[] = [];
+    await clientSubscribeSubscribe.mock.calls[
+      clientSubscribeSubscribe.mock.calls.length - 1
+    ][0]({ data: { data: delta } });
+    expect(combineDelta).toBeCalledTimes(1);
+    expect(combineData).toBeCalledTimes(2);
+    expect(callback).toBeCalledTimes(2);
+    expect(callback.mock.calls[1][0].isUpdate).toBe(true);
+    expect(callback.mock.calls[1][0].delta).toBe(combinedDelta);
+    subscription.unsubscribe();
+  });
+
+  it('pass isInsert on any dependency isInsert, uses result of combineInsertionData as insertionData in next callback', async () => {
+    clearPendingQueries();
+    combineData.mockClear();
+    combineDelta.mockClear();
+    const callback = jest.fn<
+      ReturnType<UpdateCallback<CombinedData, CombinedData>>,
+      Parameters<UpdateCallback<CombinedData, CombinedData>>
+    >();
+    const subscription = derivedSubscribe(callback, client);
+    const data = { totalCount: 0 };
+    combineData.mockReturnValueOnce(data);
+    await resolveQuery({
+      data: generateData(),
+      pageInfo: {
+        hasNextPage: true,
+        endCursor: '100',
+      },
+    });
+    await resolveQuery({ data: [] });
+    expect(combineData).toBeCalledTimes(1);
+    expect(callback).toBeCalledTimes(1);
+    update.mockImplementation((data, delta) => [...data, ...delta]);
+    combineData.mockReturnValueOnce({ ...data });
+    const combinedInsertionData = {};
+    combineInsertionData.mockReturnValueOnce(combinedInsertionData);
+    subscription.load && subscription.load();
+    const lastQueryArgs =
+      clientQuery.mock.calls[clientQuery.mock.calls.length - 1][0];
+    expect(lastQueryArgs?.variables?.pagination).toEqual({
+      after: '100',
+      first,
+    });
+    await resolveQuery({
+      data: generateData(100),
+      pageInfo: {
+        hasNextPage: true,
+        endCursor: '200',
+      },
+    });
+    expect(combineInsertionData).toBeCalledTimes(1);
+    expect(combineData).toBeCalledTimes(2);
+    expect(callback).toBeCalledTimes(2);
+    expect(callback.mock.calls[1][0].isInsert).toBe(true);
+    expect(callback.mock.calls[1][0].insertionData).toBe(combinedInsertionData);
+    subscription.unsubscribe();
+  });
 });

libs/react-helpers/src/lib/generic-data-provider.ts

@@ -94,6 +94,14 @@ interface GetDelta<SubscriptionData, Delta> {
   (subscriptionData: SubscriptionData, variables?: OperationVariables): Delta;
 }
 
+export type Node = { id: string };
+export type Cursor = {
+  cursor?: string | null;
+};
+export interface Edge<T extends Node> extends Cursor {
+  node: T;
+}
+
 export function defaultAppend<Data>(
   data: Data | null,
   insertionData: Data | null,
@@ -104,7 +112,7 @@ export function defaultAppend<Data>(
   if (data && insertionData && insertionPageInfo) {
     if (!(data instanceof Array) || !(insertionData instanceof Array)) {
       throw new Error(
-        'data needs to be instance of { cursor: string }[] when using pagination'
+        'data needs to be instance of Edge[] when using pagination'
       );
     }
     if (pagination?.after) {
@@ -220,9 +228,7 @@ function makeDataProviderInternal<QueryData, Data, SubscriptionData, Delta>({
       if (!start) {
         paginationVariables.after = undefined;
       } else if (data && data[start - 1]) {
-        paginationVariables.after = (
+        paginationVariables.after = (data[start - 1] as Cursor).cursor;
-          data[start - 1] as { cursor: string }
-        ).cursor;
       } else {
         let skip = 1;
         while (!data[start - 1 - skip] && skip <= start) {
@@ -232,9 +238,7 @@ function makeDataProviderInternal<QueryData, Data, SubscriptionData, Delta>({
         if (skip === start) {
           paginationVariables.after = undefined;
         } else {
-          paginationVariables.after = (
+          paginationVariables.after = (data[start - 1 - skip] as Cursor).cursor;
-            data[start - 1 - skip] as { cursor: string }
-          ).cursor;
         }
       }
     } else if (!pageInfo.hasNextPage) {
@@ -282,7 +286,7 @@ function makeDataProviderInternal<QueryData, Data, SubscriptionData, Delta>({
       if (data && pagination) {
         if (!(data instanceof Array)) {
           throw new Error(
-            'data needs to be instance of { cursor: string }[] when using pagination'
+            'data needs to be instance of Edge[] when using pagination'
           );
         }
         pageInfo = pagination.getPageInfo(res.data);
@@ -493,26 +497,35 @@ export function makeDataProvider<QueryData, Data, SubscriptionData, Delta>(
  */
 type DependencySubscribe = Subscribe<any, any>; // eslint-disable-line @typescript-eslint/no-explicit-any
 type DependencyUpdateCallback = Parameters<DependencySubscribe>['0'];
+export type DerivedPart = Parameters<DependencyUpdateCallback>['0'];
 export type CombineDerivedData<Data> = (
-  data: Parameters<DependencyUpdateCallback>['0']['data'][],
+  data: DerivedPart['data'][],
   variables?: OperationVariables
 ) => Data | null;
 
-export type CombineDerivedDelta<Delta> = (
+export type CombineDerivedDelta<Data, Delta> = (
-  parts: Parameters<DependencyUpdateCallback>['0'][],
+  data: Data,
+  parts: DerivedPart[],
   variables?: OperationVariables
 ) => Delta | undefined;
 
+export type CombineInsertionData<Data> = (
+  data: Data,
+  parts: DerivedPart[],
+  variables?: OperationVariables
+) => Data | undefined;
+
 function makeDerivedDataProviderInternal<Data, Delta>(
   dependencies: DependencySubscribe[],
   combineData: CombineDerivedData<Data>,
-  combineDelta?: CombineDerivedDelta<Delta>
+  combineDelta?: CombineDerivedDelta<Data, Delta>,
+  combineInsertionData?: CombineInsertionData<Data>
 ): Subscribe<Data, Delta> {
   let subscriptions: ReturnType<DependencySubscribe>[] | undefined;
   let client: ApolloClient<object>;
   const callbacks: UpdateCallback<Data, Delta>[] = [];
   let variables: OperationVariables | undefined;
-  const parts: Parameters<DependencyUpdateCallback>['0'][] = [];
+  const parts: DerivedPart[] = [];
   let data: Data | null = null;
   let error: Error | undefined;
   let loading = true;
@@ -541,7 +554,8 @@ function makeDerivedDataProviderInternal<Data, Delta>(
   const combine = (updatedPartIndex: number) => {
     let delta: Delta | undefined;
     let isUpdate = false;
-    const isInsert = false;
+    let isInsert = false;
+    let insertionData: Data | undefined;
     let newError: Error | undefined;
     let newLoading = false;
     let newLoaded = true;
@@ -558,17 +572,6 @@ function makeDerivedDataProviderInternal<Data, Delta>(
           variables
         )
       : data;
-    if (newLoaded) {
-      const updatedPart = parts[updatedPartIndex];
-      if (updatedPart.isUpdate) {
-        isUpdate = true;
-        if (updatedPart.delta && combineDelta) {
-          delta = combineDelta(parts, variables);
-        }
-        delete updatedPart.isUpdate;
-        delete updatedPart.delta;
-      }
-    }
     if (
       newLoading !== loading ||
       newError !== error ||
@@ -579,10 +582,30 @@ function makeDerivedDataProviderInternal<Data, Delta>(
       error = newError;
       loaded = newLoaded;
       data = newData;
+      if (newLoaded) {
+        const updatedPart = parts[updatedPartIndex];
+        if (updatedPart.isUpdate) {
+          isUpdate = true;
+          if (updatedPart.delta && combineDelta && data) {
+            delta = combineDelta(data, parts, variables);
+          }
+          delete updatedPart.isUpdate;
+          delete updatedPart.delta;
+        }
+        if (updatedPart.isInsert) {
+          isInsert = updatedPartIndex === 0;
+          if (updatedPart.insertionData && combineInsertionData && data) {
+            insertionData = combineInsertionData(data, parts, variables);
+          }
+          delete updatedPart.insertionData;
+          delete updatedPart.isInsert;
+        }
+      }
       notifyAll({
         isUpdate,
         isInsert,
         delta,
+        insertionData,
       });
     }
   };
@@ -641,10 +664,16 @@ function makeDerivedDataProviderInternal<Data, Delta>(
 export function makeDerivedDataProvider<Data, Delta>(
   dependencies: DependencySubscribe[],
   combineData: CombineDerivedData<Data>,
-  combineDelta?: CombineDerivedDelta<Delta>
+  combineDelta?: CombineDerivedDelta<Data, Delta>,
+  combineInsertionData?: CombineInsertionData<Data>
 ): Subscribe<Data, Delta> {
   const getInstance = memoize<Data, Delta>(() =>
-    makeDerivedDataProviderInternal(dependencies, combineData, combineDelta)
+    makeDerivedDataProviderInternal(
+      dependencies,
+      combineData,
+      combineDelta,
+      combineInsertionData
+    )
   );
   return (callback, client, variables) =>
     getInstance(variables)(callback, client, variables);

libs/react-helpers/src/lib/get-nodes.ts

@@ -1,18 +1,16 @@
 import type { Schema } from '@vegaprotocol/types';
 
-export type Node<T> = {
+type Edge<T> = {
-  __typename?: string;
   node: T;
 };
 
-export type Connection<A> = {
+type Connection<A> = {
-  __typename?: string;
   edges?: Schema.Maybe<Array<Schema.Maybe<A>>>;
 };
 
 export function getNodes<
   T,
-  A extends Node<T> = Node<T>,
+  A extends Edge<T> = Edge<T>,
   B extends Connection<A> = Connection<A>
 >(data?: B | null, filterBy?: (item?: T | null) => boolean) {
   const edges = data?.edges || [];

libs/react-helpers/src/lib/pagination.ts

@@ -1,5 +1,6 @@
 import type { IGetRowsParams } from 'ag-grid-community';
-import type { Load } from './generic-data-provider';
+import type { Load, DerivedPart } from './generic-data-provider';
+import type { Node, Edge } from './generic-data-provider';
 import type { MutableRefObject } from 'react';
 
 const getLastRow = (
@@ -57,3 +58,39 @@ export const makeInfiniteScrollGetRows =
       failCallback();
     }
   };
+
+export const paginatedCombineDelta = <
+  DataNode extends Node,
+  DeltaNode extends Node
+>(
+  data: (Edge<DataNode> | null)[],
+  parts: DerivedPart[]
+): DataNode[] | undefined => {
+  if (!parts[0].isUpdate) {
+    return;
+  }
+  const updatedIds = (parts[0].delta as DeltaNode[]).map((node) => node.id);
+  return data
+    .filter<Edge<DataNode>>(
+      (edge): edge is Edge<DataNode> =>
+        edge !== null && updatedIds.includes(edge.node.id)
+    )
+    .map((edge) => edge.node);
+};
+
+export const paginatedCombineInsertionData = <DataNode extends Node>(
+  data: (Edge<DataNode> | null)[],
+  parts: DerivedPart[]
+): Edge<DataNode>[] | undefined => {
+  if (!parts[0].isInsert) {
+    return;
+  }
+  const insertedIds = (parts[0].insertionData as DataNode[]).map(
+    (node) => node.id
+  );
+  // get updated orders
+  return data.filter<Edge<DataNode>>(
+    (edge): edge is Edge<DataNode> =>
+      edge !== null && insertedIds.includes(edge.node.id)
+  );
+};

libs/trades/src/lib/trades-data-provider.ts

@@ -2,8 +2,10 @@ import {
   makeDataProvider,
   makeDerivedDataProvider,
   defaultAppend as append,
+  paginatedCombineDelta as combineDelta,
+  paginatedCombineInsertionData as combineInsertionData,
 } from '@vegaprotocol/react-helpers';
-import type { PageInfo } from '@vegaprotocol/react-helpers';
+import type { PageInfo, Edge } from '@vegaprotocol/react-helpers';
 import type { Market } from '@vegaprotocol/market-list';
 import { marketsProvider } from '@vegaprotocol/market-list';
 import type {
@@ -63,10 +65,7 @@ const update = (
 };
 
 export type Trade = Omit<TradeFieldsFragment, 'market'> & { market?: Market };
-export type TradeEdge = {
+export type TradeEdge = Edge<Trade>;
-  cursor: string;
-  node: Trade;
-};
 
 const getPageInfo = (responseData: TradesQuery): PageInfo | null =>
   responseData.market?.tradesConnection?.pageInfo || null;
@@ -108,20 +107,6 @@ export const tradesWithMarketProvider = makeDerivedDataProvider<
       };
     });
   },
-  (parts): Trade[] | undefined => {
+  combineDelta<Trade, ReturnType<typeof getDelta>['0']>,
-    if (!parts[0].isUpdate) {
+  combineInsertionData<Trade>
-      return;
-    }
-    // map FillsSub_trades[] from subscription to updated Trade[]
-    return (parts[0].delta as ReturnType<typeof getDelta>).map(
-      (deltaTrade) => ({
-        ...((parts[0].data as ReturnType<typeof getData>)?.find(
-          (edge) => edge?.node.id === deltaTrade.id
-        )?.node as Trade),
-        market: (parts[1].data as Market[]).find(
-          (market) => market.id === deltaTrade.marketId
-        ),
-      })
-    );
-  }
 );