/********************************************************************
* CONFIDENTIAL - All source code is the propriety and confidential  *
* information of David Sorber.                                      *
*                                                                   *
* Copyright 2024 David Sorber                                       *
* Unpublished -- all rights reserved under the copyright laws       *
* of the United States.                                             *
********************************************************************/
#include <algorithm>
#include <cerrno>
#include <chrono>
#include <cstring>
#include <string_view>
#include <vector>

#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/sendfile.h>
#include <unistd.h>

#include "CopyManager.h"
#include "Logger.h"
#include "Utilities.h"


DBS::CopyManager::CopyManager(CopyManagerOptions& options)
    : r_options(options),
      m_terminate(false),
      p_mainThread(nullptr)
{
}

DBS::CopyManager::~CopyManager()
{
    // Add processor thread terminators
    for (uint32_t idx = 0; idx < r_options.getNumCopyThreads(); ++idx)
    {
        m_entryQueue.push_back(ENTRY_TERMINATOR);
    }

    // Join and delete thread terminators
    for (auto& thread : m_copyThreads)
    {
        thread->join();
        delete thread;
    }
    m_copyThreads.clear();

    // Cleanup the main thread
    if (p_mainThread != nullptr)
    {
        if (p_mainThread->joinable())
        {
            p_mainThread->join();
        }
        delete p_mainThread;
        p_mainThread = nullptr;
    }
}

int DBS::CopyManager::run()
{
    // Copy top level source directory itself to destination directory rather
    // than just the contents of the source directory. To support this option
    // modify the destination directory path and pre-create the top level
    // directory.
    if (r_options.getCopyTopLevel())
    {
        // Determine source top level directory
        std::string srcPath{r_options.getSrcPath()};
        if (srcPath.back() == '/')
        {
            srcPath.erase(srcPath.size() - 1);
        }

        std::vector<std::string> tokens;
        tokenize(srcPath, "/", tokens);

        sfs::path newDestPath{r_options.getDestPath()};
        newDestPath /= tokens.back();

        std::error_code errorCode;
        if (! sfs::create_directories(newDestPath, errorCode))
        {
            // Error code 0 means the directory already existed
            if (errorCode.value() != 0)
            {
                LOG(error) << "create_directories(" << newDestPath << ") -- "
                           << errorCode.value() << " -- " << errorCode.message();
                return -1;
            }
        }
        // TODO: handle top level directory metadata

        // Reassign destination directory
        r_options.setDestPath(newDestPath);
    }

    // Spawn processor threads
    for (uint32_t idx = 0; idx < r_options.getNumCopyThreads(); ++idx)
    {
        m_copyThreads.push_back(
            new std::thread(&CopyManager::copyThreadBody, this));
    }

    // Spawn the main thread
    p_mainThread = new std::thread(&CopyManager::mainThreadBody, this);
    p_mainThread->join();

    return 0;
}

void DBS::CopyManager::mainThreadBody()
{
    const std::string& srcPath(r_options.getSrcPath());
    for (const auto& dirEntry : sfs::recursive_directory_iterator(srcPath))
    {
        // Flow control -- delay until the entry queue is bewlo the specified
        // threshold size before proceeding
        while (m_entryQueue.size() >= r_options.getFlowControlThreshold())
        {
            std::this_thread::sleep_for(std::chrono::milliseconds(2));
        }

//        std::cout << "PATH: " << dirEntry << std::endl;
        m_entryQueue.push_back(dirEntry);
    }

    // Add processor thread terminators
    for (uint32_t idx = 0; idx < r_options.getNumCopyThreads(); ++idx)
    {
        m_entryQueue.push_back(ENTRY_TERMINATOR);
    }
}

void DBS::CopyManager::copyThreadBody()
{
    sfs::directory_entry entry;
    struct stat statBuf{};
    int rc = 0;
    std::error_code errorCode;
    int fromFd = 0;
    int toFd = 0;

    while (true)
    {
        // Grab entry from queue
        entry = m_entryQueue.pop_front();
        if (entry == ENTRY_TERMINATOR)
        {
            break;
        }

        if (entry.is_directory())
        {
            // Directory entry
//            const char* destDirPath = entry.path().string().c_str();
//            rc = stat(destDirPath, &statBuf);

            // Create destination path
            auto destDirPath = sfs::path{r_options.getDestPath()} /
                                     sfs::relative(entry.path(),
                                              r_options.getSrcPath());

            if (! sfs::create_directories(destDirPath, errorCode))
            {
                // Error code 0 means the directory already existed
                if (errorCode.value() != 0)
                {
                    LOG(error) << "create_directories(" << entry.path() << ") -- "
                               << errorCode.value() << " -- " << errorCode.message();
                }
            }

            LOG(debug) << "Created directory: " << entry.path();
        }
        else if (entry.is_regular_file())
        {
            // Regular file entry

            // Remove filename from file path
            auto fileDirPath = entry.path();
            fileDirPath = fileDirPath.remove_filename();
//            LOG(error) << "FILEDIRPATH: " << fileDirPath;

            // Create destination directory path (sans filename)
            auto destDirPath = sfs::path{r_options.getDestPath()} /
                                        sfs::relative(fileDirPath,
                                                 r_options.getSrcPath());
//            LOG(error) << "DIRPATH:    " << destDirPath;

            // Create destination directory path if it doesn't already exist
            if (! sfs::create_directories(destDirPath, errorCode))
            {
                // Error code 0 means the directory already existed
                if (errorCode.value() != 0)
                {
                    LOG(error) << "create_directories(" << entry.path() << ") -- "
                               << errorCode.value() << " -- " << errorCode.message();
                }
            }

            // Create destination file path including filename
            auto destFilePath = destDirPath / entry.path().filename();

            fromFd = open(entry.path().string().c_str(), O_RDONLY);
            if (fromFd == -1)
            {
                LOG(error) << "Unable to open: " << entry.path() << " -- "
                << errno << " -- " << std::strerror(errno);
                continue;
            }

            fstat(fromFd, &statBuf);
            // TODO: check return code

            toFd = open(destFilePath.string().c_str(),
                        O_WRONLY | O_CREAT, statBuf.st_mode);
            if (toFd == -1)
            {
                LOG(error) << "Unable to open: " << entry.path() << " -- "
                           << errno << " -- " << std::strerror(errno);
                continue;
            }

            // Determine sendfile() "count" parameter which must be no more than
            // MAX_COPY_SIZE per sendfile() documentation.
            ssize_t count = std::min(sfs::file_size(entry.path()), MAX_COPY_SIZE);

            // Iterate as needed calling sendfile() to copy data from the source file
            // to the destination file
            rc = 1;
            while (rc > 0)
            {
                rc = sendfile(toFd, fromFd, 0, count);
            }
            if (rc < 0)
            {
                LOG(error) << "Error while writing: " << destFilePath << " -- "
                           << errno << " -- " << std::strerror(errno);
                continue;
            }

	    close(fromFd);
	    close(toFd);
            LOG(debug) << "Copy: " << entry.path().string()
                       << " to: " << destFilePath.string();
        }
        else
        {
            // Unsupported entry type
            LOG(error) << "Unsupported entry type: " << entry.path();
        }

    }
}