XuqmGroup-Server/tenant-service/src/main/java/com/xuqm/tenant/service/SystemUpdateService.java

package com.xuqm.tenant.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Service;

import com.xuqm.tenant.config.PrivateDeploymentProperties;

import javax.sql.DataSource;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Statement;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Consumer;
import java.util.stream.Collectors;

@Service
public class SystemUpdateService {

    private static final Logger log = LoggerFactory.getLogger(SystemUpdateService.class);

    // nginx 最后重启，确保它能获取到其他服务修复后的配置
    private static final List<String> OTHER_SERVICES = List.of(
            "file-service", "tenant-web", "im-service", "push-service", "update-service", "license-service", "nginx"
    );

    private static final Set<String> ALLOWED_LOG_SERVICES = Set.of(
            "tenant-service", "file-service", "im-service", "push-service",
            "update-service", "license-service", "nginx", "tenant-web"
    );

    @Value("${PRIVATE_DEPLOY_ROOT:/opt/xuqm-private}")
    private String deployRoot;

    private final DataSource dataSource;
    private final PrivateDeploymentProperties deployProps;

    public SystemUpdateService(DataSource dataSource, PrivateDeploymentProperties deployProps) {
        this.dataSource = dataSource;
        this.deployProps = deployProps;
    }

    public String getDeployRoot() { return deployRoot; }

    // ── 公开接口 ────────────────────────────────────────────────────────────────

    /**
     * 返回当前正在运行的服务名列表。
     * 使用 docker ps --format "{{.Label \"com.docker.compose.service\"}}" 枚举运行中容器的服务标签，
     * 不依赖 compose 文件路径，公有云/私有云均可用。
     * 结果已过滤为 ALLOWED_LOG_SERVICES 白名单内的服务。
     */
    public List<String> getRunningServices() {
        try {
            Process p = new ProcessBuilder(
                    "docker", "ps",
                    "--format", "{{.Label \"com.docker.compose.service\"}}"
            ).redirectErrorStream(true).start();
            String out = new String(p.getInputStream().readAllBytes(), StandardCharsets.UTF_8).trim();
            p.waitFor();
            if (out.isEmpty()) return List.of();
            return Arrays.stream(out.split("\n"))
                    .map(String::trim)
                    .filter(s -> !s.isEmpty() && ALLOWED_LOG_SERVICES.contains(s))
                    .distinct()
                    .collect(Collectors.toList());
        } catch (Exception e) {
            log.error("failed to list running services", e);
            return List.of();
        }
    }

    /**
     * 获取指定服务最近 N 行日志（上限 1000）。
     * 通过 docker ps 标签找到容器 ID 后使用 docker logs，不依赖 compose 文件路径。
     */
    public String getServiceLogs(String service, int lines) {
        if (!ALLOWED_LOG_SERVICES.contains(service)) {
            throw new IllegalArgumentException("不允许查看此服务的日志: " + service);
        }
        int safeLines = Math.min(Math.max(lines, 10), 1000);
        try {
            // 找到对应服务的容器 ID（可能有多个，取第一个）
            Process psProc = new ProcessBuilder(
                    "docker", "ps",
                    "--filter", "label=com.docker.compose.service=" + service,
                    "--format", "{{.ID}}"
            ).redirectErrorStream(true).start();
            String containerId = new String(psProc.getInputStream().readAllBytes(), StandardCharsets.UTF_8).trim();
            psProc.waitFor();

            if (containerId.isEmpty()) {
                return "（服务 " + service + " 当前没有运行中的容器）";
            }
            // 取第一行（如有多个容器）
            String firstId = containerId.split("\n")[0].trim();

            Process logsProc = new ProcessBuilder(
                    "docker", "logs", "--tail", String.valueOf(safeLines), "--timestamps", firstId
            ).redirectErrorStream(true).start();
            String out = new String(logsProc.getInputStream().readAllBytes(), StandardCharsets.UTF_8);
            int exitCode = logsProc.waitFor();
            if (exitCode != 0 && out.isBlank()) {
                throw new RuntimeException("docker logs 返回非零退出码: " + exitCode);
            }
            return out;
        } catch (IllegalArgumentException e) {
            throw e;
        } catch (Exception e) {
            log.error("failed to fetch logs for service {}", service, e);
            throw new RuntimeException("获取日志失败: " + e.getMessage());
        }
    }

    /** 读取镜像内打包的 VERSION 文件，返回当前版本号，文件不存在时返回 "unknown"。 */
    public String readCurrentVersion() {
        // 优先读镜像内 /app/VERSION
        Path containerVersion = Paths.get("/app/VERSION");
        try {
            if (Files.exists(containerVersion)) {
                return Files.readString(containerVersion).trim();
            }
        } catch (IOException ignored) {}
        // 兼容旧路径：宿主机挂载目录
        Path hostVersion = Paths.get(deployRoot, "VERSION");
        try {
            if (Files.exists(hostVersion)) {
                return Files.readString(hostVersion).trim();
            }
        } catch (IOException ignored) {}
        return "unknown";
    }

    /**
     * 读取本地各服务的版本号（从 Docker 镜像 LABEL 中读取）。
     */
    public Map<String, String> readLocalServiceVersions() {
        Map<String, String> versions = new LinkedHashMap<>();
        List<String> allServices = new ArrayList<>(OTHER_SERVICES);
        allServices.add("tenant-service");
        for (String svc : allServices) {
            try {
                Process p = new ProcessBuilder(
                        "docker", "inspect", "--format",
                        "{{index .Config.Labels \"com.xuqm.version\"}}",
                        "xuqmgroup/" + svc + ":latest"
                ).redirectErrorStream(true).start();
                String out = new String(p.getInputStream().readAllBytes(), StandardCharsets.UTF_8).trim();
                p.waitFor();
                versions.put(svc, out.isEmpty() ? "unknown" : out);
            } catch (Exception e) {
                versions.put(svc, "unknown");
            }
        }
        return versions;
    }

