package org.openjproxy.grpc.server;

import com.google.protobuf.ByteString;
import com.openjproxy.grpc.CallResourceRequest;
import com.openjproxy.grpc.CallResourceResponse;
import com.openjproxy.grpc.CallType;
import com.openjproxy.grpc.ConnectionDetails;
import com.openjproxy.grpc.DbName;
import com.openjproxy.grpc.LobDataBlock;
import com.openjproxy.grpc.LobReference;
import com.openjproxy.grpc.LobType;
import com.openjproxy.grpc.OpResult;
import com.openjproxy.grpc.ReadLobRequest;
import com.openjproxy.grpc.ResourceType;
import com.openjproxy.grpc.ResultSetFetchRequest;
import com.openjproxy.grpc.ResultType;
import com.openjproxy.grpc.SessionInfo;
import com.openjproxy.grpc.SessionTerminationStatus;
import com.openjproxy.grpc.SqlErrorType;
import com.openjproxy.grpc.StatementRequest;
import com.openjproxy.grpc.StatementServiceGrpc;
import com.openjproxy.grpc.TransactionInfo;
import com.openjproxy.grpc.TransactionStatus;
import com.zaxxer.hikari.HikariDataSource;
import io.grpc.Status;
import io.grpc.stub.ServerCallStreamObserver;
import io.grpc.stub.StreamObserver;
import lombok.Builder;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.io.input.ReaderInputStream;
import org.apache.commons.lang3.StringUtils;
import org.openjproxy.constants.CommonConstants;
import org.openjproxy.database.DatabaseUtils;
import org.openjproxy.datasource.ConnectionPoolProviderRegistry;
import org.openjproxy.datasource.PoolConfig;
import org.openjproxy.grpc.ProtoConverter;
import org.openjproxy.grpc.dto.OpQueryResult;
import org.openjproxy.grpc.dto.Parameter;
import org.openjproxy.grpc.server.lob.LobProcessor;
import org.openjproxy.grpc.server.pool.ConnectionPoolConfigurer;
import org.openjproxy.grpc.server.pool.DataSourceConfigurationManager;
import org.openjproxy.grpc.server.resultset.ResultSetWrapper;
import org.openjproxy.grpc.server.statement.ParameterHandler;
import org.openjproxy.grpc.server.statement.StatementFactory;
import org.openjproxy.grpc.server.utils.ConnectionHashGenerator;
import org.openjproxy.grpc.server.utils.DateTimeUtils;
import org.openjproxy.grpc.server.utils.DriverUtils;
import org.openjproxy.grpc.server.utils.MethodNameGenerator;
import org.openjproxy.grpc.server.utils.MethodReflectionUtils;
import org.openjproxy.grpc.server.utils.SessionInfoUtils;
import org.openjproxy.grpc.server.utils.StatementRequestValidator;
import org.openjproxy.grpc.server.utils.UrlParser;
import org.openjproxy.grpc.server.xa.XADataSourceFactory;
import org.openjproxy.xa.pool.XABackendSession;
import org.openjproxy.xa.pool.XATransactionRegistry;
import org.openjproxy.xa.pool.XidKey;
import org.openjproxy.xa.pool.spi.XAConnectionPoolProvider;

import javax.annotation.PostConstruct;
import javax.sql.DataSource;
import javax.sql.XAConnection;
import javax.sql.XADataSource;
import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.io.Reader;
import java.io.Writer;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.sql.Array;
import java.sql.Blob;
import java.sql.CallableStatement;
import java.sql.Clob;
import java.sql.Connection;
import java.sql.Date;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLDataException;
import java.sql.SQLException;
import java.sql.Savepoint;
import java.sql.Statement;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.ServiceLoader;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import static org.openjproxy.constants.CommonConstants.MAX_LOB_DATA_BLOCK_SIZE;
import static org.openjproxy.grpc.server.Constants.EMPTY_LIST;
import static org.openjproxy.grpc.server.Constants.EMPTY_MAP;
import static org.openjproxy.grpc.server.GrpcExceptionHandler.sendSQLExceptionMetadata;

@Slf4j
public class StatementServiceImpl extends StatementServiceGrpc.StatementServiceImplBase {

