/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.master.file;

import alluxio.AlluxioURI;
import alluxio.exception.ExceptionMessage;
import alluxio.exception.FileDoesNotExistException;
import alluxio.exception.InvalidPathException;
import alluxio.grpc.ListStatusPartialPOptions;
import alluxio.master.file.contexts.ListStatusContext;
import alluxio.master.file.meta.Inode;
import alluxio.master.file.meta.InodeTree;
import alluxio.master.file.meta.LockedInodePath;
import alluxio.master.metastore.ReadOnlyInodeStore;
import alluxio.resource.CloseableIterator;
import alluxio.util.io.PathUtils;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;

class ListStatusPartial {

  /**
   * If performing a partial listing, from an offset, this checks to ensure the offset
   * exists within the starting path.
   * @param path the listing path
   * @param context the listing context
   * @return the components of the path to the offset
   */
  static List<String> checkPartialListingOffset(
      InodeTree inodeTree,  AlluxioURI path, ListStatusContext context)
      throws FileDoesNotExistException, InvalidPathException {
    Optional<ListStatusPartialPOptions.Builder> pOptions = context.getPartialOptions();
    if (!pOptions.isPresent()) {
      return Collections.emptyList();
    }
    ListStatusPartialPOptions.Builder partialOptions = pOptions.get();
    List<String> partialPathNames = Collections.emptyList();
    if (context.isPartialListing() && partialOptions.getOffsetId() != 0) {
      try {
        // See if the inode from where to start the listing exists.
        partialPathNames = inodeTree.getPathInodeNames(partialOptions.getOffsetId());
      } catch (FileDoesNotExistException e) {
        throw new FileDoesNotExistException(
            ExceptionMessage.INODE_NOT_FOUND_PARTIAL_LISTING.getMessage(e.getMessage()),
            e.getCause());
      }
      String[] pathComponents = PathUtils.getPathComponents(path.getPath());
      // the offset path must be at least as long as the starting path
      if (partialPathNames.size() < pathComponents.length) {
        throw new FileDoesNotExistException(ExceptionMessage.INODE_NOT_FOUND_PARTIAL_LISTING
            .getMessage(path));
      }
      for (int i = 0; i < pathComponents.length; i++) {
        if (!partialPathNames.get(i).equals(pathComponents[i])) {
          throw new FileDoesNotExistException(ExceptionMessage.INODE_NOT_FOUND_PARTIAL_LISTING
              .getMessage(path));
        }
      }
    }
    return partialPathNames;
  }

  /**
   * If this is a partial listing, this will compute the path components from where to start
   * the listing from that come after the root listing path.
   *
   * @param path the root listing path
   * @param context the context of the operation
   * @param pathNames the full path from where the partial listing is expected to start,
   *                  null if this is the first listing
   * @param rootPath the locked root path of the listing
   * @return the path components after the root path from where to start the partial listing,
   * or null if the listing should start from the beginning of the root path
   */
  static List<String> computePartialListingPaths(
      AlluxioURI path, ListStatusContext context,
      List<String> pathNames, LockedInodePath rootPath)
      throws InvalidPathException {
    if (pathNames.isEmpty()) {
      Optional<ListStatusPartialPOptions.Builder> partialOptions = context.getPartialOptions();
      if (partialOptions.isPresent()) {
        // use the startAfter option, since this is the first listing
        if (!partialOptions.get().getStartAfter().isEmpty()) {
          String startAfter = partialOptions.get().getStartAfter();
          if (startAfter.startsWith(AlluxioURI.SEPARATOR)) {
            // this path starts from the root, so we must remove the prefix
            String startAfterCheck = startAfter.substring(0,
                Math.min(path.getPath().length(), startAfter.length()));
            if (!path.getPath().startsWith(startAfterCheck)) {
              throw new InvalidPathException(
                  ExceptionMessage.START_AFTER_DOES_NOT_MATCH_PATH
                      .getMessage(startAfter, path.getPath()));
            }
            startAfter = startAfter.substring(
                Math.min(startAfter.length(), path.getPath().length()));
          }
          startAfter = startAfter.startsWith(AlluxioURI.SEPARATOR) ? startAfter
                : AlluxioURI.SEPARATOR + startAfter;
          String[] startAfterComponents = PathUtils.getPathComponents(
                startAfter);
          ArrayList<String> startAfterList = new ArrayList<>(startAfterComponents.length - 1);
          startAfterList.addAll(Arrays.asList(startAfterComponents)
              .subList(1, startAfterComponents.length));
          return startAfterList;
        }
      }
      // otherwise, start from the beginning of the listing
      return Collections.emptyList();
    }
    // compute where to start from in each depth, we skip past the rootInodes, since that is
    // where we start the traversal from
    if (pathNames.size() <= rootPath.size()) {
      return Collections.emptyList();
    }
    ArrayList<String> partialPath = new ArrayList<>(pathNames.size() - rootPath.size());
    partialPath.addAll(pathNames.subList(rootPath.size(), pathNames.size()));
    return partialPath;
  }

