实现采集器断线恢复与来源异常持久化

yeg

2026-08-30 ec3308617909c8d6280e0d00ffb18350e372fbdf

实现采集器断线恢复与来源异常持久化

 ruoyi-admin/src/main/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizer.java

@@ -10,6 +10,7 @@
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore;
import com.ruoyi.device.discovery.DeviceDiscoveryCandidateStore.CollectorCandidate;
import com.ruoyi.system.mapper.TCollectorMapper;
import com.ruoyi.system.service.ICollectorDisconnectRecoveryService;

/** 将模拟采集器的发现写入 Redis，并同步已登记采集器的实时在线事实。 */
@Component
@@ -18,14 +19,24 @@
    private final BtsDevelopmentSimulator simulator;
    private final DeviceDiscoveryCandidateStore candidateStore;
    private final TCollectorMapper collectorMapper;
    private final ICollectorDisconnectRecoveryService disconnectRecovery;

    @Autowired
    public DevelopmentCollectorStatusSynchronizer(BtsDevelopmentSimulator simulator,
            DeviceDiscoveryCandidateStore candidateStore, TCollectorMapper collectorMapper)
            DeviceDiscoveryCandidateStore candidateStore, TCollectorMapper collectorMapper,
            ICollectorDisconnectRecoveryService disconnectRecovery)
    {
        this.simulator = simulator;
        this.candidateStore = candidateStore;
        this.collectorMapper = collectorMapper;
        this.disconnectRecovery = disconnectRecovery;
    }

    /** 保留单元测试及独立模拟器装配使用的旧构造入口。 */
    public DevelopmentCollectorStatusSynchronizer(BtsDevelopmentSimulator simulator,
            DeviceDiscoveryCandidateStore candidateStore, TCollectorMapper collectorMapper)
    {
        this(simulator, candidateStore, collectorMapper, (codes, at) -> { });
    }

    /** 完整替换 dev-sim 采集器快照，删除配置后候选立即消失。 */
