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

#pragma once

#include <stdio.h>

#include <sys/stat.h>

#include <memory>

#include <string>

#include <vector>

#include "io/fs/file_reader.h"

#include "io/fs/file_writer.h"

#include "io/fs/local_file_system.h"

#include "olap/lru_cache.h"

#include "olap/olap_common.h"

#include "olap/olap_define.h"

#include "olap/utils.h"

#include "util/debug_util.h"

namespace doris {

using FixedFileHeader = struct _FixedFileHeader {

    // the length of the entire file

    uint32_t file_length;

    // Checksum of the file's contents except the FileHeader

    uint32_t checksum;

    // Protobuf length of section

    uint32_t protobuf_length;

    // Checksum of Protobuf part

    uint32_t protobuf_checksum;

} __attribute__((packed));

using FixedFileHeaderV2 = struct _FixedFileHeaderV2 {

    uint64_t magic_number;

    uint32_t version;

    // the length of the entire file

    uint64_t file_length;

    // Checksum of the file's contents except the FileHeader

    uint32_t checksum;

    // Protobuf length of section

    uint64_t protobuf_length;

    // Checksum of Protobuf part

    uint32_t protobuf_checksum;

} __attribute__((packed));

template <typename MessageType, typename ExtraType = uint32_t>

class FileHeader {

public:

    FileHeader(const std::string& file_path) : _file_path(file_path) {

        memset(&_fixed_file_header, 0, sizeof(_fixed_file_header));

        memset(&_extra_fixed_header, 0, sizeof(_extra_fixed_header));

        _fixed_file_header_size = sizeof(_fixed_file_header);

    }

    ~FileHeader() = default;

    // To calculate the length of the proto part, it needs to be called after the proto is operated,

    // and prepare must be called before calling serialize

    Status prepare();

    // call prepare() first, serialize() will write fixed header and protobuffer.

    // Write the header to the starting position of the incoming file handle

    Status serialize();

    // read from file, validate file length, signature and alder32 of protobuffer.

    // Read the header from the beginning of the incoming file handle

    Status deserialize();

    // Check the validity of Header

    // it is actually call deserialize().

    Status validate();

    // write to memory

    Status serialize_to_memory(uint8_t* buffer, size_t buffer_size);

    // read from memory

    Status deserialize_from_memory(const uint8_t* buffer, size_t buffer_size);

    uint64_t file_length() const { return _fixed_file_header.file_length; }

    uint32_t checksum() const { return _fixed_file_header.checksum; }

    const ExtraType& extra() const { return _extra_fixed_header; }

    ExtraType* mutable_extra() { return &_extra_fixed_header; }

    const MessageType& message() const { return _proto; }

    MessageType* mutable_message() { return &_proto; }

    uint64_t size() const {

        return _fixed_file_header_size + sizeof(_extra_fixed_header) +

               _fixed_file_header.protobuf_length;

    }

    void set_file_length(uint64_t file_length) { _fixed_file_header.file_length = file_length; }

    void set_checksum(uint32_t checksum) { _fixed_file_header.checksum = checksum; }

private:

    std::string _file_path;

    FixedFileHeaderV2 _fixed_file_header;

    uint32_t _fixed_file_header_size;

    std::string _proto_string;

    ExtraType _extra_fixed_header;

