实现 22 所推式来源闭环：接收端点、落地表与读落地表的适配器

yeg

2026-09-28 ccf7fa9210887633944bd4d78f90463b82807e6e

实现 22 所推式来源闭环：接收端点、落地表与读落地表的适配器

按 ADR-0065 落地 PUSH_INGEST（22 所那台）：只读取数程序把厂商原始行推到
POST /device-ingest/{adapterCode}，后端解释后落进落地表 t_device_ingest_record，
适配器再从落地表读回统一观测，与直连来源共用同一契约、接缝形状不改。

- 接收端点：按来源实例区分，请求头携带该来源令牌（与厂商库口令共用
  LocalCredentials 的取值顺序，不入库）；凭据错、来源未启用、来源类型不对或
  载荷不合法一律整批不落库，并按非 2xx 应答——取数程序只在 2xx 后推进水位。
- 落地表：一行 = 一条厂商原始行，按收到的键值原样保存原始行，另存解析出的
  身份、设备时间、协议序号与测量列。幂等键为「来源定位 + Data_Point」
  （22 所每个测试新建一个库，Data_Point 在每个测试内重新编号）；
  重复推送幂等忽略，同身份内容不同只报冲突、不覆盖。
- 适配器：候选来自拓扑上报并带各自最近数据时间，机器地址取实例 host
  （登记检测柜时带进 IP 字段）；心跳缺失或超过容忍窗口时发现明确失败，
  沿用既有路径把已登记设备置为离线、撤下旧快照，离线不伪造测量值。
  水位按 (设备时间, 协议序号) 只放行严格更晚的行。
- 取数程序 scripts/j22-collector.ps1（中文注释 + UTF-8 with BOM）：只读
  Windows 认证、固定 SELECT、不开入站端口、不传厂商库凭据；每条命令显式设
  超时（连接串与命令都设）、无加锁读与长事务、增量过滤不排序（排序在客户端
  做）、收到 2xx 才推进水位。
- 落地行未新增又查不到既有行时明确失败，而不是当作重复忽略：避免
  insert ignore 把"转换失败被跳过"变成静默丢数后仍然回 2xx。

后端全量测试 371 项通过（battery-business 18、ruoyi-admin 273 含 4 跳过）。

 battery-business/src/main/java/com/battery/business/domain/TDeviceIngestRecord.java

New file
@@ -0,0 +1,79 @@
package com.battery.business.domain;

import java.io.Serializable;
import java.math.BigDecimal;
import java.util.Date;

/**
 * 推式来源落地表的一行（ADR-0065）：厂商原始行按收到的键值原样保存在 {@code payloadJson}，
 * 同时带上接收端点解析出的身份、时间、协议序号与测量列。
 *
 * <p>解释只做一遍：厂商列名到这些字段的映射在接收端点里，适配器只把落地行投影成统一观测，
 * 因此适配器里不出现任何厂商列名。{@code recordKey} 是来源行身份，重复推送由它幂等忽略。
 */
public class TDeviceIngestRecord implements Serializable
{
    /** 载荷类别：拓扑上报，声明该来源当前有哪些通道。 */
    public static final String KIND_TOPOLOGY = "TOPOLOGY";
    /** 载荷类别：逐点数据行。 */
    public static final String KIND_SAMPLE = "SAMPLE";
    /** 载荷类别：存活心跳，只刷新"取数程序还在推"这一条事实。 */
    public static final String KIND_HEARTBEAT = "HEARTBEAT";

    private Long ingestId;
    private String adapterCode;
    private String payloadKind;
    private Integer devtype;
    private Integer devid;
    private Integer subdevid;
    private Integer chlid;
    private Long recordSequence;
    private String recordKey;
    private String sourceRef;
    private Date deviceTimeUtc;
    private String deviceTimeRaw;
    private String timezoneId;
    private Integer cycleNo;
    private Integer stepNo;
    private BigDecimal relativeTimeSeconds;
    private BigDecimal voltageV;
    private BigDecimal currentA;
    private BigDecimal chargeCapacityAh;
    private BigDecimal dischargeCapacityAh;
    private BigDecimal chargeEnergyWh;
    private BigDecimal dischargeEnergyWh;
    private String payloadJson;
    private byte[] contentSha256;
    private Date receivedAtUtc;

    public Long getIngestId() { return ingestId; } public void setIngestId(Long v) { ingestId = v; }
    public String getAdapterCode() { return adapterCode; } public void setAdapterCode(String v) { adapterCode = v; }
    public String getPayloadKind() { return payloadKind; } public void setPayloadKind(String v) { payloadKind = v; }
    public Integer getDevtype() { return devtype; } public void setDevtype(Integer v) { devtype = v; }
    public Integer getDevid() { return devid; } public void setDevid(Integer v) { devid = v; }
    public Integer getSubdevid() { return subdevid; } public void setSubdevid(Integer v) { subdevid = v; }
    public Integer getChlid() { return chlid; } public void setChlid(Integer v) { chlid = v; }
    public Long getRecordSequence() { return recordSequence; } public void setRecordSequence(Long v) { recordSequence = v; }
    public String getRecordKey() { return recordKey; } public void setRecordKey(String v) { recordKey = v; }
    public String getSourceRef() { return sourceRef; } public void setSourceRef(String v) { sourceRef = v; }
    public Date getDeviceTimeUtc() { return deviceTimeUtc; } public void setDeviceTimeUtc(Date v) { deviceTimeUtc = v; }
    public String getDeviceTimeRaw() { return deviceTimeRaw; } public void setDeviceTimeRaw(String v) { deviceTimeRaw = v; }
    public String getTimezoneId() { return timezoneId; } public void setTimezoneId(String v) { timezoneId = v; }
    public Integer getCycleNo() { return cycleNo; } public void setCycleNo(Integer v) { cycleNo = v; }
    public Integer getStepNo() { return stepNo; } public void setStepNo(Integer v) { stepNo = v; }
    public BigDecimal getRelativeTimeSeconds() { return relativeTimeSeconds; }
    public void setRelativeTimeSeconds(BigDecimal v) { relativeTimeSeconds = v; }
    public BigDecimal getVoltageV() { return voltageV; } public void setVoltageV(BigDecimal v) { voltageV = v; }
    public BigDecimal getCurrentA() { return currentA; } public void setCurrentA(BigDecimal v) { currentA = v; }
    public BigDecimal getChargeCapacityAh() { return chargeCapacityAh; }
    public void setChargeCapacityAh(BigDecimal v) { chargeCapacityAh = v; }
    public BigDecimal getDischargeCapacityAh() { return dischargeCapacityAh; }
    public void setDischargeCapacityAh(BigDecimal v) { dischargeCapacityAh = v; }
    public BigDecimal getChargeEnergyWh() { return chargeEnergyWh; }
    public void setChargeEnergyWh(BigDecimal v) { chargeEnergyWh = v; }
    public BigDecimal getDischargeEnergyWh() { return dischargeEnergyWh; }
    public void setDischargeEnergyWh(BigDecimal v) { dischargeEnergyWh = v; }
    public String getPayloadJson() { return payloadJson; } public void setPayloadJson(String v) { payloadJson = v; }
    public byte[] getContentSha256() { return contentSha256; } public void setContentSha256(byte[] v) { contentSha256 = v; }
    public Date getReceivedAtUtc() { return receivedAtUtc; } public void setReceivedAtUtc(Date v) { receivedAtUtc = v; }
}

 battery-business/src/main/java/com/battery/business/domain/vo/TDeviceIngestChannel.java

New file
@@ -0,0 +1,23 @@
package com.battery.business.domain.vo;

import java.io.Serializable;
import java.util.Date;

/**
 * 推式来源拓扑上报里的一条通道，附带它在落地表里最近的采样行时间。
 * 最近数据时间是一个观测事实，不是"在用"判定：登记哪些通道由操作员决定（ADR-0063）。
 */
public class TDeviceIngestChannel implements Serializable
{
    private Integer devtype;
    private Integer devid;
    private Integer subdevid;
    private Integer chlid;
    private Date lastDataAt;

    public Integer getDevtype() { return devtype; } public void setDevtype(Integer v) { devtype = v; }
    public Integer getDevid() { return devid; } public void setDevid(Integer v) { devid = v; }
    public Integer getSubdevid() { return subdevid; } public void setSubdevid(Integer v) { subdevid = v; }
    public Integer getChlid() { return chlid; } public void setChlid(Integer v) { chlid = v; }
    public Date getLastDataAt() { return lastDataAt; } public void setLastDataAt(Date v) { lastDataAt = v; }
}

 battery-business/src/main/java/com/battery/business/mapper/TDeviceIngestRecordMapper.java

New file
@@ -0,0 +1,42 @@
package com.battery.business.mapper;

import java.util.Date;
import java.util.List;
import org.apache.ibatis.annotations.Param;
import com.battery.business.domain.TDeviceIngestRecord;
import com.battery.business.domain.vo.TDeviceIngestChannel;

/**
 * 推式来源落地表的读写：接收端点写入厂商原始行，适配器读回统一观测（ADR-0065）。
 *
 * <p>写入一律用 insert ignore 与来源行身份唯一键：重复推送不改动既有事实，
 * 由调用方比较内容摘要判定"同内容重复"与"同身份不同内容"。
 */
public interface TDeviceIngestRecordMapper
{
    int insertIngestRecord(TDeviceIngestRecord record);

    /** 心跳行按来源唯一，用 upsert 覆盖接收时间：它承载"取数程序还在推"这一条事实。 */
    int upsertHeartbeat(TDeviceIngestRecord record);

    /** 返回该来源行已落库的内容摘要（小写十六进制）；没有这一行时为空。 */
    String selectIngestRecordHashByKey(@Param("adapterCode") String adapterCode,
            @Param("payloadKind") String payloadKind, @Param("recordKey") String recordKey);

    /** 该来源最近一次收到推送的时间；从未收到推送时为空。 */
    Date selectHeartbeatAt(@Param("adapterCode") String adapterCode);

    /** 拓扑上报声明的通道清单，每条附带其最近一次采样行的设备时间。 */
    List<TDeviceIngestChannel> selectTopologyChannels(@Param("adapterCode") String adapterCode);

    /** 某通道最近一次采样行；没有采样行时为空。水位与实时观测都取它。 */
    TDeviceIngestRecord selectLatestSample(@Param("adapterCode") String adapterCode,
            @Param("devtype") int devtype, @Param("devid") int devid, @Param("subdevid") int subdevid,
            @Param("chlid") int chlid);

    /** 严格晚于 (设备时间, 协议序号) 的采样行，按二元组升序；水位为空表示不取数据。 */
    List<TDeviceIngestRecord> selectSamplesSince(@Param("adapterCode") String adapterCode,
            @Param("devtype") int devtype, @Param("devid") int devid, @Param("subdevid") int subdevid,
            @Param("chlid") int chlid, @Param("watermarkTime") Date watermarkTime,
            @Param("watermarkSequence") long watermarkSequence, @Param("limit") int limit);
}

 battery-business/src/main/java/com/battery/business/service/DeviceIngestErrorCodes.java

New file
@@ -0,0 +1,20 @@
package com.battery.business.service;

/**
 * 推式来源接收端点（{@code POST /device-ingest/{adapterCode}}）使用的稳定业务错误码。
 * 每个码都对应一个明确的 HTTP 状态，取数程序据此决定是否推进自己的读取位置：
 * 只有 2xx 才算"对方收下了"（ADR-0065）。
 */
public final class DeviceIngestErrorCodes
{
    /** 来源凭据缺失或不匹配：拒绝并点名取值位置，不落任何数据。 */
    public static final int TOKEN_INVALID = 53101;
    /** 来源未启用（类型不是推式来源的情形归 {@code DeviceAdapterErrorCodes.UNSUPPORTED_TYPE}）：拒绝并说明状态。 */
    public static final int SOURCE_DISABLED = 53102;
    /** 请求体不符合推式来源的载荷契约，或落地行写不进去：点名出错的行与字段。 */
    public static final int PAYLOAD_INVALID = 53103;

    private DeviceIngestErrorCodes()
    {
    }
}

 battery-business/src/main/java/com/battery/business/service/IDeviceIngestService.java

New file
@@ -0,0 +1,22 @@
package com.battery.business.service;

/**
 * 推式来源的接收端点业务：按来源实例校验凭据，把厂商原始行落进落地表（ADR-0065）。
 *
 * <p>拒绝一律不留痕：凭据错误、来源未启用或载荷不合法时整批不落库，
 * 取数程序因此不会推进自己的读取位置，重试也不会产生半批数据。
 */
public interface IDeviceIngestService
{
    /**
     * 接收一批原始行。返回结果表示本批的落库情况，返回即代表"对方已收下"。
     *
     * @param adapterCode 来源实例编码（端点路径）
     * @param token       该来源的凭据（请求头）
     * @param body        请求体原文：厂商原始行，不做任何换算
     */
    IngestResult ingest(String adapterCode, String token, String body);

    /** 本批落库结果：新增、幂等忽略与同身份内容冲突的条数。 */
    record IngestResult(int accepted, int ignored, int conflicts) { }
}

 battery-business/src/main/java/com/battery/business/service/impl/DeviceAdapterServiceImpl.java

@@ -35,10 +35,18 @@
    private static final String DISABLED = "1";
    private static final String READ_ONLY_CAPABILITIES =
            "{\"discovery\":true,\"realtimeObservation\":true,\"historyDownload\":true,\"control\":false}";
    /** 适配器类型按存储形态划分；BTS 保留给开发模拟器与将来的协议路线（ADR-0065）。 */
    private static final List<String> SUPPORTED_TYPES = List.of("BTS", "NEWARE_MYSQL");
    /** 直连厂商存储的来源，其通道登记清单是长期事实，候选不按时间过期。 */
    private static final List<String> PERSISTENT_CANDIDATE_TYPES = List.of("NEWARE_MYSQL");
    /**
     * 适配器类型按存储形态划分；BTS 保留给开发模拟器与将来的协议路线（ADR-0065）。
     * `NEWARE_MYSQL` 只能写字面量：battery-business 不依赖 battery-device-neware，
     * 拿不到那边的常量；推式来源的实现就在本模块，所以用它的常量。
     */
    private static final List<String> SUPPORTED_TYPES = List.of("BTS", "NEWARE_MYSQL", PushIngestAdapter.ADAPTER_TYPE);
    /**
     * 通道清单长期存在的来源，候选不按时间过期：直连厂商存储的来源读的是厂商的登记表，
     * 推式来源读的是取数程序上报的拓扑——两者都不是"轮询发现"的结果。
     */
    private static final List<String> PERSISTENT_CANDIDATE_TYPES =
            List.of("NEWARE_MYSQL", PushIngestAdapter.ADAPTER_TYPE);

    private final TDeviceAdapterMapper mapper;
    private final DeviceAdapterRegistry registry;

 battery-business/src/main/java/com/battery/business/service/impl/DeviceIngestServiceImpl.java

