package com.easyagents.federation.sql.federation;

import com.easyagents.federation.sql.source.SourceId;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.time.Instant;
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.Objects;
import java.util.Set;

/**
 * 一次编译期间不可变引用的统计快照。
 *
 * <p>构造时会复制完整统计映射，确保版本、数据与有效期属于同一个冻结视图。</p>
 */
public final class FederationStatisticsSnapshot {

    private final String version;
    private final Instant capturedAt;
    private final Map<TableKey, FederationTableStatistics> statisticsByTable;
    private final Instant validUntil;

    /**
     * 创建统计快照。
     *
     * @param version 稳定快照版本
     * @param statisticsByTable 按物理表索引的统计映射
     */
    public FederationStatisticsSnapshot(
        String version,
        Map<TableKey, FederationTableStatistics> statisticsByTable
    ) {
        this(version, Instant.now(), statisticsByTable);
    }

    /**
     * 创建带固定评估时刻的统计快照。
     *
     * @param version 稳定快照版本
     * @param capturedAt 快照捕获和有效期评估时刻
     * @param statisticsByTable 按物理表索引的统计映射
     */
    public FederationStatisticsSnapshot(
        String version,
        Instant capturedAt,
        Map<TableKey, FederationTableStatistics> statisticsByTable
    ) {
        this.version = version == null || version.isBlank() ? "none" : version;
        this.capturedAt = Objects.requireNonNull(capturedAt, "capturedAt");
        this.statisticsByTable = Map.copyOf(new LinkedHashMap<>(
            statisticsByTable == null ? Map.of() : statisticsByTable
        ));
        this.validUntil = this.statisticsByTable.values().stream()
            .filter(statistics -> {
                FederationStatisticsStatus status = statistics.effectiveStatus(capturedAt);
                return status == FederationStatisticsStatus.COMPLETE
                    || status == FederationStatisticsStatus.PARTIAL;
            })
            .map(FederationTableStatistics::expiresAt)
            .min(Instant::compareTo)
            .orElse(Instant.MAX);
    }

    /**
     * 返回空统计快照。
     *
     * @return 空快照
     */
    public static FederationStatisticsSnapshot empty() {
        return new FederationStatisticsSnapshot("none", Instant.now(), Map.of());
    }

    /**
     * 返回快照版本。
     *
     * @return 稳定版本
     */
    public String version() {
        return version;
    }

    /**
     * 返回本次编译统一使用的统计有效期评估时刻。
     *
     * @return 快照捕获时刻
     */
    public Instant capturedAt() {
        return capturedAt;
    }

    /**
     * 返回该快照内最早的统计失效时间。
     *
     * @return 最早失效时间；空快照为 {@link Instant#MAX}
     */
    public Instant validUntil() {
        return validUntil;
    }

    /**
     * 查询冻结快照中的表统计。
     *
     * @param sourceId 物理源
     * @param schema Source Definition 暴露的逻辑 Schema 名称
     * @param table 表名称
     * @return 表统计；缺失时为 {@code null}
     */
    public FederationTableStatistics statistics(
        SourceId sourceId,
        String schema,
        String table
    ) {
        return statisticsByTable.get(new TableKey(sourceId, schema, table));
    }

    /**
     * 截取指定物理表的稳定统计身份与最早失效时间。
     *
     * <p>该结果只依赖查询实际引用的表。未引用表的刷新不会使已有计划失效。</p>
     *
     * @param tables 查询实际引用的物理表键
     * @return 查询级统计选择结果
     */
    public Selection select(Set<TableKey> tables) {
        List<TableKey> ordered = (tables == null ? Set.<TableKey>of() : tables).stream()
            .sorted(Comparator
                .comparing((TableKey key) -> key.sourceId().value())
                .thenComparing(TableKey::schema)
                .thenComparing(TableKey::table))
            .toList();
        MessageDigest digest = sha256();
        Instant selectedValidUntil = Instant.MAX;
        for (TableKey key : ordered) {
            update(digest, key.sourceId().value());
            update(digest, key.schema());
            update(digest, key.table());
            FederationTableStatistics statistics = statisticsByTable.get(key);
            if (statistics == null) {
                update(digest, "missing");
                continue;
            }
            FederationStatisticsStatus effectiveStatus = statistics.effectiveStatus(capturedAt);
            update(digest, effectiveStatus.name());
            update(digest, Double.toString(statistics.estimatedRows()));
            update(digest, Long.toString(statistics.averageRowWidthBytes()));
            update(digest, statistics.collectedAt().toString());
            update(digest, statistics.source());
            update(digest, statistics.expiresAt().toString());
            statistics.columns().entrySet().stream()
                .sorted(Map.Entry.comparingByKey(String.CASE_INSENSITIVE_ORDER))
                .forEach(entry -> {
                    update(digest, entry.getKey().toLowerCase(Locale.ROOT));
                    update(digest, entry.getValue().toString());
                });
            statistics.uniqueKeys().stream()
                .map(columns -> columns.stream()
                    .map(value -> value.toLowerCase(Locale.ROOT))
                    .sorted()
                    .toList())
                .map(columns -> String.join("\u0001", columns))
                .sorted()
                .forEach(value -> update(digest, value));
            if ((effectiveStatus == FederationStatisticsStatus.COMPLETE
                || effectiveStatus == FederationStatisticsStatus.PARTIAL)
                && statistics.expiresAt().isBefore(selectedValidUntil)) {
                selectedValidUntil = statistics.expiresAt();
            }
        }
        return new Selection(
            HexFormat.of().formatHex(digest.digest()),
            selectedValidUntil
        );
    }

    /**
     * 查询级统计选择结果。
     *
     * @param fingerprint 实际引用表统计的稳定指纹
     * @param validUntil 实际引用表统计的最早失效时间
     */
    public record Selection(String fingerprint, Instant validUntil) {

        /** 校验查询级统计选择结果。 */
        public Selection {
            fingerprint = Objects.requireNonNull(fingerprint, "fingerprint");
            validUntil = validUntil == null ? Instant.MAX : validUntil;
        }
    }

    private static MessageDigest sha256() {
        try {
            return MessageDigest.getInstance("SHA-256");
        } catch (NoSuchAlgorithmException exception) {
            throw new IllegalStateException("SHA-256 is unavailable", exception);
        }
    }

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

    /**
     * 数据源表统计键。
     *
     * @param sourceId 物理源
     * @param schema Source Definition 暴露的逻辑 Schema 名称
     * @param table 表名称
     */
    public record TableKey(SourceId sourceId, String schema, String table) {

        /**
         * 规范化物理表键，保证常见数据库标识符大小写差异不影响命中。
         */
        public TableKey {
            sourceId = Objects.requireNonNull(sourceId, "sourceId");
            schema = normalize(schema);
            table = normalize(table);
        }

        private static String normalize(String value) {
            return value == null ? "" : value.toLowerCase(Locale.ROOT);
        }
    }
}
