credentials)
{
this.adapters = adapters;
this.records = records;
this.clock = clock;
this.credentials = credentials;
}
@Override
@Transactional(rollbackFor = Exception.class)
public IngestResult ingest(String adapterCode, String token, String body)
{
TDeviceAdapter adapter = requirePushSource(adapterCode);
requireToken(adapter, token);
JSONObject payload = parse(body);
String kind = payload.getString("kind");
Instant receivedAt = clock.instant();
Counters counters = new Counters();
if (TDeviceIngestRecord.KIND_HEARTBEAT.equals(kind))
{
// 心跳没有行：它只刷新"取数程序还在推"这一条事实。
}
else if (TDeviceIngestRecord.KIND_TOPOLOGY.equals(kind))
{
storeTopology(adapterCode, adapter.getTimezoneId(), payload, receivedAt, counters);
}
else if (TDeviceIngestRecord.KIND_SAMPLE.equals(kind))
{
storeSamples(adapterCode, adapter.getTimezoneId(), payload, receivedAt, counters);
}
else
{
throw payloadInvalid("未知的载荷类别 kind：" + kind);
}
// 每一批成功接收都刷新心跳：收到采样行同样是"程序还在推"的证据，缺一次心跳不会误判离线。
touchHeartbeat(adapterCode, receivedAt);
return new IngestResult(counters.accepted, counters.ignored, counters.conflicts);
}
private TDeviceAdapter requirePushSource(String adapterCode)
{
TDeviceAdapter adapter = StringUtils.isBlank(adapterCode) ? null
: adapters.selectDeviceAdapterByCode(adapterCode);
// 来源不存在、类型不对与未启用都明确拒绝：这几种情况一律不落数据。
if (adapter == null) throw new ServiceException("来源不存在：" + adapterCode, DeviceAdapterErrorCodes.NOT_FOUND);
if (!PushIngestAdapter.ADAPTER_TYPE.equalsIgnoreCase(String.valueOf(adapter.getAdapterType())))
throw new ServiceException("该来源不是推式来源：" + adapterCode, DeviceAdapterErrorCodes.UNSUPPORTED_TYPE);
if (!"0".equals(adapter.getStatus()))
throw new ServiceException("来源未启用：" + adapterCode, DeviceIngestErrorCodes.SOURCE_DISABLED);
return adapter;
}
/** 端点令牌与厂商库口令共用同一套取值顺序；配置缺失同样按凭据错误拒绝（失败关闭）。 */
private void requireToken(TDeviceAdapter adapter, String token)
{
String key = option(adapter.getSourceConfig(), TOKEN_ENV_OPTION, DEFAULT_TOKEN_ENV);
String expected;
try
{
expected = credentials.apply(key);
}
catch (RuntimeException failure)
{
throw new ServiceException("来源凭据不可用：" + failure.getMessage(), DeviceIngestErrorCodes.TOKEN_INVALID);
}
if (StringUtils.isBlank(token) || !MessageDigest.isEqual(
expected.getBytes(StandardCharsets.UTF_8), token.trim().getBytes(StandardCharsets.UTF_8)))
throw new ServiceException("来源凭据不匹配：" + adapter.getAdapterCode(),
DeviceIngestErrorCodes.TOKEN_INVALID);
}
/**
* 拓扑上报：声明该来源当前有哪些通道。行里只有厂商自己的通道号，身份映射在本方法里完成。
* 拓扑是"这个通道还在"的长期声明，重复上报被幂等忽略（不多占空间，也不改写既有行）。
*/
private void storeTopology(String adapterCode, String timezoneId, JSONObject payload, Instant receivedAt,
Counters counters)
{
JSONArray rows = rows(payload);
for (int index = 0; index < rows.size(); index++)
{
JSONObject row = rowAt(rows, index);
Integer channel = row.getInteger("channel");
if (channel == null || channel < 1) throw payloadInvalid("第 " + index + " 行的 channel 不合法：" + row);
TDeviceIngestRecord value = base(adapterCode, timezoneId, TDeviceIngestRecord.KIND_TOPOLOGY, channel,
receivedAt);
value.setRecordKey("topology:" + channel);
value.setPayloadJson(JSON.toJSONString(row));
store(value, counters);
}
}
/**
* 逐点数据行：一批来自同一张厂商表（同一测试同一通道），因此身份与来源定位取批次头上的声明。
* 行本身原样保存，只额外解析出水位与投影需要的字段。
*/
private void storeSamples(String adapterCode, String timezoneId, JSONObject payload, Instant receivedAt,
Counters counters)
{
Integer channel = payload.getInteger("channel");
if (channel == null || channel < 1) throw payloadInvalid("批次缺少合法的 channel：" + payload);
String sourceRef = sourceRef(payload);
JSONArray rows = rows(payload);
for (int index = 0; index < rows.size(); index++)
{
JSONObject row = rowAt(rows, index);
TDeviceIngestRecord value = base(adapterCode, timezoneId, TDeviceIngestRecord.KIND_SAMPLE, channel,
receivedAt);
value.setSourceRef(sourceRef);
value.setRecordSequence(requiredSequence(row));
value.setDeviceTimeRaw(requiredText(row, "record_time"));
value.setDeviceTimeUtc(Date.from(filetime(value.getDeviceTimeRaw())));
value.setCycleNo(row.getInteger("Cycle_Index"));
value.setStepNo(row.getInteger("Step_Index"));
value.setRelativeTimeSeconds(row.getBigDecimal("Step_Time"));
value.setVoltageV(row.getBigDecimal("Voltage"));
value.setCurrentA(row.getBigDecimal("eCurrent"));
value.setChargeCapacityAh(row.getBigDecimal("ChargeCapacity"));
value.setDischargeCapacityAh(row.getBigDecimal("DischargeCapacity"));
value.setChargeEnergyWh(row.getBigDecimal("ChargeEnergy"));
value.setDischargeEnergyWh(row.getBigDecimal("DischargeEnergy"));
// 幂等键含来源定位：22 所每个测试新建一个库，Data_Point 在每个测试内重新编号，只看它会让两次测试撞键。
value.setRecordKey(sourceRef + ":" + value.getRecordSequence());
value.setPayloadJson(JSON.toJSONString(row));
store(value, counters);
}
}
// ------------------------------------------------------------------ 落地行
private TDeviceIngestRecord base(String adapterCode, String timezoneId, String kind, int channel,
Instant receivedAt)
{
TDeviceIngestRecord value = new TDeviceIngestRecord();
value.setAdapterCode(adapterCode);
value.setPayloadKind(kind);
// 身份映射与适配器读落地行时用的是同一组常量：这个形态只在 PushIngestAdapter 里定义一次。
value.setDevtype(PushIngestAdapter.DEVICE_TYPE);
value.setDevid(PushIngestAdapter.DEVICE_ID);
value.setSubdevid(PushIngestAdapter.SUB_DEVICE_ID);
value.setChlid(channel);
value.setRecordSequence(0L);
// 来源时区是"当时怎么解释这条来源"的一部分，跟着行落库：厂商给的是绝对时间，
// 换算出的是 UTC，时区只用于展示；实例时区以后改了，历史行仍保留当时的口径。
value.setTimezoneId(timezoneId);
value.setReceivedAtUtc(Date.from(receivedAt));
return value;
}
/**
* 落一行并用来源行身份判幂等：撞键即视为重复推送，比较内容摘要后区分"同内容重复"与"同身份不同内容"。
* 内容不同只记警告、不覆盖——既有那一行是已经收到的事实。
*/
/**
* 落一行并用来源行身份判幂等：撞键即视为重复推送，比较内容摘要后区分"同内容重复"与"同身份不同内容"。
* 内容不同只记警告、不覆盖——既有那一行是已经收到的事实。
*
* 落库没有新增、又查不到这一行时**明确失败**，而不是当作重复忽略：`insert ignore` 把
* 约束冲突与"转换失败被跳过"都变成"没插入"，后者若不报错，取数程序会拿到 2xx 推进水位，
* 那条设备行就此消失。宁可让这一批重试，也不静默丢数据。
*/
private void store(TDeviceIngestRecord value, Counters counters)
{
value.setContentSha256(canonicalHash(value.getPayloadJson()));
int inserted = records.insertIngestRecord(value);
if (inserted > 0)
{
counters.accepted++;
return;
}
String existing = records.selectIngestRecordHashByKey(value.getAdapterCode(), value.getPayloadKind(),
value.getRecordKey());
if (existing == null)
throw payloadInvalid("落地行未能写入且查不到既有行，原因不明：" + value.getRecordKey());
if (!existing.equalsIgnoreCase(HexFormat.of().formatHex(value.getContentSha256())))
{
counters.conflicts++;
log.warn("落地记录内容冲突，拒绝覆盖 adapterCode={}, payloadKind={}, recordKey={}",
value.getAdapterCode(), value.getPayloadKind(), value.getRecordKey());
return;
}
counters.ignored++;
}
private void touchHeartbeat(String adapterCode, Instant receivedAt)
{
TDeviceIngestRecord heartbeat = base(adapterCode, null, TDeviceIngestRecord.KIND_HEARTBEAT, 0, receivedAt);
heartbeat.setRecordKey(HEARTBEAT_KEY);
heartbeat.setPayloadJson("{}");
heartbeat.setContentSha256(canonicalHash(heartbeat.getPayloadJson()));
records.upsertHeartbeat(heartbeat);
}
/**
* 内容摘要按规范化后的键值算：程序重复推同一行时字段顺序不同也算同一条，
* 但任何一个值的改动都会被识别为冲突。
*/
private static byte[] canonicalHash(String payloadJson)
{
Map content = new TreeMap<>(JSON.parseObject(payloadJson));
try
{
return MessageDigest.getInstance("SHA-256").digest(JSON.toJSONBytes(content));
}
catch (Exception exception)
{
throw new IllegalStateException("无法计算落地记录内容摘要", exception);
}
}
// ------------------------------------------------------------------ 请求解析
private JSONObject parse(String body)
{
if (StringUtils.isBlank(body)) throw payloadInvalid("请求体为空");
JSONObject payload;
try
{
payload = JSON.parseObject(body);
}
catch (RuntimeException failure)
{
throw payloadInvalid("请求体不是合法的 JSON：" + failure.getMessage());
}
if (payload == null) throw payloadInvalid("请求体不是 JSON 对象");
return payload;
}
private JSONArray rows(JSONObject payload)
{
JSONArray rows = payload.getJSONArray("rows");
if (rows == null) throw payloadInvalid("请求体缺少 rows 数组");
if (rows.size() > MAX_ROWS_PER_REQUEST)
throw payloadInvalid("一批最多 " + MAX_ROWS_PER_REQUEST + " 行，本批 " + rows.size() + " 行");
return rows;
}
private JSONObject rowAt(JSONArray rows, int index)
{
JSONObject row = rows.getJSONObject(index);
if (row == null) throw payloadInvalid("第 " + index + " 行不是 JSON 对象");
return row;
}
/** 来源定位（库/表）取批次头：幂等键需要它，回放与排查也需要它。 */
private String sourceRef(JSONObject payload)
{
String database = payload.getString("database");
String table = payload.getString("table");
if (StringUtils.isBlank(database) || StringUtils.isBlank(table))
throw payloadInvalid("逐点数据批缺少 database 或 table：无法确定来源行身份");
return database.trim() + "/" + table.trim();
}
private long requiredSequence(JSONObject row)
{
Long sequence = row.getLong("Data_Point");
if (sequence == null) throw payloadInvalid("行缺少 Data_Point：" + row);
return sequence;
}
private String requiredText(JSONObject row, String key)
{
Object value = row.get(key);
if (value == null) throw payloadInvalid("行缺少 " + key + "：" + row);
String text = String.valueOf(value).trim();
if (text.isEmpty()) throw payloadInvalid("行的 " + key + " 为空：" + row);
return text;
}
/**
* Windows FILETIME（100 ns 刻度，自 1601-01-01 UTC）换算为 UTC 时刻。
* 用毫秒精度换算：落地表与实时记录的时间列都是 {@code DATETIME(3)}，多的精度存不下。
*/
private static Instant filetime(String ticks)
{
BigDecimal value = new BigDecimal(ticks);
long epochMillis = value.divide(BigDecimal.valueOf(10000), 0, RoundingMode.DOWN).longValue()
- FILETIME_EPOCH_DIFFERENCE_SECONDS * 1000L;
return Instant.ofEpochMilli(epochMillis);
}
private static String option(String sourceConfig, String key, String fallback)
{
if (StringUtils.isBlank(sourceConfig)) return fallback;
try
{
JSONObject config = JSON.parseObject(sourceConfig);
String value = config == null ? null : config.getString(key);
return StringUtils.isBlank(value) ? fallback : value.trim();
}
catch (RuntimeException ignored)
{
return fallback;
}
}
private static ServiceException payloadInvalid(String message)
{
return new ServiceException(message, DeviceIngestErrorCodes.PAYLOAD_INVALID);
}
private static final class Counters
{
private int accepted;
private int ignored;
private int conflicts;
}
}