package org.infinispan.persistence.jdbc.stringbased;

import org.infinispan.Cache;
import org.infinispan.configuration.cache.Configuration;
import org.infinispan.configuration.cache.ConfigurationBuilder;
import org.infinispan.configuration.global.GlobalConfiguration;
import org.infinispan.configuration.global.GlobalConfigurationBuilder;
import org.infinispan.context.Flag;
import org.infinispan.eviction.EvictionType;
import org.infinispan.manager.DefaultCacheManager;
import org.infinispan.persistence.jdbc.configuration.JdbcStringBasedStoreConfigurationBuilder;
import org.infinispan.persistence.jdbc.configuration.TableManipulationConfiguration;
import org.infinispan.persistence.jdbc.connectionfactory.ConnectionFactory;
import org.infinispan.persistence.jdbc.impl.table.TableManager;
import org.infinispan.persistence.jdbc.impl.table.TableManagerFactory;
import org.infinispan.test.TestingUtil;
import org.infinispan.test.fwk.TestCacheManagerFactory;
import org.testng.annotations.Test;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;

import static org.infinispan.test.fwk.UnitTestDatabaseManager.buildTableManipulation;
import static org.testng.AssertJUnit.assertEquals;
import static org.testng.AssertJUnit.assertNotNull;
import static org.testng.AssertJUnit.assertNull;


@Test(groups = "functional", testName = "persistence.jdbc.stringbased.AbstractStringBasedCacheStore")
public abstract class AbstractStringBasedCacheStore {

   protected DefaultCacheManager dcm;
   protected ConnectionFactory connectionFactory;
   protected TableManager tableManager;
   protected TableManipulationConfiguration tableConfiguration;
   protected Cache<String, String> cache;


   @Test
   public void testPutGetRemoveWithoutPassivationWithPreload() throws Exception {
      try {
         dcm = configureCacheManager(false, true, false);
         cache = dcm.getCache();
         tableManager.setCacheName(cache.getName());
         assertCleanCacheAndStore(cache);

         cache.put("k1", "v1");
         cache.put("k2", "v2");

         // test passivation==false, database should contain all entries which are in the cache
         assertNotNull(getValueByKey("k1"));
         assertNotNull(getValueByKey("k2"));

         cache.stop();
         cache.start();

         assertNotNull(getValueByKey("k1"));
         assertNotNull(getValueByKey("k2"));
         assertEquals("v1", cache.get("k1"));
         assertEquals("v2", cache.get("k2"));
         // when the entry is removed from the cache, it should be also removed from the cache store (the store
         // and the cache are the same sets of keys)
         cache.remove("k1");
         assertNull(cache.get("k1"));
         assertNull(getValueByKey("k1"));
      } finally {
         TestingUtil.killCacheManagers(dcm);
      }
   }

   @Test
   public void testPutGetRemoveWithPassivationWithoutPreload() throws Exception {
      try {
         dcm = configureCacheManager(true, false, true);
         cache = dcm.getCache();
         tableManager.setCacheName(cache.getName());
         assertCleanCacheAndStore(cache);

         cache.put("k1", "v1");
         cache.put("k2", "v2");
         //not yet in store (eviction.max-entries=2)
         assertNull(getValueByKey("k1"));
         assertNull(getValueByKey("k2"));
         cache.put("k3", "v3");
         assertEquals("v3", cache.get("k3"));
         //now some key is evicted and stored in store
         assertEquals(2, cache.getAdvancedCache().withFlags(Flag.SKIP_CACHE_LOAD).size());
         assertEquals(1, getAllRows().size());
         //retrieve from store to cache and remove from store, another key must be evicted
         cache.get("k1");
         cache.get("k2");
         assertEquals(2, cache.getAdvancedCache().withFlags(Flag.SKIP_CACHE_LOAD).size());
         assertEquals(1,  getAllRows().size());

         cache.stop();
         cache.start();

         assertEquals(0, cache.getAdvancedCache().withFlags(Flag.SKIP_CACHE_LOAD).size());
         // test purge==false, entries should remain in the database after restart
         assertEquals(3, getAllRows().size());
         assertNotNull(getValueByKey("k1"));
         assertEquals("v1", cache.get("k1"));

         cache.remove("k1");
         assertNull(cache.get("k1"));
         assertNull(getValueByKey("k1"));
      } finally {
         TestingUtil.killCacheManagers(dcm);
      }
   }

   public DefaultCacheManager configureCacheManager(boolean passivation, boolean preload, boolean eviction) throws Exception {
      GlobalConfiguration glob = new GlobalConfigurationBuilder().nonClusteredDefault().build();
      ConfigurationBuilder builder = TestCacheManagerFactory.getDefaultCacheConfiguration(false);
      JdbcStringBasedStoreConfigurationBuilder storeBuilder = builder
              .persistence()
              .passivation(passivation)
              .addStore(JdbcStringBasedStoreConfigurationBuilder.class)
              .preload(preload);

      connectionFactory = getConnectionFactory(storeBuilder);
      buildTableManipulation(storeBuilder.table());

      tableConfiguration = storeBuilder.create().table();
      tableManager = TableManagerFactory.getManager(connectionFactory, storeBuilder.create());

      if (eviction) {
         builder.memory().evictionType(EvictionType.COUNT).size(2);
      } else {
         builder.memory().evictionType(EvictionType.COUNT).size(-1);
      }

      Configuration configuration = builder.build();

      return new DefaultCacheManager(glob, configuration, true);
   }

   protected abstract ConnectionFactory getConnectionFactory(JdbcStringBasedStoreConfigurationBuilder storeBuilder);

   protected void assertCleanCacheAndStore(Cache cache) throws Exception {
      cache.clear();
      deleteAllRows();
      assertEquals(0, cache.size());
      assertNull(getValueByKey("k1"));
      assertNull(getValueByKey("k2"));
   }

   public Object getValueByKey(String key) throws Exception {
      Object result = null;
      Connection connection = connectionFactory.getConnection();
         PreparedStatement ps = connection.prepareStatement(tableManager.getSelectIdRowSql());
         ps.setString(1, key);
         ResultSet rs = ps.executeQuery();
         if (rs.next()) {
            result = rs.getObject(tableConfiguration.idColumnName()); //start from 1, not 0
         }
         connectionFactory.releaseConnection(connection);

      return result;

   }

   private List<String> getAllRows() throws Exception {
      Connection connection = connectionFactory.getConnection();
      Statement s = connection.createStatement();
      ResultSet rs = s.executeQuery(tableManager.getLoadAllRowsSql());
      List<String> rows = new ArrayList<String>();
      while (rs.next()) {
         rows.add(rs.toString());
      }
      connectionFactory.releaseConnection(connection);
      return rows;
   }

   private void deleteAllRows() throws Exception {
      Connection connection = connectionFactory.getConnection();
      Statement s = connection.createStatement();
      s.executeUpdate(tableManager.getDeleteAllRowsSql());
      connectionFactory.releaseConnection(connection);
   }
}
