pipe-ner-server/zk-api-service/src/main/java/com/zksy/api/service/impl/EquipmentStatusApiServiceImpl.java

package com.zksy.api.service.impl;

import static com.zksy.common.utils.CollectionSideEffects.*;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.zksy.api.domain.EquipmentBase;
import com.zksy.api.domain.EquipmentStatus;
import com.zksy.api.domain.AlarmData;
import com.zksy.api.event.DeviceStatusChangePublisher;
import com.zksy.api.service.CollectionAlarmWriter;
import com.zksy.api.mapper.EquipmentBaseApiMapper;
import com.zksy.api.mapper.EquipmentStatusApiMapper;
import com.zksy.api.service.EquipmentStatusApiService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import java.time.LocalDateTime;
import java.util.Objects;

/**
 * 设备报警状态服务实现
 * 根据设备编码更新 equipment_status 表的报警状态
 */
@Slf4j
@Service
public class EquipmentStatusApiServiceImpl implements EquipmentStatusApiService {

    @Autowired
    private EquipmentBaseApiMapper equipmentBaseApiMapper;

    @Autowired
    private EquipmentStatusApiMapper equipmentStatusApiMapper;

    @Autowired
    private DeviceStatusChangePublisher deviceStatusChangePublisher;

    @Autowired
    private CollectionAlarmWriter collectionAlarmWriter;

    @Override
    @Transactional
    public boolean updateAlarmStatus(String deviceCode, Integer alarmStatus) {
        if (deviceCode == null || deviceCode.isEmpty()) {
            log.warn("更新设备报警状态失败：deviceCode为空");
            return false;
        }

        // 1. 根据设备编码查询设备基础信息，获取 equipmentId
        EquipmentBase equipmentBase = equipmentBaseApiMapper.selectOne(
                new LambdaQueryWrapper<EquipmentBase>()
                        .eq(EquipmentBase::getEquipmentCode, deviceCode)
                        .last("LIMIT 1"));
        if (equipmentBase == null) {
            log.warn("更新设备报警状态失败：未找到设备编码对应的设备信息，deviceCode={}", deviceCode);
            return false;
        }

        String equipmentId = equipmentBase.getEquipmentId();

        // 2. 查询是否已有状态记录
        EquipmentStatus existingStatus = equipmentStatusApiMapper.selectOne(
                new LambdaQueryWrapper<EquipmentStatus>()
                        .eq(EquipmentStatus::getEquipmentId, equipmentId));

        if (existingStatus != null) {
            Integer previousAlarmStatus = existingStatus.getAlarmStatus();
            // 更新已有记录
            existingStatus.setAlarmStatus(alarmStatus);
            existingStatus.setStatusUpdateTime(LocalDateTime.now());
            int affected = equipmentStatusApiMapper.updateById(existingStatus);
            if (affected <= 0) {
                return false;
            }
            log.info("更新设备报警状态成功: deviceCode={}, equipmentId={}, alarmStatus={}",
                    deviceCode, equipmentId, alarmStatus);
            if (!Objects.equals(previousAlarmStatus, alarmStatus)) {
                deviceStatusChangePublisher.publishAfterCommit(deviceCode, equipmentId,
                        existingStatus.getCurrentStatus(), existingStatus.getAlarmStatus(),
                        existingStatus.getOnlineStatus(), "zk-api:updateAlarmStatus");
            }
        } else {
            // 不存在则创建新记录
            EquipmentStatus newStatus = new EquipmentStatus();
            newStatus.setEquipmentId(equipmentId);
            newStatus.setAlarmStatus(alarmStatus);
            newStatus.setCurrentStatus(1); // 默认在用
            newStatus.setOnlineStatus(1);  // 默认在线
            newStatus.setStatusUpdateTime(LocalDateTime.now());
            newStatus.setCreateTime(LocalDateTime.now());
            int affected = equipmentStatusApiMapper.insert(newStatus);
            if (affected <= 0) {
                return false;
            }
            log.info("创建设备报警状态记录成功: deviceCode={}, equipmentId={}, alarmStatus={}",
                    deviceCode, equipmentId, alarmStatus);
            deviceStatusChangePublisher.publishAfterCommit(deviceCode, equipmentId,
                    newStatus.getCurrentStatus(), newStatus.getAlarmStatus(),
                    newStatus.getOnlineStatus(), "zk-api:updateAlarmStatus");
        }

        return true;
    }

