fix(storage-browser): move useProcessTasks concurrency option to dispatch handler (#6525)

Author: calebpollman

.changeset/nasty-months-teach.md

@@ -0,0 +1,5 @@
+---
+"@aws-amplify/ui-react-storage": patch
+---
+
+fix(storage-browser): move useProcessTasks concurrency option to dispatch handler

packages/react-storage/src/components/StorageBrowser/tasks/__tests__/useProcessTasks.spec.ts

@@ -91,11 +91,7 @@ describe('useProcessTasks', () => {
 
   it('handles concurrent tasks as expected', async () => {
     const { result } = renderHook(() =>
-      useProcessTasks(action, {
-        concurrency: 2,
-        items,
-        onTaskProgress,
-      })
+      useProcessTasks(action, { items, onTaskProgress })
     );
 
     const processTasks = result.current[1];
@@ -105,7 +101,7 @@ describe('useProcessTasks', () => {
     expect(result.current[0].tasks[2].status).toBe('QUEUED');
 
     act(() => {
-      processTasks({ config });
+      processTasks({ config, options: { concurrency: 2 } });
     });
 
     expect(action).toHaveBeenCalledTimes(2);
@@ -392,13 +388,13 @@ describe('useProcessTasks', () => {
 
   it('returns `error` and `message` for a failed task and provides the expected values to `onTaskError`', async () => {
     const { result } = renderHook(() =>
-      useProcessTasks(action, { concurrency: 2, items, onTaskError })
+      useProcessTasks(action, { items, onTaskError })
     );
 
     const processTasks = result.current[1];
 
     act(() => {
-      processTasks({ config });
+      processTasks({ config, options: { concurrency: 2 } });
     });
 
     expect(result.current[0].tasks[0].status).toBe('PENDING');

packages/react-storage/src/components/StorageBrowser/tasks/types.ts

@@ -3,6 +3,7 @@ import type {
   TaskData,
   TaskResult,
   TaskResultStatus,
+  TaskHandlerOptions,
 } from '../actions';
 
 /**
@@ -17,8 +18,6 @@ export type TaskStatus = TaskResultStatus | 'QUEUED' | 'PENDING';
 export type StatusCounts = Record<TaskStatus | 'TOTAL', number>;
 
 export interface ProcessTasksOptions<TTask extends Task, TItems = []> {
-  // has no impact in atomic mode
-  concurrency?: number;
   items?: TItems;
   onTaskCancel?: (task: TTask) => void;
   onTaskComplete?: (task: TTask) => void;
@@ -65,9 +64,19 @@ export type UseProcessTasksState<TTask, TInput> = [
   HandleProcessTasks<TInput>,
 ];
 
+interface HandleTasksOptions extends TaskHandlerOptions {
+  concurrency?: number;
+}
+
+export interface HandleBatchTasksInput<TData extends TaskData>
+  extends Omit<TaskHandlerInput<TData, HandleTasksOptions>, 'data'> {}
+
+export interface HandleSingleTaskInput<TData extends TaskData>
+  extends TaskHandlerInput<TData> {}
+
 export type InferHandleTasksInput<
   TItems,
   TData extends TaskData,
 > = TItems extends NonNullable<TItems>
-  ? Omit<TaskHandlerInput<TData>, 'data'>
-  : TaskHandlerInput<TData>;
+  ? HandleBatchTasksInput<TData>
+  : HandleSingleTaskInput<TData>;

packages/react-storage/src/components/StorageBrowser/tasks/useProcessTasks.ts

@@ -1,10 +1,12 @@
 import React from 'react';
 import { isFunction } from '@aws-amplify/ui';
 
-import type { ActionHandler, TaskData, TaskHandlerInput } from '../actions';
+import type { ActionHandler, TaskData } from '../actions';
 
 import type {
+  HandleBatchTasksInput,
   HandleProcessTasks,
+  HandleSingleTaskInput,
   InferHandleTasksInput,
   Task,
   ProcessTasksOptions,
@@ -23,9 +25,10 @@ const QUEUED_TASK_BASE = {
   status: 'QUEUED' as const,
 };
 
-const isTaskHandlerInput = <T extends TaskData>(
-  input: TaskHandlerInput<T> | Omit<TaskHandlerInput<T>, 'data'>
-): input is TaskHandlerInput<T> => !!(input as TaskHandlerInput<T>).data;
+const isSingleTaskInput = <TData extends TaskData>(
+  input: HandleSingleTaskInput<TData> | HandleBatchTasksInput<TData>
+): input is HandleSingleTaskInput<TData> =>
+  !!(input as HandleSingleTaskInput<TData>).data;
 
 export function useProcessTasks<
   TData,
@@ -38,7 +41,7 @@ export function useProcessTasks<
   handler: ActionHandler<TData, TValue>,
   options?: ProcessTasksOptions<TTask, TItems>
 ): UseProcessTasksState<TTask, TInput> {
-  const { concurrency, items, ...callbacks } = options ?? {};
+  const { items, ...callbacks } = options ?? {};
 
   const callbacksRef = React.useRef(callbacks);
 
@@ -133,14 +136,14 @@ export function useProcessTasks<
     flush();
   }, [createTask, flush, updateTask, items, refreshTaskData]);
 
-  const processNextTask = (_input: TInput) => {
-    const hasInputData = isTaskHandlerInput(_input);
-    if (hasInputData) {
+  const processTask = (_input: TInput) => {
+    const isSingleTask = isSingleTaskInput(_input);
+    if (isSingleTask) {
       createTask(_input.data);
       flush();
     }
 
-    const { data } = hasInputData
+    const { data } = isSingleTask
       ? _input
       : [...tasksRef.current.values()].find(
           ({ status }) => status === 'QUEUED'
@@ -203,10 +206,10 @@ export function useProcessTasks<
         const task = getTask();
         if (task && isFunction(onTaskComplete)) onTaskComplete(task);
 
-        // ignore process next task for single operation inputs
-        if (hasInputData) return;
+        // ignore process next task for single task
+        if (isSingleTask) return;
 
-        processNextTask(_input);
+        processTask(_input);
       });
 
     updateTask(data.id, { cancel, status: 'PENDING' });
@@ -222,14 +225,26 @@ export function useProcessTasks<
       return;
     }
 
+    // if single task, run `processTask` once
+    if (isSingleTaskInput(input)) {
+      processTask(input);
+      return;
+    }
+
+    const { concurrency, ...options } = input.options ?? {};
+
+    // reconstruct `input` without `concurrency`
+    const _input = { ...input, options };
+
+    // for batch tasks, if no `concurrency` process tasks individually
     if (!concurrency) {
-      processNextTask(input);
+      processTask(_input);
       return;
     }
 
     let count = 0;
     while (count < concurrency) {
-      processNextTask(input);
+      processTask(_input);
       count++;
     }
   };

packages/react-storage/src/components/StorageBrowser/useAction/__tests__/useHandler.spec.ts

@@ -120,7 +120,10 @@ describe('useHandler', () => {
     dispatch();
 
     expect(mockUseProcessDispatch).toHaveBeenCalledTimes(1);
-    expect(mockUseProcessDispatch).toHaveBeenCalledWith({ config });
+    expect(mockUseProcessDispatch).toHaveBeenCalledWith({
+      config,
+      options: { concurrency: DEFAULT_ACTION_CONCURRENCY },
+    });
   });
 
   it('provides the expected values to `useProcessTasks` when called with `options` that include `items`', () => {
@@ -143,7 +146,6 @@ describe('useHandler', () => {
 
     expect(useProcessTasksMock).toHaveBeenCalledTimes(1);
     expect(useProcessTasksMock).toHaveBeenCalledWith(handler, {
-      concurrency: DEFAULT_ACTION_CONCURRENCY,
       items: input.items,
       onTaskError: input.onTaskError,
       onTaskProgress: input.onTaskProgress,

packages/react-storage/src/components/StorageBrowser/useAction/useHandler.ts

@@ -46,10 +46,7 @@ export function useHandler<
   handler: THandler,
   options?: UseHandlerOptionsWithItems<TTask> | UseHandlerOptions<TTask>
 ): HandleTasksState<TTask> | HandleTaskState<TTask> {
-  const [state, handleProcessing] = useProcessTasks(handler, {
-    ...options,
-    concurrency: DEFAULT_ACTION_CONCURRENCY,
-  });
+  const [state, handleProcessing] = useProcessTasks(handler, options);
   const getConfig = useGetActionInput();
 
   const { reset, isProcessing, tasks, ...rest } = state;
@@ -64,7 +61,10 @@ export function useHandler<
 
       handleProcessing({
         config,
-        ...(hasData ? { data: input.data } : undefined),
+        ...(hasData
+          ? { data: input.data }
+          : // if no `data` provided, provide `concurrency` to `options`
+            { options: { concurrency: DEFAULT_ACTION_CONCURRENCY } }),
       });
     },
     [getConfig, handleProcessing, reset]