package com.easyagents.federation.sql.adapter.jdbc;

import com.easyagents.federation.sql.api.FederationSqlErrorCode;
import com.easyagents.federation.sql.api.FederationSqlException;
import com.easyagents.federation.sql.execute.FederationColumn;
import com.easyagents.federation.sql.execute.FederationFragmentExecutionContext;
import com.easyagents.federation.sql.execute.FederationFragmentExecutor;
import com.easyagents.federation.sql.execute.FederationResultCursor;
import com.easyagents.federation.sql.execute.SqlParameter;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.SQLTimeoutException;
import java.sql.SQLTransientConnectionException;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;

/**
 * 直接使用 PreparedStatement 执行目标数据库 SQL 的流式 JDBC 执行器。
 */
final class JdbcFederationFragmentExecutor implements FederationFragmentExecutor {

    /**
     * 获取连接、应用只读限制、绑定参数并返回持有全部资源的流式游标。
     *
     * @param context 执行上下文
     * @return 流式游标
     */
    @Override
    public FederationResultCursor execute(FederationFragmentExecutionContext context) {
        Connection connection = null;
        PreparedStatement statement = null;
        boolean registered = false;
        boolean connectionAcquired = false;
        try {
            context.executionGuard().ensureAllowed();
            long connectionStarted = System.nanoTime();
            connection = context.dataSource().getConnection();
            connectionAcquired = true;
            context.observer().connectionAcquired(System.nanoTime() - connectionStarted);
            context.executionGuard().ensureAllowed();
            configureConnection(connection, context);
            statement = connection.prepareStatement(
                context.sql(),
                ResultSet.TYPE_FORWARD_ONLY,
                ResultSet.CONCUR_READ_ONLY
            );
            applyOptions(statement, context);
            bindParameters(statement, context.parameters());
            context.statementLifecycle().register(statement);
            registered = true;
            context.executionGuard().ensureAllowed();
            long executionStarted = System.nanoTime();
            ResultSet resultSet = statement.executeQuery();
            context.executionGuard().ensureAllowed();
            context.observer().databaseExecutionCompleted(System.nanoTime() - executionStarted);
            List<FederationColumn> columns = readColumns(resultSet.getMetaData());
            return new JdbcFederationResultCursor(
                context.queryId(),
                columns,
                resultSet,
                statement,
                connection,
                context.statementLifecycle(),
                context.executionGuard(),
                context.observer()
            );
        } catch (SQLException | RuntimeException exception) {
            if (!connectionAcquired) {
                // 连接池等待可能跨过统一截止时间；总超时或显式取消应保持为查询终态。
                context.executionGuard().ensureAllowed();
            }
            if (registered) {
                context.statementLifecycle().unregister(statement);
            }
            closeAfterFailure(statement, connection, exception);
            if (exception instanceof FederationSqlException federationSqlException) {
                throw federationSqlException;
            }
            boolean connectionTimedOut = !connectionAcquired
                && (exception instanceof SQLTransientConnectionException
                    || exception instanceof SQLTimeoutException);
            if (connectionTimedOut) {
                throw new FederationSqlException(
                    FederationSqlErrorCode.CONNECTION_ACQUISITION_TIMEOUT,
                    "timed out while acquiring a JDBC connection",
                    exception
                );
            }
            if (!connectionAcquired) {
                throw new FederationSqlException(
                    FederationSqlErrorCode.CONNECTION_ACQUISITION_FAILED,
                    "failed to acquire a JDBC connection",
                    exception
                );
            }
            throw JdbcFailureClassifier.classify(
                context.statementLifecycle(),
                exception,
                "JDBC query timed out",
                "JDBC query was cancelled",
                "JDBC query execution failed"
            );
        }
    }

    private static void applyOptions(
        PreparedStatement statement,
        FederationFragmentExecutionContext context
    ) throws SQLException {
        int fetchSize = effectiveFetchSize(context);
        if (fetchSize != 0) {
            statement.setFetchSize(fetchSize);
        }
        if (context.options().maxRows() > 0) {
            statement.setMaxRows(context.options().maxRows());
        }
        int queryTimeout = context.executionGuard().boundedQueryTimeoutSeconds(
            context.options().queryTimeoutSeconds()
        );
        if (queryTimeout > 0) {
            statement.setQueryTimeout(queryTimeout);
        }
    }

    private static void configureConnection(
        Connection connection,
        FederationFragmentExecutionContext context
    ) throws SQLException {
        if (!connection.isReadOnly()) {
            connection.setReadOnly(true);
        }
        String product = context.compatibility().databaseProduct().toLowerCase(Locale.ROOT);
        // PostgreSQL 只有在事务模式下才会按正 fetchSize 使用服务端游标。
        if (product.contains("postgres") && context.options().fetchSize() > 0
            && connection.getAutoCommit()) {
            connection.setAutoCommit(false);
        }
    }

    private static int effectiveFetchSize(FederationFragmentExecutionContext context) {
        String product = context.compatibility().databaseProduct().toLowerCase(Locale.ROOT);
        if (product.contains("mysql")
            && "legacy".equalsIgnoreCase(context.adapterOptions().get("mysqlStreamingMode"))) {
            // Connector/J 旧式逐行流需要显式 MIN_VALUE；默认仍使用正 fetchSize + useCursorFetch。
            return Integer.MIN_VALUE;
        }
        return context.options().fetchSize();
    }

    private static void bindParameters(PreparedStatement statement, List<SqlParameter> parameters)
        throws SQLException {
        for (int index = 0; index < parameters.size(); index++) {
            SqlParameter parameter = parameters.get(index);
            int jdbcIndex = index + 1;
            if (parameter.value() == null) {
                statement.setNull(jdbcIndex, parameter.jdbcType());
            } else {
                statement.setObject(jdbcIndex, parameter.value(), parameter.jdbcType());
            }
        }
    }

    private static List<FederationColumn> readColumns(ResultSetMetaData metadata) throws SQLException {
        List<FederationColumn> columns = new ArrayList<>(metadata.getColumnCount());
        for (int index = 1; index <= metadata.getColumnCount(); index++) {
            columns.add(new FederationColumn(
                index,
                metadata.getColumnLabel(index),
                metadata.getColumnType(index),
                metadata.getColumnTypeName(index),
                metadata.isNullable(index) != ResultSetMetaData.columnNoNulls
            ));
        }
        return List.copyOf(columns);
    }

    private static void closeAfterFailure(
        PreparedStatement statement,
        Connection connection,
        Throwable original
    ) {
        closeAndSuppress(statement, original);
        closeAndSuppress(connection, original);
    }

    private static void closeAndSuppress(AutoCloseable closeable, Throwable original) {
        if (closeable == null) {
            return;
        }
        try {
            closeable.close();
        } catch (Exception closeException) {
            original.addSuppressed(closeException);
        }
    }
}