New file
@@ -0,0 +1,386 @@
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;
    }

    @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;
    }

    /**
     * 落一行并用来源行身份判幂等：撞键即视为重复推送，比较内容摘要后区分"同内容重复"与"同身份不同内容"。
     * 内容不同只记警告、不覆盖——既有那一行是已经收到的事实。
     */
    /**
     * 落一行并用来源行身份判幂等：撞键即视为重复推送，比较内容摘要后区分"同内容重复"与"同身份不同内容"。
     * 内容不同只记警告、不覆盖——既有那一行是已经收到的事实。
     *
     * <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;
    }
}

 battery-business/src/main/java/com/battery/business/service/impl/PushIngestAdapter.java

New file
@@ -0,0 +1,284 @@
package com.battery.business.service.impl;

import java.math.BigDecimal;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import com.battery.device.api.DeviceAdapter;
import com.battery.device.api.DeviceAdapterException;
import com.battery.device.api.TestDataObservationAdapter;
import com.battery.device.api.model.AdapterCapabilities;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.device.api.model.DetectedChannel;
import com.battery.device.api.model.DetectedDeviceMetadata;
import com.battery.device.api.model.DeviceConnectionStatus;
import com.battery.device.api.model.DeviceWorkStatus;
import com.battery.device.api.model.RealtimeSnapshot;
import com.battery.device.api.registry.AdapterInstanceConfig;
import com.battery.business.domain.TDeviceIngestRecord;
import com.battery.business.domain.vo.TDeviceIngestChannel;
import com.battery.business.mapper.TDeviceIngestRecordMapper;

/**
 * 推式来源的观测实现（存储形态 {@code PUSH_INGEST}，ADR-0065）：读本项目的落地表，
 * 交出与直连来源完全相同的统一观测。
 *
 * <p>这类来源的机器没有可用的网络协议（22 所：TCP 与命名管道都未启用、Windows 账号未获提供），
 * 只能在它本机跑一个只读取数程序，把厂商原始行推到接收端点。程序本身很蠢——固定查询、原样推出、
 * 不开入站端口、不在网络上传递厂商库凭据；解释（身份映射、时间换算、幂等键取值）全部在后端完成，
 * 所以这里只做投影：落地行的字段由接收端点解析好，本类不出现任何厂商列名。
 *
 * <p>三条本形态特有的事实：
 * <ul>
 *   <li><b>来源是否在线由心跳判定</b>：落地表在本地随时可读，因此"能否读到表"不能表示来源可达。
 *       取数程序每轮推送一次心跳，心跳缺失即认为来源不可达——发现路径明确失败，
 *       已登记设备随之离线、旧快照被撤下（既有路径），页面不会显示伪造的实时值。</li>
 *   <li><b>最近数据时间只看采样行</b>：没有新采样行就是没有，不按拓扑上报或心跳时间代替（ADR-0063）。</li>
 *   <li><b>通道工作状态不可观测</b>：落地表里没有设备侧状态，报"无法观测"而不由数据行推断。</li>
 * </ul>
 */
public final class PushIngestAdapter implements DeviceAdapter, TestDataObservationAdapter
{
    /** 适配器类型按存储形态划分：这个形态读本项目的落地表（ADR-0065）。 */
    public static final String ADAPTER_TYPE = "PUSH_INGEST";
    /**
     * 身份映射（后端完成）：22 所的存储形态没有设备类型与单元维度，一台电脑就是登记意义上的一个柜，
     * 通道号就是厂商自己的通道号（表名 {@code Ws<N>_IV_0} 里的 N）。厂商通道号进 {@code chlid}，
     * {@code subdevid} 固定为 1，使界面通道号显示为「1-26」而不是「0-26」。
     * 若甲方要求按接口再分成几个柜，那是新增通道→接口映射的新决策，不在本单。
     *
     * <p>接收端点写落地行时用的是同一组常量（{@code DeviceIngestServiceImpl.base}），
     * 使"身份映射只在这个形态里定义一次"成立。
     */
    public static final int DEVICE_TYPE = 0;
    public static final int DEVICE_ID = 0;
    public static final int SUB_DEVICE_ID = 1;
    /**
     * 落地表在本地，读它不产生厂商负载，所以按展示节奏取数（控制台 5 秒一轮）；
     * 只取水位之后的新行，取数更勤不会产生重复记录。厂商写库节奏仍是 60 秒，值不会来得更快。
     */
    private static final long PERSISTENCE_INTERVAL_MS = 5000L;
    private static final int SAMPLE_BATCH_LIMIT = 500;
    /** 心跳容忍窗口的下限：写库与推送都是 60 秒一轮，窗口至少覆盖三轮。 */
    private static final Duration MINIMUM_STALE_AFTER = Duration.ofSeconds(90);

    private final AdapterInstanceConfig config;
    private final TDeviceIngestRecordMapper records;
    private final Clock clock;

    public PushIngestAdapter(AdapterInstanceConfig config, TDeviceIngestRecordMapper records)
    {
        this(config, records, Clock.systemUTC());
    }

    public PushIngestAdapter(AdapterInstanceConfig config, TDeviceIngestRecordMapper records, Clock clock)
    {
        this.config = config;
        this.records = records;
        this.clock = clock;
    }

    @Override
    public String adapterCode()
    {
        return config.adapterCode();
    }

    @Override
    public AdapterCapabilities capabilities()
    {
        return config.capabilities();
    }

    /**
     * 候选来自取数程序的拓扑上报：每条候选附带落地表里该通道最近的采样行时间。
     * 不自动判定"在用"，登记哪些通道由操作员决定（ADR-0063）。
     *
     * <p>候选的机器地址取实例配置的 {@code host}：这个形态里它表示**取数程序所在那台机器**，
     * 登记检测柜时会被带进 IP 字段（推送目标地址是取数程序自己的配置项，不在这边）。
     *
     * <p>心跳缺失时明确失败（而不是返回空清单）：来源不可达是一个可见的失败事实，
     * 业务侧据此把已登记设备置为离线并撤下旧快照。
     */
    @Override
    public List<DetectedChannel> discoverChannels()
    {
        requireFreshHeartbeat();
        List<DetectedChannel> channels = new ArrayList<>();
        List<TDeviceIngestChannel> reported = records.selectTopologyChannels(config.adapterCode());
        for (TDeviceIngestChannel channel : reported == null ? List.<TDeviceIngestChannel>of() : reported)
        {
            Instant lastDataAt = channel.getLastDataAt() == null ? null : channel.getLastDataAt().toInstant();
            channels.add(new DetectedChannel(identity(channel.getDevid(), channel.getSubdevid(), channel.getChlid()),
                    config.host(), true, DetectedDeviceMetadata.empty(), lastDataAt));
        }
        return List.copyOf(channels);
    }

    /** 实时快照取自该通道最近一次采样行；没有采样行时不返回该通道，避免沿用过期观测。 */
    @Override
    public Map<ChannelIdentity, RealtimeSnapshot> inquire(Set<ChannelIdentity> channels)
    {
        if (channels == null || channels.isEmpty()) return Map.of();
        Map<ChannelIdentity, RealtimeSnapshot> result = new HashMap<>();
        for (ChannelIdentity channel : channels)
        {
            CabinetRealtime latest = latestCabinet(channel);
            if (latest == null) continue;
            result.put(channel, snapshot(channel, latest));
        }
        return Map.copyOf(result);
    }

    @Override
    public CabinetRealtime latestCabinet(ChannelIdentity channel)
    {
        TDeviceIngestRecord row = latestSample(channel);
        return row == null ? null : cabinetRealtime(row);
    }

    /**
     * 22 所的 IV 表里有温箱温度、压力与计数器，是现场唯一见到环境量的来源，
     * 但类目语义与量纲尚未确认。按 ADR-0064 逐来源确认可得性，确认之前不交出测量值。
     */
    @Override
    public CollectorRealtime latestCollector(String adapterCode, String collectorCode)
    {
        return null;
    }

    /**
     * 22 所没有独立的工步层：工步与循环号只在采样行的 {@code Step_Index} / {@code Cycle_Index} 里，
     * 从采样推导工步是另一项决策（与"工步来自设备工步层"的既有口径冲突），不在本单范围，因此返回空。
     */
    @Override
    public List<StepRecord> stepRecords(ChannelIdentity channel)
    {
        return List.of();
    }

    /** 同上：循环统计不从采样推导，本单不交出循环记录。 */
    @Override
    public List<CycleRecord> cycleRecords(ChannelIdentity channel)
    {
        return List.of();
    }

    @Override
    public long persistenceIntervalMillis(ChannelIdentity channel)
    {
        return PERSISTENCE_INTERVAL_MS;
    }

    /** 水位是落地表里该通道已存在的最新一行：设备绝对时间与协议序号。 */
    @Override
    public SampleWatermark currentWatermark(ChannelIdentity channel)
    {
        TDeviceIngestRecord row = latestSample(channel);
        return row == null || row.getDeviceTimeUtc() == null
                ? new SampleWatermark(null, 0L)
                : new SampleWatermark(row.getDeviceTimeUtc().toInstant(), row.getRecordSequence());
    }

    /**
     * 严格晚于水位的落地行。按 {@code (设备时间, 协议序号)} 过滤与排序：同一绝对秒内的多行各自成立，
     * 不按秒合并。水位为空表示调用方尚未取得水位，返回空列表而不把来源里的历史并入本次接收。
     *
     * <p>这里不判心跳：落地行是"我们已经收到并解释过的设备事实"，
     * 来源离线只影响发现与实时观测，不影响这些已经收到的行。
     */
    @Override
    public List<CabinetRealtime> samplesSince(ChannelIdentity channel, SampleWatermark watermark)
    {
        if (watermark == null) return List.of();
        ChannelIdentity key = requireOwnChannel(channel);
        java.util.Date watermarkTime = watermark.deviceTime() == null ? null : java.util.Date.from(watermark.deviceTime());
        List<TDeviceIngestRecord> rows = records.selectSamplesSince(config.adapterCode(), key.devtype(), key.devid(),
                key.subdevid(), key.chlid(), watermarkTime, watermark.sequenceId(), SAMPLE_BATCH_LIMIT);
        List<CabinetRealtime> samples = new ArrayList<>();
        for (TDeviceIngestRecord row : rows == null ? List.<TDeviceIngestRecord>of() : rows) samples.add(cabinetRealtime(row));
        return List.copyOf(samples);
    }

    // ------------------------------------------------------------------ 投影与判定

    /** 心跳缺失即来源不可达：取数程序停了、或它自己连不上厂商库。 */
    private void requireFreshHeartbeat()
    {
        java.util.Date receivedAt = records.selectHeartbeatAt(config.adapterCode());
        if (receivedAt == null)
            throw new DeviceAdapterException("SOURCE_OFFLINE",
                    "尚未收到取数程序的推送：请确认程序已启动并指向 " + config.host() + ":" + config.port(), null);
        Duration staleAfter = staleAfter();
        Duration silent = Duration.between(receivedAt.toInstant(), clock.instant());
        if (silent.compareTo(staleAfter) > 0)
            throw new DeviceAdapterException("SOURCE_OFFLINE", "取数程序已停止推送：最近一次推送在 "
                    + silent.toSeconds() + " 秒前，超过容忍窗口 " + staleAfter.toSeconds() + " 秒", null);
    }

    /** 容忍窗口按实例配置的轮询周期算：写库与推送都是这个节奏，连丢三轮即认为离线。 */
    private Duration staleAfter()
    {
        Duration window = Duration.ofMillis(config.pollIntervalMs()).multipliedBy(3);
        return window.compareTo(MINIMUM_STALE_AFTER) < 0 ? MINIMUM_STALE_AFTER : window;
    }

    private TDeviceIngestRecord latestSample(ChannelIdentity channel)
    {
        ChannelIdentity key = requireOwnChannel(channel);
        return records.selectLatestSample(config.adapterCode(), key.devtype(), key.devid(), key.subdevid(),
                key.chlid());
    }

    private ChannelIdentity requireOwnChannel(ChannelIdentity channel)
    {
        if (channel == null || !config.adapterCode().equals(channel.adapterCode()))
            throw new DeviceAdapterException("通道身份不属于本实例："
                    + (channel == null ? "空" : channel.adapterCode()));
        return channel;
    }

    private CabinetRealtime cabinetRealtime(TDeviceIngestRecord row)
    {
        // 充放电容量按电流方向取对应的一组：与直连来源同一口径（22 所的电流符号约定待现场对着上位机确认）。
        BigDecimal current = row.getCurrentA() == null ? BigDecimal.ZERO : row.getCurrentA();
        boolean charging = current.signum() >= 0;
        BigDecimal capacity = charging ? row.getChargeCapacityAh() : row.getDischargeCapacityAh();
        BigDecimal energy = charging ? row.getChargeEnergyWh() : row.getDischargeEnergyWh();
        return new CabinetRealtime(row.getRecordSequence() == null ? 0L : row.getRecordSequence(),
                row.getDeviceTimeUtc() == null ? null : row.getDeviceTimeUtc().toInstant(),
                current.doubleValue(), decimal(row.getVoltageV()), decimal(capacity), decimal(energy),
                row.getCycleNo() == null ? 0 : row.getCycleNo(), row.getStepNo() == null ? 0 : row.getStepNo(),
                null, null, row.getReceivedAtUtc() == null ? null : row.getReceivedAtUtc().toInstant(),
                row.getDeviceTimeRaw(), row.getTimezoneId(),
                row.getRelativeTimeSeconds() == null ? null : row.getRelativeTimeSeconds().longValue(),
                "push:" + row.getRecordKey());
    }

    private RealtimeSnapshot snapshot(ChannelIdentity channel, CabinetRealtime latest)
    {
        // 读落地表路线下通道工作状态不可观测（ADR-0063）：不按数据行推断，也不沿用上一次状态。
        return new RealtimeSnapshot(channel, DeviceWorkStatus.UNOBSERVABLE, latest.cycleNo(), latest.stepNo(),
                latest.stepType(), null, BigDecimal.valueOf(latest.currentA()), BigDecimal.valueOf(latest.voltageV()),
                BigDecimal.valueOf(latest.capacityAh()), BigDecimal.valueOf(latest.energyWh()), null,
                latest.relativeTimeSeconds(), null, latest.serverReceivedAt(), Map.of(),
                DeviceConnectionStatus.ONLINE, latest.deviceTime(), List.of(), null);
    }

    private ChannelIdentity identity(Integer devid, Integer subdevid, Integer chlid)
    {
        return new ChannelIdentity(config.adapterCode(), DEVICE_TYPE,
                devid == null ? DEVICE_ID : devid, subdevid == null ? SUB_DEVICE_ID : subdevid,
                chlid == null ? 0 : chlid);
    }

    private static double decimal(BigDecimal value)
    {
        return value == null ? 0d : value.doubleValue();
    }
}

 battery-business/src/main/java/com/battery/business/service/impl/PushIngestAdapterFactory.java

New file
@@ -0,0 +1,34 @@
package com.battery.business.service.impl;

import com.battery.device.api.DeviceAdapter;
import com.battery.device.api.registry.AdapterInstanceConfig;
import com.battery.device.api.registry.DeviceAdapterFactory;
import com.battery.business.mapper.TDeviceIngestRecordMapper;

/**
 * 按存储形态 {@code PUSH_INGEST} 构造读落地表的适配器实例（ADR-0065）。
 *
 * <p>这个形态没有厂商连接：落地表就在业务库里，读写都走业务数据源与 Mapper，
 * 所以工厂不需要凭据，装配不会因为"忘了放口令"失败。凭据（接收端点令牌）只在接收端点上校验。
 */
public final class PushIngestAdapterFactory implements DeviceAdapterFactory
{
    private final TDeviceIngestRecordMapper records;

    public PushIngestAdapterFactory(TDeviceIngestRecordMapper records)
    {
        this.records = records;
    }

    @Override
    public String adapterType()
    {
        return PushIngestAdapter.ADAPTER_TYPE;
    }

    @Override
    public DeviceAdapter create(AdapterInstanceConfig config)
    {
        return new PushIngestAdapter(config, records);
    }
}

 battery-business/src/main/resources/mapper/battery/TDeviceIngestRecordMapper.xml