    private final Map<String, DataSource> datasourceMap = new ConcurrentHashMap<>();
    // Map for storing XADataSources (native database XADataSource, not Atomikos)
    private final Map<String, XADataSource> xaDataSourceMap = new ConcurrentHashMap<>();
    // XA Pool Provider for pooling XAConnections (loaded via SPI)
    private XAConnectionPoolProvider xaPoolProvider;
    // XA Transaction Registries (one per connection hash for isolated transaction management)
    private final Map<String, XATransactionRegistry> xaRegistries = new ConcurrentHashMap<>();
    private final SessionManager sessionManager;
    private final CircuitBreaker circuitBreaker;
    
    // Per-datasource slow query segregation managers
    private final Map<String, SlowQuerySegregationManager> slowQuerySegregationManagers = new ConcurrentHashMap<>();
    
    // Server configuration for creating segregation managers
    private final ServerConfiguration serverConfiguration;
    
    // Multinode XA coordinator for distributing transaction limits
    private static final MultinodeXaCoordinator xaCoordinator = new MultinodeXaCoordinator();
    
    // Cluster health tracker for monitoring health changes
    private final ClusterHealthTracker clusterHealthTracker = new ClusterHealthTracker();
    
    // Unpooled connection details (for passthrough mode when pooling is disabled)
    @Builder
    @Getter
    private static class UnpooledConnectionDetails {
        private final String url;
        private final String username;
        private final String password;
        private final long connectionTimeout;
    }
    private final Map<String, UnpooledConnectionDetails> unpooledConnectionDetailsMap = new ConcurrentHashMap<>();
    
    private static final List<String> INPUT_STREAM_TYPES = Arrays.asList("RAW", "BINARY VARYING", "BYTEA");
    private final Map<String, DbName> dbNameMap = new ConcurrentHashMap<>();

    private final static String RESULT_SET_METADATA_ATTR_PREFIX = "rsMetadata|";

    static {
        DriverUtils.registerDrivers();
    }

    public StatementServiceImpl(SessionManager sessionManager, CircuitBreaker circuitBreaker, ServerConfiguration serverConfiguration) {
        this.sessionManager = sessionManager;
        this.circuitBreaker = circuitBreaker;
        this.serverConfiguration = serverConfiguration;
        initializeXAPoolProvider();
    }

    /**
     * Initialize XA Pool Provider if XA pooling is enabled in configuration.
     * Loads the provider via ServiceLoader (Commons Pool 2 by default).
     */
    private void initializeXAPoolProvider() {
        // XA pooling is always enabled
        // Select the provider with the HIGHEST priority (100 = highest, 0 = lowest)
        
        try {
            ServiceLoader<XAConnectionPoolProvider> loader = ServiceLoader.load(XAConnectionPoolProvider.class);
            XAConnectionPoolProvider selectedProvider = null;
            int highestPriority = Integer.MIN_VALUE;
            
            for (XAConnectionPoolProvider provider : loader) {
                if (provider.isAvailable()) {
                    log.debug("Found available XA Pool Provider: {} (priority: {})", 
                            provider.getClass().getName(), provider.getPriority());
                    
                    if (provider.getPriority() > highestPriority) {
                        selectedProvider = provider;
                        highestPriority = provider.getPriority();
                    }
                }
            }
            
            if (selectedProvider != null) {
                this.xaPoolProvider = selectedProvider;
                log.info("Selected XA Pool Provider: {} (priority: {})", 
                        selectedProvider.getClass().getName(), selectedProvider.getPriority());
            } else {
                log.warn("No available XA Pool Provider found via ServiceLoader, XA pooling will be unavailable");
            }
        } catch (Exception e) {
            log.error("Failed to load XA Pool Provider: {}", e.getMessage(), e);
        }
    }

    /**
     * Gets the target server identifier from the incoming request.
     * Simply echoes back what the client sent without any override.
     */
    private String getTargetServer(SessionInfo incomingSessionInfo) {
        // Echo back the targetServer from incoming request, or return empty string if not present
        if (incomingSessionInfo != null && 
            incomingSessionInfo.getTargetServer() != null && 
            !incomingSessionInfo.getTargetServer().isEmpty()) {
            return incomingSessionInfo.getTargetServer();
        }
        
        // Return empty string if client didn't send targetServer
        return "";
    }
    
