Run an errors bulk action as one atom run, not one write per issue

apps/web/src/components/errors/issue-batch.test.ts

@@ -0,0 +1,86 @@
+import type { ErrorIssueId } from "@maple/domain/http"
+import { Cause, Effect, Exit } from "effect"
+import * as Reactivity from "effect/unstable/reactivity/Reactivity"
+import { describe, expect, it } from "vitest"
+
+import { BULK_CONCURRENCY, batchOutcome, forEachIssue, ISSUES_KEY, issueKey } from "./issue-batch"
+
+const ids = (count: number) => Array.from({ length: count }, (_, i) => `issue-${i}` as ErrorIssueId)
+
+const run = <A, E>(effect: Effect.Effect<A, E, Reactivity.Reactivity>) =>
+	Effect.runPromise(Effect.provide(effect, Reactivity.layer))
+
+describe("forEachIssue", () => {
+	it("runs every issue and keeps each exit at its issue's position", async () => {
+		const exits = await run(
+			forEachIssue(ids(6), (issueId) =>
+				issueId === "issue-2" ? Effect.fail(`nope ${issueId}`) : Effect.succeed(issueId),
+			),
+		)
+
+		expect(exits).toHaveLength(6)
+		expect(exits.map(Exit.isSuccess)).toEqual([true, true, false, true, true, true])
+		expect(exits[3]).toEqual(Exit.succeed("issue-3"))
+		expect(exits[2]).toEqual(Exit.fail("nope issue-2"))
+	})
+
+	it("runs issues in parallel, bounded by the bulk concurrency", async () => {
+		let inFlight = 0
+		let peak = 0
+		await run(
+			forEachIssue(ids(BULK_CONCURRENCY * 3), () =>
+				Effect.gen(function* () {
+					inFlight += 1
+					yield* Effect.yieldNow
+					peak = Math.max(peak, inFlight)
+					inFlight -= 1
+				}),
+			),
+		)
+
+		expect(peak).toBe(BULK_CONCURRENCY)
+	})
+
+	it("invalidates the list and each issue once, after the whole batch", async () => {
+		const invalidated: Array<string> = []
+		await run(
+			Effect.gen(function* () {
+				const reactivity = yield* Reactivity.Reactivity
+				reactivity.registerUnsafe([ISSUES_KEY], () => invalidated.push(ISSUES_KEY))
+				reactivity.registerUnsafe([issueKey("issue-1" as ErrorIssueId)], () =>
+					invalidated.push(issueKey("issue-1" as ErrorIssueId)),
+				)
+				yield* forEachIssue(ids(3), (issueId) =>
+					Effect.sync(() => {
+						expect(invalidated).toEqual([])
+						return issueId
+					}),
+				)
+			}),
+		)
+
+		expect(invalidated).toEqual([ISSUES_KEY, "errorIssue:issue-1"])
+	})
+})
+
+describe("batchOutcome", () => {
+	it("counts per-issue failures and surfaces the first one", () => {
+		const seen: Array<Exit.Exit<unknown, unknown>> = []
+		const failure = Exit.fail("second")
+		const outcome = batchOutcome(
+			3,
+			Exit.succeed([Exit.succeed(1), failure, Exit.fail("third")]),
+			(exit) => seen.push(exit),
+		)
+
+		expect(outcome).toEqual({ succeeded: 1, failed: 2, firstFailure: failure })
+		expect(seen).toHaveLength(2)
+	})
+
+	it("treats a failed batch as every issue failing", () => {
+		const exit = Exit.failCause(Cause.die(new Error("boom")))
+		const outcome = batchOutcome(4, exit, () => {})
+
+		expect(outcome).toEqual({ succeeded: 0, failed: 4, firstFailure: exit })
+	})
+})

apps/web/src/components/errors/issue-batch.ts

