/*
 * 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.orm.dao;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;

import jakarta.persistence.EntityManager;
import jakarta.persistence.TypedQuery;
import jakarta.persistence.criteria.CriteriaQuery;
import jakarta.persistence.criteria.Order;
import jakarta.persistence.metamodel.SingularAttribute;

import org.apache.ambari.annotations.Experimental;
import org.apache.ambari.annotations.ExperimentalFeature;
import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.api.query.JpaPredicateVisitor;
import org.apache.ambari.server.api.query.JpaSortBuilder;
import org.apache.ambari.server.cleanup.TimeBasedCleanupPolicy;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.controller.AlertCurrentRequest;
import org.apache.ambari.server.controller.AlertHistoryRequest;
import org.apache.ambari.server.controller.spi.Predicate;
import org.apache.ambari.server.controller.utilities.PredicateHelper;
import org.apache.ambari.server.events.AggregateAlertRecalculateEvent;
import org.apache.ambari.server.events.publishers.AlertEventPublisher;
import org.apache.ambari.server.orm.RequiresSession;
import org.apache.ambari.server.orm.entities.AlertCurrentEntity;
import org.apache.ambari.server.orm.entities.AlertCurrentEntity_;
import org.apache.ambari.server.orm.entities.AlertHistoryEntity;
import org.apache.ambari.server.orm.entities.AlertHistoryEntity_;
import org.apache.ambari.server.orm.entities.AlertNoticeEntity;
import org.apache.ambari.server.state.AlertState;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.MaintenanceState;
import org.apache.ambari.server.state.alert.Scope;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.Singleton;
import com.google.inject.persist.Transactional;

/**
 * The {@link AlertsDAO} class manages the {@link AlertHistoryEntity} and
 * {@link AlertCurrentEntity} instances. Each {@link AlertHistoryEntity} is
 * known as an "alert" that has been triggered and received.
 * <p/>
 * If alert caching is enabled, then updates to {@link AlertCurrentEntity} are
 * not immediately persisted to JPA. Instead, they are kept in a cache and
 * periodically flushed. This means that many queries will need to swap in the
 * cached {@link AlertCurrentEntity} with that returned from the EclipseLink JPA
 * entity manager.
 */
@Singleton
@Experimental(feature = ExperimentalFeature.ALERT_CACHING)
public class AlertsDAO implements Cleanable {
  /**
   * Logger.
   */
  private static final Logger LOG = LoggerFactory.getLogger(AlertsDAO.class);

  /**
   * A template of JPQL for getting the number of hosts in various states.
   */
  private static final String ALERT_COUNT_SQL_TEMPLATE = "SELECT NEW %s("
      + "SUM(CASE WHEN history.alertState = :okState AND alert.maintenanceState = :maintenanceStateOff THEN 1 ELSE 0 END), "
      + "SUM(CASE WHEN history.alertState = :warningState AND alert.maintenanceState = :maintenanceStateOff THEN 1 ELSE 0 END), "
      + "SUM(CASE WHEN history.alertState = :criticalState AND alert.maintenanceState = :maintenanceStateOff THEN 1 ELSE 0 END), "
      + "SUM(CASE WHEN history.alertState = :unknownState AND alert.maintenanceState = :maintenanceStateOff THEN 1 ELSE 0 END), "
      + "SUM(CASE WHEN alert.maintenanceState != :maintenanceStateOff THEN 1 ELSE 0 END)) "
      + "FROM AlertCurrentEntity alert JOIN alert.alertHistory history WHERE history.clusterId = :clusterId";

