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

#include <crc32c/crc32c.h>

#include "cloud/config.h"

#include "io/fs/file_writer.h"

#include "io/fs/packed_file_writer.h"

namespace doris {

DeleteBitmapFileWriter::DeleteBitmapFileWriter(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) {}

DeleteBitmapFileWriter::DeleteBitmapFileWriter(int64_t tablet_id, const std::string& rowset_id,

                                               std::optional<StorageResource>& storage_resource,

                                               bool enable_packed_file, int64_t txn_id)

        : _tablet_id(tablet_id),

          _rowset_id(rowset_id),

          _storage_resource(storage_resource),

          _enable_packed_file(enable_packed_file),

          _txn_id(txn_id) {}

DeleteBitmapFileWriter::~DeleteBitmapFileWriter() {}

Status DeleteBitmapFileWriter::init() {

#ifdef BE_TEST

    _path = "./log/" + _rowset_id + "_delete_bitmap.db";

    io::Path path = _path;

    auto parent_path = path.parent_path();

    bool exists = false;

    RETURN_IF_ERROR(io::global_local_filesystem()->exists(parent_path, &exists));

    if (!exists) {

        RETURN_IF_ERROR(io::global_local_filesystem()->create_directory(parent_path));

    }

    RETURN_IF_ERROR(io::global_local_filesystem()->create_file(_path, &_file_writer));

    return Status::OK();

#endif

    if (!_storage_resource) {

        return Status::InternalError("invalid storage resource for tablet_id={}", _tablet_id);

    }

    _path = _storage_resource->remote_delete_bitmap_path(_tablet_id, _rowset_id);

    io::FileWriterOptions opts;

    if (_enable_packed_file) {

        // Create underlying file writer

        io::FileWriterPtr inner_writer;

        // Disable write_file_cache for inner writer when using PackedFileWriter.

        // Small files will be cached separately by PackedFileManager using the

        // small file path as cache key.

        opts.write_file_cache = false;

        RETURN_IF_ERROR(_storage_resource->fs->create_file(_path, &inner_writer, &opts));

        // Wrap with PackedFileWriter

        io::PackedAppendContext append_info;

        append_info.resource_id = _storage_resource->fs->id();

        append_info.tablet_id = _tablet_id;

        append_info.rowset_id = _rowset_id;

        append_info.txn_id = _txn_id;

        append_info.write_file_cache = false;

        _file_writer = std::make_unique<io::PackedFileWriter>(std::move(inner_writer),

                                                              io::Path(_path), append_info);

    } else {

        RETURN_IF_ERROR(_storage_resource->fs->create_file(_path, &_file_writer, &opts));

    }

    return Status::OK();

}

Status DeleteBitmapFileWriter::close() {

    if (!_file_writer) {

        return Status::InternalError("fail to close delete bitmap file={} because writer is null",

                                     _path);

    }

    auto st = _file_writer->close();

    if (!st.ok()) {

        LOG(WARNING) << "failed to close delete bitmap file=" << _path << ", st=" << st.to_string();

        return st;

    }

    // Check if file was written to packed file

    if (_enable_packed_file) {

        auto* packed_writer = static_cast<io::PackedFileWriter*>(_file_writer.get());

        io::PackedSliceLocation loc;

        st = packed_writer->get_packed_slice_location(&loc);

        if (!st.ok()) {

            LOG(WARNING) << "failed to get packed slice location for delete bitmap file=" << _path

                         << ", st=" << st.to_string();

            return st;

        }

        if (!loc.packed_file_path.empty()) {

            _is_packed = true;

            _packed_location = loc;

        }

    }

    return Status::OK();

}

Status DeleteBitmapFileWriter::get_packed_slice_location(io::PackedSliceLocation* location) const {

    if (!_is_packed) {

        return Status::InternalError("delete bitmap file is not packed");

    }

    *location = _packed_location;

    return Status::OK();

}

Status DeleteBitmapFileWriter::write(const DeleteBitmapPB& delete_bitmap) {

    if (delete_bitmap.rowset_ids_size() == 0) {

        return Status::InternalError("empty delete bitmap for file={}", _path);

    }

    if (!_file_writer) {

        return Status::InternalError("fail to write delete bitmap file={} because writer is null",

                                     _path);

    }

    // 0. write magic

    RETURN_IF_ERROR(_file_writer->append({DELETE_BITMAP_MAGIC, MAGIC_SIZE}));

    // 1. write delete bitmap length

    uint64_t delete_bitmap_len = delete_bitmap.ByteSizeLong();

    uint8_t len_buf[LENGTH_SIZE];

    encode_fixed64_le(len_buf, delete_bitmap_len);

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

    // 2. write delete bitmap

    std::string content = delete_bitmap.SerializeAsString();

    RETURN_IF_ERROR(_file_writer->append(content));

    // 3. write checksum

    uint8_t checksum_buf[CHECKSUM_SIZE];

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

    encode_fixed32_le(checksum_buf, checksum);

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

    return Status::OK();

}

} // namespace doris