/*
 * Copyright © 2016 Cask Data, Inc.
 *
 * Licensed 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 co.cask.cdap.internal.app.namespace;

import co.cask.cdap.common.conf.CConfiguration;
import co.cask.cdap.common.conf.Constants;
import co.cask.cdap.common.namespace.NamespaceQueryAdmin;
import co.cask.cdap.common.namespace.NamespacedLocationFactory;
import co.cask.cdap.explore.client.ExploreFacade;
import co.cask.cdap.explore.service.ExploreException;
import co.cask.cdap.proto.NamespaceMeta;
import co.cask.cdap.proto.id.NamespaceId;
import com.google.common.base.Strings;
import org.apache.twill.filesystem.Location;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.nio.file.FileAlreadyExistsException;
import java.sql.SQLException;

/**
 * Performs common namespace admin operations on storage providers (HBase, Filesystem, Hive, etc)
 */
abstract class AbstractStorageProviderNamespaceAdmin implements StorageProviderNamespaceAdmin {
  private static final Logger LOG = LoggerFactory.getLogger(AbstractStorageProviderNamespaceAdmin.class);

  private final CConfiguration cConf;
  private final NamespacedLocationFactory namespacedLocationFactory;
  private final ExploreFacade exploreFacade;
  private final NamespaceQueryAdmin namespaceQueryAdmin;


  AbstractStorageProviderNamespaceAdmin(CConfiguration cConf,
                                        NamespacedLocationFactory namespacedLocationFactory,
                                        ExploreFacade exploreFacade,
                                        NamespaceQueryAdmin namespaceQueryAdmin) {
    this.cConf = cConf;
    this.namespacedLocationFactory = namespacedLocationFactory;
    this.exploreFacade = exploreFacade;
    this.namespaceQueryAdmin = namespaceQueryAdmin;
  }

  /**
   * Create a namespace in the File System and Hive.
   *
   * @param namespaceMeta {@link NamespaceMeta} for the namespace to create
   * @throws IOException if there are errors while creating the namespace in the File System
   * @throws ExploreException if there are errors while deleting the namespace in Hive
   * @throws SQLException if there are errors while deleting the namespace in Hive
   */
  @Override
  public void create(NamespaceMeta namespaceMeta) throws IOException, ExploreException, SQLException {

    createLocation(namespaceMeta);

    if (cConf.getBoolean(Constants.Explore.EXPLORE_ENABLED)) {
      try {
        exploreFacade.createNamespace(namespaceMeta);
      } catch (ExploreException | SQLException e) {
        // if we failed to create a namespace in explore then delete the earlier created location for the namespace
        deleteLocation(namespaceMeta.getNamespaceId());
        throw e;
      }
    }
  }

  /**
   * Deletes the namespace directory on the FileSystem and Hive.
   *
   * @param namespaceId {@link NamespaceId} for the namespace to delete
   * @throws IOException if there are errors while deleting the namespace in the File System
   * @throws ExploreException if there are errors while deleting the namespace in Hive
   * @throws SQLException if there are errors while deleting the namespace in Hive
   */
  @Override
  public void delete(NamespaceId namespaceId) throws IOException, ExploreException, SQLException {

    deleteLocation(namespaceId);

    if (cConf.getBoolean(Constants.Explore.EXPLORE_ENABLED)) {
      exploreFacade.removeNamespace(namespaceId.toId());
    }
  }

  private void deleteLocation(NamespaceId namespaceId) throws IOException {
    // TODO: CDAP-1581: Implement soft delete
    Location namespaceHome = namespacedLocationFactory.get(namespaceId.toId());
    try {
      if (hasCustomLocation(namespaceQueryAdmin.get(namespaceId.toId()))) {
        LOG.debug("Custom location mapping %s was found while deleting namespace %s. Skipping location delete.",
                  namespaceHome, namespaceId);
      } else {
        // a custom location was not provided for this namespace so cdap is responsible for managing the lifecycle of
        // the location hence delete it.
        if (namespaceHome.exists() && !namespaceHome.delete(true)) {
          throw new IOException(String.format("Error while deleting home directory '%s' for namespace '%s'",
                                              namespaceHome, namespaceId));
        }
      }
    } catch (Exception e) {
      throw new IOException(String.format("Error while deleting home directory %s for namespace %s ", namespaceHome,
                                          namespaceId), e);
    }
  }