    /**
     * 读取版本清单（versions.json）。
     *
     * 优先级：
     * 1. 若 .env 中配置了 VERSIONS_MANIFEST_URL，则从远端拉取并缓存到本地
     * 2. 回退到本地 {deployRoot}/versions.json
     *
     * 私有部署服务器须在 .env 中添加：
     *   VERSIONS_MANIFEST_URL=https://<ops-server>/api/versions/manifest
     * 以便自动感知公有端发布的新版本。
     */
    @SuppressWarnings("unchecked")
    public Map<String, Object> readRemoteVersions() {
        String manifestUrl = readEnvValue(Paths.get(deployRoot, ".env"), "VERSIONS_MANIFEST_URL");
        if (manifestUrl != null && !manifestUrl.isBlank()) {
            try {
                HttpClient client = HttpClient.newBuilder()
                        .connectTimeout(Duration.ofSeconds(5))
                        .build();
                HttpRequest req = HttpRequest.newBuilder()
                        .uri(URI.create(manifestUrl.trim()))
                        .timeout(Duration.ofSeconds(10))
                        .GET()
                        .build();
                HttpResponse<String> resp = client.send(req, HttpResponse.BodyHandlers.ofString());
                if (resp.statusCode() == 200) {
                    Map<String, Object> remote = new ObjectMapper().readValue(resp.body(), new TypeReference<>() {});
                    // 缓存到本地，避免网络故障时无法读取
                    Files.writeString(Paths.get(deployRoot, "versions.json"), resp.body(),
                            StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING);
                    log.info("versions.json refreshed from {}", manifestUrl);
                    return remote;
                } else {
                    log.warn("versions manifest fetch returned HTTP {}", resp.statusCode());
                }
            } catch (Exception e) {
                log.warn("Failed to fetch versions from {}: {}", manifestUrl, e.getMessage());
            }
        }

        // 回退到本地缓存
        Path localVersions = Paths.get(deployRoot, "versions.json");
        if (Files.exists(localVersions)) {
            try {
                String json = Files.readString(localVersions);
                return new ObjectMapper().readValue(json, new TypeReference<>() {});
            } catch (Exception e) {
                log.warn("Failed to read local versions.json: {}", e.getMessage());
            }
        }
        return Map.of();
    }

    /**
     * 检查是否有可用更新。
     * 对比本地平台版本与远端 versions.json 中的平台版本。
     */
    @SuppressWarnings("unchecked")
    public Map<String, Object> checkForUpdates() {
        String localPlatform = readCurrentVersion();
        Map<String, Object> remote = readRemoteVersions();
        String remotePlatform = remote.getOrDefault("platformVersion", "").toString();
        boolean hasUpdate = !remotePlatform.isEmpty() && !remotePlatform.equals(localPlatform);

        Map<String, Object> result = new LinkedHashMap<>();
        result.put("currentVersion", localPlatform);
        result.put("latestVersion", remotePlatform);
        result.put("hasUpdate", hasUpdate);
        result.put("releasedAt", remote.getOrDefault("releasedAt", ""));
        result.put("changelog", remote.getOrDefault("changelog", ""));

        // 各服务版本对比
        Map<String, String> localVersions = readLocalServiceVersions();
        Map<String, Object> remoteServices = remote.get("services") instanceof Map
                ? (Map<String, Object>) remote.get("services") : Map.of();

        Map<String, Object> services = new LinkedHashMap<>();
        for (String svc : localVersions.keySet()) {
            Map<String, Object> svcInfo = new LinkedHashMap<>();
            svcInfo.put("current", localVersions.get(svc));
            Object remoteSvc = remoteServices.get(svc);
            if (remoteSvc instanceof Map) {
                Map<String, Object> rMap = (Map<String, Object>) remoteSvc;
                svcInfo.put("latest", rMap.getOrDefault("version", ""));
                svcInfo.put("changed", rMap.getOrDefault("changed", false));
            } else {
                svcInfo.put("latest", "");
                svcInfo.put("changed", false);
            }
            services.put(svc, svcInfo);
        }
        result.put("services", services);

        return result;
    }

    /**
     * 选择性更新：只拉取指定服务的镜像并重建。
     * @param services 要更新的服务列表，为空则更新所有
     */
    public void runSelectiveUpdate(Consumer<String> emit, List<String> services) {
        String composeFile = deployRoot + "/docker-compose.yml";

        boolean loginOk = dockerLogin(emit);
        if (!loginOk) {
            emit.accept("  [错误] 镜像仓库登录失败，更新中止。请检查 " + deployRoot + "/.env 中的 REGISTRY/REGISTRY_USER/REGISTRY_PASSWORD 配置。");
            emit.accept("DONE");
            return;
        }
        patchConfigs(emit);
        runSchemaMigrations(emit);

        List<String> toUpdate = services != null && !services.isEmpty()
                ? services : new ArrayList<>(OTHER_SERVICES);
        // 确保 tenant-service 在最后
        toUpdate.remove("tenant-service");

        emit.accept(">>> 拉取镜像（" + toUpdate.size() + " 个服务）...");
        for (String svc : toUpdate) {
            emit.accept("  pulling " + svc + " ...");
            exec(emit, "docker", "compose", "-f", composeFile, "pull", "--quiet", svc);
        }
        emit.accept("  pulling tenant-service ...");
        exec(emit, "docker", "compose", "-f", composeFile, "pull", "--quiet", "tenant-service");
        emit.accept(">>> 镜像拉取完成");

        restartAndSelfUpdate(emit, composeFile);
    }

    /** 拉取最新镜像并重建所有容器。 */
    public void runUpdate(Consumer<String> emit) {
        runSelectiveUpdate(emit, null);
    }