  private static final String ALERT_COUNT_PER_HOST_SQL_TEMPLATE = "SELECT NEW %s("
      + "history.hostName, "
      + "SUM(CASE WHEN history.alertState = :okState AND alert.maintenanceState = :maintenanceStateOff THEN 1 ELSE 0 END), "
      + "SUM(CASE WHEN history.alertState = :warningState AND alert.maintenanceState = :maintenanceStateOff THEN 1 ELSE 0 END), "
      + "SUM(CASE WHEN history.alertState = :criticalState AND alert.maintenanceState = :maintenanceStateOff THEN 1 ELSE 0 END), "
      + "SUM(CASE WHEN history.alertState = :unknownState AND alert.maintenanceState = :maintenanceStateOff THEN 1 ELSE 0 END), "
      + "SUM(CASE WHEN alert.maintenanceState != :maintenanceStateOff THEN 1 ELSE 0 END)) "
      + "FROM AlertCurrentEntity alert JOIN alert.alertHistory history WHERE history.clusterId = :clusterId GROUP BY history.hostName";

  /**
   * JPA entity manager
   */
  @Inject
  private Provider<EntityManager> m_entityManagerProvider;

  /**
   * DAO utilities for dealing mostly with {@link TypedQuery} results.
   */
  @Inject
  private DaoUtils m_daoUtils;

  /**
   * Publishes alert events when particular DAO methods are called.
   */
  @Inject
  private AlertEventPublisher m_alertEventPublisher;

  /**
   * Used to lookup clusters.
   */
  @Inject
  private Provider<Clusters> m_clusters;

  /**
   * Configuration.
   */
  private final Configuration m_configuration;

  /**
   * A cache of current alert information. The {@link AlertCurrentEntity}
   * instances cached are currently managed. This allows the cached instances to
   * be easiler flushed from the cache to JPA.
   * <p/>
   * This also means that the cache is holding onto a rather large map of JPA
   * entities. This could lead to OOM errors over time if the indirectly
   * referenced entity map contains more than just {@link AlertCurrentEntity}.
   */
  private LoadingCache<AlertCacheKey, AlertCurrentEntity> m_currentAlertCache = null;

  /**
   * Batch size to query the DB and use the results in an IN clause.
   */
  private static final int BATCH_SIZE = 999;

  /**
   * Constructor.
   *
   */
  @Inject
  public AlertsDAO(Configuration configuration) {
    m_configuration = configuration;

    if( m_configuration.isAlertCacheEnabled() ){
      int maximumSize = m_configuration.getAlertCacheSize();

      LOG.info("Alert caching is enabled (size={}, flushInterval={}m)", maximumSize,
          m_configuration.getAlertCacheFlushInterval());

      // construct a cache for current alerts which will prevent database hits
      // on every heartbeat
      m_currentAlertCache = CacheBuilder.newBuilder().maximumSize(
          maximumSize).build(new CacheLoader<AlertCacheKey, AlertCurrentEntity>() {
            @Override
            public AlertCurrentEntity load(AlertCacheKey key) throws Exception {
              LOG.debug("Cache miss for alert key {}, fetching from JPA", key);

              final AlertCurrentEntity alertCurrentEntity;

              long clusterId = key.getClusterId();
              String alertDefinitionName = key.getAlertDefinitionName();
              String hostName = key.getHostName();

              if (StringUtils.isEmpty(hostName)) {
                alertCurrentEntity = findCurrentByNameNoHostInternalInJPA(clusterId,
                    alertDefinitionName);
              } else {
                alertCurrentEntity = findCurrentByHostAndNameInJPA(clusterId, hostName,
                    alertDefinitionName);
              }

              if (null == alertCurrentEntity) {
                LOG.trace("Cache lookup failed for {} because the alert does not yet exist", key);
                throw new AlertNotYetCreatedException();
              }

              return alertCurrentEntity;
            }
          });
    }
  }

  /**
   * Gets an alert with the specified ID.
   *
   * @param alertId
   *          the ID of the alert to retrieve.
   * @return the alert or {@code null} if none exists.
   */
  @RequiresSession
  public AlertHistoryEntity findById(long alertId) {
    return m_entityManagerProvider.get().find(AlertHistoryEntity.class, alertId);
  }

