/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.ambari.server.security.authorization;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.regex.Pattern;

import jakarta.persistence.EntityManager;
import jakarta.persistence.OptimisticLockException;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.hooks.HookContextFactory;
import org.apache.ambari.server.hooks.HookService;
import org.apache.ambari.server.ldap.domain.AmbariLdapConfiguration;
import org.apache.ambari.server.ldap.service.AmbariLdapConfigurationProvider;
import org.apache.ambari.server.orm.dao.GroupDAO;
import org.apache.ambari.server.orm.dao.MemberDAO;
import org.apache.ambari.server.orm.dao.PermissionDAO;
import org.apache.ambari.server.orm.dao.PrincipalDAO;
import org.apache.ambari.server.orm.dao.PrincipalTypeDAO;
import org.apache.ambari.server.orm.dao.PrivilegeDAO;
import org.apache.ambari.server.orm.dao.ResourceDAO;
import org.apache.ambari.server.orm.dao.UserAuthenticationDAO;
import org.apache.ambari.server.orm.dao.UserDAO;
import org.apache.ambari.server.orm.entities.GroupEntity;
import org.apache.ambari.server.orm.entities.MemberEntity;
import org.apache.ambari.server.orm.entities.PermissionEntity;
import org.apache.ambari.server.orm.entities.PrincipalEntity;
import org.apache.ambari.server.orm.entities.PrincipalTypeEntity;
import org.apache.ambari.server.orm.entities.PrivilegeEntity;
import org.apache.ambari.server.orm.entities.ResourceTypeEntity;
import org.apache.ambari.server.orm.entities.UserAuthenticationEntity;
import org.apache.ambari.server.orm.entities.UserEntity;
import org.apache.ambari.server.security.authentication.AccountDisabledException;
import org.apache.ambari.server.security.authentication.TooManyLoginFailuresException;
import org.apache.ambari.server.security.authentication.UserNotFoundException;
import org.apache.ambari.server.security.ldap.LdapBatchDto;
import org.apache.ambari.server.security.ldap.LdapGroupDto;
import org.apache.ambari.server.security.ldap.LdapUserDto;
import org.apache.ambari.server.security.ldap.LdapUserGroupMemberDto;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.security.crypto.password.PasswordEncoder;

import com.google.common.base.Strings;
import com.google.common.collect.Sets;
import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.Singleton;
import com.google.inject.persist.Transactional;

/**
 * Provides high-level access to Users and Roles in database
 */
@Singleton
public class Users {

  private static final Logger LOG = LoggerFactory.getLogger(Users.class);

  /**
   * The maximum number of retries when handling OptimisticLockExceptions
   */
  private static final int MAX_RETRIES = 10;

  @Inject
  private Provider<EntityManager> entityManagerProvider;

  @Inject
  private UserDAO userDAO;

  @Inject
  private UserAuthenticationDAO userAuthenticationDAO;

  @Inject
  private GroupDAO groupDAO;

  @Inject
  private MemberDAO memberDAO;

  @Inject
  private PrincipalDAO principalDAO;

  @Inject
  private PermissionDAO permissionDAO;

  @Inject
  private PrivilegeDAO privilegeDAO;

  @Inject
  private ResourceDAO resourceDAO;

  @Inject
  private PrincipalTypeDAO principalTypeDAO;

  @Inject
  private PasswordEncoder passwordEncoder;

  @Inject
  protected AmbariLdapConfigurationProvider ldapConfigurationProvider;

  @Inject
  protected Configuration configuration;

  @Inject
  private Provider<HookService> hookServiceProvider;

  @Inject
  private HookContextFactory hookContextFactory;

  public List<User> getAllUsers() {
    List<UserEntity> userEntities = userDAO.findAll();
    List<User> users = new ArrayList<>(userEntities.size());

    for (UserEntity userEntity : userEntities) {
      users.add(new User(userEntity));
    }

    return users;
  }

  public List<UserEntity> getAllUserEntities() {
    return userDAO.findAll();
  }