    /** 保留数据，重置容器和数据库表结构。 */
    public void runReset(Consumer<String> emit) {
        String composeFile = deployRoot + "/docker-compose.yml";

        patchConfigs(emit);
        resetDatabaseSchema(emit);
        restartAndSelfUpdate(emit, composeFile);
    }

    // ── 数据库重置（保留核心数据）──────────────────────────────────────────────

    /**
     * 需要保留数据的核心表及其主键列。
     * 导出 → 删表 → 重启后 Hibernate 重建 → 恢复数据。
     */
    private static final Map<String, String> PRESERVE_TABLES = new java.util.LinkedHashMap<>();
    static {
        PRESERVE_TABLES.put("t_tenant",             "id");
        PRESERVE_TABLES.put("t_ops_admin",          "id");
        PRESERVE_TABLES.put("t_app",                "id");
        PRESERVE_TABLES.put("t_feature_service",    "id");
        PRESERVE_TABLES.put("t_risk_config",        "id");
        PRESERVE_TABLES.put("app_licenses",         "app_key");
        PRESERVE_TABLES.put("t_sensitive_word",             "id");
        PRESERVE_TABLES.put("t_operation_log",             "id");
        PRESERVE_TABLES.put("t_service_activation_request","id");
        PRESERVE_TABLES.put("t_email_verification",        "id");
        PRESERVE_TABLES.put("t_migrate_key",               "id");
    }

    private void resetDatabaseSchema(Consumer<String> emit) {
        emit.accept(">>> 重置数据库表结构（保留核心数据）...");
        try (Connection conn = dataSource.getConnection();
             Statement stmt = conn.createStatement()) {

            // 1. 导出核心表数据到临时表
            Map<String, String> backupTables = new java.util.LinkedHashMap<>();
            for (Map.Entry<String, String> entry : PRESERVE_TABLES.entrySet()) {
                String table = entry.getKey();
                String tmpTable = "_backup_" + table;
                try {
                    // 检查源表是否存在
                    try (ResultSet rs = stmt.executeQuery(
                            "SELECT COUNT(*) FROM information_schema.TABLES " +
                            "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '" + table + "'")) {
                        if (!rs.next() || rs.getInt(1) == 0) continue;
                    }
                    stmt.execute("DROP TABLE IF EXISTS `" + tmpTable + "`");
                    stmt.execute("CREATE TABLE `" + tmpTable + "` AS SELECT * FROM `" + table + "`");
                    long count = 0;
                    try (ResultSet rs = stmt.executeQuery("SELECT COUNT(*) FROM `" + tmpTable + "`")) {
                        if (rs.next()) count = rs.getLong(1);
                    }
                    backupTables.put(table, tmpTable);
                    emit.accept("  已备份 " + table + " (" + count + " 行)");
                } catch (Exception e) {
                    emit.accept("  [警告] 备份 " + table + " 失败: " + e.getMessage());
                }
            }

            // 2. 禁用外键检查，删除所有业务表
            stmt.execute("SET FOREIGN_KEY_CHECKS = 0");
            try (ResultSet rs = stmt.executeQuery(
                    "SELECT TABLE_NAME FROM information_schema.TABLES " +
                    "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_TYPE = 'BASE TABLE'")) {
                List<String> tables = new java.util.ArrayList<>();
                while (rs.next()) {
                    String name = rs.getString("TABLE_NAME");
                    if (!"_schema_migrations".equals(name) && !name.startsWith("_backup_")) {
                        tables.add(name);
                    }
                }
                for (String table : tables) {
                    stmt.execute("DROP TABLE IF EXISTS `" + table + "`");
                }
                emit.accept("  已删除 " + tables.size() + " 张业务表");
            }

            // 3. 清空迁移记录，保留备份表
            stmt.execute("DELETE FROM _schema_migrations");
            stmt.execute("SET FOREIGN_KEY_CHECKS = 1");

            // 4. 写入恢复脚本，供启动后执行
            saveRestoreScript(backupTables);
            emit.accept(">>> 数据库表结构已重置，容器重启后将自动重建并恢复数据");
        } catch (Exception e) {
            emit.accept("  [错误] 重置数据库失败: " + e.getMessage());
            log.error("reset database schema failed", e);
        }
    }

    /**
     * 将恢复指令写入文件，供 onApplicationReady 读取执行。
     * 格式：每行 "源表名:备份表名:主键列"
     */
    private void saveRestoreScript(Map<String, String> backupTables) {
        try {
            Path script = Paths.get(deployRoot, ".db-restore-pending");
            List<String> lines = backupTables.entrySet().stream()
                    .map(e -> e.getKey() + ":" + e.getValue() + ":" + PRESERVE_TABLES.get(e.getKey()))
                    .collect(Collectors.toList());
            Files.write(script, lines, StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING);
            log.info("wrote restore script: {}", script);
        } catch (IOException e) {
            log.error("failed to write restore script", e);
        }
    }

    /**
     * 启动时检查是否有待恢复的数据。
     * 在 Hibernate 建表之后、迁移之前执行。
     */
    @EventListener(ApplicationReadyEvent.class)
    public void onApplicationReady() {
        restoreFromBackup();
        runSchemaMigrations(line -> log.info("[migration] {}", line));
    }

