package org.infinispan.distribution;

import org.infinispan.Cache;
import org.infinispan.commands.VisitableCommand;
import org.infinispan.commands.read.GetKeyValueCommand;
import org.infinispan.commands.write.InvalidateL1Command;
import org.infinispan.configuration.cache.ConfigurationBuilder;
import org.infinispan.interceptors.base.CommandInterceptor;
import org.infinispan.interceptors.distribution.L1WriteSynchronizer;
import org.infinispan.statetransfer.StateTransferLock;
import org.infinispan.test.TestingUtil;
import org.infinispan.test.fwk.CheckPoint;
import org.infinispan.transaction.TransactionMode;
import org.infinispan.util.concurrent.IsolationLevel;
import org.junit.Assert;
import org.mockito.AdditionalAnswers;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.testng.annotations.Test;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.Callable;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import static org.junit.Assert.fail;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.withSettings;
import static org.testng.AssertJUnit.assertEquals;
import static org.testng.AssertJUnit.assertNotNull;

/**
 * Base class for various L1 tests for use with distributed cache.  Note these only currently work for synchronous based
 * caches
 *
 * @author wburns
 * @since 6.0
 */
@Test(groups = "functional", testName = "distribution.BaseDistSyncL1Test")
public abstract class BaseDistSyncL1Test extends BaseDistFunctionalTest<Object, String> {

   protected static final String key = "key-to-the-cache";
   protected static final String firstValue = "first-put";
   protected static final String secondValue = "second-put";

   protected IsolationLevel isolationLevel = IsolationLevel.READ_COMMITTED;

   @Override
   protected ConfigurationBuilder buildConfiguration() {
      ConfigurationBuilder builder = super.buildConfiguration();
      builder.locking().isolationLevel(isolationLevel);
      return builder;
   }

   protected void addBlockingInterceptorBeforeTx(Cache<?, ?> cache, final CyclicBarrier barrier,
                                                 Class<? extends VisitableCommand> commandClass) {
      addBlockingInterceptorBeforeTx(cache, barrier, commandClass, true);
   }

   protected void addBlockingInterceptorBeforeTx(Cache<?, ?> cache, final CyclicBarrier barrier,
                                                 Class<? extends VisitableCommand> commandClass,
                                                 boolean blockAfterCommand) {
      addBlockingInterceptor(cache, barrier, commandClass, getDistributionInterceptorClass(), blockAfterCommand);
   }

   protected void addBlockingInterceptor(Cache<?, ?> cache, final CyclicBarrier barrier,
                                         Class<? extends VisitableCommand> commandClass,
                                         Class<? extends CommandInterceptor> interceptorPosition,
                                         boolean blockAfterCommand) {
      cache.getAdvancedCache().addInterceptorBefore(new BlockingInterceptor(barrier, commandClass, blockAfterCommand),
                                                    interceptorPosition);
   }

   protected abstract Class<? extends CommandInterceptor> getDistributionInterceptorClass();

   protected abstract Class<? extends CommandInterceptor> getL1InterceptorClass();

   protected void assertL1StateOnLocalWrite(Cache<?,?> cache, Cache<?, ?> updatingCache, Object key, Object valueWrite) {
      // Default just assumes it invalidated the cache
      assertIsNotInL1(cache, key);
   }

