/*
 * 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;

import static org.apache.geode.internal.cache.LocalRegion.InitializationLevel.ANY_INIT;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.logging.log4j.Logger;

import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.DiskStore;
import org.apache.geode.cache.EntryDestroyedException;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.execute.FunctionService;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.cache.LocalRegion.InitializationLevel;
import org.apache.geode.internal.cache.execute.InternalRegionFunctionContext;
import org.apache.geode.internal.cache.partitioned.PRLocallyDestroyedException;
import org.apache.geode.internal.cache.persistence.PRPersistentConfig;
import org.apache.geode.internal.cache.wan.parallel.ParallelGatewaySenderQueue;
import org.apache.geode.internal.logging.LogService;

/**
 * An utility class to retrieve colocated regions in a colocation hierarchy in various scenarios
 *
 * @since GemFire 6.0
 */
public class ColocationHelper {
  private static final Logger logger = LogService.getLogger();

  /**
   * Whether to ignore missing parallel queues on restart if they are not attached to the region.
   * See bug 50120. Mutable for tests.
   */
  @MutableForTesting
  public static boolean IGNORE_UNRECOVERED_QUEUE =
      Boolean.getBoolean(DistributionConfig.GEMFIRE_PREFIX + "IGNORE_UNRECOVERED_QUEUE");

  /**
   * An utility method to retrieve colocated region of a given partitioned region
   *
   * @return colocated PartitionedRegion
   * @throws IllegalStateException for missing colocated region
   * @since GemFire 5.8Beta
   */
  public static PartitionedRegion getColocatedRegion(final PartitionedRegion partitionedRegion) {
    Assert.assertTrue(partitionedRegion != null); // precondition1
    String colocatedWith = partitionedRegion.getPartitionAttributes().getColocatedWith();
    if (colocatedWith == null) {
      // the region is not colocated with any region
      return null;
    }
    Region prRoot = PartitionedRegionHelper.getPRRoot(partitionedRegion.getCache());
    PartitionRegionConfig prConf =
        (PartitionRegionConfig) prRoot.get(getRegionIdentifier(colocatedWith));
    if (prConf == null) {
      partitionedRegion.getCache().getCancelCriterion().checkCancelInProgress(null);
      throw new IllegalStateException(
          String.format(
              "Region specified in 'colocated-with' (%s) for region %s does not exist. It should be created before setting 'colocated-with' attribute for this region.",
              new Object[] {colocatedWith, partitionedRegion.getFullPath()}));
    }
    int prID = prConf.getPRId();
    PartitionedRegion colocatedPR = null;
    try {
      colocatedPR = PartitionedRegion.getPRFromId(prID);
      if (colocatedPR != null) {
        colocatedPR.waitOnBucketMetadataInitialization();
      } else {
        partitionedRegion.getCache().getCancelCriterion().checkCancelInProgress(null);
        throw new IllegalStateException(
            String.format(
                "Region specified in 'colocated-with' (%s) for region %s does not exist. It should be created before setting 'colocated-with' attribute for this region.",
                new Object[] {colocatedWith, partitionedRegion.getFullPath()}));
      }
    } catch (PRLocallyDestroyedException e) {
      if (logger.isDebugEnabled()) {
        logger.debug(
            "PRLocallyDestroyedException : Region with prId={} is locally destroyed on this node",
            prID, e);
      }
    }
    return colocatedPR;
  }

