package com.ai.trackwalker.capability.uav;

import com.ai.trackwalker.common.Jsonb;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.sql.Timestamp;
import java.time.Instant;
import java.time.format.DateTimeParseException;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

@Component
public class UavAccessEventProjector {
    private static final Set<String> TERMINAL = Set.of("COMPLETED", "CANCELLED", "ABORTED", "FAILED");

    private final JdbcTemplate jdbc;

    public UavAccessEventProjector(JdbcTemplate jdbc) {
        this.jdbc = jdbc;
    }

    @KafkaListener(
            topics = "${rail.uav-access.event-topic:uav.access.events.v1}",
            groupId = "${rail.uav-access.event-consumer-group:rail-platform-uav-access-v1}"
    )
    @Transactional
    public void consume(Map<String, Object> envelope) {
        String eventId = text(envelope.get("event_id"));
        String eventType = text(envelope.get("event_type"));
        if (eventId.isBlank() || eventType.isBlank()) return;
        insertInbox(eventId, eventType, envelope);
        List<Map<String, Object>> inbox = jdbc.queryForList(
                "select processed_at from uav_access_event_inbox where event_id=?",
                eventId
        );
        if (!inbox.isEmpty() && inbox.get(0).get("processed_at") != null) return;
        try {
            Map<String, Object> payload = map(envelope.get("payload"));
            switch (eventType) {
                case "uav.access.mission.status.changed.v1" -> projectMission(payload);
                case "uav.access.telemetry.received.v1" -> projectTelemetry(payload);
                default -> {
                    // Raw vendor events stay auditable in the Inbox until a typed projector is added.
                }
            }
            jdbc.update(
                    "update uav_access_event_inbox set processed_at=?,processing_error=null where event_id=?",
                    Timestamp.from(Instant.now()),
                    eventId
            );
        } catch (RuntimeException exception) {
            jdbc.update(
                    "update uav_access_event_inbox set processing_error=? where event_id=?",
                    bounded(exception.getClass().getSimpleName() + ": " + exception.getMessage()),
                    eventId
            );
            throw exception;
        }
    }

    private void insertInbox(String eventId, String eventType, Map<String, Object> envelope) {
        try {
            jdbc.update("""
                            insert into uav_access_event_inbox(
                                event_id,event_type,aggregate_id,occurred_at,received_at,payload
                            ) values (?,?,?,?,?,?::jsonb)
                            """,
                    eventId,
                    eventType,
                    envelope.get("aggregate_id"),
                    timestamp(text(envelope.get("occurred_at"))),
                    Timestamp.from(Instant.now()),
                    Jsonb.write(envelope)
            );
        } catch (DuplicateKeyException ignored) {
            // A prior failed delivery remains unprocessed and is retried below.
        }
    }

    private void projectMission(Map<String, Object> payload) {
        String missionId = text(payload.get("mission_id"));
        if (missionId.isBlank()) return;
        List<Map<String, Object>> rows = jdbc.queryForList(
                "select * from flight_missions where id=? for update",
                missionId
        );
        if (rows.isEmpty()) return;
        Map<String, Object> mission = rows.get(0);
        String current = text(mission.get("status"));
        String incoming = text(payload.get("status")).toUpperCase();
        if (incoming.isBlank()) incoming = "UNKNOWN";
        if (TERMINAL.contains(current) && !current.equals(incoming)) return;
        if (isRegression(current, incoming)) return;
        int progress = integer(payload.get("progress"), integer(mission.get("progress"), 0));
        if ("COMPLETED".equals(incoming)) progress = 100;
        Instant now = Instant.now();
        Timestamp startedAt = List.of("FLYING", "PAUSED", "RETURNING", "UPLOADING", "COMPLETED")
                .contains(incoming) ? Timestamp.from(now) : null;
        Timestamp completedAt = TERMINAL.contains(incoming) ? Timestamp.from(now) : null;
        jdbc.update("""
                        update flight_missions set
                            status=?,progress=?,current_waypoint=?,vendor_status=?,
                            sync_status='SYNCHRONIZED',last_reconciled_at=?,
                            started_at=coalesce(started_at,?),completed_at=coalesce(completed_at,?),
                            error_code=?,error_message=?,version_no=version_no+1,updated_at=?
                        where id=?
                        """,
                incoming,
                Math.max(0, Math.min(100, progress)),
                nullableInteger(payload.get("current_waypoint")),
                text(payload.get("vendor_status")),
                Timestamp.from(now),
                startedAt,
                completedAt,
                errorCode(payload),
                errorMessage(payload),
                Timestamp.from(now),
                missionId
        );
        String taskStatus = InspectionTaskWorkflow.taskStatusForMission(incoming);
        jdbc.update(
                "update inspection_tasks set status=?,updated_at=?,version_no=version_no+1 where id=?",
                taskStatus,
                Timestamp.from(now),
                mission.get("task_id")
        );
        completeCommands(missionId, incoming, now);
    }