@@ -33,7 +44,8 @@
    public synchronized void synchronizeOnlineStatus()
    {
        Instant observedAt = Instant.now();
        List<CollectorCandidate> candidates = simulator.snapshot().collectors().stream()
        var simulatorState = simulator.snapshot();
        List<CollectorCandidate> candidates = simulatorState.collectors().stream()
                .map(item -> new CollectorCandidate(BtsDevelopmentSimulator.ADAPTER_CODE, item.deviceCode(),
                        item.ip(), item.mac(), item.name(),
                        item.manufacturer(), item.model(), item.channelCount(), item.online(), observedAt,
@@ -42,5 +54,7 @@
        candidateStore.replaceCollectors(BtsDevelopmentSimulator.ADAPTER_CODE, candidates);
        candidates.forEach(candidate -> collectorMapper.updateRegisteredObservation(candidate.deviceCode(),
                candidate.online() ? "ONLINE" : "OFFLINE"));
        disconnectRecovery.reconcile(simulatorState.collectors().stream()
                .filter(item -> item.online()).map(item -> item.deviceCode()).collect(java.util.stream.Collectors.toSet()), observedAt);
    }
}

 ruoyi-admin/src/main/resources/db/migration/V18__collector_disconnect_recovery.sql

New file
@@ -0,0 +1,26 @@
-- Ticket11：持久化采集器来源级断线、缺失范围和恢复边界。
CREATE TABLE t_test_collector_exception_interval (
    interval_id BIGINT NOT NULL AUTO_INCREMENT,
    test_id BIGINT NOT NULL,
    battery_relation_id BIGINT NOT NULL,
    collector_id BIGINT NOT NULL,
    collector_channel_code VARCHAR(64) NOT NULL,
    exception_type VARCHAR(64) NOT NULL DEFAULT 'COLLECTOR_DISCONNECTED',
    opened_at_utc DATETIME(3) NOT NULL,
    missing_from_utc DATETIME(3) NOT NULL,
    recovered_at_utc DATETIME(3) NULL,
    missing_to_utc DATETIME(3) NULL,
    interval_status VARCHAR(16) NOT NULL DEFAULT 'OPEN',
    detail VARCHAR(1000) NULL,
    PRIMARY KEY (interval_id),
    KEY idx_collector_exception_test (test_id, interval_status, opened_at_utc),
    KEY idx_collector_exception_source (collector_id, collector_channel_code, interval_status),
    CONSTRAINT fk_collector_exception_test FOREIGN KEY (test_id) REFERENCES t_test (test_id),
    CONSTRAINT fk_collector_exception_battery FOREIGN KEY (battery_relation_id) REFERENCES t_test_battery (relation_id),
    CONSTRAINT fk_collector_exception_collector FOREIGN KEY (collector_id) REFERENCES t_collector (collector_id),
    CONSTRAINT chk_collector_exception_status CHECK (interval_status IN ('OPEN', 'CLOSED')),
    CONSTRAINT chk_collector_exception_range CHECK (missing_to_utc IS NULL OR missing_to_utc >= missing_from_utc)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='采集器来源级断线缺失区间';

ALTER TABLE t_test_collector_exception_interval
    ADD UNIQUE KEY uk_collector_exception_open (test_id, battery_relation_id, collector_id, collector_channel_code, exception_type, interval_status);

 ruoyi-admin/src/test/java/com/ruoyi/config/DevelopmentCollectorStatusSynchronizerTest.java

@@ -16,9 +16,10 @@
    void productionConstructorIsExplicitlyAutowired()
    {
        var constructors = DevelopmentCollectorStatusSynchronizer.class.getDeclaredConstructors();
        assertEquals(1, constructors.length);
        assertTrue(constructors[0].isAnnotationPresent(Autowired.class));
        assertEquals(3, constructors[0].getParameterCount());
        assertEquals(2, constructors.length);
        var production = java.util.Arrays.stream(constructors)
                .filter(item -> item.isAnnotationPresent(Autowired.class)).findFirst().orElseThrow();
        assertEquals(4, production.getParameterCount());
    }
    @Test
    void synchronizesOnlyConfiguredCollectorsWithoutRegisteringThem() throws Exception

 ruoyi-admin/src/test/java/com/ruoyi/config/FlywayMigrationContractTest.java

@@ -128,6 +128,13 @@
        assertTrue(v17.contains("AFTER ended_at_utc"));
        assertFalse(v17.matches("(?s).*\\bDROP\\s+TABLE\\b.*"));

        String v18 = resource("db/migration/V18__collector_disconnect_recovery.sql");
        assertTrue(v18.contains("CREATE TABLE t_test_collector_exception_interval"));
        assertTrue(v18.contains("missing_from_utc"));
        assertTrue(v18.contains("recovered_at_utc"));
        assertTrue(v18.contains("uk_collector_exception_open"));
        assertFalse(v18.matches("(?s).*\\bDROP\\s+TABLE\\b.*"));

        String testMapper = resource("mapper/system/TTestMapper.xml");
        assertTrue(testMapper.contains("selectReceivingChannelsForAdapter"));
        assertTrue(testMapper.contains("RECEPTION_RESUMED"));

 ruoyi-admin/src/test/java/com/ruoyi/system/service/impl/CollectorDisconnectRecoveryServiceImplTest.java

New file
@@ -0,0 +1,67 @@
package com.ruoyi.system.service.impl;

import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import java.time.Instant;
import java.util.List;
import java.util.Set;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import com.ruoyi.system.domain.TTestCollectorExceptionInterval;
import com.ruoyi.system.domain.TTestCollectorSource;
import com.ruoyi.system.mapper.TTestMapper;

/** 采集器断线对账的来源级幂等与恢复边界测试。 */
@ExtendWith(MockitoExtension.class)
class CollectorDisconnectRecoveryServiceImplTest
{
    @Mock private TTestMapper mapper;
    @InjectMocks private CollectorDisconnectRecoveryServiceImpl service;

    @Test
    void opensOnlyOfflineSourceAndKeepsOtherSourcesIndependent()
    {
        TTestCollectorSource offline = source(10L, 20L, 8L, "COL-1", "S-01");
        TTestCollectorSource online = source(10L, 20L, 9L, "COL-2", "S-01");
        when(mapper.selectReceivingCollectorSources()).thenReturn(List.of(offline, online));
        when(mapper.selectOpenCollectorException(10L, 20L, 8L, "S-01")).thenReturn(null);
        when(mapper.selectOpenCollectorException(10L, 20L, 9L, "S-01")).thenReturn(null);

        service.reconcile(Set.of("COL-2"), Instant.parse("2026-08-30T09:00:00Z"));

        verify(mapper).insertCollectorExceptionInterval(any(TTestCollectorExceptionInterval.class));
        verify(mapper, never()).closeCollectorExceptionInterval(any(), any(), any());
    }

    @Test
    void closesOpenIntervalAtRecoveryBoundary()
    {
        TTestCollectorSource source = source(10L, 20L, 8L, "COL-1", "S-01");
        TTestCollectorExceptionInterval open = new TTestCollectorExceptionInterval();
        open.setIntervalId(99L);
        open.setIntervalStatus("OPEN");
        when(mapper.selectReceivingCollectorSources()).thenReturn(List.of(source));
        when(mapper.selectOpenCollectorException(10L, 20L, 8L, "S-01")).thenReturn(open);

        service.reconcile(Set.of("COL-1"), Instant.parse("2026-08-30T09:01:00Z"));

        verify(mapper).closeCollectorExceptionInterval(eq(99L), any(), any());
        verify(mapper, never()).insertCollectorExceptionInterval(any());
    }

    private TTestCollectorSource source(Long testId, Long relationId, Long collectorId,
            String collectorCode, String channelCode)
    {
        TTestCollectorSource source = new TTestCollectorSource();
        source.setTestId(testId); source.setBatteryRelationId(relationId); source.setCollectorId(collectorId);
        source.setCollectorCode(collectorCode); source.setBatteryNo("BAT-01"); source.setChannelCode(channelCode);
        return source;
    }
}

 ruoyi-system/src/main/java/com/ruoyi/system/domain/TTestCollectorExceptionInterval.java

New file
@@ -0,0 +1,54 @@
package com.ruoyi.system.domain;

import java.util.Date;

/** 采集器来源断线期间的只追加事实。关闭动作只补写恢复边界。 */
public class TTestCollectorExceptionInterval
{
    private Long intervalId;
    private Long testId;
    private Long batteryRelationId;
    private Long collectorId;
    private String collectorCode;
    private String collectorName;
    private String collectorChannelCode;
    private String measurementName;
    private String exceptionType;
    private Date openedAtUtc;
    private Date missingFromUtc;
    private Date recoveredAtUtc;
    private Date missingToUtc;
    private String intervalStatus;
    private String detail;

    public Long getIntervalId() { return intervalId; }
    public void setIntervalId(Long intervalId) { this.intervalId = intervalId; }
    public Long getTestId() { return testId; }
    public void setTestId(Long testId) { this.testId = testId; }
    public Long getBatteryRelationId() { return batteryRelationId; }
    public void setBatteryRelationId(Long batteryRelationId) { this.batteryRelationId = batteryRelationId; }
    public Long getCollectorId() { return collectorId; }
    public void setCollectorId(Long collectorId) { this.collectorId = collectorId; }
    public String getCollectorCode() { return collectorCode; }
    public void setCollectorCode(String collectorCode) { this.collectorCode = collectorCode; }
    public String getCollectorName() { return collectorName; }
    public void setCollectorName(String collectorName) { this.collectorName = collectorName; }
    public String getCollectorChannelCode() { return collectorChannelCode; }
    public void setCollectorChannelCode(String collectorChannelCode) { this.collectorChannelCode = collectorChannelCode; }
    public String getMeasurementName() { return measurementName; }
    public void setMeasurementName(String measurementName) { this.measurementName = measurementName; }
    public String getExceptionType() { return exceptionType; }
    public void setExceptionType(String exceptionType) { this.exceptionType = exceptionType; }
    public Date getOpenedAtUtc() { return openedAtUtc; }
    public void setOpenedAtUtc(Date openedAtUtc) { this.openedAtUtc = openedAtUtc; }
    public Date getMissingFromUtc() { return missingFromUtc; }
    public void setMissingFromUtc(Date missingFromUtc) { this.missingFromUtc = missingFromUtc; }
    public Date getRecoveredAtUtc() { return recoveredAtUtc; }
    public void setRecoveredAtUtc(Date recoveredAtUtc) { this.recoveredAtUtc = recoveredAtUtc; }
    public Date getMissingToUtc() { return missingToUtc; }
    public void setMissingToUtc(Date missingToUtc) { this.missingToUtc = missingToUtc; }
    public String getIntervalStatus() { return intervalStatus; }
    public void setIntervalStatus(String intervalStatus) { this.intervalStatus = intervalStatus; }
    public String getDetail() { return detail; }
    public void setDetail(String detail) { this.detail = detail; }
}

 ruoyi-system/src/main/java/com/ruoyi/system/domain/TTestCollectorSource.java

New file
@@ -0,0 +1,25 @@
package com.ruoyi.system.domain;

/** 接收中的测试采集器来源，供断线恢复协调器使用。 */
public class TTestCollectorSource
{
    private Long testId;
    private Long batteryRelationId;
    private String batteryNo;
    private Long collectorId;
    private String collectorCode;
    private String channelCode;

    public Long getTestId() { return testId; }
    public void setTestId(Long testId) { this.testId = testId; }
    public Long getBatteryRelationId() { return batteryRelationId; }
    public void setBatteryRelationId(Long batteryRelationId) { this.batteryRelationId = batteryRelationId; }
    public String getBatteryNo() { return batteryNo; }
    public void setBatteryNo(String batteryNo) { this.batteryNo = batteryNo; }
    public Long getCollectorId() { return collectorId; }
    public void setCollectorId(Long collectorId) { this.collectorId = collectorId; }
    public String getCollectorCode() { return collectorCode; }
    public void setCollectorCode(String collectorCode) { this.collectorCode = collectorCode; }
    public String getChannelCode() { return channelCode; }
    public void setChannelCode(String channelCode) { this.channelCode = channelCode; }
}

 ruoyi-system/src/main/java/com/ruoyi/system/mapper/TTestMapper.java

@@ -11,6 +11,8 @@
import com.ruoyi.system.domain.TTestPlanStep;
import com.ruoyi.system.domain.TTestResult;
import com.ruoyi.system.domain.TTestWarningThreshold;
import com.ruoyi.system.domain.TTestCollectorSource;
import com.ruoyi.system.domain.TTestCollectorExceptionInterval;
import com.ruoyi.system.domain.TDetectionChannel;
import com.ruoyi.system.domain.vo.TestBatteryOptionVo;
import com.ruoyi.system.domain.vo.TestChannelOptionVo;
@@ -87,4 +89,15 @@
            @Param("updatedAtUtc") Date updatedAtUtc);
    int updateReceptionStopped(@Param("testId") Long testId, @Param("updateBy") String updateBy,
            @Param("endedAtUtc") Date endedAtUtc);

    /** 查询仍在接收中的采集器来源。 */
    List<TTestCollectorSource> selectReceivingCollectorSources();
    /** 查询来源当前开放的断线区间。 */
    TTestCollectorExceptionInterval selectOpenCollectorException(@Param("testId") Long testId,
            @Param("batteryRelationId") Long batteryRelationId, @Param("collectorId") Long collectorId,
            @Param("channelCode") String channelCode);
    int insertCollectorExceptionInterval(TTestCollectorExceptionInterval interval);
    int closeCollectorExceptionInterval(@Param("intervalId") Long intervalId,
            @Param("recoveredAtUtc") Date recoveredAtUtc, @Param("missingToUtc") Date missingToUtc);
    List<TTestCollectorExceptionInterval> selectCollectorExceptionIntervals(Long testId);
}

 ruoyi-system/src/main/java/com/ruoyi/system/service/ICollectorDisconnectRecoveryService.java