  private void createLocation(NamespaceMeta namespaceMeta) throws IOException {
    Location namespaceHome;
    if (hasCustomLocation(namespaceMeta)) {
      // a custom location was provided
      // check that its an absolute path
      File customLocation = new File(namespaceMeta.getConfig().getRootDirectory());
      if (!customLocation.isAbsolute()) {
        throw new IOException(String.format("Cannot create the namespace '%s' with the given custom " +
                                              "location %s. Custom location must be absolute path.",
                                            namespaceMeta.getName(), customLocation));
      }
      // since this is a custom location we expect it to exists. Get the custom location for the namespace from
      // namespaceLocationFactory since the location needs to be aware of local/distributed fs.
      Location customNamespacedLocation = namespacedLocationFactory.get(namespaceMeta.getNamespaceId().toId());
      if (!customNamespacedLocation.exists()) {
        // TODO: Add username in the below exception message.
        throw new IOException(String.format(
          "The provided home directory '%s' for namespace '%s' does not exists. Please create it on filesystem " +
            "with sufficient privileges for the user and then try creating a namespace.",
          customNamespacedLocation.toString(), namespaceMeta.getNamespaceId()));
      }
      // we also expect it to empty since non-empty directories can lead to various inconsistencies CDAP-6743
      if (customLocation.list().length != 0) {
        // TODO: Add username in the below exception message.
        throw new IOException(String.format(
          "The provided home directory '%s' for namespace '%s' is not empty. Namespace can only be mapped to an " +
            "empty directory. Please try creating the namespace again with an empty directory mapping.",
          customNamespacedLocation.toString(), namespaceMeta.getNamespaceId()));
      }
    } else {
      // no namespace custom location was provided one must be created by cdap
      namespaceHome = namespacedLocationFactory.get(namespaceMeta.getNamespaceId().toId());
      if (namespaceHome.exists()) {
        throw new FileAlreadyExistsException(namespaceHome.toString());

      }
      // create namespace home dir
      if (!namespaceHome.mkdirs()) {
        throw new IOException(String.format("Error while creating home directory '%s' for namespace '%s'",
                                            namespaceHome, namespaceMeta.getNamespaceId()));
      }
    }
  }

  private boolean hasCustomLocation(NamespaceMeta namespaceMeta) {
    return !Strings.isNullOrEmpty(namespaceMeta.getConfig().getRootDirectory());
  }
}

@@ -6105,70 +6105,8 @@
 ) %7B%0A
-        // TODO: Add username in the below exception message.%0A
     
@@ -6314,16 +6314,19 @@
 he user 
+%25s 
 and then
@@ -6423,32 +6423,84 @@
 getNamespaceId()
+,%0A          namespaceMeta.getConfig().getPrincipal()
 ));%0A      %7D%0A    
@@ -6656,70 +6656,8 @@
 ) %7B%0A
-        // TODO: Add username in the below exception message.%0A
     
@@ -6777,77 +6777,8 @@
 pty.
- Namespace can only be mapped to an %22 +%0A            %22empty directory.
  Ple
@@ -6808,16 +6808,33 @@
 mespace 
+%22 +%0A            %22
 again wi
@@ -6862,16 +6862,58 @@
  mapping
+ and sufficient privileges for the user %25s
 .%22,%0A    
@@ -6977,32 +6977,84 @@
 getNamespaceId()
+,%0A          namespaceMeta.getConfig().getPrincipal()
 ));%0A      %7D%0A    

Address comments