Fix concurrency issues in Admin Management

src/main/java/dev/vality/orgmanager/repository/MemberRepository.java

@@ -5,6 +5,7 @@
 import dev.vality.orgmanager.service.dto.UserDto;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Modifying;
 import org.springframework.data.jpa.repository.NativeQuery;
 import org.springframework.stereotype.Repository;
 
@@ -15,6 +16,14 @@
 public interface MemberRepository extends JpaRepository<MemberEntity, String> {
 
 
+    /**
+     * Пользователь общий для всех организаций, поэтому создаётся без конфликта по ключу.
+     */
+    @Modifying(flushAutomatically = true)
+    @NativeQuery("INSERT INTO org_manager.member (id, email) VALUES (?1, ?2) " +
+            " ON CONFLICT (id) DO UPDATE SET email = excluded.email")
+    void upsert(String userId, String email);
+
     @NativeQuery("SELECT m.id, " +
             "              m.email,  " +
             "              mr.id as memberRoleId, " +

src/main/java/dev/vality/orgmanager/repository/OrganizationRepository.java

@@ -15,26 +15,26 @@ public interface OrganizationRepository
         extends JpaRepository<OrganizationEntity, String>, JpaSpecificationExecutor<OrganizationEntity> {
 
     @NativeQuery(" SELECT * FROM org_manager.organization AS o " +
-                    " WHERE o.id IN " +
-                    " ( " +
-                    "  SELECT mo.organization_id FROM org_manager.member_to_organization AS mo " +
-                    "   WHERE mo.member_id = ?1 " +
-                    "   UNION " +
-                    "  SELECT id FROM org_manager.organization WHERE owner = ?1 " +
-                    " ) " +
-                    " ORDER BY o.id DESC")
+            " WHERE o.id IN " +
+            " ( " +
+            "  SELECT mo.organization_id FROM org_manager.member_to_organization AS mo " +
+            "   WHERE mo.member_id = ?1 " +
+            "   UNION " +
+            "  SELECT id FROM org_manager.organization WHERE owner = ?1 " +
+            " ) " +
+            " ORDER BY o.id DESC")
     List<OrganizationEntity> findAllByMember(String userId);
 
     @NativeQuery(" SELECT * FROM org_manager.organization AS o " +
-                    " WHERE o.id IN " +
-                    "  ( " +
-                    "    SELECT mo.organization_id FROM org_manager.member_to_organization AS mo " +
-                    "     WHERE mo.member_id = ?1 " +
-                    "      UNION " +
-                    "    SELECT id FROM org_manager.organization WHERE owner = ?1 " +
-                    "   ) " +
-                    " AND o.id < ?2 " +
-                    " ORDER BY o.id DESC")
+            " WHERE o.id IN " +
+            "  ( " +
+            "    SELECT mo.organization_id FROM org_manager.member_to_organization AS mo " +
+            "     WHERE mo.member_id = ?1 " +
+            "      UNION " +
+            "    SELECT id FROM org_manager.organization WHERE owner = ?1 " +
+            "   ) " +
+            " AND o.id < ?2 " +
+            " ORDER BY o.id DESC")
     List<OrganizationEntity> findAllByMember(String userId, String continuationId);
 
     Set<OrganizationEntity> findAllByOwner(String owner);
@@ -43,4 +43,16 @@ public interface OrganizationRepository
 
     Optional<OrganizationEntity> findByParty(String party);
 
+    /**
+     * Блокирует party до конца транзакции: строки для блокировки ещё нет.
+     */
+    @NativeQuery("SELECT 1 FROM (SELECT pg_advisory_xact_lock(hashtext(?1))) AS locked")
+    Integer lockByParty(String party);
+
+    /**
+     * Блокирует строку организации до конца транзакции.
+     */
+    @NativeQuery("SELECT id FROM org_manager.organization WHERE id = ?1 FOR UPDATE")
+    Optional<String> lockById(String organizationId);
+
 }

src/main/java/dev/vality/orgmanager/service/AdminCommonService.java

@@ -29,6 +29,11 @@ public OrganizationEntity findOrganization(String organizationId) throws Organiz
                 .orElseThrow(OrganizationNotFound::new);
     }
 