   protected void assertL1GetWithConcurrentUpdate(final Cache<Object, String> nonOwnerCache, Cache<Object, String> ownerCache,
                                                  final Object key, String originalValue, String updateValue)
         throws InterruptedException, ExecutionException, TimeoutException, BrokenBarrierException {
      CyclicBarrier barrier = new CyclicBarrier(2);
      addBlockingInterceptorBeforeTx(nonOwnerCache, barrier, GetKeyValueCommand.class);

      try {
         Future<String> future = nonOwnerCache.getAsync(key);

         // Now wait for the get to return and block it for now
         barrier.await(5, TimeUnit.SECONDS);

         assertEquals(originalValue, ownerCache.put(key, updateValue));

         // Now let owner key->updateValue go through
         barrier.await(5, TimeUnit.SECONDS);

         // This should be originalValue still as we did the get
         assertEquals(originalValue, future.get(5, TimeUnit.SECONDS));

         // Remove the interceptor now since we don't want to block ourselves - if using phaser this isn't required
         removeAllBlockingInterceptorsFromCache(nonOwnerCache);

         assertL1StateOnLocalWrite(nonOwnerCache, ownerCache, key, updateValue);
         // The nonOwnerCache should retrieve new value as it isn't in L1
         assertEquals(updateValue, nonOwnerCache.get(key));
         assertIsInL1(nonOwnerCache, key);
      }
      finally {
         removeAllBlockingInterceptorsFromCache(nonOwnerCache);
      }
   }

   @Test
   public void testNoEntryInL1GetWithConcurrentInvalidation() throws InterruptedException, ExecutionException, TimeoutException, BrokenBarrierException {
      final Cache<Object, String> nonOwnerCache = getFirstNonOwner(key);
      final Cache<Object, String> ownerCache = getFirstOwner(key);

      // Put the first value in the owner, so the L1 is empty
      ownerCache.put(key, firstValue);

      assertL1GetWithConcurrentUpdate(nonOwnerCache, ownerCache, key, firstValue, secondValue);
   }

   @Test
   public void testEntryInL1GetWithConcurrentInvalidation() throws InterruptedException, ExecutionException, TimeoutException, BrokenBarrierException {
      final Cache<Object, String> nonOwnerCache = getFirstNonOwner(key);
      final Cache<Object, String> ownerCache = getFirstOwner(key);

      // Put the first value in a non owner, so the L1 has the key
      ownerCache.put(key, firstValue);
      nonOwnerCache.get(key);

      assertIsInL1(nonOwnerCache, key);

      assertL1GetWithConcurrentUpdate(nonOwnerCache, ownerCache, key, firstValue, secondValue);
   }

   @Test
   public void testEntryInL1GetWithConcurrentPut() throws InterruptedException, ExecutionException, TimeoutException, BrokenBarrierException {
      final Cache<Object, String> nonOwnerCache = getFirstNonOwner(key);
      final Cache<Object, String> ownerCache = getFirstOwner(key);

      // Put the first value in a non owner, so the L1 has the key
      ownerCache.put(key, firstValue);
      nonOwnerCache.get(key);

      assertIsInL1(nonOwnerCache, key);

      assertL1GetWithConcurrentUpdate(nonOwnerCache, nonOwnerCache, key, firstValue, secondValue);
   }

   @Test
   public void testNoEntryInL1GetWithConcurrentPut() throws InterruptedException, ExecutionException, TimeoutException, BrokenBarrierException {
      final Cache<Object, String> nonOwnerCache = getFirstNonOwner(key);
      final Cache<Object, String> ownerCache = getFirstOwner(key);

      // Put the first value in the owner, so the L1 is empty
      ownerCache.put(key, firstValue);

      assertL1GetWithConcurrentUpdate(nonOwnerCache, nonOwnerCache, key, firstValue, secondValue);
   }