  /**
   * An utility to make sure that a member contains all of the partitioned regions that are
   * colocated with a given region on other members. TODO rebalance - this is rather inefficient,
   * and probably all this junk should be in the advisor.
   */
  public static boolean checkMembersColocation(PartitionedRegion partitionedRegion,
      InternalDistributedMember member) {
    List<PartitionRegionConfig> colocatedRegions = new ArrayList<PartitionRegionConfig>();
    List<PartitionRegionConfig> tempcolocatedRegions = new ArrayList<PartitionRegionConfig>();
    Region prRoot = PartitionedRegionHelper.getPRRoot(partitionedRegion.getCache());
    PartitionRegionConfig regionConfig =
        (PartitionRegionConfig) prRoot.get(partitionedRegion.getRegionIdentifier());
    // The region was probably concurrently destroyed
    if (regionConfig == null) {
      return false;
    }
    tempcolocatedRegions.add(regionConfig);
    colocatedRegions.addAll(tempcolocatedRegions);
    PartitionRegionConfig prConf = null;
    do {
      PartitionRegionConfig tempToBeColocatedWith = tempcolocatedRegions.remove(0);
      for (Iterator itr = prRoot.keySet().iterator(); itr.hasNext();) {
        String prName = (String) itr.next();
        try {
          prConf = (PartitionRegionConfig) prRoot.get(prName);
        } catch (EntryDestroyedException ignore) {
          continue;
        }
        if (prConf == null) {
          // darrel says: I'm seeing an NPE in this code after pr->rem
          // merge so I added this check and continue
          continue;
        }
        if (prConf.getColocatedWith() != null) {
          if (prConf.getColocatedWith().equals(tempToBeColocatedWith.getFullPath())
              || (Region.SEPARATOR + prConf.getColocatedWith())
                  .equals(tempToBeColocatedWith.getFullPath())) {
            colocatedRegions.add(prConf);
            tempcolocatedRegions.add(prConf);
          }
        }
      }
    } while (!tempcolocatedRegions.isEmpty());

    PartitionRegionConfig tempColocatedWith = regionConfig;
    prConf = null;
    while (true) {
      String colocatedWithRegionName = tempColocatedWith.getColocatedWith();
      if (colocatedWithRegionName == null)
        break;
      else {
        prConf = (PartitionRegionConfig) prRoot.get(getRegionIdentifier(colocatedWithRegionName));
        if (prConf == null) {
          break;
        }
        colocatedRegions.add(tempColocatedWith);
        tempColocatedWith = prConf;
      }
    }

    // Now check to make sure that all of the colocated regions
    // Have this member.

    // We don't need a hostname because the equals method doesn't check it.
    for (PartitionRegionConfig config : colocatedRegions) {
      if (config.isColocationComplete() && !config.containsMember(member)) {
        return false;
      }
    }

    // Check to make sure all of the persisted regions that are colocated
    // with this region have been created.
    if (hasOfflineColocatedChildRegions(partitionedRegion)) {
      return false;
    }

    return true;
  }

  /**
   * Returns true if there are regions that are persisted on this member and were previously
   * colocated with the given region, but have not yet been created.
   *
   * @param region The parent region
   * @return true if there are any child regions that are persisted on this member, but have not yet
   *         been created.
   */
  private static boolean hasOfflineColocatedChildRegions(PartitionedRegion region) {
    boolean hasOfflineChildren = false;
    final InitializationLevel oldLevel = LocalRegion.setThreadInitLevelRequirement(ANY_INIT);
    try {
      InternalCache cache = region.getCache();
      Collection<DiskStore> stores = cache.listDiskStores();
      // Look through all of the disk stores for offline colocated child regions
      for (DiskStore diskStore : stores) {
        // Look at all of the partitioned regions.
        for (Map.Entry<String, PRPersistentConfig> entry : ((DiskStoreImpl) diskStore).getAllPRs()
            .entrySet()) {

          PRPersistentConfig config = entry.getValue();
          String childName = entry.getKey();

          // Check to see if they're colocated with this region.
          if (region.getFullPath().equals(config.getColocatedWith())) {
            PartitionedRegion childRegion = (PartitionedRegion) cache.getRegion(childName);

            if (childRegion == null) {
              // If the child region is offline, return true
              // unless it is a parallel queue that the user has removed.
              if (!ignoreUnrecoveredQueue(region, childName)) {
                region.addMissingColocatedRegionLogger(childName);
                hasOfflineChildren = true;
              }
            } else {
              // Otherwise, look for offline children of that region.
              if (hasOfflineColocatedChildRegions(childRegion)) {
                hasOfflineChildren = true;
                // Add the offline children of this child to the region's missingChildren list
                region.addMissingColocatedRegionLogger(childRegion);
              }
            }
          }
        }
      }
    } finally {
      LocalRegion.setThreadInitLevelRequirement(oldLevel);
    }
    return hasOfflineChildren;
  }


  private static boolean ignoreUnrecoveredQueue(PartitionedRegion region, String childName) {
    // Hack for #50120 if the childRegion is an async queue, but we
    // no longer define the async queue, ignore it.
    if (!ParallelGatewaySenderQueue.isParallelQueue(childName)) {
      return false;
    }

    String senderId = ParallelGatewaySenderQueue.getSenderId(childName);
    if (!region.getAsyncEventQueueIds().contains(senderId)
        && !region.getParallelGatewaySenderIds().contains(senderId) && IGNORE_UNRECOVERED_QUEUE) {
      return true;
    }

    // TODO Auto-generated method stub
    return false;
  }

