/*
 * © Copyright 1996-2012 ECMWF.
 *
 * This software is licensed under the terms of the Apache Licence Version 2.0
 * which can be obtained at http://www.apache.org/licenses/LICENSE-2.0.
 * In applying this licence, ECMWF does not waive the privileges and immunities
 * granted to it by virtue of its status as an intergovernmental organisation nor
 * does it submit to any jurisdiction.
 */

#include "odb.h"
#include "mars.h"

#include "eckit/eckit_version.h"
#include "eckit/exception/Exceptions.h"
#include "eckit/filesystem/PathName.h"
#include "eckit/io/Buffer.h"
#include "eckit/io/Length.h"
#include "eckit/io/SeekableHandle.h"
#include "eckit/utils/StringTools.h"

#include "metkit/odb/OdbToRequest.h"

#include "odc/api/Odb.h"

#include <algorithm>
#include <memory>
#include <set>
#include <string>

using namespace eckit;
using namespace metkit::odb;


boolean is_part_of_mars_language(const char* n) {
    static request* archive = NULL;
    const char* s           = NULL;

    if (!archive) {
        request* r = mars_language();
        while (r && !EQ(r->name, "ARCHIVE"))
            r = r->next;
        if (r == NULL) {
            marslog(LOG_EROR, const_cast<char*>("ARCHIVE request not found in language. OOOPPPSSS!!!"));
            marsexit(1);
        }

        archive = r;
    }

    if ((s = get_value(archive, n, 0)) != NULL)
        return true;

    return false;
}

err odb_to_request_from_file(request* r, const char* fileName) {
    try {
        marslog(LOG_DBUG, const_cast<char*>("odb_to_request_from_file: fileName = %s."), fileName);

        PathName pn(fileName);
        if (!pn.exists()) {
            marslog(LOG_EROR, (char*)"oda_to_request_from_file: file '%s' does not exist.", (char*)fileName);
            return TOO_SHORT_ERR;
        }
        if (!pn.size()) {
            marslog(LOG_EROR, (char*)"oda_to_request_from_file: file '%s' empty.", fileName);
            return TOO_SHORT_ERR;
        }

        // Extract the MARS keywords needed to archive this data

        bool singleRequest   = true;
        bool constantColumns = true;
        OdbToRequest o2r("archive", singleRequest, constantColumns);

        metkit::mars::MarsRequest rq;
        try {
            std::unique_ptr<DataHandle> dh(pn.fileHandle());
            dh->openForRead();
            std::vector<metkit::mars::MarsRequest> requests = o2r.odbToRequest(*dh);
            ASSERT(requests.size() == 1);
            rq = requests[0];
        }
        catch (Exception& ex) {
            return HYPERCUBE_ERROR;
        }

        // Convert the metkit mars rq to one for MARS client

        std::ostringstream ss_rq;
        rq.dump(ss_rq, "", "");
        Log::debug() << "odb_to_request_from_file: " << ss_rq.str() << std::endl;

        request* n = string2request(ss_rq.str().c_str());
        if (!n) {
            marslog(LOG_EROR, "Error creating a MARS request from data");
            return TOO_SHORT_ERR;
        }
        reqmerge(r, n);
        free_all_requests(n);
    }
    catch (Exception& ex) {
        marslog(LOG_EROR, (char*)"Error making a request from file %s\n", ex.what());
        marslog(LOG_EROR, (char*)"Exception ignored");
        return TOO_SHORT_ERR;
    }
    return NOERR;
}

typedef std::map<std::string, std::set<std::string> > Parameters;

Parameters parameters(request* r) {
    Parameters ret;

    for (parameter* p = r->params; p; p = p->next) {
        std::string parameterName = p->name;
        size_t n                  = count_values(r, p->name);
        if (n == 0) {
            marslog(LOG_EROR, (char*)"parameters: no values of param '%s'", p->name);
            ASSERT(n != 0);
        }
        std::set<std::string> values;
        for (size_t i = 0; i < n; ++i)
            values.insert(get_value(r, p->name, i));
        ret[parameterName] = values;
    }
    return ret;
}

bool set_values_compare(const std::set<std::string>& s1, const std::set<std::string>& s2) {

    if (s1.size() != s2.size())
        return false;

    for (const auto& v : s1) {
        if (s2.find(v) != s2.end())
            continue;
        if (s2.find(StringTools::upper(v)) != s2.end())
            continue;
        if (s2.find(StringTools::lower(v)) != s2.end())
            continue;
        return false;
    }

    return true;
}

err odb_compare_attributes_of_first_request(request* first, request* second) {
    typedef Parameters P;

    P firstParams  = parameters(first);
    P secondParams = parameters(second);

    for (P::iterator it = firstParams.begin(); it != firstParams.end(); ++it) {
        const std::string& paramName        = it->first;
        const std::set<std::string>& values = it->second;

        P::iterator jt = secondParams.find(paramName);
        if (jt == secondParams.end())
            jt = secondParams.find(StringTools::upper(paramName));
        if (jt == secondParams.end())
            jt = secondParams.find(StringTools::lower(paramName));
        if (jt == secondParams.end()) {
            marslog(LOG_EROR, (char*)"odb_compare_attributes_of_first_request: second request has no param '%s'", paramName.c_str());
            return -1;
        }
        const std::set<std::string>& otherValues = jt->second;
        if (!set_values_compare(values, otherValues)) {
            std::stringstream ss;
            if (values.size() == 1 && otherValues.size() == 1) {
                ss << "Values of '" << paramName << "' differ: " << *values.begin() << " <> " << *otherValues.begin();
                marslog(LOG_EROR, (char*)"odb_compare_attributes_of_first_request: %s", ss.str().c_str());
            }
            else {
                marslog(LOG_EROR, (char*)"odb_compare_attributes_of_first_request: values for param '%s' differ", paramName.c_str());
                std::ostream_iterator<std::string> out(ss, ", ");
                std::set_symmetric_difference(values.begin(), values.end(), otherValues.begin(), otherValues.end(), out);
                marslog(LOG_EROR, (char*)"odb_compare_attributes_of_first_request: values present in one of the sets, but not in the other: %s",
                        ss.str().c_str());
            }
            return -1;
        }
    }
    return NOERR;
}