    /**
     * Processes cluster health from the client request and triggers pool rebalancing if needed.
     * This should be called for every request that includes SessionInfo with cluster health.
     */
    private void processClusterHealth(SessionInfo sessionInfo) {
        if (sessionInfo == null) {
            log.debug("[XA-REBALANCE-DEBUG] processClusterHealth: sessionInfo is null");
            return;
        }
        
        String clusterHealth = sessionInfo.getClusterHealth();
        String connHash = sessionInfo.getConnHash();
        
        log.debug("[XA-REBALANCE] processClusterHealth called: connHash={}, clusterHealth='{}', isXA={}, hasXARegistry={}", 
                connHash, clusterHealth, sessionInfo.getIsXA(), xaRegistries.containsKey(connHash));
        
        if (clusterHealth != null && !clusterHealth.isEmpty() && 
            connHash != null && !connHash.isEmpty()) {
            
            // Check if cluster health has changed
            boolean healthChanged = clusterHealthTracker.hasHealthChanged(connHash, clusterHealth);
            
            log.debug("[XA-REBALANCE] Cluster health check for {}: changed={}, current health='{}', isXA={}", 
                    connHash, healthChanged, clusterHealth, sessionInfo.getIsXA());
            
            if (healthChanged) {
                int healthyServerCount = clusterHealthTracker.countHealthyServers(clusterHealth);
                log.info("[XA-REBALANCE] Cluster health changed for {}, healthy servers: {}, triggering pool rebalancing, isXA={}", 
                        connHash, healthyServerCount, sessionInfo.getIsXA());
                
                // Update the pool coordinator with new healthy server count
                ConnectionPoolConfigurer.getPoolCoordinator().updateHealthyServers(connHash, healthyServerCount);
                
                // Apply pool size changes to non-XA HikariDataSource if present
                DataSource ds = datasourceMap.get(connHash);
                if (ds instanceof HikariDataSource) {
                    log.info("[XA-REBALANCE-DEBUG] Applying size changes to HikariDataSource for {}", connHash);
                    ConnectionPoolConfigurer.applyPoolSizeChanges(connHash, (HikariDataSource) ds);
                } else {
                    log.info("[XA-REBALANCE-DEBUG] No HikariDataSource found for {}", connHash);
                }
                
                // Apply pool size changes to XA registry if present
                XATransactionRegistry xaRegistry = xaRegistries.get(connHash);
                if (xaRegistry != null) {
                    log.info("[XA-REBALANCE-DEBUG] Found XA registry for {}, resizing", connHash);
                    MultinodePoolCoordinator.PoolAllocation allocation = 
                            ConnectionPoolConfigurer.getPoolCoordinator().getPoolAllocation(connHash);
                    
                    if (allocation != null) {
                        int newMaxPoolSize = allocation.getCurrentMaxPoolSize();
                        int newMinIdle = allocation.getCurrentMinIdle();
                        
                        log.info("[XA-REBALANCE-DEBUG] Resizing XA backend pool for {}: maxPoolSize={}, minIdle={}", 
                                connHash, newMaxPoolSize, newMinIdle);
                        
                        xaRegistry.resizeBackendPool(newMaxPoolSize, newMinIdle);
                    } else {
                        log.warn("[XA-REBALANCE-DEBUG] No pool allocation found for {}", connHash);
                    }
                } else if (sessionInfo.getIsXA()) {
                    // Only log missing XA registry for actual XA connections
                    log.info("[XA-REBALANCE-DEBUG] No XA registry found for XA connection {}", connHash);
                }
            } else {
                log.debug("[XA-REBALANCE-DEBUG] Cluster health unchanged for {}", connHash);
            }
        } else {
            log.info("[XA-REBALANCE-DEBUG] Skipping cluster health processing: clusterHealth={}, connHash={}", 
                    clusterHealth != null && !clusterHealth.isEmpty() ? "present" : "empty", 
                    connHash != null && !connHash.isEmpty() ? "present" : "empty");
        }
    }

