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

import org.elasticsearch.Version;
import org.elasticsearch.client.Request;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.ml.job.config.AnalysisConfig;
import org.elasticsearch.client.ml.job.config.DataDescription;
import org.elasticsearch.client.ml.job.config.Detector;
import org.elasticsearch.client.ml.job.config.Job;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndex;
import org.elasticsearch.xpack.test.rest.XPackRestTestHelper;

import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class MlMappingsUpgradeIT extends AbstractUpgradeTestCase {

    private static final String JOB_ID = "ml-mappings-upgrade-job";

    @Override
    protected Collection<String> templatesToWaitFor() {
        return Stream.concat(XPackRestTestHelper.ML_POST_V660_TEMPLATES.stream(),
            super.templatesToWaitFor().stream()).collect(Collectors.toSet());
    }

    /**
     * The purpose of this test is to ensure that when a job is open through a rolling upgrade we upgrade the results
     * index mappings when it is assigned to an upgraded node even if no other ML endpoint is called after the upgrade
     */
    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/37763")
    public void testMappingsUpgrade() throws Exception {

        switch (CLUSTER_TYPE) {
            case OLD:
                createAndOpenTestJob();
                break;
            case MIXED:
                // We don't know whether the job is on an old or upgraded node, so cannot assert that the mappings have been upgraded
                break;
            case UPGRADED:
                assertUpgradedMappings();
                break;
            default:
                throw new UnsupportedOperationException("Unknown cluster type [" + CLUSTER_TYPE + "]");
        }
    }

    private void createAndOpenTestJob() throws IOException {

        Detector.Builder d = new Detector.Builder("metric", "responsetime");
        d.setByFieldName("airline");
        AnalysisConfig.Builder analysisConfig = new AnalysisConfig.Builder(Collections.singletonList(d.build()));
        analysisConfig.setBucketSpan(TimeValue.timeValueMinutes(10));
        Job.Builder job = new Job.Builder(JOB_ID);
        job.setAnalysisConfig(analysisConfig);
        job.setDataDescription(new DataDescription.Builder());

        Request putJob = new Request("PUT", "_ml/anomaly_detectors/" + JOB_ID);
        putJob.setJsonEntity(Strings.toString(job.build()));
        Response response = client().performRequest(putJob);
        assertEquals(200, response.getStatusLine().getStatusCode());

        Request openJob = new Request("POST", "_ml/anomaly_detectors/" + JOB_ID + "/_open");
        response = client().performRequest(openJob);
        assertEquals(200, response.getStatusLine().getStatusCode());
    }

    @SuppressWarnings("unchecked")
    private void assertUpgradedMappings() throws Exception {

        assertBusy(() -> {
            Request getMappings = new Request("GET", AnomalyDetectorsIndex.resultsWriteAlias(JOB_ID) + "/_mappings");
            Response response = client().performRequest(getMappings);

            Map<String, Object> responseLevel = entityAsMap(response);
            assertNotNull(responseLevel);
            Map<String, Object> indexLevel = (Map<String, Object>) responseLevel.get(".ml-anomalies-shared");
            assertNotNull(indexLevel);
            Map<String, Object> mappingsLevel = (Map<String, Object>) indexLevel.get("mappings");
            assertNotNull(mappingsLevel);
            Map<String, Object> metaLevel = (Map<String, Object>) mappingsLevel.get("_meta");
            assertEquals(Collections.singletonMap("version", Version.CURRENT.toString()), metaLevel);
            Map<String, Object> propertiesLevel = (Map<String, Object>) mappingsLevel.get("properties");
            assertNotNull(propertiesLevel);
            // TODO: as the years go by, the field we assert on here should be changed
            // to the most recent field we've added that is NOT of type "keyword"
            Map<String, Object> fieldLevel = (Map<String, Object>) propertiesLevel.get("multi_bucket_impact");
            assertEquals(Collections.singletonMap("type", "double"), fieldLevel);
        });
    }
}

@@ -1660,89 +1660,8 @@
  */%0A
-    @AwaitsFix(bugUrl = %22https://github.com/elastic/elasticsearch/issues/37763%22)%0A
     
@@ -1705,32 +1705,32 @@
 ows Exception %7B%0A
+
 %0A        switch 
@@ -2762,24 +2762,177 @@
 .Builder());
+%0A        // Use a custom index because other rolling upgrade tests meddle with the shared index%0A        job.setResultsIndexName(%22mappings-upgrade-test%22);
 %0A%0A        Re
@@ -3878,36 +3878,379 @@
 t%3E indexLevel = 
-(Map
+null;%0A            // The name of the concrete index underlying the results index alias may or may not have been changed%0A            // by the upgrade process (depending on what other tests are being run and the order they're run in),%0A            // so navigating to the next level of the tree must account for both cases%0A            for (Map.Entry
 %3CString, Object%3E
@@ -4249,17 +4249,24 @@
  Object%3E
-)
+ entry :
  respons
@@ -4276,36 +4276,272 @@
 vel.
-get(%22.ml-anomalies-shared%22);
+entrySet()) %7B%0A                if (entry.getKey().startsWith(%22.ml-anomalies-%22) && entry.getKey().contains(%22mappings-upgrade-test%22)) %7B%0A                    indexLevel = (Map%3CString, Object%3E) entry.getValue();%0A                    break;%0A                %7D%0A            %7D
 %0A   

Fix MlMappingsUpgradeIT testMappingsUpgrade (#37769)