// 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 "io/fs/packed_file_writer.h"

#include <bvar/recorder.h>
#include <bvar/window.h>
#include <glog/logging.h>

#include "cloud/config.h"
#include "common/config.h"
#include "common/status.h"
#include "runtime/exec_env.h"
#include "util/slice.h"

namespace doris::io {

bvar::IntRecorder packed_file_writer_first_append_to_close_ms_recorder;
bvar::Window<bvar::IntRecorder> packed_file_writer_first_append_to_close_ms_window(
        "packed_file_writer_first_append_to_close_ms",
        &packed_file_writer_first_append_to_close_ms_recorder, /*window_size=*/10);

PackedFileWriter::PackedFileWriter(FileWriterPtr inner_writer, Path path,
                                   PackedAppendContext append_info)
        : _inner_writer(std::move(inner_writer)),
          _file_path(path.native()),
          _packed_file_manager(PackedFileManager::instance()),
          _append_info(std::move(append_info)) {
    DCHECK(_inner_writer != nullptr);
    DCHECK(!_file_path.empty());
}

PackedFileWriter::~PackedFileWriter() {
    if (_state == State::OPENED) {
        LOG(WARNING) << "PackedFileWriter destroyed without being closed, file: " << _file_path;
    }
}

Status PackedFileWriter::appendv(const Slice* data, size_t data_cnt) {
    if (_state != State::OPENED) {
        return Status::InternalError("Cannot append to closed or closing writer for file: " +
                                     _file_path);
    }

    if (!_first_append_timestamp.has_value()) {
        _first_append_timestamp = std::chrono::steady_clock::now();
    }

    // Calculate total size to append
    size_t total_size = 0;
    for (size_t i = 0; i < data_cnt; ++i) {
        total_size += data[i].size;
    }

    if (total_size == 0) {
        return Status::OK();
    }

    // Check if we should switch to direct write mode
    if (!_is_direct_write && _bytes_appended + total_size > config::small_file_threshold_bytes) {
        RETURN_IF_ERROR(_switch_to_direct_write());
        _is_direct_write = true;
    }

    // Write data based on current mode
    if (_is_direct_write) {
        RETURN_IF_ERROR(_inner_writer->appendv(data, data_cnt));
    } else {
        _buffer.reserve(_bytes_appended + total_size);
        // Buffer small file data
        for (size_t i = 0; i < data_cnt; ++i) {
            _buffer.append(data[i].data, data[i].size);
        }
    }

    _bytes_appended += total_size;
    return Status::OK();
}

Status PackedFileWriter::close(bool non_block) {
    if (_state == State::CLOSED) {
        return Status::OK();
    }

    auto record_close_latency = [this]() {
        if (_close_latency_recorded || !_first_append_timestamp.has_value()) {
            return;
        }
        auto now = std::chrono::steady_clock::now();
        auto latency_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
                                  now - *_first_append_timestamp)
                                  .count();
        packed_file_writer_first_append_to_close_ms_recorder << latency_ms;
        if (auto* sampler = packed_file_writer_first_append_to_close_ms_recorder.get_sampler()) {
            sampler->take_sample();
        }
        _close_latency_recorded = true;
    };

    if (_state == State::ASYNC_CLOSING) {
        if (non_block) {
            return Status::InternalError("Don't submit async close multi times");
        }
        if (!_is_direct_write) {
            RETURN_IF_ERROR(_wait_packed_upload());
        } else {
            RETURN_IF_ERROR(_inner_writer->close(false));
        }
        _state = State::CLOSED;
        if (!non_block) {
            record_close_latency();
        }
        return Status::OK();
    }

    if (non_block) {
        return _close_async();
    } else {
        return _close_sync();
    }
}

Status PackedFileWriter::_close_async() {
    if (!_is_direct_write) {
        // Send small file data to packed manager
        RETURN_IF_ERROR(_send_to_packed_manager());
    } else {
        // For large files, just close the inner writer asynchronously
        RETURN_IF_ERROR(_inner_writer->close(true));
    }
    _state = State::ASYNC_CLOSING;
    return Status::OK();
}