   @Test
   public void testNoEntryInL1MultipleConcurrentGetsWithInvalidation() throws TimeoutException, InterruptedException, ExecutionException, BrokenBarrierException {
      final Cache<Object, String> nonOwnerCache = getFirstNonOwner(key);
      final Cache<Object, String> ownerCache = getFirstOwner(key);

      ownerCache.put(key, firstValue);

      CyclicBarrier invalidationBarrier = new CyclicBarrier(2);
      // We want to block right before the invalidation would hit the L1 interceptor to prevent it from invaliding until we want
      nonOwnerCache.getAdvancedCache().addInterceptorBefore(
            new BlockingInterceptor(invalidationBarrier, InvalidateL1Command.class, false), getL1InterceptorClass());

      try {
         assertEquals(firstValue, nonOwnerCache.get(key));

         Future<String> futurePut = ownerCache.putAsync(key, secondValue);

         // Wait for the invalidation to be processing
         invalidationBarrier.await(5, TimeUnit.SECONDS);

         // Now remove the value - assimilates that a get came earlier to owner registering as a invalidatee, however
         // the invalidation blocked the update from going through
         nonOwnerCache.getAdvancedCache().getDataContainer().remove(key);

         // Hack, but we remove the blocking interceptor while a call is in it, it still retains a reference to the next
         // interceptor to invoke and when we unblock it will continue forward
         // This is done because we can't have 2 interceptors of the same class.
         removeAllBlockingInterceptorsFromCache(nonOwnerCache);

         CyclicBarrier getBarrier = new CyclicBarrier(2);
         addBlockingInterceptorBeforeTx(nonOwnerCache, getBarrier, GetKeyValueCommand.class);

         Future<String> futureGet = nonOwnerCache.getAsync(key);

         // Wait for the get to retrieve the remote value but not try to update L1 yet
         getBarrier.await(5, TimeUnit.SECONDS);

         // Let the invalidation unblock now
         invalidationBarrier.await(5, TimeUnit.SECONDS);

         // Wait for the invalidation complete fully
         assertEquals(firstValue, futurePut.get(5, TimeUnit.SECONDS));

         // Now let our get go through
         getBarrier.await(5, TimeUnit.SECONDS);

         // Technically this could be firstValue or secondValue depending on the ordering of if the put has updated
         // it's in memory contents (since the L1 is sent asynchronously with the update) - For Tx this is always
         // firstValue - the point though is to ensure it doesn't write to the L1
         assertNotNull(futureGet.get(5, TimeUnit.SECONDS));

         // Remove the interceptor now since we don't want to block ourselves - if using phaser this isn't required
         removeAllBlockingInterceptorsFromCache(nonOwnerCache);

         // The value shouldn't be in the L1 still
         assertIsNotInL1(nonOwnerCache, key);

         // It is possible that the async L1LastChance will blow away this get, so we have to make sure to check
         // it eventually
         eventually(new Condition() {
            @Override
            public boolean isSatisfied() throws Exception {
               // The nonOwnerCache should retrieve new value as it isn't in L1
               assertEquals(secondValue, nonOwnerCache.get(key));
               return isInL1(nonOwnerCache, key);
            }
         });
      }
      finally {
         removeAllBlockingInterceptorsFromCache(nonOwnerCache);
      }
   }

   /**
    * See ISPN-3657
    */
   @Test
   public void testGetAfterWriteAlreadyInvalidatedCurrentGet() throws InterruptedException, TimeoutException,
                                                                      BrokenBarrierException, ExecutionException {
      final Cache<Object, String> nonOwnerCache = getFirstNonOwner(key);
      final Cache<Object, String> ownerCache = getFirstOwner(key);

      ownerCache.put(key, firstValue);

      CyclicBarrier nonOwnerGetBarrier = new CyclicBarrier(2);
      // We want to block after it retrieves the value from remote owner so the L1 value will be invalidated
      addBlockingInterceptor(nonOwnerCache, nonOwnerGetBarrier, GetKeyValueCommand.class,
                             getDistributionInterceptorClass(), true);

      try {

         Future<String> future = fork(new Callable<String>() {
            @Override
            public String call() throws Exception {
               return nonOwnerCache.get(key);
            }
         });

         // Wait for the get to register L1 before it has sent remote
         nonOwnerGetBarrier.await(10, TimeUnit.SECONDS);

         // Now force the L1 sync to be blown away by an update
         ownerCache.put(key, secondValue);

         assertEquals(secondValue, nonOwnerCache.get(key));

         // It should be in L1 now with the second value
         assertIsInL1(nonOwnerCache, key);
         assertEquals(secondValue, nonOwnerCache.getAdvancedCache().getDataContainer().get(key).getValue());

         // Now let the original get complete
         nonOwnerGetBarrier.await(10, TimeUnit.SECONDS);

         assertEquals(firstValue, future.get(10, TimeUnit.SECONDS));

         // It should STILL be in L1 now with the second value
         assertIsInL1(nonOwnerCache, key);
         assertEquals(secondValue, nonOwnerCache.getAdvancedCache().getDataContainer().get(key).getValue());

      } finally {
         removeAllBlockingInterceptorsFromCache(nonOwnerCache);
      }
   }

