package com.airlock.iam.common.infrastructure.state.redis;

import com.airlock.iam.elog.api.EDefaultLogType;
import com.airlock.iam.elog.api.ELogger;
import com.airlock.iam.elog.api.ELoggerFactory;
import com.google.common.annotations.VisibleForTesting;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.time.Duration;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import org.redisson.api.RedissonClient;

public class LockKeepAlivePoller {
   private static final ELogger LOG = ELoggerFactory.getLogger(LockKeepAlivePoller.class);
   private static LockKeepAlivePoller instance;
   private final LockOwnerId lockOwnerId = LockOwnerId.getInstance();
   private Duration pollingInterval;
   private ScheduledExecutorService schedulerService;

   public static synchronized LockKeepAlivePoller getInstance() {
      if (instance == null) {
         instance = new LockKeepAlivePoller(RedisStateRepository.LOCK_LIFETIME_POLLING_INTERVAL);
      }

      return instance;
   }

   private LockKeepAlivePoller(Duration pollingInterval) {
      this.pollingInterval = pollingInterval;
   }

   public void startPolling(Supplier<Set<RedissonClient>> redissonClientsSupplier) {
      this.startPollingInternal(() -> this.createTasks(redissonClientsSupplier.get()));
   }

   public void startPolling(Set<Runnable> tasks) {
      this.startPollingInternal(() -> tasks);
   }

   private void startPollingInternal(Supplier<Set<Runnable>> runnableSupplier) {
      if (this.schedulerService != null && !this.schedulerService.isTerminated()) {
         this.schedulerService.shutdownNow();
      }

      LOG.debug("Starting to keep alive locks on redis...");
      this.schedulerService = Executors.newScheduledThreadPool(1);
      ExecutorService executorService = Executors.newCachedThreadPool();
      Duration taskTimeout = this.pollingInterval.multipliedBy(8L).dividedBy(10L);
      this.schedulerService
         .scheduleAtFixedRate(
            () -> this.executeTaskWithTimeout(executorService, runnableSupplier.get(), taskTimeout), 0L, this.pollingInterval.toNanos(), TimeUnit.NANOSECONDS
         );
   }

   public void startPolling(Runnable task) {
      this.startPolling(Set.of(task));
   }

   private Set<Runnable> createTasks(Set<RedissonClient> redissonClients) {
      return redissonClients.stream().map(RedisPollingTaskFactory::createPollingTask).collect(Collectors.toSet());
   }

   public void stopPolling() {
      LOG.debug("Stopping polling to keep alive locks...");
      this.schedulerService.shutdownNow();
   }

   @VisibleForTesting
   void changePollingInterval(Duration pollingInterval) {
      this.pollingInterval = pollingInterval;
   }

   @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED_BAD_PRACTICE")
   private void executeTaskWithTimeout(ExecutorService executorService, Set<Runnable> tasks, Duration timeout) {
      for (Runnable task : tasks) {
         executorService.submit(
            () -> {
               Future<?> timeoutFuture = executorService.submit(task);

               try {
                  timeoutFuture.get(timeout.toNanos(), TimeUnit.NANOSECONDS);
               } catch (TimeoutException e) {
                  LOG.warn(
                     EDefaultLogType.GENERIC_LOG_TYPE,
                     "Timeout occurred during extension of lock lifetimes for lock owner %s. Aborting task.".formatted(this.lockOwnerId.value()),
                     e
                  );
                  timeoutFuture.cancel(true);
               } catch (InterruptedException e) {
                  Thread.currentThread().interrupt();
               } catch (ExecutionException e) {
                  if (e.getCause() instanceof NullPointerException) {
                     LOG.warn(
                        EDefaultLogType.GENERIC_LOG_TYPE,
                        "Failed to connect to redis during extension of lock lifetimes. This situation is known to occur during config activations."
                     );
                  } else {
                     LOG.warn(
                        EDefaultLogType.GENERIC_LOG_TYPE,
                        "Exception occurred during extension of lock lifetimes for lock owner %s: ".formatted(this.lockOwnerId.value()),
                        e
                     );
                  }
               }
            }
         );
      }
   }
}