// 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 "load/group_commit/wal/wal_writer.h"

#include <crc32c/crc32c.h>

#include <gen_cpp/AgentService_types.h>

#include <gen_cpp/FrontendService_types.h>

#include "common/config.h"

#include "common/status.h"

#include "io/fs/encrypted_fs_factory.h"

#include "io/fs/file_system.h"

#include "io/fs/file_writer.h"

#include "io/fs/local_file_system.h"

#include "io/fs/path.h"

#include "load/group_commit/wal/wal_manager.h"

#include "runtime/cluster_info.h"

#include "storage/storage_engine.h"

#include "util/thrift_rpc_helper.h"

namespace doris {

const char* k_wal_magic = "WAL1";

const uint32_t k_wal_magic_length = 4;

WalWriter::WalWriter(const std::string& file_name) : _file_name(file_name) {}

WalWriter::~WalWriter() {}

Status determine_wal_fs(int64_t db_id, int64_t tb_id, io::FileSystemSPtr& fs) {

    if (!config::enable_wal_tde) {

        fs = io::global_local_filesystem();

        return Status::OK();

    }

#ifndef BE_TEST

    TNetworkAddress master_addr = ExecEnv::GetInstance()->cluster_info()->master_fe_addr;

    TGetTableTDEInfoRequest req;

    req.__set_db_id(db_id);

    req.__set_table_id(tb_id);

    TGetTableTDEInfoResult ret;

    RETURN_IF_ERROR(ThriftRpcHelper::rpc<FrontendServiceClient>(

            master_addr.hostname, master_addr.port,

            [&req, &ret](FrontendServiceConnection& client) {

                client->getTableTDEInfo(ret, req);

            }));

    if (auto st = Status::create(ret.status); !st) {

        return st;

    }

    auto encrypt_algorithm = [&ret]() -> EncryptionAlgorithmPB {

        switch (ret.algorithm) {

        case doris::TEncryptionAlgorithm::AES256:

            return EncryptionAlgorithmPB::AES_256_CTR;

        case doris::TEncryptionAlgorithm::SM4:

            return EncryptionAlgorithmPB::SM4_128_CTR;

        default:

            return EncryptionAlgorithmPB::PLAINTEXT;

        }

    }();

    auto local_fs = io::global_local_filesystem();

    fs = io::make_file_system(local_fs, encrypt_algorithm);

#else

    fs = io::global_local_filesystem();

#endif

    return Status::OK();

}

Status WalWriter::init(const io::FileSystemSPtr& fs) {

    io::Path wal_path = _file_name;

    auto parent_path = wal_path.parent_path();

    bool exists = false;

    RETURN_IF_ERROR(fs->exists(parent_path, &exists));

    if (!exists) {

        RETURN_IF_ERROR(fs->create_directory(parent_path));

    }

    RETURN_IF_ERROR(fs->create_file(_file_name, &_file_writer));

    LOG(INFO) << "create wal " << _file_name;

    return Status::OK();

}

Status WalWriter::finalize() {

    if (!_file_writer) {

        return Status::InternalError("wal writer is null,fail to close file={}", _file_name);

    }

    auto st = _file_writer->close();

    if (!st.ok()) {

        LOG(WARNING) << "fail to close wal " << _file_name;

    }

    return Status::OK();

}

Status WalWriter::append_blocks(const PBlockArray& blocks) {

    if (!_file_writer) {

        return Status::InternalError("wal writer is null,fail to write file={}", _file_name);

    }

    size_t total_size = 0;

    size_t offset = 0;

    for (const auto& block : blocks) {

        uint8_t len_buf[sizeof(uint64_t)];

        uint64_t block_length = block->ByteSizeLong();

        total_size += LENGTH_SIZE + block_length + CHECKSUM_SIZE;

        encode_fixed64_le(len_buf, block_length);

        RETURN_IF_ERROR(_file_writer->append({len_buf, sizeof(uint64_t)}));

        offset += LENGTH_SIZE;

        std::string content = block->SerializeAsString();

        RETURN_IF_ERROR(_file_writer->append(content));

        offset += block_length;

        uint8_t checksum_buf[sizeof(uint32_t)];

        uint32_t checksum = crc32c::Crc32c(content.data(), block_length);

        encode_fixed32_le(checksum_buf, checksum);

        RETURN_IF_ERROR(_file_writer->append({checksum_buf, sizeof(uint32_t)}));

        offset += CHECKSUM_SIZE;

    }

    if (offset != total_size) {

        return Status::InternalError(

                "failed to write block to wal expected= " + std::to_string(total_size) +

                ",actually=" + std::to_string(offset));

    }

    return Status::OK();

}

Status WalWriter::append_header(std::string col_ids) {

    if (!_file_writer) {

        return Status::InternalError("wal writer is null,fail to write file={}", _file_name);

    }

    size_t total_size = 0;

    uint64_t length = col_ids.size();

    total_size += k_wal_magic_length;

    total_size += VERSION_SIZE;

    total_size += LENGTH_SIZE;

    total_size += length;

    size_t offset = 0;

    RETURN_IF_ERROR(_file_writer->append({k_wal_magic, k_wal_magic_length}));

    offset += k_wal_magic_length;

    uint8_t version_buf[sizeof(uint32_t)];

    encode_fixed32_le(version_buf, WAL_VERSION);

    RETURN_IF_ERROR(_file_writer->append({version_buf, sizeof(uint32_t)}));

    offset += VERSION_SIZE;

    uint8_t len_buf[sizeof(uint64_t)];

    encode_fixed64_le(len_buf, length);

    RETURN_IF_ERROR(_file_writer->append({len_buf, sizeof(uint64_t)}));

    offset += LENGTH_SIZE;

    RETURN_IF_ERROR(_file_writer->append(col_ids));

    offset += length;

    if (offset != total_size) {

        return Status::InternalError(

                "failed to write header to wal expected= " + std::to_string(total_size) +

                ",actually=" + std::to_string(offset));

    }

    return Status::OK();

}

} // namespace doris