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

    /** 拓扑是声明：撞键即刷新声明内容，不追加新行（与生产 SQL 的 upsert 语义一致）。 */
    @Override
    public int upsertTopology(TDeviceIngestRecord record)
    {
        int index = -1;
        for (int i = 0; i < rows.size(); i++)
            if (same(rows.get(i), record)) index = i;
        if (index < 0)
        {
            rows.add(record);
            return 1;
        }
        rows.set(index, 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.setVendorChannelNo(row.getVendorChannelNo());
            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> selectSamplesOfSource(String adapterCode, int devtype, int devid, int subdevid,
            int chlid, String sourceRef)
    {
        return samples(adapterCode, devtype, devid, subdevid, chlid).stream()
                .filter(row -> sourceRef.equals(row.getSourceRef()))
                .sorted(java.util.Comparator.comparing(TDeviceIngestRecord::getDeviceTimeUtc)
                        .thenComparing(TDeviceIngestRecord::getRecordSequence))
                .toList();
    }

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