/// A simple wrapper class to make an existing FILE* look like a DataHandle

class CloneCopyStdFileHandle : public DataHandle {
public:
    CloneCopyStdFileHandle(FILE* f) :
        f_(f) {}

    DataHandle* clone() const override {
        return new CloneCopyStdFileHandle(f_);
    }

    void print(std::ostream& s) const override {
        s << "CloneCopyStdFileHandle()";
    }

    Length openForRead() override { return 0; }
    void openForWrite(const Length&) override {}
    void openForAppend(const Length&) override {}
    void close() override{};

    long read(void* out, long length) override {
        return ::fread(out, 1, length, f_);
    }

    long write(const void* buffer, long length) override {
        return ::fwrite(buffer, 1, length, f_);
    }

    Offset seek(const Offset& offset) override {
        return ::fseek(f_, offset, SEEK_SET);
    }

private:
    FILE* f_;
};


class FilteredRestrictedODBHandle : public DataHandle, public HandleHolder {

public:  // methods
    FilteredRestrictedODBHandle(DataHandle* dh) :
        HandleHolder(dh),
        frame_(),
        frameData_(0),
        pos_(0) {}

    ~FilteredRestrictedODBHandle() override {}

    Length openForRead() override {
        return handle().openForRead();
    }

    void print(std::ostream& s) const override {
        s << "FilteredRestrictedODBHandle(" << handle() << ")";
    }
    void close() override { handle().close(); }
    bool canSeek() const override { return false; }
    DataHandle* clone() const override {
        return new FilteredRestrictedODBHandle(handle().clone());
    }

    void getNextFrame() {

        while (true) {

            // We have these inside the loop so that we don't end up storing lots of data in
            // memory if we have a long sequence of restricted (i.e. skipped) frames
            //
            // n.b. if we stored the frameData_ in the object rather than the frame object
            //      we could have these on the stack, rather than belonging to the Reader
            //      (and thus having ownership shared with the Frame).

            PeekHandle* peek_in         = new PeekHandle(handle());
            SeekableHandle* seekable_in = new SeekableHandle(peek_in);
            seekable_in->openForRead();

            odc::api::Reader reader(seekable_in, /* aggregated */ false);
            frame_ = reader.next();

            if (!frame_)
                break;

            // Do we have a column called restricted? Otherwise assume unrestricted

            if (frame_.hasColumn("restricted")) {

                bool onlyConstantColumns        = false;
                odc::api::Span restrictedSpan   = frame_.span({"restricted"}, onlyConstantColumns);
                std::set<long> restrictedValues = restrictedSpan.getIntegerValues("restricted");

                if (restrictedValues.size() == 1) {
                    if (restrictedValues.find(0) != restrictedValues.end()) {
                        // Unrestricted data;
                        break;
                    }
                    else {
                        // Restricted data. Discard.
                        // Ensure entire frame has been read --> next Seekable handle will work.
                        Buffer discard = frame_.encodedData();
                        continue;
                    }
                }
                else {
                    // A mix of restricted/unrestricted data
                    frame_ = frame_.filter("SELECT * WHERE restricted = 0 OR restricted IS NULL");
                    break;
                }
            }
            else {
                // Not restricted, so we can use this data
                break;
            }
        }
    }

    long read(void* out, long length) override {

        if (pos_ == Length(frameData_.size())) {
            pos_ = 0;
            getNextFrame();

            if (frame_) {
                frameData_ = frame_.encodedData();
            }
            else {
                frameData_ = Buffer(0);
                return 0;
            }
        }

        ASSERT(pos_ < Length(frameData_.size()));
        long readlength = std::min<long>(length, (frameData_.size() - pos_));
        ::memcpy(out, &frameData_[pos_], readlength);
        pos_ += readlength;
        return readlength;
    }

private:  // members
    odc::api::Frame frame_;
    Buffer frameData_;
    Offset pos_;
};


long long odb_filter(const char* sql, FILE* fin, FILE* fout, long long total_to_read) {
    try {

        marslog(LOG_INFO, const_cast<char*>("odb_filter: sql = '%s', total_to_read = %lld"),
                ((sql == 0) ? "NULL" : sql),
                total_to_read);

        if (total_to_read == 0)
            return 0;


        // Loop over each of the frame individually
        // --> We consider the restricted nature of any observations before we apply
        //     any user-specified filters


        std::unique_ptr<DataHandle> input_dh(new CloneCopyStdFileHandle(fin));

        if (!mars.privileged) {
            input_dh.reset(new FilteredRestrictedODBHandle(input_dh.release()));
        }

        // Is an SQL filter specified (just do this once)

        std::string sqlstring;
        if (sql && strlen(sql) != 0) {
            sqlstring = StringTools::unQuote(StringTools::trim(sql));
        }

        // If non-null apply SQL filter, and save into output DH

        CloneCopyStdFileHandle output_dh(fout);
        odc::api::filter(sqlstring, *input_dh, output_dh);

        marslog(LOG_DBUG, const_cast<char*>(" => odb_filter"));
    }
    catch (Exception& ex) {
        marslog(LOG_EROR, (char*)"Error in odb_filter %s\n", ex.what());
        return -1;
    }

    // TODO: make sure the below is true
    return total_to_read;
}