  public UserEntity getUserEntity(String userName) {
    return (userName == null) ? null : userDAO.findUserByName(userName);
  }

  public UserEntity getUserEntity(Integer userId) {
    return (userId == null) ? null : userDAO.findByPK(userId);
  }

  public User getUser(UserEntity userEntity) {
    return (null == userEntity) ? null : new User(userEntity);
  }

  public User getUser(Integer userId) {
    return getUser(getUserEntity(userId));
  }

  public User getUser(String userName) {
    return getUser(getUserEntity(userName));
  }

  /**
   * Enables/disables user.
   *
   * @param userName user name
   * @param active   true if active; false if not active
   * @throws AmbariException if user does not exist
   */
  public synchronized void setUserActive(String userName, boolean active) throws AmbariException {
    UserEntity userEntity = userDAO.findUserByName(userName);
    if (userEntity != null) {
      setUserActive(userEntity, active);
    } else {
      throw new AmbariException("User " + userName + " doesn't exist");
    }
  }

  /**
   * Enables/disables user.
   *
   * @param userEntity the user
   * @param active     true if active; false if not active
   * @throws AmbariException if user does not exist
   */
  public synchronized void setUserActive(UserEntity userEntity, final boolean active) throws AmbariException {
    if (userEntity != null) {
      Command command = new Command() {
        @Override
        public void perform(UserEntity userEntity) {
          userEntity.setActive(active);
        }
      };

      safelyUpdateUserEntity(userEntity, command, MAX_RETRIES);
    }
  }

  /**
   * Validates the user account such that the user is allowed to log in.
   *
   * @param userEntity the user entity
   * @param userName   the Ambari username
   */
  public void validateLogin(UserEntity userEntity, String userName) {
    if (userEntity == null) {
      LOG.info("User not found");
      throw new UserNotFoundException(userName);
    } else {
      if (!userEntity.getActive()) {
        LOG.info("User account is disabled: {}", userName);
        throw new AccountDisabledException(userName);
      }

      int maxConsecutiveFailures = configuration.getMaxAuthenticationFailures();
      if (maxConsecutiveFailures > 0 && userEntity.getConsecutiveFailures() >= maxConsecutiveFailures) {
        LOG.info("User account is locked out due to too many authentication failures ({}/{}): {}",
            userEntity.getConsecutiveFailures(), maxConsecutiveFailures, userName);
        throw new TooManyLoginFailuresException(userName);
      }
    }
  }

  /**
   * Converts group to LDAP group.
   *
   * @param groupName group name
   * @throws AmbariException if group does not exist
   */
  public synchronized void setGroupLdap(String groupName) throws AmbariException {
    GroupEntity groupEntity = groupDAO.findGroupByName(groupName);
    if (groupEntity != null) {
      groupEntity.setGroupType(GroupType.LDAP);
      groupDAO.merge(groupEntity);
    } else {
      throw new AmbariException("Group " + groupName + " doesn't exist");
    }
  }

  /**
   * Creates new, active, user with provided userName, local username, and display name.
   *
   * @param userName      user name
   * @param localUserName the local username to use; if <code>null</code> or empty, userName will be used
   * @param displayName   the name to display for presentation; if <code>null</code> or empty, userName will be used
   * @return the new UserEntity
   * @throws AmbariException if user already exists
   */
  public UserEntity createUser(String userName, String localUserName, String displayName) throws AmbariException {
    return createUser(userName, localUserName, displayName, true);
  }