New file
@@ -0,0 +1,113 @@
<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
"http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.battery.business.mapper.TDeviceIngestRecordMapper">
    <resultMap id="TDeviceIngestRecordResultMap" type="com.battery.business.domain.TDeviceIngestRecord">
        <id property="ingestId" column="ingest_id"/>
        <result property="adapterCode" column="adapter_code"/>
        <result property="payloadKind" column="payload_kind"/>
        <result property="devtype" column="devtype"/>
        <result property="devid" column="devid"/>
        <result property="subdevid" column="subdevid"/>
        <result property="chlid" column="chlid"/>
        <result property="recordSequence" column="record_sequence"/>
        <result property="recordKey" column="record_key"/>
        <result property="sourceRef" column="source_ref"/>
        <result property="deviceTimeUtc" column="device_time_utc"/>
        <result property="deviceTimeRaw" column="device_time_raw"/>
        <result property="timezoneId" column="timezone_id"/>
        <result property="cycleNo" column="cycle_no"/>
        <result property="stepNo" column="step_no"/>
        <result property="relativeTimeSeconds" column="relative_time_seconds"/>
        <result property="voltageV" column="voltage_v"/>
        <result property="currentA" column="current_a"/>
        <result property="chargeCapacityAh" column="charge_capacity_ah"/>
        <result property="dischargeCapacityAh" column="discharge_capacity_ah"/>
        <result property="chargeEnergyWh" column="charge_energy_wh"/>
        <result property="dischargeEnergyWh" column="discharge_energy_wh"/>
        <result property="payloadJson" column="payload_json"/>
        <result property="contentSha256" column="content_sha256"/>
        <result property="receivedAtUtc" column="received_at_utc"/>
    </resultMap>

    <!-- 驼峰映射未开启（见 mybatis-config.xml），列名带下划线的查询一律显式映射。 -->
    <resultMap id="TDeviceIngestChannelResultMap" type="com.battery.business.domain.vo.TDeviceIngestChannel">
        <result property="devtype" column="devtype"/>
        <result property="devid" column="devid"/>
        <result property="subdevid" column="subdevid"/>
        <result property="chlid" column="chlid"/>
        <result property="lastDataAt" column="last_data_at"/>
    </resultMap>

    <!-- 落地行只追加：来源行身份（adapter_code + 类别 + record_key）撞键即视为重复推送，不改动既有事实。 -->
    <insert id="insertIngestRecord" parameterType="com.battery.business.domain.TDeviceIngestRecord"
            useGeneratedKeys="true" keyProperty="ingestId">
        insert ignore into t_device_ingest_record
        (adapter_code, payload_kind, devtype, devid, subdevid, chlid, record_sequence, record_key, source_ref,
         device_time_utc, device_time_raw, timezone_id, cycle_no, step_no, relative_time_seconds,
         voltage_v, current_a, charge_capacity_ah, discharge_capacity_ah, charge_energy_wh, discharge_energy_wh,
         payload_json, content_sha256, received_at_utc)
        values (#{adapterCode}, #{payloadKind}, #{devtype}, #{devid}, #{subdevid}, #{chlid}, #{recordSequence},
         #{recordKey}, #{sourceRef}, #{deviceTimeUtc}, #{deviceTimeRaw}, #{timezoneId}, #{cycleNo}, #{stepNo},
         #{relativeTimeSeconds}, #{voltageV}, #{currentA}, #{chargeCapacityAh}, #{dischargeCapacityAh},
         #{chargeEnergyWh}, #{dischargeEnergyWh}, cast(#{payloadJson} as json), #{contentSha256}, #{receivedAtUtc})
    </insert>

    <!-- 心跳单独占一行并覆盖接收时间：它只表示"取数程序还在推"，不是设备事实，没有保留历史的意义。 -->
    <insert id="upsertHeartbeat" parameterType="com.battery.business.domain.TDeviceIngestRecord">
        insert into t_device_ingest_record
        (adapter_code, payload_kind, devtype, devid, subdevid, chlid, record_sequence, record_key, source_ref,
         device_time_utc, device_time_raw, timezone_id, cycle_no, step_no, relative_time_seconds,
         voltage_v, current_a, charge_capacity_ah, discharge_capacity_ah, charge_energy_wh, discharge_energy_wh,
         payload_json, content_sha256, received_at_utc)
        values (#{adapterCode}, #{payloadKind}, #{devtype}, #{devid}, #{subdevid}, #{chlid}, #{recordSequence},
         #{recordKey}, #{sourceRef}, #{deviceTimeUtc}, #{deviceTimeRaw}, #{timezoneId}, #{cycleNo}, #{stepNo},
         #{relativeTimeSeconds}, #{voltageV}, #{currentA}, #{chargeCapacityAh}, #{dischargeCapacityAh},
         #{chargeEnergyWh}, #{dischargeEnergyWh}, cast(#{payloadJson} as json), #{contentSha256}, #{receivedAtUtc})
        on duplicate key update received_at_utc = values(received_at_utc)
    </insert>

    <select id="selectIngestRecordHashByKey" resultType="string">
        select lower(hex(content_sha256)) from t_device_ingest_record
        where adapter_code = #{adapterCode} and payload_kind = #{payloadKind} and record_key = #{recordKey}
        limit 1
    </select>

    <select id="selectHeartbeatAt" resultType="java.util.Date">
        select max(received_at_utc) from t_device_ingest_record
        where adapter_code = #{adapterCode} and payload_kind = 'HEARTBEAT'
    </select>

    <!-- 最近数据时间取该通道自己的采样行：没有新采样行就是没有，不按拓扑上报时间代替。 -->
    <select id="selectTopologyChannels" resultMap="TDeviceIngestChannelResultMap">
        select t.devtype, t.devid, t.subdevid, t.chlid,
        (select max(prior.device_time_utc) from t_device_ingest_record prior
         where prior.adapter_code = t.adapter_code and prior.payload_kind = 'SAMPLE'
           and prior.devtype = t.devtype and prior.devid = t.devid
           and prior.subdevid = t.subdevid and prior.chlid = t.chlid) as last_data_at
        from t_device_ingest_record t
        where t.adapter_code = #{adapterCode} and t.payload_kind = 'TOPOLOGY'
        order by t.devid, t.subdevid, t.chlid
    </select>

    <select id="selectLatestSample" resultMap="TDeviceIngestRecordResultMap">
        select * from t_device_ingest_record
        where adapter_code = #{adapterCode} and payload_kind = 'SAMPLE'
          and devtype = #{devtype} and devid = #{devid} and subdevid = #{subdevid} and chlid = #{chlid}
        order by device_time_utc desc, record_sequence desc
        limit 1
    </select>

    <!-- 只取严格晚于水位的行；同秒多行按协议序号各自成立，不按秒合并。 -->
    <select id="selectSamplesSince" resultMap="TDeviceIngestRecordResultMap">
        select * from t_device_ingest_record
        where adapter_code = #{adapterCode} and payload_kind = 'SAMPLE'
          and devtype = #{devtype} and devid = #{devid} and subdevid = #{subdevid} and chlid = #{chlid}
        <if test="watermarkTime != null">
          and (device_time_utc &gt; #{watermarkTime}
               or (device_time_utc = #{watermarkTime} and record_sequence &gt; #{watermarkSequence}))
        </if>
        order by device_time_utc, record_sequence
        limit #{limit}
    </select>
</mapper>

 battery-business/src/test/java/com/battery/business/service/impl/DeviceIngestServiceTest.java

New file
@@ -0,0 +1,201 @@
package com.battery.business.service.impl;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

import java.time.Clock;
import java.time.Instant;
import java.time.ZoneOffset;
import java.util.Date;
import org.junit.jupiter.api.Test;
import com.ruoyi.common.exception.ServiceException;
import com.battery.business.domain.TDeviceAdapter;
import com.battery.business.domain.TDeviceIngestRecord;
import com.battery.business.mapper.TDeviceAdapterMapper;
import com.battery.business.service.DeviceAdapterErrorCodes;
import com.battery.business.service.DeviceIngestErrorCodes;
import org.mockito.Mockito;

/**
 * 推式来源接收端点的判定口径：凭据、来源状态与载荷不合法时**整批不落库**（取数程序因此不推进水位），
 * 合法批次里厂商原始行原样落地、身份与时间由后端解析、重复推送幂等、同身份不同内容不覆盖。
 */
class DeviceIngestServiceTest
{
    private static final String ADAPTER = "j22";
    private static final String TOKEN = "j22-token";
    /** 现场实测的一行：FILETIME 134349692000000000 等于 2026-09-27 07:53:20 UTC（北京 15:53）。 */
    private static final String SAMPLE_BATCH = """
            {"kind":"SAMPLE","database":"DB_20260927_155209","table":"Ws26_IV_0","channel":26,"rows":[
              {"record_time":"134349692000000000","Data_Point":420,"Schedule_Index":2818575540,
               "Test_Time":600.0,"Step_Time":540.0,"Step_Index":4,"Cycle_Index":1,
               "Voltage":1.26778,"eCurrent":0.5,"ChargeCapacity":0.12,"DischargeCapacity":0.0,
               "ChargeEnergy":0.3,"DischargeEnergy":0.0,"IR":0.01,"Power":0.63,"Tempbox":25.4}]}
            """;

    private final FakeIngestLanding landing = new FakeIngestLanding();
    private final TDeviceAdapterMapper adapters = Mockito.mock(TDeviceAdapterMapper.class);
    private final Clock clock = Clock.fixed(Instant.parse("2026-09-27T09:00:00Z"), ZoneOffset.UTC);
    private final DeviceIngestServiceImpl service =
            new DeviceIngestServiceImpl(adapters, landing, clock, key -> TOKEN);

    @Test
    void unknownSourceIsRejectedWithoutWritingAnything()
    {
        var failure = assertThrows(ServiceException.class, () -> service.ingest("mystery", TOKEN, "{}"));

        assertEquals(DeviceAdapterErrorCodes.NOT_FOUND, failure.getCode());
        assertTrue(landing.rows().isEmpty());
    }

    @Test
    void sourceThatIsNotAPushSourceOrIsDisabledIsRejectedWithoutWritingAnything()
    {
        Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("NEWARE_MYSQL", "0"));
        var wrongType = assertThrows(ServiceException.class, () -> service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH));
        assertEquals(DeviceAdapterErrorCodes.UNSUPPORTED_TYPE, wrongType.getCode());

        Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("PUSH_INGEST", "1"));
        var disabled = assertThrows(ServiceException.class, () -> service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH));
        assertEquals(DeviceIngestErrorCodes.SOURCE_DISABLED, disabled.getCode());

        assertTrue(landing.rows().isEmpty());
    }

    /** 凭据不入库：实例配置里只有变量名，取值缺失同样按凭据错误拒绝（失败关闭）。 */
    @Test
    void wrongMissingOrUnresolvableTokenIsRejectedWithoutWritingAnything()
    {
        Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("PUSH_INGEST", "0"));

        assertEquals(DeviceIngestErrorCodes.TOKEN_INVALID,
                assertThrows(ServiceException.class, () -> service.ingest(ADAPTER, "wrong", SAMPLE_BATCH)).getCode());
        assertEquals(DeviceIngestErrorCodes.TOKEN_INVALID,
                assertThrows(ServiceException.class, () -> service.ingest(ADAPTER, null, SAMPLE_BATCH)).getCode());

        DeviceIngestServiceImpl withoutCredential = new DeviceIngestServiceImpl(adapters, landing, clock,
                key -> { throw new IllegalStateException("缺少来源凭据：" + key); });
        assertEquals(DeviceIngestErrorCodes.TOKEN_INVALID,
                assertThrows(ServiceException.class, () -> withoutCredential.ingest(ADAPTER, TOKEN, SAMPLE_BATCH))
                        .getCode());

        assertTrue(landing.rows().isEmpty());
    }

    /** 请求体承载厂商原始行：不含已换算的数值，单位换算、时间换算、身份映射与幂等键都在后端完成。 */
    @Test
    void rawVendorRowLandsWithResolvedIdentityTimeAndUnscaledMeasurements()
    {
        Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("PUSH_INGEST", "0"));

        var result = service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH);

        assertEquals(1, result.accepted());
        TDeviceIngestRecord row = landing.rows().get(0);
        // 身份映射：22 所没有设备类型与单元维度，一台电脑登记为一个柜，通道号就是厂商通道号。
        assertEquals(0, row.getDevtype());
        assertEquals(0, row.getDevid());
        assertEquals(1, row.getSubdevid());
        assertEquals(26, row.getChlid());
        // 时间换算：FILETIME → UTC，同时保留设备给出的原始时间文本。
        assertEquals(Instant.parse("2026-09-27T07:53:20Z"), row.getDeviceTimeUtc().toInstant());
        assertEquals("134349692000000000", row.getDeviceTimeRaw());
        // 幂等键：库/表 + Data_Point（22 所每个测试新建一个库，Data_Point 在每个测试内重新编号）。
        assertEquals(420L, row.getRecordSequence());
        assertEquals("DB_20260927_155209/Ws26_IV_0:420", row.getRecordKey());
        assertEquals("DB_20260927_155209/Ws26_IV_0", row.getSourceRef());
        // 测量值按物理单位直接落库，不做缩放：1.26778 V 就是 1.26778 V。
        assertEquals(0, new java.math.BigDecimal("1.26778").compareTo(row.getVoltageV()));
        assertEquals(0, new java.math.BigDecimal("0.5").compareTo(row.getCurrentA()));
        assertEquals(0, new java.math.BigDecimal("0.12").compareTo(row.getChargeCapacityAh()));
        assertEquals(1, row.getCycleNo());
        assertEquals(4, row.getStepNo());
        assertEquals(0, new java.math.BigDecimal("540.0").compareTo(row.getRelativeTimeSeconds()));
        // 原始行的键值原样保留，可供回放与核对。
        assertTrue(row.getPayloadJson().contains("Data_Point"));
        assertTrue(row.getPayloadJson().contains("\"Schedule_Index\":2818575540"));
    }

    @Test
    void repeatedPushIsIdempotentAndConflictingContentIsReportedWithoutOverwriting()
    {
        Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("PUSH_INGEST", "0"));

        assertEquals(1, service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH).accepted());
        // 同一批再推一次：同一条记录被幂等忽略，采样行仍是一行。
        var repeat = service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH);
        assertEquals(0, repeat.accepted());
        assertEquals(1, repeat.ignored());
        assertEquals(1, landing.samples(ADAPTER, 0, 0, 1, 26).size());

        // 同身份但内容不同：报冲突且不覆盖已有事实。
        var conflict = service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH.replace("1.26778", "1.99"));
        assertEquals(1, conflict.conflicts());
        assertEquals(0, conflict.accepted());
        assertEquals(1, landing.samples(ADAPTER, 0, 0, 1, 26).size());
        assertEquals(0, new java.math.BigDecimal("1.26778")
                .compareTo(landing.samples(ADAPTER, 0, 0, 1, 26).get(0).getVoltageV()));
    }

    @Test
    void topologyHeartbeatAndSamplesShareTheSameLandingTable()
    {
        Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("PUSH_INGEST", "0"));

        service.ingest(ADAPTER, TOKEN, "{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26},{\"channel\":18}]}");
        // 心跳只刷新存活事实，不产生数据行。
        var heartbeat = service.ingest(ADAPTER, TOKEN, "{\"kind\":\"HEARTBEAT\"}");
        assertEquals(0, heartbeat.accepted());
        assertEquals(3, landing.rows().size());
        assertEquals(clock.instant(), landing.selectHeartbeatAt(ADAPTER).toInstant());

        // 收到采样行同样是"程序还在推"的证据：接受请求就会刷新心跳，不依赖心跳消息本身。
        Date before = landing.selectHeartbeatAt(ADAPTER);
        assertNotNull(before);
        assertEquals(1, service.ingest(ADAPTER, TOKEN, SAMPLE_BATCH).accepted());
        assertEquals(4, landing.rows().size());
        assertEquals(before, landing.selectHeartbeatAt(ADAPTER));
        // 拓扑上报的通道清单是长期声明，重复上报幂等；采样行按事件逐行落库。
        assertEquals(2, landing.selectTopologyChannels(ADAPTER).size());
        assertEquals(1, landing.samples(ADAPTER, 0, 0, 1, 26).size());
    }

    @Test
    void malformedBatchIsRejectedWithTheOffendingRowNamed()
    {
        Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(adapter("PUSH_INGEST", "0"));

        for (String body : new String[] {
                "{\"kind\":\"SAMPLE\",\"database\":\"DB_x\",\"table\":\"Ws26_IV_0\",\"channel\":26}",
                "{\"kind\":\"SAMPLE\",\"database\":\"DB_x\",\"table\":\"Ws26_IV_0\",\"rows\":[]}",
                "{\"kind\":\"SAMPLE\",\"channel\":26,\"rows\":[]}",
                "{\"kind\":\"SAMPLE\",\"database\":\"DB_x\",\"table\":\"Ws26_IV_0\",\"channel\":26,"
                        + "\"rows\":[{\"Voltage\":1.2}]}",
                "{\"kind\":\"SAMPLE\",\"database\":\"DB_x\",\"table\":\"Ws26_IV_0\",\"channel\":26,"
                        + "\"rows\":[{\"Data_Point\":1}]}",
                "{\"kind\":\"SOMETHING_ELSE\"}",
                "not-json" })
        {
            var failure = assertThrows(ServiceException.class, () -> service.ingest(ADAPTER, TOKEN, body), body);
            assertEquals(DeviceIngestErrorCodes.PAYLOAD_INVALID, failure.getCode(), body);
        }
        assertTrue(landing.rows().isEmpty());
    }

    private static TDeviceAdapter adapter(String type, String status)
    {
        TDeviceAdapter adapter = new TDeviceAdapter();
        adapter.setAdapterId(7L);
        adapter.setAdapterCode(ADAPTER);
        adapter.setAdapterName("22 所那台");
        adapter.setAdapterType(type);
        adapter.setStatus(status);
        adapter.setHost("127.0.0.1");
        adapter.setPort(8080);
        adapter.setTimezoneId("Asia/Shanghai");
        adapter.setSourceConfig("{\"tokenEnv\":\"BATTERY_INGEST_TOKEN\"}");
        return adapter;
    }
}

 battery-business/src/test/java/com/battery/business/service/impl/FakeIngestLanding.java

