package com.battery.business.service.impl;

import java.time.Duration;
import java.time.Instant;
import java.util.LinkedHashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore;
import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore.CabinetCandidate;
import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore.ChannelCandidate;
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.DeviceWorkStatus;
import com.battery.device.api.model.RealtimeSnapshot;
import com.battery.device.api.registry.DeviceAdapterRegistry;
import com.battery.business.domain.TDetectionChannel;
import com.battery.business.domain.vo.AdapterDiscoveryResult;
import com.battery.business.mapper.TDetectionCabinetMapper;
import com.battery.business.mapper.TTestMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/** 将适配器观测写入 Redis 候选，并同步已登记设备的实时事实。 */
@Service
public class DeviceDiscoveryPersistenceService
{
    private static final Duration MINIMUM_CANDIDATE_LIFETIME = Duration.ofSeconds(30);
    /**
     * 直连厂商存储的来源，其通道登记清单是长期存在的事实，没有"过期"的对应语义：
     * 候选是否出现在界面上只由"已登记且未删除"决定，这个时间只是让候选一直有效。
     */
    static final Instant CANDIDATE_NEVER_EXPIRES = Instant.parse("9999-12-31T23:59:59Z");
    private final DeviceDiscoveryCandidateStore candidateStore;
    private final TDetectionCabinetMapper detectionCabinetMapper;
    private final TTestMapper testMapper;

    public DeviceDiscoveryPersistenceService(DeviceDiscoveryCandidateStore candidateStore,
            TDetectionCabinetMapper detectionCabinetMapper)
    {
        this(candidateStore, detectionCabinetMapper, null);
    }

    @Autowired
    public DeviceDiscoveryPersistenceService(DeviceDiscoveryCandidateStore candidateStore,
            TDetectionCabinetMapper detectionCabinetMapper, TTestMapper testMapper)
    {
        this.candidateStore = candidateStore;
        this.detectionCabinetMapper = detectionCabinetMapper;
        this.testMapper = testMapper;
    }

    /** 执行设备发现和可用通道查询，并将本轮观测写入 Redis 候选。 */
    public AdapterDiscoveryResult discover(DeviceAdapterRegistry registry, DiscoverySource source)
    {
        return discoverNow(registry, source);
    }

    /** 执行设备发现；是否请求实时快照由接收中的通道集合单独决定。 */
    public AdapterDiscoveryResult discoverForReception(DeviceAdapterRegistry registry, DiscoverySource source)
    {
        return discoverNow(registry, source);
    }

    /** 判断适配器是否存在需要实时请求的接收中测试。 */
    public boolean shouldPollReception(String adapterCode)
    {
        return testMapper == null || testMapper.hasReceivingTestForAdapter(adapterCode) > 0;
    }

    /** 发现失败或返回空拓扑时，清除该适配器下已登记设备的过期运行事实。 */
    public void markUnavailable(String adapterCode)
    {
        Instant failedAt = Instant.now();
        // 失败立即撤下旧测量，但保留来源身份，不能等候选 TTL 到期才停止展示成功值。
        candidateStore.recordCabinets(candidateStore.snapshot(failedAt).cabinets().stream()
                .filter(item -> item.adapterCode().equals(adapterCode))
                .map(item -> new CabinetCandidate(item.adapterCode(), item.adapterId(), item.deviceCode(),
                        item.devtype(), item.devid(), item.ip(), item.mac(), item.name(), item.manufacturer(), item.model(),
                        item.channels().stream().map(channel -> new ChannelCandidate(channel.identity(), false,
                                DeviceWorkStatus.UNOBSERVABLE, failedAt, null, channel.lastDataAt(),
                                channel.sourceDisplayNo())).toList(),
                        failedAt, item.expiresAt(), item.lastDataAt()))
                .toList());
        detectionCabinetMapper.markRegisteredFactsUnobservable(adapterCode);
        detectionCabinetMapper.markRegisteredCabinetsOffline(adapterCode);
    }