   /**
    * See ISPN-3364
    * @throws Throwable
    */
   @Test
   public void testRemoteGetArrivesButWriteOccursBeforeRegistration() throws Throwable {
      final Cache<Object, String>[] owners = getOwners(key, 2);

      final Cache<Object, String> ownerCache = owners[0];
      final Cache<Object, String> backupOwnerCache = owners[1];
      final Cache<Object, String> nonOwnerCache = getFirstNonOwner(key);

      ownerCache.put(key, firstValue);

      assertIsNotInL1(nonOwnerCache, key);

      // Add a barrier to block the owner/backupowner from going further after retrieving the value before coming back into the L1
      // interceptor
      CyclicBarrier getBarrier = new CyclicBarrier(3);
      ownerCache.getAdvancedCache().addInterceptorAfter(new BlockingInterceptor(getBarrier, GetKeyValueCommand.class, true),
                                                        getL1InterceptorClass());
      backupOwnerCache.getAdvancedCache().addInterceptorAfter(new BlockingInterceptor(getBarrier, GetKeyValueCommand.class, true),
                                                       getL1InterceptorClass());

      try {
         Future<String> future = nonOwnerCache.getAsync(key);

         // Wait until get goes remote and retrieves value before going back into L1 interceptor
         getBarrier.await(10, TimeUnit.SECONDS);

         Assert.assertEquals(firstValue, ownerCache.put(key, secondValue));

         // Let the get complete finally
         getBarrier.await(10, TimeUnit.SECONDS);

         final String expectedValue;
         // This is a bit peculiar that depending on the isolation level that a different value is returned.  This is
         // caused due to the fact that the get has retrieved the value from the data container already and placed it
         // in it's context.  With read committed however the value in the context is the same value from the data
         // container.  And thus when the value is updated the context can see the update since it is same reference.
         // Repeatable read however makes a copy of the data and thus doesn't see the value.  That is why read committed
         // returns the new value and repeatable read has the original value
         if (isolationLevel == IsolationLevel.REPEATABLE_READ) {
            expectedValue = firstValue;
         } else {
            expectedValue = secondValue;
         }
         Assert.assertEquals(expectedValue, future.get(10, TimeUnit.SECONDS));

         assertIsNotInL1(nonOwnerCache, key);
      } finally {
         removeAllBlockingInterceptorsFromCache(ownerCache);
         removeAllBlockingInterceptorsFromCache(backupOwnerCache);
      }
   }

   @Test
   public void testGetBlockedInvalidation() throws Throwable {
      final Cache<Object, String> nonOwnerCache = getFirstNonOwner(key);
      final Cache<Object, String> ownerCache = getFirstOwner(key);

      ownerCache.put(key, firstValue);

      assertIsNotInL1(nonOwnerCache, key);

      CheckPoint checkPoint = new CheckPoint();
      waitUntilAboutToAcquireLock(nonOwnerCache, checkPoint);

      log.warn("Doing get here - ignore all previous");

      Future<String> getFuture = nonOwnerCache.getAsync(key);

      // Wait until we are about to write value into data container on non owner
      checkPoint.awaitStrict("pre_acquire_shared_topology_lock_invoked", 10, TimeUnit.SECONDS);

      Future<String> putFuture = ownerCache.putAsync(key, secondValue);

      try {
         putFuture.get(1, TimeUnit.SECONDS);
         fail("Should have thrown a TimeoutException");
      } catch (TimeoutException e) {
      }

      // Let the get complete finally
      checkPoint.triggerForever("pre_acquire_shared_topology_lock_released");

      Assert.assertEquals(firstValue, getFuture.get(10, TimeUnit.SECONDS));

      Assert.assertEquals(firstValue, putFuture.get(10, TimeUnit.SECONDS));

      assertIsNotInL1(nonOwnerCache, key);
   }

