/*
 * 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 and the Server Side Public License, v 1; you may not use this file except
 * in compliance with, at your election, the Elastic License 2.0 or the Server
 * Side Public License, v 1.
 */

package org.elasticsearch.datastreams.lifecycle;

import org.elasticsearch.action.datastreams.CreateDataStreamAction;
import org.elasticsearch.cluster.metadata.DataStreamLifecycle;
import org.elasticsearch.cluster.metadata.Template;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.datastreams.DataStreamsPlugin;
import org.elasticsearch.datastreams.lifecycle.action.DeleteDataStreamLifecycleAction;
import org.elasticsearch.datastreams.lifecycle.action.GetDataStreamLifecycleAction;
import org.elasticsearch.datastreams.lifecycle.action.PutDataStreamLifecycleAction;
import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.test.ESIntegTestCase;
import org.elasticsearch.test.transport.MockTransportService;

import java.util.Collection;
import java.util.List;

import static org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleFixtures.putComposableIndexTemplate;
import static org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleFixtures.randomLifecycle;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.nullValue;

public class CrudDataStreamLifecycleIT extends ESIntegTestCase {

    @Override
    protected Collection<Class<? extends Plugin>> nodePlugins() {
        return List.of(DataStreamsPlugin.class, MockTransportService.TestPlugin.class);
    }

    protected boolean ignoreExternalCluster() {
        return true;
    }

    public void testGetLifecycle() throws Exception {
        DataStreamLifecycle lifecycle = randomLifecycle();
        putComposableIndexTemplate("id1", null, List.of("with-lifecycle*"), null, null, lifecycle);
        putComposableIndexTemplate("id2", null, List.of("without-lifecycle*"), null, null, null);
        {
            String dataStreamName = "with-lifecycle-1";
            CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(dataStreamName);
            client().execute(CreateDataStreamAction.INSTANCE, createDataStreamRequest).get();
        }
        {
            String dataStreamName = "with-lifecycle-2";
            CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(dataStreamName);
            client().execute(CreateDataStreamAction.INSTANCE, createDataStreamRequest).get();
        }
        {
            String dataStreamName = "without-lifecycle";
            CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(dataStreamName);
            client().execute(CreateDataStreamAction.INSTANCE, createDataStreamRequest).get();
        }

        // Test retrieving all lifecycles
        {
            GetDataStreamLifecycleAction.Request getDataLifecycleRequest = new GetDataStreamLifecycleAction.Request(new String[] { "*" });
            GetDataStreamLifecycleAction.Response response = client().execute(
                GetDataStreamLifecycleAction.INSTANCE,
                getDataLifecycleRequest
            ).get();
            assertThat(response.getDataStreamLifecycles().size(), equalTo(3));
            assertThat(response.getDataStreamLifecycles().get(0).dataStreamName(), equalTo("with-lifecycle-1"));
            assertDataLifecycle(response.getDataStreamLifecycles().get(0).lifecycle(), lifecycle);
            assertThat(response.getDataStreamLifecycles().get(1).dataStreamName(), equalTo("with-lifecycle-2"));
            assertDataLifecycle(response.getDataStreamLifecycles().get(1).lifecycle(), lifecycle);
            assertThat(response.getDataStreamLifecycles().get(2).dataStreamName(), equalTo("without-lifecycle"));
            assertThat(response.getDataStreamLifecycles().get(2).lifecycle(), nullValue());
            assertThat(response.getRolloverConfiguration(), nullValue());
        }

        // Test retrieving all lifecycles prefixed wildcard
        {
            GetDataStreamLifecycleAction.Request getDataLifecycleRequest = new GetDataStreamLifecycleAction.Request(
                new String[] { "with-lifecycle*" }
            );
            GetDataStreamLifecycleAction.Response response = client().execute(
                GetDataStreamLifecycleAction.INSTANCE,
                getDataLifecycleRequest
            ).get();
            assertThat(response.getDataStreamLifecycles().size(), equalTo(2));
            assertThat(response.getDataStreamLifecycles().get(0).dataStreamName(), equalTo("with-lifecycle-1"));
            assertDataLifecycle(response.getDataStreamLifecycles().get(0).lifecycle(), lifecycle);
            assertThat(response.getDataStreamLifecycles().get(1).dataStreamName(), equalTo("with-lifecycle-2"));
            assertDataLifecycle(response.getDataStreamLifecycles().get(1).lifecycle(), lifecycle);
            assertThat(response.getRolloverConfiguration(), nullValue());
        }

        // Test retrieving concrete data streams
        {
            GetDataStreamLifecycleAction.Request getDataLifecycleRequest = new GetDataStreamLifecycleAction.Request(
                new String[] { "with-lifecycle-1", "with-lifecycle-2" }
            );
            GetDataStreamLifecycleAction.Response response = client().execute(
                GetDataStreamLifecycleAction.INSTANCE,
                getDataLifecycleRequest
            ).get();
            assertThat(response.getDataStreamLifecycles().size(), equalTo(2));
            assertThat(response.getDataStreamLifecycles().get(0).dataStreamName(), equalTo("with-lifecycle-1"));
            assertDataLifecycle(response.getDataStreamLifecycles().get(0).lifecycle(), lifecycle);
            assertThat(response.getRolloverConfiguration(), nullValue());
        }

        // Test include defaults
        GetDataStreamLifecycleAction.Request getDataLifecycleRequestWithDefaults = new GetDataStreamLifecycleAction.Request(
            new String[] { "*" }
        ).includeDefaults(true);
        GetDataStreamLifecycleAction.Response responseWithRollover = client().execute(
            GetDataStreamLifecycleAction.INSTANCE,
            getDataLifecycleRequestWithDefaults
        ).get();
        assertThat(responseWithRollover.getDataStreamLifecycles().size(), equalTo(3));
        assertThat(responseWithRollover.getDataStreamLifecycles().get(0).dataStreamName(), equalTo("with-lifecycle-1"));
        assertDataLifecycle(responseWithRollover.getDataStreamLifecycles().get(0).lifecycle(), lifecycle);
        assertThat(responseWithRollover.getDataStreamLifecycles().get(1).dataStreamName(), equalTo("with-lifecycle-2"));
        assertDataLifecycle(responseWithRollover.getDataStreamLifecycles().get(1).lifecycle(), lifecycle);
        assertThat(responseWithRollover.getDataStreamLifecycles().get(2).dataStreamName(), equalTo("without-lifecycle"));
        assertThat(responseWithRollover.getDataStreamLifecycles().get(2).lifecycle(), is(nullValue()));
        assertThat(responseWithRollover.getRolloverConfiguration(), notNullValue());
    }

