#include <unistd.h>
#include <fcntl.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <locale.h>
#include <time.h>

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

static char pgm_help[] =
    "   Sintax: \tWordCountReduce <output_file> <nreducers> <reducer_id>\n"
    "   Example: \tWordCountReduce ./Result/wc.txt 2 0\n";

static char *build_suffle_input_name(int reducer_id)
{
    char id[32];
    size_t size;
    char *path;

    snprintf(id, sizeof(id), "%d", reducer_id);

    /*
     * Los ficheros generados por Suffle tienen el formato:
     * tmp/_suffle_0.tmp, tmp/_suffle_1.tmp, ...
     */
    size = strlen(DAuxiliarDiretory)
         + strlen(DSuffleFilePrefix)
         + 1
         + strlen(id)
         + strlen(DAuxiliarFilesExtension)
         + 1;

    path = malloc(size);

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

    sprintf(path, "%s%s_%s%s",
            DAuxiliarDiretory,
            DSuffleFilePrefix,
            id,
            DAuxiliarFilesExtension);

    return path;
}

static char *build_reduce_output_name(const char *output_file,
                                      int nreducers,
                                      int reducer_id)
{
    char id[32];
    const char *dot;
    size_t base_len;
    size_t ext_len;
    size_t size;
    char *path;

    if (nreducers == 1) {
        path = malloc(strlen(output_file) + 1);

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

        strcpy(path, output_file);
        return path;
    }

    snprintf(id, sizeof(id), "%d", reducer_id + 1);

    dot = strrchr(output_file, '.');

    if (dot != NULL) {
        base_len = (size_t)(dot - output_file);
        ext_len = strlen(dot);

        size = base_len + 1 + strlen(id) + ext_len + 1;
        path = malloc(size);

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

        memcpy(path, output_file, base_len);
        path[base_len] = '\0';

        strcat(path, "_");
        strcat(path, id);
        strcat(path, dot);
    } else {
        size = strlen(output_file)
             + 1
             + strlen(id)
             + strlen(DResultFileExtension)
             + 1;

        path = malloc(size);

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

        sprintf(path, "%s_%s%s", output_file, id, DResultFileExtension);
    }

    return path;
}

int main(int argc, char *argv[])
{
    char *output_file;
    int nreducers;
    int reducer_id;
    char *input_file;
    char *final_output_file;
    int fd_in;
    int fd_out;
    int words;
    struct timespec start, finish;
    float time;

    if (argc != 4) {
        printMessage("Error number of arguments.\n", COLOR_RED);
        printMessage(pgm_help, COLOR_RED);
        exit(1);
    }

    output_file = argv[1];
    nreducers = atoi(argv[2]);
    reducer_id = atoi(argv[3]);

    if (nreducers < 1 || nreducers > DMAX_REDUCERS ||
        reducer_id < 0 || reducer_id >= nreducers) {
        Error("[WordCountReduce] Invalid reducer parameters");
    }

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

    input_file = build_suffle_input_name(reducer_id);
    final_output_file = build_reduce_output_name(output_file, nreducers, reducer_id);

    if (VERBOSE) {
        sprintf(Msg, "[Reduce Stage]\tStart Reducer %d to process %s file.\n",
                reducer_id + 1, input_file);
        print(Msg);
    }

    fd_in = open(input_file, O_RDONLY);

    if (fd_in < 0) {
        free(input_file);
        free(final_output_file);
        Error("[WordCountReduce] Error opening reducer input file");
    }

    fd_out = open(final_output_file, O_WRONLY | O_CREAT | O_TRUNC, 0644);

    if (fd_out < 0) {
        close(fd_in);
        free(input_file);
        free(final_output_file);
        Error("[WordCountReduce] Error opening reducer output file");
    }

    words = reduce_process_file(fd_in, fd_out);

    close(fd_in);
    close(fd_out);

    if (VERBOSE) {
        clock_gettime(CLOCK_MONOTONIC, &finish);
        time = (finish.tv_sec - start.tv_sec) +
               ((finish.tv_nsec - start.tv_nsec) / 1000000000.0);

        sprintf(Msg,
                "[Reduce Stage]\tReducer %d Done. Processed %d different words in %.3f secs.\n",
                reducer_id + 1, words, time);
        print(Msg);
    }

    free(input_file);
    free(final_output_file);

    return 0;
}