New file
@@ -0,0 +1,11 @@
package com.ruoyi.system.service;

import java.time.Instant;
import java.util.Set;

/** 维护接收中采集器来源的断线异常与恢复边界。 */
public interface ICollectorDisconnectRecoveryService
{
    /** 根据当前模拟器在线设备编码对账所有接收中来源。 */
    void reconcile(Set<String> onlineCollectorCodes, Instant observedAt);
}

 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/CollectorDisconnectRecoveryServiceImpl.java

New file
@@ -0,0 +1,62 @@
package com.ruoyi.system.service.impl;

import java.time.Instant;
import java.util.Collections;
import java.util.Set;
import java.util.stream.Collectors;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import com.alibaba.fastjson2.JSON;
import com.ruoyi.system.domain.TTestCollectorExceptionInterval;
import com.ruoyi.system.domain.TTestCollectorSource;
import com.ruoyi.system.mapper.TTestMapper;
import com.ruoyi.system.service.ICollectorDisconnectRecoveryService;

/** 将采集器在线事实对账为可恢复的来源级异常区间。 */
@Service
public class CollectorDisconnectRecoveryServiceImpl implements ICollectorDisconnectRecoveryService
{
    private final TTestMapper mapper;

    public CollectorDisconnectRecoveryServiceImpl(TTestMapper mapper) { this.mapper = mapper; }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void reconcile(Set<String> onlineCollectorCodes, Instant observedAt)
    {
        Set<String> online = onlineCollectorCodes == null ? Collections.emptySet()
                : onlineCollectorCodes.stream().filter(code -> code != null && !code.isBlank()).collect(Collectors.toSet());
        for (TTestCollectorSource source : mapper.selectReceivingCollectorSources())
        {
            TTestCollectorExceptionInterval open = mapper.selectOpenCollectorException(source.getTestId(),
                    source.getBatteryRelationId(), source.getCollectorId(), source.getChannelCode());
            boolean isOnline = online.contains(source.getCollectorCode());
            if (!isOnline && open == null)
            {
                TTestCollectorExceptionInterval interval = new TTestCollectorExceptionInterval();
                interval.setTestId(source.getTestId());
                interval.setBatteryRelationId(source.getBatteryRelationId());
                interval.setCollectorId(source.getCollectorId());
                interval.setCollectorChannelCode(source.getChannelCode());
                interval.setExceptionType("COLLECTOR_DISCONNECTED");
                java.util.Date at = java.util.Date.from(observedAt);
                interval.setOpenedAtUtc(at);
                interval.setMissingFromUtc(at);
                interval.setDetail(JSON.toJSONString(java.util.Map.of("batteryNo", source.getBatteryNo(),
                        "collectorCode", source.getCollectorCode(), "message", "采集器离线，检测柜及其他来源继续接收")));
                mapper.insertCollectorExceptionInterval(interval);
                mapper.insertTestAudit(source.getTestId(), "COLLECTOR_DISCONNECTED", null, "system", null,
                        JSON.toJSONString(java.util.Map.of("collectorId", source.getCollectorId(),
                                "channelCode", source.getChannelCode(), "missingFrom", observedAt.toString())));
            }
            else if (isOnline && open != null)
            {
                java.util.Date at = java.util.Date.from(observedAt);
                mapper.closeCollectorExceptionInterval(open.getIntervalId(), at, at);
                mapper.insertTestAudit(source.getTestId(), "COLLECTOR_RECONNECTED", null, "system", null,
                        JSON.toJSONString(java.util.Map.of("collectorId", source.getCollectorId(),
                                "channelCode", source.getChannelCode(), "recoveredAt", observedAt.toString())));
            }
        }
    }
}

 ruoyi-system/src/main/java/com/ruoyi/system/service/impl/TTestLiveServiceImpl.java

