#pragma once

#include "trygvis/sensor.h"

#include <mutex>
#include <boost/asio/buffer.hpp>
#include <utility>

namespace trygvis {
namespace sensor {
namespace io {

using namespace std;
using namespace boost::asio;

struct sample_output_stream_option {
    virtual ~sample_output_stream_option() = default;;
};

struct output_fields_option : sample_output_stream_option {
    ~output_fields_option() override = default;

    vector<string> fields;
};

struct timestamp_field_option : sample_output_stream_option {
    explicit timestamp_field_option(string name) : name(std::move(name)) {}

    ~timestamp_field_option() override = default;

    const string name;
};

class table_name_option : public sample_output_stream_option {
public:
    explicit table_name_option(string name) : name(std::move(name)) {}

    ~table_name_option() override = default;

    const string name;
};

class sample_output_stream_options : public vector<sample_output_stream_option *> {
public:
    ~sample_output_stream_options() = default;

    template<typename T>
    o<T *> find_option() const {
        for (auto it : *this) {
            auto *x = dynamic_cast<T *>(it);

            if (x != nullptr) {
                return o<T *>(x);
            }
        }

        return o<T *>();
    }
};

struct missing_required_option_error : runtime_error {
    explicit missing_required_option_error(const string &what) : runtime_error(what) {}

    ~missing_required_option_error() override = default;
};

class SampleStreamParser;

class SampleConsumer;

/**
 * Throws missing_required_option_error
 */
unique_ptr<SampleStreamParser> open_sample_stream_parser(
        shared_ptr<SampleConsumer> output,
        KeyDictionary &dict,
        sample_format_type type = sample_format_type::AUTO);

/**
 * Throws missing_required_option_error
 */
unique_ptr<SampleConsumer> open_sample_writer(
        shared_ptr<ostream> output,
        KeyDictionary &dict,
        sample_format_type type,
        sample_output_stream_options options);

static inline
unique_ptr<SampleConsumer> open_sample_output_stream(
        shared_ptr<ostream> output,
        KeyDictionary &dict,
        sample_format_type type) {
    sample_output_stream_options options;
    return open_sample_writer(std::move(output), dict, type, options);
}

class SampleConsumer {
public:
    virtual void onSample(SampleRecord const &sample) = 0;
};

class VectorSampleOutputStream : public SampleConsumer {

public:
    void onSample(SampleRecord const &sample) override;

    vector<SampleRecord> samples;
};

class ThreadSafeSampleConsumer : public SampleConsumer {
public:
    explicit ThreadSafeSampleConsumer(unique_ptr<SampleConsumer> underlying);

    ~ThreadSafeSampleConsumer() = default;

    void onSample(SampleRecord const &sample) override;

private:
    unique_ptr<SampleConsumer> underlying;
    std::mutex mutex;
};

class AddTimestampSampleConsumer : public SampleConsumer {
public:
    AddTimestampSampleConsumer(unique_ptr<SampleConsumer> underlying, KeyDictionary &dict, const string &timestamp_name);

    ~AddTimestampSampleConsumer() = default;

    void onSample(SampleRecord const &sample) override;

private:
    unique_ptr<SampleConsumer> underlying_;
    const SampleKey* timestamp_key;
};

class CsvWriterSampleConsumer : public SampleConsumer {
public:
    CsvWriterSampleConsumer(shared_ptr<ostream> stream, KeyDictionary &dict);

    void onSample(SampleRecord const &sample) override;

private:
    void writeHeader();

    KeyDictionary &dict;
    shared_ptr<ostream> stream;
    bool headerWritten;
};

class JsonWriterSampleConsumer : public SampleConsumer {
public:
    JsonWriterSampleConsumer(shared_ptr<ostream> stream, KeyDictionary &dict);

    void onSample(SampleRecord const &sample) override;

private:
    KeyDictionary &dict;
    shared_ptr<ostream> stream;
};

class KeyValueWriterSampleConsumer : public SampleConsumer {
public:
    KeyValueWriterSampleConsumer(shared_ptr<ostream> stream, KeyDictionary &dict);

    void onSample(SampleRecord const &sample) override;

private:
    KeyDictionary &dict;
    shared_ptr<ostream> stream;
};

class RrdWriterSampleConsumer : public SampleConsumer {
public:
    RrdWriterSampleConsumer(shared_ptr<ostream> stream, KeyDictionary &dict, const SampleKey *timestamp_key, o<output_fields_option *> output_fields);

    void onSample(SampleRecord const &sample) override;

private:
    vector<const SampleKey *> keys;
    shared_ptr<ostream> stream;
    const SampleKey *timestamp_key;
};

class SqlWriterSampleConsumer : public SampleConsumer {
public:
    SqlWriterSampleConsumer(shared_ptr<ostream> stream, KeyDictionary &dict, string table_name);

    void onSample(SampleRecord const &sample) override;

private:
    KeyDictionary &dict;
    shared_ptr<ostream> stream;
    const string table_name;
};

class SampleStreamParser {
public:
    virtual int process(mutable_buffers_1 &buffer) = 0;

    /**
     * Tells the parser that the data that has been written so far either "is it" (there is no more input) or that
     * it should just process what it has so far. It is ok to call process() after again after this method.
     *
     * TODO: should probably be renamed to "process()", "flush()" or "done()".
     */
    virtual int finish() = 0;

    virtual sample_format_type type() {
        return type_;
    }

    virtual KeyDictionary &dict() {
        return dict_;
    }

protected:
    sample_format_type type_;
    KeyDictionary &dict_;

    SampleStreamParser(const sample_format_type type, KeyDictionary &dict) : type_(type), dict_(dict) {
    }
};

class KeyValueSampleStreamParser : public SampleStreamParser {

public:
    KeyValueSampleStreamParser(shared_ptr<SampleConsumer> output, KeyDictionary &dict) :
            SampleStreamParser(sample_format_type::CSV, dict), output(std::move(output)),
            line(make_shared<vector<uint8_t>>()) {
    }

    int process(mutable_buffers_1 &buffer) override;

    int finish() override;

private:
    void process_line(shared_ptr<vector<uint8_t>> &packet);

    static const uint8_t packet_delimiter = '\n';
    shared_ptr<SampleConsumer> output;
    shared_ptr<vector<uint8_t>> line;
};

class AutoSampleParser : public SampleStreamParser {
public:
    AutoSampleParser(shared_ptr<SampleConsumer> output, KeyDictionary &dict);

    int process(mutable_buffers_1 &buffer) override;

    int finish() override;

private:
    unique_ptr<SampleStreamParser> parser;
    unique_ptr<KeyValueSampleStreamParser> keyValueParser;
};

static inline
unique_ptr<ThreadSafeSampleConsumer> thread_safe_sample_output_stream(unique_ptr<SampleConsumer> underlying) {
    return make_unique<ThreadSafeSampleConsumer>(move(underlying));
};

} // namespace io
} // namespace sensor
} // namespace trygvis