@@ -0,0 +1,49 @@
+import type { ErrorIssueId } from "@maple/domain/http"
+import { Effect, Exit } from "effect"
+import * as Reactivity from "effect/unstable/reactivity/Reactivity"
+
+/** Parallel calls per bulk action; the API serialises each issue's row anyway. */
+export const BULK_CONCURRENCY = 4
+
+export const ISSUES_KEY = "errorIssues"
+export const issueKey = (issueId: ErrorIssueId) => `errorIssue:${issueId}`
+
+/**
+ * One bulk action is ONE atom run, never N writes to a per-issue mutation atom.
+ * An `Atom.fn` is not concurrent: each write interrupts the fiber of the
+ * previous one and every pending `promiseExit` resolves with the LAST result,
+ * so `Promise.all(ids.map(set))` moved one issue and toasted that it moved all.
+ *
+ * Every issue gets its own `Exit`, so a partial failure is reported per row,
+ * and the list is invalidated once at the end rather than once per issue.
+ */
+export const forEachIssue = <A, E, R>(
+	issueIds: ReadonlyArray<ErrorIssueId>,
+	run: (issueId: ErrorIssueId) => Effect.Effect<A, E, R>,
+): Effect.Effect<ReadonlyArray<Exit.Exit<A, E>>, never, R | Reactivity.Reactivity> =>
+	Reactivity.mutation(
+		Effect.forEach(issueIds, (issueId) => Effect.exit(run(issueId)), { concurrency: BULK_CONCURRENCY }),
+		[ISSUES_KEY, ...issueIds.map(issueKey)],
+	)
+
+export interface BatchOutcome {
+	readonly succeeded: number
+	readonly failed: number
+	/** The first failure, for the toast; undefined when every issue succeeded. */
+	readonly firstFailure: Exit.Exit<unknown, unknown> | undefined
+}
+
+/** Folds the per-issue exits (or the batch's own failure) into counts for the toast. */
+export const batchOutcome = (
+	total: number,
+	exit: Exit.Exit<ReadonlyArray<Exit.Exit<unknown, unknown>>, unknown>,
+	onFailure: (failure: Exit.Exit<unknown, unknown>) => void,
+): BatchOutcome => {
+	if (Exit.isFailure(exit)) {
+		onFailure(exit)
+		return { succeeded: 0, failed: total, firstFailure: exit }
+	}
+	const failures = exit.value.filter(Exit.isFailure)
+	failures.forEach(onFailure)
+	return { succeeded: total - failures.length, failed: failures.length, firstFailure: failures[0] }
+}

apps/web/src/components/errors/use-issue-mutations.ts

@@ -1,4 +1,4 @@
-import { Exit } from "effect"
+import { Effect, Exit } from "effect"
 import { toastManager } from "@maple/ui/components/ui/toast"
 import { useAtomSet } from "@/lib/effect-atom"
 import { MapleApiAtomClient } from "@/lib/services/common/atom-client"
@@ -14,8 +14,9 @@ import {
 import { WORKFLOW_LABEL } from "@/components/icons/workflow-ring"
 import { logClientError } from "@/lib/services/common/telemetry"
 import { showErrorToast } from "@/lib/error-toast"
+import { batchOutcome, forEachIssue, ISSUES_KEY, issueKey } from "./issue-batch"
 
-const INVALIDATE = ["errorIssues"] as const
+const INVALIDATE = [ISSUES_KEY] as const
 
 function logFailure(label: string, result: Exit.Exit<unknown, unknown>) {
 	if (Exit.isSuccess(result)) return
@@ -24,6 +25,47 @@ function logFailure(label: string, result: Exit.Exit<unknown, unknown>) {
 	})
 }
 