    private void completeCommands(String missionId, String missionStatus, Instant now) {
        List<String> completedCommands = switch (missionStatus) {
            case "PREPARING" -> List.of("MISSION_DISPATCH", "MISSION_START");
            case "FLYING" -> List.of("MISSION_DISPATCH", "MISSION_START", "MISSION_RESUME");
            case "PAUSED" -> List.of("MISSION_PAUSE");
            case "RETURNING" -> List.of("RETURN_HOME");
            case "CANCELLED", "ABORTED" -> List.of("MISSION_CANCEL", "RETURN_HOME");
            case "COMPLETED" -> List.of("MISSION_DISPATCH", "MISSION_START", "MISSION_RESUME", "RETURN_HOME");
            default -> List.of();
        };
        if (completedCommands.isEmpty()) return;
        String placeholders = String.join(",", java.util.Collections.nCopies(completedCommands.size(), "?"));
        java.util.ArrayList<Object> args = new java.util.ArrayList<>();
        args.add(Timestamp.from(now));
        args.add(Timestamp.from(now));
        args.add(missionId);
        args.addAll(completedCommands);
        jdbc.update(
                "update uav_commands set status='SUCCEEDED',accepted_at=coalesce(accepted_at,?),completed_at=? "
                        + "where mission_id=? and status='ACCEPTED' and command_type in (" + placeholders + ")",
                args.toArray()
        );
    }

    private void projectTelemetry(Map<String, Object> payload) {
        String connectionId = text(payload.get("connection_id"));
        String deviceSn = text(payload.get("device_sn"));
        if (deviceSn.isBlank()) return;
        List<Map<String, Object>> devices = connectionId.isBlank()
                ? jdbc.queryForList(
                        "select * from uav_devices where vendor_device_id=? order by updated_at desc limit 1",
                        deviceSn
                )
                : jdbc.queryForList(
                        "select * from uav_devices where connection_id=? and vendor_device_id=?",
                        connectionId,
                        deviceSn
                );
        if (devices.isEmpty()) return;
        Map<String, Object> device = devices.get(0);
        String missionId = text(payload.get("mission_id"));
        if (missionId.isBlank()) {
            String vendorTaskId = text(payload.get("vendor_task_uuid"));
            List<String> missions = vendorTaskId.isBlank()
                    ? List.of()
                    : jdbc.queryForList(
                            "select id from flight_missions where vendor_mission_id=? order by created_at desc limit 1",
                            String.class,
                            vendorTaskId
                    );
            missionId = missions.isEmpty() ? null : missions.get(0);
        }
        Double longitude = decimal(payload.get("longitude"));
        Double latitude = decimal(payload.get("latitude"));
        if (longitude == null || latitude == null) return;
        Double altitude = decimal(payload.get("altitude_m"));
        if (altitude == null) altitude = 0.0;
        Instant eventTime = instant(text(payload.get("event_time")));
        Map<String, Object> metrics = map(payload.get("metrics"));
        long sequence = longValue(metrics.getOrDefault("sequence_no", eventTime.toEpochMilli()));
        jdbc.update("""
                        insert into telemetry_points(
                            mission_id,device_id,event_time,received_at,geom,metrics,quality,sequence_no
                        ) values (?,?,?,?,
                            ST_SetSRID(ST_MakePoint(?,?,?),4326),?::jsonb,'GOOD',?)
                        """,
                missionId,
                device.get("id"),
                Timestamp.from(eventTime),
                Timestamp.from(Instant.now()),
                longitude,
                latitude,
                altitude,
                Jsonb.write(metrics),
                sequence
        );
        Map<String, Object> health = new LinkedHashMap<>();
        health.putAll(map(device.get("health_detail")));
        copy(metrics, health, "battery_capacity_percent", "battery_percent");
        copy(metrics, health, "battery_percent", "battery_percent");
        copy(metrics, health, "link_quality", "link_quality");
        copy(metrics, health, "rtk_status", "rtk");
        jdbc.update("""
                        update uav_devices set geom=ST_SetSRID(ST_MakePoint(?,?),4326),
                            health_detail=?::jsonb,status='ONLINE',last_seen_at=?,source_updated_at=?,updated_at=?
                        where id=?
                        """,
                longitude, latitude, Jsonb.write(health),
                Timestamp.from(eventTime), Timestamp.from(eventTime), Timestamp.from(Instant.now()), device.get("id")
        );
        if (missionId != null) updateTrack(missionId);
    }