   @Test
   public void testGetBlockingAnotherGet() throws Throwable {
      final Cache<Object, String> nonOwnerCache = getFirstNonOwner(key);
      final Cache<Object, String> ownerCache = getFirstOwner(key);

      ownerCache.put(key, firstValue);

      assertIsNotInL1(nonOwnerCache, key);

      CheckPoint checkPoint = new CheckPoint();
      StateTransferLock lock = waitUntilAboutToAcquireLock(nonOwnerCache, checkPoint);

      try {
         log.warn("Doing get here - ignore all previous");

         Future<String> getFuture = nonOwnerCache.getAsync(key);

         // Wait until we are about to write value into data container on non owner
         checkPoint.awaitStrict("pre_acquire_shared_topology_lock_invoked", 10, TimeUnit.SECONDS);

         Future<String> getFuture2 = nonOwnerCache.getAsync(key);

         try {
            getFuture2.get(1, TimeUnit.SECONDS);
            fail("Should have thrown a TimeoutException");
         } catch (TimeoutException e) {
         }

         // Let the get complete finally
         checkPoint.triggerForever("pre_acquire_shared_topology_lock_released");

         Assert.assertEquals(firstValue, getFuture.get(10, TimeUnit.SECONDS));

         Assert.assertEquals(firstValue, getFuture2.get(10, TimeUnit.SECONDS));

         assertIsInL1(nonOwnerCache, key);
      } finally {
         TestingUtil.replaceComponent(nonOwnerCache, StateTransferLock.class, lock, true);
      }
   }

   @Test
   public void testGetBlockingAnotherGetWithMiss() throws Throwable {
      final Cache<Object, String> nonOwnerCache = getFirstNonOwner(key);
      final Cache<Object, String> ownerCache = getFirstOwner(key);

      assertIsNotInL1(nonOwnerCache, key);

      CheckPoint checkPoint = new CheckPoint();
      L1Manager l1Manager = waitUntilL1Registration(nonOwnerCache, checkPoint);

      try {
         log.warn("Doing get here - ignore all previous");

         Future<String> getFuture = nonOwnerCache.getAsync(key);

         // Wait until we are about to write value into data container on non owner
         checkPoint.awaitStrict("pre_acquire_shared_topology_lock_invoked", 10, TimeUnit.SECONDS);

         Future<String> getFuture2 = nonOwnerCache.getAsync(key);

         try {
            getFuture2.get(1, TimeUnit.SECONDS);
            fail("Should have thrown a TimeoutException");
         } catch (TimeoutException e) {
         }

         // Let the get complete finally
         checkPoint.triggerForever("pre_acquire_shared_topology_lock_released");

         Assert.assertNull(getFuture.get(10, TimeUnit.SECONDS));

         Assert.assertNull(getFuture2.get(10, TimeUnit.SECONDS));
      } finally {
         TestingUtil.replaceComponent(nonOwnerCache, L1Manager.class, l1Manager, true);
      }
   }