+// Bulk actions are one atom run each (see `forEachIssue` for why N writes to
+// the per-issue mutation atoms silently moved only one issue).
+const transitionManyAtom = MapleApiAtomClient.runtime.fn<{
+	readonly issueIds: ReadonlyArray<ErrorIssueId>
+	readonly toState: WorkflowState
+}>()(
+	Effect.fnUntraced(function* ({ issueIds, toState }) {
+		const client = yield* MapleApiAtomClient
+		return yield* forEachIssue(issueIds, (issueId) =>
+			client.errors.transitionIssue({
+				params: { issueId },
+				payload: new ErrorIssueTransitionRequest({ toState }),
+			}),
+		)
+	}),
+)
+
+const claimManyAtom = MapleApiAtomClient.runtime.fn<{ readonly issueIds: ReadonlyArray<ErrorIssueId> }>()(
+	Effect.fnUntraced(function* ({ issueIds }) {
+		const client = yield* MapleApiAtomClient
+		return yield* forEachIssue(issueIds, (issueId) =>
+			client.errors.claimIssue({ params: { issueId }, payload: new ErrorIssueClaimRequest({}) }),
+		)
+	}),
+)
+
+const setSeverityManyAtom = MapleApiAtomClient.runtime.fn<{
+	readonly issueIds: ReadonlyArray<ErrorIssueId>
+	readonly severity: IssueSeverity | null
+}>()(
+	Effect.fnUntraced(function* ({ issueIds, severity }) {
+		const client = yield* MapleApiAtomClient
+		return yield* forEachIssue(issueIds, (issueId) =>
+			client.errors.setIssueSeverity({
+				params: { issueId },
+				payload: new ErrorIssueSetSeverityRequest({ severity }),
+			}),
+		)
+	}),
+)
+
 export function useIssueMutations(onSuccess?: () => void) {
 	const transition = useAtomSet(MapleApiAtomClient.mutation("errors", "transitionIssue"), {
 		mode: "promiseExit",
@@ -33,12 +75,15 @@ export function useIssueMutations(onSuccess?: () => void) {
 	const severity = useAtomSet(MapleApiAtomClient.mutation("errors", "setIssueSeverity"), {
 		mode: "promiseExit",
 	})
+	const transitionBatch = useAtomSet(transitionManyAtom, { mode: "promiseExit" })
+	const claimBatch = useAtomSet(claimManyAtom, { mode: "promiseExit" })
+	const severityBatch = useAtomSet(setSeverityManyAtom, { mode: "promiseExit" })
 
 	const transitionTo = async (issueId: ErrorIssueId, toState: WorkflowState) => {
 		const result = await transition({
 			params: { issueId },
 			payload: new ErrorIssueTransitionRequest({ toState }),
-			reactivityKeys: [...INVALIDATE, `errorIssue:${issueId}`],
+			reactivityKeys: [...INVALIDATE, issueKey(issueId)],
 		})
 		if (Exit.isSuccess(result)) {
 			onSuccess?.()
@@ -52,40 +97,32 @@ export function useIssueMutations(onSuccess?: () => void) {
 
 	const transitionMany = async (issueIds: ReadonlyArray<ErrorIssueId>, toState: WorkflowState) => {
 		if (issueIds.length === 0) return
-		const results = await Promise.all(
-			issueIds.map((issueId) =>
-				transition({
-					params: { issueId },
-					payload: new ErrorIssueTransitionRequest({ toState }),
-					reactivityKeys: [...INVALIDATE, `errorIssue:${issueId}`],
-				}),
-			),
+		const exit = await transitionBatch({ issueIds, toState })
+		const outcome = batchOutcome(issueIds.length, exit, (failure) =>
+			logFailure("transitionMany", failure),
 		)
-		const failures = results.filter((r) => !Exit.isSuccess(r))
-		const failed = failures.length
-		failures.forEach((r) => logFailure("transitionMany", r))
-		if (failed === 0) {
+		if (outcome.failed === 0) {
 			onSuccess?.()
 			toastManager.add({
 				title: `Moved ${issueIds.length} to ${WORKFLOW_LABEL[toState]}`,
 				type: "success",
 			})
-		} else if (failed < issueIds.length) {
+		} else if (outcome.succeeded > 0) {
 			onSuccess?.()
-			showErrorToast(failures[0]!, {
-				title: `Moved ${issueIds.length - failed} of ${issueIds.length}; ${failed} failed`,
+			showErrorToast(outcome.firstFailure, {
+				title: `Moved ${outcome.succeeded} of ${issueIds.length}; ${outcome.failed} failed`,
 				type: "warning",
 			})
 		} else {
-			showErrorToast(failures[0]!, { title: "State change failed" })
+			showErrorToast(outcome.firstFailure, { title: "State change failed" })
 		}
 	}
 
 	const claimIssue = async (issueId: ErrorIssueId) => {
 		const result = await claim({
 			params: { issueId },
 			payload: new ErrorIssueClaimRequest({}),
-			reactivityKeys: [...INVALIDATE, `errorIssue:${issueId}`],
+			reactivityKeys: [...INVALIDATE, issueKey(issueId)],
 		})
 		if (Exit.isSuccess(result)) {
 			onSuccess?.()
@@ -99,31 +136,27 @@ export function useIssueMutations(onSuccess?: () => void) {
 
 	const claimMany = async (issueIds: ReadonlyArray<ErrorIssueId>) => {
 		if (issueIds.length === 0) return
-		const results = await Promise.all(
-			issueIds.map((issueId) =>
-				claim({
-					params: { issueId },
-					payload: new ErrorIssueClaimRequest({}),
-					reactivityKeys: [...INVALIDATE, `errorIssue:${issueId}`],
-				}),
-			),
-		)
-		const failed = results.filter((result) => !Exit.isSuccess(result))
-		if (failed.length === 0) {
+		const exit = await claimBatch({ issueIds })
+		const outcome = batchOutcome(issueIds.length, exit, (failure) => logFailure("claimMany", failure))
+		if (outcome.failed === 0) {
 			onSuccess?.()
 			toastManager.add({ title: `Claimed ${issueIds.length} issues`, type: "success" })
-		} else {
-			showErrorToast(failed[0]!, {
-				title: `Claimed ${issueIds.length - failed.length} of ${issueIds.length}`,
+		} else if (outcome.succeeded > 0) {
+			onSuccess?.()
+			showErrorToast(outcome.firstFailure, {
+				title: `Claimed ${outcome.succeeded} of ${issueIds.length}; ${outcome.failed} failed`,
+				type: "warning",
 			})
+		} else {
+			showErrorToast(outcome.firstFailure, { title: "Claim failed" })
 		}
 	}
 
 	const releaseIssue = async (issueId: ErrorIssueId) => {
 		const result = await release({
 			params: { issueId },
 			payload: new ErrorIssueReleaseRequest({}),
-			reactivityKeys: [...INVALIDATE, `errorIssue:${issueId}`],
+			reactivityKeys: [...INVALIDATE, issueKey(issueId)],
 		})
 		if (Exit.isSuccess(result)) {
 			onSuccess?.()
@@ -139,7 +172,7 @@ export function useIssueMutations(onSuccess?: () => void) {
 		const result = await severity({
 			params: { issueId },
 			payload: new ErrorIssueSetSeverityRequest({ severity: value }),
-			reactivityKeys: [...INVALIDATE, `errorIssue:${issueId}`],
+			reactivityKeys: [...INVALIDATE, issueKey(issueId)],
 		})
 		if (Exit.isSuccess(result)) {
 			onSuccess?.()
@@ -156,23 +189,21 @@ export function useIssueMutations(onSuccess?: () => void) {
 
 	const setSeverityMany = async (issueIds: ReadonlyArray<ErrorIssueId>, value: IssueSeverity | null) => {
 		if (issueIds.length === 0) return
-		const results = await Promise.all(
-			issueIds.map((issueId) =>
-				severity({
-					params: { issueId },
-					payload: new ErrorIssueSetSeverityRequest({ severity: value }),
-					reactivityKeys: [...INVALIDATE, `errorIssue:${issueId}`],
-				}),
-			),
+		const exit = await severityBatch({ issueIds, severity: value })
+		const outcome = batchOutcome(issueIds.length, exit, (failure) =>
+			logFailure("setSeverityMany", failure),
 		)
-		const failed = results.filter((result) => !Exit.isSuccess(result))
-		if (failed.length === 0) {
+		if (outcome.failed === 0) {
 			onSuccess?.()
 			toastManager.add({ title: `Updated severity for ${issueIds.length} issues`, type: "success" })
-		} else {
-			showErrorToast(failed[0]!, {
-				title: `Updated ${issueIds.length - failed.length} of ${issueIds.length} issues`,
+		} else if (outcome.succeeded > 0) {
+			onSuccess?.()
+			showErrorToast(outcome.firstFailure, {
+				title: `Updated ${outcome.succeeded} of ${issueIds.length}; ${outcome.failed} failed`,
+				type: "warning",
 			})
+		} else {
+			showErrorToast(outcome.firstFailure, { title: "Severity change failed" })
 		}
 	}