package com.easyagents.federation.sql.runtime;

import com.easyagents.federation.sql.api.FederationSqlErrorCode;
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.InputStream;
import java.io.Reader;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 为单源 Adapter 游标补充统一查询指标。
 */
final class MetricsFederationResultCursor
    implements FederationResultCursor, CancellationAwareFederationCursor {

    private final FederationResultCursor delegate;
    private final FederationQueryMetricsTracker metrics;
    private final AtomicBoolean closed = new AtomicBoolean();

    /**
     * 创建指标包装游标。
     *
     * @param delegate Adapter 游标
     * @param metrics 指标跟踪器
     */
    MetricsFederationResultCursor(
        FederationResultCursor delegate,
        FederationQueryMetricsTracker metrics
    ) {
        this.delegate = delegate;
        this.metrics = metrics;
    }

    /** {@inheritDoc} */
    @Override
    public QueryId queryId() {
        return delegate.queryId();
    }

    /** {@inheritDoc} */
    @Override
    public List<FederationColumn> columns() {
        return delegate.columns();
    }

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

    /** {@inheritDoc} */
    @Override
    public void markCancelled() {
        metrics.markCancelled();
    }

    /** {@inheritDoc} */
    @Override
    public void markTimedOut() {
        metrics.markTimedOut();
    }

    /** {@inheritDoc} */
    @Override
    public void markFailed(FederationSqlErrorCode errorCode) {
        metrics.markFailed(errorCode);
    }

    /** {@inheritDoc} */
    @Override
    public boolean next() {
        boolean present = delegate.next();
        if (present) {
            metrics.recordSingleSourceRow();
        } else {
            metrics.finishFragment("fragment-1");
            metrics.finishSuccessfully();
        }
        return present;
    }

    /** {@inheritDoc} */
    @Override
    public Object getObject(int columnIndex) {
        return delegate.getObject(columnIndex);
    }

    /** {@inheritDoc} */
    @Override
    public InputStream getBinaryStream(int columnIndex) {
        return delegate.getBinaryStream(columnIndex);
    }

    /** {@inheritDoc} */
    @Override
    public Reader getCharacterStream(int columnIndex) {
        return delegate.getCharacterStream(columnIndex);
    }

    /** {@inheritDoc} */
    @Override
    public List<Object> row() {
        return delegate.row();
    }

    /** {@inheritDoc} */
    @Override
    public void close() {
        if (closed.compareAndSet(false, true)) {
            try {
                delegate.close();
            } finally {
                metrics.finishClosed();
            }
        }
    }
}