New file
@@ -0,0 +1,130 @@
package com.battery.business.service.impl;

import java.util.ArrayList;
import java.util.Date;
import java.util.HexFormat;
import java.util.List;
import com.battery.business.domain.TDeviceIngestRecord;
import com.battery.business.domain.vo.TDeviceIngestChannel;
import com.battery.business.mapper.TDeviceIngestRecordMapper;

/**
 * 落地表的内存替身：按来源行身份判重、心跳按来源覆盖、水位按 (设备时间, 协议序号) 取严格更晚的行——
 * 与生产 Mapper 的 SQL 语义一致，使接收端点与适配器可以在没有数据库的情况下接起来测。
 */
final class FakeIngestLanding implements TDeviceIngestRecordMapper
{
    private final List<TDeviceIngestRecord> rows = new ArrayList<>();

    @Override
    public int insertIngestRecord(TDeviceIngestRecord record)
    {
        if (rows.stream().anyMatch(existing -> same(existing, record))) return 0;
        rows.add(record);
        return 1;
    }

    @Override
    public int upsertHeartbeat(TDeviceIngestRecord record)
    {
        rows.removeIf(existing -> same(existing, record));
        rows.add(record);
        return 1;
    }

    @Override
    public String selectIngestRecordHashByKey(String adapterCode, String payloadKind, String recordKey)
    {
        return rows.stream().filter(row -> same(row, adapterCode, payloadKind, recordKey)).findFirst()
                .map(row -> HexFormat.of().formatHex(row.getContentSha256())).orElse(null);
    }

    @Override
    public Date selectHeartbeatAt(String adapterCode)
    {
        return rows.stream()
                .filter(row -> adapterCode.equals(row.getAdapterCode())
                        && TDeviceIngestRecord.KIND_HEARTBEAT.equals(row.getPayloadKind()))
                .map(TDeviceIngestRecord::getReceivedAtUtc).max(Date::compareTo).orElse(null);
    }

    @Override
    public List<TDeviceIngestChannel> selectTopologyChannels(String adapterCode)
    {
        List<TDeviceIngestChannel> channels = new ArrayList<>();
        for (TDeviceIngestRecord row : rows)
        {
            if (!adapterCode.equals(row.getAdapterCode())
                    || !TDeviceIngestRecord.KIND_TOPOLOGY.equals(row.getPayloadKind())) continue;
            TDeviceIngestChannel channel = new TDeviceIngestChannel();
            channel.setDevtype(row.getDevtype()); channel.setDevid(row.getDevid());
            channel.setSubdevid(row.getSubdevid()); channel.setChlid(row.getChlid());
            channel.setLastDataAt(rows.stream().filter(sample -> isSample(sample, adapterCode, row))
                    .map(TDeviceIngestRecord::getDeviceTimeUtc).filter(java.util.Objects::nonNull)
                    .max(Date::compareTo).orElse(null));
            channels.add(channel);
        }
        channels.sort(java.util.Comparator.comparing(TDeviceIngestChannel::getDevid)
                .thenComparing(TDeviceIngestChannel::getSubdevid).thenComparing(TDeviceIngestChannel::getChlid));
        return channels;
    }

    @Override
    public TDeviceIngestRecord selectLatestSample(String adapterCode, int devtype, int devid, int subdevid,
            int chlid)
    {
        return samples(adapterCode, devtype, devid, subdevid, chlid).stream()
                .max(java.util.Comparator.comparing(TDeviceIngestRecord::getDeviceTimeUtc)
                        .thenComparing(TDeviceIngestRecord::getRecordSequence))
                .orElse(null);
    }

    @Override
    public List<TDeviceIngestRecord> selectSamplesSince(String adapterCode, int devtype, int devid, int subdevid,
            int chlid, Date watermarkTime, long watermarkSequence, int limit)
    {
        List<TDeviceIngestRecord> later = samples(adapterCode, devtype, devid, subdevid, chlid).stream()
                .filter(row -> after(row, watermarkTime, watermarkSequence))
                .sorted(java.util.Comparator.comparing(TDeviceIngestRecord::getDeviceTimeUtc)
                        .thenComparing(TDeviceIngestRecord::getRecordSequence))
                .toList();
        return later.size() > limit ? later.subList(0, limit) : later;
    }

    /** 全部落地行，供断言"拒绝时没有落任何数据"使用。 */
    List<TDeviceIngestRecord> rows() { return List.copyOf(rows); }

    List<TDeviceIngestRecord> samples(String adapterCode, int devtype, int devid, int subdevid, int chlid)
    {
        return rows.stream().filter(row -> adapterCode.equals(row.getAdapterCode())
                && TDeviceIngestRecord.KIND_SAMPLE.equals(row.getPayloadKind())
                && row.getDevtype() == devtype && row.getDevid() == devid && row.getSubdevid() == subdevid
                && row.getChlid() == chlid).toList();
    }

    private static boolean isSample(TDeviceIngestRecord row, String adapterCode, TDeviceIngestRecord channel)
    {
        return adapterCode.equals(row.getAdapterCode()) && TDeviceIngestRecord.KIND_SAMPLE.equals(row.getPayloadKind())
                && row.getDevtype().equals(channel.getDevtype()) && row.getDevid().equals(channel.getDevid())
                && row.getSubdevid().equals(channel.getSubdevid()) && row.getChlid().equals(channel.getChlid());
    }

    private static boolean after(TDeviceIngestRecord row, Date watermarkTime, long watermarkSequence)
    {
        if (row.getDeviceTimeUtc() == null) return false;
        if (watermarkTime == null) return true;
        int byTime = row.getDeviceTimeUtc().compareTo(watermarkTime);
        return byTime > 0 || (byTime == 0 && row.getRecordSequence() > watermarkSequence);
    }

    private static boolean same(TDeviceIngestRecord row, TDeviceIngestRecord other)
    {
        return same(row, other.getAdapterCode(), other.getPayloadKind(), other.getRecordKey());
    }

    private static boolean same(TDeviceIngestRecord row, String adapterCode, String payloadKind, String recordKey)
    {
        return adapterCode.equals(row.getAdapterCode()) && payloadKind.equals(row.getPayloadKind())
                && recordKey.equals(row.getRecordKey());
    }
}

 battery-business/src/test/java/com/battery/business/service/impl/PushIngestAdapterContractTest.java

New file
@@ -0,0 +1,215 @@
package com.battery.business.service.impl;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneOffset;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.junit.jupiter.api.Test;
import com.battery.device.api.DeviceAdapterException;
import com.battery.device.api.TestDataObservationAdapter.CabinetRealtime;
import com.battery.device.api.TestDataObservationAdapter.SampleWatermark;
import com.battery.device.api.model.AdapterCapabilities;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.device.api.model.DetectedChannel;
import com.battery.device.api.model.DeviceWorkStatus;
import com.battery.device.api.model.RealtimeSnapshot;
import com.battery.device.api.registry.AdapterInstanceConfig;
import com.battery.business.domain.TDeviceAdapter;
import com.battery.business.mapper.TDeviceAdapterMapper;
import org.mockito.Mockito;

/**
 * 推式来源适配器的观测契约：落地行投影成统一观测（身份、时间、单位、幂等身份），
 * 发现候选带最近数据时间且心跳缺失即明确失败，水位按 (设备时间, 协议序号) 取严格更晚的行。
 * 原始行由接收端点解析，所以这里只喂落地行，适配器里不应出现任何厂商列名。
 */
class PushIngestAdapterContractTest
{
    private static final String ADAPTER = "j22";
    private static final String TOKEN = "j22-token";
    /** 前两行同一个绝对秒（08:03:20 UTC）但协议序号不同；第三行晚一分钟。 */
    private static final String BATCH = """
            {"kind":"SAMPLE","database":"DB_20260927_155209","table":"Ws26_IV_0","channel":26,"rows":[
              {"record_time":"134349692000000000","Data_Point":420,"Step_Time":540.0,"Step_Index":4,
               "Cycle_Index":1,"Voltage":1.26778,"eCurrent":0.5,"ChargeCapacity":0.12,"ChargeEnergy":0.3},
              {"record_time":"134349698000000000","Data_Point":421,"Step_Time":600.0,"Step_Index":4,
               "Cycle_Index":1,"Voltage":1.2701,"eCurrent":0.5,"ChargeCapacity":0.121,"ChargeEnergy":0.31},
              {"record_time":"134349698000000000","Data_Point":422,"Step_Time":600.0,"Step_Index":5,
               "Cycle_Index":1,"Voltage":1.2702,"eCurrent":-0.4,"DischargeCapacity":0.05,"DischargeEnergy":0.1}]}
            """;

    private final FakeIngestLanding landing = new FakeIngestLanding();
    private final TDeviceAdapterMapper adapters = Mockito.mock(TDeviceAdapterMapper.class);
    private static final Instant NOW = Instant.parse("2026-09-27T09:00:00Z");

    @Test
    void topologyPushMakesChannelsCandidatesWithTheirOwnLatestDataTime()
    {
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26},{\"channel\":18}]}");
        ingest(BATCH);
        PushIngestAdapter adapter = adapter(NOW);

        List<DetectedChannel> channels = adapter.discoverChannels();

        assertEquals(List.of(18, 26), channels.stream().map(channel -> channel.identity().chlid()).toList());
        DetectedChannel active = channels.stream().filter(channel -> channel.identity().chlid() == 26).findFirst()
                .orElseThrow();
        // 最近数据时间来自该通道自己的采样行，不是拓扑上报时间。
        assertEquals(Instant.parse("2026-09-27T08:03:20Z"), active.lastDataAt());
        assertTrue(active.available());
        // 候选的机器地址取实例配置：登记时会被带进检测柜的 IP 字段。
        assertEquals("127.0.0.1", active.serverIp());
        // 登记范围由操作员决定：没有采样行的通道照样出现在候选里，只是没有最近数据时间。
        assertNull(channels.stream().filter(channel -> channel.identity().chlid() == 18).findFirst()
                .orElseThrow().lastDataAt());
    }

    /** 心跳缺失或过期即来源不可达：发现路径明确失败，业务侧据此把已登记设备置为离线并撤下旧快照。 */
    @Test
    void missingOrStaleHeartbeatFailsDiscoveryInsteadOfReturningStaleValues()
    {
        // 还没收到过任何推送：这是"取数程序没起来"这一条可见的失败，不是沉默的空清单。
        DeviceAdapterException missing = assertThrows(DeviceAdapterException.class, () -> adapter(NOW).discoverChannels());
        assertEquals("SOURCE_OFFLINE", missing.failureCode());

        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        // 收到推送即刷新心跳；容忍窗口是三个轮询周期，一轮没推不算离线。
        assertEquals(1, adapter(NOW.plus(Duration.ofSeconds(179))).discoverChannels().size());
        // 连丢三轮才算离线，且不必等候选过期。
        DeviceAdapterException stale = assertThrows(DeviceAdapterException.class,
                () -> adapter(NOW.plus(Duration.ofSeconds(181))).discoverChannels());
        assertEquals("SOURCE_OFFLINE", stale.failureCode());
    }

    @Test
    void samplesAreProjectedWithRawTimeUnscaledUnitsAndSourceRowIdentity()
    {
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        ingest(BATCH);
        PushIngestAdapter adapter = adapter(NOW);

        CabinetRealtime latest = adapter.latestCabinet(channel());

        // 时间按厂商格式换算为 UTC，同时保留原始值与来源时区；单位是物理单位，不做缩放。
        assertEquals(Instant.parse("2026-09-27T08:03:20Z"), latest.deviceTime());
        assertEquals("134349698000000000", latest.deviceTimeRaw());
        assertEquals("Asia/Shanghai", latest.timezoneId());
        assertEquals(1.2702, latest.voltageV(), 1e-9);
        assertEquals(-0.4, latest.currentA(), 1e-9);
        assertEquals(600L, latest.relativeTimeSeconds());
        assertEquals(1, latest.cycleNo());
        assertEquals(5, latest.stepNo());
        // 放电容量的取法按电流方向，与直连来源同一口径；来源行身份供幂等与冲突判定。
        assertEquals(0.05, latest.capacityAh(), 1e-9);
        assertEquals("push:DB_20260927_155209/Ws26_IV_0:422", latest.sourceRecordId());

        // 实时快照：通道工作状态在读落地表路线下不可观测（ADR-0063）。
        Map<ChannelIdentity, RealtimeSnapshot> snapshots = adapter.inquire(Set.of(channel()));
        assertEquals(DeviceWorkStatus.UNOBSERVABLE, snapshots.get(channel()).workStatus());
        assertEquals(Instant.parse("2026-09-27T08:03:20Z"), snapshots.get(channel()).deviceTime());
    }

    @Test
    void watermarkAndSamplesSinceKeepSameSecondRowsDistinctAndSkipNothingBeforeTheBoundary()
    {
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        ingest(BATCH);
        PushIngestAdapter adapter = adapter(NOW);

        // 水位是落地表里该通道最新一行：设备绝对时间与协议序号。
        assertEquals(new SampleWatermark(Instant.parse("2026-09-27T08:03:20Z"), 422L),
                adapter.currentWatermark(channel()));

        // 只取严格晚于水位的行：同一绝对秒内的多行靠协议序号各自成立。
        List<CabinetRealtime> rows = adapter.samplesSince(channel(),
                new SampleWatermark(Instant.parse("2026-09-27T07:53:20Z"), 420L));
        assertEquals(List.of(421L, 422L), rows.stream().map(CabinetRealtime::sequenceId).toList());
        assertEquals(0, adapter.samplesSince(channel(),
                new SampleWatermark(Instant.parse("2026-09-27T08:03:20Z"), 422L)).size());
        // 同一绝对秒内序号更大的行仍然成立：水位是二元组而不是秒。
        assertEquals(List.of(422L), adapter.samplesSince(channel(),
                new SampleWatermark(Instant.parse("2026-09-27T08:03:20Z"), 421L)).stream()
                .map(CabinetRealtime::sequenceId).toList());
        // 调用方尚未取得水位时不把来源里的历史并入本次接收。
        assertTrue(adapter.samplesSince(channel(), null).isEmpty());
        // 水位为空的二元组表示边界时刻来源里什么也没有：之后出现的行全部属于本次接收。
        assertEquals(3, adapter.samplesSince(channel(), new SampleWatermark(null, 0L)).size());
    }

    @Test
    void emptySourceHasNoWatermarkAndNoRealtimeObservation()
    {
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        PushIngestAdapter adapter = adapter(NOW);

        assertEquals(new SampleWatermark(null, 0L), adapter.currentWatermark(channel()));
        assertNull(adapter.latestCabinet(channel()));
        assertTrue(adapter.inquire(Set.of(channel())).isEmpty());
    }

    /** 采集器辅助测量按 ADR-0064 逐来源确认；22 所环境量的列语义未确认，因此不交出测量值。 */
    @Test
    void collectorAndStepCycleObservationsStayEmptyForThisStorageForm()
    {
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        ingest(BATCH);
        PushIngestAdapter adapter = adapter(NOW);

        assertNull(adapter.latestCollector(ADAPTER, "collector-1"));
        assertTrue(adapter.stepRecords(channel()).isEmpty());
        assertTrue(adapter.cycleRecords(channel()).isEmpty());
        // 落地表在本地，取数不产生厂商负载，所以按展示节奏取数；厂商写库节奏仍是 60 秒。
        assertEquals(5000L, adapter.persistenceIntervalMillis(channel()));
        AdapterCapabilities capabilities = adapter.capabilities();
        assertTrue(capabilities.discovery() && capabilities.realtimeObservation() && !capabilities.control());
    }

    @Test
    void channelIdentityFromAnotherInstanceFailsInsteadOfFallingBack()
    {
        PushIngestAdapter adapter = adapter(NOW);

        DeviceAdapterException failure = assertThrows(DeviceAdapterException.class,
                () -> adapter.currentWatermark(new ChannelIdentity("neware76", 21, 36, 1, 2)));
        assertTrue(failure.getMessage().contains("neware76"));
        assertFalse(adapter.adapterCode().isEmpty());
    }

    private ChannelIdentity channel()
    {
        return new ChannelIdentity(ADAPTER, 0, 0, 1, 26);
    }

    private PushIngestAdapter adapter(Instant now)
    {
        return new PushIngestAdapter(config(), landing, Clock.fixed(now, ZoneOffset.UTC));
    }

    private static AdapterInstanceConfig config()
    {
        return new AdapterInstanceConfig(ADAPTER, PushIngestAdapter.ADAPTER_TYPE, "127.0.0.1", 8080, 60000,
                "Asia/Shanghai", true, AdapterCapabilities.observeOnly(),
                Map.of("tokenEnv", "BATTERY_INGEST_TOKEN"));
    }

    private void ingest(String body)
    {
        TDeviceAdapter source = new TDeviceAdapter();
        source.setAdapterCode(ADAPTER);
        source.setAdapterType(PushIngestAdapter.ADAPTER_TYPE);
        source.setStatus("0");
        source.setTimezoneId("Asia/Shanghai");
        Mockito.when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(source);
        new DeviceIngestServiceImpl(adapters, landing, Clock.fixed(NOW, ZoneOffset.UTC), key -> TOKEN)
                .ingest(ADAPTER, TOKEN, body);
    }
}

 battery-business/src/test/java/com/battery/business/service/impl/PushIngestReceptionContractTest.java