  /**
   * Gets all alerts stored in the database across all clusters.
   *
   * @return all alerts or an empty list if none exist (never {@code null}).
   */
  @RequiresSession
  public List<AlertHistoryEntity> findAll() {
    TypedQuery<AlertHistoryEntity> query = m_entityManagerProvider.get().createNamedQuery(
        "AlertHistoryEntity.findAll", AlertHistoryEntity.class);

    return m_daoUtils.selectList(query);
  }

  /**
   * Gets all alerts stored in the database for the given cluster.
   *
   * @param clusterId
   *          the ID of the cluster.
   * @return all alerts in the specified cluster or an empty list if none exist
   *         (never {@code null}).
   */
  @RequiresSession
  public List<AlertHistoryEntity> findAll(long clusterId) {
    TypedQuery<AlertHistoryEntity> query = m_entityManagerProvider.get().createNamedQuery(
        "AlertHistoryEntity.findAllInCluster", AlertHistoryEntity.class);

    query.setParameter("clusterId", clusterId);

    return m_daoUtils.selectList(query);
  }

  /**
   * Gets all alerts stored in the database for the given cluster that have one
   * of the specified alert states.
   *
   * @param clusterId
   *          the ID of the cluster.
   * @param alertStates
   *          the states to match for the retrieved alerts (not {@code null}).
   * @return the alerts matching the specified states and cluster, or an empty
   *         list if none.
   */
  @RequiresSession
  public List<AlertHistoryEntity> findAll(long clusterId,
      List<AlertState> alertStates) {
    if (null == alertStates || alertStates.size() == 0) {
      return Collections.emptyList();
    }

    TypedQuery<AlertHistoryEntity> query = m_entityManagerProvider.get().createNamedQuery(
        "AlertHistoryEntity.findAllInClusterWithState",
        AlertHistoryEntity.class);

    query.setParameter("clusterId", clusterId);
    query.setParameter("alertStates", alertStates);

    return m_daoUtils.selectList(query);
  }

  /**
   * Gets all alerts stored in the database for the given cluster and that fall
   * withing the specified date range. Dates are expected to be in milliseconds
   * since the epoch, normalized to UTC time.
   *
   * @param clusterId
   *          the ID of the cluster.
   * @param startDate
   *          the date that the earliest entry must occur after, normalized to
   *          UTC, or {@code null} for all entries that occur before the given
   *          end date.
   * @param endDate
   *          the date that the latest entry must occur before, normalized to
   *          UTC, or {@code null} for all entries that occur after the given
   *          start date.
   * @return the alerts matching the specified date range.
   */
  @RequiresSession
  public List<AlertHistoryEntity> findAll(long clusterId, Date startDate,
      Date endDate) {
    if (null == startDate && null == endDate) {
      return Collections.emptyList();
    }

    TypedQuery<AlertHistoryEntity> query = null;

    if (null != startDate && null != endDate) {
      if (startDate.after(endDate)) {
        return Collections.emptyList();
      }

      query = m_entityManagerProvider.get().createNamedQuery(
          "AlertHistoryEntity.findAllInClusterBetweenDates",
          AlertHistoryEntity.class);

      query.setParameter("clusterId", clusterId);
      query.setParameter("startDate", startDate.getTime());
      query.setParameter("endDate", endDate.getTime());
    } else if (null != startDate) {
      query = m_entityManagerProvider.get().createNamedQuery(
          "AlertHistoryEntity.findAllInClusterAfterDate",
          AlertHistoryEntity.class);

      query.setParameter("clusterId", clusterId);
      query.setParameter("afterDate", startDate.getTime());
    } else if (null != endDate) {
      query = m_entityManagerProvider.get().createNamedQuery(
          "AlertHistoryEntity.findAllInClusterBeforeDate",
          AlertHistoryEntity.class);

      query.setParameter("clusterId", clusterId);
      query.setParameter("beforeDate", endDate.getTime());
    }

    if (null == query) {
      return Collections.emptyList();
    }

    return m_daoUtils.selectList(query);
  }

