/********************************************************************
 * 2014 -
 * open source under Apache License Version 2.0
 ********************************************************************/
/**
 * 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.
 */
#include "DirectoryIterator.h"
#include "EncryptionZoneIterator.h"
#include "Exception.h"
#include "ExceptionInternal.h"
#include "FileSystem.h"
#include "FileSystemImpl.h"
#include "FileSystemKey.h"
#include "Hash.h"
#include "SessionConfig.h"
#include "Thread.h"
#include "Token.h"
#include "Unordered.h"
#include "WritableUtils.h"

#include <algorithm>
#include <string>
#include <krb5/krb5.h>

using namespace Hdfs::Internal;

namespace Hdfs {

namespace Internal {

static std::string ExtractPrincipalFromTicketCache(
    const std::string & cachePath) {
    krb5_context cxt = NULL;
    krb5_ccache ccache = NULL;
    krb5_principal principal = NULL;
    krb5_error_code ec = 0;
    std::string errmsg, retval;
    char * priName = NULL;

    if (!cachePath.empty()) {
        if (0 != setenv("KRB5CCNAME", cachePath.c_str(), 1)) {
            THROW(HdfsIOException, "Cannot set env parameter \"KRB5CCNAME\"");
        }
    }

    do {
        if (0 != (ec = krb5_init_context(&cxt))) {
            break;
        }

        if (0 != (ec = krb5_cc_default(cxt, &ccache))) {
            break;
        }

        if (0 != (ec = krb5_cc_get_principal(cxt, ccache, &principal))) {
            break;
        }

        if (0 != (ec = krb5_unparse_name(cxt, principal, &priName))) {
            break;
        }
    } while (0);

    if (!ec) {
        retval = priName;
    } else {
        if (cxt) {
            errmsg = krb5_get_error_message(cxt, ec);
        } else {
            errmsg = "Cannot initialize kerberos context";
        }
    }

    if (priName != NULL) {
        krb5_free_unparsed_name(cxt, priName);
    }

    if (principal != NULL) {
        krb5_free_principal(cxt, principal);
    }

    if (ccache != NULL) {
        krb5_cc_close(cxt, ccache);
    }

    if (cxt != NULL) {
        krb5_free_context(cxt);
    }

    if (!errmsg.empty()) {
        THROW(HdfsIOException,
              "FileSystem: Failed to extract principal from ticket cache: %s",
              errmsg.c_str());
    }

    return retval;
}


static std::string ExtractPrincipalFromToken(const Token & token) {
    std::string realUser, owner;
    std::string identifier = token.getIdentifier();
    WritableUtils cin(&identifier[0], identifier.size());
    char version;

    try {
        version = cin.readByte();

        if (version != 0) {
            THROW(HdfsIOException, "Unknown version of delegation token");
        }

        owner = cin.ReadText();
        cin.ReadText();
        realUser = cin.ReadText();
        return realUser.empty() ? owner : realUser;
    } catch (const std::range_error & e) {
    }

    THROW(HdfsIOException, "Cannot extract principal from token");
}
}

FileSystem::FileSystem(const Config & conf) :
    conf(conf), impl(NULL) {
}

FileSystem::FileSystem(const FileSystem & other) :
    conf(other.conf), impl(NULL) {
    if (other.impl) {
        impl = new FileSystemWrapper(other.impl->filesystem);
    }
}

FileSystem & FileSystem::operator =(const FileSystem & other) {
    if (this == &other) {
        return *this;
    }

    conf = other.conf;

    if (impl) {
        delete impl;
        impl = NULL;
    }

    if (other.impl) {
        impl = new FileSystemWrapper(other.impl->filesystem);
    }

    return *this;
}

FileSystem::~FileSystem() {
    if (impl) {
        try {
            disconnect();
        } catch (...) {
        }
    }
}

void FileSystem::connect() {
    Internal::SessionConfig sconf(conf);
    connect(sconf.getDefaultUri().c_str(), NULL, NULL);
}

/**
 * Connect to hdfs
 * @param uri hdfs connection uri, hdfs://host:port
 */
void FileSystem::connect(const char * uri) {
    connect(uri, NULL, NULL);
}

static FileSystemWrapper * ConnectInternal(const char * uri,
        const std::string & principal, const Token * token, Config & conf) {
    if (NULL == uri || 0 == strlen(uri)) {
        THROW(InvalidParameter, "Invalid HDFS uri.");
    }

    FileSystemKey key(uri, principal.c_str());

    if (token) {
        key.addToken(*token);
    }

    return new FileSystemWrapper(shared_ptr<FileSystemInter>(new FileSystemImpl(key, conf)));
}

/**
 * Connect to hdfs with user or token
 * 	username and token cannot be set at the same time
 * @param uri connection uri.
 * @param username user used to connect to hdfs
 * @param token token used to connect to hdfs
 */
void FileSystem::connect(const char * uri, const char * username, const char * token) {
    AuthMethod auth;
    std::string principal;

    if (impl) {
        THROW(HdfsIOException, "FileSystem: already connected.");
    }

    try {
        SessionConfig sconf(conf);
        auth = RpcAuth::ParseMethod(sconf.getRpcAuthMethod());

        if (token && auth != AuthMethod::SIMPLE) {
            Token t;
            t.fromString(token);
            principal = ExtractPrincipalFromToken(t);
            impl = ConnectInternal(uri, principal, &t, conf);
            impl->filesystem->connect();
            return;
        } else if (username) {
            principal = username;
        }

        if (auth == AuthMethod::KERBEROS) {
            principal = ExtractPrincipalFromTicketCache(sconf.getKerberosCachePath());
        }

        impl = ConnectInternal(uri, principal, NULL, conf);
        impl->filesystem->connect();
    } catch (...) {
        delete impl;
        impl = NULL;
        throw;
    }
}

/**
 * disconnect from hdfs
 */
void FileSystem::disconnect() {
    delete impl;
    impl = NULL;
}

/**
 * To get default number of replication.
 * @return the default number of replication.
 */
int FileSystem::getDefaultReplication() const {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->getDefaultReplication();
}

/**
 * To get the default number of block size.
 * @return the default block size.
 */
int64_t FileSystem::getDefaultBlockSize() const {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->getDefaultBlockSize();
}

/**
 * To get the home directory.
 * @return home directory.
 */
std::string FileSystem::getHomeDirectory() const {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->getHomeDirectory();
}

/**
 * To delete a file or directory.
 * @param path the path to be deleted.
 * @param recursive if path is a directory, delete the contents recursively.
 * @return return true if success.
 */
bool FileSystem::deletePath(const char * path, bool recursive) {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->deletePath(path, recursive);
}

/**
 * To create a directory which given permission.
 * @param path the directory path which is to be created.
 * @param permission directory permission.
 * @return return true if success.
 */
bool FileSystem::mkdir(const char * path, const Permission & permission) {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->mkdir(path, permission);
}

/**
 * To create a directory which given permission.
 * If parent path does not exits, create it.
 * @param path the directory path which is to be created.
 * @param permission directory permission.
 * @return return true if success.
 */
bool FileSystem::mkdirs(const char * path, const Permission & permission) {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->mkdirs(path, permission);
}

/**
 * To get path information.
 * @param path the path which information is to be returned.
 * @return the path information.
 */
FileStatus FileSystem::getFileStatus(const char * path) const {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->getFileStatus(path);
}

/**
 * Return an array containing hostnames, offset and size of
 * portions of the given file.
 *
 * This call is most helpful with DFS, where it returns
 * hostnames of machines that contain the given file.
 *
 * The FileSystem will simply return an elt containing 'localhost'.
 *
 * @param path path is used to identify an FS since an FS could have
 *          another FS that it could be delegating the call to
 * @param start offset into the given file
 * @param len length for which to get locations for
 */
std::vector<BlockLocation> FileSystem::getFileBlockLocations(const char * path,
        int64_t start, int64_t len) {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->getFileBlockLocations(path, start, len);
}

/**
 * list the contents of a directory.
 * @param path the directory path.
 * @return Return a iterator to visit all elements in this directory.
 */
DirectoryIterator FileSystem::listDirectory(const char * path)  {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->listDirectory(path, false);
}

/**
 * list all the contents of a directory.
 * @param path The directory path.
 * @return Return a vector of file informations in the directory.
 */
std::vector<FileStatus> FileSystem::listAllDirectoryItems(const char * path) {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->listAllDirectoryItems(path, false);
}

/**
 * To set the owner and the group of the path.
 * username and groupname cannot be empty at the same time.
 * @param path the path which owner of group is to be changed.
 * @param username new user name.
 * @param groupname new group.
 */
void FileSystem::setOwner(const char * path, const char * username,
                          const char * groupname) {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    impl->filesystem->setOwner(path, username, groupname);
}

/**
 * To set the access time or modification time of a path.
 * @param path the path which access time or modification time is to be changed.
 * @param mtime new modification time.
 * @param atime new access time.
 */
void FileSystem::setTimes(const char * path, int64_t mtime, int64_t atime) {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    impl->filesystem->setTimes(path, mtime, atime);
}

/**
 * To set the permission of a path.
 * @param path the path which permission is to be changed.
 * @param permission new permission.
 */
void FileSystem::setPermission(const char * path,
                               const Permission & permission) {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    impl->filesystem->setPermission(path, permission);
}

/**
 * To set the number of replication.
 * @param path the path which number of replication is to be changed.
 * @param replication new number of replication.
 * @return return true if success.
 */
bool FileSystem::setReplication(const char * path, short replication) {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->setReplication(path, replication);
}

/**
 * To rename a path.
 * @param src old path.
 * @param dst new path.
 * @return return true if success.
 */
bool FileSystem::rename(const char * src, const char * dst) {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->rename(src, dst);
}

/**
 * To set working directory.
 * @param path new working directory.
 */
void FileSystem::setWorkingDirectory(const char * path) {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    impl->filesystem->setWorkingDirectory(path);
}

/**
 * To get working directory.
 * @return working directory.
 */
std::string FileSystem::getWorkingDirectory() const {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->getWorkingDirectory();
}

/**
 * To test if the path exist.
 * @param path the path which is to be tested.
 * @return return true if the path exist.
 */
bool FileSystem::exist(const char * path) const {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->exist(path);
}

/**
 * To get the file system status.
 * @return the file system status.
 */
FileSystemStats FileSystem::getStats() const {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->getFsStats();
}

/**
 * Truncate the file in the indicated path to the indicated size.
 * @param src The path to the file to be truncated
 * @param size The size the file is to be truncated to
 *
 * @return true if and client does not need to wait for block recovery,
 * false if client needs to wait for block recovery.
 */
bool FileSystem::truncate(const char * src, int64_t size) {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->truncate(src, size);
}

std::string FileSystem::getDelegationToken(const char * renewer) {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->getDelegationToken(renewer);
}

/**
 * Get a valid Delegation Token using the default user as renewer.
 *
 * @return Token
 * @throws IOException
 */
std::string FileSystem::getDelegationToken() {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->getDelegationToken();
}

/**
 * Renew an existing delegation token.
 *
 * @param token delegation token obtained earlier
 * @return the new expiration time
 * @throws IOException
 */
int64_t FileSystem::renewDelegationToken(const std::string & token) {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return  impl->filesystem->renewDelegationToken(token);
}

/**
 * Cancel an existing delegation token.
 *
 * @param token delegation token
 * @throws IOException
 */
void FileSystem::cancelDelegationToken(const std::string & token) {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    impl->filesystem->cancelDelegationToken(token);
}


/**
 * Create encryption zone for the directory with specific key name
 * @param path the directory path which is to be created.
 * @param keyname The key name of the encryption zone 
 * @return return true if success.
 */
bool FileSystem::createEncryptionZone(const char * path, const char * keyName) {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->createEncryptionZone(path, keyName);
}

/**
* To get encryption zone information.
* @param path the path which information is to be returned.
* @return the encryption zone information.
*/
EncryptionZoneInfo FileSystem::getEZForPath(const char * path) {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }
   
    return impl->filesystem->getEZForPath(path);
}

/**
 * list the contents of an encryption zone.
 * @return Return a iterator to visit all elements in this encryption zone.
 */
EncryptionZoneIterator FileSystem::listEncryptionZone()  {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->listEncryptionZone();
}

/**
* list all the contents of encryption zones.
* @param id the index of encryption zones.
* @return Return a vector of encryption zones information..
*/
std::vector<EncryptionZoneInfo> FileSystem::listAllEncryptionZoneItems() {
    if (!impl) {
        THROW(HdfsIOException, "FileSystem: not connected.");
    }

    return impl->filesystem->listAllEncryptionZoneItems();
}

}