package org.infinispan.server.functional;

import static org.infinispan.client.rest.RestResponse.NOT_FOUND;
import static org.infinispan.client.rest.RestResponse.NO_CONTENT;
import static org.infinispan.client.rest.RestResponse.OK;
import static org.infinispan.server.resp.configuration.RespServerConfiguration.DEFAULT_RESP_CACHE;
import static org.infinispan.server.test.core.Common.assertStatus;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotSame;

import java.io.IOException;
import java.util.List;

import org.infinispan.client.rest.RestCacheClient;
import org.infinispan.client.rest.RestClient;
import org.infinispan.client.rest.RestEntity;
import org.infinispan.commons.dataconversion.MediaType;
import org.infinispan.commons.dataconversion.internal.Json;
import org.infinispan.commons.internal.InternalCacheNames;
import org.infinispan.configuration.cache.CacheMode;
import org.infinispan.configuration.cache.ConfigurationBuilder;
import org.infinispan.persistence.remote.RemoteStore;
import org.infinispan.persistence.remote.configuration.RemoteServerConfiguration;
import org.infinispan.persistence.remote.configuration.RemoteStoreConfiguration;
import org.infinispan.persistence.remote.configuration.RemoteStoreConfigurationBuilder;
import org.infinispan.persistence.remote.upgrade.SerializationUtils;
import org.infinispan.util.KeyValuePair;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

/**
 * @since 13.0
 */
public class ClusterMigrationDynamicStoreIT extends AbstractMultiClusterIT {

   protected static final String CACHE_NAME = "migrate";
   protected static final int ENTRIES = 50;
   static final String PROTOBUF_SCHEMAS_INTERNAL_CACHE_NAME = InternalCacheNames.PROTOBUF_METADATA_CACHE_NAME;

   @BeforeEach
   public void before() {
      // Start two embedded clusters with 2-node each
      startSourceCluster();
      startTargetCluster();

      // Assert clusters are isolated and have 2 members each
      assertEquals(2, source.getMembers().size());
      assertEquals(2, target.getMembers().size());
      assertNotSame(source.getMembers(), target.getMembers());
   }

   @AfterEach
   public void after() throws Exception {
      stopTargetCluster();
      stopSourceCluster();
   }

   @Test
   public void testClusterMigration() throws Exception {
      RestClient restClientSource = source.getClient();
      RestClient restClientTarget = target.getClient();

      // Register proto schema
      addSchema(restClientSource);

      // Create and populate caches in the source cluster
      createSourceClusterCache(CACHE_NAME);
      populateCache(CACHE_NAME, restClientSource);
      createRespSourceCache();
      populateCache(DEFAULT_RESP_CACHE, restClientSource);

      // Migrate the schemas first
      assertSourceDisconnected(PROTOBUF_SCHEMAS_INTERNAL_CACHE_NAME);
      connectTargetCluster(PROTOBUF_SCHEMAS_INTERNAL_CACHE_NAME);
      assertSourceConnected(PROTOBUF_SCHEMAS_INTERNAL_CACHE_NAME);
      assertStatus(NO_CONTENT, restClientTarget.cache(PROTOBUF_SCHEMAS_INTERNAL_CACHE_NAME).exists());
      migrate(PROTOBUF_SCHEMAS_INTERNAL_CACHE_NAME, restClientTarget);

      // Create the cache in the taarget cluster
      createTargetClusterCache(CACHE_NAME);

      // Migrate the cache data
      assertSourceDisconnected(CACHE_NAME);
      connectTargetCluster(CACHE_NAME);
      assertSourceConnected(CACHE_NAME);
      assertEquals("name-13", getPersonName("13", restClientTarget));
      migrate(CACHE_NAME, restClientTarget);
      connectTargetCluster(DEFAULT_RESP_CACHE);
      assertSourceConnected(DEFAULT_RESP_CACHE);
      migrate(DEFAULT_RESP_CACHE, restClientTarget);

      // Do a second migration, should be harmless and simply override the data
      migrate(PROTOBUF_SCHEMAS_INTERNAL_CACHE_NAME, restClientTarget);
      migrate(CACHE_NAME, restClientTarget);

      // Disconnect source from the remote store
      disconnectSource(PROTOBUF_SCHEMAS_INTERNAL_CACHE_NAME, restClientTarget);
      disconnectSource(CACHE_NAME, restClientTarget);
      disconnectSource(DEFAULT_RESP_CACHE, restClientTarget);
      assertSourceDisconnected(PROTOBUF_SCHEMAS_INTERNAL_CACHE_NAME);
      assertSourceDisconnected(CACHE_NAME);
      assertSourceDisconnected(DEFAULT_RESP_CACHE);

      // Stop source cluster
      stopSourceCluster();

      // Assert all nodes are disconnected and data was migrated successfully
      for (int i = 0; i < target.getMembers().size(); i++) {
         RestClient restClient = target.getClient(i);
         assertEquals(ENTRIES, getCacheSize(CACHE_NAME, restClient));
         assertEquals(ENTRIES, getCacheSize(DEFAULT_RESP_CACHE, restClient));
         assertEquals("name-35", getPersonName("35", restClient));
         assertStatus(NO_CONTENT, restClient.cache(PROTOBUF_SCHEMAS_INTERNAL_CACHE_NAME).exists());
      }
   }