  /**
   * Creates new, user with provided userName, local username, and display name.
   *
   * @param userName      user name
   * @param localUserName the local username to use; if <code>null</code> or empty, userName will be used
   * @param displayName   the name to display for presentation; if <code>null</code> or empty, userName will be used
   * @param active        is user active
   * @return the new UserEntity
   * @throws AmbariException if user already exists
   */
  @Transactional
  public synchronized UserEntity createUser(String userName, String localUserName, String displayName, Boolean active) throws AmbariException {

    String validatedUserName = UserName.fromString(userName).toString();
    String validatedDisplayName = (StringUtils.isEmpty(displayName))
        ? validatedUserName
        : UserName.fromString(displayName).toString();
    String validatedLocalUserName = (StringUtils.isEmpty(localUserName))
        ? validatedUserName
        : UserName.fromString(localUserName).toString();

    // Ensure that the user does not already exist
    if (userDAO.findUserByName(validatedUserName) != null) {
      throw new AmbariException("User already exists");
    }

    // Create the PrincipalEntity - needed for assigning privileges/roles
    PrincipalTypeEntity principalTypeEntity = principalTypeDAO.findById(PrincipalTypeEntity.USER_PRINCIPAL_TYPE);
    if (principalTypeEntity == null) {
      principalTypeEntity = new PrincipalTypeEntity();
      principalTypeEntity.setId(PrincipalTypeEntity.USER_PRINCIPAL_TYPE);
      principalTypeEntity.setName(PrincipalTypeEntity.USER_PRINCIPAL_TYPE_NAME);
      principalTypeDAO.create(principalTypeEntity);
    }
    PrincipalEntity principalEntity = new PrincipalEntity();
    principalEntity.setPrincipalType(principalTypeEntity);
    principalDAO.create(principalEntity);

    // Create the new UserEntity Record
    UserEntity userEntity = new UserEntity();
    userEntity.setUserName(validatedUserName);
    userEntity.setDisplayName(validatedDisplayName);
    userEntity.setLocalUsername(validatedLocalUserName);

    userEntity.setPrincipal(principalEntity);
    if (active != null) {
      userEntity.setActive(active);
    }

    userDAO.create(userEntity);

    // execute user initialization hook if required ()
    executeUserHook(validatedUserName);

    return userEntity;
  }

  /**
   * Triggers the post user creation hook, if enabled
   *
   * @param username the username of the user to process
   */
  public void executeUserHook(String username) {
    hookServiceProvider.get().execute(hookContextFactory.createUserHookContext(username));
  }

  /**
   * Triggers the post user creation hook, if enabled
   *
   * @param userGroupsMap a map of user names to relevant groups
   */
  public void executeUserHook(Map<String, Set<String>> userGroupsMap) {
    hookServiceProvider.get().execute(hookContextFactory.createBatchUserHookContext(userGroupsMap));
  }

  /**
   * Removes a user from the Ambari database.
   * <p>
   * It is expected that the associated user authentication records are removed by this operation
   * as well.
   *
   * @param user the user to remove
   * @throws AmbariException
   */
  @Transactional
  public synchronized void removeUser(User user) throws AmbariException {
    UserEntity userEntity = userDAO.findByPK(user.getUserId());
    if (userEntity != null) {
      removeUser(userEntity);
    } else {
      throw new AmbariException("User " + user + " doesn't exist");
    }
  }

  /**
   * Removes a user from the Ambari database.
   * <p>
   * It is expected that the associated user authentication records are removed by this operation
   * as well.
   *
   * @param userEntity the user to remove
   * @throws AmbariException
   */
  @Transactional
  public synchronized void removeUser(UserEntity userEntity) throws AmbariException {
    if (userEntity != null) {
      if (!isUserCanBeRemoved(userEntity)) {
        throw new AmbariException("Could not remove user " + userEntity.getUserName() +
          ". System should have at least one administrator.");
      }
      userDAO.remove(userEntity);
    }
  }

  /**
   * Gets group by given name.
   *
   * @param groupName group name
   * @return group
   */
  public Group getGroup(String groupName) {
    final GroupEntity groupEntity = groupDAO.findGroupByName(groupName);
    return (null == groupEntity) ? null : new Group(groupEntity);
  }

  /**
   * Gets group by given name & type.
   *
   * @param groupName group name
   * @param groupType group type
   * @return group
   */
  public Group getGroup(String groupName, GroupType groupType) {
    final GroupEntity groupEntity = getGroupEntity(groupName, groupType);
    return (null == groupEntity) ? null : new Group(groupEntity);
  }

