/*
 * 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.geode.internal.cache.persistence;

import java.util.HashSet;
import java.util.Set;

import org.apache.logging.log4j.Logger;

import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.CacheDistributionAdvisor;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.InitialImageAdvice;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.logging.internal.log4j.api.LogService;

public class PersistenceInitialImageAdvisor {

  private static final Logger logger = LogService.getLogger();

  private final InternalPersistenceAdvisor persistenceAdvisor;
  private final String shortDiskStoreID;
  private final String regionPath;
  private final CacheDistributionAdvisor cacheDistributionAdvisor;
  private final boolean hasDiskImageToRecoverFrom;

  public PersistenceInitialImageAdvisor(InternalPersistenceAdvisor persistenceAdvisor,
      String shortDiskStoreID, String regionPath, CacheDistributionAdvisor cacheDistributionAdvisor,
      boolean hasDiskImageToRecoverFrom) {
    this.persistenceAdvisor = persistenceAdvisor;
    this.shortDiskStoreID = shortDiskStoreID;
    this.regionPath = regionPath;
    this.cacheDistributionAdvisor = cacheDistributionAdvisor;
    this.hasDiskImageToRecoverFrom = hasDiskImageToRecoverFrom;
  }

  public InitialImageAdvice getAdvice(InitialImageAdvice previousAdvice) {
    final boolean isPersistAdvisorDebugEnabled =
        logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE);

    MembershipChangeListener listener = new MembershipChangeListener(persistenceAdvisor);
    cacheDistributionAdvisor.addMembershipAndProxyListener(listener);
    persistenceAdvisor.addListener(listener);
    try {
      while (true) {
        cacheDistributionAdvisor.getAdvisee().getCancelCriterion().checkCancelInProgress(null);
        try {
          // On first pass, previous advice is null. On subsequent passes, it's the advice
          // from the previous iteration.
          InitialImageAdvice advice =
              cacheDistributionAdvisor.adviseInitialImage(previousAdvice, true);

          if (hasReplicates(advice)) {
            if (hasDiskImageToRecoverFrom) {
              removeReplicatesIfWeAreEqualToAnyOrElseClearEqualMembers(advice.getReplicates());
            }
            return advice;
          } else if (hasNonPersistentMember(advice)) {
            updateMembershipViewFromAnyPeer(advice.getNonPersistent(), hasDiskImageToRecoverFrom);
          }

          // Fix for 51698 - If there are online members that we previously failed to get a GII
          // from, retry those members rather than wait for new persistent members to recover.
          if (hasReplicates(previousAdvice)) {
            logger.info(
                "GII failed from all sources, but members are still online. Retrying the GII.");
            previousAdvice = null;
            continue;
          }

          Set<PersistentMemberID> previouslyOnlineMembers =
              persistenceAdvisor.getPersistedOnlineOrEqualMembers();

          // If there are no currently online members, and no previously online members, this member
          // should just go with what's on its own disk
          if (previouslyOnlineMembers.isEmpty()) {
            if (isPersistAdvisorDebugEnabled()) {
              logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
                  "{}-{}: No previously online members. Recovering with the data from the local disk",
                  shortDiskStoreID, regionPath);
            }
            return advice;
          }

          Set<PersistentMemberID> offlineMembers = new HashSet<>();
          Set<PersistentMemberID> membersToWaitFor =
              persistenceAdvisor.getMembersToWaitFor(previouslyOnlineMembers, offlineMembers);

          if (membersToWaitFor.isEmpty()) {
            if (isPersistAdvisorDebugEnabled()) {
              logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
                  "{}-{}: All of the previously online members are now online and waiting for us. Acquiring tie lock. Previously online members {}",
                  shortDiskStoreID, regionPath, advice.getReplicates());
            }
            if (persistenceAdvisor.acquireTieLock()) {
              return refreshInitialImageAdviceAndThenCheckMyStateWithReplicates(previousAdvice);
            }
          } else {
            if (isPersistAdvisorDebugEnabled) {
              logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
                  "{}-{}: Going to wait for these member ids: {}", shortDiskStoreID, regionPath,
                  membersToWaitFor);
            }
          }

          waitForMembershipChangeForMissingDiskStores(listener, offlineMembers, membersToWaitFor);
        } catch (InterruptedException e) {
          logger.debug("Interrupted while trying to determine latest persisted copy", e);
        }
      }
    } finally {
      persistenceAdvisor.setWaitingOnMembers(null, null);
      cacheDistributionAdvisor.removeMembershipAndProxyListener(listener);
      persistenceAdvisor.removeListener(listener);
    }
  }

  private void updateMembershipViewFromAnyPeer(Set<InternalDistributedMember> peers,
      boolean recoverFromDisk) {
    for (InternalDistributedMember peer : peers) {
      try {
        persistenceAdvisor.updateMembershipView(peer, recoverFromDisk);
        return;
      } catch (ReplyException e) {
        if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) {
          logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE, "Failed to update membership view", e);
        }
      }
    }
  }

  private InitialImageAdvice refreshInitialImageAdviceAndThenCheckMyStateWithReplicates(
      InitialImageAdvice previousAdvice) {
    if (isPersistAdvisorDebugEnabled()) {
      logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
          "{}-{}: Acquired the lock. This member will initialize", shortDiskStoreID, regionPath);
    }
    InitialImageAdvice advice = cacheDistributionAdvisor.adviseInitialImage(previousAdvice, true);
    if (hasReplicates(advice)) {
      if (isPersistAdvisorDebugEnabled()) {
        logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
            "{}-{}: Another member has initialized while we were getting the lock. We will initialize from that member",
            shortDiskStoreID, regionPath);
      }
      persistenceAdvisor.checkMyStateOnMembers(advice.getReplicates());
    }
    return advice;
  }

  private boolean hasNonPersistentMember(InitialImageAdvice advice) {
    return !advice.getNonPersistent().isEmpty();
  }

  /**
   * if one or more replicates are equal to this member: remove replicates from advice, return
   * advice for GII loop
   */
  private void removeReplicatesIfWeAreEqualToAnyOrElseClearEqualMembers(
      Set<InternalDistributedMember> replicates) {
    if (isPersistAdvisorDebugEnabled()) {
      logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
          "{}-{}: There are members currently online. Checking for our state on those members and then initializing",
          shortDiskStoreID, regionPath);
    }
    // Check with these members to make sure that they have heard of us. If any of them
    // say we have the same data on disk, we don't need to do a GII.
    boolean weAreEqualToAReplicate = persistenceAdvisor.checkMyStateOnMembers(replicates);
    if (weAreEqualToAReplicate) {
      // prevent GII by removing all replicates
      removeReplicates(replicates);
    } else {
      persistenceAdvisor.clearEqualMembers();
    }
    // Either a replicate has said we're equal and we've cleared replicates,
    // or none of them said we're equal and we've cleared our equal members.
    // We had replicates. Now one of these things is true:
    // - We've cleared replicates (meaning we're equal to one, so can load from disk)
    // - No replicates report we're equal (so must GII from one, which we indicate by clearing equal
    // members).
  }

  private boolean hasReplicates(InitialImageAdvice advice) {
    return advice != null && !advice.getReplicates().isEmpty();
  }

  private void removeReplicates(Set<InternalDistributedMember> replicates) {
    if (isPersistAdvisorDebugEnabled()) {
      logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
          "{}-{}: We have the same data on disk as one of {} recovering gracefully",
          shortDiskStoreID, regionPath, replicates);
    }
    replicates.clear();
  }

  private void waitForMembershipChangeForMissingDiskStores(MembershipChangeListener listener,
      Set<PersistentMemberID> offlineMembers, Set<PersistentMemberID> membersToWaitFor)
      throws InterruptedException {
    persistenceAdvisor.beginWaitingForMembershipChange(membersToWaitFor);
    try {
      // The persistence advisor needs to know which members are really not available because the
      // user uses this information to decide which members they haven't started yet.
      // membersToWaitFor includes members that are still waiting to start up, but are waiting for
      // members other than the current member. So we pass the set of offline members here.

      persistenceAdvisor.setWaitingOnMembers(membersToWaitFor, offlineMembers);
      listener.waitForChange();
    } finally {
      persistenceAdvisor.endWaitingForMembershipChange();
    }
  }

  private boolean isPersistAdvisorDebugEnabled() {
    return logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE);
  }
}