New file
@@ -0,0 +1,180 @@
package com.battery.business.service.impl;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.util.Date;
import java.util.List;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import com.battery.device.api.model.AdapterCapabilities;
import com.battery.device.api.registry.AdapterInstanceConfig;
import com.battery.business.domain.TDeviceAdapter;
import com.battery.business.domain.TDeviceIngestRecord;
import com.battery.business.domain.TTest;
import com.battery.business.domain.TTestBattery;
import com.battery.business.domain.TTestChannelBinding;
import com.battery.business.domain.TTestRealtimeRecord;
import com.battery.business.mapper.TDeviceAdapterMapper;
import com.battery.business.mapper.TTestMapper;
import com.battery.business.service.ITTestService;

/**
 * 推式来源接进业务接收路径端到端：取数程序推来的厂商原始行 → 落地表 → 适配器读回统一观测 →
 * 接收中测试按水位逐行保存。
 *
 * <p>用的是同一个观测契约，所以这里不需要任何推送专属的接收代码：落地行按事件落库、
 * 轮询槽位为空（同秒多行因此各自成立），来源行身份承担幂等与冲突判定。
 */
class PushIngestReceptionContractTest
{
    private static final String ADAPTER = "j22";
    private static final Instant START = Instant.parse("2026-09-27T08:00:00Z");
    /** 现场实测的一批：两行落在同一个绝对秒（08:03:20 UTC），协议序号不同。 */
    private static final String BATCH = """
            {"kind":"SAMPLE","database":"DB_20260927_155209","table":"Ws26_IV_0","channel":26,"rows":[
              {"record_time":"134349698000000000","Data_Point":421,"Step_Time":600.0,"Step_Index":4,
               "Cycle_Index":1,"Voltage":1.2701,"eCurrent":0.5,"ChargeCapacity":0.121,"ChargeEnergy":0.31},
              {"record_time":"134349698000000000","Data_Point":422,"Step_Time":600.0,"Step_Index":4,
               "Cycle_Index":1,"Voltage":1.2702,"eCurrent":0.5,"ChargeCapacity":0.122,"ChargeEnergy":0.32}]}
            """;

    private final FakeIngestLanding landing = new FakeIngestLanding();
    private final TDeviceAdapterMapper adapters = Mockito.mock(TDeviceAdapterMapper.class);
    private final TestClock clock = new TestClock(START);

    @Test
    void pushedRawRowsBecomeTestFactsWithoutPollingSlots()
    {
        PushIngestAdapter adapter = ingestPushSource();
        TTestMapper mapper = mock(TTestMapper.class);
        ITTestService tests = mock(ITTestService.class);
        TTest test = receivingTest();
        when(mapper.selectReceivingTestIds()).thenReturn(List.of(1L));
        when(mapper.selectTestByIdForUpdate(1L)).thenReturn(test);
        when(tests.selectTestById(1L)).thenReturn(test);
        TestDataReceptionServiceImpl service = new TestDataReceptionServiceImpl(mapper, tests, adapter, clock);

        // 第一次接收只取水位：开始接收之前的行即使躺在落地表里也不进本测试。
        landing.insertIngestRecord(boundaryRow("DB_20260927_155209/Ws26_IV_0:420", 420L));
        service.captureReceivedData();
        verify(mapper, never()).insertRealtimeRecord(any());

        // 之后推来的行逐行落库：同秒两行各自成立，因为槽位为空（推送记录按事件落库，不占轮询槽位）。
        ingest(BATCH);
        clock.advance(Duration.ofSeconds(60));
        service.captureReceivedData();

        ArgumentCaptor<TTestRealtimeRecord> rows = ArgumentCaptor.forClass(TTestRealtimeRecord.class);
        verify(mapper, times(2)).insertRealtimeRecord(rows.capture());
        assertEquals(List.of("push:DB_20260927_155209/Ws26_IV_0:421",
                "push:DB_20260927_155209/Ws26_IV_0:422"),
                rows.getAllValues().stream().map(TTestRealtimeRecord::getSourceRecordId).toList());
        assertTrue(rows.getAllValues().stream().allMatch(row -> row.getCaptureSlotUtc() == null));
        assertEquals("j22:0:0:1:26", rows.getAllValues().get(0).getSourceIdentity());
        assertEquals(Instant.parse("2026-09-27T08:03:20Z"),
                rows.getAllValues().get(0).getCabinetDeviceTimeUtc().toInstant());
        assertEquals("Asia/Shanghai", rows.getAllValues().get(0).getCabinetAdapterTimezoneId());
        // 测量值按物理单位直接使用，不做缩放。
        assertEquals(0, new java.math.BigDecimal("1.2701")
                .compareTo(rows.getAllValues().get(0).getVoltageV()));
        assertEquals(0, new java.math.BigDecimal("0.121")
                .compareTo(rows.getAllValues().get(0).getCapacityAh()));

        // 同一批再推一次不会重复落库：来源行身份让接收幂等。
        ingest(BATCH);
        clock.advance(Duration.ofSeconds(60));
        service.captureReceivedData();
        verify(mapper, times(2)).insertRealtimeRecord(any());
    }

    private PushIngestAdapter ingestPushSource()
    {
        TDeviceAdapter source = new TDeviceAdapter();
        source.setAdapterCode(ADAPTER);
        source.setAdapterType(PushIngestAdapter.ADAPTER_TYPE);
        source.setStatus("0");
        source.setTimezoneId("Asia/Shanghai");
        source.setSourceConfig("{\"tokenEnv\":\"BATTERY_INGEST_TOKEN\"}");
        when(adapters.selectDeviceAdapterByCode(ADAPTER)).thenReturn(source);
        ingest("{\"kind\":\"TOPOLOGY\",\"rows\":[{\"channel\":26}]}");
        return new PushIngestAdapter(new AdapterInstanceConfig(ADAPTER, PushIngestAdapter.ADAPTER_TYPE,
                "127.0.0.1", 8080, 60000, "Asia/Shanghai", true, AdapterCapabilities.observeOnly(),
                java.util.Map.of("tokenEnv", "BATTERY_INGEST_TOKEN")), landing, clock);
    }

    private void ingest(String body)
    {
        new DeviceIngestServiceImpl(adapters, landing, clock, key -> "token").ingest(ADAPTER, "token", body);
    }

    /** 开始接收之前就躺在来源里的那一行：它只用来把水位定在边界时刻。 */
    private TDeviceIngestRecord boundaryRow(String recordKey, long sequence)
    {
        TDeviceIngestRecord value = new TDeviceIngestRecord();
        value.setAdapterCode(ADAPTER);
        value.setPayloadKind(TDeviceIngestRecord.KIND_SAMPLE);
        value.setDevtype(0); value.setDevid(0); value.setSubdevid(1); value.setChlid(26);
        value.setRecordSequence(sequence);
        value.setRecordKey(recordKey);
        value.setSourceRef("DB_20260927_155209/Ws26_IV_0");
        value.setDeviceTimeUtc(Date.from(START.minusSeconds(60)));
        value.setDeviceTimeRaw("134349691400000000");
        value.setTimezoneId("Asia/Shanghai");
        value.setPayloadJson("{}");
        value.setContentSha256(new byte[32]);
        value.setReceivedAtUtc(Date.from(START.minusSeconds(60)));
        return value;
    }

    private static TTest receivingTest()
    {
        TTestChannelBinding binding = new TTestChannelBinding();
        binding.setBindingId(2L);
        binding.setChannelId(3L);
        binding.setAdapterCode(ADAPTER);
        binding.setAdapterTimezoneId("Asia/Shanghai");
        binding.setDevtype(0);
        binding.setDevid(0);
        binding.setSubdevid(1);
        binding.setChlid(26);
        TTestBattery battery = new TTestBattery();
        battery.setTestBatteryId(4L);
        battery.setBinding(binding);
        battery.setCollectorChannels(List.of());
        TTest test = new TTest();
        test.setTestId(1L);
        test.setTestStatus("RUNNING");
        test.setReceptionStatus("RECEIVING");
        test.setStartedAtUtc(Date.from(START));
        test.setReceptionBoundaryAtUtc(Date.from(START));
        test.setBatteries(List.of(battery));
        return test;
    }

    /** 可推进的测试时钟：接收节奏与推送节奏都按它推进。 */
    private static final class TestClock extends Clock
    {
        private Instant now;

        private TestClock(Instant now) { this.now = now; }

        private void advance(Duration duration) { now = now.plus(duration); }

        @Override public ZoneId getZone() { return ZoneOffset.UTC; }
        @Override public Clock withZone(ZoneId zone) { return this; }
        @Override public Instant instant() { return now; }
    }
}

 battery-device-api/src/main/java/com/battery/device/api/credentials/LocalCredentials.java

New file
@@ -0,0 +1,71 @@
package com.battery.device.api.credentials;

import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Map;
import java.util.Properties;
import java.util.function.Function;
import com.battery.device.api.DeviceAdapterException;

/**
 * 来源凭据的取值顺序（ADR-0065）：进程工作目录下的 {@code config/vendor-credentials.properties}
 * → 同名环境变量 → 同名 JVM 启动参数。
 *
 * <p>凭据不进数据库、不进仓库，也不打包；仓库只提交 {@code .example} 模板。
 * 文件优先是为了让部署只做"复制模板填一行"这一件事；保留环境变量与启动参数两条，
 * 是因为交付形态可能只提供其中某一种。三者都没有时装配失败，错误信息同时点名文件路径与键名，
 * 使"忘了放凭据"是可见的失败而不是沉默的缺失。
 *
 * <p>直连厂商库的口令（{@code NEWARE_MYSQL}）与推式来源的接收端点令牌（{@code PUSH_INGEST}）
 * 都是来源凭据，因此共用同一个文件与同一套取值顺序。
 */
public final class LocalCredentials
{
    /** 相对进程工作目录定位，使开发（仓库根）与交付（应用目录）用同一个位置。 */
    public static final Path DEFAULT_FILE = Path.of("config", "vendor-credentials.properties");

    private LocalCredentials() { }

    /** 按默认位置与进程环境取值；取不到时抛出 {@link DeviceAdapterException}。 */
    public static String resolve(String key)
    {
        return resolve(key, DEFAULT_FILE.toAbsolutePath(), System.getenv(), System::getProperty);
    }

    /**
     * 取值顺序：文件 → 环境变量 → 启动参数。文件里的空白值不算配置过，
     * 否则一个空模板会把失败推到连接阶段。
     */
    public static String resolve(String key, Path credentialsFile, Map<String, String> environment,
            Function<String, String> startupParameters)
    {
        if (key == null || key.isBlank()) throw new DeviceAdapterException("未配置来源凭据的取值名");
        String fromFile = fromFile(key, credentialsFile);
        if (fromFile != null) return fromFile;
        String fromEnvironment = environment == null ? null : environment.get(key);
        if (fromEnvironment != null && !fromEnvironment.isBlank()) return fromEnvironment.trim();
        String fromParameter = startupParameters == null ? null : startupParameters.apply(key);
        if (fromParameter != null && !fromParameter.isBlank()) return fromParameter.trim();
        throw new DeviceAdapterException("缺少来源凭据：" + credentialsFile
                + " 中没有键 " + key + "，环境变量与启动参数 -D" + key + " 也都没有");
    }

    /** 凭据文件缺失不是错误：交付形态可能只用环境变量，所以这里只负责"读到了就返回"。 */
    private static String fromFile(String key, Path credentialsFile)
    {
        if (credentialsFile == null || !Files.isRegularFile(credentialsFile)) return null;
        Properties properties = new Properties();
        try (InputStream input = Files.newInputStream(credentialsFile))
        {
            properties.load(input);
        }
        catch (IOException exception)
        {
            throw new DeviceAdapterException("来源凭据文件无法读取：" + credentialsFile, exception);
        }
        String value = properties.getProperty(key);
        return value == null || value.isBlank() ? null : value.trim();
    }
}

 battery-device-neware/src/main/java/com/battery/device/neware/NewareMysqlAdapterFactory.java