    private void updateTrack(String missionId) {
        Integer points = jdbc.queryForObject(
                "select count(*) from telemetry_points where mission_id=?",
                Integer.class,
                missionId
        );
        if (points == null || points < 2) return;
        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":"flighthub2-mqtt-eventapi"}'::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(Instant.now()), missionId);
    }

    private boolean isRegression(String current, String incoming) {
        return rank(incoming) < rank(current) && !"UNKNOWN".equals(current);
    }

    private int rank(String status) {
        return switch (status) {
            case "READY" -> 0;
            case "DISPATCHING" -> 1;
            case "DISPATCHED" -> 2;
            case "PREPARING" -> 3;
            case "FLYING", "PAUSED", "RETURNING" -> 4;
            case "UPLOADING" -> 5;
            case "COMPLETED", "CANCELLED", "ABORTED", "FAILED" -> 6;
            default -> -1;
        };
    }

    private String errorCode(Map<String, Object> payload) {
        Object exceptions = payload.get("exceptions");
        if (exceptions instanceof List<?> list && !list.isEmpty() && list.get(0) instanceof Map<?, ?> first) {
            return text(first.get("code"));
        }
        return null;
    }

    private String errorMessage(Map<String, Object> payload) {
        Object exceptions = payload.get("exceptions");
        if (exceptions instanceof List<?> list && !list.isEmpty() && list.get(0) instanceof Map<?, ?> first) {
            return text(first.get("message"));
        }
        return null;
    }

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

    private void copy(Map<String, Object> source, Map<String, Object> target, String from, String to) {
        if (source.get(from) != null) target.put(to, source.get(from));
    }

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

    private Instant instant(String value) {
        if (value == null || value.isBlank()) return Instant.now();
        try {
            return Instant.parse(value);
        } catch (DateTimeParseException ignored) {
            try {
                long epoch = Long.parseLong(value);
                if (epoch < 10_000_000_000L) epoch *= 1000;
                return Instant.ofEpochMilli(epoch);
            } catch (NumberFormatException ignoredAgain) {
                return Instant.now();
            }
        }
    }

    private Double decimal(Object value) {
        if (value instanceof Number number) return number.doubleValue();
        try {
            return value == null ? null : Double.parseDouble(String.valueOf(value));
        } catch (NumberFormatException ignored) {
            return null;
        }
    }

    private int integer(Object value, int fallback) {
        if (value instanceof Number number) return number.intValue();
        try {
            return Integer.parseInt(text(value));
        } catch (NumberFormatException ignored) {
            return fallback;
        }
    }

    private Integer nullableInteger(Object value) {
        if (value == null) return null;
        return integer(value, 0);
    }

    private long longValue(Object value) {
        if (value instanceof Number number) return number.longValue();
        try {
            return Long.parseLong(text(value));
        } catch (NumberFormatException ignored) {
            return Instant.now().toEpochMilli();
        }
    }

    private String text(Object value) {
        return value == null ? "" : String.valueOf(value);
    }

    private String bounded(String value) {
        if (value == null) return "unknown";
        return value.substring(0, Math.min(2000, value.length()));
    }
}