  /**
   * If listing using a prefix and a partial path, this will compute whether the prefix
   * exists in the partial path.
   *
   * @param partialPath the components of the path from where to start the listing
   * @return the components of the prefix split by the delimiter /
   * @throws InvalidPathException if the path in prefixComponents does not exist in partialPath
   */
  static List<String> checkPrefixListingPaths(
      ListStatusContext context, List<String> partialPath)
      throws InvalidPathException {
    Optional<ListStatusPartialPOptions.Builder> pOptions = context.getPartialOptions();
    if (!pOptions.isPresent()) {
      return Collections.emptyList();
    }
    List<String> prefixComponents;
    ListStatusPartialPOptions.Builder partialOptions = pOptions.get();
    if (!partialOptions.getPrefix().isEmpty()) {
      // compute the prefix as path components, removing the first empty string
      prefixComponents = Arrays.stream(PathUtils.getPathComponents(
              new AlluxioURI(
                  partialOptions.getPrefix()).getPath())).skip(1)
          .collect(Collectors.toList());
    } else {
      prefixComponents = Collections.emptyList();
    }
    // we only have to check the prefix if we are doing a partial listing,
    // and we are not on the initial partial listing call
    if (partialPath.isEmpty()
        || !(partialOptions.hasOffsetId() && partialOptions.getOffsetId() != 0)) {
      return prefixComponents;
    }
    // for each component the prefix must be the same as the partial path component
    // except at the last component where the prefix must be contained in the partial path component
    if (!hasPrefixComponentsCanBeLonger(partialPath, prefixComponents)) {
      throw new InvalidPathException(
          ExceptionMessage.PREFIX_DOES_NOT_MATCH_PATH
              .getMessage(prefixComponents, partialPath));
    }
    return prefixComponents;
  }

  /**
   * This will generate the iterator of children during a partial listing call
   * for the given depth and options.
   * @param inodeStore the inode store
   * @param inode the parent inode
   * @param partialPath the components of the path from where to start the listing
   * @param prefixComponents the components of the prefix to list
   * @param depth the depth of the children
   * @param context the list status context
   * @return the iterator of children
   */
  static CloseableIterator<? extends Inode> getChildrenIterator(
      ReadOnlyInodeStore inodeStore, Inode inode, List<String> partialPath,
      List<String> prefixComponents, int depth, ListStatusContext context)
      throws InvalidPathException {

    // Check if we should process all children, or just the partial listing, or just a prefix
    String prefix = null;
    if (prefixComponents.size() > depth) {
      prefix = prefixComponents.get(depth);
    }
    ListStatusPartialPOptions.Builder partialOptions = context.getPartialOptions().orElseThrow(
        () -> new RuntimeException("Method should only be called when doing partial listing"));
    if (partialOptions.hasOffsetId() || partialOptions.hasStartAfter()) {
      // If we have already processed the first entry in the partial path
      // then we just process from the start of the children, so we list from the empty string
      String listFrom = "";
      if (partialPath.size() > depth) {
        listFrom = partialPath.get(depth);
        if (prefix != null) {
          // the prefix must have the same components as the start point in the partial listing
          // because the node must have matched the prefix in the previous call to the listing
          if ((prefixComponents.size() > depth + 1 && !prefix.equals(listFrom))) {
            throw new InvalidPathException(ExceptionMessage.PREFIX_DOES_NOT_MATCH_PATH.getMessage(
                prefix, listFrom));
          }
        }
      }
      if (prefix != null) {
        return inodeStore.getChildrenPrefixFrom(inode.getId(), prefix, listFrom);
      } else {
        return inodeStore.getChildrenFrom(inode.getId(), listFrom);
      }
    } else if (prefix != null) {
      return inodeStore.getChildrenPrefix(inode.getId(), prefix);
    }
    // Perform a full listing of all children sorted by name.
    return inodeStore.getChildren(inode.asDirectory());
  }

  /**
   * Check if the given prefixComponents are a prefix of pathComponents.
   * For each component at index i in prefixComponents, it must be equal
   * to the component at index i in pathComponents, except the last component
   * in prefixComponents must be a prefix of the component at the same index
   * in pathComponents. Note that if there are more prefix components than path components,
   * this function will still return true as long as the previous conditions are satisfied.
   * @param pathComponents the prefix components
   * @param prefixComponents the path components
   * @return true if the prefixComponents are a prefix of pathComponents, false otherwise
   */
  static boolean hasPrefixComponentsCanBeLonger(
      List<String> pathComponents, List<String> prefixComponents) {
    for (int i = 0; i < Math.min(pathComponents.size(), prefixComponents.size()); i++) {
      if ((i < pathComponents.size() - 1 && !pathComponents.get(i).equals(prefixComponents.get(i)))
          || (i == pathComponents.size() - 1 && !pathComponents.get(i).startsWith(
          prefixComponents.get(i)))) {
        return false;
      }
    }
    return true;
  }
}
