package org.infinispan.client.hotrod;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.Flow;

import org.infinispan.api.Experimental;
import org.infinispan.api.async.AsyncCacheEntryProcessor;
import org.infinispan.api.common.CacheEntry;
import org.infinispan.api.common.CacheEntryVersion;
import org.infinispan.api.common.CacheOptions;
import org.infinispan.api.common.CacheWriteOptions;
import org.infinispan.api.common.MutableCacheEntry;
import org.infinispan.api.common.events.cache.CacheEntryEvent;
import org.infinispan.api.common.events.cache.CacheEntryEventType;
import org.infinispan.api.common.events.cache.CacheListenerOptions;
import org.infinispan.api.common.process.CacheEntryProcessorContext;
import org.infinispan.api.common.process.CacheEntryProcessorResult;
import org.infinispan.api.configuration.CacheConfiguration;
import org.infinispan.api.mutiny.MutinyCache;
import org.infinispan.api.mutiny.MutinyCacheEntryProcessor;
import org.infinispan.api.mutiny.MutinyContainer;
import org.infinispan.api.mutiny.MutinyQuery;
import org.infinispan.api.mutiny.MutinyStreamingCache;
import org.infinispan.client.hotrod.impl.InternalRemoteCache;

import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.Uni;

/**
 * @since 14.0
 **/
@Experimental
final class HotRodMutinyCache<K, V> implements MutinyCache<K, V> {
   private final HotRod hotrod;
   private final InternalRemoteCache<K, V> remoteCache;

   HotRodMutinyCache(HotRod hotrod, InternalRemoteCache<K, V> remoteCache) {
      this.hotrod = hotrod;
      this.remoteCache = remoteCache;
   }

   @Override
   public String name() {
      return remoteCache.getName();
   }

   @Override
   public Uni<CacheConfiguration> configuration() {
      return Uni.createFrom().completionStage(remoteCache.configuration());
   }

   @Override
   public MutinyContainer container() {
      return hotrod.mutiny();
   }

   @Override
   public Uni<V> get(K key, CacheOptions options) {
      return Uni.createFrom().completionStage(() -> remoteCache.get(key, options));
   }

   @Override
   public Uni<CacheEntry<K, V>> getEntry(K key, CacheOptions options) {
      return Uni.createFrom().completionStage(() -> remoteCache.getEntry(key, options));
   }

   @Override
   public Uni<CacheEntry<K, V>> putIfAbsent(K key, V value, CacheWriteOptions options) {
      return Uni.createFrom().completionStage(() -> remoteCache.putIfAbsent(key, value, options));
   }

   @Override
   public Uni<Boolean> setIfAbsent(K key, V value, CacheWriteOptions options) {
      return Uni.createFrom().completionStage(() -> remoteCache.setIfAbsent(key, value, options));
   }

   @Override
   public Uni<CacheEntry<K, V>> put(K key, V value, CacheWriteOptions options) {
      return Uni.createFrom().completionStage(() -> remoteCache.put(key, value, options));
   }

   @Override
   public Uni<Void> set(K key, V value, CacheWriteOptions options) {
      return Uni.createFrom().completionStage(() -> remoteCache.set(key, value, options));
   }

   @Override
   public Uni<Boolean> remove(K key, CacheOptions options) {
      return Uni.createFrom().completionStage(() -> remoteCache.remove(key, options));
   }

   @Override
   public Uni<CacheEntry<K, V>> getAndRemove(K key, CacheOptions options) {
      return Uni.createFrom().completionStage(() -> remoteCache.getAndRemove(key, options));
   }

   @Override
   public Multi<K> keys(CacheOptions options) {
      return Multi.createFrom().publisher(remoteCache.keys(options));
   }

   @Override
   public Multi<CacheEntry<K, V>> entries(CacheOptions options) {
      return Multi.createFrom().publisher(remoteCache.entries(options));
   }

   @Override
   public Multi<CacheEntry<K, V>> getAll(Set<K> keys, CacheOptions options) {
      return Multi.createFrom().publisher(remoteCache.getAll(keys, options));
   }

   @Override
   public Multi<CacheEntry<K, V>> getAll(CacheOptions options, K[] keys) {
      return Multi.createFrom().publisher(remoteCache.getAll(options, keys));
   }

   @Override
   public Uni<Void> putAll(Multi<CacheEntry<K, V>> entries, CacheWriteOptions options) {
      return Uni.createFrom().completionStage(() -> remoteCache.putAll(entries.convert().toPublisher(), options));
   }

   @Override
   public Uni<Void> putAll(Map<K, V> map, CacheWriteOptions options) {
      return Uni.createFrom().completionStage(() -> remoteCache.putAll(map, options));
   }

   @Override
   public Uni<Boolean> replace(K key, V value, CacheEntryVersion version, CacheWriteOptions options) {
      return Uni.createFrom().completionStage(() -> remoteCache.replace(key, value, version, options));
   }

   @Override
   public Uni<CacheEntry<K, V>> getOrReplaceEntry(K key, V value, CacheEntryVersion version, CacheWriteOptions options) {
      return Uni.createFrom().completionStage(() -> remoteCache.getOrReplaceEntry(key, value, version, options));
   }

   @Override
   public Multi<K> removeAll(Set<K> keys, CacheWriteOptions options) {
      return Multi.createFrom().publisher(remoteCache.removeAll(keys, options));
   }

   @Override
   public Multi<K> removeAll(Multi<K> keys, CacheWriteOptions options) {
      return Multi.createFrom().publisher(remoteCache.removeAll(keys.convert().toPublisher(), options));
   }

   @Override
   public Multi<CacheEntry<K, V>> getAndRemoveAll(Multi<K> keys, CacheWriteOptions options) {
      return Multi.createFrom().publisher(remoteCache.getAndRemoveAll(keys.convert().toPublisher(), options));
   }

   @Override
   public Uni<Long> estimateSize(CacheOptions options) {
      return Uni.createFrom().completionStage(() -> remoteCache.estimateSize(options));
   }

   @Override
   public Uni<Void> clear(CacheOptions options) {
      return Uni.createFrom().completionStage(() -> remoteCache.clear(options));
   }

   @Override
   public <R> MutinyQuery<K, V, R> query(String query, CacheOptions options) {
      return new HotRodMutinyQuery<>();
   }

   @Override
   public Multi<CacheEntryEvent<K, V>> listen(CacheListenerOptions options, CacheEntryEventType... types) {
      return Multi.createFrom().publisher(remoteCache.listen(options, types));
   }

   @Override
   public <T> Multi<CacheEntryProcessorResult<K, T>> process(Set<K> keys, MutinyCacheEntryProcessor<K, V, T> processor, CacheOptions options) {
      return Multi.createFrom().publisher(remoteCache.process(keys, new MutinyToAsyncCacheEntryProcessor<>(processor), options));
   }

   @Override
   public MutinyStreamingCache<K> streaming() {
      return new HotRodMutinyStreamingCache<>();
   }

   private record MutinyToAsyncCacheEntryProcessor<K, V, T>(MutinyCacheEntryProcessor<K, V, T> processor) implements AsyncCacheEntryProcessor<K, V, T> {

      @Override
      public Flow.Publisher<CacheEntryProcessorResult<K, T>> process(Flow.Publisher<MutableCacheEntry<K, V>> entries, CacheEntryProcessorContext context) {
         return processor.process(Multi.createFrom().publisher(entries), context).convert().toPublisher();
      }
   }
}