    MessageType _proto;

};

// FileHeader implementation

template <typename MessageType, typename ExtraType>

Status FileHeader<MessageType, ExtraType>::prepare() {

    try {

        if (!_proto.SerializeToString(&_proto_string)) {

            return Status::Error<ErrorCode::SERIALIZE_PROTOBUF_ERROR>(

                    "serialize file header to string error. [path={}]", _file_path);

        }

    } catch (...) {

        return Status::Error<ErrorCode::SERIALIZE_PROTOBUF_ERROR>(

                "serialize file header to string error. [path={}]", _file_path);

    }

    _fixed_file_header.protobuf_checksum =

            olap_adler32(olap_adler32_init(), _proto_string.c_str(), _proto_string.size());

    _fixed_file_header.checksum = 0;

    _fixed_file_header.protobuf_length = _proto_string.size();

    _fixed_file_header.file_length = size();

    _fixed_file_header.version = OLAP_DATA_VERSION_APPLIED;

    _fixed_file_header.magic_number = OLAP_FIX_HEADER_MAGIC_NUMBER;

    return Status::OK();

}

template <typename MessageType, typename ExtraType>

Status FileHeader<MessageType, ExtraType>::serialize() {

    // write to file

    io::FileWriterPtr file_writer;

    RETURN_IF_ERROR(io::global_local_filesystem()->create_file(_file_path, &file_writer));

    RETURN_IF_ERROR(

            file_writer->append({(const uint8_t*)&_fixed_file_header, _fixed_file_header_size}));

    RETURN_IF_ERROR(file_writer->append(

            {(const uint8_t*)&_extra_fixed_header, sizeof(_extra_fixed_header)}));

    RETURN_IF_ERROR(file_writer->append({_proto_string}));

    return file_writer->close();

}

template <typename MessageType, typename ExtraType>

Status FileHeader<MessageType, ExtraType>::serialize_to_memory(uint8_t* buffer,

                                                               size_t buffer_size) {

    if (buffer_size < size()) {

        return Status::Error<ErrorCode::INVALID_ARGUMENT>(

                "buffer size is too small. required={}, provided={}", size(), buffer_size);

    }

    uint8_t* ptr = buffer;

    memcpy(ptr, &_fixed_file_header, _fixed_file_header_size);

    ptr += _fixed_file_header_size;

    memcpy(ptr, &_extra_fixed_header, sizeof(_extra_fixed_header));

    ptr += sizeof(_extra_fixed_header);

    memcpy(ptr, _proto_string.data(), _proto_string.size());

    return Status::OK();

}

template <typename MessageType, typename ExtraType>

Status FileHeader<MessageType, ExtraType>::deserialize() {

    io::FileReaderSPtr file_reader;

    RETURN_IF_ERROR(io::global_local_filesystem()->open_file(_file_path, &file_reader));

    off_t real_file_length = 0;

    uint32_t real_protobuf_checksum = 0;

    size_t bytes_read = 0;

    RETURN_IF_ERROR(file_reader->read_at(

            0, {(const uint8_t*)&_fixed_file_header, _fixed_file_header_size}, &bytes_read));

    DCHECK(_fixed_file_header_size == bytes_read)

            << " deserialize read bytes dismatch, request bytes " << _fixed_file_header_size

            << " actual read " << bytes_read;

    //Status read_at(size_t offset, Slice result, size_t* bytes_read,

    //             const IOContext* io_ctx = nullptr);

    if (_fixed_file_header.magic_number != OLAP_FIX_HEADER_MAGIC_NUMBER) {

        VLOG_TRACE << "old fix header found, magic num=" << _fixed_file_header.magic_number;

        FixedFileHeader tmp_header;

        RETURN_IF_ERROR(file_reader->read_at(0, {(const uint8_t*)&tmp_header, sizeof(tmp_header)},

                                             &bytes_read));

        DCHECK(sizeof(tmp_header) == bytes_read)

                << " deserialize read bytes dismatch, request bytes " << sizeof(tmp_header)

                << " actual read " << bytes_read;

        _fixed_file_header.file_length = tmp_header.file_length;

        _fixed_file_header.checksum = tmp_header.checksum;

        _fixed_file_header.protobuf_length = tmp_header.protobuf_length;

        _fixed_file_header.protobuf_checksum = tmp_header.protobuf_checksum;

        _fixed_file_header.magic_number = OLAP_FIX_HEADER_MAGIC_NUMBER;

        _fixed_file_header.version = OLAP_DATA_VERSION_APPLIED;

        _fixed_file_header_size = sizeof(tmp_header);

    }

    VLOG_NOTICE << "fix head loaded. file_length=" << _fixed_file_header.file_length

                << ", checksum=" << _fixed_file_header.checksum

                << ", protobuf_length=" << _fixed_file_header.protobuf_length

                << ", magic_number=" << _fixed_file_header.magic_number

                << ", version=" << _fixed_file_header.version;

    RETURN_IF_ERROR(file_reader->read_at(

            _fixed_file_header_size,

            {(const uint8_t*)&_extra_fixed_header, sizeof(_extra_fixed_header)}, &bytes_read));

    std::unique_ptr<char[]> buf(new (std::nothrow) char[_fixed_file_header.protobuf_length]);

    if (nullptr == buf) {

        char errmsg[64];

        return Status::Error<ErrorCode::MEM_ALLOC_FAILED>(

                "malloc protobuf buf error. file={}, error={}", file_reader->path().native(),

                strerror_r(errno, errmsg, 64));

    }

    RETURN_IF_ERROR(file_reader->read_at(_fixed_file_header_size + sizeof(_extra_fixed_header),

                                         {buf.get(), _fixed_file_header.protobuf_length},

                                         &bytes_read));

    real_file_length = file_reader->size();

    if (file_length() != static_cast<uint64_t>(real_file_length)) {

        return Status::InternalError(

                "file length is not match. file={}, file_length={}, real_file_length={}",

                file_reader->path().native(), file_length(), real_file_length);

    }

    // check proto checksum

    real_protobuf_checksum =

            olap_adler32(olap_adler32_init(), buf.get(), _fixed_file_header.protobuf_length);

    if (real_protobuf_checksum != _fixed_file_header.protobuf_checksum) {

        // When compiling using gcc there woule be error like:

        // Cannot bind packed field '_FixedFileHeaderV2::protobuf_checksum' to 'unsigned int&'

        // so we need to using unary operator+ to evaluate one value to pass

        // to status to successfully compile.

        return Status::InternalError("checksum is not match. file={}, expect={}, actual={}",

                                     file_reader->path().native(),

                                     +_fixed_file_header.protobuf_checksum, real_protobuf_checksum);

    }

    try {

        std::string protobuf_str(buf.get(), _fixed_file_header.protobuf_length);

        if (!_proto.ParseFromString(protobuf_str)) {

            return Status::Error<ErrorCode::PARSE_PROTOBUF_ERROR>(

                    "fail to parse file content to protobuf object. file={}",

                    file_reader->path().native());

        }

    } catch (...) {

        LOG(WARNING) << "fail to load protobuf. file='" << file_reader->path().native();

        return Status::Error<ErrorCode::PARSE_PROTOBUF_ERROR>("fail to load protobuf. file={}",

                                                              file_reader->path().native());

    }

    return Status::OK();

}

template <typename MessageType, typename ExtraType>

Status FileHeader<MessageType, ExtraType>::deserialize_from_memory(const uint8_t* buffer,

                                                                   size_t buffer_size) {

    if (buffer_size < sizeof(FixedFileHeaderV2)) {

        return Status::Error<ErrorCode::INVALID_ARGUMENT>(

                "buffer size is too small to contain a valid header. provided={}", buffer_size);

    }

    const uint8_t* ptr = buffer;

    memcpy(&_fixed_file_header, ptr, _fixed_file_header_size);

    ptr += _fixed_file_header_size;

    if (_fixed_file_header.magic_number != OLAP_FIX_HEADER_MAGIC_NUMBER) {

        VLOG_TRACE << "old fix header found, magic num=" << _fixed_file_header.magic_number;

        if (buffer_size < sizeof(FixedFileHeader)) {

            return Status::Error<ErrorCode::INVALID_ARGUMENT>(

                    "buffer size is too small to contain a valid old header. provided={}",

                    buffer_size);

        }

        FixedFileHeader tmp_header;

        memcpy(&tmp_header, buffer, sizeof(tmp_header));

        _fixed_file_header.file_length = tmp_header.file_length;

        _fixed_file_header.checksum = tmp_header.checksum;

        _fixed_file_header.protobuf_length = tmp_header.protobuf_length;

        _fixed_file_header.protobuf_checksum = tmp_header.protobuf_checksum;

        _fixed_file_header.magic_number = OLAP_FIX_HEADER_MAGIC_NUMBER;

        _fixed_file_header.version = OLAP_DATA_VERSION_APPLIED;

        _fixed_file_header_size = sizeof(tmp_header);

        ptr = buffer + sizeof(tmp_header);

    }

    VLOG_NOTICE << "fix head loaded. file_length=" << _fixed_file_header.file_length

                << ", checksum=" << _fixed_file_header.checksum

                << ", protobuf_length=" << _fixed_file_header.protobuf_length

                << ", magic_number=" << _fixed_file_header.magic_number

                << ", version=" << _fixed_file_header.version;

    if (buffer_size < _fixed_file_header_size + sizeof(_extra_fixed_header) +

                              _fixed_file_header.protobuf_length) {

        return Status::Error<ErrorCode::INVALID_ARGUMENT>(

                "buffer size is too small to contain the entire header and protobuf. required={}, "

                "provided={}",

                _fixed_file_header_size + sizeof(_extra_fixed_header) +

                        _fixed_file_header.protobuf_length,

                buffer_size);

    }

    memcpy(&_extra_fixed_header, ptr, sizeof(_extra_fixed_header));

    ptr += sizeof(_extra_fixed_header);

    std::unique_ptr<char[]> buf(new (std::nothrow) char[_fixed_file_header.protobuf_length]);

    if (nullptr == buf) {

        char errmsg[64];

        return Status::Error<ErrorCode::MEM_ALLOC_FAILED>("malloc protobuf buf error. error={}",

                                                          strerror_r(errno, errmsg, 64));

    }

    memcpy(buf.get(), ptr, _fixed_file_header.protobuf_length);

    // check proto checksum

    uint32_t real_protobuf_checksum =

            olap_adler32(olap_adler32_init(), buf.get(), _fixed_file_header.protobuf_length);

    if (real_protobuf_checksum != _fixed_file_header.protobuf_checksum) {

        return Status::InternalError("checksum is not match. expect={}, actual={}",

                                     +_fixed_file_header.protobuf_checksum, real_protobuf_checksum);

    }

    try {

        std::string protobuf_str(buf.get(), _fixed_file_header.protobuf_length);

        if (!_proto.ParseFromString(protobuf_str)) {

            return Status::Error<ErrorCode::PARSE_PROTOBUF_ERROR>(

                    "fail to parse buffer content to protobuf object.");

        }

    } catch (...) {

        LOG(WARNING) << "fail to load protobuf from buffer.";

        return Status::Error<ErrorCode::PARSE_PROTOBUF_ERROR>("fail to load protobuf from buffer.");

    }

    return Status::OK();

}

template <typename MessageType, typename ExtraType>

Status FileHeader<MessageType, ExtraType>::validate() {

    return deserialize();

}

} // namespace doris