/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * License); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

#include <fcntl.h>

#include <algorithm>
#include <cerrno>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <ctime>
#include <fstream>
#include <iostream>
#include <string>
#include <unordered_map>
#include <utility>
#include <vector>

#include "cli/cli_args.h"
#include "cli/exit_codes.h"
#include "commands/commands.h"
#include "common/datatype/date_converter.h"
#include "common/schema.h"
#include "common/tablet.h"
#include "file/write_file.h"
#include "format/input_format.h"
#include "writer/tsfile_table_writer.h"

namespace tsfile_cli {
namespace {

struct DataRow {
    long long line_no;
    int64_t timestamp;
    std::vector<std::string> cells;
};

// Parse a calendar date in strict YYYY-MM-DD form into a std::tm (year offset
// from 1900, month 0-based) the way storage::Tablet expects for DATE columns.
// Validates that it is a real date (DateConverter rejects e.g. 2024-13-40),
// since the writer silently drops an invalid std::tm rather than erroring.
bool parse_date_cell(const std::string& cell, std::tm& out) {
    int y = 0;
    int m = 0;
    int d = 0;
    char extra = 0;
    if (std::sscanf(cell.c_str(), "%4d-%2d-%2d%c", &y, &m, &d, &extra) != 3) {
        return false;
    }
    out = std::tm();
    out.tm_year = y - 1900;
    out.tm_mon = m - 1;
    out.tm_mday = d;
    int32_t date_int = 0;
    return common::DateConverter::date_to_int(out, date_int) == common::E_OK;
}

bool add_typed_value(storage::Tablet& tablet, uint32_t row,
                     uint32_t schema_index, const ColumnDef& def,
                     const std::string& cell, std::string& error) {
    if (cell.empty()) {
        return true;  // null
    }
    char* e = nullptr;
    switch (def.type) {
        case common::BOOLEAN: {
            bool v = false;
            if (!parse_bool_cell(cell, v)) {
                error = "bad BOOLEAN '" + cell + "'";
                return false;
            }
            tablet.add_value(row, schema_index, v);
            return true;
        }
        case common::INT32: {
            errno = 0;
            long long v = std::strtoll(cell.c_str(), &e, 10);
            if (e == nullptr || *e != '\0') {
                error = "bad INT32 '" + cell + "'";
                return false;
            }
            if (errno == ERANGE || v < INT32_MIN || v > INT32_MAX) {
                error = "INT32 out of range '" + cell + "'";
                return false;
            }
            tablet.add_value(row, schema_index, static_cast<int32_t>(v));
            return true;
        }
        case common::INT64: {
            errno = 0;
            long long v = std::strtoll(cell.c_str(), &e, 10);
            if (e == nullptr || *e != '\0') {
                error = "bad INT64 '" + cell + "'";
                return false;
            }
            if (errno == ERANGE) {
                error = "INT64 out of range '" + cell + "'";
                return false;
            }
            tablet.add_value(row, schema_index, static_cast<int64_t>(v));
            return true;
        }
        case common::TIMESTAMP: {
            errno = 0;
            long long v = std::strtoll(cell.c_str(), &e, 10);
            if (e == nullptr || *e != '\0') {
                error = "bad TIMESTAMP '" + cell + "'";
                return false;
            }
            if (errno == ERANGE) {
                error = "TIMESTAMP out of range '" + cell + "'";
                return false;
            }
            tablet.add_value(row, schema_index, static_cast<int64_t>(v));
            return true;
        }
        case common::DATE: {
            std::tm d;
            if (!parse_date_cell(cell, d)) {
                error = "bad DATE '" + cell + "' (want YYYY-MM-DD)";
                return false;
            }
            tablet.add_value(row, schema_index, d);
            return true;
        }
        case common::FLOAT: {
            errno = 0;
            float v = std::strtof(cell.c_str(), &e);
            if (e == nullptr || *e != '\0') {
                error = "bad FLOAT '" + cell + "'";
                return false;
            }
            if (errno == ERANGE) {
                error = "FLOAT out of range '" + cell + "'";
                return false;
            }
            tablet.add_value(row, schema_index, v);
            return true;
        }
        case common::DOUBLE: {
            errno = 0;
            double v = std::strtod(cell.c_str(), &e);
            if (e == nullptr || *e != '\0') {
                error = "bad DOUBLE '" + cell + "'";
                return false;
            }
            if (errno == ERANGE) {
                error = "DOUBLE out of range '" + cell + "'";
                return false;
            }
            tablet.add_value(row, schema_index, v);
            return true;
        }
        case common::STRING:
        case common::TEXT:
        case common::BLOB: {
            // Add by index using the c-string overload to avoid the per-cell
            // name lowercasing + map lookup the by-name overload would do.
            tablet.add_value(row, schema_index, cell.c_str());
            return true;
        }
        default:
            error = "unsupported column type";
            return false;
    }
}

}  // namespace

int cmd_write(const ParsedArgs& args, std::ostream& /*out*/,
              std::ostream& err) {
    std::vector<ColumnDef> columns;
    std::string perr;
    if (!parse_columns_spec(args.columns, columns, perr)) {
        err << "Error: " << perr << "\n";
        return kExitUsage;
    }

    std::istream* in = &std::cin;
    std::ifstream fin;
    if (!args.file.empty() && args.file != "-") {
        fin.open(args.file.c_str());
        if (!fin.is_open()) {
            err << "Error: cannot open input: " << args.file << "\n";
            return kExitFile;
        }
        in = &fin;
    }

    const char delim = (args.format == ParsedArgs::Format::kTsv) ? '\t' : ',';
    const bool csv_quotes = (delim == ',');

    if (args.verbose) {
        // Echo the resolved write configuration so the user can self-check the
        // parsed flags before any rows are imported.
        err << "write: table=" << args.table << " output=" << args.output
            << " input=" << (args.file.empty() ? "-" : args.file)
            << " format=" << (delim == '\t' ? "tsv" : "csv")
            << " header=" << (args.no_header ? "none" : "yes")
            << (args.header_match ? " (match)" : "") << "\n";
        for (const ColumnDef& d : columns) {
            err << "  column " << d.name << ":" << tsdatatype_name(d.type) << ":"
                << (d.category == common::ColumnCategory::TAG ? "tag" : "field")
                << "\n";
        }
    }

    std::string line;
    long long line_no = 0;
    long long record_lines = 0;
    if (!args.no_header) {
        if (read_record(*in, csv_quotes, line, record_lines)) {
            line_no += record_lines;
            if (args.header_match) {
                std::vector<std::string> h =
                    split_line(line, delim, csv_quotes);
                const size_t expected = columns.size() + 1;
                if (h.size() != expected) {
                    err << "Error: header has " << h.size()
                        << " columns, expected " << expected
                        << " (time + --columns) (line 1)\n";
                    return kExitRuntime;
                }
                for (size_t i = 0; i < columns.size(); ++i) {
                    if (h[i + 1] != columns[i].name) {
                        err << "Error: header column " << (i + 2) << " is '"
                            << h[i + 1] << "', expected '" << columns[i].name
                            << "' (line 1)\n";
                        return kExitRuntime;
                    }
                }
            }
        }
    }

    std::vector<std::string> names;
    std::vector<common::TSDataType> types;
    std::vector<common::ColumnCategory> cats;
    std::vector<common::ColumnSchema> col_schemas;
    std::vector<size_t> tag_idx;
    for (size_t j = 0; j < columns.size(); ++j) {
        const ColumnDef& d = columns[j];
        names.push_back(d.name);
        types.push_back(d.type);
        cats.push_back(d.category);
        col_schemas.push_back(common::ColumnSchema(
            d.name, d.type, common::get_default_compressor(),
            common::get_value_encoder(d.type), d.category));
        if (d.category == common::ColumnCategory::TAG) {
            tag_idx.push_back(j);
        }
    }

    // Creating the output truncates it; refuse to clobber the input we are
    // still reading from, which would otherwise silently destroy the source
    // data.
    if (!args.file.empty() && args.file != "-" && args.output == args.file) {
        err << "Error: --output is the same as the input file: " << args.output
            << "\n";
        return kExitUsage;
    }

    storage::WriteFile file;
    int flags = O_WRONLY | O_CREAT | O_TRUNC;
#ifdef _WIN32
    flags |= O_BINARY;
#endif
    int cret = file.create(args.output, flags, 0666);
    if (cret != 0) {
        err << "Error: cannot create output " << args.output << ": "
            << error_code_message(cret) << " (code " << cret << ")\n";
        return kExitFile;
    }
    auto* schema = new storage::TableSchema(args.table, col_schemas);
    auto* writer = new storage::TsFileTableWriter(&file, schema);

    // Stream rows into fixed-size batches so memory stays bounded regardless of
    // input size; a full file is never buffered in memory.
    const size_t kBatch = 1024;
    int result_code = kExitOk;
    long long total_rows = 0;
    std::vector<DataRow> batch;
    batch.reserve(kBatch);
    // The table writer requires strictly increasing timestamps per device, and
    // a device is identified by its tag-column values. Track the last timestamp
    // seen for each device so out-of-order input is rejected with a clear,
    // located message instead of an opaque write failure.
    std::unordered_map<std::string, int64_t> last_ts_by_device;

    auto flush_batch = [&]() -> bool {
        if (batch.empty()) {
            return true;
        }
        storage::Tablet tablet(args.table, names, types, cats,
                               static_cast<int>(batch.size()));
        for (size_t i = 0; i < batch.size(); ++i) {
            uint32_t r = static_cast<uint32_t>(i);
            tablet.add_timestamp(r, batch[i].timestamp);
            for (size_t j = 0; j < columns.size(); ++j) {
                std::string cell_err;
                if (!add_typed_value(tablet, r, static_cast<uint32_t>(j),
                                     columns[j], batch[i].cells[j], cell_err)) {
                    err << "Error: " << cell_err << " (line "
                        << batch[i].line_no << ")\n";
                    return false;
                }
            }
        }
        int wt = writer->write_table(tablet);
        if (wt != 0) {
            err << "Error: failed to write rows: " << error_code_message(wt)
                << " (code " << wt << ")\n";
            return false;
        }
        total_rows += static_cast<long long>(batch.size());
        batch.clear();
        return true;
    };

    while (read_record(*in, csv_quotes, line, record_lines)) {
        line_no += record_lines;
        if (line.empty()) {
            continue;
        }
        std::vector<std::string> fields = split_line(line, delim, csv_quotes);
        if (fields.size() != columns.size() + 1) {
            err << "Error: expected " << (columns.size() + 1) << " fields, got "
                << fields.size() << " (line " << line_no << ")\n";
            result_code = kExitRuntime;
            break;
        }
        char* e = nullptr;
        errno = 0;
        long long ts = std::strtoll(fields[0].c_str(), &e, 10);
        if (e == nullptr || *e != '\0' || errno == ERANGE) {
            err << "Error: bad timestamp '" << fields[0] << "' (line "
                << line_no << ")\n";
            result_code = kExitRuntime;
            break;
        }
        DataRow r;
        r.line_no = line_no;
        r.timestamp = static_cast<int64_t>(ts);
        r.cells.assign(fields.begin() + 1, fields.end());

        std::string device_key;
        for (size_t k : tag_idx) {
            device_key += r.cells[k];
            device_key.push_back('\0');
        }
        auto seen = last_ts_by_device.find(device_key);
        if (seen != last_ts_by_device.end() && r.timestamp <= seen->second) {
            err << "Error: timestamps must be strictly increasing per device "
                   "(line "
                << line_no << ": " << r.timestamp << " <= previous "
                << seen->second << ")\n";
            result_code = kExitRuntime;
            break;
        }
        last_ts_by_device[device_key] = r.timestamp;

        batch.push_back(std::move(r));
        if (batch.size() >= kBatch && !flush_batch()) {
            result_code = kExitRuntime;
            break;
        }
    }

    if (result_code == kExitOk && !flush_batch()) {
        result_code = kExitRuntime;
    }

    if (result_code == kExitOk) {
        int fr = writer->flush();
        if (fr != 0) {
            err << "Error: failed to flush output: " << error_code_message(fr)
                << " (code " << fr << ")\n";
            result_code = kExitRuntime;
        } else {
            int cr = writer->close();
            if (cr != 0) {
                err << "Error: failed to close output: "
                    << error_code_message(cr) << " (code " << cr << ")\n";
                result_code = kExitRuntime;
            }
        }
    } else {
        writer->close();
    }
    delete writer;
    delete schema;

    if (result_code != kExitOk) {
        // The import failed; do not leave a partial/corrupt .tsfile behind.
        file.close();
        std::remove(args.output.c_str());
    } else if (args.verbose) {
        err << "wrote " << total_rows << " rows to " << args.output << "\n";
    }
    return result_code;
}

}  // namespace tsfile_cli