mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-11-08 12:24:45 +00:00
fix(ui): complete grouped Jira dispatch review
This commit is contained in:
26 files changed
+1668
-894
No files matched your search
@@ -49,6 +49,37 @@ describe("task watcher store", () => {
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("returns and persists a completed task result while allowing the caller to own notifications", async () => {
|
||||
// Given
|
||||
pollMock.mockResolvedValue({
|
||||
ok: true,
|
||||
state: "completed",
|
||||
result: { created_count: 1, failed_count: 1 },
|
||||
});
|
||||
|
||||
// When
|
||||
const result = await trackAndPollTask<{
|
||||
created_count: number;
|
||||
failed_count: number;
|
||||
}>({
|
||||
taskId: "jira-task",
|
||||
kind: "test-kind",
|
||||
meta: {},
|
||||
notifyHandler: false,
|
||||
});
|
||||
|
||||
// Then
|
||||
expect(result).toEqual({
|
||||
status: TASK_WATCHER_STATUS.READY,
|
||||
result: { created_count: 1, failed_count: 1 },
|
||||
});
|
||||
expect(useTaskWatcherStore.getState().tasks["jira-task"]?.result).toEqual({
|
||||
created_count: 1,
|
||||
failed_count: 1,
|
||||
});
|
||||
expect(onReady).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("replaces settled results of the same kind when tracking new work", async () => {
|
||||
// Given
|
||||
pollMock.mockResolvedValue({ ok: true, state: "completed" });
|
||||
@@ -201,6 +232,39 @@ describe("task watcher store", () => {
|
||||
expect(onReady).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("passes a persisted task result to its handler after resuming", async () => {
|
||||
// Given
|
||||
const taskResult = {
|
||||
created_count: 1,
|
||||
failed_count: 1,
|
||||
failed_finding_ids: ["finding-2"],
|
||||
};
|
||||
pollMock.mockResolvedValue({
|
||||
ok: true,
|
||||
state: "completed",
|
||||
result: taskResult,
|
||||
});
|
||||
useTaskWatcherStore.setState({
|
||||
tasks: {
|
||||
"resumed-jira-task": {
|
||||
taskId: "resumed-jira-task",
|
||||
kind: "test-kind",
|
||||
status: TASK_WATCHER_STATUS.PENDING,
|
||||
meta: {},
|
||||
startedAt: Date.now(),
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
// When
|
||||
await resumePendingTasks();
|
||||
|
||||
// Then
|
||||
expect(onReady).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ result: taskResult }),
|
||||
);
|
||||
});
|
||||
|
||||
it("discards settled tasks before resuming persisted work", async () => {
|
||||
// Given
|
||||
pollMock.mockResolvedValue({ ok: true, state: "completed" });
|
||||
|
||||
+138
-72
@@ -28,6 +28,9 @@ export interface WatchedTask {
|
||||
meta: Record<string, string>;
|
||||
startedAt: number;
|
||||
error?: string;
|
||||
/** Serializable task result. Persisted so another tab or a reload can
|
||||
* finish feature-specific handling without polling the task again. */
|
||||
result?: unknown;
|
||||
}
|
||||
|
||||
export interface TaskKindHandler {
|
||||
@@ -35,6 +38,21 @@ export interface TaskKindHandler {
|
||||
onError: (task: WatchedTask) => void;
|
||||
}
|
||||
|
||||
export interface TaskTrackingResult<R = unknown> {
|
||||
status: TaskWatcherStatus;
|
||||
error?: string;
|
||||
result?: R;
|
||||
}
|
||||
|
||||
export interface TrackAndPollTaskInput {
|
||||
taskId: string;
|
||||
kind: string;
|
||||
meta: Record<string, string>;
|
||||
/** Let the awaiting caller aggregate notifications. If this tab reloads,
|
||||
* the persisted task resumes with its registered handler as usual. */
|
||||
notifyHandler?: boolean;
|
||||
}
|
||||
|
||||
interface TaskWatcherState {
|
||||
tasks: Record<string, WatchedTask>;
|
||||
upsertTask: (task: WatchedTask) => void;
|
||||
@@ -42,6 +60,7 @@ interface TaskWatcherState {
|
||||
taskId: string,
|
||||
status: TaskWatcherStatus,
|
||||
error?: string,
|
||||
result?: unknown,
|
||||
) => void;
|
||||
dismissTask: (taskId: string) => void;
|
||||
}
|
||||
@@ -69,12 +88,15 @@ export const useTaskWatcherStore = create<TaskWatcherState>()(
|
||||
tasks: {},
|
||||
upsertTask: (task) =>
|
||||
set((state) => ({ tasks: { ...state.tasks, [task.taskId]: task } })),
|
||||
resolveTask: (taskId, status, error) =>
|
||||
resolveTask: (taskId, status, error, result) =>
|
||||
set((state) => {
|
||||
const task = state.tasks[taskId];
|
||||
if (!task) return state;
|
||||
return {
|
||||
tasks: { ...state.tasks, [taskId]: { ...task, status, error } },
|
||||
tasks: {
|
||||
...state.tasks,
|
||||
[taskId]: { ...task, status, error, result },
|
||||
},
|
||||
};
|
||||
}),
|
||||
dismissTask: (taskId) =>
|
||||
@@ -92,25 +114,35 @@ export const useTaskWatcherStore = create<TaskWatcherState>()(
|
||||
|
||||
// In-memory only: poll loops alive in THIS tab. Never persisted, so a reload
|
||||
// naturally re-enters through resumePendingTasks without double-polling.
|
||||
const activePolls = new Set<string>();
|
||||
const activePolls = new Map<string, Promise<TaskTrackingResult<unknown>>>();
|
||||
const suppressedHandlers = new Set<string>();
|
||||
|
||||
const settleTask = (
|
||||
taskId: string,
|
||||
status: TaskWatcherStatus,
|
||||
error?: string,
|
||||
) => {
|
||||
result?: unknown,
|
||||
): TaskTrackingResult => {
|
||||
const store = useTaskWatcherStore.getState();
|
||||
const currentTask = store.tasks[taskId];
|
||||
if (!currentTask || currentTask.status !== TASK_WATCHER_STATUS.PENDING) {
|
||||
return;
|
||||
return {
|
||||
status: currentTask?.status ?? TASK_WATCHER_STATUS.ERROR,
|
||||
...(currentTask?.error ? { error: currentTask.error } : {}),
|
||||
...(currentTask?.result !== undefined
|
||||
? { result: currentTask.result }
|
||||
: {}),
|
||||
};
|
||||
}
|
||||
|
||||
store.resolveTask(taskId, status, error);
|
||||
store.resolveTask(taskId, status, error, result);
|
||||
const task = useTaskWatcherStore.getState().tasks[taskId];
|
||||
if (!task) return;
|
||||
if (!task) {
|
||||
return { status: TASK_WATCHER_STATUS.ERROR, error: "Task unavailable." };
|
||||
}
|
||||
|
||||
const handler = handlers.get(task.kind);
|
||||
if (handler) {
|
||||
if (handler && !suppressedHandlers.has(taskId)) {
|
||||
if (status === TASK_WATCHER_STATUS.READY) handler.onReady(task);
|
||||
else handler.onError(task);
|
||||
}
|
||||
@@ -120,107 +152,141 @@ const settleTask = (
|
||||
if (status === TASK_WATCHER_STATUS.ERROR) {
|
||||
store.dismissTask(taskId);
|
||||
}
|
||||
|
||||
return {
|
||||
status,
|
||||
...(error ? { error } : {}),
|
||||
...(result !== undefined ? { result } : {}),
|
||||
};
|
||||
};
|
||||
|
||||
const runPollLoop = async (taskId: string): Promise<void> => {
|
||||
const runPollLoop = async <R>(
|
||||
taskId: string,
|
||||
): Promise<TaskTrackingResult<R>> => {
|
||||
for (let round = 0; round < MAX_POLL_ROUNDS; round++) {
|
||||
const result = await pollTaskUntilSettled(taskId);
|
||||
const result = await pollTaskUntilSettled<R>(taskId);
|
||||
|
||||
if (result.ok) {
|
||||
if (result.state === "completed") {
|
||||
settleTask(taskId, TASK_WATCHER_STATUS.READY);
|
||||
return settleTask(
|
||||
taskId,
|
||||
TASK_WATCHER_STATUS.READY,
|
||||
undefined,
|
||||
result.result,
|
||||
) as TaskTrackingResult<R>;
|
||||
} else {
|
||||
settleTask(
|
||||
return settleTask(
|
||||
taskId,
|
||||
TASK_WATCHER_STATUS.ERROR,
|
||||
`Task ended in state "${result.state}".`,
|
||||
);
|
||||
result.result,
|
||||
) as TaskTrackingResult<R>;
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// "Task timeout" just means this server round expired while the task
|
||||
// is still running — keep polling. Real errors settle immediately.
|
||||
if (result.error !== "Task timeout") {
|
||||
settleTask(taskId, TASK_WATCHER_STATUS.ERROR, result.error);
|
||||
return;
|
||||
return settleTask(
|
||||
taskId,
|
||||
TASK_WATCHER_STATUS.ERROR,
|
||||
result.error,
|
||||
result.result,
|
||||
) as TaskTrackingResult<R>;
|
||||
}
|
||||
}
|
||||
|
||||
settleTask(
|
||||
return settleTask(
|
||||
taskId,
|
||||
TASK_WATCHER_STATUS.ERROR,
|
||||
"The task is taking too long. Try again later.",
|
||||
);
|
||||
) as TaskTrackingResult<R>;
|
||||
};
|
||||
|
||||
const pollUntilDone = async (taskId: string): Promise<void> => {
|
||||
if (activePolls.has(taskId)) return;
|
||||
activePolls.add(taskId);
|
||||
const pollUntilDone = <R>(taskId: string): Promise<TaskTrackingResult<R>> => {
|
||||
const existingPoll = activePolls.get(taskId);
|
||||
if (existingPoll) return existingPoll as Promise<TaskTrackingResult<R>>;
|
||||
|
||||
try {
|
||||
const runIfPending = async () => {
|
||||
// A different tab may have completed the task while this one waited for
|
||||
// the cross-tab lock. Refresh persisted state before polling or notifying.
|
||||
await useTaskWatcherStore.persist.rehydrate();
|
||||
const task = useTaskWatcherStore.getState().tasks[taskId];
|
||||
if (task?.status !== TASK_WATCHER_STATUS.PENDING) return;
|
||||
await runPollLoop(taskId);
|
||||
};
|
||||
const pollPromise = (async (): Promise<TaskTrackingResult<R>> => {
|
||||
try {
|
||||
const runIfPending = async (): Promise<TaskTrackingResult<R>> => {
|
||||
// A different tab may have completed the task while this one waited for
|
||||
// the cross-tab lock. Refresh persisted state before polling or notifying.
|
||||
await useTaskWatcherStore.persist.rehydrate();
|
||||
const task = useTaskWatcherStore.getState().tasks[taskId];
|
||||
if (task?.status !== TASK_WATCHER_STATUS.PENDING) {
|
||||
return {
|
||||
status: task?.status ?? TASK_WATCHER_STATUS.ERROR,
|
||||
...(task?.error ? { error: task.error } : {}),
|
||||
...(task?.result !== undefined ? { result: task.result as R } : {}),
|
||||
};
|
||||
}
|
||||
return runPollLoop<R>(taskId);
|
||||
};
|
||||
|
||||
if (typeof navigator !== "undefined" && navigator.locks) {
|
||||
await navigator.locks.request(`task-watcher:${taskId}`, runIfPending);
|
||||
} else {
|
||||
await runIfPending();
|
||||
if (typeof navigator !== "undefined" && navigator.locks) {
|
||||
return await navigator.locks.request(
|
||||
`task-watcher:${taskId}`,
|
||||
runIfPending,
|
||||
);
|
||||
}
|
||||
|
||||
return await runIfPending();
|
||||
} catch {
|
||||
// A thrown poll (e.g. the server-action RPC failing on a network drop)
|
||||
// must still settle the task, or it stays PENDING in the persisted
|
||||
// store and blocks the UI until the staleness ceiling.
|
||||
return settleTask(
|
||||
taskId,
|
||||
TASK_WATCHER_STATUS.ERROR,
|
||||
"Tracking the task failed unexpectedly. Try again later.",
|
||||
) as TaskTrackingResult<R>;
|
||||
} finally {
|
||||
activePolls.delete(taskId);
|
||||
}
|
||||
} catch {
|
||||
// A thrown poll (e.g. the server-action RPC failing on a network drop)
|
||||
// must still settle the task, or it stays PENDING in the persisted
|
||||
// store and blocks the UI until the staleness ceiling.
|
||||
settleTask(
|
||||
taskId,
|
||||
TASK_WATCHER_STATUS.ERROR,
|
||||
"Tracking the task failed unexpectedly. Try again later.",
|
||||
);
|
||||
} finally {
|
||||
activePolls.delete(taskId);
|
||||
}
|
||||
})();
|
||||
|
||||
activePolls.set(taskId, pollPromise);
|
||||
return pollPromise;
|
||||
};
|
||||
|
||||
/** Track a freshly dispatched backend task and poll it to completion. The
|
||||
* poll loop lives at module scope (fired from the click handler), so it
|
||||
* survives client-side navigation without any effect subscriptions. */
|
||||
export const trackAndPollTask = async ({
|
||||
export const trackAndPollTask = async <R = unknown>({
|
||||
taskId,
|
||||
kind,
|
||||
meta,
|
||||
}: {
|
||||
taskId: string;
|
||||
kind: string;
|
||||
meta: Record<string, string>;
|
||||
}): Promise<void> => {
|
||||
notifyHandler = true,
|
||||
}: TrackAndPollTaskInput): Promise<TaskTrackingResult<R>> => {
|
||||
if (!notifyHandler) suppressedHandlers.add(taskId);
|
||||
|
||||
const existing = useTaskWatcherStore.getState().tasks[taskId];
|
||||
if (existing?.status === TASK_WATCHER_STATUS.PENDING) {
|
||||
return pollUntilDone(taskId);
|
||||
try {
|
||||
if (existing?.status === TASK_WATCHER_STATUS.PENDING) {
|
||||
return await pollUntilDone<R>(taskId);
|
||||
}
|
||||
|
||||
const store = useTaskWatcherStore.getState();
|
||||
Object.values(store.tasks)
|
||||
.filter(
|
||||
(task) =>
|
||||
task.kind === kind && task.status !== TASK_WATCHER_STATUS.PENDING,
|
||||
)
|
||||
.forEach((task) => store.dismissTask(task.taskId));
|
||||
|
||||
store.upsertTask({
|
||||
taskId,
|
||||
kind,
|
||||
status: TASK_WATCHER_STATUS.PENDING,
|
||||
meta,
|
||||
startedAt: Date.now(),
|
||||
});
|
||||
|
||||
return await pollUntilDone<R>(taskId);
|
||||
} finally {
|
||||
suppressedHandlers.delete(taskId);
|
||||
}
|
||||
|
||||
const store = useTaskWatcherStore.getState();
|
||||
Object.values(store.tasks)
|
||||
.filter(
|
||||
(task) =>
|
||||
task.kind === kind && task.status !== TASK_WATCHER_STATUS.PENDING,
|
||||
)
|
||||
.forEach((task) => store.dismissTask(task.taskId));
|
||||
|
||||
store.upsertTask({
|
||||
taskId,
|
||||
kind,
|
||||
status: TASK_WATCHER_STATUS.PENDING,
|
||||
meta,
|
||||
startedAt: Date.now(),
|
||||
});
|
||||
|
||||
return pollUntilDone(taskId);
|
||||
};
|
||||
|
||||
/** Resume polling every persisted pending task after a hard reload; tasks
|
||||
|
||||
Reference in new issue
Block a user