    private void restoreFromBackup() {
        Path script = Paths.get(deployRoot, ".db-restore-pending");
        if (!Files.exists(script)) return;

        log.info("restoring data from backup tables...");
        try (Connection conn = dataSource.getConnection();
             Statement stmt = conn.createStatement()) {
            List<String> lines = Files.readAllLines(script);
            for (String line : lines) {
                String[] parts = line.split(":");
                if (parts.length != 3) continue;
                String table = parts[0];
                String tmpTable = parts[1];
                String pk = parts[2];
                try {
                    // 检查备份表是否存在
                    try (ResultSet rs = stmt.executeQuery(
                            "SELECT COUNT(*) FROM information_schema.TABLES " +
                            "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '" + tmpTable + "'")) {
                        if (!rs.next() || rs.getInt(1) == 0) {
                            log.info("backup table {} not found, skip", tmpTable);
                            continue;
                        }
                    }
                    // 获取列列表（取目标表和备份表的交集）
                    List<String> columns = getCommonColumns(stmt, table, tmpTable);
                    if (columns.isEmpty()) {
                        log.warn("no common columns between {} and {}", table, tmpTable);
                        continue;
                    }
                    // 获取目标表列的 NOT NULL 和 DEFAULT 信息
                    Map<String, Map<String, String>> colMeta = getColumnMeta(stmt, table);
                    // 对 NOT NULL 列用 COALESCE 兜底，避免 INSERT IGNORE 静默丢行
                    List<String> selectExprs = new java.util.ArrayList<>();
                    for (String col : columns) {
                        Map<String, String> meta = colMeta.get(col);
                        if (meta != null && "NO".equals(meta.get("nullable"))) {
                            String fallback = guessNotNullDefault(meta);
                            selectExprs.add("COALESCE(`" + col + "`, " + fallback + ") AS `" + col + "`");
                        } else {
                            selectExprs.add("`" + col + "`");
                        }
                    }
                    String colList = columns.stream().map(c -> "`" + c + "`").collect(Collectors.joining(", "));
                    String selectList = String.join(", ", selectExprs);
                    String sql = "INSERT INTO `" + table + "` (" + colList + ") " +
                                 "SELECT " + selectList + " FROM `" + tmpTable + "`";
                    int rows = stmt.executeUpdate(sql);
                    log.info("restored {} rows into {}", rows, table);

                    // 删除备份表
                    stmt.execute("DROP TABLE IF EXISTS `" + tmpTable + "`");
                } catch (Exception e) {
                    log.error("restore {} failed: {}", table, e.getMessage());
                }
            }
            Files.deleteIfExists(script);
            log.info("data restore complete");
        } catch (Exception e) {
            log.error("restore from backup failed", e);
        }
    }

    private List<String> getCommonColumns(Statement stmt, String table, String tmpTable) throws Exception {
        Set<String> tableCols = new java.util.LinkedHashSet<>();
        try (ResultSet rs = stmt.executeQuery("SHOW COLUMNS FROM `" + table + "`")) {
            while (rs.next()) tableCols.add(rs.getString("Field"));
        }
        List<String> common = new java.util.ArrayList<>();
        try (ResultSet rs = stmt.executeQuery("SHOW COLUMNS FROM `" + tmpTable + "`")) {
            while (rs.next()) {
                String col = rs.getString("Field");
                if (tableCols.contains(col)) common.add(col);
            }
        }
        return common;
    }

    /** 获取目标表各列的 Null、Default 等元信息。 */
    private Map<String, Map<String, String>> getColumnMeta(Statement stmt, String table) throws Exception {
        Map<String, Map<String, String>> meta = new java.util.LinkedHashMap<>();
        try (ResultSet rs = stmt.executeQuery("SHOW COLUMNS FROM `" + table + "`")) {
            while (rs.next()) {
                Map<String, String> info = new java.util.LinkedHashMap<>();
                info.put("type", rs.getString("Type"));
                info.put("nullable", rs.getString("Null"));
                info.put("default", rs.getString("Default"));
                meta.put(rs.getString("Field"), info);
            }
        }
        return meta;
    }

    /** 为 NOT NULL 且无 DEFAULT 的列根据类型推断一个合理的兜底值。 */
    private String guessNotNullDefault(Map<String, String> meta) {
        String defaultVal = meta.get("default");
        if (defaultVal != null && !"null".equalsIgnoreCase(defaultVal) && !defaultVal.isEmpty()) {
            // MySQL SHOW COLUMNS 的 Default 字段已经是不带引号的字面量
            // 但字符串类型的值在 JDBC ResultSet 中返回时可能不带引号，需要加上
            String type = meta.getOrDefault("type", "").toLowerCase();
            if (type.startsWith("varchar") || type.startsWith("char") || type.startsWith("text") || type.startsWith("enum")) {
                return "'" + defaultVal.replace("'", "\\'") + "'";
            }
            return defaultVal;
        }
        String type = meta.getOrDefault("type", "").toLowerCase();
        if (type.startsWith("varchar") || type.startsWith("char") || type.startsWith("text")) return "''";
        if (type.startsWith("bigint") || type.startsWith("int") || type.startsWith("tinyint") || type.startsWith("smallint")) return "0";
        if (type.startsWith("decimal") || type.startsWith("double") || type.startsWith("float")) return "0";
        if (type.startsWith("datetime") || type.startsWith("timestamp")) return "CURRENT_TIMESTAMP";
        if (type.startsWith("date")) return "CURRENT_DATE";
        if (type.startsWith("json")) return "'{}'";
        if (type.startsWith("enum")) return "''";
        return "''";
    }

    // ── Schema 版本化迁移 ───────────────────────────────────────────────────────

    /**
     * 执行所有待处理的 schema 迁移。
     *
     * 迁移原则：
     *  - ddl-auto:update 自动处理新增列/表，此处仅处理 Hibernate 无法完成的变更
     *    （删列、改列名、类型转换、数据填充等）
     *  - 每个迁移有唯一 ID，执行后记录到 _schema_migrations，保证幂等
     *  - 新版本新增迁移时，在末尾追加新的 migrate_xxx() 调用即可
     */
    public void runSchemaMigrations(Consumer<String> emit) {
        emit.accept(">>> 检查数据库迁移...");
        try {
            ensureMigrationsTable();
        } catch (Exception e) {
            emit.accept("  [警告] 无法初始化迁移记录表: " + e.getMessage());
            return;
        }

        migrate_v20260101_drop_device_id_unique_index(emit);
        migrate_v20260527_push_license_operation_logs(emit);
        migrate_v20260527_fix_orphan_tenant_data(emit);
        migrate_v20260610_gray_mode_simplify_bookmark(emit);
        // 新版本迁移在此追加，例如：
        // migrate_v20260701_xxx(emit);

        emit.accept(">>> 数据库迁移检查完成");
    }