  /**
   * Finds all {@link AlertHistoryEntity} that match the provided
   * {@link AlertHistoryRequest}. This method will make JPA do the heavy lifting
   * of providing a slice of the result set.
   *
   * @param request
   * @return
   */
  @RequiresSession
  public List<AlertHistoryEntity> findAll(AlertHistoryRequest request) {
    EntityManager entityManager = m_entityManagerProvider.get();

    // convert the Ambari predicate into a JPA predicate
    HistoryPredicateVisitor visitor = new HistoryPredicateVisitor();
    PredicateHelper.visit(request.Predicate, visitor);

    CriteriaQuery<AlertHistoryEntity> query = visitor.getCriteriaQuery();
    jakarta.persistence.criteria.Predicate jpaPredicate = visitor.getJpaPredicate();

    if (null != jpaPredicate) {
      query.where(jpaPredicate);
    }

    // sorting
    JpaSortBuilder<AlertHistoryEntity> sortBuilder = new JpaSortBuilder<>();
    List<Order> sortOrders = sortBuilder.buildSortOrders(request.Sort, visitor);
    query.orderBy(sortOrders);

    // pagination
    TypedQuery<AlertHistoryEntity> typedQuery = entityManager.createQuery(query);
    if (null != request.Pagination) {
      typedQuery.setFirstResult(request.Pagination.getOffset());
      typedQuery.setMaxResults(request.Pagination.getPageSize());
    }

    return m_daoUtils.selectList(typedQuery);
  }

  /**
   * Finds all {@link AlertCurrentEntity} that match the provided
   * {@link AlertCurrentRequest}. This method will make JPA do the heavy lifting
   * of providing a slice of the result set.
   *
   * @param request
   * @return
   */
  @Transactional
  public List<AlertCurrentEntity> findAll(AlertCurrentRequest request) {
    EntityManager entityManager = m_entityManagerProvider.get();

    // convert the Ambari predicate into a JPA predicate
    CurrentPredicateVisitor visitor = new CurrentPredicateVisitor();
    PredicateHelper.visit(request.Predicate, visitor);

    CriteriaQuery<AlertCurrentEntity> query = visitor.getCriteriaQuery();
    jakarta.persistence.criteria.Predicate jpaPredicate = visitor.getJpaPredicate();

    if (null != jpaPredicate) {
      query.where(jpaPredicate);
    }

    // sorting
    JpaSortBuilder<AlertCurrentEntity> sortBuilder = new JpaSortBuilder<>();
    List<Order> sortOrders = sortBuilder.buildSortOrders(request.Sort, visitor);
    query.orderBy(sortOrders);

    // pagination
    TypedQuery<AlertCurrentEntity> typedQuery = entityManager.createQuery(query);
    if( null != request.Pagination ){
      // prevent JPA errors when -1 is passed in by accident
      int offset = request.Pagination.getOffset();
      if (offset < 0) {
        offset = 0;
      }

      typedQuery.setFirstResult(offset);
      typedQuery.setMaxResults(request.Pagination.getPageSize());
    }

    List<AlertCurrentEntity> alerts = m_daoUtils.selectList(typedQuery);

    // if caching is enabled, replace results with cached values when present
    if (m_configuration.isAlertCacheEnabled()) {
      alerts = supplementWithCachedAlerts(alerts);
    }

    return alerts;
  }

  /**
   * Gets the total count of all {@link AlertHistoryEntity} rows that match the
   * specified {@link Predicate}.
   *
   * @param predicate
   *          the predicate to apply, or {@code null} for none.
   * @return the total count of rows that would be returned in a result set.
   */
  public int getCount(Predicate predicate) {
    return 0;
  }

  /**
   * Gets the current alerts.
   *
   * @return the current alerts or an empty list if none exist (never
   *         {@code null}).
   */
  @RequiresSession
  public List<AlertCurrentEntity> findCurrent() {
    TypedQuery<AlertCurrentEntity> query = m_entityManagerProvider.get().createNamedQuery(
        "AlertCurrentEntity.findAll", AlertCurrentEntity.class);

    List<AlertCurrentEntity> alerts = m_daoUtils.selectList(query);

    // if caching is enabled, replace results with cached values when present
    if (m_configuration.isAlertCacheEnabled()) {
      alerts = supplementWithCachedAlerts(alerts);
    }

    return alerts;
  }