  /**
   * Gets a {@link GroupEntity} by name and type.
   *
   * @param groupName group name
   * @param groupType group type
   * @return group
   */
  public GroupEntity getGroupEntity(String groupName, GroupType groupType) {
    return groupDAO.findGroupByNameAndType(groupName, groupType);
  }

  /**
   * Gets group members.
   *
   * @param groupName group name
   * @return list of members
   */
  public Collection<User> getGroupMembers(String groupName) {
    final GroupEntity groupEntity = groupDAO.findGroupByName(groupName);
    if (groupEntity == null) {
      return null;
    } else {
      final Set<User> users = new HashSet<>();
      for (MemberEntity memberEntity : groupEntity.getMemberEntities()) {
        if (memberEntity.getUser() != null) {
          users.add(new User(memberEntity.getUser()));
        } else {
          LOG.error("Wrong state, not found user for member '{}' (group: '{}')",
            memberEntity.getMemberId(), memberEntity.getGroup().getGroupName());
        }
      }
      return users;
    }
  }

  /**
   * Creates new group with provided name & type
   */
  @Transactional
  public synchronized GroupEntity createGroup(String groupName, GroupType groupType) {
    // create an admin principal to represent this group
    PrincipalTypeEntity principalTypeEntity = principalTypeDAO.findById(PrincipalTypeEntity.GROUP_PRINCIPAL_TYPE);
    if (principalTypeEntity == null) {
      principalTypeEntity = new PrincipalTypeEntity();
      principalTypeEntity.setId(PrincipalTypeEntity.GROUP_PRINCIPAL_TYPE);
      principalTypeEntity.setName(PrincipalTypeEntity.GROUP_PRINCIPAL_TYPE_NAME);
      principalTypeDAO.create(principalTypeEntity);
    }
    PrincipalEntity principalEntity = new PrincipalEntity();
    principalEntity.setPrincipalType(principalTypeEntity);
    principalDAO.create(principalEntity);

    final GroupEntity groupEntity = new GroupEntity();
    groupEntity.setGroupName(groupName);
    groupEntity.setPrincipal(principalEntity);
    groupEntity.setGroupType(groupType);

    groupDAO.create(groupEntity);
    return groupEntity;
  }

  /**
   * Gets all groups.
   *
   * @return list of groups
   */
  public List<Group> getAllGroups() {
    final List<GroupEntity> groupEntities = groupDAO.findAll();
    final List<Group> groups = new ArrayList<>(groupEntities.size());

    for (GroupEntity groupEntity : groupEntities) {
      groups.add(new Group(groupEntity));
    }

    return groups;
  }

  /**
   * Gets all members of a group specified.
   *
   * @param groupName group name
   * @return list of user names
   */
  public List<String> getAllMembers(String groupName) throws AmbariException {
    final List<String> members = new ArrayList<>();
    final GroupEntity groupEntity = groupDAO.findGroupByName(groupName);
    if (groupEntity == null) {
      throw new AmbariException("Group " + groupName + " doesn't exist");
    }
    for (MemberEntity member : groupEntity.getMemberEntities()) {
      members.add(member.getUser().getUserName());
    }
    return members;
  }

  @Transactional
  public synchronized void removeGroup(Group group) throws AmbariException {
    final GroupEntity groupEntity = groupDAO.findByPK(group.getGroupId());
    if (groupEntity != null) {
      groupDAO.remove(groupEntity);
    } else {
      throw new AmbariException("Group " + group + " doesn't exist");
    }
  }

