package com.ai.trackwalker.capability.service;

import com.ai.trackwalker.capability.api.CapabilityRequests;
import com.ai.trackwalker.capability.uav.MissionStateMachine;
import com.ai.trackwalker.capability.uav.InspectionTaskWorkflow;
import com.ai.trackwalker.capability.uav.UavVendorAdapter;
import com.ai.trackwalker.capability.uav.UavVendorAdapterRegistry;
import com.ai.trackwalker.capability.workorder.WorkOrderStateMachine;
import com.ai.trackwalker.common.Ids;
import com.ai.trackwalker.common.Jsonb;
import com.ai.trackwalker.storage.ModelTestStorageService;
import org.springframework.dao.EmptyResultDataAccessException;
import org.springframework.http.HttpStatus;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;
import org.springframework.web.server.ResponseStatusException;

import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.sql.Timestamp;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HexFormat;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;

@Service
public class CapabilityCompletionService {
    private final JdbcTemplate jdbc;
    private final KafkaTemplate<String, Object> kafkaTemplate;
    private final UavVendorAdapterRegistry adapterRegistry;
    private final ModelTestStorageService storageService;

    public CapabilityCompletionService(
            JdbcTemplate jdbc,
            KafkaTemplate<String, Object> kafkaTemplate,
            UavVendorAdapterRegistry adapterRegistry,
            ModelTestStorageService storageService
    ) {
        this.jdbc = jdbc;
        this.kafkaTemplate = kafkaTemplate;
        this.adapterRegistry = adapterRegistry;
        this.storageService = storageService;
    }

    public Map<String, Object> completionOverview() {
        Map<String, Object> result = new LinkedHashMap<>();
        result.put("capabilities", capabilityStatuses());
        result.put("counts", Map.ofEntries(
                Map.entry("plans", count("inspection_plans")),
                Map.entry("objects", count("inspection_objects")),
                Map.entry("routes", count("routes")),
                Map.entry("connections", count("uav_vendor_connections")),
                Map.entry("devices", count("uav_devices")),
                Map.entry("missions", count("flight_missions")),
                Map.entry("telemetry_points", count("telemetry_points")),
                Map.entry("organizations", count("organizations")),
                Map.entry("workorder_actions", count("workorder_actions")),
                Map.entry("archives", count("archive_packages"))
        ));
        result.put("external_dependencies", capabilityStatuses().stream()
                .filter(row -> "EXTERNAL_BLOCKED".equals(row.get("status")))
                .toList());
        return result;
    }

    public List<Map<String, Object>> capabilityStatuses() {
        return jdbc.queryForList("""
                select capability_code, capability_name, stage, status, implementation_ref,
                       blocker_type, blocker_reason, updated_at
                from capability_completion_status
                order by capability_code
                """);
    }

    public List<Map<String, Object>> listPlans() {
        return jdbc.queryForList("""
                select p.id as plan_id, p.name, p.plan_type, p.trigger_type, p.schedule_rule,
                       p.schedule_config::text as schedule_config,
                       p.object_ids::text as object_ids, p.scene_set::text as scene_set,
                       p.priority, p.status, p.next_run_at, p.owner_org_id,
                       o.name as owner_org_name, p.created_by, p.created_at, p.updated_at,
                       (select count(*) from inspection_tasks t where t.plan_id=p.id) as generated_task_count
                from inspection_plans p
                left join organizations o on o.id=p.owner_org_id
                order by p.updated_at desc
                """);
    }

