/*
 * 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.FileDoesNotExistException;
import alluxio.exception.InvalidPathException;
import alluxio.master.file.meta.Inode;
import alluxio.master.file.meta.InodeDirectory;
import alluxio.master.file.meta.MountTable;
import alluxio.underfs.UnderFileStatus;
import alluxio.underfs.UnderFileSystem;
import alluxio.underfs.options.ListOptions;
import alluxio.util.io.PathUtils;

import com.google.common.base.Preconditions;

import java.io.IOException;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import javax.annotation.concurrent.NotThreadSafe;

/**
 * Class to check if the contents of the under storage is in-sync with the master.
 */
@NotThreadSafe
public final class UfsSyncChecker {

  /** UFS directories for which list was called. */
  private Map<String, UnderFileStatus[]> mListedDirectories;

  /** This manages the file system mount points. */
  private final MountTable mMountTable;

  /**
   * Create a new instance of {@link UfsSyncChecker}.
   *
   * @param mountTable to resolve path in under storage
   */
  public UfsSyncChecker(MountTable mountTable) {
    mListedDirectories = new HashMap<>();
    mMountTable = mountTable;
  }
  /**
   * Check if immediate children of directory are in sync with UFS.
   *
   * @param inode read-locked directory to check
   * @param alluxioUri path of directory to to check
   * @return true is contents of directory match
   */
  public boolean isUFSDeleteSafe(InodeDirectory inode, AlluxioURI alluxioUri)
      throws FileDoesNotExistException, InvalidPathException, IOException {
    Preconditions.checkArgument(inode.isPersisted());

    UnderFileStatus[] ufsChildren = getChildrenInUFS(alluxioUri);
    // Empty directories are not persisted
    if (ufsChildren == null) {
      return true;
    }
    // Sort-merge compare
    Arrays.sort(ufsChildren, new Comparator<UnderFileStatus>() {
      @Override
      public int compare(UnderFileStatus a, UnderFileStatus b) {
        return a.getName().compareTo(b.getName());
      }
    });
    int numInodeChildren = inode.getChildren().size();
    Inode[] inodeChildren = new Inode[numInodeChildren];
    inodeChildren = inode.getChildren().toArray(inodeChildren);
    Arrays.sort(inodeChildren, new Comparator<Inode>() {
      @Override
      public int compare(Inode a, Inode b) {
        return a.getName().compareTo(b.getName());
      }
    });
    int ufsPos = 0;
    int inodePos = 0;
    while (ufsPos < ufsChildren.length && inodePos < numInodeChildren) {
      String ufsName = ufsChildren[ufsPos].getName();
      if (ufsName.endsWith(AlluxioURI.SEPARATOR)) {
        ufsName = ufsName.substring(0, ufsName.length() - 1);
      }
      if (ufsName.equals(inodeChildren[inodePos].getName())) {
        ufsPos++;
      }
      inodePos++;
    }
    return ufsPos == ufsChildren.length;
  }

  /**
   * Get the children in under storage for given alluxio path.
   *
   * @param alluxioUri alluxio path
   * @return the list of children in under storage
   * @throws InvalidPathException if aluxioUri is invalid
   * @throws IOException if a non-alluxio error occurs
   */
  private UnderFileStatus[] getChildrenInUFS(AlluxioURI alluxioUri)
      throws InvalidPathException, IOException {
    MountTable.Resolution resolution = mMountTable.resolve(alluxioUri);
    AlluxioURI ufsUri = resolution.getUri();
    UnderFileSystem ufs = resolution.getUfs();

    AlluxioURI curUri = ufsUri;
    while (curUri != null) {
      if (mListedDirectories.containsKey(curUri.toString())) {
        List<UnderFileStatus> childrenList = new LinkedList<>();
        for (UnderFileStatus child : mListedDirectories.get(curUri.toString())) {
          String childPath = PathUtils.concatPath(curUri, child.getName());
          String prefix = PathUtils.normalizePath(ufsUri.toString(), AlluxioURI.SEPARATOR);
          // TODO(adit): only return immediate children
          if (childPath.startsWith(prefix) && childPath.length() > prefix.length()) {
            childrenList.add(new UnderFileStatus(childPath.substring(prefix.length()),
                child.isDirectory()));
          }
        }
        return childrenList.toArray(new UnderFileStatus[childrenList.size()]);
      }
      curUri = curUri.getParent();
    }
    UnderFileStatus[] children =
        ufs.listStatus(ufsUri.toString(), ListOptions.defaults().setRecursive(true));
    // Assumption: multiple mounted UFSs cannot have the same ufsUri
    if (children != null) {
      mListedDirectories.put(ufsUri.toString(), children);
    }
    // TODO(adit): only return immediate children
    return children;
  }
}

@@ -4500,64 +4500,8 @@
 R);%0A
-          // TODO(adit): only return immediate children%0A
     
@@ -4578,24 +4578,24 @@
 length()) %7B%0A
+
             
@@ -4737,32 +4737,45 @@
 %0A        return 
+trimIndirect(
 childrenList.toA
@@ -4820,16 +4820,17 @@
 size()%5D)
+)
 ;%0A      
@@ -5161,73 +5161,716 @@
     
-// TODO(adit): only return immediate children%0A    return children
+return trimIndirect(children);%0A  %7D%0A%0A  /**%0A   * Remove indirect children from children list returned from recursive listing.%0A   *%0A   * @param children list from recursive listing%0A   * @return trimmed list, null if input is null%0A   */%0A  private UnderFileStatus%5B%5D trimIndirect(UnderFileStatus%5B%5D children) %7B%0A    if (children == null) %7B%0A      return null;%0A    %7D%0A    List%3CUnderFileStatus%3E childrenList = new LinkedList%3C%3E();%0A    for (UnderFileStatus child : children) %7B%0A      int index = child.getName().indexOf(AlluxioURI.SEPARATOR);%0A      if (index %3C 0 %7C%7C index == child.getName().length()) %7B%0A        childrenList.add(child);%0A      %7D%0A    %7D%0A    return childrenList.toArray(new UnderFileStatus%5BchildrenList.size()%5D)
 ;%0A  

Fix ufsyncchecker