    private void ensureMigrationsTable() throws Exception {
        try (Connection conn = dataSource.getConnection();
             Statement stmt = conn.createStatement()) {
            stmt.execute("""
                CREATE TABLE IF NOT EXISTS _schema_migrations (
                    id          VARCHAR(128) NOT NULL PRIMARY KEY,
                    applied_at  DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
                    description VARCHAR(255)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
                """);
        }
    }

    private boolean migrationApplied(String id) {
        try (Connection conn = dataSource.getConnection();
             PreparedStatement ps = conn.prepareStatement(
                     "SELECT COUNT(*) FROM _schema_migrations WHERE id = ?")) {
            ps.setString(1, id);
            try (ResultSet rs = ps.executeQuery()) {
                return rs.next() && rs.getInt(1) > 0;
            }
        } catch (Exception e) {
            log.warn("check migration {} failed: {}", id, e.getMessage());
            return false;
        }
    }

    private void recordMigration(String id, String description) {
        try (Connection conn = dataSource.getConnection();
             PreparedStatement ps = conn.prepareStatement(
                     "INSERT IGNORE INTO _schema_migrations (id, description) VALUES (?, ?)")) {
            ps.setString(1, id);
            ps.setString(2, description);
            ps.executeUpdate();
        } catch (Exception e) {
            log.warn("record migration {} failed: {}", id, e.getMessage());
        }
    }

    // ── 各版本迁移 ──────────────────────────────────────────────────────────────

    /**
     * license-service DeviceEntity 上的 column-level unique=true 在多租户场景下产生了跨 appKey 的
     * 全局唯一约束，与正确的复合唯一索引 uk_app_key_device_id(app_key, device_id) 冲突。
     * Hibernate ddl-auto:update 不删除多余约束，必须手动 ALTER TABLE。
     * 根治方案：已同步移除 DeviceEntity.deviceId 上的 unique=true 注解，新安装不再产生该约束。
     */
    private void migrate_v20260101_drop_device_id_unique_index(Consumer<String> emit) {
        final String id = "v20260101_drop_device_id_unique_index";
        if (migrationApplied(id)) {
            emit.accept("  [已应用] " + id);
            return;
        }
        try (Connection conn = dataSource.getConnection()) {
            boolean exists;
            try (PreparedStatement ps = conn.prepareStatement("""
                    SELECT COUNT(*) FROM information_schema.STATISTICS
                    WHERE TABLE_SCHEMA = DATABASE()
                      AND TABLE_NAME  = 'devices'
                      AND INDEX_NAME  = 'device_id'
                      AND NON_UNIQUE  = 0
                    """);
                 ResultSet rs = ps.executeQuery()) {
                exists = rs.next() && rs.getInt(1) > 0;
            }
            if (exists) {
                try (Statement stmt = conn.createStatement()) {
                    stmt.execute("ALTER TABLE devices DROP INDEX device_id");
                }
                emit.accept("  [已迁移] " + id + ": 删除 devices.device_id 旧单列唯一约束");
            } else {
                emit.accept("  [已迁移] " + id + ": devices.device_id 单列约束不存在，无需处理");
            }
            recordMigration(id, "删除 devices 表 device_id 旧单列唯一约束");
        } catch (Exception e) {
            emit.accept("  [错误] " + id + ": " + e.getMessage());
            log.error("migration {} failed", id, e);
        }
    }

    /**
     * push-service 和 license-service 新增操作日志表（push_operation_log / license_operation_log）。
     * 实际表结构由各服务的 SchemaMigrationService 在各自数据库中创建，此处仅记录版本标记。
     */
    private void migrate_v20260527_push_license_operation_logs(Consumer<String> emit) {
        final String id = "v20260527_push_license_operation_logs";
        if (migrationApplied(id)) {
            emit.accept("  [已应用] " + id);
            return;
        }
        emit.accept("  [已迁移] " + id + ": push/license 操作日志表已由各服务自行创建");
        recordMigration(id, "push-service 和 license-service 新增操作日志表");
    }