    @Transactional
    public Map<String, Object> createPlan(CapabilityRequests.CreatePlanRequest request) {
        String id = Ids.next("plan");
        Instant now = Instant.now();
        String triggerType = defaultString(
                request.triggerType(),
                request.scheduleRule() == null || request.scheduleRule().isBlank() ? "MANUAL" : "PERIODIC"
        ).toUpperCase(Locale.ROOT);
        if (!List.of("PERIODIC", "MANUAL").contains(triggerType)) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "巡检触发方式仅支持 PERIODIC 或 MANUAL");
        }
        if ("PERIODIC".equals(triggerType) && (request.scheduleRule() == null || request.scheduleRule().isBlank())) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "周期巡检必须提供周期规则");
        }
        if ("PERIODIC".equals(triggerType) && request.nextRunAt() == null) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "周期巡检必须提供首次执行时间");
        }
        Map<String, Object> scheduleConfig = request.scheduleConfig() == null ? Map.of() : request.scheduleConfig();
        if ("MANUAL".equals(triggerType)
                && "SCHEDULED".equalsIgnoreCase(String.valueOf(scheduleConfig.get("timing")))
                && request.nextRunAt() == null) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "预约手动巡检必须提供执行时间");
        }
        String scheduleRule = "PERIODIC".equals(triggerType) ? request.scheduleRule() : null;
        jdbc.update("""
                        insert into inspection_plans(
                            id, name, plan_type, trigger_type, schedule_rule, schedule_config,
                            object_ids, scene_set, priority, status, next_run_at,
                            owner_org_id, created_by, created_at, updated_at
                        ) values (?,?,?,?,?,?::jsonb,?::jsonb,?::jsonb,?,'DRAFT',?,?,?,?,?)
                        """,
                id,
                request.name(),
                request.planType().toUpperCase(Locale.ROOT),
                triggerType,
                scheduleRule,
                Jsonb.write(scheduleConfig),
                Jsonb.write(request.objectIds()),
                Jsonb.write(request.sceneSet()),
                defaultString(request.priority(), "normal"),
                timestamp(request.nextRunAt()),
                defaultString(request.ownerOrgId(), "org-railway"),
                defaultString(request.createdBy(), "user-dispatcher"),
                Timestamp.from(now),
                Timestamp.from(now)
        );
        publish("inspection.plan.created", Map.of("plan_id", id, "trigger_type", triggerType));
        return Map.of("plan_id", id, "status", "DRAFT", "trigger_type", triggerType);
    }

    @Transactional
    public Map<String, Object> activatePlan(String planId) {
        int updated = jdbc.update("update inspection_plans set status='ACTIVE', updated_at=? where id=? and status in ('DRAFT','PAUSED')",
                Timestamp.from(Instant.now()), planId);
        if (updated == 0) {
            throw conflict("计划不存在或当前状态不能启用");
        }
        publish("inspection.plan.activated", Map.of("plan_id", planId));
        return Map.of("plan_id", planId, "status", "ACTIVE");
    }

    @Transactional
    public Map<String, Object> generatePlanTasks(String planId) {
        Map<String, Object> plan = requireRow("select *, object_ids::text as object_ids_text, scene_set::text as scene_set_text from inspection_plans where id=?", planId);
        if (!List.of("ACTIVE", "DRAFT").contains(String.valueOf(plan.get("status")))) {
            throw conflict("计划当前状态不能生成任务");
        }
        List<String> objectIds = Jsonb.stringList(String.valueOf(plan.get("object_ids_text")));
        List<String> taskIds = new ArrayList<>();
        Instant now = Instant.now();
        for (String objectId : objectIds) {
            Map<String, Object> object = requireRow("select * from inspection_objects where id=? and status='ACTIVE'", objectId);
            List<Map<String, Object>> routeRows = jdbc.queryForList("select id from routes where object_id=? and status='PUBLISHED' order by updated_at desc limit 1", objectId);
            String routeId = routeRows.isEmpty() ? null : String.valueOf(routeRows.get(0).get("id"));
            String taskId = Ids.next("task");
            jdbc.update("""
                            insert into inspection_tasks(
                                id, external_task_id, line_id, mileage_start, mileage_end, route_id,
                                scene_set, priority, status, created_at, updated_at, plan_id, task_type,
                                approval_status, object_scope, planned_start_at, created_by
                            ) values (?,?,?,?,?,?,?::jsonb,?,'created',?,?,?,?,?,?::jsonb,?,?)
                            """,
                    taskId,
                    "PLAN-" + planId + "-" + objectId,
                    object.get("line_id"),
                    object.get("mileage_start"),
                    object.get("mileage_end"),
                    routeId,
                    String.valueOf(plan.get("scene_set_text")),
                    plan.get("priority"),
                    Timestamp.from(now),
                    Timestamp.from(now),
                    planId,
                    plan.get("plan_type"),
                    "APPROVED",
                    Jsonb.write(List.of(objectId)),
                    plan.get("next_run_at"),
                    plan.get("created_by")
            );
            jdbc.update("insert into inspection_task_objects(task_id, object_id, required_actions) values (?,?,?::jsonb) on conflict do nothing",
                    taskId, objectId, "[]");
            taskIds.add(taskId);
        }
        jdbc.update("update inspection_plans set updated_at=? where id=?", Timestamp.from(now), planId);
        publish("inspection.plan.tasks.generated", Map.of("plan_id", planId, "task_ids", taskIds));
        return Map.of("plan_id", planId, "generated", taskIds.size(), "task_ids", taskIds);
    }

    public List<Map<String, Object>> listObjects() {
        return jdbc.queryForList("""
                select i.id as object_id, i.object_type, i.name, i.line_id, i.mileage_start,
                       i.mileage_end, ST_AsGeoJSON(i.geom) as geometry, i.attributes::text as attributes,
                       i.owner_org_id, o.name as owner_org_name, i.risk_level, i.inspection_cycle_days,
                       i.source_crs, i.transform_version, i.accuracy_m, i.status, i.created_at, i.updated_at
                from inspection_objects i
                left join organizations o on o.id=i.owner_org_id
                order by i.line_id, i.mileage_start, i.name
                """);
    }

    @Transactional
    public Map<String, Object> createObject(CapabilityRequests.CreateObjectRequest request) {
        String id = Ids.next("object");
        Instant now = Instant.now();
        int inserted = jdbc.update("""
                        insert into inspection_objects(
                            id, object_type, name, line_id, mileage_start, mileage_end, geom,
                            attributes, owner_org_id, risk_level, inspection_cycle_days, source_crs,
                            status, created_at, updated_at
                        ) values (?,?,?,?,?,?,ST_SetSRID(ST_GeomFromGeoJSON(?),4326),?::jsonb,?,?,?,?, 'ACTIVE',?,?)
                        """,
                id,
                request.objectType().toUpperCase(Locale.ROOT),
                request.name(),
                request.lineId(),
                request.mileageStart(),
                request.mileageEnd(),
                Jsonb.write(request.geometry()),
                Jsonb.write(request.attributes() == null ? Map.of() : request.attributes()),
                request.ownerOrgId(),
                defaultString(request.riskLevel(), "NORMAL"),
                request.inspectionCycleDays(),
                defaultString(request.sourceCrs(), "EPSG:4326"),
                Timestamp.from(now),
                Timestamp.from(now)
        );
        if (inserted != 1) {
            throw new ResponseStatusException(HttpStatus.INTERNAL_SERVER_ERROR, "巡检对象创建失败");
        }
        publish("inspection.object.created", Map.of("object_id", id));
        return Map.of("object_id", id, "status", "ACTIVE");
    }

    public List<Map<String, Object>> listRoutes() {
        return jdbc.queryForList("""
                select r.id as route_id, r.name, r.object_id, o.name as object_name, o.object_type,
                       r.template_id, t.name as template_name, r.status, r.current_version_id,
                       rv.version_no, rv.status as version_status, ST_AsGeoJSON(rv.geom) as geometry,
                       rv.waypoints::text as waypoints, rv.flight_parameters::text as flight_parameters,
                       rv.payload_actions::text as payload_actions, rv.validation_result::text as validation_result,
                       rv.checksum, rv.published_at, r.created_at, r.updated_at
                from routes r
                left join inspection_objects o on o.id=r.object_id
                left join route_templates t on t.id=r.template_id
                left join route_versions rv on rv.id=r.current_version_id
                order by r.updated_at desc
                """);
    }

    @Transactional
    public Map<String, Object> createRoute(CapabilityRequests.CreateRouteRequest request) {
        if (request.waypoints().size() < 2) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "航线至少需要两个航点");
        }
        String routeId = Ids.next("route");
        String versionId = Ids.next("route-version");
        Instant now = Instant.now();
        jdbc.update("insert into routes(id,name,object_id,template_id,status,current_version_id,created_by,created_at,updated_at) values (?,?,?,?,'DRAFT',?,?,?,?)",
                routeId, request.name(), blankToNull(request.objectId()), blankToNull(request.templateId()), versionId,
                defaultString(request.createdBy(), "user-dispatcher"), Timestamp.from(now), Timestamp.from(now));
        jdbc.update("""
                        insert into route_versions(
                            id,route_id,version_no,status,waypoints,flight_parameters,payload_actions,
                            validation_result,created_by,created_at
                        ) values (?,?,1,'DRAFT',?::jsonb,?::jsonb,?::jsonb,'{}'::jsonb,?,?)
                        """,
                versionId,
                routeId,
                Jsonb.write(request.waypoints()),
                Jsonb.write(request.flightParameters() == null ? Map.of() : request.flightParameters()),
                Jsonb.write(request.payloadActions() == null ? List.of() : request.payloadActions()),
                defaultString(request.createdBy(), "user-dispatcher"),
                Timestamp.from(now));
        for (int i = 0; i < request.waypoints().size(); i++) {
            Map<String, Object> waypoint = request.waypoints().get(i);
            double longitude = number(waypoint, "longitude", 0);
            double latitude = number(waypoint, "latitude", 0);
            double altitude = number(waypoint, "altitude_m", 80);
            double speed = number(waypoint, "speed_mps", number(request.flightParameters(), "speed_mps", 6));
            jdbc.update("""
                            insert into route_waypoints(
                                id,route_version_id,sequence_no,geom,altitude_m,speed_mps,
                                heading_deg,gimbal_pitch_deg,actions
                            ) values (?,?,?,ST_SetSRID(ST_MakePoint(?,?,?),4326),?,?,?,?,?::jsonb)
                            """,
                    Ids.next("wp"), versionId, i + 1, longitude, latitude, altitude, altitude, speed,
                    nullableNumber(waypoint.get("heading_deg")), nullableNumber(waypoint.get("gimbal_pitch_deg")),
                    Jsonb.write(waypoint.getOrDefault("actions", List.of())));
        }
        jdbc.update("""
                update route_versions rv
                set geom=(select ST_MakeLine(w.geom order by w.sequence_no) from route_waypoints w where w.route_version_id=rv.id)
                where rv.id=?
                """, versionId);
        Map<String, Object> validation = validateRouteVersion(versionId);
        publish("route.created", Map.of("route_id", routeId, "route_version_id", versionId));
        Map<String, Object> result = new LinkedHashMap<>();
        result.put("route_id", routeId);
        result.put("route_version_id", versionId);
        result.put("status", "DRAFT");
        result.put("validation", validation);
        return result;
    }

    @Transactional
    public Map<String, Object> validateRouteVersion(String versionId) {
        Map<String, Object> stats = requireRow("""
                select count(*) as waypoint_count, min(altitude_m) as min_altitude,
                       max(altitude_m) as max_altitude, max(speed_mps) as max_speed
                from route_waypoints where route_version_id=?
                """, versionId);
        List<String> errors = new ArrayList<>();
        List<String> warnings = new ArrayList<>();
        int waypointCount = ((Number) stats.get("waypoint_count")).intValue();
        if (waypointCount < 2) errors.add("航线至少需要两个航点");
        double minAltitude = nullableNumber(stats.get("min_altitude")) == null ? 0 : nullableNumber(stats.get("min_altitude"));
        double maxAltitude = nullableNumber(stats.get("max_altitude")) == null ? 0 : nullableNumber(stats.get("max_altitude"));
        double maxSpeed = nullableNumber(stats.get("max_speed")) == null ? 0 : nullableNumber(stats.get("max_speed"));
        if (minAltitude < 20) errors.add("最低飞行高度小于平台安全下限 20 米");
        if (maxAltitude > 500) errors.add("最高飞行高度超过平台配置上限 500 米");
        if (maxSpeed > 20) errors.add("航点速度超过平台配置上限 20 米/秒");
        if (waypointCount > 500) warnings.add("航点数量较多，需要在厂商转换时再次校验限制");
        if (maxAltitude - minAltitude > 120) warnings.add("航线高差较大，需要复核地形和返航高度");
        Map<String, Object> validation = new LinkedHashMap<>();
        validation.put("valid", errors.isEmpty());
        validation.put("errors", errors);
        validation.put("warnings", warnings);
        validation.put("waypoint_count", waypointCount);
        validation.put("min_altitude_m", minAltitude);
        validation.put("max_altitude_m", maxAltitude);
        validation.put("max_speed_mps", maxSpeed);
        validation.put("checked_at", Instant.now().toString());
        jdbc.update("update route_versions set validation_result=?::jsonb where id=?", Jsonb.write(validation), versionId);
        return validation;
    }

    @Transactional
    public Map<String, Object> publishRouteVersion(String versionId, String approvedBy) {
        Map<String, Object> version = requireRow("select route_id, status from route_versions where id=?", versionId);
        Map<String, Object> validation = validateRouteVersion(versionId);
        if (!(Boolean) validation.get("valid")) {
            throw conflict("航线校验未通过，不能发布");
        }
        Instant now = Instant.now();
        jdbc.update("update route_versions set status='PUBLISHED', approved_by=?, published_at=? where id=?",
                defaultString(approvedBy, "user-dispatcher"), Timestamp.from(now), versionId);
        jdbc.update("update routes set status='PUBLISHED', current_version_id=?, updated_at=? where id=?",
                versionId, Timestamp.from(now), version.get("route_id"));
        publish("route.published", Map.of("route_id", version.get("route_id"), "route_version_id", versionId));
        return Map.of("route_id", version.get("route_id"), "route_version_id", versionId, "status", "PUBLISHED");
    }

    public Map<String, Object> synchronizeRouteVersion(String versionId, String connectionId) {
        Map<String, Object> connection = requireRow(
                "select * from uav_vendor_connections where id=? and enabled=true",
                connectionId
        );
        Map<String, Object> route = requireRow("""
                select r.id as route_id,r.name,r.status,rv.id as route_version_id,
                       rv.status as version_status,rv.validation_result::text as validation_result,
                       rv.waypoints::text as waypoints,rv.flight_parameters::text as flight_parameters
                from route_versions rv join routes r on r.id=rv.route_id
                where rv.id=?
                """, versionId);
        if (!"PUBLISHED".equals(route.get("status")) || !"PUBLISHED".equals(route.get("version_status"))) {
            throw conflict("航线版本必须校验并发布后才能同步到无人机厂商");
        }
        UavVendorAdapter adapter = adapterRegistry.require(String.valueOf(connection.get("vendor_code")));
        Map<String, Object> artifact = adapter.prepareRoute(connection, route);
        insertVendorEvent(
                connection,
                "route-synchronized-" + versionId + "-" + Instant.now().toEpochMilli(),
                "ROUTE_SYNCHRONIZED",
                artifact
        );
        publish("uav.route.synchronized", Map.of(
                "connection_id", connectionId,
                "route_version_id", versionId,
                "import_state", artifact.getOrDefault("import_state", "UNKNOWN")
        ));
        return artifact;
    }

    public List<Map<String, Object>> listConnections() {
        return jdbc.queryForList("""
                select id as connection_id, name, vendor_code, connection_mode, environment,
                       endpoint_url, workspace_id, organization_uuid, project_uuid, integration_version,
                       reconciliation_status, capabilities::text as capabilities,
                       enabled, status, health_detail::text as health_detail,
                       last_checked_at, last_event_at, created_at, updated_at,
                       (select count(*) from uav_devices d where d.connection_id=c.id) as device_count
                from uav_vendor_connections c order by vendor_code, environment
                """);
    }

    @Transactional
    public Map<String, Object> createConnection(CapabilityRequests.CreateConnectionRequest request) {
        String id = Ids.next("connection");
        String vendor = request.vendorCode().toUpperCase(Locale.ROOT);
        String status = "SIMULATOR".equals(vendor) ? "AVAILABLE" : "MATERIALS_REQUIRED";
        Instant now = Instant.now();
        jdbc.update("""
                        insert into uav_vendor_connections(
                            id,name,vendor_code,connection_mode,environment,endpoint_url,workspace_id,
                            secret_ref,capabilities,enabled,status,health_detail,created_at,updated_at
                        ) values (?,?,?,?,?,?,?,?,?::jsonb,?,?,?::jsonb,?,?)
                        """,
                id, request.name(), vendor, request.connectionMode(), request.environment(), request.endpointUrl(),
                request.workspaceId(), request.secretRef(), "[]", request.enabled() == null || request.enabled(), status,
                Jsonb.write(Map.of("network", "unchecked", "authentication", "unchecked")),
                Timestamp.from(now), Timestamp.from(now));
        if ("FLIGHTHUB2".equals(vendor)) {
            jdbc.update("""
                    update uav_vendor_connections
                    set project_uuid=?,integration_version='OPENAPI_V2',
                        reconciliation_status='PENDING',updated_at=?
                    where id=?
                    """, request.workspaceId(), Timestamp.from(now), id);
        }
        return Map.of("connection_id", id, "status", status);
    }

    @Transactional
    public Map<String, Object> checkConnection(String connectionId) {
        Map<String, Object> connection = requireRow("select * from uav_vendor_connections where id=?", connectionId);
        String vendor = String.valueOf(connection.get("vendor_code"));
        Map<String, Object> health;
        if ("SIMULATOR".equalsIgnoreCase(vendor) || "FLIGHTHUB2".equalsIgnoreCase(vendor)) {
            health = adapterRegistry.require(vendor).checkConnection(connection);
        } else {
            health = externalMaterialsHealth(vendor);
        }
        String status = String.valueOf(health.get("status"));
        jdbc.update("""
                        update uav_vendor_connections set status=?,health_detail=?::jsonb,
                            capabilities=?::jsonb,organization_uuid=coalesce(?,organization_uuid),
                            project_uuid=coalesce(?,project_uuid),
                            integration_version=coalesce(?,integration_version),
                            reconciliation_status=coalesce(?,reconciliation_status),
                            last_checked_at=?,updated_at=?
                        where id=?
                        """,
                status, Jsonb.write(health), Jsonb.write(health.getOrDefault("capabilities", List.of())),
                health.get("organization_uuid"), health.get("project_uuid"),
                "FLIGHTHUB2".equalsIgnoreCase(vendor) ? "OPENAPI_V2" : null,
                "FLIGHTHUB2".equalsIgnoreCase(vendor)
                        ? ("AVAILABLE".equalsIgnoreCase(status) ? "SYNCHRONIZED" : "ATTENTION_REQUIRED")
                        : null,
                Timestamp.from(Instant.now()), Timestamp.from(Instant.now()), connectionId);
        return health;
    }

    @Transactional
    public Map<String, Object> synchronizeDevices(String connectionId) {
        Map<String, Object> connection = requireRow("select * from uav_vendor_connections where id=?", connectionId);
        String vendor = String.valueOf(connection.get("vendor_code"));
        UavVendorAdapter adapter = adapterRegistry.require(vendor);
        List<Map<String, Object>> devices = adapter.synchronizeDevices(connection);
        Instant now = Instant.now();
        List<String> deviceIds = new ArrayList<>();
        for (Map<String, Object> device : devices) {
            String deviceType = String.valueOf(device.getOrDefault("device_type", "UNKNOWN")).toUpperCase(Locale.ROOT);
            if ("DOCK".equals(deviceType)) {
                upsertDock(connectionId, device, now);
                continue;
            }
            if (!List.of("AIRCRAFT", "UAV", "DRONE").contains(deviceType)) {
                continue;
            }
            String internalId = "SIM-UAV-01".equals(device.get("vendor_device_id")) ? "uav-sim-01" : Ids.next("uav");
            jdbc.update("""
                            insert into uav_devices(
                                id,connection_id,vendor_device_id,vendor_code,device_type,model,serial_number,
                                capabilities,status,health_detail,geom,last_seen_at,created_at,updated_at,
                                gateway_vendor_device_id,device_model_key,firmware_version,project_uuid,source_updated_at
                            ) values (?,?,?,?,?,?,?,?::jsonb,?,?::jsonb,ST_SetSRID(ST_MakePoint(?,?),4326),?,?,?,?,?,?,?,?)
                            on conflict (connection_id,vendor_device_id) do update set
                                model=excluded.model, capabilities=excluded.capabilities, status=excluded.status,
                                health_detail=excluded.health_detail, geom=excluded.geom,
                                last_seen_at=excluded.last_seen_at, updated_at=excluded.updated_at,
                                gateway_vendor_device_id=excluded.gateway_vendor_device_id,
                                device_model_key=excluded.device_model_key,
                                firmware_version=excluded.firmware_version,
                                project_uuid=excluded.project_uuid,
                                source_updated_at=excluded.source_updated_at
                            """,
                    internalId, connectionId, device.get("vendor_device_id"), vendor, device.get("device_type"),
                    device.get("model"), device.get("serial_number"), Jsonb.write(device.get("capabilities")),
                    device.get("status"), Jsonb.write(device.get("health_detail")), device.get("longitude"),
                    device.get("latitude"), Timestamp.from(now), Timestamp.from(now), Timestamp.from(now),
                    device.get("parent_vendor_device_id"), device.get("device_model_key"),
                    mapValue(device.get("health_detail"), "firmware_version"),
                    connection.get("project_uuid") == null ? connection.get("workspace_id") : connection.get("project_uuid"),
                    Timestamp.from(now));
            String actualId = jdbc.queryForObject("select id from uav_devices where connection_id=? and vendor_device_id=?", String.class,
                    connectionId, device.get("vendor_device_id"));
            deviceIds.add(actualId);
        }
        jdbc.update("update uav_vendor_connections set last_event_at=?, updated_at=? where id=?", Timestamp.from(now), Timestamp.from(now), connectionId);
        publish("uav.devices.synchronized", Map.of("connection_id", connectionId, "device_ids", deviceIds));
        return Map.of(
                "connection_id", connectionId,
                "synchronized", devices.size(),
                "aircraft_ids", deviceIds,
                "device_ids", deviceIds
        );
    }

    public List<Map<String, Object>> listDevices() {
        return jdbc.queryForList("""
                select d.id as device_id, d.connection_id, c.name as connection_name, d.vendor_device_id,
                       d.vendor_code, d.device_type, d.model, d.serial_number,
                       d.capabilities::text as capabilities, d.status,
                       d.gateway_vendor_device_id, d.device_model_key, d.firmware_version, d.project_uuid,
                       d.health_detail::text as health_detail, ST_AsGeoJSON(d.geom) as geometry,
                       d.last_seen_at, d.created_at, d.updated_at
                from uav_devices d join uav_vendor_connections c on c.id=d.connection_id
                order by d.vendor_code, d.vendor_device_id
                """);
    }

    public List<Map<String, Object>> listMissions() {
        return jdbc.queryForList("""
                select m.id as mission_id, m.task_id, m.route_version_id, rv.route_id,
                       r.name as route_name, m.connection_id, c.name as connection_name,
                       m.device_id, d.model as device_model, m.dock_id, m.vendor_code,
                       m.vendor_mission_id, m.status, m.progress, m.current_waypoint,
                       m.validation_snapshot::text as validation_snapshot, m.error_code,
                       m.error_message, m.planned_start_at, m.started_at, m.completed_at,
                       m.version_no, m.created_at, m.updated_at
                from flight_missions m
                join uav_vendor_connections c on c.id=m.connection_id
                join uav_devices d on d.id=m.device_id
                left join route_versions rv on rv.id=m.route_version_id
                left join routes r on r.id=rv.route_id
                order by m.updated_at desc
                """);
    }

    @Transactional
    public Map<String, Object> dispatchMission(CapabilityRequests.DispatchMissionRequest request) {
        String idempotencyKey = defaultString(request.idempotencyKey(), "dispatch-" + request.taskId());
        List<Map<String, Object>> existing = jdbc.queryForList("""
                select m.id as mission_id, m.task_id, m.status, m.vendor_mission_id
                from uav_commands c join flight_missions m on m.id=c.mission_id
                where c.idempotency_key=?
                """, idempotencyKey);
        if (!existing.isEmpty()) {
            if (!request.taskId().equals(existing.get(0).get("task_id"))) {
                throw conflict("幂等键已被其他巡检任务使用");
            }
            return existing.get(0);
        }

        Map<String, Object> task = requireRow("select * from inspection_tasks where id=? for update", request.taskId());
        boolean hasMission = Boolean.TRUE.equals(jdbc.queryForObject(
                "select exists(select 1 from flight_missions where task_id=?)",
                Boolean.class,
                request.taskId()
        ));
        String assignedRouteId = task.get("route_id") == null ? null : blankToNull(String.valueOf(task.get("route_id")));
        String requestedRouteId = blankToNull(request.routeId());
        if (requestedRouteId != null && !requestedRouteId.equals(assignedRouteId)) {
            throw conflict("下发航线与任务已审批航线不一致，请先更新任务航线");
        }
        String routeId = assignedRouteId;
        String preflightBlockReason = InspectionTaskWorkflow.dispatchBlockReason(
                String.valueOf(task.get("status")),
                String.valueOf(task.get("approval_status")),
                routeId != null,
                true,
                hasMission
        );
        if (preflightBlockReason != null) {
            throw conflict(preflightBlockReason);
        }
        Map<String, Object> route = requireRow("""
                select r.id as route_id, r.name, r.status, rv.id as route_version_id,
                       rv.status as version_status, rv.validation_result::text as validation_result,
                       rv.waypoints::text as waypoints, rv.flight_parameters::text as flight_parameters
                from routes r join route_versions rv on rv.id=r.current_version_id
                where r.id=?
                """, routeId);
        if (!"PUBLISHED".equals(route.get("status")) || !"PUBLISHED".equals(route.get("version_status"))) {
            throw conflict("航线尚未发布，不能下发");
        }
        Map<String, Object> connection = requireRow("select * from uav_vendor_connections where id=? and enabled=true", request.connectionId());
        Map<String, Object> device = requireRow("select * from uav_devices where id=? and connection_id=?", request.deviceId(), request.connectionId());
        if (!"ONLINE".equals(device.get("status"))) throw conflict("无人机当前不在线");
        String dockVendorDeviceId = null;
        String requestedDockId = blankToNull(request.dockId());
        if (requestedDockId != null) {
            Map<String, Object> dock = requireRow(
                    "select * from uav_docks where id=? and connection_id=?",
                    requestedDockId,
                    request.connectionId()
            );
            dockVendorDeviceId = String.valueOf(dock.get("vendor_dock_id"));
        } else if (device.get("gateway_vendor_device_id") != null) {
            dockVendorDeviceId = String.valueOf(device.get("gateway_vendor_device_id"));
        }
        String vendor = String.valueOf(connection.get("vendor_code"));
        UavVendorAdapter adapter = adapterRegistry.require(vendor);

        String missionId = Ids.next("mission");
        String commandId = Ids.next("command");
        Instant now = Instant.now();
        jdbc.update("""
                        insert into flight_missions(
                            id,task_id,route_version_id,connection_id,device_id,dock_id,vendor_code,
                            status,progress,validation_snapshot,planned_start_at,created_at,updated_at
                        ) values (?,?,?,?,?,?,?,'DISPATCHING',0,?::jsonb,?,?,?)
                        """,
                missionId, request.taskId(), route.get("route_version_id"), request.connectionId(), request.deviceId(),
                blankToNull(request.dockId()), vendor, String.valueOf(route.get("validation_result")),
                request.plannedStartAt() == null ? task.get("planned_start_at") : timestamp(request.plannedStartAt()),
                Timestamp.from(now), Timestamp.from(now));
        jdbc.update("""
                        insert into uav_commands(
                            id,mission_id,command_type,idempotency_key,requested_by,reason,request_payload,
                            status,attempt_count,requested_at
                        ) values (?,?, 'MISSION_DISPATCH',?,?,?,?::jsonb, 'SENDING',1,?)
                        """,
                commandId, missionId, idempotencyKey, defaultString(request.requestedBy(), "user-dispatcher"),
                "巡检任务航线下发", Jsonb.write(Map.of("route_id", routeId, "device_id", request.deviceId())), Timestamp.from(now));
        Map<String, Object> adapterMission = new HashMap<>();
        adapterMission.put("mission_id", missionId);
        adapterMission.put("task_id", request.taskId());
        adapterMission.put("device_id", request.deviceId());
        adapterMission.put("connection_id", request.connectionId());
        adapterMission.put("vendor_device_id", device.get("vendor_device_id"));
        adapterMission.put("dock_vendor_device_id", dockVendorDeviceId);
        adapterMission.put("planned_start_at", request.plannedStartAt() == null
                ? task.get("planned_start_at")
                : request.plannedStartAt());
        adapterMission.put("idempotency_key", idempotencyKey);
        Map<String, Object> receipt = adapter.dispatchMission(adapterMission, route);
        String vendorMissionId = String.valueOf(receipt.get("vendor_mission_id"));
        String vendorTransactionId = String.valueOf(receipt.get("vendor_transaction_id"));
        boolean asynchronous = Boolean.TRUE.equals(receipt.get("async"));
        if (asynchronous) {
            jdbc.update("""
                            update uav_commands set vendor_transaction_id=?,response_payload=?::jsonb,
                                status='ACCEPTED',accepted_at=? where id=?
                            """,
                    vendorTransactionId, Jsonb.write(receipt), Timestamp.from(Instant.now()), commandId);
        } else {
            jdbc.update("""
                            update uav_commands set vendor_transaction_id=?,response_payload=?::jsonb,
                                status='SUCCEEDED',accepted_at=?,completed_at=? where id=?
                            """,
                    vendorTransactionId, Jsonb.write(receipt), Timestamp.from(Instant.now()),
                    Timestamp.from(Instant.now()), commandId);
        }
        jdbc.update("update flight_missions set vendor_mission_id=?, status='DISPATCHED', version_no=version_no+1, updated_at=? where id=?",
                vendorMissionId, Timestamp.from(Instant.now()), missionId);
        jdbc.update("update inspection_tasks set status='dispatched', updated_at=?, version_no=version_no+1 where id=?",
                Timestamp.from(Instant.now()), request.taskId());
        insertVendorEvent(connection, "mission-dispatched-" + missionId, "MISSION_DISPATCHED", receipt);
        publish("uav.mission.dispatched", Map.of("mission_id", missionId, "task_id", request.taskId(), "vendor", vendor));
        return Map.of("mission_id", missionId, "vendor_mission_id", vendorMissionId, "status", "DISPATCHED", "receipt", receipt);
    }

    @Transactional
    public Map<String, Object> executeMissionCommand(String missionId, CapabilityRequests.MissionCommandRequest request) {
        String idempotencyKey = defaultString(request.idempotencyKey(), request.command() + "-" + missionId + "-" + Instant.now().toEpochMilli());
        List<Map<String, Object>> existing = jdbc.queryForList("select id as command_id, status, response_payload::text as response_payload from uav_commands where idempotency_key=?", idempotencyKey);
        if (!existing.isEmpty()) return existing.get(0);

        Map<String, Object> mission = requireRow("select * from flight_missions where id=? for update", missionId);
        long currentVersion = ((Number) mission.get("version_no")).longValue();
        if (request.expectedMissionVersion() != null && request.expectedMissionVersion() != currentVersion) {
            throw conflict("任务状态已变化，请刷新后重试");
        }
        String currentStatus = String.valueOf(mission.get("status"));
        String command = request.command().toUpperCase(Locale.ROOT);
        String targetStatus = MissionStateMachine.targetForCommand(currentStatus, command);
        int progress = ((Number) mission.get("progress")).intValue();
        if ("SIMULATE_PROGRESS".equals(command)) {
            if (!"FLYING".equals(currentStatus)) throw conflict("只有飞行中的模拟任务可以推进进度");
            progress = Math.min(100, progress + 25);
            if (progress >= 100) targetStatus = "COMPLETED";
        } else if (!MissionStateMachine.canTransition(currentStatus, targetStatus)) {
            throw conflict("飞行任务不能从 " + currentStatus + " 执行 " + command + " 到 " + targetStatus);
        }

        Map<String, Object> connection = requireRow("select * from uav_vendor_connections where id=?", mission.get("connection_id"));
        UavVendorAdapter adapter = adapterRegistry.require(String.valueOf(connection.get("vendor_code")));
        String commandId = Ids.next("command");
        Instant now = Instant.now();
        jdbc.update("""
                        insert into uav_commands(
                            id,mission_id,command_type,idempotency_key,requested_by,reason,request_payload,
                            status,attempt_count,requested_at
                        ) values (?,?,?,?,?,?,?::jsonb,'SENDING',1,?)
                        """,
                commandId, missionId, command, idempotencyKey, defaultString(request.requestedBy(), "user-dispatcher"),
                request.reason(), Jsonb.write(request.parameters() == null ? Map.of() : request.parameters()), Timestamp.from(now));
        Map<String, Object> adapterParameters = new HashMap<>(
                request.parameters() == null ? Map.of() : request.parameters()
        );
        adapterParameters.put("idempotency_key", idempotencyKey);
        adapterParameters.put("platform_command_id", commandId);
        adapterParameters.put("requested_by", defaultString(request.requestedBy(), "user-dispatcher"));
        adapterParameters.put("reason", request.reason());
        Map<String, Object> receipt = adapter.executeCommand(mission, command, adapterParameters);
        boolean asynchronous = Boolean.TRUE.equals(receipt.get("async"));
        if (asynchronous) {
            jdbc.update("""
                            update uav_commands set vendor_transaction_id=?,response_payload=?::jsonb,
                                status='ACCEPTED',accepted_at=? where id=?
                            """,
                    receipt.get("vendor_transaction_id"), Jsonb.write(receipt), Timestamp.from(Instant.now()), commandId);
            insertVendorEvent(connection, "command-accepted-" + commandId, "MISSION_COMMAND_ACCEPTED", receipt);
            publish("uav.mission.command.accepted", Map.of(
                    "mission_id", missionId,
                    "command", command,
                    "status", currentStatus
            ));
            return Map.of(
                    "mission_id", missionId,
                    "command_id", commandId,
                    "status", currentStatus,
                    "progress", progress,
                    "accepted", true,
                    "receipt", receipt
            );
        }
        jdbc.update("""
                        update uav_commands set vendor_transaction_id=?,response_payload=?::jsonb,
                            status='SUCCEEDED',accepted_at=?,completed_at=? where id=?
                        """,
                receipt.get("vendor_transaction_id"), Jsonb.write(receipt), Timestamp.from(Instant.now()),
                Timestamp.from(Instant.now()), commandId);
        Instant startedAt = "FLYING".equals(targetStatus) && mission.get("started_at") == null ? now : null;
        Instant completedAt = "COMPLETED".equals(targetStatus) ? now : null;
        int targetProgress = "COMPLETED".equals(targetStatus) ? 100 : ("FLYING".equals(targetStatus) && progress == 0 ? 10 : progress);
        jdbc.update("""
                        update flight_missions set status=?, progress=?,
                            started_at=coalesce(started_at,?), completed_at=coalesce(completed_at,?),
                            version_no=version_no+1, updated_at=? where id=?
                        """,
                targetStatus, targetProgress, startedAt == null ? null : Timestamp.from(startedAt),
                completedAt == null ? null : Timestamp.from(completedAt), Timestamp.from(now), missionId);
        updateTaskFromMission(String.valueOf(mission.get("task_id")), targetStatus);
        if (List.of("FLYING", "RETURNING", "COMPLETED").contains(targetStatus)) {
            insertSimulatedTelemetry(missionId, String.valueOf(mission.get("device_id")), targetProgress);
        }
        insertVendorEvent(connection, "command-" + commandId, "MISSION_COMMAND_COMPLETED", receipt);
        publish("uav.mission.command.completed", Map.of("mission_id", missionId, "command", command, "status", targetStatus));
        return Map.of("mission_id", missionId, "command_id", commandId, "status", targetStatus, "progress", targetProgress, "receipt", receipt);
    }

    public List<Map<String, Object>> listTelemetry(String missionId) {
        return jdbc.queryForList("""
                select id, mission_id, device_id, event_time, received_at,
                       ST_X(geom) as longitude, ST_Y(geom) as latitude, ST_Z(geom) as altitude_m,
                       metrics::text as metrics, quality, sequence_no
                from telemetry_points where mission_id=? order by event_time
                """, missionId);
    }

    @Transactional
    public Map<String, Object> createUploadSession(CapabilityRequests.CreateUploadSessionRequest request) {
        if (request.totalSizeBytes() <= 0 || request.totalSizeBytes() > 200L * 1024 * 1024 * 1024) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "文件大小必须在 1 字节到 200 GB 之间");
        }
        String taskId = blankToNull(request.taskId());
        if (taskId == null && request.missionId() != null) {
            taskId = String.valueOf(requireRow("select task_id from flight_missions where id=?", request.missionId()).get("task_id"));
        }
        if (taskId == null) throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "必须提供 task_id 或 mission_id");
        requireRow("select id from inspection_tasks where id=?", taskId);
        long partSize = request.partSizeBytes() == null ? 8L * 1024 * 1024 : request.partSizeBytes();
        if (partSize < 1024 * 1024 || partSize > 64L * 1024 * 1024) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "分片大小必须在 1 MB 到 64 MB 之间");
        }
        int totalParts = Math.toIntExact((request.totalSizeBytes() + partSize - 1) / partSize);
        String id = Ids.next("upload");
        Instant now = Instant.now();
        jdbc.update("""
                        insert into upload_sessions(
                            id,task_id,mission_id,resource_type,file_name,content_type,total_size_bytes,
                            part_size_bytes,total_parts,expected_sha256,status,expires_at,created_by,created_at,updated_at
                        ) values (?,?,?,?,?,?,?,?,?,?, 'UPLOADING',?,?,?,?)
                        """,
                id, taskId, blankToNull(request.missionId()), request.resourceType().toUpperCase(Locale.ROOT),
                safeFileName(request.fileName()), defaultString(request.contentType(), "application/octet-stream"),
                request.totalSizeBytes(), partSize, totalParts, blankToNull(request.expectedSha256()),
                Timestamp.from(now.plus(7, ChronoUnit.DAYS)), defaultString(request.createdBy(), "数据接入人员"),
                Timestamp.from(now), Timestamp.from(now));
        return Map.ofEntries(
                Map.entry("upload_session_id", id),
                Map.entry("task_id", taskId),
                Map.entry("part_size_bytes", partSize),
                Map.entry("total_parts", totalParts),
                Map.entry("status", "UPLOADING"),
                Map.entry("expires_at", now.plus(7, ChronoUnit.DAYS).toString())
        );
    }

    public Map<String, Object> uploadSession(String sessionId) {
        Map<String, Object> session = requireRow("""
                select id as upload_session_id,task_id,mission_id,resource_type,file_name,content_type,
                       total_size_bytes,part_size_bytes,total_parts,expected_sha256,actual_sha256,
                       object_key,status,expires_at,created_by,created_at,updated_at
                from upload_sessions where id=?
                """, sessionId);
        session.put("parts", jdbc.queryForList("select part_no,size_bytes,sha256,etag,status,uploaded_at from upload_parts where session_id=? order by part_no", sessionId));
        return session;
    }

    @Transactional
    public Map<String, Object> uploadPart(String sessionId, int partNo, MultipartFile chunk) {
        Map<String, Object> session = requireRow("select * from upload_sessions where id=? for update", sessionId);
        if (!"UPLOADING".equals(session.get("status"))) throw conflict("上传会话当前不可写入分片");
        if (((Timestamp) session.get("expires_at")).toInstant().isBefore(Instant.now())) throw conflict("上传会话已过期");
        int totalParts = ((Number) session.get("total_parts")).intValue();
        if (partNo < 0 || partNo >= totalParts) throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "分片序号超出范围");
        ModelTestStorageService.PartObject part = storageService.storePart(sessionId, sessionId, partNo, chunk);
        jdbc.update("""
                        insert into upload_parts(session_id,part_no,size_bytes,sha256,etag,status,uploaded_at)
                        values (?,?,?,?,?,'UPLOADED',?)
                        on conflict (session_id,part_no) do update set size_bytes=excluded.size_bytes,
                            sha256=excluded.sha256,etag=excluded.etag,status=excluded.status,uploaded_at=excluded.uploaded_at
                        """,
                sessionId, partNo, part.sizeBytes(), part.sha256(), "sha256:" + part.sha256(), Timestamp.from(Instant.now()));
        jdbc.update("update upload_sessions set updated_at=? where id=?", Timestamp.from(Instant.now()), sessionId);
        return Map.of("upload_session_id", sessionId, "part_no", partNo, "size_bytes", part.sizeBytes(), "sha256", part.sha256(), "status", "UPLOADED");
    }

    @Transactional
    public Map<String, Object> completeUpload(String sessionId) {
        Map<String, Object> session = requireRow("select * from upload_sessions where id=? for update", sessionId);
        if ("COMPLETED".equals(session.get("status"))) return uploadSession(sessionId);
        int totalParts = ((Number) session.get("total_parts")).intValue();
        Integer uploaded = jdbc.queryForObject("select count(*) from upload_parts where session_id=? and status='UPLOADED'", Integer.class, sessionId);
        if (uploaded == null || uploaded != totalParts) {
            throw conflict("上传分片不完整，已上传 " + (uploaded == null ? 0 : uploaded) + "/" + totalParts);
        }
        String taskId = String.valueOf(session.get("task_id"));
        String objectKey = "inspection/" + taskId + "/" + sessionId + "/" + safeFileName(String.valueOf(session.get("file_name")));
        ModelTestStorageService.StoredObject stored = storageService.completeParts(
                objectKey,
                sessionId,
                sessionId,
                totalParts,
                String.valueOf(session.get("content_type")),
                session.get("expected_sha256") == null ? null : String.valueOf(session.get("expected_sha256"))
        );
        String resourceId = Ids.next("res");
        Instant now = Instant.now();
        jdbc.update("""
                        insert into inspection_resources(
                            id,task_id,resource_type,storage_url,capture_time,metadata,created_at,
                            upload_session_id,mission_id,sha256,size_bytes,mime_type,quality_status,
                            canonical_crs
                        ) values (?,?,?,?,?,'{}'::jsonb,?,?,?,?,?,?, 'PENDING','EPSG:4326')
                        """,
                resourceId, taskId, String.valueOf(session.get("resource_type")).toLowerCase(Locale.ROOT),
                stored.storageUrl(), Timestamp.from(now), Timestamp.from(now), sessionId, session.get("mission_id"),
                stored.sha256(), stored.sizeBytes(), session.get("content_type"));
        String preprocessJobId = Ids.next("preprocess");
        jdbc.update("insert into preprocess_jobs(id,resource_id,pipeline_version,status,quality_status,created_at) values (?,?,'basic-validation-v1','QUEUED','PENDING',?)",
                preprocessJobId, resourceId, Timestamp.from(now));
        jdbc.update("update upload_sessions set status='COMPLETED',actual_sha256=?,object_key=?,updated_at=? where id=?",
                stored.sha256(), stored.objectKey(), Timestamp.from(now), sessionId);
        jdbc.update("update inspection_tasks set status='data_uploading',updated_at=? where id=?", Timestamp.from(now), taskId);
        publish("inspection.resource.uploaded", Map.of("task_id", taskId, "resource_id", resourceId, "preprocess_job_id", preprocessJobId));
        return Map.ofEntries(
                Map.entry("upload_session_id", sessionId),
                Map.entry("resource_id", resourceId),
                Map.entry("preprocess_job_id", preprocessJobId),
                Map.entry("sha256", stored.sha256()),
                Map.entry("size_bytes", stored.sizeBytes()),
                Map.entry("status", "COMPLETED")
        );
    }

    public List<Map<String, Object>> listPreprocessJobs() {
        return jdbc.queryForList("""
                select p.id as preprocess_job_id,p.resource_id,r.task_id,r.resource_type,r.mime_type,
                       r.size_bytes,p.pipeline_version,p.status,p.quality_status,
                       p.quality_metrics::text as quality_metrics,p.error_message,p.created_at,p.started_at,p.completed_at,
                       (select count(*) from preprocess_step_runs s where s.job_id=p.id) as step_count
                from preprocess_jobs p join inspection_resources r on r.id=p.resource_id
                order by p.created_at desc
                """);
    }

    @Transactional
    public Map<String, Object> runPreprocess(String jobId) {
        Map<String, Object> job = requireRow("""
                select p.*,r.resource_type,r.storage_url,r.sha256,r.size_bytes,r.mime_type
                from preprocess_jobs p join inspection_resources r on r.id=p.resource_id where p.id=? for update
                """, jobId);
        if ("COMPLETED".equals(job.get("status"))) return Map.of("preprocess_job_id", jobId, "status", "COMPLETED", "quality_status", job.get("quality_status"));
        Instant started = Instant.now();
        jdbc.update("update preprocess_jobs set status='RUNNING',started_at=? where id=?", Timestamp.from(started), jobId);
        jdbc.update("delete from preprocess_step_runs where job_id=?", jobId);
        List<Map<String, Object>> steps = List.of(
                Map.of("code", "CHECKSUM_VERIFY", "metrics", Map.of("sha256_present", job.get("sha256") != null)),
                Map.of("code", "MIME_VALIDATE", "metrics", Map.of("mime_type", String.valueOf(job.get("mime_type")))),
                Map.of("code", "METADATA_NORMALIZE", "metrics", Map.of("canonical_crs", "EPSG:4326", "resource_type", String.valueOf(job.get("resource_type"))))
        );
        for (int i = 0; i < steps.size(); i++) {
            Map<String, Object> step = steps.get(i);
            jdbc.update("""
                            insert into preprocess_step_runs(
                                id,job_id,sequence_no,step_code,status,input_ref,output_ref,metrics,started_at,completed_at
                            ) values (?,?,?,?, 'COMPLETED',?,?,?::jsonb,?,?)
                            """,
                    Ids.next("pre-step"), jobId, i + 1, step.get("code"), job.get("storage_url"), job.get("storage_url"),
                    Jsonb.write(step.get("metrics")), Timestamp.from(started), Timestamp.from(Instant.now()));
        }
        Map<String, Object> quality = Map.of(
                "checksum_verified", job.get("sha256") != null,
                "size_bytes", job.get("size_bytes"),
                "mime_type", String.valueOf(job.get("mime_type")),
                "metadata_normalized", true
        );
        Instant completed = Instant.now();
        jdbc.update("update preprocess_jobs set status='COMPLETED',quality_status='PASSED',quality_metrics=?::jsonb,completed_at=? where id=?",
                Jsonb.write(quality), Timestamp.from(completed), jobId);
        jdbc.update("update inspection_resources set quality_status='PASSED' where id=?", job.get("resource_id"));
        publish("inspection.preprocess.completed", Map.of("preprocess_job_id", jobId, "resource_id", job.get("resource_id")));
        return Map.of("preprocess_job_id", jobId, "status", "COMPLETED", "quality_status", "PASSED", "quality_metrics", quality, "steps", steps);
    }

    public Map<String, Object> gisFeatureCollection() {
        List<Map<String, Object>> features = new ArrayList<>();
        for (Map<String, Object> row : jdbc.queryForList("select id,name,object_type,line_id,risk_level,status,ST_AsGeoJSON(geom) as geometry from inspection_objects where status='ACTIVE'")) {
            features.add(feature(row.get("id"), "inspection_object", row.get("geometry"), Map.of(
                    "name", row.get("name"), "object_type", row.get("object_type"), "line_id", row.get("line_id"),
                    "risk_level", row.get("risk_level"), "status", row.get("status")
            )));
        }
        for (Map<String, Object> row : jdbc.queryForList("""
                select r.id,r.name,r.status,rv.version_no,ST_AsGeoJSON(rv.geom) as geometry
                from routes r join route_versions rv on rv.id=r.current_version_id where rv.geom is not null
                """)) {
            features.add(feature(row.get("id"), "route", row.get("geometry"), Map.of(
                    "name", row.get("name"), "status", row.get("status"), "version", row.get("version_no")
            )));
        }
        for (Map<String, Object> row : jdbc.queryForList("select id,model,vendor_code,status,ST_AsGeoJSON(geom) as geometry from uav_devices where geom is not null")) {
            features.add(feature(row.get("id"), "uav_device", row.get("geometry"), Map.of(
                    "name", row.get("model"), "vendor", row.get("vendor_code"), "status", row.get("status")
            )));
        }
        for (Map<String, Object> row : jdbc.queryForList("select mission_id,point_count,ST_AsGeoJSON(geom) as geometry from flight_tracks where geom is not null")) {
            features.add(feature(row.get("mission_id"), "flight_track", row.get("geometry"), Map.of("point_count", row.get("point_count"))));
        }
        for (Map<String, Object> row : jdbc.queryForList("""
                select a.id,a.task_id,a.scene,a.category,a.severity,a.confidence,a.status,
                       a.location::text as location,a.location->'geometry' as geometry,
                       wo.id as workorder_id,wo.status as workorder_status
                from alarms a left join work_orders wo on wo.alarm_id=a.id
                where a.location ? 'geometry'
                """)) {
            Map<String, Object> properties = new LinkedHashMap<>();
            properties.put("alarm_id", row.get("id"));
            properties.put("task_id", row.get("task_id"));
            properties.put("scene", row.get("scene"));
            properties.put("category", row.get("category"));
            properties.put("severity", row.get("severity"));
            properties.put("confidence", row.get("confidence"));
            properties.put("status", row.get("status"));
            properties.put("location", row.get("location"));
            if (row.get("workorder_id") != null) properties.put("workorder_id", row.get("workorder_id"));
            if (row.get("workorder_status") != null) properties.put("workorder_status", row.get("workorder_status"));
            features.add(feature(row.get("id"), "alarm", row.get("geometry"), properties));
        }
        Map<String, Object> collection = new LinkedHashMap<>();
        collection.put("type", "FeatureCollection");
        collection.put("features", features);
        collection.put("canonical_crs", "EPSG:4326");
        collection.put("display_note", "高德底图启用时由前端地图适配器转换为 GCJ-02，主数据保持 WGS84");
        return collection;
    }

    public List<Map<String, Object>> listOrganizations() {
        return jdbc.queryForList("""
                select o.id as org_id, o.parent_id, o.org_type, o.name, o.code, o.status,
                       (select count(*) from platform_users u where u.org_id=o.id) as user_count,
                       (select count(*) from responsibility_rules r where r.owner_org_id=o.id and r.status='ACTIVE') as responsibility_count
                from organizations o order by o.code
                """);
    }

    public List<Map<String, Object>> listResponsibilityRules() {
        return jdbc.queryForList("""
                select r.id as rule_id, r.line_id, r.object_type, r.scene, r.owner_org_id,
                       o.name as owner_org_name, r.assignee_user_id, u.display_name as assignee_name,
                       r.priority, r.status, ST_AsGeoJSON(r.geom) as geometry
                from responsibility_rules r
                join organizations o on o.id=r.owner_org_id
                left join platform_users u on u.id=r.assignee_user_id
                order by r.priority, r.id
                """);
    }

    public List<Map<String, Object>> listNotifications(String status) {
        String normalized = blankToNull(status);
        if (normalized == null) {
            return jdbc.queryForList("""
                    select n.id as notification_id,n.alarm_id,n.channel,n.recipient,n.status,
                           n.attempt_count,n.receipt::text as receipt,n.sent_at,n.delivered_at,n.read_at,
                           n.created_at,a.scene,a.category,a.severity
                    from notification_deliveries n join alarms a on a.id=n.alarm_id
                    order by n.created_at desc limit 200
                    """);
        }
        return jdbc.queryForList("""
                select n.id as notification_id,n.alarm_id,n.channel,n.recipient,n.status,
                       n.attempt_count,n.receipt::text as receipt,n.sent_at,n.delivered_at,n.read_at,
                       n.created_at,a.scene,a.category,a.severity
                from notification_deliveries n join alarms a on a.id=n.alarm_id
                where n.status=? order by n.created_at desc limit 200
                """, normalized.toUpperCase(Locale.ROOT));
    }

    @Transactional
    public Map<String, Object> readNotification(String notificationId) {
        Map<String, Object> notification = requireRow("select id,status from notification_deliveries where id=? for update", notificationId);
        Instant now = Instant.now();
        jdbc.update("update notification_deliveries set status='READ',read_at=coalesce(read_at,?),updated_at=? where id=?",
                Timestamp.from(now), Timestamp.from(now), notificationId);
        return Map.of("notification_id", notification.get("id"), "status", "READ", "read_at", now.toString());
    }

    @Transactional
    public Map<String, Object> actOnWorkOrder(String workorderId, CapabilityRequests.WorkOrderActionRequest request) {
        Map<String, Object> workorder = requireRow("select * from work_orders where id=? for update", workorderId);
        String from = String.valueOf(workorder.get("status"));
        String action = request.action().toUpperCase(Locale.ROOT);
        if (!WorkOrderStateMachine.supports(action)) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "不支持的工单动作: " + action);
        }
        if (!WorkOrderStateMachine.canApply(from, action)) {
            throw conflict("工单状态为 " + from + "，不能执行 " + action);
        }
        String to = WorkOrderStateMachine.target(action);
        Instant now = Instant.now();
        jdbc.update("""
                        update work_orders set status=?, close_result=coalesce(?,close_result),
                            comment=coalesce(?,comment), accepted_at=case when ?='ACCEPT' then ? else accepted_at end,
                            arrived_at=case when ?='ARRIVE' then ? else arrived_at end,
                            submitted_at=case when ?='SUBMIT' then ? else submitted_at end,
                            reviewed_at=case when ? in ('APPROVE','RETURN') then ? else reviewed_at end,
                            version_no=version_no+1, updated_at=? where id=?
                        """,
                to, request.result(), request.comment(), action, Timestamp.from(now), action, Timestamp.from(now),
                action, Timestamp.from(now), action, Timestamp.from(now), Timestamp.from(now), workorderId);
        jdbc.update("""
                        insert into workorder_actions(
                            id,workorder_id,action_type,from_status,to_status,operator_id,operator_name,
                            result,comment,metadata,created_at
                        ) values (?,?,?,?,?,?,?,?,?,?::jsonb,?)
                        """,
                Ids.next("wo-action"), workorderId, action, from, to, request.operatorId(),
                defaultString(request.operatorName(), "现场处置人员"), request.result(), request.comment(),
                Jsonb.write(request.metadata() == null ? Map.of() : request.metadata()), Timestamp.from(now));
        if ("closed".equals(to)) {
            jdbc.update("update alarms set status='closed', updated_at=? where id=?", Timestamp.from(now), workorder.get("alarm_id"));
        }
        publish("workorder.action.completed", Map.of("workorder_id", workorderId, "action", action, "status", to));
        return Map.of("workorder_id", workorderId, "from_status", from, "status", to, "action", action);
    }

    public List<Map<String, Object>> listWorkOrderActions(String workorderId) {
        return jdbc.queryForList("""
                select id as action_id, action_type, from_status, to_status, operator_id,
                       operator_name, result, comment, metadata::text as metadata, created_at
                from workorder_actions where workorder_id=? order by created_at
                """, workorderId);
    }

    @Transactional
    public Map<String, Object> addWorkOrderEvidence(
            String workorderId,
            String phase,
            String uploadedBy,
            String captureTime,
            Double longitude,
            Double latitude,
            MultipartFile file
    ) {
        requireRow("select id from work_orders where id=?", workorderId);
        String normalizedPhase = defaultString(phase, "AFTER").toUpperCase(Locale.ROOT);
        if (!List.of("BEFORE", "DURING", "AFTER", "REVIEW").contains(normalizedPhase)) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "证据阶段仅支持 BEFORE、DURING、AFTER、REVIEW");
        }
        String contentType = defaultString(file.getContentType(), "application/octet-stream");
        if (!(contentType.startsWith("image/") || contentType.startsWith("video/") || contentType.startsWith("audio/") || contentType.equals("application/pdf"))) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "现场证据仅支持图片、视频、音频或 PDF");
        }
        if (file.isEmpty() || file.getSize() > 500L * 1024 * 1024) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "现场证据大小必须在 1 字节到 500 MB 之间");
        }
        String attachmentId = Ids.next("evidence");
        String objectKey = "workorders/" + workorderId + "/" + normalizedPhase.toLowerCase(Locale.ROOT) + "/" + attachmentId + "-" + safeFileName(file.getOriginalFilename());
        ModelTestStorageService.StoredObject stored = storageService.store(objectKey, file);
        Instant now = Instant.now();
        jdbc.update("""
                        insert into workorder_attachments(
                            id,workorder_id,phase,file_name,object_key,storage_url,mime_type,size_bytes,
                            sha256,capture_time,geom,uploaded_by,metadata,created_at
                        ) values (?,?,?,?,?,?,?,?,?,?,
                            case when cast(? as double precision) is null or cast(? as double precision) is null
                                then null
                                else ST_SetSRID(ST_MakePoint(cast(? as double precision),cast(? as double precision)),4326)
                            end,
                            ?,?::jsonb,?)
                        """,
                attachmentId, workorderId, normalizedPhase, safeFileName(file.getOriginalFilename()), stored.objectKey(),
                stored.storageUrl(), contentType, stored.sizeBytes(), stored.sha256(), timestamp(captureTime),
                longitude, latitude, longitude, latitude, defaultString(uploadedBy, "现场处置人员"),
                Jsonb.write(Map.of("local_fallback", stored.localFallback(), "source", "FIELD_UPLOAD")), Timestamp.from(now));
        jdbc.update("insert into workorder_actions(id,workorder_id,action_type,from_status,to_status,operator_name,result,metadata,created_at) select ?,id,'EVIDENCE_UPLOADED',status,status,?,'现场证据上传',?::jsonb,? from work_orders where id=?",
                Ids.next("wo-action"), defaultString(uploadedBy, "现场处置人员"), Jsonb.write(Map.of("attachment_id", attachmentId, "phase", normalizedPhase)), Timestamp.from(now), workorderId);
        publish("workorder.evidence.uploaded", Map.of("workorder_id", workorderId, "attachment_id", attachmentId, "phase", normalizedPhase));
        return Map.ofEntries(
                Map.entry("attachment_id", attachmentId),
                Map.entry("workorder_id", workorderId),
                Map.entry("phase", normalizedPhase),
                Map.entry("file_name", safeFileName(file.getOriginalFilename())),
                Map.entry("mime_type", contentType),
                Map.entry("size_bytes", stored.sizeBytes()),
                Map.entry("sha256", stored.sha256())
        );
    }

    public List<Map<String, Object>> listWorkOrderEvidence(String workorderId) {
        return jdbc.queryForList("""
                select id as attachment_id,workorder_id,phase,file_name,mime_type,size_bytes,sha256,
                       capture_time,ST_AsGeoJSON(geom) as geometry,uploaded_by,metadata::text as metadata,created_at
                from workorder_attachments where workorder_id=? order by created_at
                """, workorderId);
    }

    public AttachmentContent readWorkOrderEvidence(String attachmentId) {
        Map<String, Object> attachment = requireRow("select file_name,object_key,storage_url,mime_type from workorder_attachments where id=?", attachmentId);
        byte[] content = storageService.read(String.valueOf(attachment.get("object_key")), String.valueOf(attachment.get("storage_url")));
        return new AttachmentContent(String.valueOf(attachment.get("file_name")), String.valueOf(attachment.get("mime_type")), content);
    }

    @Transactional
    public Map<String, Object> createReinspection(String workorderId, CapabilityRequests.CreateReinspectionRequest request) {
        Map<String, Object> source = requireRow("""
                select wo.id as workorder_id, a.task_id, a.location::text as location,
                       t.line_id,t.mileage_start,t.mileage_end,t.route_id,t.scene_set::text as scene_set,
                       t.priority,t.external_task_id
                from work_orders wo join alarms a on a.id=wo.alarm_id
                join inspection_tasks t on t.id=a.task_id where wo.id=?
                """, workorderId);
        String taskId = Ids.next("task");
        String requestId = Ids.next("reinspection");
        Instant now = Instant.now();
        String routeId = defaultString(request.routeId(), String.valueOf(source.get("route_id")));
        jdbc.update("""
                        insert into inspection_tasks(
                            id,external_task_id,line_id,mileage_start,mileage_end,route_id,scene_set,
                            priority,status,created_at,updated_at,task_type,approval_status,parent_task_id,created_by
                        ) values (?,?,?,?,?,?,?::jsonb,?,'created',?,?,'SPECIAL','APPROVED',?,?)
                        """,
                taskId, "REINSPECT-" + workorderId, source.get("line_id"), source.get("mileage_start"),
                source.get("mileage_end"), routeId, source.get("scene_set"), source.get("priority"),
                Timestamp.from(now), Timestamp.from(now), source.get("task_id"), defaultString(request.requestedBy(), "专业复核人员"));
        jdbc.update("""
                        insert into reinspection_requests(
                            id,workorder_id,source_task_id,reinspection_task_id,route_id,required_methods,
                            status,requested_by,created_at,updated_at
                        ) values (?,?,?,?,?,?::jsonb,'TASK_CREATED',?,?,?)
                        """,
                requestId, workorderId, source.get("task_id"), taskId, routeId,
                Jsonb.write(request.requiredMethods() == null ? List.of("UAV_REFLIGHT", "AI_COMPARE") : request.requiredMethods()),
                defaultString(request.requestedBy(), "专业复核人员"), Timestamp.from(now), Timestamp.from(now));
        jdbc.update("update work_orders set status='reinspection_required', updated_at=?, version_no=version_no+1 where id=?",
                Timestamp.from(now), workorderId);
        publish("workorder.reinspection.created", Map.of("workorder_id", workorderId, "task_id", taskId));
        return Map.of("reinspection_request_id", requestId, "reinspection_task_id", taskId, "status", "TASK_CREATED");
    }

    public List<Map<String, Object>> listArchives() {
        return jdbc.queryForList("""
                select id as archive_id, business_type, business_id, archive_no,
                       manifest::text as manifest, manifest_url, checksum, retention_until,
                       status, created_by, created_at, updated_at
                from archive_packages order by created_at desc
                """);
    }

    @Transactional
    public Map<String, Object> createArchive(CapabilityRequests.CreateArchiveRequest request) {
        List<Map<String, Object>> existing = jdbc.queryForList("select id as archive_id, archive_no, status from archive_packages where business_type=? and business_id=?",
                request.businessType(), request.businessId());
        if (!existing.isEmpty()) return existing.get(0);
        Map<String, Object> manifest = buildManifest(request.businessType(), request.businessId());
        String manifestJson = Jsonb.write(manifest);
        String id = Ids.next("archive");
        String archiveNo = "RAIL-" + Instant.now().toString().substring(0, 10).replace("-", "") + "-" + id.substring(id.length() - 8).toUpperCase();
        String checksum = sha256(manifestJson);
        Instant now = Instant.now();
        jdbc.update("""
                        insert into archive_packages(
                            id,business_type,business_id,archive_no,manifest,checksum,retention_until,
                            status,created_by,created_at,updated_at
                        ) values (?,?,?,?,?::jsonb,?,current_date + interval '10 years','READY',?,?,?)
                        """,
                id, request.businessType(), request.businessId(), archiveNo, manifestJson, checksum,
                defaultString(request.createdBy(), "专业复核人员"), Timestamp.from(now), Timestamp.from(now));
        publish("archive.package.ready", Map.of("archive_id", id, "business_id", request.businessId()));
        return Map.of("archive_id", id, "archive_no", archiveNo, "checksum", checksum, "status", "READY", "manifest", manifest);
    }

    private Map<String, Object> buildManifest(String businessType, String businessId) {
        Map<String, Object> manifest = new LinkedHashMap<>();
        manifest.put("schema_version", "1.0");
        manifest.put("business_type", businessType);
        manifest.put("business_id", businessId);
        manifest.put("generated_at", Instant.now().toString());
        if ("WORKORDER".equalsIgnoreCase(businessType)) {
            Map<String, Object> workorder = requireRow("select * from work_orders where id=?", businessId);
            manifest.put("workorder", workorder);
            manifest.put("actions", listWorkOrderActions(businessId));
            manifest.put("attachments", jdbc.queryForList("select id,file_name,storage_url,sha256,phase,created_at from workorder_attachments where workorder_id=? order by created_at", businessId));
            manifest.put("reviews", jdbc.queryForList("select id,review_type,decision,reviewer_name,comment,created_at from review_records where workorder_id=? order by created_at", businessId));
            manifest.put("reinspections", jdbc.queryForList("select id,reinspection_task_id,status,created_at from reinspection_requests where workorder_id=? order by created_at", businessId));
        } else if ("TASK".equalsIgnoreCase(businessType)) {
            manifest.put("task", requireRow("select * from inspection_tasks where id=?", businessId));
            manifest.put("missions", jdbc.queryForList("select id,vendor_code,vendor_mission_id,status,progress,created_at,completed_at from flight_missions where task_id=?", businessId));
            manifest.put("resources", jdbc.queryForList("select id,resource_type,storage_url,sha256,size_bytes,created_at from inspection_resources where task_id=?", businessId));
            manifest.put("alarms", jdbc.queryForList("select id,category,severity,status,created_at from alarms where task_id=?", businessId));
        } else {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "仅支持 TASK 或 WORKORDER 归档");
        }
        return manifest;
    }

    private void updateTaskFromMission(String taskId, String missionStatus) {
        String taskStatus = InspectionTaskWorkflow.taskStatusForMission(missionStatus);
        jdbc.update("update inspection_tasks set status=?,updated_at=?,version_no=version_no+1 where id=?",
                taskStatus, Timestamp.from(Instant.now()), taskId);
    }

    private void insertSimulatedTelemetry(String missionId, String deviceId, int progress) {
        double fraction = Math.max(0, Math.min(1, progress / 100.0));
        Instant now = Instant.now();
        jdbc.update("""
                insert into telemetry_points(mission_id,device_id,event_time,received_at,geom,metrics,quality,sequence_no)
                select fm.id,fm.device_id,?,?,
                       ST_Force3D(ST_LineInterpolatePoint(ST_Force2D(rv.geom),?)),
                       ?::jsonb,'GOOD',?
                from flight_missions fm join route_versions rv on rv.id=fm.route_version_id
                where fm.id=? and rv.geom is not null
                """, Timestamp.from(now), Timestamp.from(now), fraction,
                Jsonb.write(Map.of("battery_percent", Math.max(25, 96 - progress / 2), "link_quality", 97, "rtk_status", "FIXED", "progress", progress)),
                progress, missionId);
        Integer points = jdbc.queryForObject("select count(*) from telemetry_points where mission_id=?", Integer.class, missionId);
        if (points != null && points >= 2) {
            jdbc.update("""
                    insert into flight_tracks(mission_id,geom,point_count,quality,start_at,end_at,updated_at)
                    select ?,ST_MakeLine(geom order by event_time),count(*),'{"source":"simulator"}'::jsonb,min(event_time),max(event_time),?
                    from telemetry_points where mission_id=?
                    on conflict (mission_id) do update set geom=excluded.geom,point_count=excluded.point_count,
                        quality=excluded.quality,start_at=excluded.start_at,end_at=excluded.end_at,updated_at=excluded.updated_at
                    """, missionId, Timestamp.from(now), missionId);
        }
    }

    private void upsertDock(String connectionId, Map<String, Object> device, Instant now) {
        String vendorDockId = String.valueOf(device.get("vendor_device_id"));
        jdbc.update("""
                        insert into uav_docks(
                            id,connection_id,vendor_dock_id,model,capabilities,status,geom,last_seen_at,created_at,updated_at
                        ) values (?,?,?,?,?::jsonb,?,ST_SetSRID(ST_MakePoint(?,?),4326),?,?,?)
                        on conflict (connection_id,vendor_dock_id) do update set
                            model=excluded.model,capabilities=excluded.capabilities,status=excluded.status,
                            geom=excluded.geom,last_seen_at=excluded.last_seen_at,updated_at=excluded.updated_at
                        """,
                Ids.next("dock"), connectionId, vendorDockId, device.get("model"),
                Jsonb.write(device.getOrDefault("capabilities", List.of())), device.get("status"),
                device.get("longitude"), device.get("latitude"),
                Timestamp.from(now), Timestamp.from(now), Timestamp.from(now)
        );
    }

    @SuppressWarnings("unchecked")
    private Object mapValue(Object value, String key) {
        if (value instanceof Map<?, ?> map) return map.get(key);
        if (value instanceof String json && !json.isBlank()) return Jsonb.map(json).get(key);
        return null;
    }

    private void insertVendorEvent(Map<String, Object> connection, String eventId, String eventType, Map<String, Object> payload) {
        jdbc.update("""
                        insert into vendor_event_inbox(
                            id,connection_id,vendor_code,vendor_event_id,event_type,event_time,received_at,payload,processed_at
                        ) values (?,?,?,?,?,?,?,?::jsonb,?) on conflict (connection_id,vendor_event_id) do nothing
                        """,
                Ids.next("vendor-event"), connection.get("id"), connection.get("vendor_code"), eventId, eventType,
                Timestamp.from(Instant.now()), Timestamp.from(Instant.now()), Jsonb.write(payload), Timestamp.from(Instant.now()));
    }

    private Map<String, Object> externalMaterialsHealth(String vendor) {
        String details = "DJI".equalsIgnoreCase(vendor)
                ? "缺少 AppKey、AppSecret、License、工作空间、测试设备和回调网络信息"
                : "缺少正式 SDK/API、授权文件、版本矩阵、测试设备和协议文档";
        Map<String, Object> result = new LinkedHashMap<>();
        result.put("status", "MATERIALS_REQUIRED");
        result.put("checked_at", Instant.now().toString());
        result.put("layers", List.of(
                Map.of("layer", "NETWORK", "status", "NOT_CONFIGURED", "message", "未配置厂商服务地址"),
                Map.of("layer", "AUTHENTICATION", "status", "BLOCKED", "message", details),
                Map.of("layer", "CAPABILITIES", "status", "BLOCKED", "message", "取得授权后才能读取真实能力清单"),
                Map.of("layer", "BUSINESS_HANDSHAKE", "status", "BLOCKED", "message", "缺少厂商测试环境和设备")
        ));
        result.put("capabilities", List.of());
        result.put("blocker_reason", details);
        return result;
    }

    private Map<String, Object> feature(Object id, String domainType, Object geometryJson, Map<String, Object> properties) {
        Map<String, Object> feature = new LinkedHashMap<>();
        feature.put("type", "Feature");
        feature.put("id", id);
        feature.put("geometry", Jsonb.read(String.valueOf(geometryJson)));
        Map<String, Object> extended = new LinkedHashMap<>(properties);
        extended.put("domain_type", domainType);
        feature.put("properties", extended);
        return feature;
    }

    private Map<String, Object> requireRow(String sql, Object... args) {
        try {
            return new HashMap<>(jdbc.queryForMap(sql, args));
        } catch (EmptyResultDataAccessException e) {
            throw new ResponseStatusException(HttpStatus.NOT_FOUND, "业务对象不存在");
        }
    }

    private int count(String table) {
        Integer value = jdbc.queryForObject("select count(*) from " + table, Integer.class);
        return value == null ? 0 : value;
    }

    private Timestamp timestamp(String value) {
        if (value == null || value.isBlank()) return null;
        return Timestamp.from(Instant.parse(value));
    }

    private String defaultString(String value, String defaultValue) {
        return value == null || value.isBlank() || "null".equals(value) ? defaultValue : value;
    }

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

    private String safeFileName(String value) {
        String candidate = value == null || value.isBlank() ? "unnamed.bin" : value;
        candidate = candidate.replace('\\', '_').replace('/', '_').replaceAll("[^A-Za-z0-9._\\-\\u4e00-\\u9fa5]", "_");
        return candidate.length() > 180 ? candidate.substring(candidate.length() - 180) : candidate;
    }

    private double number(Map<String, Object> source, String key, double defaultValue) {
        if (source == null || source.get(key) == null) return defaultValue;
        Object value = source.get(key);
        return value instanceof Number n ? n.doubleValue() : Double.parseDouble(String.valueOf(value));
    }

    private Double nullableNumber(Object value) {
        if (value == null) return null;
        return value instanceof Number n ? n.doubleValue() : Double.parseDouble(String.valueOf(value));
    }

    private String sha256(String value) {
        try {
            MessageDigest digest = MessageDigest.getInstance("SHA-256");
            return HexFormat.of().formatHex(digest.digest(value.getBytes(StandardCharsets.UTF_8)));
        } catch (Exception e) {
            throw new IllegalStateException("生成归档摘要失败", e);
        }
    }

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

    private void publish(String eventType, Map<String, Object> payload) {
        Map<String, Object> event = new LinkedHashMap<>();
        event.put("event_id", Ids.next("evt"));
        event.put("event_type", eventType);
        event.put("schema_version", "1.0");
        event.put("occurred_at", Instant.now().truncatedTo(ChronoUnit.MILLIS).toString());
        event.put("source", "rail-inspection-platform");
        event.put("payload", payload);
        try {
            kafkaTemplate.send("platform-events", eventType, event);
        } catch (Exception ignored) {
            // Database changes remain authoritative when Kafka is unavailable in local development.
        }
    }

    public record AttachmentContent(String fileName, String contentType, byte[] content) {
    }
}
