fix: Add locks to user mutations (#5805)

server/commands/userDemoter.ts

@@ -1,22 +1,28 @@
+import { Transaction } from "sequelize";
 import { UserRole } from "@shared/types";
 import { ValidationError } from "@server/errors";
 import { Event, User } from "@server/models";
 import CleanupDemotedUserTask from "@server/queues/tasks/CleanupDemotedUserTask";
-import { sequelize } from "@server/storage/database";
 
 type Props = {
   user: User;
   actorId: string;
   to: UserRole;
+  transaction?: Transaction;
   ip: string;
 };
 
-export default async function userDemoter({ user, actorId, to, ip }: Props) {
+export default async function userDemoter({
+  user,
+  actorId,
+  to,
+  transaction,
+  ip,
+}: Props) {
   if (user.id === actorId) {
     throw ValidationError("Unable to demote the current user");
   }
 
-  return sequelize.transaction(async (transaction) => {
   await user.demote(to, { transaction });
   await Event.create(
     {
@@ -32,5 +38,4 @@ export default async function userDemoter({ user, actorId, to, ip }: Props) {
     { transaction }
   );
   await CleanupDemotedUserTask.schedule({ userId: user.id });
-  });
 }

server/commands/userSuspender.ts

@@ -1,12 +1,12 @@
 import { Transaction } from "sequelize";
 import { User, Event, GroupUser } from "@server/models";
 import CleanupDemotedUserTask from "@server/queues/tasks/CleanupDemotedUserTask";
-import { sequelize } from "@server/storage/database";
 import { ValidationError } from "../errors";
 
 type Props = {
   user: User;
   actorId: string;
+  transaction?: Transaction;
   ip: string;
 };
 
@@ -17,13 +17,13 @@ type Props = {
 export default async function userSuspender({
   user,
   actorId,
+  transaction,
   ip,
 }: Props): Promise<void> {
   if (user.id === actorId) {
     throw ValidationError("Unable to suspend the current user");
   }
 
-  await sequelize.transaction(async (transaction: Transaction) => {
   await user.update(
     {
       suspendedById: actorId,
@@ -56,5 +56,4 @@ export default async function userSuspender({
   );
 
   await CleanupDemotedUserTask.schedule({ userId: user.id });
-  });
 }

server/commands/userUnsuspender.ts

@@ -1,11 +1,11 @@
 import { Transaction } from "sequelize";
 import { User, Event } from "@server/models";
-import { sequelize } from "@server/storage/database";
 import { ValidationError } from "../errors";
 
 type Props = {
   user: User;
   actorId: string;
+  transaction?: Transaction;
   ip: string;
 };
 
@@ -16,13 +16,13 @@ type Props = {
 export default async function userUnsuspender({
   user,
   actorId,
+  transaction,
   ip,
 }: Props): Promise<void> {
   if (user.id === actorId) {
     throw ValidationError("Unable to unsuspend the current user");
   }
 
-  await sequelize.transaction(async (transaction: Transaction) => {
   await user.update(
     {
       suspendedById: null,
@@ -45,5 +45,4 @@ export default async function userUnsuspender({
       transaction,
     }
   );
-  });
 }

server/models/User.ts

@@ -525,6 +525,7 @@ class User extends ParanoidModel {
         },
       },
       limit: 1,
+      ...options,
     });
 
     if (res.count >= 1) {
@@ -563,11 +564,14 @@ class User extends ParanoidModel {
     }
   };
 
-  promote = () =>
+  promote = (options?: SaveOptions<User>) =>
-    this.update({
+    this.update(
+      {
         isAdmin: true,
         isViewer: false,
-    });
+      },
+      options
+    );
 
   // hooks
 

server/routes/api/users/users.ts

@@ -187,8 +187,8 @@ router.post(
 router.post(
   "users.update",
   auth(),
-  transaction(),
   validate(T.UsersUpdateSchema),
+  transaction(),
   async (ctx: APIContext<T.UsersUpdateReq>) => {
     const { auth, transaction } = ctx.state;
     const actor = auth.user;
@@ -196,7 +196,11 @@ router.post(
 
     let user: User | null = actor;
     if (id) {
-      user = await User.findByPk(id);
+      user = await User.findByPk(id, {
+        rejectOnEmpty: true,
+        transaction,
+        lock: transaction.LOCK.UPDATE,
+      });
     }
     authorize(actor, "update", user);
     const includeDetails = can(actor, "readDetails", user);
@@ -240,15 +244,24 @@ router.post(
   "users.promote",
   auth(),
   validate(T.UsersPromoteSchema),
+  transaction(),
   async (ctx: APIContext<T.UsersPromoteReq>) => {
+    const { transaction } = ctx.state;
     const userId = ctx.input.body.id;
     const actor = ctx.state.auth.user;
     const teamId = actor.teamId;
-    const user = await User.findByPk(userId);
+    const user = await User.findByPk(userId, {
+      rejectOnEmpty: true,
+      transaction,
+      lock: transaction.LOCK.UPDATE,
+    });
     authorize(actor, "promote", user);
 
-    await user.promote();
+    await user.promote({
-    await Event.create({
+      transaction,
+    });
+    await Event.create(
+      {
         name: "users.promote",
         actorId: actor.id,
         userId,
@@ -257,7 +270,11 @@ router.post(
           name: user.name,
         },
         ip: ctx.request.ip,
-    });
+      },
+      {
+        transaction,
+      }
+    );
     const includeDetails = can(actor, "readDetails", user);
 
     ctx.body = {
@@ -273,20 +290,29 @@ router.post(
   "users.demote",
   auth(),
   validate(T.UsersDemoteSchema),
+  transaction(),
   async (ctx: APIContext<T.UsersDemoteReq>) => {
-    const userId = ctx.input.body.id;
+    const { transaction } = ctx.state;
-    const to = ctx.input.body.to;
+    const { to, id: userId } = ctx.input.body;
     const actor = ctx.state.auth.user;
 
     const user = await User.findByPk(userId, {
       rejectOnEmpty: true,
+      transaction,
+      lock: transaction.LOCK.UPDATE,
     });
     authorize(actor, "demote", user);
 
+    await Team.findByPk(user.teamId, {
+      transaction,
+      lock: transaction.LOCK.UPDATE,
+    });
+
     await userDemoter({
       to,
       user,
       actorId: actor.id,
+      transaction,
       ip: ctx.request.ip,
     });
     const includeDetails = can(actor, "readDetails", user);
@@ -304,11 +330,15 @@ router.post(
   "users.suspend",
   auth(),
   validate(T.UsersSuspendSchema),
+  transaction(),
   async (ctx: APIContext<T.UsersSuspendReq>) => {
+    const { transaction } = ctx.state;
     const userId = ctx.input.body.id;
     const actor = ctx.state.auth.user;
     const user = await User.findByPk(userId, {
       rejectOnEmpty: true,
+      transaction,
+      lock: transaction.LOCK.UPDATE,
     });
     authorize(actor, "suspend", user);
 
@@ -316,6 +346,7 @@ router.post(
       user,
       actorId: actor.id,
       ip: ctx.request.ip,
+      transaction,
     });
     const includeDetails = can(actor, "readDetails", user);
 
@@ -332,17 +363,22 @@ router.post(
   "users.activate",
   auth(),
   validate(T.UsersActivateSchema),
+  transaction(),
   async (ctx: APIContext<T.UsersActivateReq>) => {
+    const { transaction } = ctx.state;
     const userId = ctx.input.body.id;
     const actor = ctx.state.auth.user;
     const user = await User.findByPk(userId, {
       rejectOnEmpty: true,
+      transaction,
+      lock: transaction.LOCK.UPDATE,
     });
     authorize(actor, "activate", user);
 
     await userUnsuspender({
       user,
       actorId: actor.id,
+      transaction,
       ip: ctx.request.ip,
     });
     const includeDetails = can(actor, "readDetails", user);
@@ -465,6 +501,8 @@ router.post(
     if (id) {
       user = await User.findByPk(id, {
         rejectOnEmpty: true,
+        transaction,
+        lock: transaction.LOCK.UPDATE,
       });
     } else {
       user = actor;