package org.infinispan.interceptors.locking;

import org.infinispan.manager.EmbeddedCacheManager;
import org.infinispan.metadata.Metadata;
import org.infinispan.commands.FlagAffectedCommand;
import org.infinispan.commands.tx.VersionedPrepareCommand;
import org.infinispan.commands.tx.totalorder.TotalOrderPrepareCommand;
import org.infinispan.commands.write.WriteCommand;
import org.infinispan.commons.util.Immutables;
import org.infinispan.commons.util.InfinispanCollections;
import org.infinispan.configuration.cache.Configuration;
import org.infinispan.container.DataContainer;
import org.infinispan.container.entries.CacheEntry;
import org.infinispan.container.versioning.EntryVersionsMap;
import org.infinispan.container.versioning.VersionGenerator;
import org.infinispan.context.Flag;
import org.infinispan.context.InvocationContext;
import org.infinispan.context.impl.TxInvocationContext;
import org.infinispan.distribution.DistributionManager;
import org.infinispan.factories.annotations.Inject;
import org.infinispan.factories.scopes.Scope;
import org.infinispan.factories.scopes.Scopes;
import org.infinispan.notifications.cachelistener.CacheNotifier;
import org.infinispan.remoting.rpc.RpcManager;
import org.infinispan.remoting.transport.Address;
import org.infinispan.statetransfer.StateTransferLock;
import org.infinispan.statetransfer.StateTransferManager;
import org.infinispan.transaction.WriteSkewHelper;
import org.infinispan.transaction.xa.CacheTransaction;
import org.infinispan.util.concurrent.locks.containers.ReentrantPerEntryLockContainer;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.TimeUnit;

import static org.infinispan.transaction.WriteSkewHelper.performTotalOrderWriteSkewCheckAndReturnNewVersions;
import static org.infinispan.transaction.WriteSkewHelper.performWriteSkewCheckAndReturnNewVersions;

/**
 * Abstractization for logic related to different clustering modes: replicated or distributed. This implements the <a
 * href="http://en.wikipedia.org/wiki/Bridge_pattern">Bridge</a> pattern as described by the GoF: this plays the role of
 * the <b>Implementor</b> and various LockingInterceptors are the <b>Abstraction</b>.
 *
 * @author Mircea Markus
 * @author Pedro Ruivo
 * @since 5.1
 */
@Scope(Scopes.NAMED_CACHE)
public interface ClusteringDependentLogic {

   boolean localNodeIsOwner(Object key);

   boolean localNodeIsPrimaryOwner(Object key);

   Address getPrimaryOwner(Object key);

   void commitEntry(CacheEntry entry, Metadata metadata, FlagAffectedCommand command, InvocationContext ctx);

   List<Address> getOwners(Collection<Object> keys);

   List<Address> getOwners(Object key);

   EntryVersionsMap createNewVersionsAndCheckForWriteSkews(VersionGenerator versionGenerator, TxInvocationContext context, VersionedPrepareCommand prepareCommand);

   Address getAddress();

   /**
    * Acquires internal lock to interact with DataContainer
    *
    * @param noWaitTime if {@code true}, it tries to acquire the lock without waiting
    * @return {@code true} if the lock was acquired.
    * @throws InterruptedException if interrupted while waiting.
    */
   boolean lock(Object key, boolean noWaitTime) throws InterruptedException;

   void unlock(Object key);

   public static abstract class AbstractClusteringDependentLogic implements ClusteringDependentLogic {

      protected DataContainer dataContainer;
      protected CacheNotifier notifier;
      protected boolean totalOrder;
      private WriteSkewHelper.KeySpecificLogic keySpecificLogic;
      private ReentrantPerEntryLockContainer lockContainer;

      @Inject
      public void init(DataContainer dataContainer, CacheNotifier notifier, Configuration configuration) {
         this.dataContainer = dataContainer;
         this.notifier = notifier;
         this.totalOrder = configuration.transaction().transactionProtocol().isTotalOrder();
         this.keySpecificLogic = initKeySpecificLogic(totalOrder);
         this.lockContainer = createLockContainer(configuration);
      }

      @Override
      public EntryVersionsMap createNewVersionsAndCheckForWriteSkews(VersionGenerator versionGenerator, TxInvocationContext context, VersionedPrepareCommand prepareCommand) {
         return totalOrder ?
               totalOrderCreateNewVersionsAndCheckForWriteSkews(versionGenerator, context, prepareCommand) :
               clusteredCreateNewVersionsAndCheckForWriteSkews(versionGenerator, context, prepareCommand);
      }

      protected abstract WriteSkewHelper.KeySpecificLogic initKeySpecificLogic(boolean totalOrder);