@@ -24,6 +24,8 @@
import com.ruoyi.device.model.Measurement;
import com.ruoyi.system.domain.TTest;
import com.ruoyi.system.domain.TTestBattery;
import com.ruoyi.system.domain.TTestCollectorExceptionInterval;
import com.ruoyi.system.mapper.TTestMapper;
import com.ruoyi.system.service.ITestLiveService;
import com.ruoyi.system.service.ITTestService;

@@ -34,12 +36,13 @@
    private final ITTestService tests;
    private final DeviceDiscoveryCandidateStore candidates;
    private final BtsDevelopmentSimulator simulator;
    private final TTestMapper testMapper;
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
    private final ConcurrentMap<String, LiveSource> lastSources = new ConcurrentHashMap<>();

    public TTestLiveServiceImpl(ITTestService tests, DeviceDiscoveryCandidateStore candidates,
            BtsDevelopmentSimulator simulator)
    { this.tests = tests; this.candidates = candidates; this.simulator = simulator; }
            BtsDevelopmentSimulator simulator, TTestMapper testMapper)
    { this.tests = tests; this.candidates = candidates; this.simulator = simulator; this.testMapper = testMapper; }

    @Override public LiveState current(Long testId)
    {
@@ -55,12 +58,34 @@
            if (battery.getCollectorId() != null)
                sources.add(snapshotFor(testId, collectorSource(battery), freezeSnapshots));
        }
        List<Map<String, Object>> exceptions = sources.stream().filter(source -> !"ONLINE".equals(source.deviceStatus())
        List<TTestCollectorExceptionInterval> persisted = testMapper.selectCollectorExceptionIntervals(testId);
        if (persisted == null) persisted = List.of();
        List<Map<String, Object>> exceptions = new ArrayList<>(persisted.stream().map(interval -> {
            Map<String, Object> item = new LinkedHashMap<>();
            item.put("sourceType", "COLLECTOR");
            item.put("sourceId", interval.getCollectorId());
            item.put("collectorCode", interval.getCollectorCode());
            item.put("collectorName", interval.getCollectorName());
            item.put("channelCode", interval.getCollectorChannelCode());
            item.put("measurementName", interval.getMeasurementName());
            item.put("exceptionType", interval.getExceptionType());
            item.put("status", interval.getIntervalStatus());
            item.put("openedAt", interval.getOpenedAtUtc());
            item.put("missingFrom", interval.getMissingFromUtc());
            item.put("recoveredAt", interval.getRecoveredAtUtc());
            item.put("missingTo", interval.getMissingToUtc());
            item.put("detail", interval.getDetail());
            return item;
        }).toList());
        exceptions.addAll(sources.stream().filter(source -> !"COLLECTOR".equals(source.sourceType())
                && !"ONLINE".equals(source.deviceStatus())
                && !"WORKING".equals(source.deviceStatus()) && !"PAUSED".equals(source.deviceStatus()))
                .map(source -> Map.<String, Object>of("sourceType", source.sourceType(), "sourceId", source.sourceId(),
                        "batteryNo", source.batteryNo(), "status", source.deviceStatus(), "message", "数据来源当前不可观测"))
                .toList();
        return new LiveState(testId, test.getTestStatus(), test.getReceptionStatus(), Instant.now(), sources, exceptions);
                .toList());
        boolean abnormal = persisted.stream().anyMatch(item -> "OPEN".equals(item.getIntervalStatus()));
        String testStatus = abnormal && "RUNNING".equals(test.getTestStatus()) ? "RUNNING_ABNORMAL" : test.getTestStatus();
        return new LiveState(testId, testStatus, test.getReceptionStatus(), Instant.now(), sources, exceptions);
    }

    private LiveSource snapshotFor(Long testId, LiveSource fresh, boolean freeze)

 ruoyi-system/src/main/resources/mapper/system/TTestMapper.xml