  /**
   * Gets a current alert with the specified ID.
   *
   * @param alertId
   *          the ID of the alert to retrieve.
   * @return the alert or {@code null} if none exists.
   */
  @RequiresSession
  public AlertCurrentEntity findCurrentById(long alertId) {
    return m_entityManagerProvider.get().find(AlertCurrentEntity.class, alertId);
  }

  /**
   * Gets the current alerts for the specified definition ID.
   *
   * @param definitionId
   *          the ID of the definition to retrieve current alerts for.
   * @return the current alerts for the definition or an empty list if none
   *         exist (never {@code null}).
   */
  @RequiresSession
  public List<AlertCurrentEntity> findCurrentByDefinitionId(long definitionId) {
    TypedQuery<AlertCurrentEntity> query = m_entityManagerProvider.get().createNamedQuery(
        "AlertCurrentEntity.findByDefinitionId", AlertCurrentEntity.class);

    query.setParameter("definitionId", Long.valueOf(definitionId));

    List<AlertCurrentEntity> alerts = m_daoUtils.selectList(query);

    // if caching is enabled, replace results with cached values when present
    if (m_configuration.isAlertCacheEnabled()) {
      alerts = supplementWithCachedAlerts(alerts);
    }

    return alerts;
  }

  /**
   * Gets the current alerts for a given cluster.
   *
   * @return the current alerts for the given cluster or an empty list if none
   *         exist (never {@code null}).
   */
  @RequiresSession
  public List<AlertCurrentEntity> findCurrentByCluster(long clusterId) {
    TypedQuery<AlertCurrentEntity> query = m_entityManagerProvider.get().createNamedQuery(
        "AlertCurrentEntity.findByCluster", AlertCurrentEntity.class);

    query.setParameter("clusterId", Long.valueOf(clusterId));

    List<AlertCurrentEntity> alerts = m_daoUtils.selectList(query);

    // if caching is enabled, replace results with cached values when present
    if (m_configuration.isAlertCacheEnabled()) {
      alerts = supplementWithCachedAlerts(alerts);
    }

    return alerts;
  }

  /**
   * Retrieves the summary information for a particular scope. The result is a
   * DTO since the columns are aggregated and don't fit to an entity.
   *
   * @param clusterId
   *          the cluster id
   * @param serviceName
   *          the service name. Use {@code null} to not filter on service.
   * @param hostName
   *          the host name. Use {@code null} to not filter on host.
   * @return the summary DTO
   */
  @RequiresSession
  public AlertSummaryDTO findCurrentCounts(long clusterId, String serviceName, String hostName) {
    String sql = String.format(ALERT_COUNT_SQL_TEMPLATE,
      AlertSummaryDTO.class.getName());

    StringBuilder sb = new StringBuilder(sql);

    if (null != serviceName) {
      sb.append(" AND history.serviceName = :serviceName");
    }

    if (null != hostName) {
      sb.append(" AND history.hostName = :hostName");
    }

    TypedQuery<AlertSummaryDTO> query = m_entityManagerProvider.get().createQuery(
      sb.toString(), AlertSummaryDTO.class);

    query.setParameter("clusterId", Long.valueOf(clusterId));
    query.setParameter("okState", AlertState.OK);
    query.setParameter("warningState", AlertState.WARNING);
    query.setParameter("criticalState", AlertState.CRITICAL);
    query.setParameter("unknownState", AlertState.UNKNOWN);
    query.setParameter("maintenanceStateOff", MaintenanceState.OFF);

    if (null != serviceName) {
      query.setParameter("serviceName", serviceName);
    }

    if (null != hostName) {
      query.setParameter("hostName", hostName);
    }

    return m_daoUtils.selectSingle(query);
  }

