fix(tasks): make bulk status changes atomic

apps/api/src/task/controllers/bulk-update-tasks.ts

@@ -1,4 +1,4 @@
-import { and, eq, inArray } from "drizzle-orm";
+import { and, eq, getTableColumns, inArray, sql } from "drizzle-orm";
 import { HTTPException } from "hono/http-exception";
 import db from "../../database";
 import {
@@ -104,52 +104,70 @@ async function bulkUpdateTasks({
       }
       const projectIds = [...new Set(tasks.map((t) => t.projectId))];
 
+      // Validate every project's destination before the first write.
+      const destinations = new Map<string, string | null>();
       for (const projectId of projectIds) {
         await assertValidTaskStatus(value, projectId);
-
         const column = await db.query.columnTable.findFirst({
           where: and(
             eq(columnTable.projectId, projectId),
             eq(columnTable.slug, value),
           ),
         });
-
-        const projectTasks = tasks.filter(
-          (task) => task.projectId === projectId,
-        );
-        const projectTaskIds = projectTasks.map((task) => task.id);
-
-        const result = await db
-          .update(taskTable)
-          .set({ status: value, columnId: column?.id ?? null })
-          .where(inArray(taskTable.id, projectTaskIds));
-
-        updatedCount += result.rowCount ?? projectTaskIds.length;
-
-        const parentProjects = await getSubtaskParentProjects(projectTaskIds);
-        await publishEvent("subtask-parents.refresh", {
-          projects: parentProjects,
-        });
-
-        for (const task of projectTasks) {
+        destinations.set(projectId, column?.id ?? null);
+      }
+      const updatedTasks = await db.transaction(async (tx) => {
+        const result: (typeof taskTable.$inferSelect)[] = [];
+        for (const projectId of projectIds) {
+          const projectTaskIds = tasks
+            .filter((task) => task.projectId === projectId)
+            .map((task) => task.id);
+          const changed = await tx
+            .update(taskTable)
+            .set({
+              status: value,
+              columnId: destinations.get(projectId) ?? null,
+            })
+            .where(
+              and(
+                inArray(taskTable.id, projectTaskIds),
+                eq(taskTable.projectId, projectId),
+              ),
+            )
+            .returning({
+              ...getTableColumns(taskTable),
+              description: sql<null>`null`,
+            });
+          if (changed.length !== projectTaskIds.length)
+            throw new HTTPException(409, {
+              message: "Tasks changed projects; retry the operation",
+            });
+          result.push(...changed);
+        }
+        return result;
+      });
+      updatedCount = updatedTasks.length;
+      const parentProjects = await getSubtaskParentProjects(foundIds);
+      await publishEvent("subtask-parents.refresh", {
+        projects: parentProjects,
+      });
+      for (const updatedTask of updatedTasks) {
+        const before = tasks.find((task) => task.id === updatedTask.id)!;
+        if (before.status !== updatedTask.status)
           await publishEvent("task.status_changed", {
-            taskId: task.id,
-            projectId,
+            taskId: updatedTask.id,
+            projectId: updatedTask.projectId,
             userId,
-            oldStatus: task.status,
-            newStatus: value,
-            title: task.title,
-            assigneeId: task.userId,
+            oldStatus: before.status,
+            newStatus: updatedTask.status,
+            title: updatedTask.title,
+            assigneeId: updatedTask.userId,
             type: "status_changed",
             skipSubtaskParentRefresh: true,
           });
-        }
-
-        await publishEvent("task-relation.refresh", {
-          projectId,
-          userId,
-        });
       }
+      for (const projectId of projectIds)
+        await publishEvent("task-relation.refresh", { projectId, userId });
       break;
     }
 

tests/api-integration/bulk-task-atomic-status.test.ts

@@ -0,0 +1,42 @@
+import { eq } from "drizzle-orm";
+import { beforeEach, expect, it, vi } from "vite-plus/test";
+import db, { schema } from "../../apps/api/src/database";
+import bulkUpdateTasks from "../../apps/api/src/task/controllers/bulk-update-tasks";
+import { resetTestDatabase } from "./helpers/database";
+import {
+  createProjectFixture,
+  createWorkspaceMember,
+} from "./helpers/fixtures";
+const publish = vi.hoisted(() => vi.fn(async () => undefined));
+vi.mock("../../apps/api/src/events", () => ({ publishEvent: publish }));
+beforeEach(async () => {
+  await resetTestDatabase();
+  vi.clearAllMocks();
+});
+it("validates every destination before committing any project", async () => {
+  const { user, workspace } = await createWorkspaceMember();
+  const a = await createProjectFixture({ workspaceId: workspace.id });
+  const b = await createProjectFixture({ workspaceId: workspace.id });
+  await db
+    .delete(schema.columnTable)
+    .where(eq(schema.columnTable.id, b.columns.inProgress.id));
+  const tasks = await db
+    .insert(schema.taskTable)
+    .values([
+      { projectId: a.project.id, title: "a", status: "to-do" },
+      { projectId: b.project.id, title: "b", status: "to-do" },
+    ])
+    .returning();
+  await expect(
+    bulkUpdateTasks({
+      taskIds: tasks.map((task) => task.id),
+      operation: "updateStatus",
+      value: "in-progress",
+      userId: user.id,
+    }),
+  ).rejects.toThrow("Invalid status");
+  expect(
+    (await db.query.taskTable.findMany()).map((task) => task.status),
+  ).toEqual(["to-do", "to-do"]);
+  expect(publish).not.toHaveBeenCalled();
+});