   @Test
   public void testGetBlockingLocalPut() throws Throwable {
      final Cache<Object, String> nonOwnerCache = getFirstNonOwner(key);
      final Cache<Object, String> ownerCache = getFirstOwner(key);

      ownerCache.put(key, firstValue);

      assertIsNotInL1(nonOwnerCache, key);

      CheckPoint checkPoint = new CheckPoint();
      waitUntilAboutToAcquireLock(nonOwnerCache, checkPoint);

      log.warn("Doing get here - ignore all previous");

      Future<String> getFuture = nonOwnerCache.getAsync(key);

      // Wait until we are about to write value into data container on non owner
      checkPoint.awaitStrict("pre_acquire_shared_topology_lock_invoked", 10, TimeUnit.SECONDS);

      Future<String> putFuture = nonOwnerCache.putAsync(key, secondValue);

      try {
         putFuture.get(1, TimeUnit.SECONDS);
         fail("Should have thrown a TimeoutException");
      } catch (TimeoutException e) {
      }

      // Let the get complete finally
      checkPoint.triggerForever("pre_acquire_shared_topology_lock_released");

      Assert.assertEquals(firstValue, getFuture.get(10, TimeUnit.SECONDS));

      Assert.assertEquals(firstValue, putFuture.get(10, TimeUnit.SECONDS));

      if (nonOwnerCache.getCacheConfiguration().transaction().transactionMode() == TransactionMode.TRANSACTIONAL) {
         assertIsInL1(nonOwnerCache, key);
      } else {
         assertIsNotInL1(nonOwnerCache, key);
      }
   }

   /**
    * Replaces StateTransferLock in cache with a proxy one that will block on
    * {#link StateTransferLock#acquireSharedTopologyLock} until the checkpoint is triggered
    * @param cache The cache to replace the StateTransferLock on
    * @param checkPoint The checkpoint to use to trigger blocking
    * @return The original real StateTransferLock
    */
   protected StateTransferLock waitUntilAboutToAcquireLock(final Cache<?, ?> cache, final CheckPoint checkPoint) {
      StateTransferLock stl = TestingUtil.extractComponent(cache, StateTransferLock.class);
      final Answer<Object> forwardedAnswer = AdditionalAnswers.delegatesTo(stl);
      StateTransferLock mockLock = mock(StateTransferLock.class, withSettings().defaultAnswer(forwardedAnswer));
      doAnswer(new Answer() {
         @Override
         public Object answer(InvocationOnMock invocation) throws Throwable {
            // Wait for main thread to sync up
            checkPoint.trigger("pre_acquire_shared_topology_lock_invoked");
            // Now wait until main thread lets us through
            checkPoint.awaitStrict("pre_acquire_shared_topology_lock_released", 10, TimeUnit.SECONDS);

            return forwardedAnswer.answer(invocation);
         }
      }).when(mockLock).acquireSharedTopologyLock();
      TestingUtil.replaceComponent(cache, StateTransferLock.class, mockLock, true);
      return stl;
   }

   /**
    * Replaces L1Manager in cache with a proxy one that will block on
    * {#link L1Manager#registerL1WriteSynchronizer} until the checkpoint is triggered
    * @param cache The cache to replace the L1Manager on
    * @param checkPoint The checkpoint to use to trigger blocking
    * @return The original real L1Manager
    */
   protected L1Manager waitUntilL1Registration(final Cache<?, ?> cache, final CheckPoint checkPoint) {
      L1Manager l1Manager = TestingUtil.extractComponent(cache, L1Manager.class);
      final Answer<Object> forwardedAnswer = AdditionalAnswers.delegatesTo(l1Manager);
      L1Manager mockL1 = mock(L1Manager.class, withSettings().defaultAnswer(forwardedAnswer));
      doAnswer(new Answer() {
         @Override
         public Object answer(InvocationOnMock invocation) throws Throwable {
            // Wait for main thread to sync up
            checkPoint.trigger("pre_acquire_shared_topology_lock_invoked");
            // Now wait until main thread lets us through
            checkPoint.awaitStrict("pre_acquire_shared_topology_lock_released", 10, TimeUnit.SECONDS);

            return forwardedAnswer.answer(invocation);
         }
      }).when(mockL1).registerL1WriteSynchronizer(Mockito.anyObject(), Mockito.any(L1WriteSynchronizer.class));
      TestingUtil.replaceComponent(cache, L1Manager.class, mockL1, true);
      return l1Manager;
   }
}
