/*
 * 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.
 */
package org.apache.kafka.coordinator.group;

import org.apache.kafka.clients.consumer.internals.AutoOffsetResetStrategy;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigException;

import java.util.Arrays;
import java.util.Locale;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * Represents the strategy for resetting offsets in share consumer groups when no previous offset is found
 * for a partition or when an offset is out of range.
 *
 * Supports three strategies:
 * <ul>
 *   <li>{@code EARLIEST} - Reset the offset to the earliest available offset
 *   <li>{@code LATEST} - Reset the offset to the latest available offset
 *   <li>{@code BY_DURATION} - Reset the offset to a timestamp that is the specified duration before the current time
 * </ul>
 */
public class ShareGroupAutoOffsetResetStrategy {

    public static final ShareGroupAutoOffsetResetStrategy EARLIEST = new ShareGroupAutoOffsetResetStrategy(AutoOffsetResetStrategy.EARLIEST, StrategyType.EARLIEST);
    public static final ShareGroupAutoOffsetResetStrategy LATEST = new ShareGroupAutoOffsetResetStrategy(AutoOffsetResetStrategy.LATEST, StrategyType.LATEST);

    public enum StrategyType {
        LATEST, EARLIEST, BY_DURATION;

        @Override
        public String toString() {
            return super.toString().toLowerCase(Locale.ROOT);
        }
    }

    private final AutoOffsetResetStrategy delegate;
    private final StrategyType type;

    private ShareGroupAutoOffsetResetStrategy(AutoOffsetResetStrategy delegate, StrategyType type) {
        this.delegate = delegate;
        this.type = type;
    }

    /**
     * Factory method to create a ShareGroupAutoOffsetResetStrategy from a string representation.
     */
    public static ShareGroupAutoOffsetResetStrategy fromString(String offsetStrategy) {
        AutoOffsetResetStrategy baseStrategy = AutoOffsetResetStrategy.fromString(offsetStrategy);
        AutoOffsetResetStrategy.StrategyType baseType = baseStrategy.type();

        StrategyType shareGroupType = switch (baseType) {
            case EARLIEST -> StrategyType.EARLIEST;
            case LATEST -> StrategyType.LATEST;
            case BY_DURATION -> StrategyType.BY_DURATION;
            default -> throw new IllegalArgumentException("Unsupported strategy for ShareGroup: " + baseType);
        };

        return new ShareGroupAutoOffsetResetStrategy(baseStrategy, shareGroupType);
    }

    /**
     * Returns the share group strategy type.
     */
    public StrategyType type() {
        return type;
    }

    /**
     * Returns the name of the share group offset reset strategy.
     */
    public String name() {
        return type.toString();
    }

    /**
     * Delegates the timestamp calculation to the base strategy.
     * @return the timestamp for the OffsetResetStrategy,
     * if the strategy is EARLIEST or LATEST or duration is provided
     * else return Optional.empty()
     */
    public Long timestamp() {
        return delegate.timestamp().get();
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        ShareGroupAutoOffsetResetStrategy that = (ShareGroupAutoOffsetResetStrategy) o;
        return type == that.type && Objects.equals(delegate, that.delegate);
    }

    @Override
    public int hashCode() {
        return Objects.hash(delegate, type);
    }

    @Override
    public String toString() {
        return "ShareGroupAutoOffsetResetStrategy{" +
                "type=" + type +
                ", delegate=" + delegate +
                '}';
    }

    /**
     * Factory method for creating EARLIEST strategy.
     */
    public static ShareGroupAutoOffsetResetStrategy earliest() {
        return new ShareGroupAutoOffsetResetStrategy(AutoOffsetResetStrategy.EARLIEST, StrategyType.EARLIEST);
    }

    /**
     * Factory method for creating LATEST strategy.
     */
    public static ShareGroupAutoOffsetResetStrategy latest() {
        return new ShareGroupAutoOffsetResetStrategy(AutoOffsetResetStrategy.LATEST, StrategyType.LATEST);
    }

    public static class Validator implements ConfigDef.Validator {
        @Override
        public void ensureValid(String name, Object value) {
            String offsetStrategy = (String) value;
            try {
                fromString(offsetStrategy);
            } catch (Exception e) {
                throw new ConfigException(name, value, "Invalid value `" + offsetStrategy + "` for configuration " +
                        name + ". The value must be either 'earliest', 'latest' or of the format 'by_duration:<PnDTnHnMn.nS>'.");
            }
        }

        public String toString() {
            String values = Arrays.stream(StrategyType.values())
                .map(strategyType -> {
                    if (strategyType == StrategyType.BY_DURATION) {
                        return strategyType + ":PnDTnHnMn.nS";
                    }
                    return strategyType.toString();
                }).collect(Collectors.joining(", "));
            return "[" + values + "]";
        }
    }
}
