feat: Fetcher updates (#1090)

* fetcher updates

* add placeholder for fetcher

* update fetcher mock

* fetcher start update

* update todo

* Apply suggestion from @Copilot

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* fix copilot comments

* remove extra comment

* lint fix

* change lastFlattented initial value to -1

* move lastFlattenedPagesCount to a ref and remove shouldContinue condition

* fix typo and add mock for configs

* update page size test case

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: Assem-Uber

src/views/workflow-history-v2/workflow-history-v2.tsx

@@ -4,7 +4,7 @@ import usePageFilters from '@/components/page-filters/hooks/use-page-filters';
 import decodeUrlParams from '@/utils/decode-url-params';
 
 import workflowHistoryFiltersConfig from '../workflow-history/config/workflow-history-filters.config';
-import WORKFLOW_HISTORY_PAGE_SIZE_CONFIG from '../workflow-history/config/workflow-history-page-size.config';
+import { WORKFLOW_HISTORY_PAGE_SIZE_CONFIG } from '../workflow-history/config/workflow-history-page-size.config';
 import { WorkflowHistoryContext } from '../workflow-history/workflow-history-context-provider/workflow-history-context-provider';
 import workflowPageQueryParamsConfig from '../workflow-page/config/workflow-page-query-params.config';
 import { type WorkflowPageTabContentParams } from '../workflow-page/workflow-page-tab-content/workflow-page-tab-content.types';

src/views/workflow-history/__tests__/workflow-history.test.tsx

@@ -33,7 +33,9 @@ jest.mock('../hooks/use-workflow-history-fetcher', () => {
   const actual = jest.requireActual('../hooks/use-workflow-history-fetcher');
   return {
     __esModule: true,
-    default: jest.fn((params) => actual.default(params, 0)), // 0ms throttle for tests
+    default: jest.fn((params, onEventsChange) =>
+      actual.default(params, onEventsChange, 0)
+    ), // 0ms throttle for tests
   };
 });
 

src/views/workflow-history/config/workflow-history-page-size.config.ts

@@ -1,3 +1,2 @@
-const WORKFLOW_HISTORY_PAGE_SIZE_CONFIG = 200;
-
-export default WORKFLOW_HISTORY_PAGE_SIZE_CONFIG;
+export const WORKFLOW_HISTORY_PAGE_SIZE_CONFIG = 1000;
+export const WORKFLOW_HISTORY_FIRST_PAGE_SIZE_CONFIG = 200;

src/views/workflow-history/helpers/__tests__/workflow-history-fetcher.test.tsx

@@ -15,6 +15,15 @@ const RETRY_COUNT = 3;
 let queryClient: QueryClient;
 let hoistedFetcher: WorkflowHistoryFetcher;
 
+jest.mock(
+  '@/views/workflow-history/config/workflow-history-page-size.config',
+  () => ({
+    __esModule: true,
+    WORKFLOW_HISTORY_FIRST_PAGE_SIZE_CONFIG: 200,
+    WORKFLOW_HISTORY_PAGE_SIZE_CONFIG: 1000,
+  })
+);
+
 describe(WorkflowHistoryFetcher.name, () => {
   beforeEach(() => {
     queryClient = new QueryClient({
@@ -262,6 +271,35 @@ describe(WorkflowHistoryFetcher.name, () => {
     const state = fetcher.getCurrentState();
     expect(state.data?.pages.length).toBe(pageCountBefore);
   });
+
+  it('should use WORKFLOW_HISTORY_FIRST_PAGE_SIZE_CONFIG for the first page', async () => {
+    const { fetcher, getCapturedPageSizes } = setup(queryClient);
+
+    fetcher.start((state) => !state?.data?.pages?.length);
+
+    await waitFor(() => {
+      const state = fetcher.getCurrentState();
+      expect(state.data?.pages).toHaveLength(1);
+    });
+
+    const pageSizes = getCapturedPageSizes();
+    expect(pageSizes[0]).toBe('200');
+  });
+
+  it('should use WORKFLOW_HISTORY_PAGE_SIZE_CONFIG for subsequent pages', async () => {
+    const { fetcher, getCapturedPageSizes } = setup(queryClient);
+
+    fetcher.start((state) => (state.data?.pages.length || 0) < 3);
+
+    await waitFor(() => {
+      const state = fetcher.getCurrentState();
+      expect(state.data?.pages).toHaveLength(3);
+    });
+
+    const pageSizes = getCapturedPageSizes();
+    expect(pageSizes[1]).toBe('1000');
+    expect(pageSizes[2]).toBe('1000');
+  });
 });
 
 function setup(client: QueryClient, options: { failOnPages?: number[] } = {}) {
@@ -273,7 +311,10 @@ function setup(client: QueryClient, options: { failOnPages?: number[] } = {}) {
     pageSize: 10,
   };
 
-  mockHistoryEndpoint(workflowHistoryMultiPageFixture, options.failOnPages);
+  const { getCapturedPageSizes } = mockHistoryEndpoint(
+    workflowHistoryMultiPageFixture,
+    options.failOnPages
+  );
   const fetcher = new WorkflowHistoryFetcher(client, params);
   hoistedFetcher = fetcher;
 
@@ -292,13 +333,16 @@ function setup(client: QueryClient, options: { failOnPages?: number[] } = {}) {
     fetcher,
     params,
     waitForData,
+    getCapturedPageSizes,
   };
 }
 
 function mockHistoryEndpoint(
   responses: GetWorkflowHistoryResponse[],
   failOnPages: number[] = []
 ) {
+  const capturedPageSizes: string[] = [];
+
   mswMockEndpoints([
     {
       path: '/api/domains/:domain/:cluster/workflows/:workflowId/:runId/history',
@@ -307,6 +351,9 @@ function mockHistoryEndpoint(
       httpResolver: async ({ request }) => {
         const url = new URL(request.url);
         const nextPage = url.searchParams.get('nextPage');
+        const pageSize = url.searchParams.get('pageSize');
+
+        capturedPageSizes.push(pageSize ?? '');
 
         // Determine current page number based on nextPage param
         let pageNumber = 1;
@@ -334,4 +381,8 @@ function mockHistoryEndpoint(
       },
     },
   ]);
+
+  return {
+    getCapturedPageSizes: () => capturedPageSizes,
+  };
 }

src/views/workflow-history/helpers/workflow-history-fetcher.ts

@@ -7,6 +7,11 @@ import {
 } from '@/route-handlers/get-workflow-history/get-workflow-history.types';
 import request from '@/utils/request';
 
+import {
+  WORKFLOW_HISTORY_FIRST_PAGE_SIZE_CONFIG,
+  WORKFLOW_HISTORY_PAGE_SIZE_CONFIG,
+} from '../config/workflow-history-page-size.config';
+
 import {
   type WorkflowHistoryQueryResult,
   type QueryResultOnChangeCallback,
@@ -41,16 +46,18 @@ export default class WorkflowHistoryFetcher {
   }
 
   start(shouldContinue: ShouldContinueCallback = () => true): void {
-    if (shouldContinue) {
-      this.shouldContinue = shouldContinue;
-    }
-    // If already started, return
-    if (this.isStarted) return;
+    this.shouldContinue = shouldContinue;
+
+    // remove current listener (if exists) to have fresh emits only
+    this.unsubscribe?.();
+    this.unsubscribe = null;
+
     this.isStarted = true;
     let emitCount = 0;
     const currentState = this.observer.getCurrentResult();
     const fetchedFirstPage = currentState.status !== 'pending';
-    const shouldEnableQuery = !fetchedFirstPage && shouldContinue(currentState);
+    const shouldEnableQuery =
+      !fetchedFirstPage && this.shouldContinue(currentState);
 
     if (shouldEnableQuery) {
       this.observer.setOptions({
@@ -81,8 +88,6 @@ export default class WorkflowHistoryFetcher {
       emit(currentState);
     }
 
-    // remove current listener (if exists) and add new one
-    this.unsubscribe?.();
     this.unsubscribe = this.observer.subscribe((res) => emit(res));
   }
 
@@ -126,7 +131,9 @@ export default class WorkflowHistoryFetcher {
             url: `/api/domains/${params.domain}/${params.cluster}/workflows/${params.workflowId}/${params.runId}/history`,
             query: {
               nextPage: pageParam,
-              pageSize: params.pageSize,
+              pageSize: pageParam
+                ? WORKFLOW_HISTORY_PAGE_SIZE_CONFIG
+                : WORKFLOW_HISTORY_FIRST_PAGE_SIZE_CONFIG,
               waitForNewEvent: params.waitForNewEvent ?? false,
             } satisfies WorkflowHistoryQueryParams,
           })

src/views/workflow-history/hooks/__tests__/use-workflow-history-fetcher.test.tsx

@@ -19,13 +19,17 @@ let mockOnChangeCallback: jest.Mock;
 let mockUnsubscribe: jest.Mock;
 
 function setup() {
-  const hookResult = renderHook(() => useWorkflowHistoryFetcher(mockParams));
+  const mockOnEventsChange = jest.fn();
+  const hookResult = renderHook(() =>
+    useWorkflowHistoryFetcher(mockParams, mockOnEventsChange)
+  );
 
   return {
     ...hookResult,
     mockFetcherInstance,
     mockOnChangeCallback,
     mockUnsubscribe,
+    mockOnEventsChange,
   };
 }
 

src/views/workflow-history/hooks/use-workflow-history-fetcher.ts

@@ -6,6 +6,7 @@ import {
   useQueryClient,
 } from '@tanstack/react-query';
 
+import { type HistoryEvent } from '@/__generated__/proto-ts/uber/cadence/api/v1/HistoryEvent';
 import useThrottledState from '@/hooks/use-throttled-state';
 import {
   type WorkflowHistoryQueryParams,
@@ -19,13 +20,18 @@ import { type ShouldContinueCallback } from '../helpers/workflow-history-fetcher
 
 export default function useWorkflowHistoryFetcher(
   params: WorkflowHistoryQueryParams & RouteParams,
+  onEventsChange: (events: HistoryEvent[]) => void,
   throttleMs: number = 2000
 ) {
   const queryClient = useQueryClient();
   const fetcherRef = useRef<WorkflowHistoryFetcher | null>(null);
+  const lastFlattenedPagesCountRef = useRef<number>(-1);
 
   if (!fetcherRef.current) {
     fetcherRef.current = new WorkflowHistoryFetcher(queryClient, params);
+
+    // Fetch first page
+    fetcherRef.current.start((state) => !state?.data?.pages?.length);
   }
 
   const [historyQuery, setHistoryQuery] = useThrottledState<
@@ -40,21 +46,25 @@ export default function useWorkflowHistoryFetcher(
 
   useEffect(() => {
     if (!fetcherRef.current) return;
-
     const unsubscribe = fetcherRef.current.onChange((state) => {
       const pagesCount = state.data?.pages?.length || 0;
+      // If the pages count is greater than the last flattened pages count, then we need to flatten the pages and call the onEventsChange callback
+      // Depending on ref variable instead of historyQuery is because historyQuery is throttled.
+      if (pagesCount > lastFlattenedPagesCountRef.current) {
+        lastFlattenedPagesCountRef.current = pagesCount;
+        onEventsChange(
+          state.data?.pages?.flatMap((page) => page.history?.events || []) || []
+        );
+      }
       // immediately set if there is the first page without throttling other wise throttle
       const executeImmediately = pagesCount <= 1;
       setHistoryQuery(() => state, executeImmediately);
     });
 
-    // Fetch first page
-    fetcherRef.current.start((state) => !state?.data?.pages?.length);
-
     return () => {
       unsubscribe();
     };
-  }, [setHistoryQuery]);
+  }, [setHistoryQuery, onEventsChange]);
 
   useEffect(() => {
     return () => {

src/views/workflow-history/workflow-history.tsx

@@ -25,7 +25,7 @@ import workflowPageQueryParamsConfig from '../workflow-page/config/workflow-page
 import { useSuspenseDescribeWorkflow } from '../workflow-page/hooks/use-describe-workflow';
 
 import workflowHistoryFiltersConfig from './config/workflow-history-filters.config';
-import WORKFLOW_HISTORY_PAGE_SIZE_CONFIG from './config/workflow-history-page-size.config';
+import { WORKFLOW_HISTORY_PAGE_SIZE_CONFIG } from './config/workflow-history-page-size.config';
 import compareUngroupedEvents from './helpers/compare-ungrouped-events';
 import getSortableEventId from './helpers/get-sortable-event-id';
 import getVisibleGroupsHasMissingEvents from './helpers/get-visible-groups-has-missing-events';
@@ -73,6 +73,8 @@ export default function WorkflowHistory({ params }: Props) {
       pageSize: wfHistoryRequestArgs.pageSize,
       waitForNewEvent: wfHistoryRequestArgs.waitForNewEvent,
     },
+    //TODO: @assem.hafez replace this with grouper callback
+    () => {},
     2000
   );