  /**
   * Test the user for Ambari Admistrator privileges.
   *
   * @param userEntity the user to test
   * @return true if the user has Ambari Administrator privileges; otherwise false
   */
  public synchronized boolean hasAdminPrivilege(UserEntity userEntity) {
    PrincipalEntity principalEntity = userEntity.getPrincipal();
    if (principalEntity != null) {
      Set<PrivilegeEntity> roles = principalEntity.getPrivileges();
      if (roles != null) {
        PermissionEntity adminPermission = permissionDAO.findAmbariAdminPermission();
        Integer adminPermissionId = (adminPermission == null) ? null : adminPermission.getId();

        if (adminPermissionId != null) {
          for (PrivilegeEntity privilegeEntity : roles) {
            PermissionEntity rolePermission = privilegeEntity.getPermission();
            if ((rolePermission != null) && (adminPermissionId.equals(rolePermission.getId()))) {
              return true;
            }
          }
        }
      }
    }

    return false;
  }

  /**
   * Grants Ambari Administrator privilege to provided user.
   *
   * @param userId user id
   */
  public synchronized void grantAdminPrivilege(Integer userId) {
    grantAdminPrivilege(userDAO.findByPK(userId));
  }

  /**
   * Grants Ambari Administrator privilege to provided user.
   *
   * @param userEntity the user
   */
  public synchronized void grantAdminPrivilege(UserEntity userEntity) {
    final PrivilegeEntity adminPrivilege = new PrivilegeEntity();
    adminPrivilege.setPermission(permissionDAO.findAmbariAdminPermission());
    adminPrivilege.setPrincipal(userEntity.getPrincipal());
    adminPrivilege.setResource(resourceDAO.findAmbariResource());
    if (!userEntity.getPrincipal().getPrivileges().contains(adminPrivilege)) {
      privilegeDAO.create(adminPrivilege);
      userEntity.getPrincipal().getPrivileges().add(adminPrivilege);
      principalDAO.merge(userEntity.getPrincipal()); //explicit merge for Derby support
      userDAO.merge(userEntity);
    }
  }

  /**
   * Grants privilege to provided group.
   *
   * @param groupId        group id
   * @param resourceId     resource id
   * @param resourceType   resource type
   * @param permissionName permission name
   */
  public synchronized void grantPrivilegeToGroup(Integer groupId, Long resourceId, ResourceType resourceType, String permissionName) {
    final GroupEntity group = groupDAO.findByPK(groupId);
    final PrivilegeEntity privilege = new PrivilegeEntity();
    ResourceTypeEntity resourceTypeEntity = new ResourceTypeEntity();
    resourceTypeEntity.setId(resourceType.getId());
    resourceTypeEntity.setName(resourceType.name());
    privilege.setPermission(permissionDAO.findPermissionByNameAndType(permissionName, resourceTypeEntity));
    privilege.setPrincipal(group.getPrincipal());
    privilege.setResource(resourceDAO.findById(resourceId));
    if (!group.getPrincipal().getPrivileges().contains(privilege)) {
      privilegeDAO.create(privilege);
      group.getPrincipal().getPrivileges().add(privilege);
      principalDAO.merge(group.getPrincipal()); //explicit merge for Derby support
      groupDAO.merge(group);
      privilegeDAO.merge(privilege);
    }
  }

  /**
   * Revokes Ambari Administrator privileges from provided user.
   *
   * @param userId user id
   */
  public synchronized void revokeAdminPrivilege(Integer userId) {
    revokeAdminPrivilege(userDAO.findByPK(userId));
  }

  /**
   * Revokes Ambari Administrator privileges from provided user.
   *
   * @param userEntity the user
   */
  public synchronized void revokeAdminPrivilege(UserEntity userEntity) {
    for (PrivilegeEntity privilege : userEntity.getPrincipal().getPrivileges()) {
      if (privilege.getPermission().getPermissionName().equals(PermissionEntity.AMBARI_ADMINISTRATOR_PERMISSION_NAME)) {
        userEntity.getPrincipal().getPrivileges().remove(privilege);
        principalDAO.merge(userEntity.getPrincipal()); //explicit merge for Derby support
        userDAO.merge(userEntity);
        privilegeDAO.remove(privilege);
        break;
      }
    }
  }

