package com.easyagents.federation.sql.runtime;

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.FederationQueryMetricsSnapshot;
import com.easyagents.federation.sql.execute.FederationResultCursor;
import com.easyagents.federation.sql.execute.QueryId;
import java.io.FilterInputStream;
import java.io.FilterReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.Reader;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 由 Core 统一释放 Runtime lease、准入许可并追踪活动游标的包装器。
 */
final class ManagedFederationResultCursor implements FederationResultCursor {

    private final FederationResultCursor delegate;
    private final AutoCloseable resources;
    private final Runnable onClose;
    private final AtomicBoolean closed = new AtomicBoolean();

    /**
     * 创建 Core 托管游标。
     *
     * @param delegate Adapter 游标
     * @param resources Core 资源
     * @param onClose 关闭后的追踪清理动作
     */
    ManagedFederationResultCursor(
        FederationResultCursor delegate,
        AutoCloseable resources,
        Runnable onClose
    ) {
        this.delegate = delegate;
        this.resources = resources;
        this.onClose = onClose;
    }

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

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

    /** {@inheritDoc} */
    @Override
    public FederationQueryMetricsSnapshot metrics() {
        return delegate.metrics();
    }

    /**
     * 在 Engine 主动取消后、资源关闭前定稿取消指标。
     */
    void markCancelled() {
        if (delegate instanceof CancellationAwareFederationCursor aware) {
            aware.markCancelled();
        }
    }

    /**
     * 在 Engine 执行时限到达后、资源关闭前定稿超时指标。
     */
    void markTimedOut() {
        if (delegate instanceof CancellationAwareFederationCursor aware) {
            aware.markTimedOut();
        }
    }

    /**
     * 移动至下一行，Adapter 异常时同步释放 Core 资源。
     *
     * @return 是否存在下一行
     */
    @Override
    public boolean next() {
        try {
            return delegate.next();
        } catch (RuntimeException exception) {
            markFailure(exception);
            closeAndSuppress(exception);
            throw exception;
        }
    }

    /**
     * 读取当前行列值，Adapter 异常时同步释放 Core 资源。
     *
     * @param columnIndex 从 1 开始的列序号
     * @return 列值
     */
    @Override
    public Object getObject(int columnIndex) {
        try {
            return delegate.getObject(columnIndex);
        } catch (RuntimeException exception) {
            markFailure(exception);
            closeAndSuppress(exception);
            throw exception;
        }
    }

    /**
     * 委托流式读取二进制列，并在 Adapter 异常时同步释放 Core 资源。
     *
     * @param columnIndex 从 1 开始的列序号
     * @return 二进制流
     */
    @Override
    public InputStream getBinaryStream(int columnIndex) {
        try {
            InputStream stream = delegate.getBinaryStream(columnIndex);
            return stream == null ? null : managed(stream);
        } catch (RuntimeException exception) {
            markFailure(exception);
            closeAndSuppress(exception);
            throw exception;
        }
    }

    /**
     * 委托流式读取字符列，并在 Adapter 异常时同步释放 Core 资源。
     *
     * @param columnIndex 从 1 开始的列序号
     * @return 字符流
     */
    @Override
    public Reader getCharacterStream(int columnIndex) {
        try {
            Reader reader = delegate.getCharacterStream(columnIndex);
            return reader == null ? null : managed(reader);
        } catch (RuntimeException exception) {
            markFailure(exception);
            closeAndSuppress(exception);
            throw exception;
        }
    }

    /**
     * 复制当前行，Adapter 异常时同步释放 Core 资源。
     *
     * @return 当前行值
     */
    @Override
    public List<Object> row() {
        try {
            return delegate.row();
        } catch (RuntimeException exception) {
            markFailure(exception);
            closeAndSuppress(exception);
            throw exception;
        }
    }

