package com.battery.business.service.impl;
|
|
import java.math.BigDecimal;
|
import java.math.RoundingMode;
|
import java.nio.charset.StandardCharsets;
|
import java.security.MessageDigest;
|
import java.time.Clock;
|
import java.time.Instant;
|
import java.util.Date;
|
import java.util.HexFormat;
|
import java.util.Map;
|
import java.util.TreeMap;
|
import java.util.function.Function;
|
import org.slf4j.Logger;
|
import org.slf4j.LoggerFactory;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.stereotype.Service;
|
import org.springframework.transaction.annotation.Transactional;
|
import com.alibaba.fastjson2.JSON;
|
import com.alibaba.fastjson2.JSONArray;
|
import com.alibaba.fastjson2.JSONObject;
|
import com.ruoyi.common.exception.ServiceException;
|
import com.ruoyi.common.utils.StringUtils;
|
import com.battery.device.api.credentials.LocalCredentials;
|
import com.battery.business.domain.TDeviceIngestRecord;
|
import com.battery.business.domain.TDeviceAdapter;
|
import com.battery.business.mapper.TDeviceAdapterMapper;
|
import com.battery.business.mapper.TDeviceIngestRecordMapper;
|
import com.battery.business.service.DeviceAdapterErrorCodes;
|
import com.battery.business.service.DeviceIngestErrorCodes;
|
import com.battery.business.service.IDeviceIngestService;
|
|
/**
|
* 推式来源的接收端点实现（ADR-0065）：校验来源凭据，把厂商原始行落进落地表。
|
*
|
* <p>这里也是**唯一**解释厂商列名的地方：原始行里有 22 所 IV 表的列（{@code record_time}、
|
* {@code Data_Point}、{@code Voltage}、{@code eCurrent} 等），本类把它们映射成落地表的身份、时间、
|
* 协议序号与测量列，同一份原始行原样存进 {@code payload_json}（键值不变，格式由 JSON 列规范化）。适配器因此只做投影、不认识厂商列名。
|
*
|
* <p>三条判定口径：
|
* <ul>
|
* <li><b>凭据不入库</b>：端点令牌与厂商库口令共用同一套取值顺序（本地凭据文件 → 环境变量 →
|
* 启动参数），实例配置里只有变量名。令牌缺失或不匹配时拒绝，且不落任何数据。</li>
|
* <li><b>时间换算</b>：22 所的 {@code record_time} 是 Windows FILETIME（100 ns 刻度，自 1601-01-01 UTC），
|
* 直接换算出 UTC 时刻；原始字符串同时保留。厂商给的是绝对时间，不需要按时区再解释一次。</li>
|
* <li><b>幂等与冲突</b>：来源行身份是「库/表 + Data_Point」，重复推送被唯一键幂等忽略；
|
* 同身份但内容不同只记警告、不覆盖既有事实。</li>
|
* </ul>
|
*/
|
@Service
|
public class DeviceIngestServiceImpl implements IDeviceIngestService
|
{
|
private static final Logger log = LoggerFactory.getLogger(DeviceIngestServiceImpl.class);
|
/** 端点令牌的变量名默认值；实例配置用 {@code tokenEnv} 覆盖它。 */
|
private static final String TOKEN_ENV_OPTION = "tokenEnv";
|
private static final String DEFAULT_TOKEN_ENV = "BATTERY_INGEST_TOKEN";
|
/** 心跳行按来源唯一，键名固定。 */
|
private static final String HEARTBEAT_KEY = "heartbeat";
|
/** 一批的上限：程序按批推，一次推来整表说明配置有问题，直接拒绝而不是把它读进内存。 */
|
private static final int MAX_ROWS_PER_REQUEST = 2000;
|
/** Windows FILETIME 的起点（1601-01-01 UTC）与 Unix 纪元之间的秒数差。 */
|
private static final long FILETIME_EPOCH_DIFFERENCE_SECONDS = 11644473600L;
|
|
private final TDeviceAdapterMapper adapters;
|
private final TDeviceIngestRecordMapper records;
|
private final Clock clock;
|
/** 端点令牌的取值入口；测试注入固定值，生产走本地凭据文件 → 环境变量 → 启动参数。 */
|
private final Function<String, String> credentials;
|
|
@Autowired
|
public DeviceIngestServiceImpl(TDeviceAdapterMapper adapters, TDeviceIngestRecordMapper records)
|
{
|
this(adapters, records, Clock.systemUTC(), LocalCredentials::resolve);
|
}
|
|
/** 允许测试固定接收时间与凭据取值，生产运行仍使用 UTC 系统时钟与本地凭据文件。 */
|
public DeviceIngestServiceImpl(TDeviceAdapterMapper adapters, TDeviceIngestRecordMapper records, Clock clock,
|
Function<String, String> credentials)
|
{
|
this.adapters = adapters;
|
this.records = records;
|
this.clock = clock;
|
this.credentials = credentials;
|
}
|
|
/**
|
* 接收推式来源程序的一批载荷：校验来源与令牌，按载荷类别写入落地表，并刷新该来源的心跳。
|
*
|
* <p>前置条件：来源已登记为启用中的推式实例，端点令牌与本地凭据一致，载荷为合法 JSON 且类别已知。
|
* 整批在同一事务内落库，任一行失败则整批回滚；重复推送按来源行身份幂等忽略，内容不一致的行记为冲突
|
* 而不覆盖既有事实。心跳、拓扑与采样都会刷新心跳时间。</p>
|
*
|
* @param adapterCode 来源实例编码，取自接收端点路径
|
* @param token 端点令牌，取自请求头
|
* @param body 载荷正文，含 kind 与对应的行数据
|
* @return 本批的接收计数：成功写入、幂等忽略、内容冲突的行数
|
* @throws ServiceException 来源不是启用的推式实例、令牌不匹配、载荷格式非法或行数超出单批上限
|
*/
|
@Override
|
@Transactional(rollbackFor = Exception.class)
|
public IngestResult ingest(String adapterCode, String token, String body)
|
{
|
// STEP 1: 校验来源是启用中的推式实例，并校验端点令牌。
|
TDeviceAdapter adapter = requirePushSource(adapterCode);
|
requireToken(adapter, token);
|
// STEP 2: 解析载荷并按 kind 分流；未知类别直接拒绝，不按猜测处理。
|
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);
|
}
|
// STEP 3: 刷新心跳时间；每一批成功接收都是"程序还在推"的证据。
|
// 每一批成功接收都刷新心跳：收到采样行同样是"程序还在推"的证据，缺一次心跳不会误判离线。
|
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);
|
}
|
|
/**
|
* 拓扑上报：声明该来源当前有哪些通道。行里只有厂商自己的通道号，身份映射在本方法里完成。
|
* 拓扑是"这个通道还在"的长期声明，重复上报被幂等忽略（不多占空间，也不改写既有行）。
|
*
|
* <p>行里可以额外带 {@code channelNo}：来源自己软件里显示的通道号。
|
* 它只用于界面显示与登记（操作员要拿它跟厂商软件对上），不参与通道身份，所以缺了也不影响接收。
|
*/
|
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.setVendorChannelNo(row.containsKey("channelNo") ? row.getInteger("channelNo") : null);
|
value.setRecordKey("topology:" + channel);
|
value.setPayloadJson(JSON.toJSONString(row));
|
// 拓扑是声明不是设备事实：重复上报刷新声明内容（例如这次才带上的来源通道号），
|
// 不追加新行、也不按"内容不同即冲突"处理——声明本来就允许被刷新。
|
value.setContentSha256(canonicalHash(value.getPayloadJson()));
|
if (records.upsertTopology(value) > 0) counters.accepted++; else counters.ignored++;
|
}
|
}
|
|
/**
|
* 逐点数据行：一批来自同一张厂商表（同一测试同一通道），因此身份与来源定位取批次头上的声明。
|
* 行本身原样保存，只额外解析出水位与投影需要的字段。
|
*/
|
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;
|
}
|
|
/**
|
* 落一行并用来源行身份判幂等：撞键即视为重复推送，比较内容摘要后区分"同内容重复"与"同身份不同内容"。
|
* 内容不同只记警告、不覆盖——既有那一行是已经收到的事实。
|
*/
|
/**
|
* 落一行并用来源行身份判幂等：撞键即视为重复推送，比较内容摘要后区分"同内容重复"与"同身份不同内容"。
|
* 内容不同只记警告、不覆盖——既有那一行是已经收到的事实。
|
*
|
* <p>落库没有新增、又查不到这一行时**明确失败**，而不是当作重复忽略：`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<String, Object> 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;
|
}
|
}
|