    @Override
    @Transactional
    public boolean updateOnlineStatus(String deviceCode, Integer onlineStatus) {
        if (deviceCode == null || deviceCode.isEmpty()) {
            log.warn("更新设备在线状态失败：deviceCode为空");
            return false;
        }

        EquipmentBase equipmentBase = equipmentBaseApiMapper.selectOne(
                new LambdaQueryWrapper<EquipmentBase>()
                        .eq(EquipmentBase::getEquipmentCode, deviceCode)
                        .last("LIMIT 1"));
        if (equipmentBase == null) {
            log.warn("更新设备在线状态失败：未找到设备编码对应的设备信息，deviceCode={}", deviceCode);
            return false;
        }

        String equipmentId = equipmentBase.getEquipmentId();

        EquipmentStatus existingStatus = equipmentStatusApiMapper.selectOne(
                new LambdaQueryWrapper<EquipmentStatus>()
                        .eq(EquipmentStatus::getEquipmentId, equipmentId));

        if (existingStatus != null) {
            Integer previousOnlineStatus = existingStatus.getOnlineStatus();
            existingStatus.setOnlineStatus(onlineStatus);
            existingStatus.setStatusUpdateTime(LocalDateTime.now());
            int affected = equipmentStatusApiMapper.updateById(existingStatus);
            if (affected <= 0) {
                return false;
            }
            log.info("更新设备在线状态成功: deviceCode={}, equipmentId={}, onlineStatus={}",
                    deviceCode, equipmentId, onlineStatus);
            if (!Objects.equals(previousOnlineStatus, onlineStatus)) {
                deviceStatusChangePublisher.publishAfterCommit(deviceCode, equipmentId,
                        existingStatus.getCurrentStatus(), existingStatus.getAlarmStatus(),
                        existingStatus.getOnlineStatus(), "zk-api:updateOnlineStatus");
            }
            // 设备由在线转为离线时，新增离线报警记录
            if (previousOnlineStatus != null && previousOnlineStatus == 1 && onlineStatus == 0) {
                saveOfflineAlarmAfterCommit(deviceCode);
            }
            // 收到有效数据重新上线后，自动处理对应的离线告警；恢复失败不影响在线状态和后续采集。
            if (previousOnlineStatus != null && previousOnlineStatus == 0 && onlineStatus == 1) {
                recoverOfflineAlarmAfterCommit(deviceCode);
            }
        } else {
            EquipmentStatus newStatus = new EquipmentStatus();
            newStatus.setEquipmentId(equipmentId);
            newStatus.setOnlineStatus(onlineStatus);
            newStatus.setAlarmStatus(0);
            newStatus.setCurrentStatus(1);
            newStatus.setStatusUpdateTime(LocalDateTime.now());
            newStatus.setCreateTime(LocalDateTime.now());
            int affected = equipmentStatusApiMapper.insert(newStatus);
            if (affected <= 0) {
                return false;
            }
            log.info("创建设备在线状态记录成功: deviceCode={}, equipmentId={}, onlineStatus={}",
                    deviceCode, equipmentId, onlineStatus);
            deviceStatusChangePublisher.publishAfterCommit(deviceCode, equipmentId,
                    newStatus.getCurrentStatus(), newStatus.getAlarmStatus(),
                    newStatus.getOnlineStatus(), "zk-api:updateOnlineStatus");
        }

        return true;
    }

    /**
     * 新增设备离线报警记录
     */
    private void saveOfflineAlarmAfterCommit(String deviceCode) {
        if (TransactionSynchronizationManager.isSynchronizationActive()) {
            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
                @Override
                public void afterCommit() {
                    run(deviceCode, "离线报警独立事务", () -> saveOfflineAlarm(deviceCode));
                }
            });
        } else {
            run(deviceCode, "离线报警", () -> saveOfflineAlarm(deviceCode));
        }
    }

    private void saveOfflineAlarm(String deviceCode) {
        AlarmData alarmData = new AlarmData();
        alarmData.setDeviceCode(deviceCode);
        alarmData.setWarningCode("OFFLINE");
        alarmData.setAlarmLevel(4);
        alarmData.setAlarmStatus(0);
        alarmData.setCreateTime(LocalDateTime.now());
        alarmData.setCreateTime(LocalDateTime.now());
        if (collectionAlarmWriter.saveOrUpdateUnhandled(alarmData)) {
            log.info("设备离线报警数据入库成功: deviceCode={}, alarmId={}", deviceCode, alarmData.getId());
        }
    }

    private void recoverOfflineAlarmAfterCommit(String deviceCode) {
        if (TransactionSynchronizationManager.isSynchronizationActive()) {
            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
                @Override
                public void afterCommit() {
                    run(deviceCode, "离线告警恢复独立事务", () ->
                            collectionAlarmWriter.markRecovered(deviceCode, "OFFLINE"));
                }
            });
        } else {
            run(deviceCode, "离线告警恢复", () ->
                    collectionAlarmWriter.markRecovered(deviceCode, "OFFLINE"));
        }
    }
}