package com.ruoyi.system.service.impl;
|
|
import java.nio.charset.StandardCharsets;
|
import java.security.MessageDigest;
|
import java.time.Instant;
|
import java.util.Date;
|
import java.util.HexFormat;
|
import java.util.List;
|
import java.util.Objects;
|
import com.alibaba.fastjson2.JSON;
|
import org.springframework.stereotype.Service;
|
import org.springframework.transaction.PlatformTransactionManager;
|
import org.springframework.transaction.support.TransactionTemplate;
|
import org.springframework.transaction.annotation.Transactional;
|
import com.ruoyi.common.exception.ServiceException;
|
import com.ruoyi.device.api.DeviceAdapter;
|
import com.ruoyi.device.model.ChannelIdentity;
|
import com.ruoyi.device.model.HistoryAvailability;
|
import com.ruoyi.device.model.HistoryPage;
|
import com.ruoyi.device.model.HistorySample;
|
import com.ruoyi.device.simulator.BtsSimulatorStateStore;
|
import com.ruoyi.device.simulator.BtsSimulatorStateStore.CollectorMeasurementRecord;
|
import com.ruoyi.device.registry.DeviceAdapterRegistry;
|
import com.ruoyi.system.domain.TTest;
|
import com.ruoyi.system.domain.TTestBackfillCursor;
|
import com.ruoyi.system.domain.TTestBattery;
|
import com.ruoyi.system.domain.TTestChannelBinding;
|
import com.ruoyi.system.domain.TTestCollectorChannel;
|
import com.ruoyi.system.domain.TTestReceptionInterval;
|
import com.ruoyi.system.domain.TTestIntegrityIssue;
|
import com.ruoyi.system.mapper.TTestMapper;
|
import com.ruoyi.system.service.ITestBackfillService;
|
import com.ruoyi.system.service.ITTestService;
|
|
/** 以测试动作边界筛选来源历史，并以数据库游标保证补采可恢复。 */
|
@Service
|
public class TestBackfillServiceImpl implements ITestBackfillService
|
{
|
private static final int PAGE_SIZE = 1000;
|
private static final int MAX_INTEGRITY_RETRIES = 3;
|
private final TTestMapper mapper;
|
private final DeviceAdapterRegistry registry;
|
private final BtsSimulatorStateStore collectorHistoryStore;
|
private final ITTestService tests;
|
private final TransactionTemplate transactionTemplate;
|
|
public TestBackfillServiceImpl(TTestMapper mapper, DeviceAdapterRegistry registry,
|
BtsSimulatorStateStore collectorHistoryStore, ITTestService tests,
|
PlatformTransactionManager transactionManager)
|
{
|
this.mapper = mapper;
|
this.registry = registry;
|
this.collectorHistoryStore = collectorHistoryStore;
|
this.tests = tests;
|
this.transactionTemplate = new TransactionTemplate(transactionManager);
|
}
|
|
@Override
|
@Transactional(rollbackFor = Exception.class)
|
public void reconcileNaturalEnd()
|
{
|
List<TTest> summaries = mapper.selectTestsForNaturalEnd();
|
if (summaries == null) return;
|
for (TTest summary : summaries)
|
{
|
TTest test = tests.selectTestById(summary.getTestId());
|
if (test == null || test.getBatteries() == null || test.getBatteries().isEmpty()) continue;
|
boolean allFinished = true;
|
for (TTestBattery battery : test.getBatteries())
|
{
|
TTestChannelBinding binding = battery.getBinding();
|
if (binding == null) { allFinished = false; break; }
|
var channel = mapper.selectReceptionChannelForUpdate(binding.getChannelId());
|
if (channel == null || !"FINISHED".equalsIgnoreCase(channel.getWorkStatus()))
|
{ allFinished = false; break; }
|
}
|
if (allFinished) endNaturally(test.getTestId(), "NATURAL_FINISH");
|
}
|
}
|
|
@Override
|
public void resumePendingBackfill()
|
{
|
List<TTest> pending = mapper.selectTestsForBackfill();
|
if (pending == null) return;
|
for (TTest test : pending)
|
{
|
try { transactionTemplate.executeWithoutResult(status -> backfill(test)); }
|
catch (RuntimeException exception)
|
{
|
mapper.updateTestBackfillWaiting(test.getTestId(), new Date(), exception.getMessage());
|
}
|
}
|
}
|
|
@Override
|
@Transactional(rollbackFor = Exception.class)
|
public void endNaturally(Long testId, String reason)
|
{
|
if (testId == null) throw new ServiceException("测试ID不能为空");
|
Date endedAt = new Date();
|
if (mapper.updateTestNaturallyEnded(testId, endedAt) != 1) return;
|
mapper.closeOpenReceptionIntervals(testId, endedAt, reason == null ? "NATURAL_FINISH" : reason);
|
mapper.deleteChannelLeasesByTestId(testId);
|
mapper.deleteCollectorChannelLeasesByTestId(testId);
|
mapper.insertTestAudit(testId, "RECEPTION_NATURALLY_ENDED", null, "system", null,
|
JSON.toJSONString(java.util.Map.of("testStatus", "COMPLETION_PENDING",
|
"receptionStatus", "ENDED", "reason", reason == null ? "NATURAL_FINISH" : reason)));
|
}
|
|
@Transactional(rollbackFor = Exception.class)
|
protected void backfill(TTest test)
|
{
|
if (mapper.updateTestBackfillStarted(test.getTestId(), new Date()) == 0) return;
|
// 来源按检测柜、采集器分别提交；任何一页失败都不推进该来源游标。
|
for (TTestBattery battery : test.getBatteries() == null ? List.<TTestBattery>of() : test.getBatteries())
|
{
|
TTestChannelBinding binding = battery.getBinding();
|
if (binding != null) backfillCabinet(test, battery, binding);
|
for (TTestCollectorChannel mapping : battery.getCollectorChannels() == null
|
? List.<TTestCollectorChannel>of() : battery.getCollectorChannels())
|
backfillCollector(test, battery, mapping);
|
}
|
Date checkedAt = new Date();
|
boolean complete = checkIntegrity(test, checkedAt);
|
if (complete)
|
{
|
mapper.updateTestCompleted(test.getTestId(), checkedAt);
|
mapper.insertTestAudit(test.getTestId(), "TEST_COMPLETED", null, "system", null,
|
JSON.toJSONString(java.util.Map.of("backfillStatus", "COMPLETED", "integrity", "PASSED")));
|
}
|
else
|
{
|
mapper.updateTestBackfillWaiting(test.getTestId(), checkedAt, "区间完整性检查未通过");
|
mapper.insertTestAudit(test.getTestId(), "INTEGRITY_CHECK_BLOCKED", null, "system", null,
|
JSON.toJSONString(java.util.Map.of("backfillStatus", "WAITING", "integrity", "OPEN")));
|
}
|
}
|
|
private void backfillCabinet(TTest test, TTestBattery battery, TTestChannelBinding binding)
|
{
|
String adapterCode = binding.getAdapterCode();
|
if (adapterCode == null || adapterCode.isBlank()) throw new ServiceException("检测柜适配器身份缺失");
|
ChannelIdentity identity = new ChannelIdentity(adapterCode, binding.getDevtype(), binding.getDevid(),
|
binding.getSubdevid(), binding.getChlid());
|
List<TTestReceptionInterval> intervals = intervals(test.getTestId(), binding.getBindingId(), "CABINET_CHANNEL",
|
binding.getChannelId(), null);
|
if (intervals.isEmpty()) return;
|
HistoryAvailability availability = registry.execute(adapterCode,
|
adapter -> adapter.inquireHistory(java.util.Set.of(identity)).stream().findFirst()
|
.orElseThrow(() -> new ServiceException("检测柜未返回历史清单")));
|
TTestBackfillCursor cursor = cursor(test, binding.getBindingId(), "CABINET_CHANNEL", binding.getChannelId(), null,
|
availability.btsTestId());
|
cursor.setProtocolMetadata(JSON.toJSONString(java.util.Map.of("availableRecordCount",
|
availability.availableRecordCount(), "uploadComplete", availability.uploadComplete(),
|
"adapterCode", adapterCode)));
|
cursor.setUpdatedAtUtc(new Date());
|
mapper.updateBackfillCursor(cursor);
|
long position = Math.max(1L, cursor.getNextPosition() == null ? 1L : cursor.getNextPosition());
|
while (true)
|
{
|
final long requestPosition = position;
|
HistoryPage page = registry.execute(adapterCode,
|
adapter -> adapter.downloadHistory(identity, availability.btsTestId(), requestPosition, PAGE_SIZE));
|
for (HistorySample sample : page.samples())
|
{
|
if (!within(sample.deviceTime(), intervals)) continue;
|
saveBtsSample(test, binding, intervals, availability.btsTestId(), sample);
|
}
|
long next = page.nextPosition() > position ? page.nextPosition() : position + page.samples().size();
|
cursor.setNextPosition(next);
|
cursor.setSourceStatus(page.samples().isEmpty() && page.nextPosition() <= position ? "COMPLETED" : "RUNNING");
|
cursor.setUpdatedAtUtc(new Date());
|
mapper.updateBackfillCursor(cursor);
|
if (page.samples().isEmpty() || page.nextPosition() <= position) break;
|
position = next;
|
}
|
cursor.setSourceStatus("COMPLETED");
|
cursor.setUpdatedAtUtc(new Date());
|
mapper.updateBackfillCursor(cursor);
|
}
|
|
private void saveBtsSample(TTest test, TTestChannelBinding binding, List<TTestReceptionInterval> intervals,
|
String btsTestId, HistorySample sample)
|
{
|
byte[] hash = hash(sample.sequenceId() + "|" + sample.stepNo() + "|" + sample.cycleNo() + "|"
|
+ sample.stepType() + "|" + sample.deviceTimeRaw() + "|" + sample.voltageV() + "|"
|
+ sample.currentA() + "|" + sample.capacityAh() + "|" + sample.energyWh());
|
String old = mapper.selectSampleContentHash(binding.getChannelId(), btsTestId, sample.sequenceId());
|
if (old != null && !old.equalsIgnoreCase(HexFormat.of().formatHex(hash)))
|
{
|
mapper.insertBackfillConflict(test.getTestId(), binding.getChannelId(),
|
"BTS testid=" + btsTestId + ", seqid=" + sample.sequenceId());
|
return;
|
}
|
mapper.insertBtsSample(test.getTestId(), binding.getBindingId(), binding.getChannelId(),
|
intervalId(intervals, sample.deviceTime()), btsTestId, sample.sequenceId(), Date.from(sample.deviceTime()),
|
sample.deviceTimeRaw(), new Date(), sample.cycleNo(), sample.stepNo(), sample.stepType(), sample.testTimeMs(),
|
sample.voltageV(), sample.currentA(), sample.capacityAh(), sample.energyWh(), hash);
|
Long sampleId = mapper.selectBtsSampleId(binding.getChannelId(), btsTestId, sample.sequenceId());
|
if (sampleId != null)
|
sample.measurements().forEach((code, measurement) -> mapper.insertBtsMeasurement(sampleId, code,
|
measurement.value(), measurement.unit() == null ? "" : measurement.unit(), measurement.sourceType()));
|
}
|
|
private void backfillCollector(TTest test, TTestBattery battery, TTestCollectorChannel mapping)
|
{
|
if (battery.getCollectorId() == null) return;
|
if (battery.getBinding() == null) throw new ServiceException("采集器绑定缺少检测柜通道");
|
String collectorCode = mapper.selectCollectorCode(battery.getCollectorId());
|
if (collectorCode == null || collectorCode.isBlank()) throw new ServiceException("采集器来源不存在");
|
List<TTestReceptionInterval> intervals = intervals(test.getTestId(), battery.getBinding().getBindingId(),
|
"COLLECTOR_CHANNEL", battery.getCollectorId(), mapping.getChannelCode());
|
if (intervals.isEmpty()) return;
|
TTestBackfillCursor cursor = cursor(test, battery.getBinding().getBindingId(), "COLLECTOR_CHANNEL",
|
battery.getCollectorId(), mapping.getChannelCode(), collectorCode);
|
long offset = Math.max(0L, (cursor.getNextPosition() == null ? 1L : cursor.getNextPosition()) - 1L);
|
while (true)
|
{
|
var page = collectorHistoryStore.loadCollectorHistoryPage(collectorCode, offset, PAGE_SIZE, false);
|
for (CollectorMeasurementRecord record : page.records())
|
{
|
if (!mapping.getMeasurementCode().equalsIgnoreCase(record.metricCode()) || !within(record.deviceTime(), intervals)) continue;
|
saveCollectorSample(test, battery, mapping, intervals, record);
|
}
|
offset = page.nextOffset();
|
cursor.setNextPosition(offset + 1L);
|
cursor.setSourceStatus(page.hasMore() ? "RUNNING" : "COMPLETED");
|
cursor.setUpdatedAtUtc(new Date());
|
mapper.updateBackfillCursor(cursor);
|
if (!page.hasMore()) break;
|
}
|
}
|
|
private void saveCollectorSample(TTest test, TTestBattery battery, TTestCollectorChannel mapping,
|
List<TTestReceptionInterval> intervals, CollectorMeasurementRecord record)
|
{
|
byte[] hash = hash(record.sequenceId() + "|" + record.deviceTimeRaw() + "|" + record.metricCode() + "|"
|
+ record.value() + "|" + record.unit());
|
String old = mapper.selectCollectorSampleContentHash(battery.getCollectorId(), record.channelCode(),
|
record.sequenceId(), record.metricCode());
|
if (old != null && !old.equalsIgnoreCase(HexFormat.of().formatHex(hash)))
|
{
|
mapper.insertBackfillConflict(test.getTestId(), null,
|
"采集器=" + record.collectorCode() + ", seqid=" + record.sequenceId());
|
return;
|
}
|
mapper.insertCollectorSample(test.getTestId(), battery.getBinding().getBindingId(),
|
intervalId(intervals, record.deviceTime()), battery.getCollectorId(),
|
record.channelCode(), record.sequenceId(), Date.from(record.deviceTime()), record.deviceTimeRaw(),
|
Date.from(record.serverReceivedAt()), record.metricCode(), record.value(), record.unit(), hash);
|
}
|
|
private boolean checkIntegrity(TTest test, Date checkedAt)
|
{
|
boolean complete = true;
|
for (TTestReceptionInterval interval : mapper.selectTestReceptionIntervals(test.getTestId()))
|
{
|
String sourceType = interval.getSourceType();
|
List<Long> sequenceIds = "CABINET_CHANNEL".equals(sourceType)
|
? mapper.selectBtsSequencesByInterval(interval.getIntervalId())
|
: mapper.selectCollectorSequencesByInterval(interval.getIntervalId());
|
List<IntervalIntegrityChecker.Gap> gaps = IntervalIntegrityChecker.gaps(sequenceIds);
|
if (sequenceIds == null || sequenceIds.isEmpty())
|
{
|
complete = false;
|
TTestIntegrityIssue old = mapper.selectOpenIntegrityIssue(interval.getIntervalId(), sourceType,
|
interval.getSourceId(), interval.getSourceChannelCode(), null, null);
|
if (old == null)
|
{
|
TTestIntegrityIssue issue = new TTestIntegrityIssue();
|
issue.setTestId(test.getTestId()); issue.setIntervalId(interval.getIntervalId());
|
issue.setBindingId(interval.getBindingId()); issue.setSourceType(sourceType);
|
issue.setSourceId(interval.getSourceId()); issue.setSourceChannelCode(interval.getSourceChannelCode());
|
issue.setIssueType("NO_SAMPLES"); issue.setRetryCount(1);
|
issue.setDetail("接收区间没有可确认的历史采样"); issue.setFirstDetectedAtUtc(checkedAt);
|
issue.setLastDetectedAtUtc(checkedAt); mapper.insertIntegrityIssue(issue);
|
mapper.insertTestAudit(test.getTestId(), "INTEGRITY_GAP_DETECTED", null, "system", null,
|
JSON.toJSONString(java.util.Map.of("intervalId", interval.getIntervalId(), "sourceType", sourceType,
|
"issueType", "NO_SAMPLES")));
|
}
|
else
|
{
|
int retry = old.getRetryCount() == null ? 1 : old.getRetryCount() + 1;
|
int cappedRetry = Math.min(retry, MAX_INTEGRITY_RETRIES);
|
mapper.touchIntegrityIssue(old.getIssueId(), cappedRetry,
|
old.getDetail(), checkedAt);
|
mapper.insertTestAudit(test.getTestId(), "INTEGRITY_GAP_RETRIED", null, "system", null,
|
JSON.toJSONString(java.util.Map.of("issueId", old.getIssueId(), "retryCount", cappedRetry)));
|
if (cappedRetry < MAX_INTEGRITY_RETRIES)
|
mapper.rewindBackfillCursor(test.getTestId(), interval.getBindingId(), sourceType,
|
interval.getSourceId(), interval.getSourceChannelCode(), 1L);
|
}
|
continue;
|
}
|
if (gaps.isEmpty())
|
{
|
int resolved = mapper.resolveIntegrityIssues(test.getTestId(), interval.getIntervalId(), sourceType,
|
interval.getSourceId(), interval.getSourceChannelCode(), checkedAt);
|
if (resolved > 0) mapper.insertTestAudit(test.getTestId(), "INTEGRITY_GAP_RESOLVED", null, "system", null,
|
JSON.toJSONString(java.util.Map.of("intervalId", interval.getIntervalId())));
|
continue;
|
}
|
complete = false;
|
for (TTestIntegrityIssue open : mapper.selectOpenIntegrityIssues(interval.getIntervalId(), sourceType,
|
interval.getSourceId(), interval.getSourceChannelCode()))
|
{
|
boolean stillOpen = gaps.stream().anyMatch(gap -> Objects.equals(gap.from(), open.getMissingFromSequence())
|
&& Objects.equals(gap.to(), open.getMissingToSequence()));
|
if (!stillOpen && mapper.resolveIntegrityIssue(open.getIssueId(), checkedAt) > 0)
|
mapper.insertTestAudit(test.getTestId(), "INTEGRITY_GAP_RESOLVED", null, "system", null,
|
JSON.toJSONString(java.util.Map.of("issueId", open.getIssueId())));
|
}
|
for (IntervalIntegrityChecker.Gap gap : gaps)
|
{
|
TTestIntegrityIssue old = mapper.selectOpenIntegrityIssue(interval.getIntervalId(), sourceType,
|
interval.getSourceId(), interval.getSourceChannelCode(), gap.from(), gap.to());
|
if (old == null)
|
{
|
TTestIntegrityIssue issue = new TTestIntegrityIssue();
|
issue.setTestId(test.getTestId()); issue.setIntervalId(interval.getIntervalId());
|
issue.setBindingId(interval.getBindingId()); issue.setSourceType(sourceType);
|
issue.setSourceId(interval.getSourceId()); issue.setSourceChannelCode(interval.getSourceChannelCode());
|
issue.setIssueType("SEQUENCE_GAP"); issue.setMissingFromSequence(gap.from());
|
issue.setMissingToSequence(gap.to()); issue.setRetryCount(1);
|
issue.setDetail("接收区间内存在未解释序号缺口"); issue.setFirstDetectedAtUtc(checkedAt);
|
issue.setLastDetectedAtUtc(checkedAt); mapper.insertIntegrityIssue(issue);
|
mapper.insertTestAudit(test.getTestId(), "INTEGRITY_GAP_DETECTED", null, "system", null,
|
JSON.toJSONString(java.util.Map.of("intervalId", interval.getIntervalId(), "sourceType", sourceType,
|
"missingFrom", gap.from(), "missingTo", gap.to())));
|
}
|
else
|
{
|
int retry = old.getRetryCount() == null ? 1 : old.getRetryCount() + 1;
|
int cappedRetry = Math.min(retry, MAX_INTEGRITY_RETRIES);
|
mapper.touchIntegrityIssue(old.getIssueId(), cappedRetry,
|
old.getDetail(), checkedAt);
|
mapper.insertTestAudit(test.getTestId(), "INTEGRITY_GAP_RETRIED", null, "system", null,
|
JSON.toJSONString(java.util.Map.of("issueId", old.getIssueId(), "retryCount", cappedRetry)));
|
if (cappedRetry >= MAX_INTEGRITY_RETRIES) continue;
|
}
|
// next_position 是分页偏移而非 seqid；从安全起点重放，依靠幂等键避免覆盖原始事实。
|
mapper.rewindBackfillCursor(test.getTestId(), interval.getBindingId(), sourceType,
|
interval.getSourceId(), interval.getSourceChannelCode(), 1L);
|
}
|
}
|
return complete && mapper.countOpenIntegrityIssues(test.getTestId()) == 0
|
&& mapper.countOpenExceptions(test.getTestId()) == 0;
|
}
|
|
private TTestBackfillCursor cursor(TTest test, Long bindingId, String sourceType, Long sourceId,
|
String channelCode, String protocolTestId)
|
{
|
TTestBackfillCursor value = mapper.selectBackfillCursorForUpdate(test.getTestId(), bindingId, sourceType, sourceId, channelCode);
|
if (value != null)
|
{
|
if (value.getProtocolTestId() != null && !Objects.equals(value.getProtocolTestId(), protocolTestId))
|
throw new ServiceException("历史协议测试标识发生变化");
|
return value;
|
}
|
value = new TTestBackfillCursor();
|
value.setTestId(test.getTestId()); value.setBindingId(bindingId); value.setSourceType(sourceType);
|
value.setSourceId(sourceId); value.setSourceChannelCode(channelCode); value.setProtocolTestId(protocolTestId);
|
value.setNextPosition(1L); value.setSourceStatus("PENDING"); value.setProtocolMetadata("{}");
|
value.setUpdatedAtUtc(new Date());
|
mapper.insertBackfillCursor(value);
|
return value;
|
}
|
|
private List<TTestReceptionInterval> intervals(Long testId, Long bindingId, String sourceType, Long sourceId,
|
String channelCode)
|
{
|
return mapper.selectTestReceptionIntervals(testId).stream()
|
.filter(item -> Objects.equals(item.getBindingId(), bindingId) && Objects.equals(item.getSourceType(), sourceType)
|
&& Objects.equals(item.getSourceId(), sourceId)
|
&& Objects.equals(item.getSourceChannelCode(), channelCode)).toList();
|
}
|
|
private boolean within(Instant time, List<TTestReceptionInterval> intervals)
|
{
|
return time != null && intervals.stream().anyMatch(item -> !time.isBefore(item.getStartBoundaryAtUtc().toInstant())
|
&& (item.getEndBoundaryAtUtc() == null || time.isBefore(item.getEndBoundaryAtUtc().toInstant())));
|
}
|
|
private Long intervalId(List<TTestReceptionInterval> intervals, Instant time)
|
{
|
return intervals.stream().filter(item -> !time.isBefore(item.getStartBoundaryAtUtc().toInstant())
|
&& (item.getEndBoundaryAtUtc() == null || time.isBefore(item.getEndBoundaryAtUtc().toInstant())))
|
.map(TTestReceptionInterval::getIntervalId).findFirst().orElse(null);
|
}
|
|
private byte[] hash(String value)
|
{
|
try { return MessageDigest.getInstance("SHA-256").digest(value.getBytes(StandardCharsets.UTF_8)); }
|
catch (Exception exception) { throw new IllegalStateException("SHA-256不可用", exception); }
|
}
|
}
|