      protected void notifyCommitEntry(boolean created, boolean removed,
                                       boolean evicted, CacheEntry entry, InvocationContext ctx,
                                       FlagAffectedCommand command) {
         // Eviction has no notion of pre/post event since 4.2.0.ALPHA4.
         // EvictionManagerImpl.onEntryEviction() triggers both pre and post events
         // with non-null values, so we should do the same here as an ugly workaround.
         if (removed && evicted) {
            notifier.notifyCacheEntryEvicted(
                  entry.getKey(), entry.getValue(), ctx, command);
         } else if (removed) {
            notifier.notifyCacheEntryRemoved(
                  entry.getKey(), null, entry.getValue(), false, ctx, command);
         } else {
            // TODO: We're not very consistent (will JSR-107 solve it?):
            // Current tests expect entry modified to be fired when entry
            // created but not when entry removed

            // Notify entry modified after container has been updated
            notifier.notifyCacheEntryModified(entry.getKey(),
                                              entry.getValue(), created, false, ctx, command);

            // Notify entry created event after container has been updated
            if (created)
               notifier.notifyCacheEntryCreated(
                     entry.getKey(), entry.getValue(), false, ctx, command);
         }
      }

      private ReentrantPerEntryLockContainer createLockContainer(Configuration configuration) {
         //we need a lock container to synchronized the keys being moved between the data container and the persistence
         //with manual eviction.
         if (configuration.persistence().usingStores() && configuration.persistence().passivation()) {
            return new ReentrantPerEntryLockContainer(configuration.locking().concurrencyLevel());
         }
         return null;
      }

      private EntryVersionsMap totalOrderCreateNewVersionsAndCheckForWriteSkews(VersionGenerator versionGenerator, TxInvocationContext context,
                                                                                VersionedPrepareCommand prepareCommand) {
         if (context.isOriginLocal()) {
            throw new IllegalStateException("This must not be reached");
         }

         EntryVersionsMap updatedVersionMap = new EntryVersionsMap();

         if (!((TotalOrderPrepareCommand) prepareCommand).skipWriteSkewCheck()) {
            updatedVersionMap = performTotalOrderWriteSkewCheckAndReturnNewVersions(prepareCommand, dataContainer,
                                                                                    versionGenerator, context, keySpecificLogic);
         }

         for (WriteCommand c : prepareCommand.getModifications()) {
            for (Object k : c.getAffectedKeys()) {
               if (keySpecificLogic.performCheckOnKey(k)) {
                  if (!updatedVersionMap.containsKey(k)) {
                     updatedVersionMap.put(k, null);
                  }
               }
            }
         }

         context.getCacheTransaction().setUpdatedEntryVersions(updatedVersionMap);
         return updatedVersionMap;
      }

      private EntryVersionsMap clusteredCreateNewVersionsAndCheckForWriteSkews(VersionGenerator versionGenerator, TxInvocationContext context,
                                                                               VersionedPrepareCommand prepareCommand) {
         // Perform a write skew check on mapped entries.
         EntryVersionsMap uv = performWriteSkewCheckAndReturnNewVersions(prepareCommand, dataContainer,
                                                                         versionGenerator, context,
                                                                         keySpecificLogic);

         CacheTransaction cacheTransaction = context.getCacheTransaction();
         EntryVersionsMap uvOld = cacheTransaction.getUpdatedEntryVersions();
         if (uvOld != null && !uvOld.isEmpty()) {
            uvOld.putAll(uv);
            uv = uvOld;
         }
         cacheTransaction.setUpdatedEntryVersions(uv);
         return (uv.isEmpty()) ? null : uv;
      }

      @Override
      public final boolean lock(Object key, boolean noWaitTime) throws InterruptedException {
         if (lockContainer == null) {
            return true;
         }
         final long timeout = noWaitTime ? 0 : 1;
         return lockContainer.acquireLock(null, key, timeout, TimeUnit.DAYS) != null;
      }

      @Override
      public final void unlock(Object key) {
         if (lockContainer != null) {
            lockContainer.releaseLock(null, key);
         }
      }

      protected final void commitCacheEntry(CacheEntry entry, Metadata metadata) {
         forceLock(entry.getKey());
         entry.commit(dataContainer, metadata);
         unlock(entry.getKey());
      }

      private void forceLock(Object key) {
         boolean interrupted = false;
         boolean locked = false;
         do {
            try {
               locked = lock(key, false);
            } catch (InterruptedException e) {
               interrupted = true;
            }
         } while (!locked);
         if (interrupted) {
            Thread.currentThread().interrupt();
         }
      }
   }

   /**
    * This logic is used in local mode caches.
    */
   public static class LocalLogic extends AbstractClusteringDependentLogic {

