package pl.allegro.tech.hermes.management.domain.consistency;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import pl.allegro.tech.hermes.management.config.kafka.KafkaClustersProperties;
import pl.allegro.tech.hermes.management.domain.auth.RequestUser;
import pl.allegro.tech.hermes.management.domain.topic.TopicService;
import pl.allegro.tech.hermes.management.infrastructure.kafka.MultiDCAwareService;

import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

import static java.util.Arrays.asList;

@Component
public class KafkaHermesConsistencyService {

    private static final Logger logger = LoggerFactory.getLogger(KafkaHermesConsistencyService.class);

    private static final String AVRO_SUFFIX = "_avro";
    private static final List<String> IGNORED_TOPIC = asList("__consumer_offsets");
    private final TopicService topicService;
    private final MultiDCAwareService multiDCAwareService;
    private final KafkaClustersProperties kafkaClustersProperties;

    public KafkaHermesConsistencyService(
        TopicService topicService,
        MultiDCAwareService multiDCAwareService,
        KafkaClustersProperties kafkaClustersProperties) {
        this.topicService = topicService;
        this.kafkaClustersProperties = kafkaClustersProperties;
        this.multiDCAwareService = multiDCAwareService;
    }

    public Set<String> listInconsistentTopics() {
        List<String> topicsFromHermes = topicService.listQualifiedTopicNames();

        return multiDCAwareService.listTopicFromAllDC()
            .stream()
            .filter(topic -> !IGNORED_TOPIC.contains(topic) && !topicsFromHermes
                .contains(mapToHermesFormat(topic)))
            .collect(Collectors.toSet());
    }

    public void removeTopic(String topicName, RequestUser requestUser) {
        logger.info("Removing topic {} only on brokers. Requested by {}", topicName, requestUser.getUsername());
        multiDCAwareService.removeTopicByName(topicName);
        logger.info("Successfully removed topic {} on brokers. Requested by {}", topicName, requestUser.getUsername());
    }

    private String mapToHermesFormat(String topic) {
        String prefix = kafkaClustersProperties.getDefaultNamespace() + kafkaClustersProperties.getNamespaceSeparator();

        int beginIndex = topic.startsWith(prefix) ? prefix.length() : 0;
        int endIndex = topic.endsWith(AVRO_SUFFIX) ? topic.length() - AVRO_SUFFIX.length() : topic.length();

        return topic.substring(beginIndex, endIndex);
    }
}
