// 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 "cloud/delete_bitmap_file_reader.h"

#include "cloud/delete_bitmap_file_writer.h"
#include "common/status.h"
#include "io/fs/file_reader.h"
#include "io/fs/packed_file_reader.h"
#include "util/coding.h"

namespace doris {

DeleteBitmapFileReader::DeleteBitmapFileReader(int64_t tablet_id, const std::string& rowset_id,
                                               std::optional<StorageResource>& storage_resource)
        : _tablet_id(tablet_id), _rowset_id(rowset_id), _storage_resource(storage_resource) {}

DeleteBitmapFileReader::DeleteBitmapFileReader(int64_t tablet_id, const std::string& rowset_id,
                                               std::optional<StorageResource>& storage_resource,
                                               const PackedSliceLocationPB& packed_location)
        : _tablet_id(tablet_id),
          _rowset_id(rowset_id),
          _storage_resource(storage_resource),
          _is_packed(true),
          _packed_offset(packed_location.offset()),
          _packed_size(packed_location.size()),
          _packed_file_path(packed_location.packed_file_path()),
          _packed_file_size(packed_location.has_packed_file_size()
                                    ? packed_location.packed_file_size()
                                    : -1) {}

DeleteBitmapFileReader::~DeleteBitmapFileReader() = default;

Status DeleteBitmapFileReader::init() {
#ifdef BE_TEST
    _path = "./log/" + _rowset_id + "_delete_bitmap.db";
    bool exists = false;
    RETURN_IF_ERROR(io::global_local_filesystem()->exists(_path, &exists));
    if (!exists) {
        return Status::NotFound("{} doesn't exist", _path);
    }
    RETURN_IF_ERROR(io::global_local_filesystem()->open_file(_path, &_file_reader));
    return Status::OK();
#endif
    if (!_storage_resource) {
        return Status::InternalError("invalid storage resource for tablet_id={}", _tablet_id);
    }

    if (_is_packed) {
        // Read from packed file
        io::FileReaderSPtr inner_reader;
        io::FileReaderOptions opts;
        if (_packed_file_size > 0) {
            opts.file_size = _packed_file_size;
        }
        opts.cache_type = io::FileCachePolicy::NO_CACHE;
        RETURN_IF_ERROR(_storage_resource->fs->open_file(io::Path(_packed_file_path), &inner_reader,
                                                         &opts));

        _path = _storage_resource->remote_delete_bitmap_path(_tablet_id, _rowset_id);
        _file_reader = std::make_shared<io::PackedFileReader>(
                std::move(inner_reader), io::Path(_path), _packed_offset, _packed_size);
    } else {
        // Read from standalone file
        _path = _storage_resource->remote_delete_bitmap_path(_tablet_id, _rowset_id);
        io::FileReaderOptions opts;
        RETURN_IF_ERROR(_storage_resource->fs->open_file(_path, &_file_reader, &opts));
    }
    return Status::OK();
}

Status DeleteBitmapFileReader::close() {
    if (_file_reader) {
        return _file_reader->close();
    }
    return Status::OK();
}

Status DeleteBitmapFileReader::read(DeleteBitmapPB& delete_bitmap) {
    size_t bytes_read = 0;
    size_t offset = 0;
    // 0. read magic number
    {
        uint8_t magic_buf[DeleteBitmapFileWriter::MAGIC_SIZE];
        RETURN_IF_ERROR(_file_reader->read_at(
                offset, {magic_buf, DeleteBitmapFileWriter::MAGIC_SIZE}, &bytes_read));
        offset += DeleteBitmapFileWriter::MAGIC_SIZE;
        if (bytes_read != DeleteBitmapFileWriter::MAGIC_SIZE ||
            memcmp(magic_buf, DeleteBitmapFileWriter::DELETE_BITMAP_MAGIC,
                   DeleteBitmapFileWriter::MAGIC_SIZE) != 0) {
            return Status::InternalError(
                    "read delete bitmap failed from {} because magic is not "
                    "matched",
                    _path);
        }
    }
    // 1. read delete bitmap proto length
    size_t delete_bitmap_len = 0;
    {
        uint8_t len_buf[DeleteBitmapFileWriter::LENGTH_SIZE];
        RETURN_IF_ERROR(_file_reader->read_at(
                offset, {len_buf, DeleteBitmapFileWriter::LENGTH_SIZE}, &bytes_read));
        offset += DeleteBitmapFileWriter::LENGTH_SIZE;
        delete_bitmap_len = decode_fixed64_le(len_buf);
        if (delete_bitmap_len == 0) {
            return Status::InternalError("read delete bitmap failed from {} because length is 0",
                                         _path);
        }
        if (offset == _file_reader->size()) {
            LOG(WARNING) << "read delete bitmap failed because reach end of file=" << _path
                         << ", file size=" << _file_reader->size() << ", offset=" << offset
                         << ", delete_bitmap_len=" << delete_bitmap_len;
            return Status::InternalError(
                    "read delete bitmap failed from {} because reach end of file", _path);
        }
    }
    // 2. read delete bitmap
    std::string delete_bitmap_buf;
    {
        delete_bitmap_buf.resize(delete_bitmap_len);
        RETURN_IF_ERROR(_file_reader->read_at(
                offset, {delete_bitmap_buf.c_str(), delete_bitmap_len}, &bytes_read));
        offset += delete_bitmap_len;
        if (!delete_bitmap.ParseFromString(delete_bitmap_buf)) {
            LOG(WARNING) << "deserialize delete bitmap failed from file=" << _path
                         << ", file size=" << _file_reader->size()
                         << ", delete_bitmap_len=" << delete_bitmap_len
                         << ", bytes_read=" << bytes_read;
            return Status::InternalError("deserialize delete bitmap failed from {}", _path);
        }
    }
    // 3. checksum
    uint8_t checksum_len_buf[DeleteBitmapFileWriter::CHECKSUM_SIZE];
    RETURN_IF_ERROR(_file_reader->read_at(
            offset, {checksum_len_buf, DeleteBitmapFileWriter::CHECKSUM_SIZE}, &bytes_read));
    offset += DeleteBitmapFileWriter::CHECKSUM_SIZE;
    uint32_t checksum = decode_fixed32_le(checksum_len_buf);
    uint32_t computed_checksum = crc32c::Crc32c(delete_bitmap_buf.data(), delete_bitmap_len);
    if (computed_checksum != checksum) {
        return Status::InternalError("delete bitmap checksum failed from file=" + _path +
                                     ", computed checksum=" + std::to_string(computed_checksum) +
                                     ", expected=" + std::to_string(checksum));
    }
    return Status::OK();
}

} // namespace doris