@@ -1,16 +1,12 @@
package com.battery.device.neware;

import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Properties;
import java.util.function.Function;
import com.battery.device.api.DeviceAdapter;
import com.battery.device.api.DeviceAdapterException;
import com.battery.device.api.credentials.LocalCredentials;
import com.battery.device.api.model.AdapterCapabilities;
import com.battery.device.api.registry.AdapterInstanceConfig;
import com.battery.device.api.registry.DeviceAdapterFactory;
@@ -33,7 +29,7 @@
    /** 现场已知的全权限账号；写进交付配置就等于把误写风险带进现场。 */
    private static final List<String> FULL_PRIVILEGE_USERS = List.of("btsserver", "root");
    /** 凭据文件相对进程工作目录定位，使开发（仓库根）与交付（应用目录）用同一个位置。 */
    static final Path CREDENTIALS_FILE = Path.of("config", "vendor-credentials.properties");
    static final Path CREDENTIALS_FILE = LocalCredentials.DEFAULT_FILE;

    @Override
    public String adapterType()
@@ -48,7 +44,7 @@
        String user = readOnlyUser(config.sourceConfig());
        String key = option(config.sourceConfig(), PASSWORD_ENV_OPTION,
                NewareMysqlAdapter.DEFAULT_PASSWORD_ENV);
        String password = credential(key, CREDENTIALS_FILE.toAbsolutePath(), System.getenv(),
        String password = LocalCredentials.resolve(key, CREDENTIALS_FILE.toAbsolutePath(), System.getenv(),
                System::getProperty);
        SingleConnectionVendorDataSource source = new SingleConnectionVendorDataSource(
                jdbcUrl(config.host(), config.port(), database), user, password);
@@ -67,41 +63,6 @@
        if (FULL_PRIVILEGE_USERS.contains(user.toLowerCase(Locale.ROOT)))
            throw new DeviceAdapterException("厂商数据源只允许只读账号，拒绝全权限账号：" + user);
        return user;
    }

    /**
     * 按「本地凭据文件 → 环境变量 → 启动参数」取值；键名由实例配置的 {@code passwordEnv} 指定。
     * 文件优先，是为了让部署只做"复制模板填一行"这一件事。
     */
    static String credential(String key, Path credentialsFile, Map<String, String> environment,
            Function<String, String> startupParameters)
    {
        if (key == null || key.isBlank()) throw new DeviceAdapterException("未配置厂商库口令的取值名");
        String fromFile = credentialFromFile(key, credentialsFile);
        if (fromFile != null) return fromFile;
        String fromEnvironment = environment == null ? null : environment.get(key);
        if (fromEnvironment != null && !fromEnvironment.isBlank()) return fromEnvironment.trim();
        String fromParameter = startupParameters == null ? null : startupParameters.apply(key);
        if (fromParameter != null && !fromParameter.isBlank()) return fromParameter.trim();
        throw new DeviceAdapterException("缺少厂商库连接口令：" + credentialsFile
                + " 中没有键 " + key + "，环境变量与启动参数 -D" + key + " 也都没有");
    }

    /** 凭据文件缺失不是错误：交付形态可能只用环境变量，所以这里只负责"读到了就返回"。 */
    private static String credentialFromFile(String key, Path credentialsFile)
    {
        if (credentialsFile == null || !Files.isRegularFile(credentialsFile)) return null;
        Properties properties = new Properties();
        try (InputStream input = Files.newInputStream(credentialsFile))
        {
            properties.load(input);
        }
        catch (IOException exception)
        {
            throw new DeviceAdapterException("厂商库凭据文件无法读取：" + credentialsFile, exception);
        }
        String value = properties.getProperty(key);
        return value == null || value.isBlank() ? null : value.trim();
    }

    /**

 battery-device-neware/src/test/java/com/battery/device/neware/NewareMysqlAdapterContractTest.java

@@ -16,6 +16,7 @@
import com.battery.device.api.TestDataObservationAdapter.CycleRecord;
import com.battery.device.api.TestDataObservationAdapter.SampleWatermark;
import com.battery.device.api.TestDataObservationAdapter.StepRecord;
import com.battery.device.api.credentials.LocalCredentials;
import com.battery.device.api.model.AdapterCapabilities;
import com.battery.device.api.model.ChannelIdentity;
import com.battery.device.api.model.DetectedChannel;
@@ -429,22 +430,22 @@
        {
            // 文件、环境变量、启动参数都在时取文件。
            java.nio.file.Files.writeString(file, key + "=from-file\n");
            assertEquals("from-file", NewareMysqlAdapterFactory.credential(key, file,
            assertEquals("from-file", LocalCredentials.resolve(key, file,
                    Map.of(key, "from-env"), name -> "from-parameter"));

            // 没有文件时退到环境变量。
            java.nio.file.Files.delete(file);
            assertEquals("from-env", NewareMysqlAdapterFactory.credential(key, file,
            assertEquals("from-env", LocalCredentials.resolve(key, file,
                    Map.of(key, "from-env"), name -> "from-parameter"));

            // 前两者都没有时退到启动参数。
            assertEquals("from-parameter", NewareMysqlAdapterFactory.credential(key, file,
            assertEquals("from-parameter", LocalCredentials.resolve(key, file,
                    Map.of(), name -> "from-parameter"));

            // 三者都没有：失败信息里同时出现文件路径与键名。
            var failure = org.junit.jupiter.api.Assertions.assertThrows(
                    com.battery.device.api.DeviceAdapterException.class,
                    () -> NewareMysqlAdapterFactory.credential(key, file, Map.of(), name -> null));
                    () -> LocalCredentials.resolve(key, file, Map.of(), name -> null));
            assertTrue(failure.getMessage().contains(file.toString()));
            assertTrue(failure.getMessage().contains(key));
        }
@@ -465,7 +466,7 @@
        try
        {
            java.nio.file.Files.writeString(file, key + "=\n");
            assertEquals("from-env", NewareMysqlAdapterFactory.credential(key, file, Map.of(key, "from-env"),
            assertEquals("from-env", LocalCredentials.resolve(key, file, Map.of(key, "from-env"),
                    name -> null));
        }
        finally

 ruoyi-admin/src/main/java/com/battery/business/runtime/DeviceAdapterRuntimeConfig.java

@@ -8,6 +8,8 @@
import com.battery.device.bts.BtsDeviceAdapterFactory;
import com.battery.device.api.registry.DeviceAdapterRegistry;
import com.battery.device.neware.NewareMysqlAdapterFactory;
import com.battery.business.mapper.TDeviceIngestRecordMapper;
import com.battery.business.service.impl.PushIngestAdapterFactory;

/**
 * 装配单进程设备适配器运行时。
@@ -15,15 +17,17 @@
 * <p>注册表按存储形态持有全部可用实现；实例按 {@code t_device_adapter} 配置或代码装配注册。
 * 同一进程内可并存多个来源实例，取数按通道身份里的适配器编码解析到各自实例。
 * 某个类型是否真的有实例可用只由配置决定，因此这里不再按 profile 收窄可注册的类型。
 *
 * <p>推式来源（{@code PUSH_INGEST}）读的是本项目的落地表，所以它的工厂拿业务 Mapper 而不是厂商连接。
 */
@Configuration
@EnableScheduling
public class DeviceAdapterRuntimeConfig
{
    @Bean(destroyMethod = "close")
    public DeviceAdapterRegistry deviceAdapterRegistry()
    public DeviceAdapterRegistry deviceAdapterRegistry(TDeviceIngestRecordMapper ingestRecords)
    {
        return new DeviceAdapterRegistry(List.of(new NoopDeviceAdapterFactory(), new BtsDeviceAdapterFactory(),
                new NewareMysqlAdapterFactory()));
                new NewareMysqlAdapterFactory(), new PushIngestAdapterFactory(ingestRecords)));
    }
}

 ruoyi-admin/src/main/java/com/battery/business/web/device/DeviceIngestController.java

New file
@@ -0,0 +1,75 @@
package com.battery.business.web.device;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestHeader;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import com.ruoyi.common.annotation.Anonymous;
import com.ruoyi.common.exception.ServiceException;
import com.battery.business.service.DeviceAdapterErrorCodes;
import com.battery.business.service.DeviceIngestErrorCodes;
import com.battery.business.service.IDeviceIngestService;
import com.battery.business.service.IDeviceIngestService.IngestResult;

/**
 * 推式来源的接收端点（ADR-0065）：甲方机器上的只读取数程序把厂商原始行推到这里。
 *
 * <p>端点匿名放行但自己校验来源凭据（令牌随请求头来）：取数程序不带登录态，也不能带。
 * 拒绝一律用非 2xx 状态码——程序只在收到 2xx 后才推进自己的读取位置，
 * 所以"响应 200 但体内是错误码"是不能接受的：那会让程序把没落库的数据当成已送达。
 */
@Anonymous
@RestController
@RequestMapping("/device-ingest")
public class DeviceIngestController
{
    /** 来源凭据的请求头名；每个推式来源实例各有一个令牌。 */
    public static final String TOKEN_HEADER = "X-Device-Ingest-Token";

    private final IDeviceIngestService ingest;

    @Autowired
    public DeviceIngestController(IDeviceIngestService ingest)
    {
        this.ingest = ingest;
    }

    @PostMapping("/{adapterCode}")
    public ResponseEntity<?> receive(@PathVariable String adapterCode,
            @RequestHeader(value = TOKEN_HEADER, required = false) String token,
            @RequestBody(required = false) String body)
    {
        try
        {
            IngestResult result = ingest.ingest(adapterCode, token, body);
            return ResponseEntity.ok(result);
        }
        catch (ServiceException failure)
        {
            return ResponseEntity.status(httpStatus(failure.getCode()))
                    .body(new IngestFailure(failure.getCode(), failure.getMessage()));
        }
    }

    /** 业务错误码到 HTTP 状态的映射：只影响取数程序"是否推进水位"这一个判断。 */
    private static HttpStatus httpStatus(Integer code)
    {
        if (code == null) return HttpStatus.INTERNAL_SERVER_ERROR;
        return switch (code)
        {
            case DeviceAdapterErrorCodes.NOT_FOUND -> HttpStatus.NOT_FOUND;
            case DeviceIngestErrorCodes.TOKEN_INVALID -> HttpStatus.UNAUTHORIZED;
            case DeviceAdapterErrorCodes.UNSUPPORTED_TYPE, DeviceIngestErrorCodes.SOURCE_DISABLED -> HttpStatus.FORBIDDEN;
            case DeviceIngestErrorCodes.PAYLOAD_INVALID -> HttpStatus.BAD_REQUEST;
            default -> HttpStatus.INTERNAL_SERVER_ERROR;
        };
    }

    /** 拒绝应答：取数程序据此记录原因，并保持自己的读取位置不动。 */
    public record IngestFailure(Integer code, String message) { }
}

 ruoyi-admin/src/main/resources/db/migration-final/V24__push_ingest_landing_table.sql