      private EmbeddedCacheManager cacheManager;

      @Inject
      public void init(EmbeddedCacheManager cacheManager) {
         this.cacheManager = cacheManager;
      }

      @Override
      public boolean localNodeIsOwner(Object key) {
         return true;
      }

      @Override
      public boolean localNodeIsPrimaryOwner(Object key) {
         return true;
      }

      @Override
      public Address getPrimaryOwner(Object key) {
         throw new IllegalStateException("Cannot invoke this method for local caches");
      }

      @Override
      public List<Address> getOwners(Collection<Object> keys) {
         return null;
      }

      @Override
      public List<Address> getOwners(Object key) {
         return null;
      }

      @Override
      public Address getAddress() {
         Address address = cacheManager.getAddress();
         if (address == null) {
            address = LOCAL_MODE_ADDRESS;
         }
         return address;
      }

      @Override
      public void commitEntry(CacheEntry entry, Metadata metadata, FlagAffectedCommand command, InvocationContext ctx) {
         // Cache flags before they're reset
         // TODO: Can the reset be done after notification instead?
         boolean created = entry.isCreated();
         boolean removed = entry.isRemoved();
         boolean evicted = entry.isEvicted();

         commitCacheEntry(entry, metadata);

         // Notify after events if necessary
         notifyCommitEntry(created, removed, evicted, entry, ctx, command);
      }

      @Override
      public EntryVersionsMap createNewVersionsAndCheckForWriteSkews(VersionGenerator versionGenerator, TxInvocationContext context, VersionedPrepareCommand prepareCommand) {
         throw new IllegalStateException("Cannot invoke this method for local caches");
      }

      @Override
      protected WriteSkewHelper.KeySpecificLogic initKeySpecificLogic(boolean totalOrder) {
         return null; //not used
      }
   }

   static final Address LOCAL_MODE_ADDRESS = new Address() {
      @Override
      public String toString() {
         return "Local Address";
      }

      @Override
      public int compareTo(Address o) {
         return 0;
      }
   };

   /**
    * This logic is used in invalidation mode caches.
    */
   public static class InvalidationLogic extends AbstractClusteringDependentLogic {

      private StateTransferManager stateTransferManager;
      private RpcManager rpcManager;

      @Inject
      public void init(RpcManager rpcManager, StateTransferManager stateTransferManager) {
         this.rpcManager = rpcManager;
         this.stateTransferManager = stateTransferManager;
      }

      @Override
      public boolean localNodeIsOwner(Object key) {
         return stateTransferManager.getCacheTopology().getWriteConsistentHash().isKeyLocalToNode(rpcManager.getAddress(), key);
      }

      @Override
      public boolean localNodeIsPrimaryOwner(Object key) {
         return stateTransferManager.getCacheTopology().getWriteConsistentHash().locatePrimaryOwner(key).equals(rpcManager.getAddress());
      }

      @Override
      public Address getPrimaryOwner(Object key) {
         return stateTransferManager.getCacheTopology().getWriteConsistentHash().locatePrimaryOwner(key);
      }

      @Override
      public void commitEntry(CacheEntry entry, Metadata metadata,
            FlagAffectedCommand command, InvocationContext ctx) {
         // Cache flags before they're reset
         // TODO: Can the reset be done after notification instead?
         boolean created = entry.isCreated();
         boolean removed = entry.isRemoved();
         boolean evicted = entry.isEvicted();

         commitCacheEntry(entry, metadata);

         // Notify after events if necessary
         notifyCommitEntry(created, removed, evicted, entry, ctx, command);
      }

      @Override
      public List<Address> getOwners(Collection<Object> keys) {
         return null;    //todo [anistor] should I actually return this based on current CH?
      }

      @Override
      public List<Address> getOwners(Object key) {
         return null;
      }

      @Override
      public Address getAddress() {
         return rpcManager.getAddress();
      }

      @Override
      protected WriteSkewHelper.KeySpecificLogic initKeySpecificLogic(boolean totalOrder) {
         return null; //not used because write skew check is not allowed with invalidation
      }
   }

   /**
    * This logic is used in replicated mode caches.
    */
   public static class ReplicationLogic extends InvalidationLogic {

      private StateTransferLock stateTransferLock;

      @Inject
      public void init(StateTransferLock stateTransferLock) {
         this.stateTransferLock = stateTransferLock;
      }

      @Override
      public void commitEntry(CacheEntry entry, Metadata metadata, FlagAffectedCommand command, InvocationContext ctx) {
         stateTransferLock.acquireSharedTopologyLock();
         try {
            super.commitEntry(entry, metadata, command, ctx);
         } finally {
            stateTransferLock.releaseSharedTopologyLock();
         }
      }

