package com.ai.trackwalker.model;

import com.ai.trackwalker.api.dto.ModelManagementRequests;
import com.ai.trackwalker.common.Ids;
import com.ai.trackwalker.common.Jsonb;
import org.springframework.boot.web.client.RestTemplateBuilder;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.server.ResponseStatusException;

import java.net.URI;
import java.sql.Timestamp;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;

@Service
public class ModelRuntimeDeploymentService {
    private final JdbcTemplate jdbc;
    private final RestTemplateBuilder restTemplateBuilder;

    public ModelRuntimeDeploymentService(JdbcTemplate jdbc, RestTemplateBuilder restTemplateBuilder) {
        this.jdbc = jdbc;
        this.restTemplateBuilder = restTemplateBuilder;
    }

    public List<Map<String, Object>> list(String modelGroupId) {
        requireModelGroup(modelGroupId);
        List<Map<String, Object>> rows = jdbc.queryForList("""
                select d.*, mg.code as model_group_code, mg.name as model_group_name,
                       v.version as model_version, v.display_name as model_version_name
                from model_runtime_deployments d
                join model_groups mg on mg.id=d.model_group_id
                join model_versions v on v.id=d.model_version_id
                where d.model_group_id=?
                order by d.active desc,
                         case d.environment when 'production' then 1 when 'staging' then 2 else 3 end,
                         d.updated_at desc
                """, modelGroupId);
        return rows.stream().map(this::parseRow).toList();
    }

    public List<Map<String, Object>> activeDeployments() {
        List<Map<String, Object>> rows = jdbc.queryForList("""
                select d.*, mg.code as model_group_code, mg.name as model_group_name,
                       v.version as model_version, v.display_name as model_version_name
                from model_runtime_deployments d
                join model_groups mg on mg.id=d.model_group_id
                join model_versions v on v.id=d.model_version_id
                where d.active=true order by mg.name
                """);
        return rows.stream().map(this::parseRow).toList();
    }

    public Map<String, Object> get(String deploymentId) {
        List<Map<String, Object>> rows = jdbc.queryForList("""
                select d.*, mg.code as model_group_code, mg.name as model_group_name,
                       v.version as model_version, v.display_name as model_version_name
                from model_runtime_deployments d
                join model_groups mg on mg.id=d.model_group_id
                join model_versions v on v.id=d.model_version_id
                where d.id=?
                """, deploymentId);
        if (rows.isEmpty()) {
            throw notFound("模型运行配置不存在");
        }
        return parseRow(rows.get(0));
    }

    @Transactional
    public Map<String, Object> create(String modelGroupId, ModelManagementRequests.SaveRuntimeDeploymentRequest request) {
        requireGroupVersion(modelGroupId, request.modelVersionId());
        validateRequest(request);
        if (jdbc.queryForObject("select count(*) from model_runtime_deployments where model_group_id=? and name=?",
                Integer.class, modelGroupId, request.name()) > 0) {
            throw conflict("同一模型组中运行配置名称不能重复");
        }
        String id = Ids.next("runtime");
        Instant now = Instant.now();
        jdbc.update("""
                insert into model_runtime_deployments(
                    id, model_group_id, model_version_id, name, environment, service_url,
                    health_path, runtime_path, inference_path, test_path, runtime_engine,
                    execution_provider, device, precision, artifact_uri, auth_secret_ref,
                    replicas, max_concurrency, request_timeout_seconds, config, status,
                    health_status, active, revision, created_by, updated_by, created_at, updated_at
                ) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?::jsonb,'ready','unknown',false,1,?,?,?,?)
                """,
                id, modelGroupId, request.modelVersionId(), request.name().trim(), request.environment(),
                normalizeServiceUrl(request.serviceUrl()), normalizePath(request.healthPath(), "/health"),
                normalizePath(request.runtimePath(), "/api/v1/runtime"), normalizePath(request.inferencePath(), null),
                normalizePath(request.testPath(), null), request.runtimeEngine(), request.executionProvider(),
                request.device(), request.precision(), blankToNull(request.artifactUri()), blankToNull(request.authSecretRef()),
                defaultInt(request.replicas(), 1), defaultInt(request.maxConcurrency(), 1),
                defaultInt(request.requestTimeoutSeconds(), 60), Jsonb.write(defaultMap(request.config())),
                operator(request.operator()), operator(request.operator()), Timestamp.from(now), Timestamp.from(now));
        Map<String, Object> created = get(id);
        audit(id, "created", request.operator(), Map.of(), created, "创建模型运行配置");
        return created;
    }