    @Override
    public void connect(ConnectionDetails connectionDetails, StreamObserver<SessionInfo> responseObserver) {
        if (StringUtils.isBlank(connectionDetails.getUrl()) &&
            StringUtils.isBlank(connectionDetails.getUser()) &&
            StringUtils.isBlank(connectionDetails.getPassword())) {
            // Empty connection details - return empty session info - used for initial health checks only
            responseObserver.onNext(SessionInfo.newBuilder().build());
            responseObserver.onCompleted();
            return;
        }

        String connHash = ConnectionHashGenerator.hashConnectionDetails(connectionDetails);

        // Use default XA configuration values (deprecated pass-through properties no longer supported)
        int maxXaTransactions = org.openjproxy.constants.CommonConstants.DEFAULT_MAX_XA_TRANSACTIONS;
        long xaStartTimeoutMillis = org.openjproxy.constants.CommonConstants.DEFAULT_XA_START_TIMEOUT_MILLIS;
        
        log.info("connect connHash = {}, isXA = {}, maxXaTransactions = {}, xaStartTimeout = {}ms", 
                connHash, connectionDetails.getIsXA(), maxXaTransactions, xaStartTimeoutMillis);

        // Check if this is an XA connection request
        if (connectionDetails.getIsXA()) {
            // Check if multinode configuration is present for XA coordination
            List<String> serverEndpoints = connectionDetails.getServerEndpointsList();
            int actualMaxXaTransactions = maxXaTransactions;
            
            if (serverEndpoints != null && !serverEndpoints.isEmpty()) {
                // Multinode: calculate divided XA transaction limits
                MultinodeXaCoordinator.XaAllocation xaAllocation = 
                        xaCoordinator.calculateXaLimits(connHash, maxXaTransactions, serverEndpoints);
                
                actualMaxXaTransactions = xaAllocation.getCurrentMaxTransactions();
                
                log.info("Multinode XA coordination enabled for {}: {} servers, divided max transactions: {}", 
                        connHash, serverEndpoints.size(), actualMaxXaTransactions);
            }
            
            // Branch based on XA pooling configuration
            // XA Pool Provider SPI (always enabled)
            if (xaPoolProvider != null) {
                handleXAConnectionWithPooling(connectionDetails, connHash, actualMaxXaTransactions, 
                        xaStartTimeoutMillis, responseObserver);
            } else {
                log.error("XA Pool Provider not initialized");
                responseObserver.onError(Status.INTERNAL
                        .withDescription("XA Pool Provider not available")
                        .asRuntimeException());
            }
            return;
        }
        
        // Handle non-XA connection - check if pooling is enabled
        DataSource ds = this.datasourceMap.get(connHash);
        UnpooledConnectionDetails unpooledDetails = this.unpooledConnectionDetailsMap.get(connHash);
        
        if (ds == null && unpooledDetails == null) {
            try {
                // Get datasource-specific configuration from client properties
                Properties clientProperties = ConnectionPoolConfigurer.extractClientProperties(connectionDetails);
                DataSourceConfigurationManager.DataSourceConfiguration dsConfig = 
                        DataSourceConfigurationManager.getConfiguration(clientProperties);
                
                // Check if pooling is enabled
                if (!dsConfig.isPoolEnabled()) {
                    // Unpooled mode: store connection details for direct connection creation
                    unpooledDetails = UnpooledConnectionDetails.builder()
                            .url(UrlParser.parseUrl(connectionDetails.getUrl()))
                            .username(connectionDetails.getUser())
                            .password(connectionDetails.getPassword())
                            .connectionTimeout(dsConfig.getConnectionTimeout())
                            .build();
                    this.unpooledConnectionDetailsMap.put(connHash, unpooledDetails);
                    
                    log.info("Unpooled (passthrough) mode enabled for dataSource '{}' with connHash: {}", 
                            dsConfig.getDataSourceName(), connHash);
                } else {
                    // Pooled mode: create datasource with Connection Pool SPI (HikariCP by default)
                    // Get pool sizes - apply multinode coordination if needed
                    int maxPoolSize = dsConfig.getMaximumPoolSize();
                    int minIdle = dsConfig.getMinimumIdle();
                    
                    List<String> serverEndpoints = connectionDetails.getServerEndpointsList();
                    if (serverEndpoints != null && !serverEndpoints.isEmpty()) {
                        // Multinode: calculate divided pool sizes
                        MultinodePoolCoordinator.PoolAllocation allocation = 
                                ConnectionPoolConfigurer.getPoolCoordinator().calculatePoolSizes(
                                        connHash, maxPoolSize, minIdle, serverEndpoints);
                        
                        maxPoolSize = allocation.getCurrentMaxPoolSize();
                        minIdle = allocation.getCurrentMinIdle();
                        
                        log.info("Multinode pool coordination enabled for {}: {} servers, divided pool sizes: max={}, min={}", 
                                connHash, serverEndpoints.size(), maxPoolSize, minIdle);
                    }
                    
                    // Build PoolConfig from connection details and configuration
                    PoolConfig poolConfig = PoolConfig.builder()
                            .url(UrlParser.parseUrl(connectionDetails.getUrl()))
                            .username(connectionDetails.getUser())
                            .password(connectionDetails.getPassword())
                            .maxPoolSize(maxPoolSize)
                            .minIdle(minIdle)
                            .connectionTimeoutMs(dsConfig.getConnectionTimeout())
                            .idleTimeoutMs(dsConfig.getIdleTimeout())
                            .maxLifetimeMs(dsConfig.getMaxLifetime())
                            .metricsPrefix("OJP-Pool-" + dsConfig.getDataSourceName())
                            .build();
                    
                    // Create DataSource using the SPI (HikariCP by default)
                    ds = ConnectionPoolProviderRegistry.createDataSource(poolConfig);
                    this.datasourceMap.put(connHash, ds);
                    
                    // Create a slow query segregation manager for this datasource
                    createSlowQuerySegregationManagerForDatasource(connHash, maxPoolSize);
                    
                    log.info("Created new DataSource for dataSource '{}' with connHash: {} using provider: {}, maxPoolSize={}, minIdle={}", 
                            dsConfig.getDataSourceName(), connHash, 
                            ConnectionPoolProviderRegistry.getDefaultProvider().map(p -> p.id()).orElse("unknown"),
                            maxPoolSize, minIdle);
                }
                
            } catch (Exception e) {
                log.error("Failed to create datasource for connection hash {}: {}", connHash, e.getMessage(), e);
                SQLException sqlException = new SQLException("Failed to create datasource: " + e.getMessage(), e);
                sendSQLExceptionMetadata(sqlException, responseObserver);
                return;
            }
        }

        this.sessionManager.registerClientUUID(connHash, connectionDetails.getClientUUID());

        // For regular connections, just return session info without creating a session yet (lazy allocation)
        // Server does not populate targetServer - client will set it on future requests
        SessionInfo sessionInfo = SessionInfo.newBuilder()
                .setConnHash(connHash)
                .setClientUUID(connectionDetails.getClientUUID())
                .setIsXA(false)
                .build();

        responseObserver.onNext(sessionInfo);

        this.dbNameMap.put(connHash, DatabaseUtils.resolveDbName(connectionDetails.getUrl()));

        responseObserver.onCompleted();
    }
    