    /**
     * 私有化部署下修复租户数据：
     * 1. 多个租户 → 保留最早的，删除其余
     * 2. 孤儿数据（tenant_id 不存在于 t_tenant）→ 指向唯一租户
     * 仅在 DEPLOYMENT_MODE=PRIVATE 时执行。
     */
    private void migrate_v20260527_fix_orphan_tenant_data(Consumer<String> emit) {
        if (!deployProps.isPrivate()) {
            return;
        }
        final String id = "v20260527_fix_orphan_tenant_data";
        if (migrationApplied(id)) {
            emit.accept("  [已应用] " + id);
            return;
        }
        try (Connection conn = dataSource.getConnection()) {
            // 1. 找到最早的租户
            String keepId;
            try (PreparedStatement ps = conn.prepareStatement(
                    "SELECT id FROM t_tenant ORDER BY created_at ASC LIMIT 1");
                 ResultSet rs = ps.executeQuery()) {
                if (!rs.next()) {
                    emit.accept("  [跳过] " + id + ": 无租户数据");
                    recordMigration(id, "私有化合并多租户（无数据）");
                    return;
                }
                keepId = rs.getString("id");
            }

            boolean changed = false;

            // 2. 多租户合并：删除多余租户
            int extraCount;
            try (PreparedStatement ps = conn.prepareStatement(
                    "SELECT COUNT(*) FROM t_tenant WHERE id != ?")) {
                ps.setString(1, keepId);
                try (ResultSet rs = ps.executeQuery()) {
                    rs.next();
                    extraCount = rs.getInt(1);
                }
            }
            if (extraCount > 0) {
                emit.accept("  [合并] 保留最早租户 " + keepId + "，删除 " + extraCount + " 个多余租户");
                try (Statement stmt = conn.createStatement()) {
                    stmt.executeUpdate("UPDATE t_tenant SET parent_id = '" + keepId
                            + "' WHERE parent_id IS NOT NULL AND parent_id != '" + keepId + "'");
                }
                try (Statement stmt = conn.createStatement()) {
                    int deleted = stmt.executeUpdate("DELETE FROM t_tenant WHERE id != '" + keepId + "'");
                    emit.accept("    删除多余租户: " + deleted + " 个");
                }
                changed = true;
            }

            // 3. 修复孤儿数据：tenant_id 与唯一租户不匹配的记录（包括指向不存在租户的情况）
            String[][] tables = {
                    {"t_app", "tenant_id"},
                    {"t_operation_log", "tenant_id"},
                    {"t_migrate_key", "tenant_id"},
            };
            for (String[] tbl : tables) {
                try (Statement stmt = conn.createStatement()) {
                    int rows = stmt.executeUpdate("UPDATE " + tbl[0] + " SET " + tbl[1] + " = '"
                            + keepId + "' WHERE " + tbl[1] + " != '" + keepId + "'");
                    if (rows > 0) {
                        emit.accept("    " + tbl[0] + ": 修复 " + rows + " 条记录");
                        changed = true;
                    }
                }
            }

            if (changed) {
                emit.accept("  [已迁移] " + id + ": 修复完成，保留租户 " + keepId);
                recordMigration(id, "私有化修复租户数据，保留 " + keepId);
            } else {
                emit.accept("  [跳过] " + id + ": 数据正常，无需修复");
                recordMigration(id, "私有化修复租户数据（无需修复）");
            }
        } catch (Exception e) {
            emit.accept("  [错误] " + id + ": " + e.getMessage());
            log.error("migration {} failed", id, e);
        }
    }

    /**
     * v20260610_gray_mode_simplify 由 update-service 的 SchemaMigrationRunner 负责执行，
     * 此处仅在 tenant-service 迁移记录表中留下标记，避免版本跳跃引起混淆。
     * update-service 在每次重启时会幂等地应用其自身迁移，无需在此重复执行 SQL。
     */
    private void migrate_v20260610_gray_mode_simplify_bookmark(Consumer<String> emit) {
        final String id = "v20260610_gray_mode_simplify";
        if (migrationApplied(id)) {
            emit.accept("  [已应用] " + id + " (update-service 负责执行)");
            return;
        }
        // update-service 重启后会自行执行实际迁移，此处只记录版本标记
        recordMigration(id, "GrayMode 简化（IM_PUSH_USERS/CUSTOMER_SYNC/CUSTOMER_CALLBACK → MEMBERS），由 update-service 执行");
        emit.accept("  [已记录] " + id + ": update-service 将在启动时执行实际迁移");
    }

    // ── 重启核心 ────────────────────────────────────────────────────────────────

    private void restartAndSelfUpdate(Consumer<String> emit, String composeFile) {
        emit.accept(">>> 重建各服务容器...");
        for (String svc : OTHER_SERVICES) {
            emit.accept("  restarting " + svc + " ...");
            exec(emit, "docker", "compose", "-f", composeFile,
                    "up", "-d", "--no-deps", "--force-recreate", svc);
            emit.accept("  " + svc + " ✓");
        }

        emit.accept(">>> 启动自更新助手容器...");
        String selfImage = getCurrentImage();
        if (selfImage == null) {
            emit.accept(">>> [错误] 无法获取当前 tenant-service 镜像名，请手动执行：");
            emit.accept(">>>   docker compose -f " + composeFile + " up -d --no-deps --force-recreate tenant-service");
            emit.accept("DONE");
            return;
        }
        boolean helperStarted = spawnSelfUpdater(composeFile, selfImage);
        if (helperStarted) {
            emit.accept(">>> 助手容器已就绪，tenant-service 即将重建（连接将短暂中断）...");
            emit.accept("RESTART_SELF");
        } else {
            emit.accept(">>> [警告] 助手容器启动失败，请手动执行：");
            emit.accept(">>>   docker compose -f " + composeFile + " up -d --no-deps --force-recreate tenant-service");
            emit.accept("DONE");
        }
    }

    // ── 配置文件热修复 ──────────────────────────────────────────────────────────

    private void patchConfigs(Consumer<String> emit) {
        emit.accept(">>> 检查并修复配置文件...");
        patchNginxFileRoute(emit);
        patchNginxUpdateTimeout(emit);
        patchNginxWebSocket(emit);
        patchDockerComposeFileService(emit);
        patchDockerComposeUpdateService(emit);
    }

