完成了contactcross的初步

qaup-collision/src/main/java/com/qaup/collision/common/model/MovingObject.java

@@ -47,6 +47,11 @@ public class MovingObject {
      */
     private Double altitude;
 
+    /**
+     * Source timestamp from upstream payload, epoch milliseconds.
+     */
+    private Long sourceTimestampMs;
+
     /**
      * 对象类型枚举 - 四种基本分类
      */

qaup-collision/src/main/java/com/qaup/collision/datacollector/dao/DataCollectorDao.java

@@ -121,6 +121,7 @@ public class DataCollectorDao {
                         aircraft.setCurrentSpeed(0.0); 
                         aircraft.setCurrentHeading(0.0);
                         aircraft.setAltitude(0.0); // 假设默认高度为0，如果API有提供可以设置
+                        aircraft.setSourceTimestampMs(normalizeUpstreamTimestampToMs(rawData.getTime()));
                         return aircraft;
                     }).filter(java.util.Objects::nonNull).collect(Collectors.toList());
 
@@ -167,6 +168,7 @@ public class DataCollectorDao {
                         vehicle.setCurrentSpeed(0.0); // 暂时设置为0.0，DataCollectorService会处理
                         vehicle.setCurrentHeading(0.0);
                         vehicle.setAltitude(0.0); // 假设默认高度为0，如果API有提供可以设置
+                        vehicle.setSourceTimestampMs(normalizeUpstreamTimestampToMs(rawData.getTime()));
                         return vehicle;
                     }).filter(java.util.Objects::nonNull).collect(Collectors.toList());
 
