/*
 * 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.clients.consumer.internals;

import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.requests.ListOffsetsRequest;
import org.apache.kafka.common.utils.Utils;

import java.time.Duration;
import java.time.Instant;
import java.util.Arrays;
import java.util.Locale;
import java.util.Objects;
import java.util.Optional;
import java.util.stream.Collectors;

public class AutoOffsetResetStrategy {
    public enum StrategyType {
        LATEST, EARLIEST, NONE, BY_DURATION;

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

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

    private final StrategyType type;
    private final Optional<Duration> duration;

    private AutoOffsetResetStrategy(StrategyType type) {
        this.type = type;
        this.duration = Optional.empty();
    }

    private AutoOffsetResetStrategy(Duration duration) {
        this.type = StrategyType.BY_DURATION;
        this.duration = Optional.of(duration);
    }

    /**
     *  Returns the AutoOffsetResetStrategy from the given string.
     */
    public static AutoOffsetResetStrategy fromString(String offsetStrategy) {
        if (offsetStrategy == null) {
            throw new IllegalArgumentException("Auto offset reset strategy is null");
        }

        if (StrategyType.BY_DURATION.toString().equals(offsetStrategy)) {
            throw new IllegalArgumentException("<:duration> part is missing in by_duration auto offset reset strategy.");
        }

        if (Arrays.asList(Utils.enumOptions(StrategyType.class)).contains(offsetStrategy)) {
            StrategyType type = StrategyType.valueOf(offsetStrategy.toUpperCase(Locale.ROOT));
            switch (type) {
                case EARLIEST:
                    return EARLIEST;
                case LATEST:
                    return LATEST;
                case NONE:
                    return NONE;
                default:
                    throw new IllegalArgumentException("Unknown auto offset reset strategy: " + offsetStrategy);
            }
        }

        if (offsetStrategy.startsWith(StrategyType.BY_DURATION + ":")) {
            String isoDuration = offsetStrategy.substring(StrategyType.BY_DURATION.toString().length() + 1);
            try {
                Duration duration = Duration.parse(isoDuration);
                if (duration.isNegative()) {
                    throw new IllegalArgumentException("Negative duration is not supported in by_duration offset reset strategy.");
                }
                return new AutoOffsetResetStrategy(duration);
            } catch (Exception e) {
                throw new IllegalArgumentException("Unable to parse duration string in by_duration offset reset strategy.", e);
            }
        }

        throw new IllegalArgumentException("Unknown auto offset reset strategy: " + offsetStrategy);
    }

    /**
     * Returns the offset reset strategy type.
     */
    public StrategyType type() {
        return type;
    }

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

    /**
     * Return the timestamp to be used for the ListOffsetsRequest.
     * @return the timestamp for the OffsetResetStrategy,
     * if the strategy is EARLIEST or LATEST or duration is provided
     * else return Optional.empty()
     */
    public Optional<Long> timestamp() {
        if (type == StrategyType.EARLIEST)
            return Optional.of(ListOffsetsRequest.EARLIEST_TIMESTAMP);
        else if (type == StrategyType.LATEST)
            return Optional.of(ListOffsetsRequest.LATEST_TIMESTAMP);
        else if (type == StrategyType.BY_DURATION && duration.isPresent()) {
            Instant now = Instant.now();
            return Optional.of(now.minus(duration.get()).toEpochMilli());
        } else
            return Optional.empty();
    }

    public Optional<Duration> duration() {
        return duration;
    }

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

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

    @Override
    public String toString() {
        return "AutoOffsetResetStrategy{" +
                "type=" + type +
                (duration.map(value -> ", duration=" + value).orElse("")) +
                '}';
    }

    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', 'none' or of the format 'by_duration:<PnDTnHnMn.nS.>'.");
            }
        }

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