    @Transactional
    public Map<String, Object> update(String deploymentId, ModelManagementRequests.SaveRuntimeDeploymentRequest request) {
        Map<String, Object> before = get(deploymentId);
        if (Boolean.TRUE.equals(before.get("active"))) {
            throw conflict("生效中的运行配置不可直接修改，请先切换到其他配置");
        }
        String modelGroupId = String.valueOf(before.get("model_group_id"));
        requireGroupVersion(modelGroupId, request.modelVersionId());
        validateRequest(request);
        Integer duplicates = jdbc.queryForObject("""
                select count(*) from model_runtime_deployments
                where model_group_id=? and name=? and id<>?
                """, Integer.class, modelGroupId, request.name(), deploymentId);
        if (duplicates != null && duplicates > 0) {
            throw conflict("同一模型组中运行配置名称不能重复");
        }
        jdbc.update("""
                update model_runtime_deployments set
                    model_version_id=?, name=?, environment=?, service_url=?, health_path=?, runtime_path=?,
                    inference_path=?, test_path=?, runtime_engine=?, execution_provider=?, device=?, precision=?,
                    artifact_uri=?, auth_secret_ref=?, replicas=?, max_concurrency=?, request_timeout_seconds=?,
                    config=?::jsonb, status='ready', health_status='unknown', health_message=null,
                    health_snapshot='{}'::jsonb, last_health_at=null, connectivity_status='unknown',
                    connectivity_message=null, last_connectivity_at=null, revision=revision+1,
                    updated_by=?, updated_at=? where id=?
                """,
                request.modelVersionId(), request.name().trim(), request.environment(), normalizeServiceUrl(request.serviceUrl()),
                normalizePath(request.healthPath(), "/health"), normalizePath(request.runtimePath(), "/api/v1/runtime"),
                normalizePath(request.inferencePath(), null), normalizePath(request.testPath(), null),
                request.runtimeEngine(), request.executionProvider(), request.device(), request.precision(),
                blankToNull(request.artifactUri()), blankToNull(request.authSecretRef()), defaultInt(request.replicas(), 1),
                defaultInt(request.maxConcurrency(), 1), defaultInt(request.requestTimeoutSeconds(), 60),
                Jsonb.write(defaultMap(request.config())), operator(request.operator()), Timestamp.from(Instant.now()), deploymentId);
        Map<String, Object> updated = get(deploymentId);
        audit(deploymentId, "updated", request.operator(), before, updated, "修改模型运行配置");
        return updated;
    }

    @Transactional
    public Map<String, Object> delete(String deploymentId, String operator) {
        Map<String, Object> before = get(deploymentId);
        if (Boolean.TRUE.equals(before.get("active"))) {
            throw conflict("生效中的运行配置不能删除");
        }
        jdbc.update("delete from model_runtime_deployments where id=?", deploymentId);
        audit(deploymentId, "deleted", operator, before, Map.of(), "删除模型运行配置");
        return Map.of("id", deploymentId, "deleted", true);
    }

    public Map<String, Object> checkHealth(String deploymentId) {
        Map<String, Object> deployment = get(deploymentId);
        HealthProbe probe = probe(deployment);
        jdbc.update("""
                update model_runtime_deployments set health_status=?, health_message=?,
                    health_snapshot=?::jsonb, last_health_at=?, connectivity_status=?,
                    connectivity_message=?, last_connectivity_at=?, updated_at=? where id=?
                """, probe.status(), probe.message(), Jsonb.write(probe.snapshot()),
                Timestamp.from(Instant.now()), probe.connectivity().status(), probe.connectivity().message(),
                Timestamp.from(Instant.now()), Timestamp.from(Instant.now()), deploymentId);
        return get(deploymentId);
    }

    public Map<String, Object> checkConnectivity(String deploymentId) {
        Map<String, Object> deployment = get(deploymentId);
        ConnectivityProbe probe = probeConnectivity(deployment, restClient(deployment));
        jdbc.update("""
                update model_runtime_deployments set connectivity_status=?, connectivity_message=?,
                    last_connectivity_at=?, updated_at=? where id=?
                """, probe.status(), probe.message(), Timestamp.from(Instant.now()),
                Timestamp.from(Instant.now()), deploymentId);
        return get(deploymentId);
    }