      @Override
      protected WriteSkewHelper.KeySpecificLogic initKeySpecificLogic(boolean totalOrder) {
         return totalOrder ?
               new WriteSkewHelper.KeySpecificLogic() {
                  @Override
                  public boolean performCheckOnKey(Object key) {
                     //in total order, all nodes perform the write skew check
                     return true;
                  }
               } :
               new WriteSkewHelper.KeySpecificLogic() {
                  @Override
                  public boolean performCheckOnKey(Object key) {
                     //in two phase commit, only the primary owner should perform the write skew check
                     return localNodeIsPrimaryOwner(key);
                  }
               };
      }
   }

   /**
    * This logic is used in distributed mode caches.
    */
   public static class DistributionLogic extends AbstractClusteringDependentLogic {

      private DistributionManager dm;
      private Configuration configuration;
      private RpcManager rpcManager;
      private StateTransferLock stateTransferLock;

      @Inject
      public void init(DistributionManager dm, Configuration configuration,
                       RpcManager rpcManager, StateTransferLock stateTransferLock) {
         this.dm = dm;
         this.configuration = configuration;
         this.rpcManager = rpcManager;
         this.stateTransferLock = stateTransferLock;
      }

      @Override
      public boolean localNodeIsOwner(Object key) {
         return dm.getLocality(key).isLocal();
      }

      @Override
      public Address getAddress() {
         return rpcManager.getAddress();
      }

      @Override
      public boolean localNodeIsPrimaryOwner(Object key) {
         final Address address = rpcManager.getAddress();
         return dm.getPrimaryLocation(key).equals(address);
      }

      @Override
      public Address getPrimaryOwner(Object key) {
         return dm.getPrimaryLocation(key);
      }

      @Override
      public void commitEntry(CacheEntry entry, Metadata metadata, FlagAffectedCommand command, InvocationContext ctx) {
         // Don't allow the CH to change (and state transfer to invalidate entries)
         // between the ownership check and the commit
         stateTransferLock.acquireSharedTopologyLock();
         try {
            boolean doCommit = true;
            // ignore locality for removals, even if skipOwnershipCheck is not true
            boolean skipOwnershipCheck = command != null &&
                  command.hasFlag(Flag.SKIP_OWNERSHIP_CHECK);

            boolean isForeignOwned = !skipOwnershipCheck && !localNodeIsOwner(entry.getKey());
            if (isForeignOwned && !entry.isRemoved()) {
               if (configuration.clustering().l1().enabled()) {
                  // transform for L1
                  long lifespan;
                  if (metadata != null) {
                     lifespan = metadata.lifespan();
                  } else {
                     lifespan = entry.getLifespan();
                  }
                  if (lifespan < 0 || lifespan > configuration.clustering().l1().lifespan()) {
                     Metadata.Builder builder;
                     if (metadata != null) {
                        builder = metadata.builder();
                     } else {
                        builder = entry.getMetadata().builder();
                     }
                     Metadata newMetadata = builder
                           .lifespan(configuration.clustering().l1().lifespan())
                           .build();
                     metadata = newMetadata;
                  }
               } else {
                  doCommit = false;
               }
            }

            boolean created = false;
            boolean removed = false;
            boolean evicted = false;
            if (!isForeignOwned) {
               created = entry.isCreated();
               removed = entry.isRemoved();
               evicted = entry.isEvicted();
            }

            if (doCommit)
               commitCacheEntry(entry, metadata);
            else
               entry.rollback();

            if (!isForeignOwned) {
               notifyCommitEntry(created, removed, evicted, entry, ctx, command);
            }
         } finally {
            stateTransferLock.releaseSharedTopologyLock();
         }
      }

      @Override
      public List<Address> getOwners(Collection<Object> affectedKeys) {
         if (affectedKeys.isEmpty()) {
            return InfinispanCollections.emptyList();
         }
         return Immutables.immutableListConvert(dm.locateAll(affectedKeys));
      }

      @Override
      public List<Address> getOwners(Object key) {
         return Immutables.immutableListConvert(dm.locate(key));
      }

      @Override
      protected WriteSkewHelper.KeySpecificLogic initKeySpecificLogic(boolean totalOrder) {
         return totalOrder ?
               new WriteSkewHelper.KeySpecificLogic() {
                  @Override
                  public boolean performCheckOnKey(Object key) {
                     //in total order, all the owners can perform the write skew check.
                     return localNodeIsOwner(key);
                  }
               } :
               new WriteSkewHelper.KeySpecificLogic() {
                  @Override
                  public boolean performCheckOnKey(Object key) {
                     //in two phase commit, only the primary owner should perform the write skew check
                     return localNodeIsPrimaryOwner(key);
                  }
               };
      }
   }
}
