/*
 * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
 * or more contributor license agreements. Licensed under the Elastic License
 * 2.0; you may not use this file except in compliance with the Elastic License
 * 2.0.
 */

package org.elasticsearch.xpack.autoscaling.existence;

import org.elasticsearch.cluster.metadata.IndexMetadata;
import org.elasticsearch.cluster.node.DiscoveryNodeRole;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.settings.Setting;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingCapacity;
import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderContext;
import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderResult;
import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderService;
import org.elasticsearch.xpack.core.ilm.LifecycleExecutionState;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

/**
 * This decider looks at all indices and ensures a minimum capacity is available if any indices are in the frozen ILM phase, since that
 * is designated for partially mounted indices/frozen tier only. Effectively, this scales the tier into existence.
 *
 * This works in concert with the `WaitForDataTierStep` in ILM that ensures we wait for autoscaling to spin up the first frozen tier node.
 */
public class FrozenExistenceDeciderService implements AutoscalingDeciderService {
    public static final String NAME = "frozen_existence";
    static final ByteSizeValue MINIMUM_FROZEN_MEMORY = ByteSizeValue.ofGb(1);
    static final ByteSizeValue MINIMUM_FROZEN_STORAGE = ByteSizeValue.ofGb(8);

    @Override
    public String name() {
        return NAME;
    }

    @Override
    public AutoscalingDeciderResult scale(Settings configuration, AutoscalingDeciderContext context) {
        List<String> indicesNeedingFrozen = StreamSupport.stream(context.state().metadata().spliterator(), false)
            .filter(this::needsTier)
            .map(imd -> imd.getIndex().getName())
            .limit(10)
            .collect(Collectors.toList());
        AutoscalingCapacity.Builder builder = AutoscalingCapacity.builder();
        if (indicesNeedingFrozen.size() > 0) {
            builder.total(MINIMUM_FROZEN_STORAGE, MINIMUM_FROZEN_MEMORY);
            builder.node(MINIMUM_FROZEN_STORAGE, MINIMUM_FROZEN_MEMORY);
        } else {
            builder.total(0L, 0L);
        }

        return new AutoscalingDeciderResult(builder.build(), new FrozenExistenceReason(indicesNeedingFrozen));
    }

    boolean needsTier(IndexMetadata idxMeta) {
        return LifecycleExecutionState.isFrozenPhase(idxMeta);
    }

    @Override
    public List<Setting<?>> deciderSettings() {
        return Collections.emptyList();
    }

    @Override
    public List<DiscoveryNodeRole> roles() {
        return org.elasticsearch.common.collect.List.of(DiscoveryNodeRole.DATA_FROZEN_NODE_ROLE);
    }

    public static class FrozenExistenceReason implements AutoscalingDeciderResult.Reason {
        private final List<String> indices;

        public FrozenExistenceReason(List<String> indices) {
            this.indices = indices;
        }

        public FrozenExistenceReason(StreamInput in) throws IOException {
            this.indices = in.readStringList();
        }

        @Override
        public String summary() {
            return "indices " + indices;
        }

        public List<String> indices() {
            return indices;
        }

        @Override
        public String getWriteableName() {
            return NAME;
        }

        @Override
        public void writeTo(StreamOutput out) throws IOException {
            out.writeStringCollection(indices);
        }

        @Override
        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
            builder.startObject();
            builder.field("indices", indices);
            builder.endObject();
            return builder;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;
            FrozenExistenceReason that = (FrozenExistenceReason) o;
            return indices.equals(that.indices);
        }

        @Override
        public int hashCode() {
            return Objects.hash(indices);
        }
    }

}
