/*
 * 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 <algorithm>
#include <climits>
#include <limits>
#include <memory>
#include <string>
#include <vector>

#include "cli/exit_codes.h"
#include "commands/commands.h"
#include "common/device_id.h"
#include "common/schema.h"
#include "format/result_set_format.h"
#include "reader/filter/tag_filter.h"
#include "reader/tsfile_reader.h"

namespace tsfile_cli {
namespace {

bool can_push_down_row_window(const ParsedArgs& args, long long offset,
                              long long limit) {
    return !args.has_start && !args.has_end && offset <= INT_MAX &&
           (limit < 0 || limit <= INT_MAX);
}

int to_reader_row_bound(long long value) {
    return value < 0 ? -1 : static_cast<int>(value);
}

}  // namespace

std::unique_ptr<storage::Filter> build_table_tag_filter(
    const ParsedArgs& args, storage::TsFileReader& reader,
    const std::string& table_name, std::ostream& err) {
    if (!args.has_tag_filter) {
        return std::unique_ptr<storage::Filter>();
    }
    auto schema = reader.get_table_schema(table_name);
    if (!schema) {
        err << "Error: no schema found for table " << table_name << "\n";
        return std::unique_ptr<storage::Filter>();
    }

    storage::TagFilterBuilder builder(schema.get());
    storage::Filter* filter = nullptr;
    switch (args.tag_filter_op) {
        case ParsedArgs::TagFilterOp::kEq:
            filter = builder.eq(args.tag_filter_column, args.tag_filter_value);
            break;
        case ParsedArgs::TagFilterOp::kNeq:
            filter = builder.neq(args.tag_filter_column, args.tag_filter_value);
            break;
        case ParsedArgs::TagFilterOp::kLt:
            filter = builder.lt(args.tag_filter_column, args.tag_filter_value);
            break;
        case ParsedArgs::TagFilterOp::kLteq:
            filter =
                builder.lteq(args.tag_filter_column, args.tag_filter_value);
            break;
        case ParsedArgs::TagFilterOp::kGt:
            filter = builder.gt(args.tag_filter_column, args.tag_filter_value);
            break;
        case ParsedArgs::TagFilterOp::kGteq:
            filter =
                builder.gteq(args.tag_filter_column, args.tag_filter_value);
            break;
        case ParsedArgs::TagFilterOp::kRegexp:
            filter =
                builder.reg_exp(args.tag_filter_column, args.tag_filter_value);
            break;
        case ParsedArgs::TagFilterOp::kNotRegexp:
            filter = builder.not_reg_exp(args.tag_filter_column,
                                         args.tag_filter_value);
            break;
        case ParsedArgs::TagFilterOp::kBetween:
            filter = builder.between_and(args.tag_filter_column,
                                         args.tag_filter_value,
                                         args.tag_filter_value2);
            break;
        case ParsedArgs::TagFilterOp::kNotBetween:
            filter = builder.not_between_and(args.tag_filter_column,
                                             args.tag_filter_value,
                                             args.tag_filter_value2);
            break;
        case ParsedArgs::TagFilterOp::kNone:
            break;
    }
    if (filter == nullptr) {
        err << "Error: invalid tag filter column '" << args.tag_filter_column
            << "' for table " << table_name << "\n";
        return std::unique_ptr<storage::Filter>();
    }
    return std::unique_ptr<storage::Filter>(filter);
}

std::vector<std::string> collect_tree_query_paths(
    const ParsedArgs& args, storage::TsFileReader& reader) {
    std::vector<std::string> paths;
    const bool has_projection = !args.measurements.empty();
    auto include_measurement = [&](const std::string& m) {
        return !has_projection ||
               std::find(args.measurements.begin(), args.measurements.end(),
                         m) != args.measurements.end();
    };

    if (!args.device.empty()) {
        // A single device was requested: resolve its series and keep only the
        // ones matching the projection. Filtering against the device's real
        // schema means a provided measurement that doesn't exist on the device
        // is dropped rather than queried blindly (matching the no-device path).
        auto did = std::make_shared<storage::StringArrayDeviceID>(args.device);
        std::vector<storage::MeasurementSchema> sch;
        if (reader.get_timeseries_schema(did, sch) == 0) {
            for (auto& m : sch) {
                if (include_measurement(m.measurement_name_)) {
                    paths.push_back(args.device + "." + m.measurement_name_);
                }
            }
        }
        return paths;
    }

    // No device filter: collect every device/series from a single whole-file
    // metadata call instead of querying each device one by one.
    storage::DeviceTimeseriesMetadataMap meta =
        reader.get_timeseries_metadata();
    for (auto& kv : meta) {
        if (!kv.first) {
            continue;
        }
        const std::string dev = kv.first->get_device_name();
        for (auto& ts : kv.second) {
            if (!ts) {
                continue;
            }
            const std::string m = ts->get_measurement_name().to_std_string();
            if (include_measurement(m)) {
                paths.push_back(dev + "." + m);
            }
        }
    }
    return paths;
}

int run_row_query(const ParsedArgs& args, storage::TsFileReader& reader,
                  OutputFormat fmt, std::ostream& out, std::ostream& err,
                  long long offset, long long limit) {
    const int64_t start = args.has_start ? static_cast<int64_t>(args.start)
                                         : std::numeric_limits<int64_t>::min();
    const int64_t end = args.has_end ? static_cast<int64_t>(args.end)
                                     : std::numeric_limits<int64_t>::max();

    storage::ResultSet* rs = nullptr;
    int qret = 0;
    const bool push_down = can_push_down_row_window(args, offset, limit);
    std::unique_ptr<storage::Filter> tag_filter;

    if (is_table_model(args, reader)) {
        std::string table_name = args.table;
        if (table_name.empty()) {
            auto schemas = reader.get_all_table_schemas();
            if (schemas.empty() || !schemas[0]) {
                err << "Error: no table found in file\n";
                return kExitRuntime;
            }
            table_name = schemas[0]->get_table_name();
        }
        std::vector<std::string> cols = args.measurements;
        if (cols.empty()) {
            auto ts = reader.get_table_schema(table_name);
            if (ts) {
                cols = ts->get_measurement_names();
            }
        }
        tag_filter = build_table_tag_filter(args, reader, table_name, err);
        if (args.has_tag_filter && tag_filter == nullptr) {
            return kExitUsage;
        }
        if (push_down) {
            qret = reader.queryByRow(table_name, cols,
                                     to_reader_row_bound(offset),
                                     to_reader_row_bound(limit), rs,
                                     tag_filter.get());
        } else {
            qret = reader.query(table_name, cols, start, end, rs,
                                tag_filter.get());
        }
    } else {
        if (args.has_tag_filter) {
            err << "Error: tag filter flags are only valid for table model\n";
            return kExitUsage;
        }
        std::vector<std::string> paths = collect_tree_query_paths(args, reader);
        if (paths.empty()) {
            err << "Error: no time series found\n";
            return kExitRuntime;
        }
        if (push_down) {
            qret = reader.queryByRow(paths, to_reader_row_bound(offset),
                                     to_reader_row_bound(limit), rs);
        } else {
            qret = reader.query(paths, start, end, rs);
        }
    }

    if (qret != 0 || rs == nullptr) {
        err << "Error: query failed: " << error_code_message(qret) << "\n";
        if (rs != nullptr) {
            reader.destroy_query_data_set(rs);
        }
        return kExitRuntime;
    }

    int wret = push_down
                   ? emit_result_set(rs, fmt, args.no_header, out)
                   : emit_result_set(rs, fmt, args.no_header, out, offset,
                                     limit);
    reader.destroy_query_data_set(rs);
    if (wret != 0) {
        err << "Error: failed to read rows: " << error_code_message(wret)
            << "\n";
        return kExitRuntime;
    }
    return kExitOk;
}

}  // namespace tsfile_cli