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

import cz.moneta.test.harness.connectors.VaultConnector;
import cz.moneta.test.harness.connectors.messaging.KafkaConnector;
import cz.moneta.test.harness.constants.HarnessConfigConstants;
import cz.moneta.test.harness.context.StoreAccessor;
import cz.moneta.test.harness.endpoints.Endpoint;
import cz.moneta.test.harness.messaging.model.ReceivedMessage;

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

public class KafkaEndpoint implements Endpoint {

    private static final String CONFIG_BOOTSTRAP_SERVERS = "messaging.kafka.bootstrap-servers";
    private static final String CONFIG_SCHEMA_REGISTRY_URL = "messaging.kafka.schema-registry-url";
    private static final String CONFIG_VAULT_PATH = "vault.path.messaging.kafka";

    private static final String CONFIG_API_KEY = "messaging.kafka.api-key";
    private static final String CONFIG_API_SECRET = "messaging.kafka.api-secret";
    private static final String CONFIG_SCHEMA_REGISTRY_API_KEY = "messaging.kafka.schema-registry-api-key";
    private static final String CONFIG_SCHEMA_REGISTRY_API_SECRET = "messaging.kafka.schema-registry-api-secret";

    private static final String VAULT_API_KEY = "apiKey";
    private static final String VAULT_API_SECRET = "apiSecret";
    private static final String VAULT_SCHEMA_REGISTRY_API_KEY = "schemaRegistryApiKey";
    private static final String VAULT_SCHEMA_REGISTRY_API_SECRET = "schemaRegistryApiSecret";

    private final StoreAccessor store;
    private volatile KafkaConnector connector;
    private final Object connectorLock = new Object();

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

    public void send(String topic, String key, String payload, Map<String, String> headers) {
        getConnector().send(topic, key, payload, headers);
    }

    public ReceivedMessage receive(String topic, Predicate<ReceivedMessage> filter, Duration timeout) {
        List<ReceivedMessage> messages = getConnector().receive(topic, filter, timeout);
        return messages.isEmpty() ? null : messages.get(0);
    }

    @Override
    public void close() {
        KafkaConnector current = connector;
        if (current != null) {
            current.close();
        }
    }

    private KafkaConnector getConnector() {
        KafkaConnector current = connector;
        if (current == null) {
            synchronized (connectorLock) {
                current = connector;
                if (current == null) {
                    current = createConnector();
                    connector = current;
                }
            }
        }
        return current;
    }

    private KafkaConnector createConnector() {
        String bootstrapServers = requireConfig(CONFIG_BOOTSTRAP_SERVERS);
        String schemaRegistryUrl = requireConfig(CONFIG_SCHEMA_REGISTRY_URL);
        String apiKey = resolveSecret(CONFIG_API_KEY, VAULT_API_KEY);
        String apiSecret = resolveSecret(CONFIG_API_SECRET, VAULT_API_SECRET);
        String schemaRegistryApiKey = resolveSecret(CONFIG_SCHEMA_REGISTRY_API_KEY, VAULT_SCHEMA_REGISTRY_API_KEY);
        String schemaRegistryApiSecret = resolveSecret(CONFIG_SCHEMA_REGISTRY_API_SECRET, VAULT_SCHEMA_REGISTRY_API_SECRET);

        return new KafkaConnector(
                bootstrapServers,
                apiKey,
                apiSecret,
                schemaRegistryUrl,
                schemaRegistryApiKey,
                schemaRegistryApiSecret
        );
    }

    private String resolveSecret(String localConfigKey, String vaultKey) {
        return Optional.ofNullable(store.getConfig(localConfigKey))
                .or(() -> readFromVault(vaultKey))
                .orElseThrow(() -> new IllegalStateException("Missing messaging secret: " + localConfigKey));
    }

    private Optional<String> readFromVault(String key) {
        String path = store.getConfig(CONFIG_VAULT_PATH);
        if (path == null || path.isBlank()) {
            return Optional.empty();
        }
        String basePath = store.getConfig("vault.path.base");
        String resolvedPath = path.startsWith("/") || basePath == null || basePath.isBlank()
                ? path
                : basePath + "/" + path;

        return createVaultConnector().flatMap(vault -> vault.getValue(resolvedPath, key));
    }

    private Optional<VaultConnector> createVaultConnector() {
        return Optional.ofNullable(store.getConfig(HarnessConfigConstants.VAULT_URL_CONFIG))
                .flatMap(url -> Optional.ofNullable(store.getConfig(HarnessConfigConstants.VAULT_USERNAME_CONFIG))
                        .flatMap(username -> Optional.ofNullable(store.getConfig(HarnessConfigConstants.VAULT_PASSWORD_CONFIG))
                                .map(password -> new VaultConnector(url, username, password))));
    }

    private String requireConfig(String key) {
        return Optional.ofNullable(store.getConfig(key))
                .orElseThrow(() -> new IllegalStateException("Missing required config: " + key));
    }
}