+    public OrganizationEntity lockOrganization(String organizationId) throws OrganizationNotFound {
+        organizationRepository.lockById(organizationId).orElseThrow(OrganizationNotFound::new);
+        return findOrganization(organizationId);
+    }
+
     public String toStoredMetadata(String metadata) throws InvalidRequest {
         if (metadata == null) {
             return null;

src/main/java/dev/vality/orgmanager/service/AdminInvitationService.java

@@ -62,7 +62,7 @@ public class AdminInvitationService {
     public Invitation create(String organizationId, CreateInvitationRequest request)
             throws OrganizationNotFound, InvalidRequest {
         log.info("Create invitation: organizationId={}, email={}", organizationId, request.getEmail());
-        commonService.findOrganization(organizationId);
+        commonService.lockOrganization(organizationId);
         String email = commonService.requireText(request.getEmail(), "Email");
         String metadata = commonService.toStoredMetadata(request.getMetadata());
         Set<MemberRoleEntity> roles = new LinkedHashSet<>();
@@ -137,7 +137,7 @@ public ListInvitationsResult list(String organizationId, ListInvitationsRequest
     public void revoke(String organizationId, String invitationId, RevokeInvitationRequest request)
             throws OrganizationNotFound, InvitationNotFound, InvalidInvitationState {
         log.info("Revoke invitation: organizationId={}, invitationId={}", organizationId, invitationId);
-        commonService.findOrganization(organizationId);
+        commonService.lockOrganization(organizationId);
         InvitationEntity invitation = findInvitation(organizationId, invitationId);
         InvitationStatus status = converter.effectiveInvitationStatus(invitation);
         if (status != InvitationStatus.pending) {

src/main/java/dev/vality/orgmanager/service/AdminMemberService.java

@@ -10,6 +10,7 @@
 import dev.vality.orgmanagement.MemberRole;
 import dev.vality.orgmanagement.MemberRoleNotFound;
 import dev.vality.orgmanagement.OrganizationNotFound;
+import dev.vality.orgmanagement.RoleScope;
 import dev.vality.orgmanager.converter.AdminManagementConverter;
 import dev.vality.orgmanager.entity.MemberEntity;
 import dev.vality.orgmanager.entity.MemberRoleEntity;
@@ -24,7 +25,6 @@
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
-import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
@@ -82,45 +82,33 @@ public Member add(String organizationId, AddMemberRequest request)
         log.info("Add member: organizationId={}, userId={}", organizationId, request.getUserId());
         String userId = commonService.requireText(request.getUserId(), "User id");
         String email = commonService.requireText(request.getEmail(), "Email");
-        OrganizationEntity organization = commonService.findOrganization(organizationId);
-        MemberEntity member = memberRepository.findById(userId)
-                .orElseGet(() -> MemberEntity.builder()
-                        .id(userId)
-                        .roles(new HashSet<>())
-                        .organizations(new HashSet<>())
-                        .build());
+        OrganizationEntity organization = commonService.lockOrganization(organizationId);
+        memberRepository.upsert(userId, email);
+        MemberEntity member = memberRepository.findById(userId).orElseThrow();
         member.setEmail(email);
-        member = memberRepository.save(member);
-        Set<MemberEntity> members = new HashSet<>(collectionOrEmpty(organization.getMembers()));
-        members.add(member);
-        organization.setMembers(members);
+        Set<MemberEntity> members = organizationMembers(organization);
+        if (!containsMember(members, userId)) {
+            members.add(member);
+        }
         organizationRepository.save(organization);
         return converter.toMember(member, organizationId);
     }
 
     @Transactional
     public void remove(String organizationId, String userId) throws OrganizationNotFound, MemberNotFound {
         log.info("Remove member: organizationId={}, userId={}", organizationId, userId);
-        OrganizationEntity organization = commonService.findOrganization(organizationId);
+        OrganizationEntity organization = commonService.lockOrganization(organizationId);
         MemberEntity member = findMember(organization, userId);
 
-        Set<MemberRoleEntity> retainedRoles = new HashSet<>();
-        List<MemberRoleEntity> removedRoles = new ArrayList<>();
-        for (MemberRoleEntity role : collectionOrEmpty(member.getRoles())) {
-            if (organizationId.equals(role.getOrganizationId()) && role.isActive()) {
-                role.setActive(false);
-                removedRoles.add(role);
-            } else {
-                retainedRoles.add(role);
-            }
-        }
-        member.setRoles(retainedRoles);
+        List<MemberRoleEntity> removedRoles = memberRoles(member).stream()
+                .filter(role -> organizationId.equals(role.getOrganizationId()) && role.isActive())
+                .toList();
+        removedRoles.forEach(role -> role.setActive(false));
+        removedRoles.forEach(memberRoles(member)::remove);
         memberRoleRepository.saveAll(removedRoles);
         memberRepository.save(member);
 
-        Set<MemberEntity> members = new HashSet<>(collectionOrEmpty(organization.getMembers()));
-        members.remove(member);
-        organization.setMembers(members);
+        organizationMembers(organization).removeIf(candidate -> candidate.getId().equals(userId));
         organizationRepository.save(organization);
     }
 
@@ -131,15 +119,17 @@ public MemberRole assignRole(
             AssignMemberRoleRequest request) throws OrganizationNotFound, MemberNotFound, InvalidRequest {
         log.info("Assign member role: organizationId={}, userId={}, roleId={}",
                 organizationId, userId, request.getRoleId());
-        OrganizationEntity organization = commonService.findOrganization(organizationId);
+        OrganizationEntity organization = commonService.lockOrganization(organizationId);
         MemberEntity member = findMember(organization, userId);
         commonService.validateRoleAssignment(request.getRoleId(), request.getScope());
+        MemberRoleEntity assigned = findAssignedRole(member, organizationId, request);
+        if (assigned != null) {
+            return converter.toMemberRole(assigned);
+        }
         MemberRoleEntity role = commonService.toMemberRoleEntity(
                 organizationId, request.getRoleId(), request.getScope());
         role = memberRoleRepository.save(role);
-        Set<MemberRoleEntity> roles = new HashSet<>(collectionOrEmpty(member.getRoles()));
-        roles.add(role);
-        member.setRoles(roles);
+        memberRoles(member).add(role);
         memberRepository.save(member);
         return converter.toMemberRole(role);
     }
@@ -149,7 +139,7 @@ public void removeRole(String organizationId, String userId, String memberRoleId
             throws OrganizationNotFound, MemberNotFound, MemberRoleNotFound {
         log.info("Remove member role: organizationId={}, userId={}, memberRoleId={}",
                 organizationId, userId, memberRoleId);
-        OrganizationEntity organization = commonService.findOrganization(organizationId);
+        OrganizationEntity organization = commonService.lockOrganization(organizationId);
         MemberEntity member = findMember(organization, userId);
         MemberRoleEntity role = collectionOrEmpty(member.getRoles()).stream()
                 .filter(candidate -> candidate.getId().equals(memberRoleId))
@@ -158,13 +148,55 @@ public void removeRole(String organizationId, String userId, String memberRoleId
                 .findFirst()
                 .orElseThrow(MemberRoleNotFound::new);
         role.setActive(false);
-        Set<MemberRoleEntity> roles = new HashSet<>(collectionOrEmpty(member.getRoles()));
-        roles.remove(role);
-        member.setRoles(roles);
+        memberRoles(member).remove(role);
         memberRoleRepository.save(role);
         memberRepository.save(member);
     }
 
+    /**
+     * Уже назначенная роль с той же областью действия, если она есть.
+     */
+    private MemberRoleEntity findAssignedRole(
+            MemberEntity member,
+            String organizationId,
+            AssignMemberRoleRequest request) {
+        RoleScope scope = request.getScope();
+        return memberRoles(member).stream()
+                .filter(MemberRoleEntity::isActive)
+                .filter(role -> organizationId.equals(role.getOrganizationId()))
+                .filter(role -> role.getRoleId().equals(request.getRoleId()))
+                .filter(role -> scope == null
+                        ? role.getScopeId() == null
+                        : scope.getScopeId().equals(role.getScopeId())
+                        && scope.getResourceId().equals(role.getResourceId()))
+                .findFirst()
+                .orElse(null);
+    }
+
+    /**
+     * Коллекция меняется на месте: подстановка новой пересоздаёт связующую таблицу целиком.
+     */
+    private Set<MemberRoleEntity> memberRoles(MemberEntity member) {
+        if (member.getRoles() == null) {
+            member.setRoles(new HashSet<>());
+        }
+        return member.getRoles();
+    }
+
+    /**
+     * Поиск по идентификатору: equals участника учитывает email.
+     */
+    private boolean containsMember(Set<MemberEntity> members, String userId) {
+        return members.stream().anyMatch(member -> member.getId().equals(userId));
+    }
+
+    private Set<MemberEntity> organizationMembers(OrganizationEntity organization) {
+        if (organization.getMembers() == null) {
+            organization.setMembers(new HashSet<>());
+        }
+        return organization.getMembers();
+    }
+
     private MemberEntity findMember(OrganizationEntity organization, String userId) throws MemberNotFound {
         return collectionOrEmpty(organization.getMembers()).stream()
                 .filter(member -> member.getId().equals(userId))

src/main/java/dev/vality/orgmanager/service/AdminOrganizationService.java

@@ -17,7 +17,6 @@
 import jakarta.persistence.criteria.Predicate;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
@@ -49,11 +48,12 @@ public class AdminOrganizationService {
     @Transactional
     public Organization create(CreateOrganizationRequest request) throws PartyAlreadyBound, InvalidRequest {
         log.info("Create organization: partyId={}, ownerId={}", request.getPartyId(), request.getOwnerId());
-        commonService.requireText(request.getPartyId(), "Party id");
+        String partyId = commonService.requireText(request.getPartyId(), "Party id");
         commonService.requireText(request.getOwnerId(), "Owner id");
         String name = commonService.requireText(request.getName(), "Organization name");
         String metadata = commonService.toStoredMetadata(request.getMetadata());
-        if (organizationRepository.existsByParty(request.getPartyId())) {
+        organizationRepository.lockByParty(partyId);
+        if (organizationRepository.existsByParty(partyId)) {
             throw new PartyAlreadyBound();
         }
         OrganizationEntity entity = OrganizationEntity.builder()
@@ -67,11 +67,7 @@ public Organization create(CreateOrganizationRequest request) throws PartyAlread
                 .members(new HashSet<>())
                 .roles(new HashSet<>())
                 .build();
-        try {
-            return converter.toOrganization(organizationRepository.saveAndFlush(entity));
-        } catch (DataIntegrityViolationException exception) {
-            throw new PartyAlreadyBound();
-        }
+        return converter.toOrganization(organizationRepository.saveAndFlush(entity));
     }
 
     @Transactional(readOnly = true)
@@ -108,7 +104,7 @@ public Organization modify(String organizationId, ModifyOrganizationRequest requ
             throws OrganizationNotFound, InvalidRequest {
         log.info("Modify organization: organizationId={}, request={}", organizationId, request);
         ModifyOrganizationRequest safeRequest = request == null ? new ModifyOrganizationRequest() : request;
-        OrganizationEntity organization = commonService.findOrganization(organizationId);
+        OrganizationEntity organization = commonService.lockOrganization(organizationId);
         if (safeRequest.isSetName()) {
             organization.setName(commonService.requireText(safeRequest.getName(), "Organization name"));
         }
@@ -150,7 +146,7 @@ private Organization changeStatus(
             String organizationId,
             StoredOrganizationStatus expected,
             StoredOrganizationStatus target) throws OrganizationNotFound, InvalidOrganizationState {
-        OrganizationEntity organization = commonService.findOrganization(organizationId);
+        OrganizationEntity organization = commonService.lockOrganization(organizationId);
         String current = organization.getStatus();
         if (!expected.matches(current)) {
             throw new InvalidOrganizationState(

src/test/java/dev/vality/orgmanager/repository/AbstractRepositoryTest.java

@@ -48,7 +48,7 @@ public void setUp() throws Exception {
         invitationRepository.deleteAll();
         memberContextRepository.deleteAll();
         organizationRoleRepository.deleteAll();
-        // Связи с каскадом PERSIST иначе отменяют удаление при flush
+        // Без разрыва связей каскад PERSIST отменяет удаление при flush
         var organizations = organizationRepository.findAll();
         organizations.forEach(it -> it.getMembers().clear());
         organizationRepository.saveAll(organizations);