    @Transactional
    public Map<String, Object> activate(String deploymentId, ModelManagementRequests.SwitchRuntimeDeploymentRequest request) {
        Map<String, Object> checked = checkHealth(deploymentId);
        String healthStatus = String.valueOf(checked.get("health_status"));
        Map<String, Object> config = mapValue(checked.get("config"));
        boolean force = request != null && Boolean.TRUE.equals(request.force());
        boolean fallbackAllowed = Boolean.TRUE.equals(config.get("allow_fallback"));
        if ("unreachable".equals(healthStatus) || "incompatible".equals(healthStatus)) {
            throw conflict("运行端点不可切换: " + checked.get("health_message"));
        }
        if (!"healthy".equals(healthStatus) && !("degraded".equals(healthStatus) && (fallbackAllowed || force))) {
            throw conflict("运行配置未达到切换条件，请先完成健康检查或明确允许降级运行");
        }
        String groupId = String.valueOf(checked.get("model_group_id"));
        List<Map<String, Object>> activeRows = jdbc.queryForList(
                "select id, name, model_version_id from model_runtime_deployments where model_group_id=? and active=true for update",
                groupId);
        Map<String, Object> before = activeRows.isEmpty() ? Map.of() : activeRows.get(0);
        if (deploymentId.equals(before.get("id"))) {
            return checked;
        }
        Instant now = Instant.now();
        jdbc.update("""
                update model_runtime_deployments set active=false, status='ready', updated_at=?, updated_by=?
                where model_group_id=? and active=true
                """, Timestamp.from(now), operator(request == null ? null : request.operator()), groupId);
        jdbc.update("""
                update model_runtime_deployments set active=true, status='active', updated_at=?, updated_by=? where id=?
                """, Timestamp.from(now), operator(request == null ? null : request.operator()), deploymentId);
        Map<String, Object> after = get(deploymentId);
        audit(deploymentId, "activated", request == null ? null : request.operator(), before, after,
                request == null ? "切换模型运行配置" : request.reason());
        return after;
    }

    public Optional<RuntimeRoute> activeRoute(String modelGroupCode) {
        List<Map<String, Object>> rows = jdbc.queryForList("""
                select d.service_url, d.inference_path, d.test_path, d.runtime_path,
                       d.request_timeout_seconds, d.config::text as config,
                       d.id as deployment_id, d.name as deployment_name,
                       v.id as model_version_id, v.version as model_version
                from model_runtime_deployments d
                join model_groups mg on mg.id=d.model_group_id
                join model_versions v on v.id=d.model_version_id
                where mg.code=? and d.active=true limit 1
                """, modelGroupCode);
        if (rows.isEmpty()) {
            return Optional.empty();
        }
        Map<String, Object> row = rows.get(0);
        return Optional.of(new RuntimeRoute(
                String.valueOf(row.get("deployment_id")), String.valueOf(row.get("deployment_name")),
                String.valueOf(row.get("service_url")), String.valueOf(row.get("inference_path")),
                String.valueOf(row.get("test_path")), String.valueOf(row.get("runtime_path")),
                String.valueOf(row.get("model_version_id")), String.valueOf(row.get("model_version")),
                ((Number) row.get("request_timeout_seconds")).intValue(), Jsonb.map(String.valueOf(row.get("config")))
        ));
    }