    private void patchNginxUpdateTimeout(Consumer<String> emit) {
        Path conf = Paths.get(deployRoot, "config", "nginx", "conf.d", "xuqm.conf");
        if (!Files.exists(conf)) return;
        try {
            String content = Files.readString(conf);
            if (content.contains("location ~ ^/api/system/") || content.contains("location = /api/system/update")) return;
            String anchor = "  # 核心 API（兜底，在所有具体 /api/xxx/ 之后）\n  location /api/ {";
            if (!content.contains(anchor)) {
                emit.accept("  [跳过] nginx 更新超时补丁锚点未找到，请手动检查");
                return;
            }
            String injection = "  # 一键更新/重置：操作耗时较长，需要更长超时（精确匹配，优先于 /api/ 前缀）\n"
                    + "  location ~ ^/api/system/(update|reset)$ {\n"
                    + "    set $svc tenant-service;\n"
                    + "    proxy_pass         http://$svc:9001;\n"
                    + "    proxy_set_header   Host $host;\n"
                    + "    proxy_set_header   X-Real-IP $remote_addr;\n"
                    + "    proxy_set_header   X-Forwarded-For $proxy_add_x_forwarded_for;\n"
                    + "    proxy_buffering    off;\n"
                    + "    proxy_read_timeout 600s;\n"
                    + "    proxy_send_timeout 600s;\n"
                    + "  }\n\n"
                    + anchor;
            Files.writeString(conf, content.replace(anchor, injection), StandardOpenOption.TRUNCATE_EXISTING);
            emit.accept("  [已修复] nginx: 补齐 /api/system/(update|reset) 600s 超时");
        } catch (IOException e) {
            emit.accept("  [警告] nginx 更新超时修复失败: " + e.getMessage());
        }
    }

    /**
     * 为 update-service 注入 nginx WebSocket 代理配置。
     * WebSocket 需要 HTTP/1.1 升级头和长时间超时。
     */
    private void patchNginxWebSocket(Consumer<String> emit) {
        Path conf = Paths.get(deployRoot, "config", "nginx", "conf.d", "xuqm.conf");
        if (!Files.exists(conf)) return;
        try {
            String content = Files.readString(conf);
            if (content.contains("location /ws/")) return;

            String anchor = "  # 核心 API（兜底，在所有具体 /api/xxx/ 之后）\n  location /api/ {";
            if (!content.contains(anchor)) {
                emit.accept("  [跳过] nginx WebSocket 补丁锚点未找到，请手动检查");
                return;
            }
            String injection = "  # WebSocket 实时通知（update-service 版本发布推送）\n"
                    + "  location /ws/ {\n"
                    + "    set $svc update-service;\n"
                    + "    proxy_pass         http://$svc:8084;\n"
                    + "    proxy_http_version 1.1;\n"
                    + "    proxy_set_header   Upgrade $http_upgrade;\n"
                    + "    proxy_set_header   Connection \"upgrade\";\n"
                    + "    proxy_set_header   Host $host;\n"
                    + "    proxy_set_header   X-Real-IP $remote_addr;\n"
                    + "    proxy_set_header   X-Forwarded-For $proxy_add_x_forwarded_for;\n"
                    + "    proxy_read_timeout 86400s;\n"
                    + "    proxy_send_timeout 86400s;\n"
                    + "  }\n\n"
                    + anchor;
            Files.writeString(conf, content.replace(anchor, injection), StandardOpenOption.TRUNCATE_EXISTING);
            emit.accept("  [已修复] nginx: 补齐 /ws/ WebSocket 代理（update-service）");
        } catch (IOException e) {
            emit.accept("  [警告] nginx WebSocket 修复失败: " + e.getMessage());
        }
    }

    private void patchNginxFileRoute(Consumer<String> emit) {
        Path conf = Paths.get(deployRoot, "config", "nginx", "conf.d", "xuqm.conf");
        if (!Files.exists(conf)) return;
        try {
            String content = Files.readString(conf);
            if (!content.contains("location /file/")) return;
            Files.writeString(conf, content.replace("location /file/", "location /api/file/"),
                    StandardOpenOption.TRUNCATE_EXISTING);
            emit.accept("  [已修复] nginx: location /file/ → /api/file/");
        } catch (IOException e) {
            emit.accept("  [警告] nginx 配置修复失败: " + e.getMessage());
        }
    }

    private void patchDockerComposeFileService(Consumer<String> emit) {
        Path composeFile = Paths.get(deployRoot, "docker-compose.yml");
        if (!Files.exists(composeFile)) return;
        try {
            String content = Files.readString(composeFile);
            if (content.contains("FILE_UPLOAD_DIR") && content.contains("FILE_BASE_URL")) return;

            String consoleDomain = readEnvValue(Paths.get(deployRoot, "config", "xuqm.env"), "CONSOLE_DOMAIN");
            if (consoleDomain == null) consoleDomain = "";

            String anchor = "      SPRING_DATA_REDIS_DATABASE: \"${REDIS_DATABASE:-0}\"\n";
            if (!content.contains(anchor)) {
                emit.accept("  [跳过] docker-compose 文件服务补丁锚点未找到，请手动检查");
                return;
            }
            String injection = anchor
                    + "      FILE_UPLOAD_DIR: \"/data/uploads\"\n"
                    + "      FILE_BASE_URL: \"" + consoleDomain + "\"\n";
            Files.writeString(composeFile, content.replace(anchor, injection), StandardOpenOption.TRUNCATE_EXISTING);
            emit.accept("  [已修复] docker-compose: 补齐 FILE_UPLOAD_DIR 和 FILE_BASE_URL");
        } catch (IOException e) {
            emit.accept("  [警告] docker-compose 修复失败: " + e.getMessage());
        }
    }

