package org.infinispan.server.hotrod;

import static org.infinispan.commons.util.concurrent.CompletableFutures.extractException;

import java.util.Collection;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;

import javax.security.auth.Subject;

import org.infinispan.counter.EmbeddedCounterManagerFactory;
import org.infinispan.counter.api.CounterConfiguration;
import org.infinispan.counter.exception.CounterNotFoundException;
import org.infinispan.counter.exception.CounterOutOfBoundsException;
import org.infinispan.counter.impl.CounterModuleLifecycle;
import org.infinispan.counter.impl.manager.EmbeddedCounterManager;
import org.infinispan.counter.impl.manager.InternalCounterAdmin;
import org.infinispan.server.hotrod.counter.listener.ClientCounterManagerNotificationManager;
import org.infinispan.server.hotrod.counter.listener.ListenerOperationStatus;

import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;

class CounterRequestProcessor extends BaseRequestProcessor {

   private ClientCounterManagerNotificationManager notificationManager;
   private EmbeddedCounterManager counterManager;

   CounterRequestProcessor(Channel channel, Executor executor, HotRodServer server) {
      super(channel, executor, server);
   }

   private EmbeddedCounterManager counterManager(HotRodHeader header) {
      header.cacheName = CounterModuleLifecycle.COUNTER_CACHE_NAME;
      if (counterManager == null) {
         counterManager = (EmbeddedCounterManager) EmbeddedCounterManagerFactory.asCounterManager(server.getCacheManager());
      }

      return counterManager;
   }

   private ClientCounterManagerNotificationManager notificationManager() {
      if (notificationManager == null) {
         notificationManager = server.getClientCounterNotificationManager();
      }
      return notificationManager;
   }

   void removeCounterListener(HotRodHeader header, Subject subject, String counterName, byte[] listenerId) {
      CompletableFuture.supplyAsync(() -> notificationManager()
                  .removeCounterListener(listenerId, counterName), executor)
            .whenCompleteAsync((status, throwable) -> listerOperationStatusHandler(header, status, throwable), channel.eventLoop());
   }

   void addCounterListener(HotRodHeader header, Subject subject, String counterName, byte[] listenerId) {
      CompletableFuture.supplyAsync(() -> notificationManager()
                  .addCounterListener(listenerId, header.getVersion(), counterName, channel, header.encoder()), executor)
            .whenCompleteAsync((status, throwable) -> listerOperationStatusHandler(header, status, throwable), channel.eventLoop());
   }

   void getCounterNames(HotRodHeader header, Subject subject) {
      Collection<String> counterNames = counterManager(header).getCounterNames();
      writeResponse(header, header.encoder().counterNamesResponse(header, server, channel, counterNames));
   }

   void counterRemove(HotRodHeader header, Subject subject, String counterName) {
      counterManager(header).removeAsync(counterName, true)
            .whenCompleteAsync((___, throwable) -> voidResultHandler(header, throwable), channel.eventLoop());
   }

   void counterCompareAndSwap(HotRodHeader header, Subject subject, String counterName, long expect, long update) {
      counterManager(header).getStrongCounterAsync(counterName)
            .thenCompose(strongCounter -> strongCounter.compareAndSwap(expect, update))
            .whenCompleteAsync((returnValue, throwable) -> longResultHandler(header, returnValue, throwable), channel.eventLoop());
   }

   void counterGet(HotRodHeader header, Subject subject, String counterName) {
      counterManager(header).getOrCreateAsync(counterName)
            .thenCompose(InternalCounterAdmin::value)
            .whenCompleteAsync((value, throwable) -> longResultHandler(header, value, throwable), channel.eventLoop());
   }

   void counterReset(HotRodHeader header, Subject subject, String counterName) {
      counterManager(header).getOrCreateAsync(counterName)
            .thenCompose(InternalCounterAdmin::reset)
            .whenCompleteAsync((unused, throwable) -> voidResultHandler(header, throwable), channel.eventLoop());
   }