@@ -183,6 +185,21 @@ public class DataCollectorDao {
 
     // 内部 DTOs，精确匹配外部 API 接口返回的 JSON 结构
 
+    private Long normalizeUpstreamTimestampToMs(Double rawTime) {
+        if (rawTime == null || rawTime.isNaN() || rawTime.isInfinite()) {
+            return null;
+        }
+
+        double absolute = Math.abs(rawTime);
+        if (absolute < 100_000_000_000d) {
+            return Math.round(rawTime * 1000d);
+        }
+        if (absolute < 100_000_000_000_000d) {
+            return Math.round(rawTime);
+        }
+        return Math.round(rawTime / 1000d);
+    }
+
     @Data
     @NoArgsConstructor
     @AllArgsConstructor
@@ -195,7 +212,7 @@ public class DataCollectorDao {
         @JsonProperty("latitude")
         private Double latitude;
         @JsonProperty("time")
-        private Long time;
+        private Double time;
         // 其他可能存在的但我们不关心的字段将被 @JsonIgnoreProperties 忽略
     }
 
@@ -211,7 +228,7 @@ public class DataCollectorDao {
         @JsonProperty("latitude")
         private Double latitude;
         @JsonProperty("time")
-        private Long time;
+        private Double time;
     }
 
 

qaup-collision/src/main/java/com/qaup/collision/datacollector/service/DataCollectorService.java

@@ -27,6 +27,8 @@ import org.springframework.stereotype.Service;
 import java.time.Instant;
 import java.time.LocalDateTime;
 import java.time.ZoneId;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -423,6 +425,42 @@ public class DataCollectorService {
      * 定时采集航空器数据
      * - 不进行数据持久化
      */
+    private List<Aircraft> deduplicateAircraftByLatestTimestamp(List<Aircraft> aircrafts) {
+        if (aircrafts == null || aircrafts.isEmpty()) {
+            return Collections.emptyList();
+        }
+
+        Map<String, Aircraft> latestByFlightNo = new HashMap<>();
+        for (Aircraft aircraft : aircrafts) {
+            if (aircraft == null || aircraft.getObjectId() == null || aircraft.getObjectId().isBlank()) {
+                continue;
+            }
+
+            Aircraft existing = latestByFlightNo.get(aircraft.getObjectId());
+            if (existing == null || isCandidateNewer(existing, aircraft)) {
+                latestByFlightNo.put(aircraft.getObjectId(), aircraft);
+            }
+        }
+
+        return List.copyOf(latestByFlightNo.values());
+    }
+
+    private boolean isCandidateNewer(Aircraft existing, Aircraft candidate) {
+        Long existingTs = existing.getSourceTimestampMs();
+        Long candidateTs = candidate.getSourceTimestampMs();
+
+        if (existingTs == null && candidateTs == null) {
+            return true;
+        }
+        if (existingTs == null) {
+            return true;
+        }
+        if (candidateTs == null) {
+            return false;
+        }
+        return candidateTs >= existingTs;
+    }
+
     @Scheduled(fixedRateString = "${data.collector.interval}")
     public void collectAircraftData() {
         if (collectorDisabled) {
@@ -444,7 +482,13 @@ public class DataCollectorService {
 
             log.info("采集到 {} 条航空器数据，用于实时处理", newAircrafts.size());
             
-            for (Aircraft aircraft : newAircrafts) {
+            List<Aircraft> deduplicatedAircrafts = deduplicateAircraftByLatestTimestamp(newAircrafts);
+            if (deduplicatedAircrafts.size() != newAircrafts.size()) {
+                log.info("Aircraft batch deduplicated by latest source time: raw={}, deduped={}",
+                        newAircrafts.size(), deduplicatedAircrafts.size());
+            }
+
+            for (Aircraft aircraft : deduplicatedAircrafts) {
                 try {
                     if (aircraft.getCurrentPosition() == null) {
                         log.warn("航空器 {} 位置信息缺失，跳过处理。", aircraft.getObjectId());
@@ -464,6 +508,7 @@ public class DataCollectorService {
                         .currentSpeed(null)    // 不在采集阶段计算速度
                         .currentHeading(null)  // 不在采集阶段计算方向
                         .altitude(aircraft.getAltitude())
+                        .sourceTimestampMs(aircraft.getSourceTimestampMs())
                         .build();
 
                     upsertActiveMovingObject(movingObject);
@@ -479,7 +524,7 @@ public class DataCollectorService {
                 }
             }
 
-            log.info("航空器数据处理和事件发布完成，处理数量: {}", newAircrafts.size());
+            log.info("航空器数据处理和事件发布完成，处理数量: {}", deduplicatedAircrafts.size());
             
         } catch (Exception e) {
             log.error("采集航空器数据异常", e);
@@ -557,6 +602,7 @@ public class DataCollectorService {
                         .currentSpeed(null)    // 不在采集阶段计算速度
                         .currentHeading(null)  // 不在采集阶段计算方向
                         .altitude(vehicle.getAltitude())
+                        .sourceTimestampMs(vehicle.getSourceTimestampMs())
                         .build();
 
                     // 将最新数据更新到缓存（不发送WebSocket消息，统一在周期性检测中发送）
@@ -1295,4 +1341,3 @@ public class DataCollectorService {
         log.info("缓存已清理");
     }
 }
-

qaup-collision/src/main/java/com/qaup/collision/datacollector/websocket/AdxpFlightServiceWebSocketClient.java

@@ -559,9 +559,11 @@ public class AdxpFlightServiceWebSocketClient implements WebSocketHandler {
         // TISFLIGHT 缺 BizKey 时，先用 FlightId 关联 FuId（FuId 视作可复用业务键）
         String fuNorm = normalizeBizKey(fuIdRaw);
         String normalizedFlightId = normalizeFlightNo(flightId);
-        if (fuNorm != null && normalizedFlightId != null) {
+        String normalizedResolvedFlightNo = normalizeFlightNo(resolvedFlightNo);
+        String flightNoForMatch = normalizedFlightId != null ? normalizedFlightId : normalizedResolvedFlightNo;
+        if (fuNorm != null && flightNoForMatch != null) {
             String fuFlightNo = firstSegmentFromBizLike(fuNorm);
-            if (normalizedFlightId.equalsIgnoreCase(fuFlightNo)) {
+            if (flightNoForMatch.equalsIgnoreCase(fuFlightNo)) {
                 return fuNorm;
             }
         }
@@ -1058,6 +1060,9 @@ public class AdxpFlightServiceWebSocketClient implements WebSocketHandler {
         if (flightKey != null || bizRedisKey != null) {
             try {
                 setValueOnFlightAndBizKeys(flightKey, bizRedisKey, "flightNumber", resolvedFlightNo);
+                if (normalizedBizKey != null) {
+                    setValueOnFlightAndBizKeys(flightKey, bizRedisKey, "bizKey", normalizedBizKey);
+                }
                 updateActiveBizKey(flightKey, normalizedBizKey, nowMs);
                 // contactCross 仅对进港有效，避免进出港参数串用
                 if ("IN".equalsIgnoreCase(routeType)

qaup-collision/src/main/java/com/qaup/collision/dataprocessing/service/DataProcessingService.java

@@ -297,13 +297,17 @@ public class DataProcessingService {
                     finalHeading = 0.0; // 默认朝北
                 }
 
+                long payloadTimestamp = movingObject.getSourceTimestampMs() != null
+                        ? movingObject.getSourceTimestampMs()
+                        : currentTime;
+
                 PositionUpdatePayload payload = PositionUpdatePayload.builder()
                         .objectId(movingObject.getObjectId())
                         .objectType(movingObject.getObjectType().name())
                         .position(positionPayload)
                         .heading(finalHeading)
                         .speed(new BigDecimal(finalSpeed).setScale(2, RoundingMode.HALF_UP).doubleValue())
-                        .timestamp(currentTime)
+                        .timestamp(payloadTimestamp)
                         .build();
 
                 // 发布WebSocket事件

qaup-collision/src/main/java/com/qaup/collision/websocket/broadcaster/WebSocketMessageBroadcaster.java

@@ -42,7 +42,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
 @Component
 public class WebSocketMessageBroadcaster {
 
-    private static final int REACQUIRE_AFTER_REJECTS = 3;
     private static final double EARTH_RADIUS_METERS = 6371000.0;
     private static final double MIN_DT_SECONDS = 0.05;
 
@@ -117,6 +116,9 @@ public class WebSocketMessageBroadcaster {
     @Value("${websocket.position.filter.state-ttl-ms:120000}")
     private long positionTrackStateTtlMs;
 
+    @Value("${websocket.position.filter.reacquire-gap-ms:8000}")
+    private long reacquireGapMs;
+
     private record CoalescedPositionUpdate(PositionUpdatePayload payload, long timestamp) {}
 
     private static final class PositionTrackState {
@@ -391,6 +393,7 @@ public class WebSocketMessageBroadcaster {
         }
 
         long payloadTimestamp = payload.getTimestamp() != null ? payload.getTimestamp() : System.currentTimeMillis();
+        long filterTimestampMs = System.currentTimeMillis();
         PositionTrackState state = positionTrackStates.computeIfAbsent(payload.getObjectId(), k -> new PositionTrackState());
 
         double filteredLatitude = latitude;
@@ -398,14 +401,14 @@ public class WebSocketMessageBroadcaster {
 
         synchronized (state) {
             if (!state.initialized) {
-                initializeTrackState(state, latitude, longitude, payloadTimestamp);
+                initializeTrackState(state, latitude, longitude, filterTimestampMs);
             } else {
-                if (payloadTimestamp <= state.lastPayloadTimestampMs) {
+                if (filterTimestampMs <= state.lastPayloadTimestampMs) {
                     state.rejectedCount++;
                     return null;
                 }
 
-                long dtMs = payloadTimestamp - state.lastPayloadTimestampMs;
+                long dtMs = filterTimestampMs - state.lastPayloadTimestampMs;
                 double dtSec = dtMs / 1000.0;
                 double rawDistance = calculateDistanceMeters(
                         state.lastRawLatitude,
@@ -418,17 +421,27 @@ public class WebSocketMessageBroadcaster {
 
                 double maxAllowedJump = resolveMaxAllowedJumpMeters(adaptive.maxSpeedMps(), adaptive.jumpMarginMeter(), dtSec);
                 boolean isOutlierJump = rawDistance > maxAllowedJump;
-                if (isOutlierJump && state.rejectedCount < REACQUIRE_AFTER_REJECTS - 1) {
-                    state.rejectedCount++;
-                    log.debug("Drop outlier position update: objectId={}, jumpMeter={}, allowedMeter={}",
-                            payload.getObjectId(), rawDistance, maxAllowedJump);
-                    return null;
-                }
 
                 if (isOutlierJump) {
-                    state.rejectedCount = 0;
+                    boolean longGapReacquire = dtMs >= Math.max(reacquireGapMs, 0L);
-                    filteredLatitude = latitude;
+                    if (longGapReacquire) {
-                    filteredLongitude = longitude;
+                        state.rejectedCount = 0;
+                        filteredLatitude = latitude;
+                        filteredLongitude = longitude;
+                    } else {
+                        state.rejectedCount++;
+                        double[] clipped = clipPointByMaxDistance(
+                                state.lastFilteredLatitude,
+                                state.lastFilteredLongitude,
+                                latitude,
+                                longitude,
+                                maxAllowedJump
+                        );
+                        filteredLatitude = clipped[0];
+                        filteredLongitude = clipped[1];
+                        log.debug("Clip outlier position update: objectId={}, jumpMeter={}, allowedMeter={}, dtMs={}",
+                                payload.getObjectId(), rawDistance, maxAllowedJump, dtMs);
+                    }
                 } else {
                     state.rejectedCount = 0;
                     if (rawDistance <= adaptive.jitterMeter()) {
@@ -443,7 +456,7 @@ public class WebSocketMessageBroadcaster {
 
                 state.lastRawLatitude = latitude;
                 state.lastRawLongitude = longitude;
-                state.lastPayloadTimestampMs = payloadTimestamp;
+                state.lastPayloadTimestampMs = filterTimestampMs;
                 state.lastFilteredLatitude = filteredLatitude;
                 state.lastFilteredLongitude = filteredLongitude;
                 state.lastAcceptedAtMs = System.currentTimeMillis();
@@ -556,4 +569,21 @@ public class WebSocketMessageBroadcaster {
         }
         return Math.min(value, max);
     }
+
+    private double[] clipPointByMaxDistance(
+            double startLatitude,
+            double startLongitude,
+            double targetLatitude,
+            double targetLongitude,
+            double maxDistanceMeters) {
+        double distanceMeters = calculateDistanceMeters(startLatitude, startLongitude, targetLatitude, targetLongitude);
+        if (distanceMeters <= maxDistanceMeters || distanceMeters <= 0.0 || maxDistanceMeters <= 0.0) {
+            return new double[] {targetLatitude, targetLongitude};
+        }
+
+        double ratio = maxDistanceMeters / distanceMeters;
+        double clippedLatitude = startLatitude + (targetLatitude - startLatitude) * ratio;
+        double clippedLongitude = startLongitude + (targetLongitude - startLongitude) * ratio;
+        return new double[] {clippedLatitude, clippedLongitude};
+    }
 }

qaup-collision/src/main/java/com/qaup/collision/websocket/event/PositionUpdateEvent.java

@@ -3,37 +3,39 @@ package com.qaup.collision.websocket.event;
 import com.qaup.collision.common.model.spatial.VehicleLocation;
 import com.qaup.collision.websocket.message.MessageTypeConstants;
 import com.qaup.collision.websocket.message.PositionUpdatePayload;
-
-import lombok.Getter;
-import lombok.Data;
-import lombok.Builder;
-import lombok.NoArgsConstructor;
 import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
 
 import java.util.UUID;
 
 /**
- * 位置更新事件
+ * Position update event.
- * 当车辆或航空器位置数据处理完成后发布此事件
+ *
+ * Outer event timestamp:
+ * - {@link #timestamp} is generated at publish time in epoch microseconds for envelope ordering.
+ *
+ * Payload timestamp:
+ * - {@link PositionUpdatePayload#getTimestamp()} is the sample time from data pipeline in epoch milliseconds.
  */
 @Getter
 public class PositionUpdateEvent implements WebSocketEvent {
 
     private final String eventId;
     private final long timestamp;
-    private final Object payload; // 支持VehicleLocation和PositionUpdatePayload两种类型
+    private final Object payload;
 
-    // 支持VehicleLocation构造函数（用于无人车数据）
     public PositionUpdateEvent(VehicleLocation vehicleLocation) {
         this.eventId = UUID.randomUUID().toString();
-        this.timestamp = java.time.Instant.now().toEpochMilli() * 1000; // 微秒级绝对时间戳
+        this.timestamp = java.time.Instant.now().toEpochMilli() * 1000;
         this.payload = vehicleLocation;
     }
 
-    // 支持PositionUpdatePayload构造函数（用于航空器和机场车辆数据）
     public PositionUpdateEvent(PositionUpdatePayload positionPayload) {
         this.eventId = UUID.randomUUID().toString();
-        this.timestamp = java.time.Instant.now().toEpochMilli() * 1000; // 微秒级绝对时间戳
+        this.timestamp = java.time.Instant.now().toEpochMilli() * 1000;
         this.payload = positionPayload;
     }
 
@@ -47,32 +49,13 @@ public class PositionUpdateEvent implements WebSocketEvent {
         return payload;
     }
 
-    /**
-     * 获取车辆ID（索引性质）
-     * 
-     * 按照API设计原则：
-     * - 对于无人车数据：返回数据库主键vehicleId
-     * - 对于航空器/机场车辆：返回null（因为它们没有数据库记录）
-     * 
-     * @return 车辆ID（Long类型），不适用时返回null
-     */
     public Long getVehicleId() {
         if (payload instanceof VehicleLocation) {
             return ((VehicleLocation) payload).getVehicleId();
         }
-        // 航空器和机场车辆没有数据库记录，返回null
         return null;
     }
 
-    /**
-     * 获取业务标识符（详细信息）
-     * 
-     * 按照API设计原则：
-     * - 对于无人车数据：返回车牌号
-     * - 对于航空器/机场车辆：返回objectId（可能是车牌号、航班号等）
-     * 
-     * @return 业务标识符（String类型）
-     */
     public String getLicensePlate() {
         if (payload instanceof VehicleLocation) {
             return ((VehicleLocation) payload).getLicensePlate();
@@ -82,11 +65,6 @@ public class PositionUpdateEvent implements WebSocketEvent {
         return null;
     }
 
-    /**
-     * 获取完整的标识符信息
-     * 
-     * @return 包含vehicleId和businessId的完整信息
-     */
     public VehicleIdentifier getVehicleIdentifier() {
         if (payload instanceof VehicleLocation) {
             VehicleLocation location = (VehicleLocation) payload;
@@ -98,7 +76,7 @@ public class PositionUpdateEvent implements WebSocketEvent {
         } else if (payload instanceof PositionUpdatePayload) {
             PositionUpdatePayload positionPayload = (PositionUpdatePayload) payload;
             return VehicleIdentifier.builder()
-                    .vehicleId(null) // 航空器没有数据库记录
+                    .vehicleId(null)
                     .businessId(positionPayload.getObjectId())
                     .vehicleType(positionPayload.getObjectType())
                     .build();
@@ -106,9 +84,6 @@ public class PositionUpdateEvent implements WebSocketEvent {
         return null;
     }
 
-    /**
-     * 获取车辆类型
-     */
     public String getVehicleType() {
         if (payload instanceof VehicleLocation) {
             return ((VehicleLocation) payload).getVehicleType().name();
@@ -118,16 +93,13 @@ public class PositionUpdateEvent implements WebSocketEvent {
         return null;
     }
 
-    /**
-     * 车辆标识符信息类
-     */
     @Data
     @Builder
     @NoArgsConstructor
     @AllArgsConstructor
     public static class VehicleIdentifier {
-        private Long vehicleId;      // 数据库主键（索引性质）
+        private Long vehicleId;
-        private String businessId;   // 业务标识符（车牌号、航班号等）
+        private String businessId;
-        private String vehicleType;  // 车辆类型
+        private String vehicleType;
     }
 }

qaup-collision/src/main/java/com/qaup/collision/websocket/message/PositionUpdatePayload.java

@@ -1,14 +1,13 @@
 package com.qaup.collision.websocket.message;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
-import lombok.Data;
-import lombok.Builder;
-import lombok.NoArgsConstructor;
 import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
 
 /**
- * 位置更新消息负载
+ * Position update payload.
- * 对应前端JSON格式: position_update
  */
 @Data
 @Builder
@@ -17,43 +16,43 @@ import lombok.AllArgsConstructor;
 public class PositionUpdatePayload {
 
     /**
-     * 对象ID（车辆ID、航班号等）
+     * Object ID, such as vehicle ID or flight number.
      */
     @JsonProperty("object_id")
     private String objectId;
 
     /**
-     * 对象类型（基于MovingObjectType枚举）
+     * Object type mapped from MovingObjectType.
      */
     @JsonProperty("object_type")
     private String objectType;
 
     /**
-     * 位置信息
+     * Current position.
      */
     @JsonProperty("position")
     private Position position;
 
     /**
-     * 航向角（度）
+     * Heading in degrees.
      */
     @JsonProperty("heading")
     private Double heading;
 
     /**
-     * 速度（米/秒）
+     * Speed value from backend (current behavior is km/h).
      */
     @JsonProperty("speed")
     private Double speed;
 
     /**
-     * 时间戳（微秒级）
+     * Position sample timestamp in epoch milliseconds.
      */
     @JsonProperty("timestamp")
     private Long timestamp;
 
     /**
-     * 位置坐标类
+     * Coordinate model.
      */
     @Data
     @Builder
@@ -62,13 +61,13 @@ public class PositionUpdatePayload {
     public static class Position {
 
         /**
-         * 纬度
+         * Latitude.
          */
         @JsonProperty("latitude")
         private Double latitude;
 
         /**
-         * 经度
+         * Longitude.
          */
         @JsonProperty("longitude")
         private Double longitude;