  @Transactional
  public synchronized void addMemberToGroup(String groupName, String userName)
    throws AmbariException {

    final GroupEntity groupEntity = groupDAO.findGroupByName(groupName);
    if (groupEntity == null) {
      throw new AmbariException("Group " + groupName + " doesn't exist");
    }

    UserEntity userEntity = userDAO.findUserByName(userName);
    if (userEntity == null) {
      throw new AmbariException("User " + userName + " doesn't exist");
    }

    addMemberToGroup(groupEntity, userEntity);
  }

  @Transactional
  public synchronized void addMemberToGroup(GroupEntity groupEntity, UserEntity userEntity)
      throws AmbariException {

    if (groupEntity == null) {
      throw new NullPointerException();
    }

    if (userEntity == null) {
      throw new NullPointerException();
    }

    if (!isUserInGroup(userEntity, groupEntity)) {
      final MemberEntity memberEntity = new MemberEntity();
      memberEntity.setGroup(groupEntity);
      memberEntity.setUser(userEntity);
      userEntity.getMemberEntities().add(memberEntity);
      groupEntity.getMemberEntities().add(memberEntity);
      memberDAO.create(memberEntity);
      userDAO.merge(userEntity);
      groupDAO.merge(groupEntity);
    }
  }

  @Transactional
  public synchronized void removeMemberFromGroup(String groupName, String userName)
    throws AmbariException {

    final GroupEntity groupEntity = groupDAO.findGroupByName(groupName);
    if (groupEntity == null) {
      throw new AmbariException("Group " + groupName + " doesn't exist");
    }

    UserEntity userEntity = userDAO.findUserByName(userName);
    if (userEntity == null) {
      throw new AmbariException("User " + userName + " doesn't exist");
    }

    removeMemberFromGroup(groupEntity, userEntity);
  }

  @Transactional
  public synchronized void removeMemberFromGroup(GroupEntity groupEntity, UserEntity userEntity)
      throws AmbariException {

    if (isUserInGroup(userEntity, groupEntity)) {
      MemberEntity memberEntity = null;
      for (MemberEntity entity : userEntity.getMemberEntities()) {
        if (entity.getGroup().equals(groupEntity)) {
          memberEntity = entity;
          break;
        }
      }
      userEntity.getMemberEntities().remove(memberEntity);
      groupEntity.getMemberEntities().remove(memberEntity);
      userDAO.merge(userEntity);
      groupDAO.merge(groupEntity);
      memberDAO.remove(memberEntity);
    }
  }

  /**
   * Performs a check if the user can be removed. Do not allow removing all admins from database.
   *
   * @param userEntity user to be checked
   * @return true if user can be removed
   */
  public synchronized boolean isUserCanBeRemoved(UserEntity userEntity) {
    List<PrincipalEntity> adminPrincipals = principalDAO.findByPermissionId(PermissionEntity.AMBARI_ADMINISTRATOR_PERMISSION);
    Set<UserEntity> userEntitysSet = new HashSet<>(userDAO.findUsersByPrincipal(adminPrincipals));
    return (userEntitysSet.contains(userEntity) && userEntitysSet.size() < 2) ? false : true;
  }

  /**
   * Performs a check if given user belongs to given group.
   *
   * @param userEntity  user entity
   * @param groupEntity group entity
   * @return true if user presents in group
   */
  public boolean isUserInGroup(UserEntity userEntity, GroupEntity groupEntity) {
    for (MemberEntity memberEntity : userEntity.getMemberEntities()) {
      if (memberEntity.getGroup().equals(groupEntity)) {
        return true;
      }
    }
    return false;
  }