  /**
   * A utility to check to see if a region has been created on all of the VMs that host the regions
   * this region is colocated with.
   */
  public static boolean isColocationComplete(PartitionedRegion region) {
    Region prRoot = PartitionedRegionHelper.getPRRoot(region.getCache());
    PartitionRegionConfig config = (PartitionRegionConfig) prRoot.get(region.getRegionIdentifier());
    // Fix for bug 40075. There is race between this call and the region being concurrently
    // destroyed.
    if (config == null) {
      Assert.assertTrue(region.isDestroyed() || region.isClosed,
          "Region is not destroyed, but there is no entry in the prRoot for region " + region);
      return false;
    }
    return config.isColocationComplete();
  }

  /**
   * An utility method to retrieve all partitioned regions(excluding self) in a colocation chain<br>
   * <p>
   * For example, shipmentPR is colocated with orderPR and orderPR is colocated with customerPR <br>
   * <br>
   * getAllColocationRegions(customerPR) --> List{orderPR, shipmentPR}<br>
   * getAllColocationRegions(orderPR) --> List{customerPR, shipmentPR}<br>
   * getAllColocationRegions(shipmentPR) --> List{customerPR, orderPR}<br>
   *
   * @return List of all partitioned regions (excluding self) in a colocated chain
   * @since GemFire 5.8Beta
   */
  public static Map<String, PartitionedRegion> getAllColocationRegions(
      PartitionedRegion partitionedRegion) {
    Map<String, PartitionedRegion> colocatedRegions = new HashMap<String, PartitionedRegion>();
    List<PartitionedRegion> colocatedByRegion = partitionedRegion.getColocatedByList();
    if (colocatedByRegion.size() != 0) {
      List<PartitionedRegion> tempcolocatedRegions = new ArrayList<PartitionedRegion>();
      tempcolocatedRegions.addAll(colocatedByRegion);
      do {
        PartitionedRegion pRegion = tempcolocatedRegions.remove(0);
        pRegion.waitOnBucketMetadataInitialization();
        colocatedRegions.put(pRegion.getFullPath(), pRegion);
        tempcolocatedRegions.addAll(pRegion.getColocatedByList());
      } while (!tempcolocatedRegions.isEmpty());
    }
    PartitionedRegion tempColocatedWith = partitionedRegion;
    while (true) {
      PartitionedRegion colocatedWithRegion = tempColocatedWith.getColocatedWithRegion();
      if (colocatedWithRegion == null)
        break;
      else {
        colocatedRegions.put(colocatedWithRegion.getFullPath(), colocatedWithRegion);
        tempColocatedWith = colocatedWithRegion;
      }
    }
    return colocatedRegions;
  }

  /**
   * gets local data of colocated regions on a particular data store
   *
   * @return map of region name to local colocated regions
   * @since GemFire 5.8Beta
   */
  public static Map<String, Region> getAllColocatedLocalDataSets(
      PartitionedRegion partitionedRegion, InternalRegionFunctionContext context) {
    Map<String, PartitionedRegion> colocatedRegions = getAllColocationRegions(partitionedRegion);
    Map<String, Region> colocatedLocalRegions = new HashMap<String, Region>();
    for (Iterator itr = colocatedRegions.entrySet().iterator(); itr.hasNext();) {
      Map.Entry me = (Entry) itr.next();
      final Region pr = (Region) me.getValue();
      colocatedLocalRegions.put((String) me.getKey(), context.getLocalDataSet(pr));
    }
    return colocatedLocalRegions;
  }

  public static Map<String, LocalDataSet> constructAndGetAllColocatedLocalDataSet(
      PartitionedRegion region, Set<Integer> bucketSet) {
    Map<String, LocalDataSet> colocatedLocalDataSets = new HashMap<String, LocalDataSet>();
    if (region.getColocatedWith() == null && (!region.isColocatedBy())) {
      colocatedLocalDataSets.put(region.getFullPath(), new LocalDataSet(region, bucketSet));
      return colocatedLocalDataSets;
    }
    Map<String, PartitionedRegion> colocatedRegions =
        ColocationHelper.getAllColocationRegions(region);
    for (Region colocatedRegion : colocatedRegions.values()) {
      colocatedLocalDataSets.put(colocatedRegion.getFullPath(),
          new LocalDataSet((PartitionedRegion) colocatedRegion, bucketSet));
    }
    colocatedLocalDataSets.put(region.getFullPath(), new LocalDataSet(region, bucketSet));
    return colocatedLocalDataSets;
  }

  public static Map<String, LocalDataSet> getColocatedLocalDataSetsForBuckets(
      PartitionedRegion region, Set<Integer> bucketSet) {
    if (region.getColocatedWith() == null && (!region.isColocatedBy())) {
      return Collections.emptyMap();
    }
    Map<String, LocalDataSet> ret = new HashMap<String, LocalDataSet>();
    Map<String, PartitionedRegion> colocatedRegions =
        ColocationHelper.getAllColocationRegions(region);
    for (Region colocatedRegion : colocatedRegions.values()) {
      ret.put(colocatedRegion.getFullPath(),
          new LocalDataSet((PartitionedRegion) colocatedRegion, bucketSet));
    }
    return ret;
  }