    private void patchDockerComposeUpdateService(Consumer<String> emit) {
        Path composeFile = Paths.get(deployRoot, "docker-compose.yml");
        if (!Files.exists(composeFile)) return;
        try {
            String content = Files.readString(composeFile);
            if (content.contains("FILE_SERVICE_INTERNAL_URL")) return;

            String consoleDomain = readEnvValue(Paths.get(deployRoot, "config", "xuqm.env"), "CONSOLE_DOMAIN");
            if (consoleDomain == null) consoleDomain = "";

            String envBlock = "      FILE_BASE_URL: \"" + consoleDomain + "\"\n"
                    + "      FILE_SERVICE_INTERNAL_URL: \"http://file-service:8086\"\n";
            String anchor = "      SDK_TENANT_SERVICE_URL: \"http://tenant-service:9001\"\n";
            String fallbackAnchor = "    image: ${REGISTRY}/update-service:${IMAGE_TAG}\n";

            String patched;
            if (content.contains(anchor)) {
                patched = content.replace(anchor, anchor + envBlock);
            } else if (content.contains(fallbackAnchor)) {
                String envAnchor = fallbackAnchor + "    environment:\n";
                if (!content.contains(envAnchor)) {
                    emit.accept("  [跳过] docker-compose update-service environment 段未找到，请手动检查");
                    return;
                }
                patched = content.replace(envAnchor, envAnchor + envBlock);
            } else {
                emit.accept("  [跳过] docker-compose update-service 段未找到，请手动检查");
                return;
            }
            Files.writeString(composeFile, patched, StandardOpenOption.TRUNCATE_EXISTING);
            emit.accept("  [已修复] docker-compose: 补齐 update-service 的 FILE_BASE_URL 和 FILE_SERVICE_INTERNAL_URL");
        } catch (IOException e) {
            emit.accept("  [警告] docker-compose update-service 修复失败: " + e.getMessage());
        }
    }

    // ── Docker 工具方法 ─────────────────────────────────────────────────────────

    /** @return true 表示登录成功或无需登录（无凭据时返回 false） */
    private boolean dockerLogin(Consumer<String> emit) {
        try {
            String registry = null, user = null, password = null;
            Path envFile = Paths.get(deployRoot + "/.env");
            if (!Files.exists(envFile)) {
                emit.accept("  [错误] 找不到 " + deployRoot + "/.env，无法获取镜像仓库凭据");
                return false;
            }
            for (String line : Files.readAllLines(envFile)) {
                if (line.startsWith("REGISTRY="))              registry = line.substring("REGISTRY=".length()).trim();
                else if (line.startsWith("REGISTRY_USER="))    user     = line.substring("REGISTRY_USER=".length()).trim();
                else if (line.startsWith("REGISTRY_PASSWORD=")) password = line.substring("REGISTRY_PASSWORD=".length()).trim();
            }
            if (registry == null || registry.isBlank()) {
                emit.accept("  [错误] .env 中未配置 REGISTRY，无法拉取镜像");
                return false;
            }
            if (user == null || password == null || password.isBlank()) {
                emit.accept("  [错误] .env 中未配置 REGISTRY_USER / REGISTRY_PASSWORD");
                return false;
            }
            String host = registry.contains("/") ? registry.substring(0, registry.indexOf('/')) : registry;
            ProcessBuilder pb = new ProcessBuilder("docker", "login", host, "-u", user, "--password-stdin")
                    .redirectErrorStream(true);
            Process p = pb.start();
            p.getOutputStream().write((password + "\n").getBytes());
            p.getOutputStream().flush();
            p.getOutputStream().close();
            String out = new String(p.getInputStream().readAllBytes()).trim();
            int code = p.waitFor();
            if (code == 0) {
                emit.accept("  已完成镜像仓库登录 (" + host + ")");
                return true;
            } else {
                emit.accept("  [错误] 镜像仓库登录失败: " + out);
                return false;
            }
        } catch (Exception e) {
            emit.accept("  [错误] 读取仓库凭据失败: " + e.getMessage());
            return false;
        }
    }

    private boolean spawnSelfUpdater(String composeFile, String image) {
        try {
            new ProcessBuilder("docker", "rm", "-f", "xuqm-self-updater")
                    .redirectErrorStream(true).start().waitFor();

            String shellCmd = "sleep 8 && docker compose -f " + composeFile
                    + " up -d --no-deps --force-recreate tenant-service";

            Process p = new ProcessBuilder(
                    "docker", "run", "-d", "--rm",
                    "--name", "xuqm-self-updater",
                    "-v", "/var/run/docker.sock:/var/run/docker.sock",
                    "-v", deployRoot + ":" + deployRoot,
                    "--entrypoint", "sh",
                    image,
                    "-c", shellCmd
            ).redirectErrorStream(true).start();

            String out = new String(p.getInputStream().readAllBytes()).trim();
            int code = p.waitFor();
            log.info("self-updater spawn: code={} containerId={}", code, out);
            return code == 0;
        } catch (Exception e) {
            log.error("failed to spawn self-updater", e);
            return false;
        }
    }

    private String getCurrentImage() {
        try {
            Process p = new ProcessBuilder(
                    "docker", "ps",
                    "--filter", "label=com.docker.compose.service=tenant-service",
                    "--format", "{{.Image}}"
            ).redirectErrorStream(true).start();
            String out = new String(p.getInputStream().readAllBytes()).trim();
            p.waitFor();
            return out.isEmpty() ? null : out;
        } catch (Exception e) {
            return null;
        }
    }

    private String readEnvValue(Path envFile, String key) {
        if (!Files.exists(envFile)) return null;
        try {
            for (String line : Files.readAllLines(envFile)) {
                if (line.startsWith(key + "=")) {
                    return line.substring(key.length() + 1).trim();
                }
            }
        } catch (IOException ignored) {}
        return null;
    }

    private void exec(Consumer<String> emit, String... cmd) {
        try {
            Process p = new ProcessBuilder(cmd)
                    .redirectErrorStream(true)
                    .start();
            try (BufferedReader reader = new BufferedReader(new InputStreamReader(p.getInputStream()))) {
                String line;
                while ((line = reader.readLine()) != null) {
                    if (!line.isBlank()) emit.accept("    " + line);
                }
            }
            int code = p.waitFor();
            if (code != 0) emit.accept("    [warn] exit code " + code);
        } catch (Exception e) {
            emit.accept("    [error] " + e.getMessage());
            log.error("exec failed: {}", String.join(" ", cmd), e);
        }
    }
}