    private void assertDataLifecycle(DataStreamLifecycle lifecycle, DataStreamLifecycle expected) {
        if (expected.equals(Template.NO_LIFECYCLE)) {
            assertThat(lifecycle, nullValue());
        } else {
            assertThat(lifecycle, equalTo(expected));
        }
    }

    public void testPutLifecycle() throws Exception {
        putComposableIndexTemplate("id1", null, List.of("my-data-stream*"), null, null, null);
        // Create index without a lifecycle
        String dataStreamName = "my-data-stream";
        CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(dataStreamName);
        client().execute(CreateDataStreamAction.INSTANCE, createDataStreamRequest).get();

        {
            GetDataStreamLifecycleAction.Request getDataLifecycleRequest = new GetDataStreamLifecycleAction.Request(
                new String[] { "my-data-stream" }
            );
            GetDataStreamLifecycleAction.Response response = client().execute(
                GetDataStreamLifecycleAction.INSTANCE,
                getDataLifecycleRequest
            ).get();
            assertThat(response.getDataStreamLifecycles().isEmpty(), equalTo(false));
            GetDataStreamLifecycleAction.Response.DataStreamLifecycle dataStreamLifecycle = response.getDataStreamLifecycles().get(0);
            assertThat(dataStreamLifecycle.dataStreamName(), is(dataStreamName));
            assertThat(dataStreamLifecycle.lifecycle(), is(nullValue()));
        }

        // Set lifecycle
        {
            TimeValue dataRetention = randomBoolean() ? null : TimeValue.timeValueMillis(randomMillisUpToYear9999());
            PutDataStreamLifecycleAction.Request putDataLifecycleRequest = new PutDataStreamLifecycleAction.Request(
                new String[] { "*" },
                dataRetention
            );
            assertThat(
                client().execute(PutDataStreamLifecycleAction.INSTANCE, putDataLifecycleRequest).get().isAcknowledged(),
                equalTo(true)
            );
            GetDataStreamLifecycleAction.Request getDataLifecycleRequest = new GetDataStreamLifecycleAction.Request(
                new String[] { "my-data-stream" }
            );
            GetDataStreamLifecycleAction.Response response = client().execute(
                GetDataStreamLifecycleAction.INSTANCE,
                getDataLifecycleRequest
            ).get();
            assertThat(response.getDataStreamLifecycles().size(), equalTo(1));
            assertThat(response.getDataStreamLifecycles().get(0).dataStreamName(), equalTo("my-data-stream"));
            assertThat(response.getDataStreamLifecycles().get(0).lifecycle().getEffectiveDataRetention(), equalTo(dataRetention));
        }
    }