New file
@@ -0,0 +1,43 @@
-- Ticket03：推式来源（PUSH_INGEST）的落地表，ADR-0065。
-- 甲方机器上的只读取数程序把厂商原始行推给接收端点，行按收到的键值原样落在这里：不改列名、不增删字段、不做换算（JSON 列会规范化格式，但键值不变）。
-- 解释只做一遍并全部落在后端：身份映射、时间换算、幂等键取值、测量列取值都在接收端点里完成，
-- 因此这张表既能本地回放开发与回归（不依赖那台机器在线），也是适配器读回统一观测的唯一来源——
-- 适配器只做投影，不含任何厂商列名。
-- 推送记录按事件落库：唯一键让同一条来源行重复推送被幂等忽略，同身份内容不同只报冲突、不覆盖。
CREATE TABLE t_device_ingest_record (
    ingest_id BIGINT NOT NULL AUTO_INCREMENT,
    adapter_code VARCHAR(64) NOT NULL,
    -- 载荷类别：TOPOLOGY 拓扑上报（通道清单）、SAMPLE 逐点数据行、HEARTBEAT 存活心跳。
    payload_kind VARCHAR(16) NOT NULL,
    devtype INT NOT NULL DEFAULT 0,
    devid INT NOT NULL DEFAULT 0,
    subdevid INT NOT NULL DEFAULT 0,
    chlid INT NOT NULL DEFAULT 0,
    -- 厂商协议序号（22 所为 Data_Point）：水位与"同一绝对秒内多行"都靠它区分。
    record_sequence BIGINT NOT NULL DEFAULT 0,
    -- 来源行身份（22 所为 库/表 + Data_Point）；同身份只允许一行，重复推送因此天然幂等。
    record_key VARCHAR(191) NOT NULL,
    -- 厂商来源定位（库/表），供回放与现场排查，不参与判定。
    source_ref VARCHAR(191) NULL,
    device_time_utc DATETIME(3) NULL,
    device_time_raw VARCHAR(64) NULL,
    timezone_id VARCHAR(64) NULL,
    cycle_no INT NULL,
    step_no INT NULL,
    relative_time_seconds DECIMAL(20,3) NULL,
    -- 测量列按厂商给的物理单位原样落库（22 所不做缩放）；缺列保持为空，不写零代替缺失。
    voltage_v DECIMAL(30,12) NULL,
    current_a DECIMAL(30,12) NULL,
    charge_capacity_ah DECIMAL(30,12) NULL,
    discharge_capacity_ah DECIMAL(30,12) NULL,
    charge_energy_wh DECIMAL(30,12) NULL,
    discharge_energy_wh DECIMAL(30,12) NULL,
    -- 厂商原始行，按收到的键值原样保留：回放与核对解释逻辑都靠它。
    payload_json JSON NOT NULL,
    content_sha256 BINARY(32) NOT NULL,
    received_at_utc DATETIME(3) NOT NULL,
    PRIMARY KEY (ingest_id),
    UNIQUE KEY uk_ingest_record (adapter_code, payload_kind, record_key),
    -- 逐行接收按 (设备时间, 协议序号) 取严格晚于水位的数据行，索引列次序与它一致。
    KEY idx_ingest_channel (adapter_code, payload_kind, devtype, devid, subdevid, chlid, device_time_utc, record_sequence)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='推式来源落地表';

 ruoyi-admin/src/test/java/com/battery/business/runtime/DeviceAdapterRuntimeNoopStartupTest.java

@@ -2,10 +2,12 @@

import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.mockito.Mockito.mock;
import org.junit.jupiter.api.Test;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import com.battery.device.api.NoopDeviceAdapter;
import com.battery.device.api.registry.DeviceAdapterRegistry;
import com.battery.business.mapper.TDeviceIngestRecordMapper;

class DeviceAdapterRuntimeNoopStartupTest
{
@@ -15,6 +17,9 @@
        try (AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext())
        {
            context.register(DeviceAdapterRuntimeConfig.class);
            // 推式来源的工厂读本项目的落地表，因此需要它的 Mapper（真实应用里由 MyBatis 提供）；
            // 本用例断言的是"没有配置任何设备来源时仍能装配出无设备实现"，所以这里给一个空替身。
            context.registerBean(TDeviceIngestRecordMapper.class, () -> mock(TDeviceIngestRecordMapper.class));
            context.refresh();
            DeviceAdapterRegistry registry = context.getBean(DeviceAdapterRegistry.class);
            assertFalse(registry.contains("dev-sim"));

 ruoyi-admin/src/test/java/com/battery/business/runtime/PushIngestLandingMigrationContractTest.java

New file
@@ -0,0 +1,270 @@
package com.battery.business.runtime;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;

import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.sql.Timestamp;
import java.time.Instant;
import java.util.Date;
import java.util.HexFormat;
import java.util.List;
import java.util.UUID;
import org.apache.ibatis.datasource.unpooled.UnpooledDataSource;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.mybatis.spring.SqlSessionTemplate;
import org.springframework.core.io.ClassPathResource;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.init.ScriptUtils;
import com.battery.business.domain.TDeviceIngestRecord;
import com.battery.business.domain.vo.TDeviceIngestChannel;
import com.battery.business.mapper.TDeviceIngestRecordMapper;

/**
 * 推式来源落地表的迁移与落库语义（生产 Mapper + 迁移脚本 + H2 内存库）：
 * 来源行身份让重复推送幂等、同身份不同内容不覆盖；心跳行按来源覆盖接收时间；
 * 拓扑上报的通道清单带上自己的最近数据时间；水位按 (设备时间, 协议序号) 取严格更晚的行。
 */
public class PushIngestLandingMigrationContractTest
{
    private static final String PATH = "db/migration-final/V24__push_ingest_landing_table.sql";
    private static final String ADAPTER = "j22";
    private static final Instant RECEIVED_AT = Instant.parse("2026-09-27T09:20:00Z");

    private java.sql.Connection keepAlive;
    private JdbcTemplate jdbc;
    private TDeviceIngestRecordMapper mapper;

    @BeforeEach void setUp() throws Exception
    {
        var source = new UnpooledDataSource("org.h2.Driver",
                "jdbc:h2:mem:" + UUID.randomUUID() + ";MODE=MySQL;DATABASE_TO_LOWER=TRUE", "sa", "");
        keepAlive = source.getConnection();
        jdbc = new JdbcTemplate(source);
        // H2 无 MySQL 的 HEX()；测试中补一个等价别名，生产 SQL 仍在 MySQL 上验证。
        jdbc.execute("create alias HEX for 'com.battery.business.runtime.PushIngestLandingMigrationContractTest.hexOf'");
        ScriptUtils.executeSqlScript(keepAlive, new ClassPathResource(PATH));

        SqlSessionFactoryBean factory = new SqlSessionFactoryBean();
        factory.setDataSource(source);
        factory.setTypeAliasesPackage("com.battery.business.domain");
        factory.setMapperLocations(new ClassPathResource("mapper/battery/TDeviceIngestRecordMapper.xml"));
        mapper = new SqlSessionTemplate(factory.getObject()).getMapper(TDeviceIngestRecordMapper.class);
    }

    @AfterEach void closeDatabase() throws Exception { if (keepAlive != null) keepAlive.close(); }

    @Test
    void migrationCreatesTheLandingTableWithoutWritingAnyData()
    {
        String migration = resource(PATH);
        // 迁移只建表：不导入演示数据、不改写任何既有表。
        assertFalse(migration.matches("(?s).*(INSERT|UPDATE|DELETE|DROP)\\b.*"));
        assertTrue(migration.contains("UNIQUE KEY uk_ingest_record (adapter_code, payload_kind, record_key)"));

        assertEquals(0, jdbc.queryForObject("select count(*) from t_device_ingest_record", Integer.class));
        // 落地表保留厂商原始载荷，且测量列可以缺列（设备没给就是空，不写零代替缺失）。
        jdbc.update("insert into t_device_ingest_record(adapter_code, payload_kind, record_key, payload_json,"
                + " content_sha256, received_at_utc) values(?,?,?,?,?,?)", ADAPTER, "SAMPLE", "k",
                "{\"Voltage\":1.2}", new byte[32], Timestamp.from(RECEIVED_AT));
        assertEquals(1, jdbc.queryForObject(
                "select count(*) from t_device_ingest_record where voltage_v is null", Integer.class));
    }

    @Test
    void repeatedPushIsIgnoredBySourceRowIdentityAndDifferentContentIsKept()
    {
        TDeviceIngestRecord first = sample("DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.2}");
        assertEquals(1, mapper.insertIngestRecord(first));

        // 同身份同内容：幂等忽略，库里仍是一行。
        TDeviceIngestRecord repeat = sample("DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.2}");
        assertEquals(0, mapper.insertIngestRecord(repeat));
        assertEquals(1, jdbc.queryForObject("select count(*) from t_device_ingest_record", Integer.class));

        // 同身份不同内容：内容摘要不同，调用方据此报冲突；既有那一行不被覆盖。
        TDeviceIngestRecord conflict = sample("DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.3}");
        assertEquals(0, mapper.insertIngestRecord(conflict));
        assertEquals(HexFormat.of().formatHex(first.getContentSha256()),
                mapper.selectIngestRecordHashByKey(ADAPTER, "SAMPLE", first.getRecordKey()));
        // H2 的 JSON 列会再引一层，所以只断内容；既有那一行没有被冲突推送覆盖。
        assertTrue(jdbc.queryForObject("select payload_json from t_device_ingest_record", String.class)
                .contains("Voltage"));
    }

    @Test
    void heartbeatRowIsUpsertedPerSourceSoAHeartbeatNeverGrows()
    {
        mapper.upsertHeartbeat(heartbeat(RECEIVED_AT));
        assertEquals(Timestamp.from(RECEIVED_AT).toInstant(), mapper.selectHeartbeatAt(ADAPTER).toInstant());

        // 下一轮心跳只推进接收时间：它表示"取数程序还在推"，不是设备事实，没有保留历史的意义。
        Instant later = RECEIVED_AT.plusSeconds(60);
        mapper.upsertHeartbeat(heartbeat(later));
        assertEquals(Timestamp.from(later).toInstant(), mapper.selectHeartbeatAt(ADAPTER).toInstant());
        assertEquals(1, jdbc.queryForObject("select count(*) from t_device_ingest_record", Integer.class));
        // 从未收到推送的来源没有心跳事实。
        assertNull(mapper.selectHeartbeatAt("other"));
    }

    @Test
    void topologyChannelsCarryTheirOwnLatestSampleTimeAndNeverExpire()
    {
        mapper.insertIngestRecord(topology(26));
        mapper.insertIngestRecord(topology(18));
        // 18 号只有更早的采样行；26 号有两行。
        mapper.insertIngestRecord(sample(26, "DB_20260927_155209/Ws26_IV_0:11", 11L, "{\"Voltage\":1.2}",
                Instant.parse("2026-09-27T07:52:20Z")));
        mapper.insertIngestRecord(sample(26, "DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.3}",
                Instant.parse("2026-09-27T07:53:20Z")));
        mapper.insertIngestRecord(sample(18, "DB_20260923_161840/Ws18_IV_0:5", 5L, "{\"Voltage\":1.1}",
                Instant.parse("2026-09-23T08:18:40Z")));

        List<TDeviceIngestChannel> channels = mapper.selectTopologyChannels(ADAPTER);
        assertEquals(List.of(18, 26), channels.stream().map(TDeviceIngestChannel::getChlid).toList());
        assertEquals(Instant.parse("2026-09-23T08:18:40Z"), channels.get(0).getLastDataAt().toInstant());
        assertEquals(Instant.parse("2026-09-27T07:53:20Z"), channels.get(1).getLastDataAt().toInstant());
        // 没有采样行的通道不给"最近数据时间"：这个事实是观测到的，不按上报时间代替。
        mapper.insertIngestRecord(topology(21));
        assertNull(mapper.selectTopologyChannels(ADAPTER).stream()
                .filter(channel -> channel.getChlid() == 21).findFirst().orElseThrow().getLastDataAt());
    }

    @Test
    void samplesSinceReturnsOnlyRowsStrictlyAfterTheWatermarkIncludingSameSecondRows()
    {
        TDeviceIngestRecord before = sample(26, "DB_20260927_155209/Ws26_IV_0:11", 11L, "{\"Voltage\":1.2}",
                Instant.parse("2026-09-27T07:52:20Z"));
        TDeviceIngestRecord sameSecond = sample(26, "DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.3}",
                Instant.parse("2026-09-27T07:53:20Z"));
        TDeviceIngestRecord sameSecondLaterSequence = sample(26, "DB_20260927_155209/Ws26_IV_0:13", 13L,
                "{\"Voltage\":1.4}", Instant.parse("2026-09-27T07:53:20Z"));
        TDeviceIngestRecord after = sample(26, "DB_20260927_155209/Ws26_IV_0:14", 14L, "{\"Voltage\":1.5}",
                Instant.parse("2026-09-27T07:54:20Z"));
        for (TDeviceIngestRecord row : List.of(before, sameSecond, sameSecondLaterSequence, after))
            mapper.insertIngestRecord(row);

        // 水位落在一行上：那一行自己不算，严格更晚的才进本次接收。
        List<TDeviceIngestRecord> rows = mapper.selectSamplesSince(ADAPTER, 0, 0, 1, 26,
                Date.from(sameSecond.getDeviceTimeUtc().toInstant()), 12L, 500);
        assertEquals(List.of(13L, 14L), rows.stream().map(TDeviceIngestRecord::getRecordSequence).toList());

        // 同一绝对秒内的多行各自成立：水位为空表示边界时刻什么也没有，之后的行全部属于本次接收。
        assertEquals(4, mapper.selectSamplesSince(ADAPTER, 0, 0, 1, 26, null, 0L, 500).size());
        // 其他通道与来源的行不混进来。
        assertTrue(mapper.selectSamplesSince("other", 0, 0, 1, 26, null, 0L, 500).isEmpty());
    }

    @Test
    void latestSampleDrivesBothRealtimeObservationAndWatermark()
    {
        assertNull(mapper.selectLatestSample(ADAPTER, 0, 0, 1, 26));
        mapper.insertIngestRecord(sample(26, "DB_20260927_155209/Ws26_IV_0:12", 12L, "{\"Voltage\":1.3}",
                Instant.parse("2026-09-27T07:53:20Z")));
        TDeviceIngestRecord latest = mapper.selectLatestSample(ADAPTER, 0, 0, 1, 26);
        assertNotNull(latest);
        assertEquals(12L, latest.getRecordSequence());
        assertEquals("2026-09-27 15:53:20", latest.getDeviceTimeRaw());
        assertEquals(Instant.parse("2026-09-27T07:53:20Z"), latest.getDeviceTimeUtc().toInstant());
        // 原始行逐字留在落地表里，回放与核对解释逻辑都靠它（H2 的 JSON 列会再引一层，所以只断内容）。
        assertTrue(latest.getPayloadJson().contains("Voltage"));
    }

    /** 供 H2 别名使用的 HEX 等价实现；H2 没有 MySQL 的 HEX()。 */
    public static String hexOf(byte[] value)
    {
        return value == null ? null : HexFormat.of().formatHex(value);
    }

    /** 内容摘要固定 32 字节：列是 BINARY(32)，短值会被补齐，比较时对不上。 */
    private static byte[] digest(String content)
    {
        try
        {
            return java.security.MessageDigest.getInstance("SHA-256")
                    .digest(content.getBytes(StandardCharsets.UTF_8));
        }
        catch (Exception exception)
        {
            throw new IllegalStateException(exception);
        }
    }

    private TDeviceIngestRecord sample(String recordKey, long sequence, String payload)
    {
        return sample(26, recordKey, sequence, payload, Instant.parse("2026-09-27T07:53:20Z"));
    }

    private TDeviceIngestRecord sample(int channel, String recordKey, long sequence, String payload)
    {
        return sample(channel, recordKey, sequence, payload, Instant.parse("2026-09-27T07:53:20Z"));
    }

    private TDeviceIngestRecord sample(int channel, String recordKey, long sequence, String payload,
            Instant deviceTime)
    {
        TDeviceIngestRecord value = new TDeviceIngestRecord();
        value.setAdapterCode(ADAPTER);
        value.setPayloadKind(TDeviceIngestRecord.KIND_SAMPLE);
        value.setDevtype(0); value.setDevid(0); value.setSubdevid(1); value.setChlid(channel);
        value.setRecordSequence(sequence);
        value.setRecordKey(recordKey);
        value.setSourceRef("DB_20260927_155209/Ws26_IV_0");
        value.setDeviceTimeUtc(Timestamp.from(deviceTime));
        value.setDeviceTimeRaw("2026-09-27 15:53:20");
        value.setTimezoneId("Asia/Shanghai");
        value.setPayloadJson(payload);
        value.setContentSha256(digest(payload));
        value.setReceivedAtUtc(Timestamp.from(RECEIVED_AT));
        return value;
    }

    private TDeviceIngestRecord topology(int channel)
    {
        TDeviceIngestRecord value = new TDeviceIngestRecord();
        value.setAdapterCode(ADAPTER);
        value.setPayloadKind(TDeviceIngestRecord.KIND_TOPOLOGY);
        value.setDevtype(0); value.setDevid(0); value.setSubdevid(1); value.setChlid(channel);
        value.setRecordSequence(0L);
        value.setRecordKey("topology:" + channel);
        value.setPayloadJson("{\"channel\":" + channel + "}");
        value.setContentSha256(digest("topology:" + channel));
        value.setReceivedAtUtc(Timestamp.from(RECEIVED_AT));
        return value;
    }

    private TDeviceIngestRecord heartbeat(Instant receivedAt)
    {
        TDeviceIngestRecord value = new TDeviceIngestRecord();
        value.setAdapterCode(ADAPTER);
        value.setPayloadKind(TDeviceIngestRecord.KIND_HEARTBEAT);
        value.setDevtype(0); value.setDevid(0); value.setSubdevid(1); value.setChlid(0);
        value.setRecordSequence(0L);
        value.setRecordKey("heartbeat");
        value.setPayloadJson("{}");
        value.setContentSha256(new byte[32]);
        value.setReceivedAtUtc(Timestamp.from(receivedAt));
        return value;
    }

    private static String resource(String path)
    {
        try (InputStream stream = PushIngestLandingMigrationContractTest.class.getClassLoader()
                .getResourceAsStream(path))
        {
            if (stream == null) throw new IllegalStateException("缺少迁移文件：" + path);
            return new String(stream.readAllBytes(), StandardCharsets.UTF_8);
        }
        catch (java.io.IOException exception)
        {
            throw new IllegalStateException("无法读取迁移文件：" + path, exception);
        }
    }
}

 ruoyi-admin/src/test/java/com/battery/business/web/device/DeviceIngestControllerTest.java

New file
@@ -0,0 +1,71 @@
package com.battery.business.web.device;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

import java.lang.reflect.Method;
import org.junit.jupiter.api.Test;
import org.springframework.test.web.servlet.MockMvc;
import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;
import org.springframework.test.web.servlet.result.MockMvcResultMatchers;
import org.springframework.test.web.servlet.setup.MockMvcBuilders;
import com.ruoyi.common.annotation.Anonymous;
import com.ruoyi.common.exception.ServiceException;
import com.battery.business.service.DeviceAdapterErrorCodes;
import com.battery.business.service.DeviceIngestErrorCodes;
import com.battery.business.service.IDeviceIngestService;
import com.battery.business.service.IDeviceIngestService.IngestResult;

/**
 * 接收端点的应答契约：拒绝一律用非 2xx 状态码——取数程序只在收到 2xx 后才推进自己的读取位置，
 * 所以"响应 200 但体内是错误码"是不能接受的：那会让程序把没落库的数据当成已送达。
 */
class DeviceIngestControllerTest
{
    private final IDeviceIngestService ingest = mock(IDeviceIngestService.class);
    private final MockMvc mvc = MockMvcBuilders.standaloneSetup(new DeviceIngestController(ingest)).build();

    @Test
    void acceptedBatchAnswersOkSoTheProgramMayAdvanceItsWatermark() throws Exception
    {
        when(ingest.ingest(any(), any(), any())).thenReturn(new IngestResult(2, 0, 0));

        mvc.perform(MockMvcRequestBuilders.post("/device-ingest/j22")
                        .header(DeviceIngestController.TOKEN_HEADER, "token").content("{}"))
                .andExpect(MockMvcResultMatchers.status().isOk())
                .andExpect(MockMvcResultMatchers.jsonPath("$.accepted").value(2));
    }

    @Test
    void rejectionsAnswerWithNonOkStatusesSoTheProgramKeepsItsPosition() throws Exception
    {
        assertStatus(DeviceIngestErrorCodes.TOKEN_INVALID, 401);
        assertStatus(DeviceIngestErrorCodes.SOURCE_DISABLED, 403);
        assertStatus(DeviceAdapterErrorCodes.UNSUPPORTED_TYPE, 403);
        assertStatus(DeviceIngestErrorCodes.PAYLOAD_INVALID, 400);
        assertStatus(DeviceAdapterErrorCodes.NOT_FOUND, 404);
        assertStatus(59999, 500);
    }

    /** 端点匿名放行（取数程序没有登录态），凭据由请求头里的来源令牌自己校验。 */
    @Test
    void endpointIsAnonymousAndAuthenticatesWithTheSourceTokenHeader() throws Exception
    {
        Method receive = DeviceIngestController.class.getMethod("receive", String.class, String.class, String.class);
        assertTrue(DeviceIngestController.class.isAnnotationPresent(Anonymous.class));
        assertEquals("X-Device-Ingest-Token", DeviceIngestController.TOKEN_HEADER);
        assertTrue(receive.getParameters()[1].toString().contains("token"));
    }

    private void assertStatus(int code, int expected) throws Exception
    {
        // 用 doThrow/when 形式重新打桩：直接调 mock 方法会被上一次的桩立即抛异常。
        org.mockito.Mockito.doThrow(new ServiceException("拒绝", code)).when(ingest).ingest(any(), any(), any());

        mvc.perform(MockMvcRequestBuilders.post("/device-ingest/j22").content("{}"))
                .andExpect(MockMvcResultMatchers.status().is(expected));
    }
}

 scripts/j22-collector.ps1

New file
@@ -0,0 +1,286 @@
#Requires -Version 5.1
<#
    22 所检测柜数据的只读取数程序 —— 运行在 22 所那台机器上
    ==========================================================================
    它做的全部事情（甲方五分钟即可审完）：

      1. 用 Windows 认证连**本机** SQL Server（Integrated Security）。这个文件里没有任何口令，
         网络上也不传任何厂商库凭据——认证在那台机器内部完成。
      2. 只执行固定的 SELECT。不写、不建、不改，不调用存储过程。
      3. 把读到的行**原样**推到我们的接收端点（HTTP，令牌放在请求头里）：不做单位换算、
         不改列名、不做业务判断。解释（身份映射、时间换算、幂等键）全部在后端完成。
      4. 只有收到 **2xx** 应答之后才推进自己对该表的读取位置。推送失败或没应答时，
         整批下一轮重推；端点按来源行身份幂等，重推不会产生重复记录。
      5. 不开任何入站端口，只发起出站连接。

    查询纪律（SQL 为什么写成这样）：
      * 所有命令都经同一个 Invoke-Query，它给**每一条**命令设 CommandTimeout；连接串里设
        Connect Timeout。SQL Server 服务端没有查询超时——客户端不设就是无限，
        所以两处都必须显式设置，且代码里不存在绕过它的查询路径。
      * 不加锁读、不长事务：全文没有 FOR UPDATE / HOLDLOCK / XLOCK / UPDLOCK / TABLOCK，
        没有 BEGIN TRAN，也不读会长期持有的视图。每条语句即取即断。
      * 增量条件是 record_time > @since，且**刻意不排序**：对没有索引的列排序会让服务端
        把整张表排序一遍，而那台机器是生产机。排序与截断都在本程序里做（客户端侧）。
      * 唯一一条 order by 落在 sys.databases 这个系统目录视图上，不碰任何测试表。
      * 活跃库探测读的是 sys.dm_db_index_usage_stats（目录视图，不是测试表；该实例只有
        5131 个库、每库一张表，聚合代价有限）。现场可用 sp_help 复核：
        若发现这轮探测或增量查询的代价偏高，就调整这里的策略。

    发送的列：厂商 IV 表自己的列，原样。要改列清单只改这一处；
    后端负责一切解释。

    交付注意事项：本文件是 **UTF-8 with BOM**。Windows PowerShell 5.1 会把没有 BOM 的
    UTF-8 按 GBK 解码，中文注释会破坏脚本（报"字符串缺少终止符"，或者静默什么都不做、
    退出码却是 0）。**重新保存本文件时必须保留 BOM。**
    另有两条坑：多行脚本不能粘贴执行（一律跑 .ps1 文件）；record_time 是 Int64，
    超过 JavaScript 安全整数范围，必须以字符串传输。
#>
param(
    # 推送目标，例如 http://<服务器>:8080/device-ingest/j22
    [Parameter(Mandatory = $true)][string]$Endpoint,
    # 该来源的令牌；对应值放在后端进程工作目录的 config/vendor-credentials.properties（不入库）
    [Parameter(Mandatory = $true)][string]$Token,

    [string]$SqlInstance = '.\SQLEXPRESS',
    [int]$IntervalSec = 60,
    # 只在第一次看到某张表时生效：从这里往前这么久开始取，而不是读整表历史——
    # 在生产库上读整表正是查询纪律要避免的
    [int]$LookbackMinutes = 10,
    [int]$MaxRowsPerChannel = 500,
    [int]$ConnectTimeoutSec = 10,
    [int]$CommandTimeoutSec = 15,
    [int]$HttpTimeoutSec = 30,
    [string]$StateFile = (Join-Path $PSScriptRoot 'j22-collector-state.json'),
    [string]$LogFile = (Join-Path $PSScriptRoot 'j22-collector.log'),
    [switch]$Once
)

$ErrorActionPreference = 'Continue'

# ---------------------------------------------------------------- 日志
function Write-Log([string]$message) {
    $line = (Get-Date -Format 'yyyy-MM-dd HH:mm:ss') + ' ' + $message
    try { Add-Content -LiteralPath $LogFile -Value $line -Encoding UTF8 } catch { }
    Write-Host $line
}

# ---------------------------------------------------------------- 读取位置
# 读取位置按「库里哪张表」记："DB_<...>/Ws<N>_IV_0" -> 端点已确认的最后一条 record_time（FILETIME 刻度）
function Get-ReadPositions {
    if (Test-Path -LiteralPath $StateFile) {
        try { return Get-Content -LiteralPath $StateFile -Raw | ConvertFrom-Json } catch { }
    }
    return New-Object psobject
}

function Save-ReadPositions($positions) {
    try { $positions | ConvertTo-Json -Depth 4 | Set-Content -LiteralPath $StateFile -Encoding UTF8 }
    catch { Write-Log ('读取位置未能落盘: ' + $_.Exception.Message) }
}

# ---------------------------------------------------------------- 厂商库访问
# 全程序唯一构造 SqlCommand 的地方，所以命令超时不可能被漏掉。所有查询都走这里。
# 取值在 reader 打开期间拷出，不留 reader 本身。
function Invoke-Query([System.Data.SqlClient.SqlConnection]$connection, [string]$sql, $parameters) {
    $command = $connection.CreateCommand()
    $command.CommandText = $sql
    $command.CommandTimeout = $CommandTimeoutSec
    if ($parameters) {
        foreach ($name in $parameters.Keys) { [void]$command.Parameters.AddWithValue($name, $parameters[$name]) }
    }
    $reader = $command.ExecuteReader()
    try {
        $rows = New-Object System.Collections.ArrayList
        while ($reader.Read()) {
            $values = New-Object object[] $reader.FieldCount
            [void]$reader.GetValues($values)
            [void]$rows.Add($values)
        }
        return $rows
    }
    finally { $reader.Close() }
}

function Open-VendorConnection {
    $connection = New-Object System.Data.SqlClient.SqlConnection(
        "Server=$SqlInstance;Database=master;Integrated Security=True;Connect Timeout=$ConnectTimeoutSec")
    $connection.Open()
    return $connection
}

# 候选库：只要最近写入过的。DMV 读不到时退回按建库时间取最近 60 个。
# 两条查询都失败就抛出——那说明这一轮**没有读到厂商库**，因此这一轮不许上报心跳。
function Find-ActiveDatabases([System.Data.SqlClient.SqlConnection]$connection) {
    $databases = New-Object System.Collections.ArrayList
    $failure = $null
    try {
        $rows = Invoke-Query $connection @"
select DB_NAME(database_id) as db
from sys.dm_db_index_usage_stats
where database_id > 4 and DB_NAME(database_id) like 'DB[_]%'
group by database_id
having max(last_user_update) > dateadd(minute, -$LookbackMinutes, getdate())
"@ $null
        foreach ($row in $rows) { [void]$databases.Add([string]$row[0]) }
        return $databases
    }
    catch {
        $failure = $_.Exception.Message
        Write-Log ('  活跃库探测失败，退回按建库时间取库列表: ' + $failure)
    }

    try {
        $rows = Invoke-Query $connection @"
select top 60 name from sys.databases where name like 'DB[_]%' order by create_date desc
"@ $null
        foreach ($row in $rows) { [void]$databases.Add([string]$row[0]) }
        return $databases
    }
    catch {
        throw "vendor database not readable: $failure / " + $_.Exception.Message
    }
}

function Find-ChannelTables([System.Data.SqlClient.SqlConnection]$connection, [string]$database) {
    $tables = New-Object System.Collections.ArrayList
    $rows = Invoke-Query $connection "select name from [$database].sys.tables where name like 'Ws%[_]IV[_]0'" $null
    foreach ($row in $rows) { [void]$tables.Add([string]$row[0]) }
    return $tables
}

# ---------------------------------------------------------------- 推送
# 只有 2xx 才返回 $true：调用方据此决定是否推进读取位置，所以没被确认的批次整批下一轮重推
function Send-Payload([string]$json) {
    try {
        Invoke-RestMethod -Uri $Endpoint -Method Post -Body $json `
            -ContentType 'application/json; charset=utf-8' `
            -Headers @{ 'X-Device-Ingest-Token' = $Token } `
            -TimeoutSec $HttpTimeoutSec | Out-Null
        return $true
    }
    catch {
        $status = ''
        try { $status = [int]$_.Exception.Response.StatusCode } catch { }
        Write-Log ("  推送失败（状态=$status）: " + $_.Exception.Message)
        return $false
    }
}

function Send-Samples([string]$database, [string]$table, [int]$channel, $rows) {
    $payload = [pscustomobject]@{
        kind     = 'SAMPLE'
        database = $database
        table    = $table
        channel  = $channel
        rows     = $rows
    } | ConvertTo-Json -Depth 6 -Compress
    return Send-Payload $payload
}

function Send-Topology($channels) {
    if ($channels.Count -eq 0) { return $true }
    $rows = @()
    foreach ($channel in $channels) { $rows += [pscustomobject]@{ channel = $channel } }
    $payload = [pscustomobject]@{ kind = 'TOPOLOGY'; rows = $rows } | ConvertTo-Json -Depth 4 -Compress
    return Send-Payload $payload
}

# 心跳只表示"程序还活着"。它在一轮成功读到厂商库之后才发，
# 所以厂商库连不上（或那台机器权限/网络有问题）时来源显示离线，而不是"绿色但没数据"。
function Send-Heartbeat {
    return Send-Payload '{"kind":"HEARTBEAT"}'
}

# ---------------------------------------------------------------- 一轮
function Invoke-Cycle([System.Data.SqlClient.SqlConnection]$connection) {
    $positions = Get-ReadPositions
    $seenChannels = New-Object System.Collections.ArrayList
    $sentRows = 0

    foreach ($database in (Find-ActiveDatabases $connection)) {
        foreach ($table in (Find-ChannelTables $connection $database)) {
            if ($table -notmatch '^Ws(\d+)_IV_0$') { continue }
            $channel = [int]$Matches[1]
            if (-not $seenChannels.Contains($channel)) { [void]$seenChannels.Add($channel) }

            $key = "$database/$table"
            $since = 0L
            if ($positions.PSObject.Properties.Name -contains $key) { $since = [long]$positions.$key }
            else {
                # 第一次看到这张表：从"现在往前 LookbackMinutes"开始，不读整表历史
                $since = [DateTime]::UtcNow.AddMinutes(-$LookbackMinutes).ToFileTimeUtc()
            }

            $rows = @()
            try {
                $read = Invoke-Query $connection @"
select record_time, Data_Point, Schedule_Index, Step_Index, Cycle_Index, Step_Time,
       Voltage, eCurrent, ChargeCapacity, DischargeCapacity, ChargeEnergy, DischargeEnergy
from [$database].dbo.[$table]
where record_time > @since
"@ @{ '@since' = $since }
                foreach ($row in $read) {
                    # record_time 是 Int64，超过 JavaScript 安全整数范围：必须以字符串传输
                    $rows += [pscustomobject]@{
                        record_time       = [string]$row[0]
                        Data_Point        = [int]$row[1]
                        Schedule_Index    = [string]$row[2]
                        Step_Index        = [int]$row[3]
                        Cycle_Index       = [int]$row[4]
                        Step_Time         = [double]$row[5]
                        Voltage           = [double]$row[6]
                        eCurrent          = [double]$row[7]
                        ChargeCapacity    = [double]$row[8]
                        DischargeCapacity = [double]$row[9]
                        ChargeEnergy      = [double]$row[10]
                        DischargeEnergy   = [double]$row[11]
                    }
                }
            }
            catch { Write-Log ("  读取失败 [$key]: " + $_.Exception.Message); continue }

            if ($rows.Count -eq 0) { continue }

            # 排序与截断在这里做：服务端不做排序
            $rows = @($rows | Sort-Object { [long]$_.record_time } | Select-Object -First $MaxRowsPerChannel)
            $highest = [long]$rows[-1].record_time

            if (-not (Send-Samples $database $table $channel $rows)) { continue }
            $positions | Add-Member -NotePropertyName $key -NotePropertyValue $highest -Force
            $sentRows += $rows.Count
            Write-Log ("  已推 {0,4} 行 ch={1} {2}" -f $rows.Count, $channel, $key)
        }
    }

    # 拓扑：这台机器当前有数据表的通道。端点忽略重复声明，所以每轮都发没有副作用；
    # 它同时是候选"最近数据时间"的来源之外的通道清单（最近数据时间取采样行）。
    [void](Send-Topology $seenChannels)
    Save-ReadPositions $positions
    return $sentRows
}

# ---------------------------------------------------------------- 主循环
Write-Log '===== 取数程序启动 ====='
Write-Log "端点 = $Endpoint"
Write-Log "间隔 = ${IntervalSec}s，回溯 = ${LookbackMinutes}min，实例 = $SqlInstance"

$connection = $null
while ($true) {
    try {
        if ($null -eq $connection -or $connection.State -ne 'Open') {
            $connection = Open-VendorConnection
            Write-Log '已连上本机 SQL Server（只读，Windows 认证）'
        }
        $sent = Invoke-Cycle $connection
        # 这一轮读到了厂商库：这是"来源还在"的证据
        if (Send-Heartbeat) { Write-Log "本轮结束，共推 $sent 行" }
    }
    catch {
        Write-Log ('本轮出错: ' + $_.Exception.Message)
        try { if ($connection) { $connection.Close() } } catch { }
        $connection = $null
    }

    if ($Once) { break }
    Start-Sleep -Seconds $IntervalSec
}