package com.easyagents.federation.sql.federation;

import com.easyagents.federation.sql.source.SourceId;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HexFormat;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.HashSet;
import java.util.Set;

/**
 * 调用方传入的不可变查询范围，描述 SQL 可见的物理数据源 Binding。
 *
 * <p>该定义不保存 DataSource、连接池或凭据。Core 只在编译和执行期间解析它，
 * 虚拟数据源的持久化、发布和分布式一致性由调用方负责。</p>
 *
 * @param definitionId 调用方定义标识
 * @param revision 查询范围版本
 * @param bindings Binding 名称到物理数据源定义的映射
 * @param defaultBinding 默认 Binding 名称
 * @param logicalTables 查询方可见的逻辑表映射；空列表保留原始物理表解析语义
 * @param executionPolicy 调用方联邦资源上限
 */
public record FederationQueryScopeDefinition(
    String definitionId,
    long revision,
    Map<String, FederationSourceBindingDefinition> bindings,
    String defaultBinding,
    List<FederationLogicalTableDefinition> logicalTables,
    FederationExecutionPolicy executionPolicy
) implements Serializable {

    /**
     * 校验并创建不可变查询范围。
     */
    public FederationQueryScopeDefinition {
        if (definitionId == null || definitionId.isBlank()) {
            throw new IllegalArgumentException("definitionId must not be blank");
        }
        if (revision < 0) {
            throw new IllegalArgumentException("scope revision must not be negative");
        }
        if (bindings == null || bindings.isEmpty()) {
            throw new IllegalArgumentException("scope bindings must not be empty");
        }
        LinkedHashMap<String, FederationSourceBindingDefinition> copied = new LinkedHashMap<>();
        Set<String> normalizedBindingNames = new HashSet<>();
        bindings.forEach((bindingName, binding) -> {
            if (bindingName == null || bindingName.isBlank() || binding == null) {
                throw new IllegalArgumentException("binding name and definition must be provided");
            }
            if (!normalizedBindingNames.add(bindingName.toUpperCase(Locale.ROOT))) {
                throw new IllegalArgumentException(
                    "binding names must be unique ignoring unquoted identifier case"
                );
            }
            copied.put(bindingName, binding);
        });
        bindings = Collections.unmodifiableMap(copied);
        if (defaultBinding == null || !bindings.containsKey(defaultBinding)) {
            throw new IllegalArgumentException("defaultBinding must reference a declared binding");
        }
        List<FederationLogicalTableDefinition> copiedTables = logicalTables == null
            ? List.of()
            : List.copyOf(logicalTables);
        Set<String> normalizedLogicalNames = new HashSet<>();
        for (FederationLogicalTableDefinition table : copiedTables) {
            if (table == null) {
                throw new IllegalArgumentException("logical table definition must not be null");
            }
            if (!bindings.containsKey(table.bindingName())) {
                throw new IllegalArgumentException(
                    "logical table binding must reference a declared binding: "
                        + table.bindingName()
                );
            }
            if (!normalizedLogicalNames.add(table.logicalName().toUpperCase(Locale.ROOT))) {
                throw new IllegalArgumentException(
                    "logical table names must be unique ignoring unquoted identifier case"
                );
            }
        }
        logicalTables = copiedTables;
        executionPolicy = executionPolicy == null
            ? FederationExecutionPolicy.basic()
            : executionPolicy;
    }

    /**
     * 创建不声明逻辑表映射的兼容查询范围。
     *
     * @param definitionId 调用方定义标识
     * @param revision 查询范围版本
     * @param bindings Binding 名称到物理数据源定义的映射
     * @param defaultBinding 默认 Binding 名称
     * @param executionPolicy 调用方联邦资源上限
     */
    public FederationQueryScopeDefinition(
        String definitionId,
        long revision,
        Map<String, FederationSourceBindingDefinition> bindings,
        String defaultBinding,
        FederationExecutionPolicy executionPolicy
    ) {
        this(
            definitionId,
            revision,
            bindings,
            defaultBinding,
            List.of(),
            executionPolicy
        );
    }

    /**
     * 创建单物理数据源查询范围。
     *
     * @param sourceId 物理数据源标识，同时作为默认 Binding 名称
     * @param minimumRevision 最低 Definition 版本
     * @return 单源查询范围
     */
    public static FederationQueryScopeDefinition single(
        SourceId sourceId,
        long minimumRevision
    ) {
        if (sourceId == null) {
            throw new IllegalArgumentException("sourceId must not be null");
        }
        return single(
            "source:" + sourceId.value(),
            minimumRevision,
            sourceId.value(),
            sourceId,
            minimumRevision
        );
    }

    /**
     * 创建调用方管理的虚拟联邦查询范围。
     *
     * @param definitionId 查询范围标识
     * @param revision 查询范围版本
     * @param bindings SQL 逻辑 Binding 到物理数据源的映射
     * @param defaultBinding 默认 Binding
     * @param executionPolicy 调用方资源上限
     * @return 虚拟联邦查询范围
     */
    public static FederationQueryScopeDefinition virtual(
        String definitionId,
        long revision,
        Map<String, FederationSourceBindingDefinition> bindings,
        String defaultBinding,
        FederationExecutionPolicy executionPolicy
    ) {
        return new FederationQueryScopeDefinition(
            definitionId,
            revision,
            bindings,
            defaultBinding,
            List.of(),
            executionPolicy
        );
    }

    /**
     * 创建带逻辑表映射的虚拟联邦查询范围。
     *
     * @param definitionId 查询范围标识
     * @param revision 查询范围版本
     * @param bindings SQL 逻辑 Binding 到物理数据源的映射
     * @param defaultBinding 默认 Binding
     * @param logicalTables 查询方可见的逻辑表映射
     * @param executionPolicy 调用方资源上限
     * @return 虚拟联邦查询范围
     */
    public static FederationQueryScopeDefinition virtual(
        String definitionId,
        long revision,
        Map<String, FederationSourceBindingDefinition> bindings,
        String defaultBinding,
        List<FederationLogicalTableDefinition> logicalTables,
        FederationExecutionPolicy executionPolicy
    ) {
        return new FederationQueryScopeDefinition(
            definitionId,
            revision,
            bindings,
            defaultBinding,
            logicalTables,
            executionPolicy
        );
    }

    /**
     * 创建带独立范围版本的单物理数据源查询范围。
     *
     * @param definitionId 查询范围标识
     * @param scopeRevision 查询范围版本
     * @param bindingName 默认 Binding 名称
     * @param sourceId 物理数据源标识
     * @param minimumRevision 最低 Definition 版本
     * @return 单源查询范围
     */
    public static FederationQueryScopeDefinition single(
        String definitionId,
        long scopeRevision,
        String bindingName,
        SourceId sourceId,
        long minimumRevision
    ) {
        return new FederationQueryScopeDefinition(
            definitionId,
            scopeRevision,
            Map.of(bindingName, FederationSourceBindingDefinition.of(sourceId, minimumRevision)),
            bindingName,
            List.of(),
            FederationExecutionPolicy.basic()
        );
    }

    /**
     * 返回默认 Binding。
     *
     * @return 默认 Binding 定义
     */
    public FederationSourceBindingDefinition defaultBindingDefinition() {
        return bindings.get(defaultBinding);
    }

    /**
     * 返回用于计划缓存与跨节点一致性判断的稳定摘要。
     *
     * @return SHA-256 十六进制摘要
     */
    public String checksum() {
        try {
            MessageDigest digest = MessageDigest.getInstance("SHA-256");
            update(digest, "federation-query-scope-v2");
            update(digest, definitionId);
            update(digest, Long.toString(revision));
            update(digest, defaultBinding);
            List<Map.Entry<String, FederationSourceBindingDefinition>> orderedBindings =
                new ArrayList<>(bindings.entrySet());
            orderedBindings.sort(Map.Entry.comparingByKey());
            updateCount(digest, orderedBindings.size());
            for (Map.Entry<String, FederationSourceBindingDefinition> entry : orderedBindings) {
                update(digest, "binding");
                update(digest, entry.getKey());
                FederationSourceBindingDefinition binding = entry.getValue();
                update(digest, binding.sourceId().value());
                update(digest, Long.toString(binding.minimumRevision()));
                List<Map.Entry<String, String>> mappings =
                    new ArrayList<>(binding.schemaMappings().entrySet());
                mappings.sort(Comparator.comparing(Map.Entry::getKey));
                updateCount(digest, mappings.size());
                for (Map.Entry<String, String> mapping : mappings) {
                    update(digest, "schema-mapping");
                    update(digest, mapping.getKey());
                    update(digest, mapping.getValue());
                }
            }
            List<FederationLogicalTableDefinition> orderedTables =
                new ArrayList<>(logicalTables);
            orderedTables.sort(Comparator.comparing(
                table -> table.logicalName().toUpperCase(Locale.ROOT)
            ));
            updateCount(digest, orderedTables.size());
            for (FederationLogicalTableDefinition table : orderedTables) {
                update(digest, "logical-table");
                update(digest, table.logicalName());
                update(digest, table.bindingName());
                update(digest, table.schemaName());
                update(digest, table.sourceTableName());
            }
            update(digest, "execution-policy");
            update(digest, Integer.toString(executionPolicy.maximumReferencedSources()));
            update(digest, Integer.toString(executionPolicy.maximumFragments()));
            update(digest, Integer.toString(executionPolicy.maximumConcurrentFragments()));
            update(digest, Long.toString(executionPolicy.maximumIntermediateRows()));
            update(digest, Long.toString(executionPolicy.maximumIntermediateBytes()));
            update(digest, Long.toString(executionPolicy.maximumExecutionTimeMillis()));
            return HexFormat.of().formatHex(digest.digest());
        } catch (NoSuchAlgorithmException exception) {
            throw new IllegalStateException("SHA-256 is not available", exception);
        }
    }

    private static void update(MessageDigest digest, String value) {
        byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
        updateCount(digest, bytes.length);
        digest.update(bytes);
    }

    private static void updateCount(MessageDigest digest, int value) {
        digest.update(ByteBuffer.allocate(Integer.BYTES).putInt(value).array());
    }
}
