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", "bugcollect-service", "nginx"
    );

    // 镜像名与 compose service 名不一致的例外（历史命名遗留）
    private static final Map<String, String> SERVICE_IMAGE_OVERRIDES = Map.of(
            "bugcollect-service", "xuqm-bugcollect-service"
    );

    // 使用 docker compose profile 的服务（未声明 profile 的服务不在此表中）
    private static final Map<String, String> SERVICE_COMPOSE_PROFILES = Map.of(
            "bugcollect-service", "bugcollect"
    );

    // 健康检查配置：新容器需在此时间内保持 running 状态才视为健康
    private static final int HEALTH_CHECK_TIMEOUT_SEC  = 60;
    private static final int HEALTH_STABLE_REQUIRED_SEC = 10;
    private static final int HEALTH_CHECK_INTERVAL_SEC  = 5;

    private static final Set<String> ALLOWED_LOG_SERVICES = Set.of(
            "tenant-service", "file-service", "im-service", "push-service",
            "update-service", "license-service", "bugcollect-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) {
            String imageName = resolveServiceImageName(svc);
            if (imageName == null) { versions.put(svc, "unknown"); continue; }
            try {
                Process p = new ProcessBuilder(
                        "docker", "inspect", "--format",
                        "{{index .Config.Labels \"com.xuqm.version\"}}",
                        imageName
                ).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");

        // 拉取前先保存所有服务的旧镜像 ID，用于启动失败时回滚
        List<String> allToSnapshot = new ArrayList<>(toUpdate);
        allToSnapshot.add("tenant-service");
        Map<String, String> oldImageIds = captureCurrentImageIds(allToSnapshot);
        emit.accept("  已快照 " + oldImageIds.size() + " 个服务的旧版本镜像（更新失败时自动回滚）");

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

        restartAndSelfUpdate(emit, composeFile, oldImageIds);
    }

    /** 拉取最新镜像并重建所有容器。 */
    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, Map.of());
    }

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

    /**
     * 需要保留数据的核心表及其主键列。
     * 导出 → 删表 → 重启后 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 将在启动时执行实际迁移");
    }

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

    /**
     * 重建各服务容器，并对每个服务进行健康检查。
     * 若新容器在 HEALTH_CHECK_TIMEOUT_SEC 内未保持稳定运行，自动回滚到旧镜像。
     * @param oldImageIds 拉取新镜像前保存的旧镜像 ID（sha256）；为空时跳过回滚
     */
    private void restartAndSelfUpdate(Consumer<String> emit, String composeFile, Map<String, String> oldImageIds) {
        emit.accept(">>> 重建各服务容器（含健康检查与自动回滚）...");
        List<String> rolledBack = new ArrayList<>();
        List<String> failed     = new ArrayList<>();

        for (String svc : OTHER_SERVICES) {
            emit.accept("  restarting " + svc + " ...");
            exec(emit, composeCmd(composeFile, svc, "up", "-d", "--no-deps", "--force-recreate", svc));
            // 拿到 compose up 之后最新创建的容器 ID，排除旧容器干扰
            String newContainerId = getNewestContainerId(svc);
            emit.accept("  [debug] " + svc + " newContainerId=" + newContainerId);
            // Log all containers visible for this service label
            try {
                Process dbgPs = new ProcessBuilder("docker", "ps", "-a",
                        "--filter", "label=com.docker.compose.service=" + svc,
                        "--format", "{{.ID}} {{.Names}} {{.Status}} {{.CreatedAt}}")
                        .redirectErrorStream(true).start();
                String dbgOut = new String(dbgPs.getInputStream().readAllBytes(), StandardCharsets.UTF_8).trim();
                dbgPs.waitFor();
                for (String l : dbgOut.split("\n")) { if (!l.isBlank()) emit.accept("  [debug] " + l); }
            } catch (Exception ignored) {}

            boolean healthy = waitForServiceStable(emit, svc, newContainerId, HEALTH_CHECK_TIMEOUT_SEC);
            if (healthy) {
                emit.accept("  " + svc + " ✓");
            } else {
                String oldId = oldImageIds.get(svc);
                if (oldId != null && !oldId.isBlank()) {
                    emit.accept("  [警告] " + svc + " 启动失败，正在回滚旧版本...");
                    boolean rollbackOk = rollbackService(emit, composeFile, svc, oldId);
                    if (rollbackOk) {
                        rolledBack.add(svc);
                    } else {
                        failed.add(svc);
                    }
                } else {
                    emit.accept("  [错误] " + svc + " 启动失败且无旧镜像 ID，无法自动回滚");
                    failed.add(svc);
                }
                // 输出该服务最近日志，辅助排查
                try {
                    String tail = getServiceLogs(svc, 30);
                    emit.accept("  --- " + svc + " 近期日志（末30行）---");
                    for (String l : tail.split("\n")) {
                        if (!l.isBlank()) emit.accept("  " + l);
                    }
                    emit.accept("  ---");
                } catch (Exception ignored) {}
            }
        }

        if (!rolledBack.isEmpty()) {
            emit.accept(">>> [警告] 以下服务已自动回滚到旧版本: " + String.join(", ", rolledBack));
            emit.accept(">>> 请检查代码或配置后重新发版。");
        }
        if (!failed.isEmpty()) {
            emit.accept(">>> [严重] 以下服务更新失败且回滚无效，需人工介入: " + String.join(", ", failed));
        }

        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, oldImageIds.getOrDefault("tenant-service", ""));
        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");
        }
    }

    // ── 镜像快照与健康检查 ──────────────────────────────────────────────────────

    /**
     * 在拉取新镜像前，保存各服务当前运行容器的镜像 ID（sha256）。
     * 存为 Map&lt;serviceName, imageId&gt;，用于更新失败时 docker tag 回旧版本。
     */
    private Map<String, String> captureCurrentImageIds(List<String> services) {
        Map<String, String> ids = new LinkedHashMap<>();
        for (String svc : services) {
            try {
                Process ps = new ProcessBuilder(
                        "docker", "ps",
                        "--filter", "label=com.docker.compose.service=" + svc,
                        "--format", "{{.ID}}"
                ).redirectErrorStream(true).start();
                String containerId = new String(ps.getInputStream().readAllBytes(), StandardCharsets.UTF_8).trim();
                ps.waitFor();
                if (containerId.isEmpty()) continue;
                containerId = containerId.split("\n")[0].trim();

                Process inspect = new ProcessBuilder(
                        "docker", "inspect", "--format", "{{.Image}}", containerId
                ).redirectErrorStream(true).start();
                String imageId = new String(inspect.getInputStream().readAllBytes(), StandardCharsets.UTF_8).trim();
                inspect.waitFor();
                if (!imageId.isEmpty()) {
                    ids.put(svc, imageId);
                }
            } catch (Exception e) {
                log.warn("captureCurrentImageIds: failed for {}: {}", svc, e.getMessage());
            }
        }
        return ids;
    }

    /**
     * 获取指定服务最新创建的容器 ID（含已停止容器）。
     * 在 docker compose up --force-recreate 之后立即调用，确保拿到新容器而非旧容器。
     */
    private String getNewestContainerId(String service) {
        try {
            // Do NOT use -n 1: Docker applies -n globally before the label filter,
            // so if the last-created container is a different service, the filter
            // returns empty. Without -n, docker ps sorts newest-first and the
            // label filter returns only this service's containers.
            Process p = new ProcessBuilder(
                    "docker", "ps", "-a",
                    "--filter", "label=com.docker.compose.service=" + service,
                    "--format", "{{.ID}}"
            ).redirectErrorStream(true).start();
            String out = new String(p.getInputStream().readAllBytes(), StandardCharsets.UTF_8).trim();
            p.waitFor();
            return out.isEmpty() ? null : out.split("\n")[0].trim();
        } catch (Exception e) {
            return null;
        }
    }

    /**
     * 轮询指定容器的状态，直到该容器持续 HEALTH_STABLE_REQUIRED_SEC 秒保持 running。
     * 通过 containerId 精确定位新容器，避免 --force-recreate 停掉旧容器时的误判。
     * 若容器已 exited，立即返回 false（快速失败）。
     *
     * @param containerId 新容器 ID；为 null 时退化为服务名轮询
     */
    private boolean waitForServiceStable(Consumer<String> emit, String service, String containerId, int timeoutSeconds) {
        int elapsed       = 0;
        int stableSeconds = 0;

        while (elapsed < timeoutSeconds) {
            try { Thread.sleep(HEALTH_CHECK_INTERVAL_SEC * 1000L); }
            catch (InterruptedException e) { Thread.currentThread().interrupt(); return false; }
            elapsed       += HEALTH_CHECK_INTERVAL_SEC;
            stableSeconds += HEALTH_CHECK_INTERVAL_SEC;

            try {
                String statusLine;
                if (containerId != null) {
                    // 直接 inspect 新容器，避免旧容器干扰
                    Process ins = new ProcessBuilder(
                            "docker", "inspect", "--format",
                            "{{.State.Status}} {{.State.ExitCode}}", containerId
                    ).redirectErrorStream(true).start();
                    statusLine = new String(ins.getInputStream().readAllBytes(), StandardCharsets.UTF_8).trim();
                    ins.waitFor();
                    emit.accept("  [debug] " + service + " t=" + elapsed + "s inspect " + containerId + " → " + statusLine);
                } else {
                    // 退化模式：查 running 容器
                    Process runPs = new ProcessBuilder(
                            "docker", "ps",
                            "--filter", "label=com.docker.compose.service=" + service,
                            "--filter", "status=running",
                            "--format", "{{.ID}}"
                    ).redirectErrorStream(true).start();
                    String runOut = new String(runPs.getInputStream().readAllBytes(), StandardCharsets.UTF_8).trim();
                    runPs.waitFor();
                    statusLine = runOut.isEmpty() ? "unknown 0" : "running 0";
                }

                if (statusLine.startsWith("running")) {
                    emit.accept("  [健康检查] " + service + " running ("
                            + stableSeconds + "/" + HEALTH_STABLE_REQUIRED_SEC + "s)");
                    if (stableSeconds >= HEALTH_STABLE_REQUIRED_SEC) {
                        return true;
                    }
                } else if (statusLine.startsWith("exited")) {
                    // 快速失败：新容器已退出
                    emit.accept("  [健康检查] " + service + " 已退出 (" + statusLine + ")，快速判定失败");
                    return false;
                } else {
                    // created / paused / restarting 等中间状态
                    stableSeconds = 0;
                    emit.accept("  [健康检查] " + service + " 等待启动... status=" + statusLine
                            + " (" + elapsed + "/" + timeoutSeconds + "s)");
                }
            } catch (Exception e) {
                stableSeconds = 0;
                emit.accept("  [健康检查] " + service + " 状态查询异常: " + e.getMessage());
            }
        }
        emit.accept("  [健康检查] " + service + " 超时（" + timeoutSeconds + "s 内未就绪）");
        return false;
    }

    /**
     * 将指定服务回滚到旧镜像。
     * 先将旧镜像 ID 重新 tag 为 :latest，再 docker compose up 重建容器。
     * @return true 表示回滚后服务成功启动
     */
    private boolean rollbackService(Consumer<String> emit, String composeFile, String service, String oldImageId) {
        String imageName = resolveServiceImageName(service);
        if (imageName != null) {
            try {
                Process tag = new ProcessBuilder("docker", "tag", oldImageId, imageName)
                        .redirectErrorStream(true).start();
                String tagOut = new String(tag.getInputStream().readAllBytes(), StandardCharsets.UTF_8).trim();
                int tagCode = tag.waitFor();
                if (tagCode != 0) {
                    emit.accept("  [回滚警告] docker tag 返回 " + tagCode + (tagOut.isEmpty() ? "" : ": " + tagOut));
                }
            } catch (Exception e) {
                emit.accept("  [回滚警告] 重新标记镜像失败: " + e.getMessage());
            }
        }

        exec(emit, "docker", "compose", "-f", composeFile, "up", "-d", "--no-deps", "--force-recreate", service);
        String rollbackContainerId = getNewestContainerId(service);

        boolean ok = waitForServiceStable(emit, service, rollbackContainerId, 60);
        if (ok) {
            emit.accept("  [回滚] " + service + " 已回滚到旧版本 ✓");
        } else {
            emit.accept("  [严重] " + service + " 回滚后仍无法启动，请人工介入！");
            emit.accept("  [诊断] docker logs $(docker ps -a --filter label=com.docker.compose.service=" + service + " -q --latest)");
        }
        return ok;
    }

    /** 从 .env 读取 REGISTRY 和 IMAGE_TAG，拼接服务完整镜像名（registry/image:tag）。
     *  部分服务镜像名与 compose service 名不同，通过 SERVICE_IMAGE_OVERRIDES 处理。 */
    private String resolveServiceImageName(String service) {
        try {
            Path envFile = Paths.get(deployRoot, ".env");
            String registry = readEnvValue(envFile, "REGISTRY");
            String imageTag  = readEnvValue(envFile, "IMAGE_TAG");
            if (registry == null) return null;
            if (imageTag == null || imageTag.isBlank()) imageTag = "latest";
            String imageName = SERVICE_IMAGE_OVERRIDES.getOrDefault(service, service);
            return registry + "/" + imageName + ":" + imageTag;
        } catch (Exception e) {
            return null;
        }
    }

    /** 构造 docker compose 命令，自动注入该服务所需的 --profile 参数。 */
    private String[] composeCmd(String composeFile, String service, String... subCmd) {
        List<String> cmd = new ArrayList<>();
        cmd.add("docker"); cmd.add("compose");
        String profile = SERVICE_COMPOSE_PROFILES.get(service);
        if (profile != null) { cmd.add("--profile"); cmd.add(profile); }
        cmd.add("-f"); cmd.add(composeFile);
        cmd.addAll(Arrays.asList(subCmd));
        return cmd.toArray(new String[0]);
    }

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

    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;
        }
    }

    /**
     * 启动自更新助手容器，负责重建 tenant-service（当前进程无法重建自身）。
     * 包含健康检查：若新容器在 60s 内未保持运行，自动回滚到旧镜像。
     *
     * @param oldTenantImageId 更新前保存的旧镜像 ID，空字符串表示无法回滚
     */
    private boolean spawnSelfUpdater(String composeFile, String image, String oldTenantImageId) {
        try {
            new ProcessBuilder("docker", "rm", "-f", "xuqm-self-updater")
                    .redirectErrorStream(true).start().waitFor();

            String tenantImageName = resolveServiceImageName("tenant-service");
            if (tenantImageName == null) tenantImageName = "";

            // Shell 脚本：重建 → 等待健康 → 不健康则回滚
            // 健康检查：60s 内每 10s 轮询一次，连续 running 即视为成功
            String shellCmd = "sleep 8 && "
                    + "OLD_ID='" + oldTenantImageId.replace("'", "") + "' && "
                    + "IMG='" + tenantImageName.replace("'", "") + "' && "
                    + "docker compose -f " + composeFile + " up -d --no-deps --force-recreate tenant-service && "
                    + "HEALTHY=false && "
                    + "for i in 1 2 3 4 5 6; do "
                    + "  sleep 10; "
                    + "  if docker ps --filter 'label=com.docker.compose.service=tenant-service' "
                    + "               --filter 'status=running' -q 2>/dev/null | grep -q .; then "
                    + "    HEALTHY=true; break; "
                    + "  fi; "
                    + "  if docker ps -a --filter 'label=com.docker.compose.service=tenant-service' "
                    + "                  --filter 'status=exited' -q 2>/dev/null | grep -q .; then "
                    + "    break; "
                    + "  fi; "
                    + "done; "
                    + "if [ \"$HEALTHY\" != \"true\" ] && [ -n \"$OLD_ID\" ] && [ -n \"$IMG\" ]; then "
                    + "  echo '[ROLLBACK] tenant-service unhealthy, reverting to old image...'; "
                    + "  docker tag \"$OLD_ID\" \"$IMG\"; "
                    + "  docker compose -f " + composeFile + " up -d --no-deps --force-recreate tenant-service; "
                    + "fi";

            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);
        }
    }
}