package com.easyagents.federation.sql.runtime;

import com.easyagents.federation.sql.adapter.AdapterCompatibility;
import com.easyagents.federation.sql.adapter.FederationSqlAdapterProvider;
import com.easyagents.federation.sql.api.FederationSqlErrorCode;
import com.easyagents.federation.sql.api.FederationSqlException;
import com.easyagents.federation.sql.source.FederationDataSourceHandle;
import com.easyagents.federation.sql.source.FederationSourceDefinition;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.sql.SqlDialect;

/**
 * 单个 sourceId + revision 的节点本地 Calcite 与 DataSource 运行对象。
 */
final class SourceRuntime {

    private final FederationSourceDefinition definition;
    private final String sourceChecksum;
    private final String runtimeFingerprint;
    private final FederationDataSourceHandle handle;
    private final FederationSqlAdapterProvider adapter;
    private final SqlDialect dialect;
    private final SchemaPlus rootSchema;
    private final SchemaPlus defaultSchema;
    private final AdapterCompatibility compatibility;
    private final Consumer<SourceRuntime> closedListener;
    private final AtomicInteger leases = new AtomicInteger();
    private final AtomicBoolean retiring = new AtomicBoolean();
    private final AtomicBoolean closed = new AtomicBoolean();

    SourceRuntime(
        FederationSourceDefinition definition,
        FederationDataSourceHandle handle,
        FederationSqlAdapterProvider adapter,
        SqlDialect dialect,
        SchemaPlus rootSchema,
        SchemaPlus defaultSchema,
        AdapterCompatibility compatibility,
        Consumer<SourceRuntime> closedListener
    ) {
        this.definition = definition;
        this.sourceChecksum = definition.checksum();
        this.runtimeFingerprint = handle.fingerprint().cacheKey();
        this.handle = handle;
        this.adapter = adapter;
        this.dialect = dialect;
        this.rootSchema = rootSchema;
        this.defaultSchema = defaultSchema;
        this.compatibility = compatibility;
        this.closedListener = closedListener;
    }

    /**
     * 返回不可变数据源定义。
     *
     * @return 数据源定义
     */
    FederationSourceDefinition definition() {
        return definition;
    }

    /**
     * 返回 Runtime 构建时缓存的 Definition 校验和。
     *
     * @return Definition 校验和
     */
    String sourceChecksum() {
        return sourceChecksum;
    }

    /**
     * 返回 Runtime 构建时缓存的数据库与驱动指纹。
     *
     * @return Runtime 指纹
     */
    String runtimeFingerprint() {
        return runtimeFingerprint;
    }

    FederationDataSourceHandle handle() {
        return handle;
    }

    FederationSqlAdapterProvider adapter() {
        return adapter;
    }

    SqlDialect dialect() {
        return dialect;
    }

    SchemaPlus rootSchema() {
        return rootSchema;
    }

    SchemaPlus defaultSchema() {
        return defaultSchema;
    }

    AdapterCompatibility compatibility() {
        return compatibility;
    }

    RuntimeLease acquire() {
        while (true) {
            if (retiring.get() || closed.get()) {
                throw new FederationSqlException(
                    FederationSqlErrorCode.SOURCE_REVISION_NOT_READY,
                    "source runtime is retiring: " + definition.sourceId()
                );
            }
            leases.incrementAndGet();
            if (!retiring.get() && !closed.get()) {
                return new RuntimeLease(this);
            }
            release();
        }
    }

    void retire() {
        retiring.set(true);
        closeWhenUnused();
    }

    void forceClose() {
        retiring.set(true);
        closeHandle();
    }

    private void release() {
        int remaining = leases.decrementAndGet();
        if (remaining < 0) {
            throw new IllegalStateException("runtime lease count became negative");
        }
        closeWhenUnused();
    }

    private void closeWhenUnused() {
        if (retiring.get() && leases.get() == 0) {
            closeHandle();
        }
    }

    private synchronized void closeHandle() {
        if (closed.get()) {
            closedListener.accept(this);
            return;
        }
        RuntimeException failure = null;
        for (int attempt = 0; attempt < 2; attempt++) {
            try {
                handle.close();
                closed.set(true);
                closedListener.accept(this);
                return;
            } catch (RuntimeException exception) {
                if (failure == null) {
                    failure = exception;
                } else {
                    failure.addSuppressed(exception);
                }
            }
        }
        throw failure;
    }

    /**
     * 对旧 Runtime 的引用租约，防止流式游标消费期间提前关闭 Handle。
     */
    static final class RuntimeLease implements AutoCloseable {

        private final SourceRuntime runtime;
        private final AtomicBoolean released = new AtomicBoolean();

        private RuntimeLease(SourceRuntime runtime) {
            this.runtime = runtime;
        }

        SourceRuntime runtime() {
            return runtime;
        }

        /**
         * 幂等释放 Runtime 租约。
         */
        @Override
        public void close() {
            if (released.compareAndSet(false, true)) {
                runtime.release();
            }
        }
    }
}