Status PackedFileWriter::_close_sync() {
    if (!_is_direct_write) {
        // Send small file data to pack manager and wait for upload
        RETURN_IF_ERROR(_send_to_packed_manager());
        RETURN_IF_ERROR(_wait_packed_upload());
    } else {
        // For large files, close the inner writer synchronously
        RETURN_IF_ERROR(_inner_writer->close(false));
    }
    _state = State::CLOSED;
    if (!_close_latency_recorded) {
        auto now = std::chrono::steady_clock::now();
        if (_first_append_timestamp.has_value()) {
            auto latency_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
                                      now - *_first_append_timestamp)
                                      .count();
            packed_file_writer_first_append_to_close_ms_recorder << latency_ms;
            if (auto* sampler =
                        packed_file_writer_first_append_to_close_ms_recorder.get_sampler()) {
                sampler->take_sample();
            }
            _close_latency_recorded = true;
        }
    }
    return Status::OK();
}

Status PackedFileWriter::_wait_packed_upload() {
    DCHECK(!_is_direct_write);
    // Only wait if we have data that was sent to packed manager
    if (_bytes_appended > 0 && _packed_file_manager != nullptr && _packed_slice_handle != nullptr) {
        return _packed_file_manager->wait_upload_done(_packed_slice_handle);
    }
    return Status::OK();
}

void PackedFileWriter::_release_buffer() {
    std::string().swap(_buffer);
}

Status PackedFileWriter::_switch_to_direct_write() {
    DCHECK(!_is_direct_write);

    // If we have buffered data, write it to inner writer first
    if (_buffer.size() > 0) {
        Slice buffer_slice(_buffer.data(), _buffer.size());
        RETURN_IF_ERROR(_inner_writer->appendv(&buffer_slice, 1));
        _release_buffer();
    }

    return Status::OK();
}

Status PackedFileWriter::_send_to_packed_manager() {
    DCHECK(!_is_direct_write);

    if (_packed_file_manager == nullptr) {
        return Status::InternalError("PackedFileManager is not available");
    }
    LOG(INFO) << "send_to_packed_manager: " << _file_path << " buffer size: " << _buffer.size();

    if (_append_info.resource_id.empty()) {
        return Status::InternalError("Missing resource id for packed file append");
    }

    if (_append_info.txn_id <= 0) {
        return Status::InvalidArgument("Missing valid txn id for packed file append: " +
                                       _file_path);
    }

    Slice data_slice(_buffer.data(), _buffer.size());
    RETURN_IF_ERROR(_packed_file_manager->append_small_file(_file_path, data_slice, _append_info,
                                                            &_packed_slice_handle));
    if (_packed_slice_handle == nullptr) {
        // append_small_file() skips data larger than `small_file_threshold_bytes` without
        // writing it anywhere. appendv() switches to direct write before the buffer can
        // reach that size, unless the threshold was lowered in between. Report the failure
        // instead of closing successfully on a file that does not exist.
        return Status::InternalError(
                "Packed file did not accept the buffered data for {}, size={}, threshold={}",
                _file_path, data_slice.get_size(), config::small_file_threshold_bytes);
    }
    _release_buffer();
    return Status::OK();
}

Status PackedFileWriter::get_packed_slice_location(PackedSliceLocation* location) const {
    DCHECK(_state == State::CLOSED)
            << " file_path: " << _file_path << " bytes_appended: " << _bytes_appended;
    if (_is_direct_write || _packed_slice_handle == nullptr) {
        *location = PackedSliceLocation {};
        return Status::OK();
    }
    *location = _packed_slice_handle->location();
    LOG(INFO) << "get_packed_slice_location: " << _file_path
              << " packed_path: " << location->packed_file_path << " " << location->offset << " "
              << location->size;
    return Status::OK();
}
} // namespace doris::io