  /**
   * A utility method to retrieve all child partitioned regions that are directly colocated to the
   * specified partitioned region.<br>
   * <p>
   * For example, shipmentPR is colocated with orderPR and orderPR is colocated with customerPR.
   * <br>
   * getColocatedChildRegions(customerPR) will return List{orderPR}<br>
   * getColocatedChildRegions(orderPR) will return List{shipmentPR}<br>
   * getColocatedChildRegions(shipmentPR) will return empty List{}<br>
   *
   * @return list of all child partitioned regions colocated with the region
   * @since GemFire 5.8Beta
   */
  public static List<PartitionedRegion> getColocatedChildRegions(
      PartitionedRegion partitionedRegion) {
    List<PartitionedRegion> colocatedChildRegions = new ArrayList<PartitionedRegion>();
    Region prRoot = PartitionedRegionHelper.getPRRoot(partitionedRegion.getCache());
    PartitionRegionConfig prConf = null;
    // final List allPRNamesList = new ArrayList(prRoot.keySet());
    Iterator itr = prRoot.keySet().iterator();
    while (itr.hasNext()) {
      try {
        String prName = (String) itr.next();
        if (prName.equals(partitionedRegion.getRegionIdentifier())) {
          // region can't be a child of itself
          continue;
        }
        try {
          prConf = (PartitionRegionConfig) prRoot.get(prName);
        } catch (EntryDestroyedException ignore) {
          continue;
        }
        if (prConf == null) {
          // darrel says: I'm seeing an NPE in this code after pr->rem
          // merge so I added this check and continue
          continue;
        }
        int prID = prConf.getPRId();
        PartitionedRegion prRegion = PartitionedRegion.getPRFromId(prID);
        if (prRegion != null) {
          if (prRegion.getColocatedWith() != null) {
            if (prRegion.getColocatedWith().equals(partitionedRegion.getFullPath())
                || ("/" + prRegion.getColocatedWith()).equals(partitionedRegion.getFullPath())) {
              // only regions directly colocatedWith partitionedRegion are
              // added to the list...
              prRegion.waitOnBucketMetadataInitialization();
              colocatedChildRegions.add(prRegion);
            }
          }
        }
      } catch (PRLocallyDestroyedException e) {
        if (logger.isDebugEnabled()) {
          logger.debug("PRLocallyDestroyedException : Region ={} is locally destroyed on this node",
              prConf.getPRId(), e);
        }
      } catch (RegionDestroyedException e) {
        if (logger.isDebugEnabled()) {
          logger.debug("RegionDestroyedException : Region ={} is destroyed.", prConf.getPRId(), e);
        }
      }
    }

    // Fix for 44484 - Make the list of colocated child regions
    // is always in the same order on all nodes.
    Collections.sort(colocatedChildRegions, new Comparator<PartitionedRegion>() {
      @Override
      public int compare(PartitionedRegion o1, PartitionedRegion o2) {
        if (o1.isShadowPR() == o2.isShadowPR()) {
          return o1.getFullPath().compareTo(o2.getFullPath());
        }
        if (o1.isShadowPR()) {
          return 1;
        }
        return -1;
      }
    });
    return colocatedChildRegions;
  }

  // TODO why do we have this method here?
  public static Function getFunctionInstance(Serializable function) {
    Function functionInstance = null;
    if (function instanceof String) {
      functionInstance = FunctionService.getFunction((String) function);
      Assert.assertTrue(functionInstance != null,
          "Function " + function + " is not registered on this node ");
    } else {
      functionInstance = (Function) function;
    }
    return functionInstance;
  }

  public static PartitionedRegion getLeaderRegion(PartitionedRegion prRegion) {
    PartitionedRegion parentRegion;

    while ((parentRegion = getColocatedRegion(prRegion)) != null) {
      prRegion = parentRegion;
    }

    return prRegion;
  }

  private static String getRegionIdentifier(String regionName) {
    if (regionName.startsWith("/")) {
      return regionName.replace("/", "#");
    } else {
      return "#" + regionName.replace("/", "#");
    }
  }

  /**
   * Test to see if there are any persistent child regions of a partitioned region.
   */
  public static boolean hasPersistentChildRegion(PartitionedRegion region) {
    for (PartitionedRegion child : getColocatedChildRegions(region)) {
      if (child.getDataPolicy().withPersistence()) {
        return true;
      }
    }
    return false;
  }
}