    private AdapterDiscoveryResult discoverNow(DeviceAdapterRegistry registry, DiscoverySource source)
    {
        DiscoveryObservation observation = registry.execute(source.adapterCode(), runtime -> {
            List<DetectedChannel> channels = runtime.discoverChannels();
            Set<ChannelIdentity> available = channels.stream()
                    .filter(DetectedChannel::available)
                    .map(DetectedChannel::identity)
                    .collect(java.util.stream.Collectors.toCollection(LinkedHashSet::new));
            Set<ChannelIdentity> requested = receivingIdentities(source.adapterCode(), available);
            Map<ChannelIdentity, RealtimeSnapshot> snapshots = requested.isEmpty()
                    ? Map.of() : runtime.inquire(requested);
            return new DiscoveryObservation(channels, snapshots);
        });
        Instant observedAt = Instant.now();
        return persist(source, observation.channels(), observation.snapshots(), observedAt);
    }

    /** upsert 本轮发现和实时状态；未返回设备由 expiresAt 控制淘汰。 */
    private AdapterDiscoveryResult persist(DiscoverySource source, List<DetectedChannel> detected,
            Map<ChannelIdentity, RealtimeSnapshot> snapshots, Instant observedAt)
    {
        Map<CabinetKey, List<DetectedChannel>> grouped = detected.stream().collect(
                java.util.stream.Collectors.groupingBy(
                        item -> new CabinetKey(item.identity().devtype(), item.identity().devid()),
                        LinkedHashMap::new, java.util.stream.Collectors.toList()));
        Duration lifetime = Duration.ofMillis(Math.max(MINIMUM_CANDIDATE_LIFETIME.toMillis(),
                (long) source.pollIntervalMs() * 3));
        Instant expiresAt = source.persistentCandidates() ? CANDIDATE_NEVER_EXPIRES : observedAt.plus(lifetime);
        List<CabinetCandidate> candidates = grouped.entrySet().stream()
                .map(entry -> toCandidate(source, entry.getKey(), entry.getValue(), snapshots,
                        observedAt, expiresAt))
                .toList();
        candidateStore.recordCabinets(candidates);
        if (detected.isEmpty()) markUnavailable(source.adapterCode());
        synchronizeRegisteredFacts(source, detected, snapshots, observedAt);
        return new AdapterDiscoveryResult(grouped.size(), detected.size(), observedAt);
    }

    private Set<ChannelIdentity> receivingIdentities(String adapterCode, Set<ChannelIdentity> available)
    {
        if (testMapper == null) return available;
        List<TDetectionChannel> receiving = testMapper.selectReceivingChannelsForAdapter(adapterCode);
        if (receiving == null || receiving.isEmpty()) return Set.of();
        Set<ChannelIdentity> identities = receiving.stream()
                .map(channel -> new ChannelIdentity(adapterCode, channel.getDevtype(), channel.getDevid(),
                        channel.getSubdevid(), channel.getChlid()))
                .collect(java.util.stream.Collectors.toCollection(LinkedHashSet::new));
        identities.retainAll(available);
        return identities;
    }