  /**
   * Retrieves the summary information for all the hosts in the provided cluster.
   * The result is mapping from hostname to summary DTO.
   *
   * @param clusterId
   *          the cluster id
   * @return map from hostnames to summary DTO
   */
  @RequiresSession
  public Map<String, AlertSummaryDTO> findCurrentPerHostCounts(long clusterId) {
    String sql = String.format(ALERT_COUNT_PER_HOST_SQL_TEMPLATE, HostAlertSummaryDTO.class.getName());

    StringBuilder sb = new StringBuilder(sql);

    TypedQuery<HostAlertSummaryDTO> query = m_entityManagerProvider.get().createQuery(sb.toString(), HostAlertSummaryDTO.class);

    query.setParameter("clusterId", Long.valueOf(clusterId));
    query.setParameter("okState", AlertState.OK);
    query.setParameter("warningState", AlertState.WARNING);
    query.setParameter("criticalState", AlertState.CRITICAL);
    query.setParameter("unknownState", AlertState.UNKNOWN);
    query.setParameter("maintenanceStateOff", MaintenanceState.OFF);

    Map<String, AlertSummaryDTO> map = new HashMap<>();
    List<HostAlertSummaryDTO> resultList = m_daoUtils.selectList(query);
    for (HostAlertSummaryDTO result : resultList) {
      map.put(result.getHostName(), result);
    }
    return map;
  }

  /**
   * Retrieve the summary alert information for all hosts. This is different
   * from {@link #findCurrentCounts(long, String, String)} since this will
   * return only alerts related to hosts and those values will be the total
   * number of hosts affected, not the total number of alerts.
   *
   * @param clusterId
   *          the cluster id
   * @return the summary DTO for host alerts.
   */
  @RequiresSession
  public AlertHostSummaryDTO findCurrentHostCounts(long clusterId) {
    String sql = String.format(ALERT_COUNT_PER_HOST_SQL_TEMPLATE, HostAlertSummaryDTO.class.getName());

    StringBuilder sb = new StringBuilder(sql);

    TypedQuery<HostAlertSummaryDTO> query = m_entityManagerProvider.get().createQuery(sb.toString(), HostAlertSummaryDTO.class);

    query.setParameter("clusterId", Long.valueOf(clusterId));
    query.setParameter("okState", AlertState.OK);
    query.setParameter("criticalState", AlertState.CRITICAL);
    query.setParameter("warningState", AlertState.WARNING);
    query.setParameter("unknownState", AlertState.UNKNOWN);
    query.setParameter("maintenanceStateOff", MaintenanceState.OFF);

    int okCount = 0;
    int warningCount = 0;
    int criticalCount = 0;
    int unknownCount = 0;

    List<HostAlertSummaryDTO> resultList = m_daoUtils.selectList(query);
    for (HostAlertSummaryDTO result : resultList) {
      if (result.getHostName() == null) {
        continue;
      }
      if (result.getCriticalCount() > 0) {
        criticalCount++;
      }
      else if (result.getWarningCount() > 0) {
        warningCount++;
      }
      else if (result.getUnknownCount() > 0) {
        unknownCount++;
      }
      else {
        okCount++;
      }
    }

    AlertHostSummaryDTO hostSummary = new AlertHostSummaryDTO(okCount,
            unknownCount, warningCount, criticalCount);

    return hostSummary;
  }

  /**
   * Gets the current alerts for a given service.
   *
   * @return the current alerts for the given service or an empty list if none
   *         exist (never {@code null}).
   */
  @RequiresSession
  public List<AlertCurrentEntity> findCurrentByService(long clusterId,
      String serviceName) {
    TypedQuery<AlertCurrentEntity> query = m_entityManagerProvider.get().createNamedQuery(
      "AlertCurrentEntity.findByService", AlertCurrentEntity.class);

    query.setParameter("clusterId", clusterId);
    query.setParameter("serviceName", serviceName);
    query.setParameter("inlist", EnumSet.of(Scope.ANY, Scope.SERVICE));

    List<AlertCurrentEntity> alerts = m_daoUtils.selectList(query);

    // if caching is enabled, replace results with cached values when present
    if (m_configuration.isAlertCacheEnabled()) {
      alerts = supplementWithCachedAlerts(alerts);
    }

    return alerts;
  }

