/*
 * 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.ml.integration;

import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
import org.elasticsearch.action.support.IndicesOptions;
import org.elasticsearch.client.internal.OriginSettingClient;
import org.elasticsearch.cluster.metadata.IndexMetadata;
import org.elasticsearch.cluster.routing.OperationRouting;
import org.elasticsearch.cluster.service.ClusterApplierService;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.cluster.service.MasterService;
import org.elasticsearch.common.settings.ClusterSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.indices.TestIndexNameExpressionResolver;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.xpack.core.ClientHelper;
import org.elasticsearch.xpack.core.action.util.QueryPage;
import org.elasticsearch.xpack.core.ml.action.DeleteJobAction;
import org.elasticsearch.xpack.core.ml.action.OpenJobAction;
import org.elasticsearch.xpack.core.ml.action.PutJobAction;
import org.elasticsearch.xpack.core.ml.job.config.AnalysisConfig;
import org.elasticsearch.xpack.core.ml.job.config.Job;
import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndex;
import org.elasticsearch.xpack.core.ml.job.results.Bucket;
import org.elasticsearch.xpack.ml.inference.ingest.InferenceProcessor;
import org.elasticsearch.xpack.ml.job.persistence.BucketsQueryBuilder;
import org.elasticsearch.xpack.ml.job.persistence.JobResultsPersister;
import org.elasticsearch.xpack.ml.job.persistence.JobResultsProvider;
import org.elasticsearch.xpack.ml.support.BaseMlIntegTestCase;
import org.elasticsearch.xpack.ml.utils.persistence.ResultsPersisterService;
import org.junit.Before;

import java.util.Arrays;
import java.util.Date;
import java.util.HashSet;
import java.util.concurrent.atomic.AtomicReference;

import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.nullValue;

/**
 * Test that ML does not touch unnecessary indices when removing job index aliases
 */
public class JobStorageDeletionTaskIT extends BaseMlIntegTestCase {

    private static final long bucketSpan = AnalysisConfig.Builder.DEFAULT_BUCKET_SPAN.getMillis();
    private static final String UNRELATED_INDEX = "unrelated-data";

    private JobResultsProvider jobResultsProvider;
    private JobResultsPersister jobResultsPersister;

    @Before
    public void createComponents() {
        Settings settings = nodeSettings(0, Settings.EMPTY);
        ThreadPool tp = mockThreadPool();
        ClusterSettings clusterSettings = new ClusterSettings(
            settings,
            new HashSet<>(
                Arrays.asList(
                    InferenceProcessor.MAX_INFERENCE_PROCESSORS,
                    MasterService.MASTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING,
                    ResultsPersisterService.PERSIST_RESULTS_MAX_RETRIES,
                    OperationRouting.USE_ADAPTIVE_REPLICA_SELECTION_SETTING,
                    ClusterService.USER_DEFINED_METADATA,
                    ClusterApplierService.CLUSTER_SERVICE_SLOW_TASK_LOGGING_THRESHOLD_SETTING
                )
            )
        );
        ClusterService clusterService = new ClusterService(settings, clusterSettings, tp, null);
        OriginSettingClient originSettingClient = new OriginSettingClient(client(), ClientHelper.ML_ORIGIN);
        ResultsPersisterService resultsPersisterService = new ResultsPersisterService(tp, originSettingClient, clusterService, settings);
        jobResultsProvider = new JobResultsProvider(client(), settings, TestIndexNameExpressionResolver.newInstance());
        jobResultsPersister = new JobResultsPersister(originSettingClient, resultsPersisterService);
    }

    public void testUnrelatedIndexNotTouched() throws Exception {
        internalCluster().ensureAtLeastNumDataNodes(1);
        ensureStableCluster(1);

        client().admin().indices().prepareCreate(UNRELATED_INDEX).get();

        enableIndexBlock(UNRELATED_INDEX, IndexMetadata.SETTING_READ_ONLY);

        Job.Builder job = createJob("delete-aliases-test-job", ByteSizeValue.ofMb(2));
        PutJobAction.Request putJobRequest = new PutJobAction.Request(job);
        client().execute(PutJobAction.INSTANCE, putJobRequest).actionGet();

        OpenJobAction.Request openJobRequest = new OpenJobAction.Request(job.getId());
        client().execute(OpenJobAction.INSTANCE, openJobRequest).actionGet();
        awaitJobOpenedAndAssigned(job.getId(), null);

        DeleteJobAction.Request deleteJobRequest = new DeleteJobAction.Request(job.getId());
        deleteJobRequest.setForce(true);
        client().execute(DeleteJobAction.INSTANCE, deleteJobRequest).actionGet();

        // If the deletion of aliases touches the unrelated index with the block
        // then the line above will throw a ClusterBlockException

        disableIndexBlock(UNRELATED_INDEX, IndexMetadata.SETTING_READ_ONLY);
    }