   void counterAddAndGet(HotRodHeader header, Subject subject, String counterName, long value) {
      counterManager(header).getOrCreateAsync(counterName)
            .thenAccept(counter -> {
               if (counter.isWeakCounter()) {
                  counter.asWeakCounter().add(value).whenCompleteAsync((___, t) -> longResultHandler(header, 0L, t), channel.eventLoop());
               } else {
                  counter.asStrongCounter().addAndGet(value).whenCompleteAsync((rv, t) -> longResultHandler(header, rv, t), channel.eventLoop());
               }
            })
            .exceptionallyAsync(throwable -> {
               checkCounterThrowable(header, throwable);
               return null;
            }, channel.eventLoop());
   }

   void counterSet(HotRodHeader header, Subject subject, String counterName, long value) {
      counterManager(header).getStrongCounterAsync(counterName)
            .thenCompose(strongCounter -> strongCounter.getAndSet(value))
            .whenCompleteAsync((returnValue, throwable) -> longResultHandler(header, returnValue, throwable), channel.eventLoop());
   }

   void getCounterConfiguration(HotRodHeader header, Subject subject, String counterName) {
      counterManager(header).getConfigurationAsync(counterName)
            .whenCompleteAsync((configuration, throwable) -> handleGetCounterConfiguration(header, configuration, throwable), channel.eventLoop());
   }

   private void handleGetCounterConfiguration(HotRodHeader header, CounterConfiguration configuration, Throwable throwable) {
      if (throwable != null) {
         checkCounterThrowable(header, throwable);
      } else {
         ByteBuf response = configuration == null ? missingCounterResponse(header) :
               header.encoder().counterConfigurationResponse(header, server, channel, configuration);
         writeResponse(header, response);
      }
   }

   void isCounterDefined(HotRodHeader header, Subject subject, String counterName) {
      counterManager(header).isDefinedAsync(counterName)
            .whenCompleteAsync((value, throwable) -> booleanResultHandler(header, value, throwable), channel.eventLoop());
   }

   void createCounter(HotRodHeader header, Subject subject, String counterName, CounterConfiguration configuration) {
      counterManager(header).defineCounterAsync(counterName, configuration)
            .whenCompleteAsync((value, throwable) -> booleanResultHandler(header, value, throwable), channel.eventLoop());
   }

   private ByteBuf createResponseFrom(HotRodHeader header, ListenerOperationStatus status) {
      return switch (status) {
         case OK -> header.encoder().emptyResponse(header, server, channel, OperationStatus.OperationNotExecuted);
         case OK_AND_CHANNEL_IN_USE -> header.encoder().emptyResponse(header, server, channel, OperationStatus.Success);
         case COUNTER_NOT_FOUND -> missingCounterResponse(header);
      };
   }

   private void listerOperationStatusHandler(HotRodHeader header, ListenerOperationStatus status, Throwable throwable) {
      if (throwable != null) {
         checkCounterThrowable(header, throwable);
      } else {
         writeResponse(header, createResponseFrom(header, status));
      }
   }

   private void checkCounterThrowable(HotRodHeader header, Throwable throwable) {
      Throwable cause = extractException(throwable);
      if (cause instanceof CounterOutOfBoundsException) {
         writeResponse(header, header.encoder().emptyResponse(header, server, channel, OperationStatus.NotExecutedWithPrevious));
      } else if (cause instanceof CounterNotFoundException) {
         writeResponse(header, missingCounterResponse(header));
      } else {
         writeException(header, cause);
      }
   }

   private ByteBuf missingCounterResponse(HotRodHeader header) {
      return header.encoder().emptyResponse(header, server, channel, OperationStatus.KeyDoesNotExist);
   }

   private void booleanResultHandler(HotRodHeader header, Boolean value, Throwable throwable) {
      if (throwable != null) {
         checkCounterThrowable(header, throwable);
      } else {
         OperationStatus status = value ? OperationStatus.Success : OperationStatus.OperationNotExecuted;
         writeResponse(header, header.encoder().emptyResponse(header, server, channel, status));
      }
   }

   private void longResultHandler(HotRodHeader header, Long value, Throwable throwable) {
      if (throwable != null) {
         checkCounterThrowable(header, throwable);
      } else {
         writeResponse(header, header.encoder().longResponse(header, server, channel, value));
      }
   }

   private void voidResultHandler(HotRodHeader header, Throwable throwable) {
      if (throwable != null) {
         checkCounterThrowable(header, throwable);
      } else {
         writeResponse(header, header.encoder().emptyResponse(header, server, channel, OperationStatus.Success));
      }
   }
}