  /**
   * Locate the current alert for the provided service and alert name. This
   * method will first consult the cache if configured with
   * {@link Configuration#isAlertCacheEnabled()}.
   *
   * @param clusterId
   *          the cluster id
   * @param hostName
   *          the name of the host (not {@code null}).
   * @param alertName
   *          the name of the alert (not {@code null}).
   * @return the current record, or {@code null} if not found
   */
  public AlertCurrentEntity findCurrentByHostAndName(long clusterId, String hostName,
      String alertName) {

    if( m_configuration.isAlertCacheEnabled() ){
      AlertCacheKey key = new AlertCacheKey(clusterId, alertName, hostName);

      try {
        return m_currentAlertCache.get(key);
      } catch (ExecutionException executionException) {
        Throwable cause = executionException.getCause();
        if (!(cause instanceof AlertNotYetCreatedException)) {
          LOG.warn("Unable to retrieve alert for key {} from the cache", key);
        }
      }
    }

    return findCurrentByHostAndNameInJPA(clusterId, hostName, alertName);
  }

  /**
   * Locate the current alert for the provided service and alert name.
   *
   * @param clusterId
   *          the cluster id
   * @param hostName
   *          the name of the host (not {@code null}).
   * @param alertName
   *          the name of the alert (not {@code null}).
   * @return the current record, or {@code null} if not found
   */
  @RequiresSession
  private AlertCurrentEntity findCurrentByHostAndNameInJPA(long clusterId, String hostName,
      String alertName) {
    TypedQuery<AlertCurrentEntity> query = m_entityManagerProvider.get().createNamedQuery(
      "AlertCurrentEntity.findByHostAndName", AlertCurrentEntity.class);

    query.setParameter("clusterId", Long.valueOf(clusterId));
    query.setParameter("hostName", hostName);
    query.setParameter("definitionName", alertName);

    return m_daoUtils.selectOne(query);
  }

  /**
   * Removes alert history and current alerts for the specified alert defintiion
   * ID. This will invoke {@link EntityManager#clear()} when completed since the
   * JPQL statement will remove entries without going through the EM.
   *
   * @param definitionId
   *          the ID of the definition to remove.
   */
  @Transactional
  public void removeByDefinitionId(long definitionId) {
    EntityManager entityManager = m_entityManagerProvider.get();
    TypedQuery<AlertCurrentEntity> currentQuery = entityManager.createNamedQuery(
        "AlertCurrentEntity.removeByDefinitionId", AlertCurrentEntity.class);

    currentQuery.setParameter("definitionId", definitionId);
    currentQuery.executeUpdate();

    TypedQuery<AlertHistoryEntity> historyQuery = entityManager.createNamedQuery(
        "AlertHistoryEntity.removeByDefinitionId", AlertHistoryEntity.class);

    historyQuery.setParameter("definitionId", definitionId);
    historyQuery.executeUpdate();

    entityManager.clear();

    // if caching is enabled, invalidate the cache to force the latest values
    // back from the DB
    if (m_configuration.isAlertCacheEnabled()) {
      m_currentAlertCache.invalidateAll();
    }
  }

  /**
   * Remove a current alert whose history entry matches the specfied ID.
   *
   * @param   historyId the ID of the history entry.
   * @return  the number of alerts removed.
   */
  @Transactional
  public int removeCurrentByHistoryId(long historyId) {
    TypedQuery<AlertCurrentEntity> query = m_entityManagerProvider.get().createNamedQuery(
      "AlertCurrentEntity.removeByHistoryId", AlertCurrentEntity.class);

    query.setParameter("historyId", historyId);
    int rowsRemoved = query.executeUpdate();

    // if caching is enabled, invalidate the cache to force the latest values
    // back from the DB
    if (m_configuration.isAlertCacheEnabled()) {
      m_currentAlertCache.invalidateAll();
    }

    return rowsRemoved;
  }

