package cz.moneta.test.harness.endpoints.messaging;

import cz.moneta.test.harness.context.StoreAccessor;
import cz.moneta.test.harness.endpoints.Endpoint;
import cz.moneta.test.harness.exception.MessagingTimeoutException;
import cz.moneta.test.harness.messaging.model.Destination;
import cz.moneta.test.harness.messaging.model.MqMessageFormat;
import cz.moneta.test.harness.messaging.model.Queue;
import cz.moneta.test.harness.messaging.model.ReceivedMessage;
import cz.moneta.test.harness.messaging.model.Topic;

import java.time.Duration;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.function.Predicate;

public class MessagingEndpoint implements Endpoint {

    private final StoreAccessor store;
    private volatile KafkaEndpoint kafkaEndpoint;
    private volatile IbmMqEndpoint ibmMqEndpoint;
    private final Object endpointLock = new Object();

    public MessagingEndpoint(StoreAccessor store) {
        this.store = store;
    }

    public void send(String destinationName,
                     String key,
                     String payload,
                     MqMessageFormat formatOverride,
                     Map<String, String> headers) {
        Destination destination = resolveDestination(destinationName);
        if (destination instanceof Topic topic) {
            getKafkaEndpoint().send(topic.getTopicName(), key, payload, headers);
            return;
        }

        Queue queue = (Queue) destination;
        MqMessageFormat format = Optional.ofNullable(formatOverride).orElse(queue.getFormat());
        getIbmMqEndpoint().send(queue.getQueueName(), payload, format, headers);
    }

    public ReceivedMessage receive(String destinationName,
                                   Predicate<ReceivedMessage> filter,
                                   Duration timeout,
                                   MqMessageFormat formatOverride) {
        Destination destination = resolveDestination(destinationName);
        if (destination instanceof Topic topic) {
            return getKafkaEndpoint().receive(topic.getTopicName(), filter, timeout);
        }

        Queue queue = (Queue) destination;
        MqMessageFormat format = Optional.ofNullable(formatOverride).orElse(queue.getFormat());
        List<ReceivedMessage> messages = getIbmMqEndpoint().browse(queue.getQueueName(), filter, format, timeout);
        if (messages.isEmpty()) {
            throw new MessagingTimeoutException("No IBM MQ message found for destination: " + destinationName);
        }
        return messages.get(0);
    }

    public Destination resolveDestination(String destinationName) {
        String prefix = "messaging.destination." + destinationName + ".";
        String type = Optional.ofNullable(store.getConfig(prefix + "type"))
                .map(v -> v.toLowerCase(Locale.ROOT))
                .orElseThrow(() -> new IllegalStateException("Missing destination config: " + prefix + "type"));

        return switch (type) {
            case "kafka" -> {
                String topic = requireConfig(prefix + "topic");
                yield new Topic(destinationName, topic);
            }
            case "ibmmq" -> {
                String queue = requireConfig(prefix + "queue");
                String format = store.getConfig(prefix + "format", MqMessageFormat.JSON.name().toLowerCase(Locale.ROOT));
                yield new Queue(destinationName, queue, MqMessageFormat.fromConfig(format, MqMessageFormat.JSON));
            }
            default -> throw new IllegalStateException("Unsupported destination type '" + type + "' for destination: " + destinationName);
        };
    }

    @Override
    public void close() {
        KafkaEndpoint kafka = kafkaEndpoint;
        if (kafka != null) {
            kafka.close();
        }
        IbmMqEndpoint mq = ibmMqEndpoint;
        if (mq != null) {
            mq.close();
        }
    }

    private KafkaEndpoint getKafkaEndpoint() {
        KafkaEndpoint current = kafkaEndpoint;
        if (current == null) {
            synchronized (endpointLock) {
                current = kafkaEndpoint;
                if (current == null) {
                    current = new KafkaEndpoint(store);
                    kafkaEndpoint = current;
                }
            }
        }
        return current;
    }

    private IbmMqEndpoint getIbmMqEndpoint() {
        IbmMqEndpoint current = ibmMqEndpoint;
        if (current == null) {
            synchronized (endpointLock) {
                current = ibmMqEndpoint;
                if (current == null) {
                    current = new IbmMqEndpoint(store);
                    ibmMqEndpoint = current;
                }
            }
        }
        return current;
    }

    private String requireConfig(String key) {
        return Optional.ofNullable(store.getConfig(key))
                .filter(v -> !Objects.equals(v.trim(), ""))
                .orElseThrow(() -> new IllegalStateException("Missing required config: " + key));
    }
}