    /**
     * Handle XA connection using XA Pool Provider SPI (NEW PATH - enabled by default).
     * Creates pooled XA DataSource and allocates a XABackendSession immediately for the client.
     * <p>
     * Note: We allocate eagerly (not deferred) because XA applications expect getConnection()
     * to work immediately after creating an XAConnection, before xaStart() is called.
     * </p>
     */
    private void handleXAConnectionWithPooling(ConnectionDetails connectionDetails, String connHash,
                                               int actualMaxXaTransactions, long xaStartTimeoutMillis,
                                               StreamObserver<SessionInfo> responseObserver) {
        log.info("Using XA Pool Provider SPI for connHash: {}", connHash);
        
        // Get current serverEndpoints configuration
        List<String> currentServerEndpoints = connectionDetails.getServerEndpointsList();
        String currentEndpointsHash = (currentServerEndpoints == null || currentServerEndpoints.isEmpty()) 
                ? "NONE" 
                : String.join(",", currentServerEndpoints);
        
        // Check if we already have an XA registry for this connection hash
        XATransactionRegistry registry = xaRegistries.get(connHash);
        log.info("XA registry cache lookup for {}: exists={}, current serverEndpoints hash: {}", 
                connHash, registry != null, currentEndpointsHash);
        
        // Calculate what the pool sizes SHOULD be based on current configuration
        int expectedMaxPoolSize;
        int expectedMinIdle;
        boolean poolEnabled;
        try {
            Properties clientProperties = ConnectionPoolConfigurer.extractClientProperties(connectionDetails);
            DataSourceConfigurationManager.XADataSourceConfiguration xaConfig = 
                    DataSourceConfigurationManager.getXAConfiguration(clientProperties);
            expectedMaxPoolSize = xaConfig.getMaximumPoolSize();
            expectedMinIdle = xaConfig.getMinimumIdle();
            poolEnabled = xaConfig.isPoolEnabled();
            
            // Apply multinode coordination to get expected divided sizes
            if (currentServerEndpoints != null && !currentServerEndpoints.isEmpty()) {
                MultinodePoolCoordinator.PoolAllocation allocation = 
                        ConnectionPoolConfigurer.getPoolCoordinator().calculatePoolSizes(
                                connHash, expectedMaxPoolSize, expectedMinIdle, currentServerEndpoints);
                expectedMaxPoolSize = allocation.getCurrentMaxPoolSize();
                expectedMinIdle = allocation.getCurrentMinIdle();
            }
        } catch (Exception e) {
            log.warn("Failed to calculate expected pool sizes, will skip validation: {}", e.getMessage());
            expectedMaxPoolSize = -1;
            expectedMinIdle = -1;
            poolEnabled = true; // Default to pooled mode if config fails
        }
        
        // Check if registry exists and needs recreation due to configuration mismatch
        boolean needsRecreation = false;
        if (registry != null) {
            String registryEndpointsHash = registry.getServerEndpointsHash();
            int registryMaxPool = registry.getMaxPoolSize();
            int registryMinIdle = registry.getMinIdle();
            
            // Check if serverEndpoints changed
            if (registryEndpointsHash == null || !registryEndpointsHash.equals(currentEndpointsHash)) {
                log.warn("XA registry for {} has serverEndpoints mismatch: registry='{}' vs current='{}'. Will recreate.", 
                        connHash, registryEndpointsHash, currentEndpointsHash);
                needsRecreation = true;
            }
            // Check if pool sizes don't match expected values (indicates wrong coordination on first creation)
            else if (expectedMaxPoolSize > 0 && registryMaxPool != expectedMaxPoolSize) {
                log.warn("XA registry for {} has maxPoolSize mismatch: registry={} vs expected={}. Will recreate with correct multinode coordination.",
                        connHash, registryMaxPool, expectedMaxPoolSize);
                needsRecreation = true;
            }
            else if (expectedMinIdle > 0 && registryMinIdle != expectedMinIdle) {
                log.warn("XA registry for {} has minIdle mismatch: registry={} vs expected={}. Will recreate with correct multinode coordination.",
                        connHash, registryMinIdle, expectedMinIdle);
                needsRecreation = true;
            }
            
            if (needsRecreation) {
                // Close and remove old registry
                try {
                    registry.close();
                } catch (Exception e) {
                    log.warn("Failed to close old XA registry during recreation: {}", e.getMessage());
                }
                xaRegistries.remove(connHash);
                registry = null;
            }
        }
        
        if (registry == null) {
            log.info("Creating NEW XA registry for connHash: {} with serverEndpoints: {}", connHash, currentEndpointsHash);
            
            // Check if XA pooling is enabled
            if (!poolEnabled) {
                log.info("XA unpooled mode enabled for connHash: {}", connHash);
                
                // Handle unpooled XA connection
                handleUnpooledXAConnection(connectionDetails, connHash, responseObserver);
                return;
            }
            
            try {
                // Parse URL to remove OJP-specific prefix (same as non-XA path)
                String parsedUrl = UrlParser.parseUrl(connectionDetails.getUrl());
                
                // Get XA datasource configuration from client properties (uses XA-specific properties)
                Properties clientProperties = ConnectionPoolConfigurer.extractClientProperties(connectionDetails);
                DataSourceConfigurationManager.XADataSourceConfiguration xaConfig = 
                        DataSourceConfigurationManager.getXAConfiguration(clientProperties);
                
                // Get default pool sizes from XA configuration
                int maxPoolSize = xaConfig.getMaximumPoolSize();
                int minIdle = xaConfig.getMinimumIdle();
                
                log.info("XA pool BEFORE multinode coordination for {}: requested max={}, min={}", 
                        connHash, maxPoolSize, minIdle);
                
                // Apply multinode pool coordination if server endpoints provided
                List<String> serverEndpoints = connectionDetails.getServerEndpointsList();
                log.info("XA serverEndpoints list: null={}, size={}, endpoints={}", 
                        serverEndpoints == null, 
                        serverEndpoints == null ? 0 : serverEndpoints.size(),
                        serverEndpoints);
                
                if (serverEndpoints != null && !serverEndpoints.isEmpty()) {
                    // Multinode: divide pool sizes among servers
                    MultinodePoolCoordinator.PoolAllocation allocation = 
                            ConnectionPoolConfigurer.getPoolCoordinator().calculatePoolSizes(
                                    connHash, maxPoolSize, minIdle, serverEndpoints);
                    
                    maxPoolSize = allocation.getCurrentMaxPoolSize();
                    minIdle = allocation.getCurrentMinIdle();
                    
                    log.info("XA multinode pool coordination for {}: {} servers, divided sizes: max={}, min={}", 
                            connHash, serverEndpoints.size(), maxPoolSize, minIdle);
                } else {
                    log.info("XA multinode coordination SKIPPED for {}: serverEndpoints null or empty", connHash);
                }
                
                log.info("XA pool AFTER multinode coordination for {}: final max={}, min={}", 
                        connHash, maxPoolSize, minIdle);

                
                // Build configuration map for XA Pool Provider
                Map<String, String> xaPoolConfig = new HashMap<>();
                xaPoolConfig.put("xa.datasource.className", getXADataSourceClassName(parsedUrl));
                xaPoolConfig.put("xa.url", parsedUrl);
                xaPoolConfig.put("xa.username", connectionDetails.getUser());
                xaPoolConfig.put("xa.password", connectionDetails.getPassword());
                // Use calculated pool sizes (with multinode coordination if applicable)
                xaPoolConfig.put("xa.maxPoolSize", String.valueOf(maxPoolSize));
                xaPoolConfig.put("xa.minIdle", String.valueOf(minIdle));
                xaPoolConfig.put("xa.connectionTimeoutMs", String.valueOf(xaConfig.getConnectionTimeout()));
                xaPoolConfig.put("xa.idleTimeoutMs", String.valueOf(xaConfig.getIdleTimeout()));
                xaPoolConfig.put("xa.maxLifetimeMs", String.valueOf(xaConfig.getMaxLifetime()));
                // Evictor configuration
                xaPoolConfig.put("xa.timeBetweenEvictionRunsMs", String.valueOf(xaConfig.getTimeBetweenEvictionRuns()));
                xaPoolConfig.put("xa.numTestsPerEvictionRun", String.valueOf(xaConfig.getNumTestsPerEvictionRun()));
                xaPoolConfig.put("xa.softMinEvictableIdleTimeMs", String.valueOf(xaConfig.getSoftMinEvictableIdleTime()));
                
                // Create pooled XA DataSource via provider
                log.info("[XA-POOL-CREATE] Creating XA pool for connHash={}, serverEndpointsHash={}, config=(max={}, min={})",
                        connHash, currentEndpointsHash, maxPoolSize, minIdle);
                Object pooledXADataSource = xaPoolProvider.createXADataSource(xaPoolConfig);
                
                // Create XA Transaction Registry with serverEndpoints hash and pool sizes for validation
                registry = new XATransactionRegistry(xaPoolProvider, pooledXADataSource, currentEndpointsHash, maxPoolSize, minIdle);
                xaRegistries.put(connHash, registry);
                
                // Initialize pool with minIdle connections immediately after creation
                // Without this, the pool starts empty and only creates connections on demand
                log.info("[XA-POOL-INIT] Initializing XA pool with minIdle={} connections for connHash={}", minIdle, connHash);
                registry.resizeBackendPool(maxPoolSize, minIdle);
                
                // Create slow query segregation manager for XA