  /**
   * Remove all current alerts that are disabled.
   *
   * @return the number of alerts removed.
   */
  @Transactional
  public int removeCurrentDisabledAlerts() {
    TypedQuery<AlertCurrentEntity> query = m_entityManagerProvider.get().createNamedQuery(
      "AlertCurrentEntity.findDisabled", AlertCurrentEntity.class);

    int rowsRemoved = 0;
    List<AlertCurrentEntity> currentEntities = m_daoUtils.selectList(query);
    if (currentEntities != null) {
      for (AlertCurrentEntity currentEntity : currentEntities) {
        remove(currentEntity);
        rowsRemoved++;
      }
    }

    // if caching is enabled, invalidate the cache to force the latest values
    // back from the DB
    if (m_configuration.isAlertCacheEnabled()) {
      m_currentAlertCache.invalidateAll();
    }

    return rowsRemoved;
  }

  /**
   * Remove the current alert that matches the given service. This is used in
   * cases where the service was removed from the cluster.
   * <p>
   * This method will also fire an {@link AggregateAlertRecalculateEvent} in
   * order to recalculate all aggregates.
   *
   * @param clusterId
   *          the ID of the cluster.
   * @param serviceName
   *          the name of the service that the current alerts are being removed
   *          for (not {@code null}).
   * @return the number of alerts removed.
   */
  @Transactional
  public int removeCurrentByService(long clusterId, String serviceName) {
    TypedQuery<AlertCurrentEntity> query = m_entityManagerProvider.get().createNamedQuery(
      "AlertCurrentEntity.findByServiceName", AlertCurrentEntity.class);

    query.setParameter("serviceName", serviceName);

    int removedItems = 0;
    List<AlertCurrentEntity> currentEntities = m_daoUtils.selectList(query);
    if (currentEntities != null) {
      for (AlertCurrentEntity currentEntity : currentEntities) {
        remove(currentEntity);
        removedItems++;
      }
    }

    // if caching is enabled, invalidate the cache to force the latest values
    // back from the DB
    if (m_configuration.isAlertCacheEnabled()) {
      m_currentAlertCache.invalidateAll();
    }

    // publish the event to recalculate aggregates
    m_alertEventPublisher.publish(new AggregateAlertRecalculateEvent(clusterId));
    return removedItems;
  }

  /**
   * Remove the current alert that matches the given host. This is used in cases
   * where the host was removed from the cluster.
   * <p>
   * This method will also fire an {@link AggregateAlertRecalculateEvent} in
   * order to recalculate all aggregates.
   *
   * @param hostName
   *          the name of the host that the current alerts are being removed for
   *          (not {@code null}).
   * @return the number of alerts removed.
   */
  @Transactional
  public int removeCurrentByHost(String hostName) {
    TypedQuery<AlertCurrentEntity> query = m_entityManagerProvider.get().createNamedQuery(
      "AlertCurrentEntity.findByHost", AlertCurrentEntity.class);

    query.setParameter("hostName", hostName);
    List<AlertCurrentEntity> currentEntities = m_daoUtils.selectList(query);
    int removedItems = 0;
    if (currentEntities != null) {
      for (AlertCurrentEntity currentEntity : currentEntities) {
        remove(currentEntity);
        removedItems++;
      }
    }

    // if caching is enabled, invalidate the cache to force the latest values
    // back from the DB
    if (m_configuration.isAlertCacheEnabled()) {
      m_currentAlertCache.invalidateAll();
    }

    // publish the event to recalculate aggregates for every cluster since a host could potentially have several clusters
    try {
      Map<String, Cluster> clusters = m_clusters.get().getClusters();
      for (Map.Entry<String, Cluster> entry : clusters.entrySet()) {
        m_alertEventPublisher.publish(new AggregateAlertRecalculateEvent(