    public void testDeleteLifecycle() throws Exception {
        DataStreamLifecycle lifecycle = new DataStreamLifecycle(randomMillisUpToYear9999());
        putComposableIndexTemplate("id1", null, List.of("with-lifecycle*"), null, null, lifecycle);
        putComposableIndexTemplate("id2", null, List.of("without-lifecycle*"), null, null, null);
        {
            String dataStreamName = "with-lifecycle-1";
            CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(dataStreamName);
            client().execute(CreateDataStreamAction.INSTANCE, createDataStreamRequest).get();
        }
        {
            String dataStreamName = "with-lifecycle-2";
            CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(dataStreamName);
            client().execute(CreateDataStreamAction.INSTANCE, createDataStreamRequest).get();
        }
        {
            String dataStreamName = "with-lifecycle-3";
            CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(dataStreamName);
            client().execute(CreateDataStreamAction.INSTANCE, createDataStreamRequest).get();
        }

        // Verify that we have 3 data streams with lifecycles
        {
            GetDataStreamLifecycleAction.Request getDataLifecycleRequest = new GetDataStreamLifecycleAction.Request(
                new String[] { "with-lifecycle*" }
            );
            GetDataStreamLifecycleAction.Response response = client().execute(
                GetDataStreamLifecycleAction.INSTANCE,
                getDataLifecycleRequest
            ).get();
            assertThat(response.getDataStreamLifecycles().size(), equalTo(3));
        }

        // Remove lifecycle from concrete data stream
        {
            DeleteDataStreamLifecycleAction.Request deleteDataLifecycleRequest = new DeleteDataStreamLifecycleAction.Request(
                new String[] { "with-lifecycle-1" }
            );
            assertThat(
                client().execute(DeleteDataStreamLifecycleAction.INSTANCE, deleteDataLifecycleRequest).get().isAcknowledged(),
                equalTo(true)
            );
            GetDataStreamLifecycleAction.Request getDataLifecycleRequest = new GetDataStreamLifecycleAction.Request(
                new String[] { "with-lifecycle*" }
            );
            GetDataStreamLifecycleAction.Response response = client().execute(
                GetDataStreamLifecycleAction.INSTANCE,
                getDataLifecycleRequest
            ).get();
            assertThat(response.getDataStreamLifecycles().size(), equalTo(3));
            GetDataStreamLifecycleAction.Response.DataStreamLifecycle dataStreamLifecycle = response.getDataStreamLifecycles().get(0);
            assertThat(dataStreamLifecycle.dataStreamName(), is("with-lifecycle-1"));
            assertThat(dataStreamLifecycle.lifecycle(), is(nullValue()));
            assertThat(response.getDataStreamLifecycles().get(1).dataStreamName(), equalTo("with-lifecycle-2"));
            assertThat(response.getDataStreamLifecycles().get(2).dataStreamName(), equalTo("with-lifecycle-3"));

        }

        // Remove lifecycle from all data streams
        {
            DeleteDataStreamLifecycleAction.Request deleteDataLifecycleRequest = new DeleteDataStreamLifecycleAction.Request(
                new String[] { "*" }
            );
            assertThat(
                client().execute(DeleteDataStreamLifecycleAction.INSTANCE, deleteDataLifecycleRequest).get().isAcknowledged(),
                equalTo(true)
            );
            GetDataStreamLifecycleAction.Request getDataLifecycleRequest = new GetDataStreamLifecycleAction.Request(
                new String[] { "with-lifecycle*" }
            );
            GetDataStreamLifecycleAction.Response response = client().execute(
                GetDataStreamLifecycleAction.INSTANCE,
                getDataLifecycleRequest
            ).get();
            assertThat(response.getDataStreamLifecycles().size(), equalTo(3));
            assertThat(response.getDataStreamLifecycles().get(0).dataStreamName(), equalTo("with-lifecycle-1"));
            assertThat(response.getDataStreamLifecycles().get(1).dataStreamName(), equalTo("with-lifecycle-2"));
            assertThat(response.getDataStreamLifecycles().get(2).dataStreamName(), equalTo("with-lifecycle-3"));

            for (GetDataStreamLifecycleAction.Response.DataStreamLifecycle dataStreamLifecycle : response.getDataStreamLifecycles()) {
                assertThat(dataStreamLifecycle.lifecycle(), nullValue());
            }
        }
    }
}