   private void createTargetClusterCache(String cacheName) {
      createCache(cacheName, indexedCacheBuilder(), target.getClient());
      createRespCache(target.getClient());
   }

   protected void connectTargetCluster(String cacheName) throws IOException {
      RestCacheClient client = target.getClient().cache(cacheName);
      ConfigurationBuilder builder = new ConfigurationBuilder();
      addRemoteStore(cacheName, builder);

      RemoteStoreConfiguration remoteStore = (RemoteStoreConfiguration) builder.build().persistence().stores().iterator().next();

      RestEntity restEntity = RestEntity.create(MediaType.APPLICATION_JSON, SerializationUtils.toJson(remoteStore));
      assertStatus(NO_CONTENT, client.connectSource(restEntity));

      String json = assertStatus(OK, client.sourceConnection());
      RemoteStoreConfiguration remoteStoreConfiguration = SerializationUtils.fromJson(json);

      List<RemoteServerConfiguration> servers = remoteStoreConfiguration.servers();
      assertEquals(1, servers.size());
      RemoteServerConfiguration initialConfig = remoteStore.servers().iterator().next();
      assertEquals(initialConfig.host(), servers.get(0).host());
      assertEquals(initialConfig.port(), servers.get(0).port());
   }

   private void assertNonExistentCache() {
      String name = "this-cache-doesnt-exist";
      RestCacheClient client = target.getClient().cache(name);
      assertStatus(NOT_FOUND, client.exists());

      ConfigurationBuilder builder = new ConfigurationBuilder();
      addRemoteStore(name, builder);
      RemoteStoreConfiguration remoteStore = (RemoteStoreConfiguration) builder.build().persistence().stores().iterator().next();
      RestEntity restEntity = RestEntity.create(MediaType.APPLICATION_JSON, SerializationUtils.toJson(remoteStore));
      assertStatus(NOT_FOUND, client.connectSource(restEntity));
      assertStatus(NOT_FOUND, client.sourceConnection());
   }

   protected void assertSourceConnected(String cacheName) {
      assertStatus(OK, target.getClient().cache(cacheName).sourceConnected());
   }

   protected void assertSourceDisconnected(String cacheName) {
      assertStatus(NOT_FOUND, target.getClient().cache(cacheName).sourceConnected());
   }

   protected void disconnectSource(String cacheName, RestClient client) {
      assertStatus(NO_CONTENT, client.cache(cacheName).disconnectSource());
   }

   protected void migrate(String cacheName, RestClient client) {
      assertStatus(OK, client.cache(cacheName).synchronizeData());
   }

   protected String getPersonName(String id, RestClient client) {
      String body = assertStatus(OK,client.cache(CACHE_NAME).get(id));
      Json value = Json.read(Json.read(body).at("_value").asString());
      return value.at("name").asString();
   }

   public void populateCache(String cacheName, RestClient client) {
      RestCacheClient cache = client.cache(cacheName);

      for (int i = 0; i < ENTRIES; i++) {
         String person = createPerson("name-" + i);
         assertStatus(NO_CONTENT, cache.put(String.valueOf(i), person));
      }
      assertEquals(ENTRIES, getCacheSize(cacheName, client));
   }

   private String createPerson(String name) {
      return String.format("{\"_type\":\"Person\",\"name\":\"%s\"}", name);
   }

   void addRemoteStore(String cacheName, ConfigurationBuilder builder) {
      RemoteStoreConfigurationBuilder storeConfigurationBuilder = builder.clustering()
            .cacheMode(CacheMode.DIST_SYNC).persistence().addStore(RemoteStoreConfigurationBuilder.class);
      storeConfigurationBuilder
            .remoteCacheName(cacheName)
            .segmented(false)
            .shared(true)
            .addServer()
            .host(source.driver.getServerAddress(0).getHostAddress())
            .port(11222)
            .addProperty(RemoteStore.MIGRATION, "true");
      final KeyValuePair<String, String> credentials = getCredentials();
      if (getCredentials() != null) {
         storeConfigurationBuilder.remoteSecurity()
               .authentication().enable().saslMechanism("PLAIN")
               .username(credentials.getKey())
               .password(credentials.getValue())
               .realm("default");
      }
   }

   void createSourceClusterCache(String cacheName) {
      createCache(cacheName, indexedCacheBuilder(), source.getClient());
   }

   private void createRespSourceCache() {
      createRespCache(source.getClient());
   }

   private void createRespCache(RestClient client) {
      String defaultRespConfiguration = """
{"respCache":{"distributed-cache":{"aliases":["0"],"key-partitioner":"org.infinispan.distribution.ch.impl.RESPHashFunctionPartitioner","mode":"SYNC","statistics":true,"encoding":{"key":{"media-type":"application/octet-stream"},"value":{"media-type":"application/octet-stream"}}}}}
""";
      assertStatus(OK, client.cache(DEFAULT_RESP_CACHE).createWithConfiguration(RestEntity.create(defaultRespConfiguration)));
   }

   ConfigurationBuilder indexedCacheBuilder() {
      ConfigurationBuilder builder = new ConfigurationBuilder();
      builder.clustering().cacheMode(CacheMode.DIST_SYNC);
      builder.indexing().enable().addIndexedEntities("Person");
      return builder;
   }
}