    /**
     * 先摘除 Engine 游标追踪，再关闭 Adapter 游标并确定性释放 Core 资源。
     */
    @Override
    public void close() {
        if (!closed.compareAndSet(false, true)) {
            return;
        }
        RuntimeException failure = null;
        // 追踪移除必须早于 QueryRegistration 释放，避免并发 cancel 写入伪预取消状态。
        try {
            onClose.run();
        } catch (RuntimeException exception) {
            failure = exception;
        }
        try {
            delegate.close();
        } catch (RuntimeException exception) {
            if (failure == null) {
                failure = exception;
            } else {
                failure.addSuppressed(exception);
            }
        }
        try {
            resources.close();
        } catch (Exception exception) {
            RuntimeException wrapped = exception instanceof RuntimeException runtimeException
                ? runtimeException
                : new IllegalStateException("failed to close managed query resources", exception);
            if (failure == null) {
                failure = wrapped;
            } else {
                failure.addSuppressed(wrapped);
            }
        }
        if (failure != null) {
            throw failure;
        }
    }

    private InputStream managed(InputStream stream) {
        return new FilterInputStream(stream) {
            @Override
            public int read() throws IOException {
                try {
                    return super.read();
                } catch (IOException | RuntimeException exception) {
                    failStream(exception);
                    throw exception;
                }
            }

            @Override
            public int read(byte[] bytes, int offset, int length) throws IOException {
                try {
                    return super.read(bytes, offset, length);
                } catch (IOException | RuntimeException exception) {
                    failStream(exception);
                    throw exception;
                }
            }

            @Override
            public long skip(long count) throws IOException {
                try {
                    return super.skip(count);
                } catch (IOException | RuntimeException exception) {
                    failStream(exception);
                    throw exception;
                }
            }

            @Override
            public int available() throws IOException {
                try {
                    return super.available();
                } catch (IOException | RuntimeException exception) {
                    failStream(exception);
                    throw exception;
                }
            }

            @Override
            public void reset() throws IOException {
                try {
                    super.reset();
                } catch (IOException | RuntimeException exception) {
                    failStream(exception);
                    throw exception;
                }
            }

            @Override
            public void close() throws IOException {
                try {
                    super.close();
                } catch (IOException | RuntimeException exception) {
                    failStream(exception);
                    throw exception;
                }
            }
        };
    }

    private Reader managed(Reader reader) {
        return new FilterReader(reader) {
            @Override
            public int read() throws IOException {
                try {
                    return super.read();
                } catch (IOException | RuntimeException exception) {
                    failStream(exception);
                    throw exception;
                }
            }

            @Override
            public int read(char[] characters, int offset, int length) throws IOException {
                try {
                    return super.read(characters, offset, length);
                } catch (IOException | RuntimeException exception) {
                    failStream(exception);
                    throw exception;
                }
            }

            @Override
            public long skip(long count) throws IOException {
                try {
                    return super.skip(count);
                } catch (IOException | RuntimeException exception) {
                    failStream(exception);
                    throw exception;
                }
            }

            @Override
            public boolean ready() throws IOException {
                try {
                    return super.ready();
                } catch (IOException | RuntimeException exception) {
                    failStream(exception);
                    throw exception;
                }
            }

            @Override
            public void mark(int readAheadLimit) throws IOException {
                try {
                    super.mark(readAheadLimit);
                } catch (IOException | RuntimeException exception) {
                    failStream(exception);
                    throw exception;
                }
            }

            @Override
            public void reset() throws IOException {
                try {
                    super.reset();
                } catch (IOException | RuntimeException exception) {
                    failStream(exception);
                    throw exception;
                }
            }

            @Override
            public void close() throws IOException {
                try {
                    super.close();
                } catch (IOException | RuntimeException exception) {
                    failStream(exception);
                    throw exception;
                }
            }
        };
    }

    private void failStream(Throwable exception) {
        markFailure(exception);
        closeAndSuppress(exception);
    }

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

    private void markFailure(Throwable exception) {
        if (delegate instanceof CancellationAwareFederationCursor aware) {
            FederationSqlErrorCode errorCode = exception instanceof FederationSqlException sqlException
                ? sqlException.errorCode()
                : FederationSqlErrorCode.EXECUTION_FAILED;
            if (errorCode == FederationSqlErrorCode.QUERY_TIMEOUT) {
                aware.markTimedOut();
            } else if (errorCode == FederationSqlErrorCode.QUERY_CANCELLED) {
                aware.markCancelled();
            } else {
                aware.markFailed(errorCode);
            }
        }
    }
}
