// 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.

// clang-format off
#include "util.h"

#include <bthread/butex.h>
#include <butil/iobuf.h>
#include <google/protobuf/util/json_util.h>

// FIXME: we should not rely other modules that may rely on this common module
#include "common/logging.h"
#include "meta-service/keys.h"
#include "meta-service/codec.h"
#include "meta-service/txn_kv.h"
#include "meta-service/txn_kv_error.h"

#include <iomanip>
#include <sstream>
#include <unordered_map>
#include <variant>
// clang-format on

namespace doris::cloud {

/**
 * This is a naïve implementation of hex, DONOT use it on retical path.
 */
std::string hex(std::string_view str) {
    std::stringstream ss;
    for (auto& i : str) {
        ss << std::hex << std::setw(2) << std::setfill('0') << ((int16_t)i & 0xff);
    }
    return ss.str();
}

/**
 * This is a naïve implementation of unhex.
 */
std::string unhex(std::string_view hex_str) {
    // clang-format off
    const static std::unordered_map<char, char> table = {
            {'0', 0},  {'1', 1},  {'2', 2},  {'3', 3},  {'4', 4}, 
            {'5', 5},  {'6', 6},  {'7', 7},  {'8', 8},  {'9', 9}, 
            {'a', 10}, {'b', 11}, {'c', 12}, {'d', 13}, {'e', 14}, {'f', 15},
            {'A', 10}, {'B', 11}, {'C', 12}, {'D', 13}, {'E', 14}, {'F', 15}};
    [[maybe_unused]] static int8_t lut[std::max({'9', 'f', 'F'}) + 1];
    lut[(int)'0'] = 0; lut[(int)'1'] = 1; lut[(int)'2'] = 2; lut[(int)'3'] = 3; lut[(int)'4'] = 4; lut[(int)'5'] = 5; lut[(int)'6'] = 6; lut[(int)'7'] = 7; lut[(int)'8'] = 8; lut[(int)'9'] = 9;
    lut[(int)'a'] = 10; lut[(int)'b'] = 11; lut[(int)'c'] = 12; lut[(int)'d'] = 13; lut[(int)'e'] = 14; lut[(int)'f'] = 15;
    lut[(int)'A'] = 10; lut[(int)'B'] = 11; lut[(int)'C'] = 12; lut[(int)'D'] = 13; lut[(int)'E'] = 14; lut[(int)'F'] = 15;
    // clang-format on
    size_t len = hex_str.length();
    len &= ~0x01UL;
    std::string buf(len >> 1, '\0');
    for (size_t i = 0; i < len; ++i) {
        const auto it = table.find(hex_str[i]);
        if (it == table.end()) break;
        buf[i >> 1] |= i & 0x1 ? (it->second & 0x0f) : (it->second & 0x0f) << 4;
    }
    return buf;
}

static std::string explain_fields(std::string_view text, const std::vector<std::string>& fields,
                                  const std::vector<int>& pos, bool unicode = false) {
    if (fields.size() != pos.size() || fields.size() == 0 || pos.size() == 0) {
        return std::string(text.data(), text.size());
    }
    size_t last_hyphen_pos = pos.back() + 1;
    std::stringstream ss;
    std::string blank_line(last_hyphen_pos + 1, ' ');

    // clang-format off
    static const std::string hyphen("\xe2\x94\x80"); // ─ e2 94 80
    static const std::string bar   ("\xe2\x94\x82"); // │ e2 94 82
    static const std::string angle ("\xe2\x94\x8c"); // ┌ e2 94 8c
    static const std::string arrow ("\xe2\x96\xbc"); // ▼ e2 96 bc
    // clang-format on

    // Each line with hyphens
    for (size_t i = 0; i < fields.size(); ++i) {
        std::string line = blank_line;
        line[pos[i]] = '/';
        int nbar = i;
        for (size_t j = 0; j < i; ++j) {
            line[pos[j]] = '|';
        }
        int nhyphen = 0;
        for (size_t j = pos[i] + 1; j <= last_hyphen_pos; ++j) {
            line[j] = '-';
            ++nhyphen;
        }

        if (unicode) {
            int i = line.size();
            line.resize(line.size() + 2 * (1 /*angle*/ + nbar + nhyphen), ' ');
            int j = line.size();
            while (--i >= 0) {
                if (line[i] == '-') {
                    line[--j] = hyphen[2];
                    line[--j] = hyphen[1];
                    line[--j] = hyphen[0];
                } else if (line[i] == '|') {
                    line[--j] = bar[2];
                    line[--j] = bar[1];
                    line[--j] = bar[0];
                } else if (line[i] == '/') {
                    line[--j] = angle[2];
                    line[--j] = angle[1];
                    line[--j] = angle[0];
                } else {
                    --j;
                    continue;
                }
                line[i] = i != j ? ' ' : line[i]; // Replace if needed
            }
        }

        ss << line << " " << i << ". " << fields[i] << "\n";
    }

    // Mark position indicator
    std::string line = blank_line;
    for (size_t i = 0; i < fields.size(); ++i) {
        line[pos[i]] = '|';
    }

    if (unicode) {
        int i = line.size();
        line.resize(line.size() + 2 * fields.size(), ' ');
        int j = line.size();
        while (--i >= 0) {
            if (line[i] != '|') {
                --j;
                continue;
            }
            line[--j] = bar[2];
            line[--j] = bar[1];
            line[--j] = bar[0];
            line[i] = i != j ? ' ' : line[i]; // Replace if needed
        }
    }

    ss << line << "\n";

    line = blank_line;
    for (size_t i = 0; i < fields.size(); ++i) {
        line[pos[i]] = 'v';
    }

    if (unicode) {
        int i = line.size();
        line.resize(line.size() + 2 * fields.size(), ' ');
        int j = line.size();
        while (--i >= 0) {
            if (line[i] != 'v') {
                --j;
                continue;
            }
            line[--j] = arrow[2];
            line[--j] = arrow[1];
            line[--j] = arrow[0];
            line[i] = i != j ? ' ' : line[i]; // Replace if needed
        }
    }

    ss << line << "\n";

    // Original text to explain
    ss << text << "\n";

    return ss.str();
}

std::string prettify_key(std::string_view key_hex, bool unicode) {
    // Decoded result container
    //                                    val                  tag  pos
    //                     .---------------^----------------.  .^.  .^.
    std::vector<std::tuple<std::variant<int64_t, std::string>, int, int>> fields;
    std::string unhex_key = unhex(key_hex);
    int key_space = unhex_key[0];
    std::string_view key_copy = unhex_key;
    key_copy.remove_prefix(1); // Remove the first key space byte
    int ret = decode_key(&key_copy, &fields);
    if (ret != 0) return "";

    std::vector<std::string> fields_str;
    std::vector<int> fields_pos;
    fields_str.reserve(fields.size() + 1);
    fields_pos.reserve(fields.size() + 1);
    // Key space byte
    fields_str.push_back("key space: " + std::to_string(key_space));
    fields_pos.push_back(0);

    for (auto& i : fields) {
        fields_str.emplace_back(std::get<1>(i) == EncodingTag::BYTES_TAG
                                        ? std::get<std::string>(std::get<0>(i))
                                        : std::to_string(std::get<int64_t>(std::get<0>(i))));
        fields_pos.push_back((std::get<2>(i) + 1) * 2);
    }

    return explain_fields(key_hex, fields_str, fields_pos, unicode);
}

std::string proto_to_json(const ::google::protobuf::Message& msg, bool add_whitespace) {
    std::string json;
    google::protobuf::util::JsonPrintOptions opts;
    opts.add_whitespace = add_whitespace;
    opts.preserve_proto_field_names = true;
    google::protobuf::util::MessageToJsonString(msg, &json, opts);
    return json;
}

std::vector<std::string_view> split_string(const std::string_view& str, int n) {
    std::vector<std::string_view> substrings;

    for (size_t i = 0; i < str.size(); i += n) {
        substrings.push_back(str.substr(i, n));
    }

    return substrings;
}

bool ValueBuf::to_pb(google::protobuf::Message* pb) const {
    butil::IOBuf merge;
    for (auto&& it : iters) {
        it->reset();
        while (it->has_next()) {
            auto [k, v] = it->next();
            merge.append_user_data((void*)v.data(), v.size(), +[](void*) {});
        }
    }
    butil::IOBufAsZeroCopyInputStream merge_stream(merge);
    return pb->ParseFromZeroCopyStream(&merge_stream);
}

std::string ValueBuf::value() const {
    butil::IOBuf merge;
    for (auto&& it : iters) {
        it->reset();
        while (it->has_next()) {
            auto [k, v] = it->next();
            merge.append_user_data((void*)v.data(), v.size(), +[](void*) {});
        }
    }
    return merge.to_string();
}

std::vector<std::string> ValueBuf::keys() const {
    std::vector<std::string> ret;
    for (auto&& it : iters) {
        it->reset();
        while (it->has_next()) {
            auto [k, _] = it->next();
            ret.push_back({k.data(), k.size()});
        }
    }
    return ret;
}

void ValueBuf::remove(Transaction* txn) const {
    for (auto&& it : iters) {
        it->reset();
        while (it->has_next()) {
            txn->remove(it->next().first);
        }
    }
}

TxnErrorCode ValueBuf::get(Transaction* txn, std::string_view key, bool snapshot) {
    iters.clear();
    ver = -1;

    std::string begin_key {key};
    std::string end_key {key};
    encode_int64(INT64_MAX, &end_key);
    std::unique_ptr<RangeGetIterator> it;
    TxnErrorCode err = txn->get(begin_key, end_key, &it, snapshot);
    if (err != TxnErrorCode::TXN_OK) {
        return err;
    }
    if (!it->has_next()) {
        return TxnErrorCode::TXN_KEY_NOT_FOUND;
    }
    // Extract version
    auto [k, _] = it->next();
    if (k.size() == key.size()) { // Old version KV
        DCHECK(k == key) << hex(k) << ' ' << hex(key);
        DCHECK_EQ(it->size(), 1) << hex(k) << ' ' << hex(key);
        ver = 0;
    } else {
        k.remove_prefix(key.size());
        int64_t suffix;
        if (decode_int64(&k, &suffix) != 0) [[unlikely]] {
            LOG_WARNING("failed to decode key").tag("key", hex(k));
            return TxnErrorCode::TXN_UNIDENTIFIED_ERROR;
        }
        ver = suffix >> 56 & 0xff;
    }
    bool more = it->more();
    if (!more) {
        iters.push_back(std::move(it));
        return TxnErrorCode::TXN_OK;
    }
    begin_key = it->next_begin_key();
    iters.push_back(std::move(it));
    do {
        err = txn->get(begin_key, end_key, &it, snapshot);
        if (err != TxnErrorCode::TXN_OK) {
            return err;
        }
        more = it->more();
        if (more) {
            begin_key = it->next_begin_key();
        }
        iters.push_back(std::move(it));
    } while (more);
    return TxnErrorCode::TXN_OK;
}

TxnErrorCode get(Transaction* txn, std::string_view key, ValueBuf* val, bool snapshot) {
    return val->get(txn, key, snapshot);
}

TxnErrorCode key_exists(Transaction* txn, std::string_view key, bool snapshot) {
    std::string end_key {key};
    encode_int64(INT64_MAX, &end_key);
    std::unique_ptr<RangeGetIterator> it;
    TxnErrorCode err = txn->get(key, end_key, &it, snapshot, 1);
    if (err != TxnErrorCode::TXN_OK) {
        return err;
    }
    return it->has_next() ? TxnErrorCode::TXN_OK : TxnErrorCode::TXN_KEY_NOT_FOUND;
}

void put(Transaction* txn, std::string_view key, const google::protobuf::Message& pb, uint8_t ver,
         size_t split_size) {
    std::string value;
    bool ret = pb.SerializeToString(&value); // Always success
    DCHECK(ret) << hex(key) << ' ' << pb.ShortDebugString();
    put(txn, key, value, ver, split_size);
}

void put(Transaction* txn, std::string_view key, std::string_view value, uint8_t ver,
         size_t split_size) {
    auto split_vec = split_string(value, split_size);
    int64_t suffix_base = ver;
    suffix_base <<= 56;
    for (size_t i = 0; i < split_vec.size(); ++i) {
        std::string k(key);
        encode_int64(suffix_base + i, &k);
        txn->put(k, split_vec[i]);
    }
}

} // namespace doris::cloud