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.FederationExecutionGuard;
import com.easyagents.federation.sql.execute.FederationExecutionObserver;
import com.easyagents.federation.sql.execute.FederationResultCursor;
import com.easyagents.federation.sql.execute.QueryId;
import com.easyagents.federation.sql.execute.StatementLifecycle;
import java.io.FilterInputStream;
import java.io.FilterReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.Reader;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 持有 ResultSet、Statement、Connection 和 Engine 资源的 JDBC 流式游标。
 */
final class JdbcFederationResultCursor implements FederationResultCursor {

    private final QueryId queryId;
    private final List<FederationColumn> columns;
    private final ResultSet resultSet;
    private final PreparedStatement statement;
    private final Connection connection;
    private final StatementLifecycle statementLifecycle;
    private final FederationExecutionGuard executionGuard;
    private final FederationExecutionObserver observer;
    private final AtomicBoolean closed = new AtomicBoolean();
    private final AtomicBoolean firstRowObserved = new AtomicBoolean();
    private final long resultSetCreatedNanos = System.nanoTime();

    /**
     * 创建 JDBC 流式游标。
     *
     * @param queryId 查询标识
     * @param columns 结果列
     * @param resultSet JDBC ResultSet
     * @param statement JDBC Statement
     * @param connection JDBC Connection
     * @param statementLifecycle Statement 生命周期回调
     * @param executionGuard 查询取消与截止时间检查器
     * @param observer Fragment 执行阶段观察器
     */
    JdbcFederationResultCursor(
        QueryId queryId,
        List<FederationColumn> columns,
        ResultSet resultSet,
        PreparedStatement statement,
        Connection connection,
        StatementLifecycle statementLifecycle,
        FederationExecutionGuard executionGuard,
        FederationExecutionObserver observer
    ) {
        this.queryId = queryId;
        this.columns = List.copyOf(columns);
        this.resultSet = resultSet;
        this.statement = statement;
        this.connection = connection;
        this.statementLifecycle = statementLifecycle;
        this.executionGuard = executionGuard;
        this.observer = observer;
    }

    /**
     * 创建不采集阶段指标的兼容 JDBC 游标。
     *
     * @param queryId 查询标识
     * @param columns 结果列
     * @param resultSet JDBC ResultSet
     * @param statement JDBC Statement
     * @param connection JDBC Connection
     * @param statementLifecycle Statement 生命周期
     */
    JdbcFederationResultCursor(
        QueryId queryId,
        List<FederationColumn> columns,
        ResultSet resultSet,
        PreparedStatement statement,
        Connection connection,
        StatementLifecycle statementLifecycle
    ) {
        this(
            queryId,
            columns,
            resultSet,
            statement,
            connection,
            statementLifecycle,
            FederationExecutionGuard.none(),
            FederationExecutionObserver.none()
        );
    }

    /**
     * 返回查询标识。
     *
     * @return 查询标识
     */
    @Override
    public QueryId queryId() {
        return queryId;
    }

    /**
     * 返回结果列。
     *
     * @return 结果列
     */
    @Override
    public List<FederationColumn> columns() {
        return columns;
    }

    /**
     * 移动到下一行；读取结束时保留资源直至调用方关闭游标。
     *
     * @return 是否存在下一行
     */
    @Override
    public boolean next() {
        ensureOpen();
        try {
            boolean present = resultSet.next();
            ensureAllowedAfterRead();
            if (present && firstRowObserved.compareAndSet(false, true)) {
                observer.firstRowAvailable(System.nanoTime() - resultSetCreatedNanos);
            }
            return present;
        } catch (SQLException exception) {
            closeWithSuppressed(exception);
            throw JdbcFailureClassifier.classify(
                statementLifecycle,
                exception,
                "JDBC result read timed out",
                "JDBC query was cancelled",
                "failed to advance JDBC result cursor"
            );
        }
    }

    /**
     * 读取当前行指定列。
     *
     * @param columnIndex 从 1 开始的列序号
     * @return 列值
     */
    @Override
    public Object getObject(int columnIndex) {
        ensureOpen();
        try {
            Object value = resultSet.getObject(columnIndex);
            ensureAllowedAfterRead();
            return value;
        } catch (SQLException exception) {
            closeWithSuppressed(exception);
            throw JdbcFailureClassifier.classify(
                statementLifecycle,
                exception,
                "JDBC result read timed out",
                "JDBC query was cancelled",
                "failed to read JDBC result column " + columnIndex
            );
        }
    }

    /**
     * 以 JDBC 流读取二进制列。
     *
     * @param columnIndex 从 1 开始的列序号
     * @return 二进制流；SQL NULL 返回 null
     */
    @Override
    public InputStream getBinaryStream(int columnIndex) {
        ensureOpen();
        try {
            InputStream stream = resultSet.getBinaryStream(columnIndex);
            ensureAllowedAfterRead();
            return stream == null ? null : new GuardedInputStream(stream, columnIndex);
        } catch (SQLException exception) {
            throw readFailure(columnIndex, exception);
        }
    }

    /**
     * 以 JDBC 流读取字符列。
     *
     * @param columnIndex 从 1 开始的列序号
     * @return 字符流；SQL NULL 返回 null
     */
    @Override
    public Reader getCharacterStream(int columnIndex) {
        ensureOpen();
        try {
            Reader reader = resultSet.getCharacterStream(columnIndex);
            ensureAllowedAfterRead();
            return reader == null ? null : new GuardedReader(reader, columnIndex);
        } catch (SQLException exception) {
            throw readFailure(columnIndex, exception);
        }
    }