@@ -108,6 +108,23 @@
        <result property="exceptionInfo" column="exception_info" />
        <result property="delFlag" column="del_flag" />
    </resultMap>
    <resultMap id="TTestCollectorExceptionIntervalResultMap" type="TTestCollectorExceptionInterval">
        <id property="intervalId" column="interval_id" />
        <result property="testId" column="test_id" />
        <result property="batteryRelationId" column="battery_relation_id" />
        <result property="collectorId" column="collector_id" />
        <result property="collectorCode" column="collector_code" />
        <result property="collectorName" column="collector_name" />
        <result property="collectorChannelCode" column="collector_channel_code" />
        <result property="measurementName" column="measurement_name" />
        <result property="exceptionType" column="exception_type" />
        <result property="openedAtUtc" column="opened_at_utc" />
        <result property="missingFromUtc" column="missing_from_utc" />
        <result property="recoveredAtUtc" column="recovered_at_utc" />
        <result property="missingToUtc" column="missing_to_utc" />
        <result property="intervalStatus" column="interval_status" />
        <result property="detail" column="detail" />
    </resultMap>

    <sql id="selectTestBase">
        select t.test_id, t.test_name, t.creator_id, creator.nick_name creator_name,
@@ -458,4 +475,48 @@
            update_by = #{updateBy}, update_time = #{endedAtUtc}
        where test_id = #{testId} and reception_status in ('RECEIVING', 'PAUSED') and del_flag = 0
    </update>

    <select id="selectReceivingCollectorSources" resultType="com.ruoyi.system.domain.TTestCollectorSource">
        select test_fact.test_id testId, battery.relation_id batteryRelationId, battery.battery_no batteryNo,
               collector.collector_id collectorId, collector.collector_code collectorCode,
               mapping.channel_code channelCode
        from t_test test_fact
        join t_test_battery battery on battery.test_id = test_fact.test_id and battery.del_flag = 0
        join t_collector collector on collector.collector_id = battery.collector_id
            and collector.del_flag = 0
        join t_test_collector_channel mapping on mapping.battery_relation_id = battery.relation_id
            and mapping.test_id = test_fact.test_id
        where test_fact.reception_status = 'RECEIVING' and test_fact.del_flag = 0
        order by test_fact.test_id, battery.relation_id, mapping.mapping_id
    </select>
    <select id="selectOpenCollectorException" resultMap="TTestCollectorExceptionIntervalResultMap">
        select * from t_test_collector_exception_interval
        where test_id = #{testId} and battery_relation_id = #{batteryRelationId}
          and collector_id = #{collectorId} and collector_channel_code = #{channelCode}
          and exception_type = 'COLLECTOR_DISCONNECTED' and interval_status = 'OPEN'
        order by interval_id desc limit 1
    </select>
    <insert id="insertCollectorExceptionInterval" parameterType="TTestCollectorExceptionInterval" useGeneratedKeys="true" keyProperty="intervalId">
        insert into t_test_collector_exception_interval
        (test_id, battery_relation_id, collector_id, collector_channel_code, exception_type,
         opened_at_utc, missing_from_utc, detail)
        values (#{testId}, #{batteryRelationId}, #{collectorId}, #{collectorChannelCode},
                #{exceptionType}, #{openedAtUtc}, #{missingFromUtc}, #{detail})
    </insert>
    <update id="closeCollectorExceptionInterval">
        update t_test_collector_exception_interval
        set recovered_at_utc = #{recoveredAtUtc}, missing_to_utc = #{missingToUtc}, interval_status = 'CLOSED'
        where interval_id = #{intervalId} and interval_status = 'OPEN'
    </update>
    <select id="selectCollectorExceptionIntervals" parameterType="Long" resultMap="TTestCollectorExceptionIntervalResultMap">
        select exception_fact.*, collector.collector_code, collector.collector_name,
               mapping.measurement_name
        from t_test_collector_exception_interval exception_fact
        left join t_collector collector on collector.collector_id = exception_fact.collector_id
        left join t_test_collector_channel mapping on mapping.battery_relation_id = exception_fact.battery_relation_id
            and mapping.collector_id = exception_fact.collector_id
            and mapping.channel_code = exception_fact.collector_channel_code
        where exception_fact.test_id = #{testId}
        order by exception_fact.opened_at_utc, exception_fact.interval_id
    </select>
</mapper>