    static String normalizeServiceUrl(String value) {
        try {
            URI uri = URI.create(value == null ? "" : value.trim());
            String scheme = uri.getScheme() == null ? "" : uri.getScheme().toLowerCase(Locale.ROOT);
            if (!("http".equals(scheme) || "https".equals(scheme)) || uri.getHost() == null || uri.getUserInfo() != null) {
                throw new IllegalArgumentException();
            }
            String normalized = uri.toString();
            return normalized.endsWith("/") ? normalized.substring(0, normalized.length() - 1) : normalized;
        } catch (Exception e) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "服务地址必须是有效的 HTTP/HTTPS 地址，且不能包含账号信息");
        }
    }

    static String normalizePath(String value, String fallback) {
        String path = value == null || value.isBlank() ? fallback : value.trim();
        if (path == null || path.isBlank() || path.contains("..") || path.contains("?") || path.contains("#")) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "接口路径格式不正确");
        }
        return path.startsWith("/") ? path : "/" + path;
    }

    private HealthProbe probe(Map<String, Object> deployment) {
        RestTemplate client = restClient(deployment);
        ConnectivityProbe connectivity = probeConnectivity(deployment, client);
        Map<String, Object> snapshot = new LinkedHashMap<>(connectivity.snapshot());
        if (!"connected".equals(connectivity.status())) {
            return new HealthProbe("unreachable", connectivity.message(), snapshot, connectivity);
        }

        try {
            String serviceUrl = String.valueOf(deployment.get("service_url"));
            ResponseEntity<Map> runtimeResponse = client.getForEntity(serviceUrl + deployment.get("runtime_path"), Map.class);
            Map<String, Object> runtime = runtimeResponse.getBody() == null ? Map.of() : runtimeResponse.getBody();
            snapshot.put("profile", runtime.get("profile"));
            snapshot.put("runtime", runtime.get("runtime"));
            snapshot.put("execution_provider", runtime.get("execution_provider"));
            snapshot.put("available_providers", runtime.get("available_providers"));
            Map<String, Object> config = mapValue(deployment.get("config"));
            String expectedProfile = stringValue(config.get("expected_profile"));
            if (expectedProfile != null && !expectedProfile.equals(String.valueOf(runtime.get("profile")))) {
                return new HealthProbe("incompatible",
                        "运行配置档案不匹配，期望 " + expectedProfile + "，实际 " + runtime.get("profile"), snapshot,
                        connectivity);
            }
            Map<String, Object> model = matchingModel(runtime.get("models"), String.valueOf(deployment.get("model_group_code")),
                    String.valueOf(deployment.get("model_version")));
            if (model.isEmpty()) {
                return new HealthProbe("incompatible", "运行端点未注册当前模型组或版本", snapshot, connectivity);
            }
            snapshot.put("model", model);
            if (!"pointcloud-analyzer".equals(deployment.get("model_group_code"))
                    && Boolean.FALSE.equals(runtime.get("execution_provider_ready"))) {
                return new HealthProbe("incompatible", "配置的执行提供程序在运行端点中不可用", snapshot,
                        connectivity);
            }
            List<String> missingArtifacts = missingArtifactPaths(model);
            if (!missingArtifacts.isEmpty()) {
                snapshot.put("missing_artifacts", missingArtifacts);
                return new HealthProbe("degraded", "服务连接正常，待安装模型制品: "
                        + String.join("、", missingArtifacts), snapshot, connectivity);
            }
            return new HealthProbe("healthy", "运行端点、模型版本和制品均已就绪", snapshot, connectivity);
        } catch (Exception e) {
            snapshot.put("runtime_error", e.getClass().getSimpleName());
            return new HealthProbe("degraded", "服务连接正常，但无法读取运行时清单: " + conciseMessage(e),
                    snapshot, connectivity);
        }
    }

    private RestTemplate restClient(Map<String, Object> deployment) {
        int timeout = Math.min(30, ((Number) deployment.get("request_timeout_seconds")).intValue());
        return restTemplateBuilder
                .setConnectTimeout(Duration.ofSeconds(Math.min(10, timeout)))
                .setReadTimeout(Duration.ofSeconds(timeout))
                .build();
    }

    private ConnectivityProbe probeConnectivity(Map<String, Object> deployment, RestTemplate client) {
        String serviceUrl = String.valueOf(deployment.get("service_url"));
        Map<String, Object> snapshot = new LinkedHashMap<>();
        try {
            ResponseEntity<Map> health = client.getForEntity(serviceUrl + deployment.get("health_path"), Map.class);
            snapshot.put("health_http_status", health.getStatusCode().value());
            snapshot.put("health_response", health.getBody() == null ? Map.of() : health.getBody());
            if (!health.getStatusCode().is2xxSuccessful()) {
                return new ConnectivityProbe("unreachable", "健康接口返回 HTTP "
                        + health.getStatusCode().value(), snapshot);
            }
            return new ConnectivityProbe("connected", "服务连接正常（HTTP "
                    + health.getStatusCode().value() + "）", snapshot);
        } catch (Exception e) {
            snapshot.put("error", e.getClass().getSimpleName());
            return new ConnectivityProbe("unreachable", "无法连接运行端点: " + conciseMessage(e), snapshot);
        }
    }

    private List<String> missingArtifactPaths(Map<String, Object> model) {
        List<String> missing = new ArrayList<>();
        if (!Boolean.TRUE.equals(model.get("artifact_installed"))) {
            missing.add(Optional.ofNullable(stringValue(model.get("artifact_path"))).orElse("主模型制品"));
        }
        if (Boolean.FALSE.equals(model.get("secondary_artifact_installed"))) {
            missing.add(Optional.ofNullable(stringValue(model.get("secondary_artifact_path"))).orElse("辅助模型制品"));
        }
        return missing;
    }

    private Map<String, Object> matchingModel(Object value, String modelGroup, String modelVersion) {
        if (!(value instanceof List<?> list)) {
            return Map.of();
        }
        Map<String, Object> groupFallback = Map.of();
        for (Object item : list) {
            if (!(item instanceof Map<?, ?> raw)) {
                continue;
            }
            Map<String, Object> model = new LinkedHashMap<>();
            raw.forEach((key, entry) -> model.put(String.valueOf(key), entry));
            if (!modelGroup.equals(String.valueOf(model.get("model_group")))) {
                continue;
            }
            groupFallback = model;
            Object registeredVersion = model.get("model_version");
            if (registeredVersion == null || modelVersion.equals(String.valueOf(registeredVersion))) {
                return model;
            }
        }
        return groupFallback.get("model_version") == null ? groupFallback : Map.of();
    }

    private void validateRequest(ModelManagementRequests.SaveRuntimeDeploymentRequest request) {
        normalizeServiceUrl(request.serviceUrl());
        normalizePath(request.healthPath(), "/health");
        normalizePath(request.runtimePath(), "/api/v1/runtime");
        normalizePath(request.inferencePath(), null);
        normalizePath(request.testPath(), null);
        if (!List.of("local", "development", "staging", "production").contains(request.environment())) {
            throw badRequest("环境必须是 local、development、staging 或 production");
        }
        if (!List.of("FP32", "FP16", "INT8", "FP64").contains(request.precision().toUpperCase(Locale.ROOT))) {
            throw badRequest("精度仅支持 FP64、FP32、FP16 或 INT8");
        }
    }

    private void requireGroupVersion(String modelGroupId, String modelVersionId) {
        Integer count = jdbc.queryForObject("""
                select count(*) from model_versions v join model_groups mg on mg.code=v.model_group
                where mg.id=? and v.id=?
                """, Integer.class, modelGroupId, modelVersionId);
        if (count == null || count == 0) {
            throw badRequest("模型版本不属于当前模型组");
        }
    }

    private void requireModelGroup(String modelGroupId) {
        Integer count = jdbc.queryForObject("select count(*) from model_groups where id=?", Integer.class, modelGroupId);
        if (count == null || count == 0) {
            throw notFound("模型组不存在");
        }
    }

    private Map<String, Object> parseRow(Map<String, Object> source) {
        Map<String, Object> row = new LinkedHashMap<>(source);
        row.put("config", Jsonb.map(String.valueOf(row.getOrDefault("config", "{}"))));
        row.put("health_snapshot", Jsonb.map(String.valueOf(row.getOrDefault("health_snapshot", "{}"))));
        return row;
    }

    @SuppressWarnings("unchecked")
    private Map<String, Object> mapValue(Object value) {
        return value instanceof Map<?, ?> map ? (Map<String, Object>) map : Map.of();
    }

    private Map<String, Object> defaultMap(Map<String, Object> value) {
        return value == null ? Map.of() : value;
    }

    private int defaultInt(Integer value, int fallback) {
        return value == null ? fallback : value;
    }

    private String blankToNull(String value) {
        return value == null || value.isBlank() ? null : value.trim();
    }

    private String stringValue(Object value) {
        return value == null || String.valueOf(value).isBlank() ? null : String.valueOf(value);
    }

    private String operator(String value) {
        return value == null || value.isBlank() ? "模型运维管理员" : value;
    }

    private String conciseMessage(Exception e) {
        String message = e.getMessage();
        return message == null || message.isBlank() ? e.getClass().getSimpleName() : message;
    }

    private void audit(String entityId, String action, String operator, Map<String, Object> before,
                       Map<String, Object> after, String reason) {
        jdbc.update("""
                insert into model_audit_logs(id, entity_type, entity_id, action, operator_name,
                    before_snapshot, after_snapshot, reason, created_at)
                values (?,'runtime_deployment',?,?,?,?::jsonb,?::jsonb,?,?)
                """, Ids.next("audit"), entityId, action, operator(operator), Jsonb.write(before), Jsonb.write(after),
                reason, Timestamp.from(Instant.now()));
    }

    private ResponseStatusException notFound(String message) {
        return new ResponseStatusException(HttpStatus.NOT_FOUND, message);
    }

    private ResponseStatusException badRequest(String message) {
        return new ResponseStatusException(HttpStatus.BAD_REQUEST, message);
    }

    private ResponseStatusException conflict(String message) {
        return new ResponseStatusException(HttpStatus.CONFLICT, message);
    }

    public record RuntimeRoute(
            String deploymentId,
            String deploymentName,
            String serviceUrl,
            String inferencePath,
            String testPath,
            String runtimePath,
            String modelVersionId,
            String modelVersion,
            int requestTimeoutSeconds,
            Map<String, Object> config
    ) {
    }

    private record HealthProbe(String status, String message, Map<String, Object> snapshot,
                               ConnectivityProbe connectivity) {
    }

    private record ConnectivityProbe(String status, String message, Map<String, Object> snapshot) {
    }
}