    /**
     * 发现候选与正式设备是两份事实；正式登记后仍需把成功观测同步到 MySQL。
     * 暂停期间仍同步设备状态，但不请求该测试的测量；离线不覆盖最后成功观测时间。
     */
    private void synchronizeRegisteredFacts(DiscoverySource source, List<DetectedChannel> detected,
            Map<ChannelIdentity, RealtimeSnapshot> snapshots, Instant observedAt)
    {
        Map<CabinetKey, Boolean> cabinetOnline = new LinkedHashMap<>();
        for (DetectedChannel channel : detected)
        {
            ChannelIdentity identity = channel.identity();
            boolean successfulObservation = channel.available() && snapshots.containsKey(identity);
            RealtimeSnapshot snapshot = snapshots.get(identity);
            DeviceWorkStatus workStatus = snapshot == null || snapshot.workStatus() == null
                    ? DeviceWorkStatus.UNKNOWN : snapshot.workStatus();
            detectionCabinetMapper.updateRegisteredChannelObservation(source.adapterCode(), identity.devtype(),
                    identity.devid(), identity.subdevid(), identity.chlid(),
                    // 来源自己报了叫法就跟着刷新显示号；没报就保持原样，不去改操作员已登记的显示
                    channel.sourceDisplayNo(),
                    channel.available() ? "ONLINE" : "OFFLINE",
                    workStatus.name(), observedAt, successfulObservation);
            if (channel.available() && !successfulObservation)
                detectionCabinetMapper.touchRegisteredChannelObservation(source.adapterCode(), identity.devtype(),
                        identity.devid(), identity.subdevid(), identity.chlid(), observedAt);
            CabinetKey cabinet = new CabinetKey(identity.devtype(), identity.devid());
            cabinetOnline.merge(cabinet, channel.available(), Boolean::logicalOr);
        }
        cabinetOnline.forEach((cabinet, online) -> detectionCabinetMapper.updateRegisteredCabinetObservation(
                source.adapterCode(), cabinet.devtype(), cabinet.devid(), online ? "ONLINE" : "OFFLINE", observedAt,
                online && detected.stream().anyMatch(channel -> channel.available()
                        && channel.identity().devtype() == cabinet.devtype()
                        && channel.identity().devid() == cabinet.devid()
                        && snapshots.containsKey(channel.identity()))));
        cabinetOnline.forEach((cabinet, online) -> {
            boolean successfulSnapshot = online && detected.stream().anyMatch(channel -> channel.available()
                    && channel.identity().devtype() == cabinet.devtype()
                    && channel.identity().devid() == cabinet.devid()
                    && snapshots.containsKey(channel.identity()));
            if (online && !successfulSnapshot)
                detectionCabinetMapper.touchRegisteredCabinetObservation(source.adapterCode(), cabinet.devtype(),
                        cabinet.devid(), observedAt);
        });
    }

    private CabinetCandidate toCandidate(DiscoverySource source, CabinetKey key, List<DetectedChannel> channels,
            Map<ChannelIdentity, RealtimeSnapshot> snapshots, Instant observedAt, Instant expiresAt)
    {
        DetectedDeviceMetadata metadata = channels.get(0).deviceMetadata();
        String name = metadata.name() == null
                ? source.adapterName() + " " + key.devtype() + "-" + key.devid()
                : metadata.name();
        String ip = channels.stream().map(DetectedChannel::serverIp)
                .filter(value -> value != null && !value.isBlank()).findFirst().orElse(null);
        List<ChannelCandidate> channelCandidates = channels.stream().map(item -> {
            RealtimeSnapshot snapshot = snapshots.get(item.identity());
            DeviceWorkStatus status = snapshot == null || snapshot.workStatus() == null
                    ? DeviceWorkStatus.UNKNOWN : snapshot.workStatus();
            return new ChannelCandidate(item.identity(), item.available(), status, observedAt,
                    snapshots.get(item.identity()), item.lastDataAt(), item.sourceDisplayNo());
        }).toList();
        Instant lastDataAt = channels.stream().map(DetectedChannel::lastDataAt)
                .filter(java.util.Objects::nonNull).max(Instant::compareTo).orElse(null);
        return new CabinetCandidate(source.adapterCode(), source.adapterId(),
                source.adapterCode() + "-" + key.devtype() + "-" + key.devid(),
                key.devtype(), key.devid(), ip, metadata.macAddress(), name,
                metadata.manufacturer(), metadata.model(), channelCandidates, observedAt, expiresAt, lastDataAt);
    }

    /** 发现来源只携带候选标识和过期计算所需配置，不复用数据库实体。 */
    public record DiscoverySource(String adapterCode, Long adapterId, String adapterName, int pollIntervalMs,
            boolean persistentCandidates)
    {
        /** 兼容按轮询过期计算候选的既有来源。 */
        public DiscoverySource(String adapterCode, Long adapterId, String adapterName, int pollIntervalMs)
        {
            this(adapterCode, adapterId, adapterName, pollIntervalMs, false);
        }
    }

    private record DiscoveryObservation(List<DetectedChannel> channels,
            Map<ChannelIdentity, RealtimeSnapshot> snapshots) { }

    private record CabinetKey(int devtype, int devid) { }
}