/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2025. All rights reserved.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *          http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */

#include "KafkaPartitionSplit.h"
#include <stdexcept>
#include <sstream>
#include <limits>

const long KafkaPartitionSplit::NO_STOPPING_OFFSET = std::numeric_limits<long>::min();
const long KafkaPartitionSplit::LATEST_OFFSET = -1;
const long KafkaPartitionSplit::EARLIEST_OFFSET = -2;
const long KafkaPartitionSplit::COMMITTED_OFFSET = -3;

const std::set<long> KafkaPartitionSplit::VALID_STARTING_OFFSET_MARKERS = {
    KafkaPartitionSplit::EARLIEST_OFFSET, KafkaPartitionSplit::LATEST_OFFSET, KafkaPartitionSplit::COMMITTED_OFFSET};

const std::set<long> KafkaPartitionSplit::VALID_STOPPING_OFFSET_MARKERS = {
    KafkaPartitionSplit::LATEST_OFFSET, KafkaPartitionSplit::COMMITTED_OFFSET, KafkaPartitionSplit::NO_STOPPING_OFFSET};

KafkaPartitionSplit::KafkaPartitionSplit(const std::shared_ptr<RdKafka::TopicPartition>& tp, long startingOffset)
    : KafkaPartitionSplit(tp, startingOffset, NO_STOPPING_OFFSET)
{
}

KafkaPartitionSplit::KafkaPartitionSplit(
    const std::shared_ptr<RdKafka::TopicPartition>& tp, long startingOffset, long stoppingOffset)
    : tp_(tp),
      startingOffset_(startingOffset),
      stoppingOffset_(stoppingOffset)
{
    verifyInitialOffset(tp, startingOffset, stoppingOffset);
}

std::string KafkaPartitionSplit::getTopic() const
{
    return tp_->topic();
}

int KafkaPartitionSplit::getPartition() const
{
    return tp_->partition();
}

const std::shared_ptr<RdKafka::TopicPartition>& KafkaPartitionSplit::getTopicPartition() const
{
    return tp_;
}

int64_t KafkaPartitionSplit::getStartingOffset() const
{
    return startingOffset_;
}

int64_t KafkaPartitionSplit::getStoppingOffset() const
{
    return stoppingOffset_;
}

std::string KafkaPartitionSplit::splitId() const
{
    return toSplitId(tp_.get());
}

std::string KafkaPartitionSplit::toSplitId(RdKafka::TopicPartition* tp)
{
    return tp->topic() + "-" + std::to_string(tp->partition());
}

void KafkaPartitionSplit::verifyInitialOffset(
    const std::shared_ptr<RdKafka::TopicPartition>& tp, long startingOffset, long stoppingOffset)
{
    if (startingOffset < 0 &&
        VALID_STARTING_OFFSET_MARKERS.find(startingOffset) == VALID_STARTING_OFFSET_MARKERS.end()) {
        std::ostringstream oss;
        oss << "Invalid starting offset " << startingOffset << " is specified for partition "
            << tp->topic() + "-" + std::to_string(tp->partition())
            << ". It should either be non-negative or be one of the [" << EARLIEST_OFFSET << "(earliest), "
            << LATEST_OFFSET << "(latest), " << COMMITTED_OFFSET << "(committed)].";
        throw std::runtime_error(oss.str());
    }

    if (stoppingOffset < 0 &&
        VALID_STOPPING_OFFSET_MARKERS.find(stoppingOffset) == VALID_STOPPING_OFFSET_MARKERS.end()) {
        std::ostringstream oss;
        oss << "Illegal stopping offset " << stoppingOffset << " is specified for partition "
            << tp->topic() + "-" + std::to_string(tp->partition())
            << ". It should either be non-negative or be one of the [" << LATEST_OFFSET << "(latest), "
            << COMMITTED_OFFSET << "(committed), " << NO_STOPPING_OFFSET << "(Long.MIN_VALUE, no_stopping_offset)].";
        throw std::runtime_error(oss.str());
    }
}