    public void testDeleteDedicatedJobWithDataInShared() throws Exception {
        internalCluster().ensureAtLeastNumDataNodes(1);
        ensureStableCluster(1);
        String jobIdDedicated = "delete-test-job-dedicated";

        Job.Builder job = createJob(jobIdDedicated, ByteSizeValue.ofMb(2)).setResultsIndexName(jobIdDedicated);
        client().execute(PutJobAction.INSTANCE, new PutJobAction.Request(job)).actionGet();
        client().execute(OpenJobAction.INSTANCE, new OpenJobAction.Request(job.getId())).actionGet();
        String dedicatedIndex = job.build().getInitialResultsIndexName();
        awaitJobOpenedAndAssigned(job.getId(), null);
        createBuckets(jobIdDedicated, 1, 10);

        String jobIdShared = "delete-test-job-shared";
        job = createJob(jobIdShared, ByteSizeValue.ofMb(2));
        client().execute(PutJobAction.INSTANCE, new PutJobAction.Request(job)).actionGet();
        client().execute(OpenJobAction.INSTANCE, new OpenJobAction.Request(job.getId())).actionGet();
        awaitJobOpenedAndAssigned(job.getId(), null);
        createBuckets(jobIdShared, 1, 10);

        // Manually switching over alias info
        IndicesAliasesRequest aliasesRequest = new IndicesAliasesRequest().addAliasAction(
            IndicesAliasesRequest.AliasActions.add()
                .alias(AnomalyDetectorsIndex.jobResultsAliasedName(jobIdDedicated))
                .isHidden(true)
                .index(AnomalyDetectorsIndex.jobResultsIndexPrefix() + "shared")
                .writeIndex(false)
                .filter(QueryBuilders.boolQuery().filter(QueryBuilders.termQuery(Job.ID.getPreferredName(), jobIdDedicated)))
        )
            .addAliasAction(
                IndicesAliasesRequest.AliasActions.add()
                    .alias(AnomalyDetectorsIndex.resultsWriteAlias(jobIdDedicated))
                    .index(AnomalyDetectorsIndex.jobResultsIndexPrefix() + "shared")
                    .isHidden(true)
                    .writeIndex(true)
            )
            .addAliasAction(
                IndicesAliasesRequest.AliasActions.remove()
                    .alias(AnomalyDetectorsIndex.resultsWriteAlias(jobIdDedicated))
                    .index(dedicatedIndex)
            );

        client().admin().indices().aliases(aliasesRequest).actionGet();

        createBuckets(jobIdDedicated, 11, 10);
        client().admin().indices().prepareRefresh(AnomalyDetectorsIndex.jobResultsIndexPrefix() + "*").get();
        AtomicReference<QueryPage<Bucket>> bucketHandler = new AtomicReference<>();
        AtomicReference<Exception> failureHandler = new AtomicReference<>();
        blockingCall(
            listener -> jobResultsProvider.buckets(
                jobIdDedicated,
                new BucketsQueryBuilder().from(0).size(22),
                listener::onResponse,
                listener::onFailure,
                client()
            ),
            bucketHandler,
            failureHandler
        );
        assertThat(failureHandler.get(), is(nullValue()));
        assertThat(bucketHandler.get().count(), equalTo(22L));

        DeleteJobAction.Request deleteJobRequest = new DeleteJobAction.Request(jobIdDedicated);
        deleteJobRequest.setForce(true);
        client().execute(DeleteJobAction.INSTANCE, deleteJobRequest).get();

        client().admin().indices().prepareRefresh(AnomalyDetectorsIndex.jobResultsIndexPrefix() + "*").get();
        // Make sure our shared index job is OK
        bucketHandler = new AtomicReference<>();
        failureHandler = new AtomicReference<>();
        blockingCall(
            listener -> jobResultsProvider.buckets(
                jobIdShared,
                new BucketsQueryBuilder().from(0).size(21),
                listener::onResponse,
                listener::onFailure,
                client()
            ),
            bucketHandler,
            failureHandler
        );
        assertThat(failureHandler.get(), is(nullValue()));
        assertThat(bucketHandler.get().count(), equalTo(11L));

        // Make sure dedicated index is gone
        assertThat(
            indicesAdmin().prepareGetIndex()
                .setIndices(dedicatedIndex)
                .setIndicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN_CLOSED_HIDDEN)
                .get()
                .indices().length,
            equalTo(0)
        );

        // Make sure all results referencing the dedicated job are gone
        assertThat(
            client().prepareSearch()
                .setIndices(AnomalyDetectorsIndex.jobResultsIndexPrefix() + "*")
                .setIndicesOptions(IndicesOptions.lenientExpandOpenHidden())
                .setTrackTotalHits(true)
                .setSize(0)
                .setSource(
                    SearchSourceBuilder.searchSource()
                        .query(QueryBuilders.boolQuery().filter(QueryBuilders.termQuery(Job.ID.getPreferredName(), jobIdDedicated)))
                )
                .get()
                .getHits()
                .getTotalHits().value,
            equalTo(0L)
        );
    }

    private void createBuckets(String jobId, int from, int count) {
        JobResultsPersister.Builder builder = jobResultsPersister.bulkPersisterBuilder(jobId);
        for (int i = from; i <= count + from; ++i) {
            Bucket bucket = new Bucket(jobId, new Date(bucketSpan * i), bucketSpan);
            builder.persistBucket(bucket);
        }
        builder.executeRequest();
    }

}
