#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <time.h>
#include <locale.h>

#include "Reduce.h"
#include "Errors.h"

#define MAX_PATH_LEN 256

static char pgm_help[] =
    "   Sintax: \tWordCountMR_v2a [-r <nreducers=1>] [-o <output_file>] <input_files>\n"
    "   Example: \tWordCountMR_v2a -r 1 -o ./Result/wc.txt ./Input/README.txt ./Input/LEEME.txt\n";

static void get_exec_dir(const char *argv0, char *dir, size_t size)
{
    const char *last_slash = strrchr(argv0, '/');
    size_t len;

    if (last_slash == NULL) {
        snprintf(dir, size, ".");
        return;
    }

    len = (size_t)(last_slash - argv0);
    if (len >= size) {
        len = size - 1;
    }

    memcpy(dir, argv0, len);
    dir[len] = '\0';
}

static char *build_exec_path(const char *dir, const char *program)
{
    size_t len_dir = strlen(dir);
    size_t len_program = strlen(program);
    char *path = malloc(len_dir + 1 + len_program + 1);

    if (path == NULL) {
        Error("[WordCountMR_v2a] malloc failed");
    }

    memcpy(path, dir, len_dir);
    path[len_dir] = '/';
    memcpy(path + len_dir + 1, program, len_program);
    path[len_dir + 1 + len_program] = '\0';

    return path;
}

static int run_child(char *const child_argv[], const char *stage)
{
    pid_t pid;
    int status;

    pid = fork();

    if (pid < 0) {
        sprintf(MsgError, "[WordCountMR_v2a] Error creating %s process", stage);
        Error(MsgError);
    }

    if (pid == 0) {
        execv(child_argv[0], child_argv);

        sprintf(MsgError, "[WordCountMR_v2a] Error executing %s", stage);
        Error(MsgError);
    }

    sprintf(Msg, "[WordCountMR_v2a] Started %s process %d.\n", stage, pid);
    printMessage(Msg, COLOR_GREEN_L);

    if (waitpid(pid, &status, 0) < 0) {
        sprintf(MsgError, "[WordCountMR_v2a] Error waiting for %s process", stage);
        Error(MsgError);
    }

    if (!WIFEXITED(status)) {
        sprintf(MsgError, "[WordCountMR_v2a] %s process did not finish normally", stage);
        Error(MsgError);
    }

    /*
     * WordCountMap devuelve el número de palabras procesadas mediante el
     * código de salida. Por eso no se trata como error que el valor sea
     * distinto de cero. Este detalle se analiza después en la pregunta 4.1.
     */
    if (strcmp(stage, "WordCountMap") != 0 && WEXITSTATUS(status) != 0) {
        sprintf(MsgError, "[WordCountMR_v2a] %s process failed", stage);
        Error(MsgError);
    }

    return WEXITSTATUS(status);
}

int main(int argc, char *argv[])
{
    int par;
    int nreducers = 1;
    int ninputfiles;
    int words;
    char *output_file = NULL;

    char **map_argv;
    char *suffle_argv[3];
    char *reduce_argv[4];

    char reducers_str[16];

    char exec_dir[MAX_PATH_LEN];
    char *map_exec;
    char *suffle_exec;
    char *reduce_exec;

    struct timespec start, finish;
    float time;

    if (argc < 2) {
        sprintf(MsgError,
                "Error number of arguments (%d arguments passed, at least 1 required).\n%s\n",
                argc - 1, pgm_help);
        printMessage(MsgError, COLOR_RED);
        exit(1);
    }

    for (par = 1; par < argc && argv[par][0] == '-'; par++) {
        switch (argv[par][1]) {
            case 'r':
                if (par + 1 >= argc) {
                    sprintf(MsgError, "Error missing value for -r.\n%s", pgm_help);
                    Error(MsgError);
                }
                nreducers = atoi(argv[++par]);
                break;

            case 'o':
                if (par + 1 >= argc) {
                    sprintf(MsgError, "Error missing value for -o.\n%s", pgm_help);
                    Error(MsgError);
                }
                output_file = argv[++par];
                break;

            default:
                sprintf(MsgError,
                        "Error optional argument %d is incorrect (%s).\n%s",
                        par, argv[par], pgm_help);
                Error(MsgError);
        }
    }

    if (output_file == NULL) {
        output_file = DReduceOutputFileName;
    }

    if (nreducers < 1 || nreducers > 16) {
        Error("[WordCountMR_v2a] Invalid number of reducers");
    }

    ninputfiles = argc - par;

    if (ninputfiles < 1) {
        Error("[WordCountMR_v2a] No input files provided");
    }

    setlocale(LC_ALL, "");
    clock_gettime(CLOCK_MONOTONIC, &start);

    get_exec_dir(argv[0], exec_dir, sizeof(exec_dir));

    map_exec = build_exec_path(exec_dir, "WordCountMap");
    suffle_exec = build_exec_path(exec_dir, "WordCountSuffle");
    reduce_exec = build_exec_path(exec_dir, "WordCountReduce");

    map_argv = malloc(sizeof(char *) * (ninputfiles + 2));

    if (map_argv == NULL) {
        Error("[WordCountMR_v2a] malloc failed");
    }

    map_argv[0] = map_exec;

    for (int i = 0; i < ninputfiles; i++) {
        map_argv[i + 1] = argv[par + i];
    }

    map_argv[ninputfiles + 1] = NULL;

    snprintf(reducers_str, sizeof(reducers_str), "%d", nreducers);

    suffle_argv[0] = suffle_exec;
    suffle_argv[1] = reducers_str;
    suffle_argv[2] = NULL;

    reduce_argv[0] = reduce_exec;
    reduce_argv[1] = output_file;
    reduce_argv[2] = reducers_str;
    reduce_argv[3] = NULL;

    words = run_child(map_argv, "WordCountMap");
    run_child(suffle_argv, "WordCountSuffle");
    run_child(reduce_argv, "WordCountReduce");

    free(map_argv);
    free(map_exec);
    free(suffle_exec);
    free(reduce_exec);

    clock_gettime(CLOCK_MONOTONIC, &finish);

    time = (finish.tv_sec - start.tv_sec) +
           ((finish.tv_nsec - start.tv_nsec) / 1000000000.0);

    sprintf(Msg,
            "[WordCountMR_v2a] WordCount completed processing %d files and %d words in %.3f secs.\n",
            ninputfiles, words, time);
    printMessage(Msg, COLOR_GREEN_D);

    return 0;
}