    /**
     * 复制当前行；Engine 不缓存返回行。
     *
     * @return 当前行列值
     */
    @Override
    public List<Object> row() {
        ensureOpen();
        List<Object> row = new ArrayList<>(columns.size());
        for (int index = 1; index <= columns.size(); index++) {
            row.add(getObject(index));
        }
        return Collections.unmodifiableList(row);
    }

    private void ensureOpen() {
        // 异步关闭可能先于消费线程到达，优先保留取消或超时终态语义。
        try {
            executionGuard.ensureAllowed();
        } catch (RuntimeException exception) {
            closeWithSuppressed(exception);
            throw exception;
        }
        if (closed.get()) {
            throw new FederationSqlException(
                FederationSqlErrorCode.EXECUTION_FAILED,
                "result cursor is closed"
            );
        }
    }

    private void ensureAllowedAfterRead() {
        try {
            executionGuard.ensureAllowed();
        } catch (RuntimeException exception) {
            closeWithSuppressed(exception);
            throw exception;
        }
    }

    /**
     * 幂等关闭 JDBC 资源并最终释放准入许可与 Runtime lease。
     */
    @Override
    public void close() {
        if (!closed.compareAndSet(false, true)) {
            return;
        }
        FederationSqlException failure = null;
        try {
            resultSet.close();
        } catch (SQLException exception) {
            failure = closeFailure("ResultSet", exception);
        }
        try {
            statementLifecycle.unregister(statement);
        } catch (RuntimeException exception) {
            failure = append(failure, closeFailure("Statement lifecycle", exception));
        }
        try {
            statement.close();
        } catch (SQLException exception) {
            failure = append(failure, closeFailure("PreparedStatement", exception));
        }
        try {
            connection.close();
        } catch (SQLException exception) {
            failure = append(failure, closeFailure("Connection", exception));
        }
        if (failure != null) {
            throw failure;
        }
    }

    private void closeWithSuppressed(Throwable original) {
        try {
            close();
        } catch (RuntimeException closeException) {
            original.addSuppressed(closeException);
        }
    }

    /**
     * 将流式列读取异常映射为统一错误并确定性关闭 JDBC 资源。
     *
     * @param columnIndex 列序号
     * @param exception JDBC 或流读取异常
     * @return 统一 Federation 异常
     */
    private FederationSqlException readFailure(int columnIndex, Throwable exception) {
        closeWithSuppressed(exception);
        return JdbcFailureClassifier.classify(
            statementLifecycle,
            exception,
            "JDBC result read timed out",
            "JDBC query was cancelled",
            "failed to stream JDBC result column " + columnIndex
        );
    }

    /**
     * 对二进制列的每次实际读取执行查询终态检查。
     */
    private final class GuardedInputStream extends FilterInputStream {

        private final int columnIndex;

        /**
         * 创建受查询生命周期保护的二进制流。
         *
         * @param delegate JDBC 驱动流
         * @param columnIndex 列序号
         */
        private GuardedInputStream(InputStream delegate, int columnIndex) {
            super(delegate);
            this.columnIndex = columnIndex;
        }

        /** {@inheritDoc} */
        @Override
        public int read() throws IOException {
            ensureOpen();
            try {
                int value = super.read();
                ensureAllowedAfterRead();
                return value;
            } catch (IOException exception) {
                throw readFailure(columnIndex, exception);
            }
        }

        /** {@inheritDoc} */
        @Override
        public int read(byte[] buffer, int offset, int length) throws IOException {
            ensureOpen();
            try {
                int read = super.read(buffer, offset, length);
                ensureAllowedAfterRead();
                return read;
            } catch (IOException exception) {
                throw readFailure(columnIndex, exception);
            }
        }

        /** {@inheritDoc} */
        @Override
        public long skip(long count) throws IOException {
            ensureOpen();
            try {
                long skipped = super.skip(count);
                ensureAllowedAfterRead();
                return skipped;
            } catch (IOException exception) {
                throw readFailure(columnIndex, exception);
            }
        }
    }

    /**
     * 对字符列的每次实际读取执行查询终态检查。
     */
    private final class GuardedReader extends FilterReader {

        private final int columnIndex;

        /**
         * 创建受查询生命周期保护的字符流。
         *
         * @param delegate JDBC 驱动 Reader
         * @param columnIndex 列序号
         */
        private GuardedReader(Reader delegate, int columnIndex) {
            super(delegate);
            this.columnIndex = columnIndex;
        }

        /** {@inheritDoc} */
        @Override
        public int read() throws IOException {
            ensureOpen();
            try {
                int value = super.read();
                ensureAllowedAfterRead();
                return value;
            } catch (IOException exception) {
                throw readFailure(columnIndex, exception);
            }
        }

        /** {@inheritDoc} */
        @Override
        public int read(char[] buffer, int offset, int length) throws IOException {
            ensureOpen();
            try {
                int read = super.read(buffer, offset, length);
                ensureAllowedAfterRead();
                return read;
            } catch (IOException exception) {
                throw readFailure(columnIndex, exception);
            }
        }

        /** {@inheritDoc} */
        @Override
        public long skip(long count) throws IOException {
            ensureOpen();
            try {
                long skipped = super.skip(count);
                ensureAllowedAfterRead();
                return skipped;
            } catch (IOException exception) {
                throw readFailure(columnIndex, exception);
            }
        }
    }

    private static FederationSqlException closeFailure(String resource, Exception cause) {
        return new FederationSqlException(
            FederationSqlErrorCode.RESOURCE_CLOSE_FAILED,
            "failed to close JDBC " + resource,
            cause
        );
    }

    private static FederationSqlException append(
        FederationSqlException failure,
        FederationSqlException next
    ) {
        if (failure == null) {
            return next;
        }
        failure.addSuppressed(next);
        return failure;
    }
}
