package com.github.alebabai.tg2vk.service.impl;

import com.github.alebabai.tg2vk.domain.User;
import com.github.alebabai.tg2vk.repository.UserRepository;
import com.github.alebabai.tg2vk.service.VkMessagesProcessor;
import com.github.alebabai.tg2vk.service.VkService;
import com.vk.api.sdk.objects.messages.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.transaction.Transactional;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.function.BiConsumer;

@Service
public class VkMessagesProcessorImpl implements VkMessagesProcessor {

    private static final Logger LOGGER = LoggerFactory.getLogger(VkMessagesProcessorImpl.class);

    @Value("${tg2vk.vk.service.processor.auto_init_pool:false}")
    private boolean autoInitPool;

    private final UserRepository userRepository;
    private final VkService vkService;
    private final LinkerServiceImpl linkerService;
    private Map<Integer, CompletableFuture<Integer>> taskPool;

    @Autowired
    public VkMessagesProcessorImpl(UserRepository userRepository,
                                   VkService vkService,
                                   LinkerServiceImpl linkerService) {
        this.userRepository = userRepository;
        this.vkService = vkService;
        this.linkerService = linkerService;
        this.taskPool = new HashMap<>();
    }

    @PostConstruct
    protected void init() {
        if (autoInitPool) {
            userRepository.findAllStarted().forEach(this::start);
        }
    }

    @Override
    @Transactional
    public void start(User user) {
        final boolean isStarted = taskPool.keySet()
                .stream()
                .anyMatch(id -> Objects.equals(id, user.getId()));
        if (!isStarted) {
            final BiConsumer<com.vk.api.sdk.objects.users.User, Message> messageHandler = linkerService.getVkMessageHandler(user.getId());
            final CompletableFuture<Integer> task = vkService.fetchMessages(user, messageHandler);
            taskPool.put(user.getId(), task);
            user.getSettings().setStarted(true);
            userRepository.save(user);
            LOGGER.debug("Start vk messages processing for {}", user);
        }
    }

    @Override
    @Transactional
    public void stop(User user) {
        Optional.ofNullable(taskPool.get(user.getId()))
                .ifPresent(task -> {
                    task.cancel(true);
                    user.getSettings().setStarted(false);
                    userRepository.save(user);
                    taskPool.remove(user.getId());
                    LOGGER.debug("Stop messages processing for {}", user);
                });
    }
}

@@ -620,56 +620,8 @@
 al;%0A
-import java.util.HashMap;%0Aimport java.util.Map;%0A
 impo
@@ -642,16 +642,16 @@
 bjects;%0A
+
 import j
@@ -716,16 +716,106 @@
 Future;%0A
+import java.util.concurrent.ConcurrentHashMap;%0Aimport java.util.concurrent.ConcurrentMap;%0A
 import j
@@ -1274,16 +1274,32 @@
 private 
+final Concurrent
 Map%3CInte
@@ -1667,32 +1667,32 @@
  linkerService;%0A
-
         this.tas
@@ -1703,16 +1703,26 @@
 l = new 
+Concurrent
 HashMap%3C

fix concurency behaviour