  /**
   * Executes batch queries to database to insert large amounts of LDAP data.
   *
   * @param batchInfo DTO with batch information
   */
  public void processLdapSync(LdapBatchDto batchInfo) {
    final Map<String, UserEntity> allUsers = new HashMap<>();
    final Map<String, GroupEntity> allGroups = new HashMap<>();

    // prefetch all user and group data to avoid heavy queries in membership creation

    for (UserEntity userEntity : userDAO.findAll()) {
      allUsers.put(userEntity.getUserName(), userEntity);
    }

    for (GroupEntity groupEntity : groupDAO.findAll()) {
      allGroups.put(groupEntity.getGroupName(), groupEntity);
    }

    final PrincipalTypeEntity groupPrincipalType = principalTypeDAO
        .ensurePrincipalTypeCreated(PrincipalTypeEntity.GROUP_PRINCIPAL_TYPE);

    /* *****
     * Remove users
     *   First remove the relevant LDAP entries for this user.
     *   If no more user authentication items exists for the user, then remove the user.
     * ***** */
    final Set<UserEntity> usersToRemove = new HashSet<>();
    final Set<UserAuthenticationEntity> authenticationEntitiesToRemove = new HashSet<>();
    for (LdapUserDto user : batchInfo.getUsersToBeRemoved()) {
      UserEntity userEntity = userDAO.findUserByName(user.getUserName());
      if (userEntity != null) {
        List<UserAuthenticationEntity> authenticationEntities = userAuthenticationDAO.findByUser(userEntity);
        Iterator<UserAuthenticationEntity> iterator = authenticationEntities.iterator();
        while (iterator.hasNext()) {
          UserAuthenticationEntity authenticationEntity = iterator.next();

          if (authenticationEntity.getAuthenticationType() == UserAuthenticationType.LDAP) {
            String dn = user.getDn();
            String authenticationKey = authenticationEntity.getAuthenticationKey();

            // DN's are case-insensitive.
            if (StringUtils.isEmpty(dn) || StringUtils.isEmpty(authenticationKey) || dn.equalsIgnoreCase(authenticationKey)) {
              authenticationEntitiesToRemove.add(authenticationEntity);
            }
            iterator.remove();
          }
        }

        if (authenticationEntities.isEmpty()) {
          allUsers.remove(userEntity.getUserName());
          usersToRemove.add(userEntity);
        }
      }
    }
    userAuthenticationDAO.remove(authenticationEntitiesToRemove);
    userDAO.remove(usersToRemove);

    // remove groups
    final Set<GroupEntity> groupsToRemove = new HashSet<>();
    for (LdapGroupDto group : batchInfo.getGroupsToBeRemoved()) {
      final GroupEntity groupEntity = groupDAO.findGroupByName(group.getGroupName());
      allGroups.remove(groupEntity.getGroupName());
      groupsToRemove.add(groupEntity);
    }
    groupDAO.remove(groupsToRemove);

    /* *****
     * Update users
     * ***** */
    final Set<UserEntity> userEntitiesToUpdate = new HashSet<>();
    for (LdapUserDto user : batchInfo.getUsersToBecomeLdap()) {
      // Ensure the username is all lowercase
      String userName = user.getUserName();

      UserEntity userEntity = userDAO.findUserByName(userName);
      if (userEntity != null) {
        LOG.trace("Enabling LDAP authentication for the user account with the username {}.", userName);

        if (configuration.getLdapSyncCollisionHandlingBehavior() == Configuration.LdapUsernameCollisionHandlingBehavior.CONVERT) {
          // If converting the user to only an LDAP user, then remove all other authentication methods
          Collection<UserAuthenticationEntity> existingEntities = userEntity.getAuthenticationEntities();
          if(existingEntities != null) {
            Iterator<UserAuthenticationEntity> iterator = existingEntities.iterator();
            while(iterator.hasNext()) {
              UserAuthenticationEntity userAuthenticationEntity = iterator.next();
              if(userAuthenticationEntity.getAuthenticationType() != UserAuthenticationType.LDAP) {
                removeAuthentication(userEntity, userAuthenticationEntity.getUserAuthenticationId());
                iterator.remove();
              }
            }
          }
        }

        try {
          addLdapAuthentication(userEntity, user.getDn(), false);
          userEntitiesToUpdate.add(userEntity);
        } catch (AmbariException e) {
          LOG.warn(String.format("Failed to enable LDAP authentication for the user account with the username %s: %s", userName, e.getLocalizedMessage()), e);
        }
      } else {
        LOG.warn("Failed to find user account for {} while enabling LDAP authentication for the user.", userName);
      }
    }
    userDAO.merge(userEntitiesToUpdate);

    // update groups
    final Set<GroupEntity> groupsToBecomeLdap = new HashSet<>();
    for (LdapGroupDto group : batchInfo.getGroupsToBecomeLdap()) {
      final GroupEntity groupEntity = groupDAO.findGroupByName(group.getGroupName());
      groupEntity.setGroupType(GroupType.LDAP);
      allGroups.put(groupEntity.getGroupName(), groupEntity);
      groupsToBecomeLdap.add(groupEntity);
    }
    groupDAO.merge(groupsToBecomeLdap);

    // prepare create principals
    final List<PrincipalEntity> principalsToCreate = new ArrayList<>();

    // Create users
    for (LdapUserDto user : batchInfo.getUsersToBeCreated()) {
      String userName = user.getUserName();
      UserEntity userEntity;

      try {
        userEntity = createUser(userName, userName, userName, true);
      } catch (AmbariException e) {
        LOG.error(String.format("Failed to create new user: %s", userName), e);
        userEntity = null;
      }

      if (userEntity != null) {
        LOG.trace("Enabling LDAP authentication for the user account with the username {}.", userName);
        try {
          addLdapAuthentication(userEntity, user.getDn(), false);
        } catch (AmbariException e) {
          LOG.warn(String.format("Failed to enable LDAP authentication for the user account with the username %s: %s", userName, e.getLocalizedMessage()), e);
        }

        userDAO.merge(userEntity);

        // Add the new user to the allUsers map.
        allUsers.put(userEntity.getUserName(), userEntity);
      }
    }

    // prepare create groups
    final Set<GroupEntity> groupsToCreate = new HashSet<>();
    for (LdapGroupDto group: batchInfo.getGroupsToBeCreated()) {
      final PrincipalEntity principalEntity = new PrincipalEntity();
      principalEntity.setPrincipalType(groupPrincipalType);
      principalsToCreate.add(principalEntity);

      final GroupEntity groupEntity = new GroupEntity();
      groupEntity.setGroupName(group.getGroupName());
      groupEntity.setPrincipal(principalEntity);
      groupEntity.setGroupType(GroupType.LDAP);

      allGroups.put(groupEntity.getGroupName(), groupEntity);
      groupsToCreate.add(groupEntity);
    }

    // create groups
    principalDAO.create(principalsToCreate);
    groupDAO.create(groupsToCreate);

    // create membership
    final Set<MemberEntity> membersToCreate = new HashSet<>();
    final Set<GroupEntity> groupsToUpdate = new HashSet<>();
    for (LdapUserGroupMemberDto member : batchInfo.getMembershipToAdd()) {
      final MemberEntity memberEntity = new MemberEntity();
      final GroupEntity groupEntity = allGroups.get(member.getGroupName());
      memberEntity.setGroup(groupEntity);
      memberEntity.setUser(allUsers.get(member.getUserName()));
      groupEntity.getMemberEntities().add(memberEntity);
      groupsToUpdate.add(groupEntity);
      membersToCreate.add(memberEntity);
    }

    // handle adminGroupMappingRules
    processLdapAdminGroupMappingRules(membersToCreate);

    memberDAO.create(membersToCreate);
    groupDAO.merge(groupsToUpdate); // needed for Derby DB as it doesn't fetch newly added members automatically

    // remove membership
    final Set<MemberEntity> membersToRemove = new HashSet<>();
    for (LdapUserGroupMemberDto member : batchInfo.getMembershipToRemove()) {
      MemberEntity memberEntity = memberDAO.findByUserAndGroup(member.getUserName(), member.getGroupName());
