smart_grid_iot/server/src/service/icc_event_subscribe.service.ts

import { Provide, Inject, Config } from "@midwayjs/core";
import { Context } from "@midwayjs/koa";
import { InjectRepository } from "@midwayjs/sequelize";
import { Repository } from "sequelize-typescript";
import DeviceAlarm from "../entity/device_alarm.entity";
import ElectricLine from "../entity/electric_line.entity";
import { getIccAccessToken, iccFetch } from "../task/icc_customer_auth";

/** 订阅注册结果 */
export interface IccRegisterResult {
	success: boolean;
	code?: string;
	errMsg?: string;
	raw?: any;
}

/** 回调事件处理统计 */
export interface IccEventHandleResult {
	/** 收到的通用事件总数 */
	received: number;
	/** 进入处理（category=alarm 且有 info）的事件数 */
	processed: number;
	/** 新增报警数 */
	inserted: number;
	/** 更新报警数 */
	updated: number;
	/** 未匹配到本系统线路的报警数 */
	unmatched: number;
}

/** 线路预取信息（回填 lineId/lineCode/lineName/customerId，避免 N+1） */
interface LineMatchInfo {
	lineId: number;
	lineName?: string;
	customerId: number;
}

/** ICC 事件订阅接口路径（version 固定 1.0.0，与 ICC 事件订阅文档一致） */
const ICC_EVENT_SUBSCRIBE_ENDPOINT = "/evo-apigw/evo-event/1.0.0/subscribe/mqinfo";
/** 查询订阅列表接口路径（version 固定 1.0.0） */
const ICC_EVENT_LIST_ENDPOINT = "/evo-apigw/evo-event/1.0.0/subscribe/subscribe-list";
/** HTTP 超时 60s（项目约定） */
const HTTP_TIMEOUT_MS = 60000;

/** 秒级时间戳转 Date（< 1e12 视为秒级 *1000；已毫秒级则原样）；非法返回 null */
function secondsToDate(value: unknown): Date | null {
	if (value === undefined || value === null || String(value).trim() === "") return null;
	const n = Number(value);
	if (!Number.isFinite(n) || n <= 0) return null;
	return new Date(n < 1e12 ? n * 1000 : n);
}

/** 数值解析：空 / null / 非法统一归一为 null */
function toInt(value: unknown): number | null {
	if (value === undefined || value === null || String(value).trim() === "") return null;
	const n = Number(value);
	return Number.isFinite(n) ? n : null;
}

/** 字符串 trim；空串归一为 null */
function toOptionalString(value: unknown): string | null {
	if (value === undefined || value === null) return null;
	const s = String(value).trim();
	return s === "" ? null : s;
}

/** 布尔解析：兼容 boolean / 0|1 / "true"|"false" / "1"|"0"；空值归一为 null */
function toBool(value: unknown): boolean | null {
	if (value === undefined || value === null || String(value).trim() === "") return null;
	if (typeof value === "boolean") return value;
	const s = String(value).trim().toLowerCase();
	if (s === "1" || s === "true" || s === "yes") return true;
	if (s === "0" || s === "false" || s === "no") return false;
	return null;
}

/**
 * ICC 事件订阅服务：
 *  1) registerSubscribe：向 ICC 事件订阅接口注册报警事件回调（monitor=monitorUrl?token=callbackToken，
 *     子系统 name/magic 取自配置 iccEvent），同名重复注册 ICC 会自动覆盖（保活语义）
 *  2) ensureSubscribe：保活入口，直接 registerSubscribe 同名覆盖（ICC 同名覆盖即保活），
 *     由每 30 分钟定时任务与手动接口调用
 *  3) handleCallback：处理 ICC 回调推送的通用事件，仅入库 category=alarm 的报警事件，
 *     按 alarm_code 幂等 upsert，uuid 相同重复推送跳过，按 nodeCode/deviceCode 匹配本系统线路
 */
@Provide()
export class IccEventSubscribeService {
	@Inject()
	ctx!: Context;

	@Config("iccEvent")
	iccEventConfig!: { monitorUrl: string; callbackToken: string; subscriberName: string; subscriberMagic: string };

	@InjectRepository(DeviceAlarm)
	alarmModel!: Repository<DeviceAlarm>;

	@InjectRepository(ElectricLine)
	lineModel!: Repository<ElectricLine>;

	/** 最近一次订阅注册结果（内存，供 subscribe-status 接口查询） */
	private lastRegisterResult: IccRegisterResult | null = null;
	private lastRegisterAt: Date | null = null;

	/**
	 * 向 ICC 注册报警事件订阅（同名重复注册 ICC 覆盖，保活语义）。
	 * 网络异常 / HTTP 错误 / ICC 业务失败均不抛错，返回失败状态并记录到内存。
	 */
	async registerSubscribe(): Promise<IccRegisterResult> {
		const baseUrl = process.env.ICC_BASE_URL?.trim();
		if (!baseUrl) {
			return this.remember({ success: false, code: "500", errMsg: "未配置 ICC_BASE_URL，请检查 server/.env" });
		}

		try {
			const token = await getIccAccessToken();
			const url = `${baseUrl.replace(/\/$/, "")}${ICC_EVENT_SUBSCRIBE_ENDPOINT}`;
			const body = {
				param: {
					monitors: [
						{
							monitor: `${this.iccEventConfig.monitorUrl}?token=${this.iccEventConfig.callbackToken}`,
							monitorType: "url",
							events: [{ category: "alarm", subscribeAll: 1, domainSubscribe: 2, eventType: 2 }]
						}
					],
					subsystem: {
						subsystemType: 0,
						name: this.iccEventConfig.subscriberName,
						magic: this.iccEventConfig.subscriberMagic
					}
				}
			};

			const res = await iccFetch(url, {
				method: "POST",
				headers: {
					"Content-Type": "application/json",
					Accept: "application/json",
					Authorization: `bearer ${token}`
				},
				body: JSON.stringify(body),
				signal: AbortSignal.timeout(HTTP_TIMEOUT_MS)
			});

			const payload: any = await res.json().catch(() => null);
			if (!res.ok) {
				return this.remember({ success: false, code: String(res.status), errMsg: `ICC 订阅接口 HTTP ${res.status}`, raw: payload });
			}

			const success = payload?.success === true || String(payload?.code ?? "") === "0";
			if (!success) {
				return this.remember({ success: false, code: String(payload?.code ?? ""), errMsg: String(payload?.errMsg ?? JSON.stringify(payload)), raw: payload });
			}
			return this.remember({ success: true, code: "0", errMsg: "", raw: payload });
		} catch (err: any) {
			const message = err instanceof Error ? err.message : String(err);
			return this.remember({ success: false, code: "500", errMsg: message });
		}
	}

	/**
	 * 保活入口：ICC 同名订阅会覆盖，因此直接 registerSubscribe 即可（保活语义）。
	 * 返回与 registerSubscribe 相同的注册结果。
	 */
	async ensureSubscribe(): Promise<IccRegisterResult> {
		return this.registerSubscribe();
	}

	/** 查询订阅状态：配置 + 最近一次注册结果（内存） */
	async getSubscribeStatus(): Promise<{
		config: { monitorUrl: string; subscriberName: string; subscriberMagic: string };
		lastRegisterResult: IccRegisterResult | null;
		lastRegisterAt: Date | null;
	}> {
		return {
			config: {
				monitorUrl: this.iccEventConfig.monitorUrl,
				subscriberName: this.iccEventConfig.subscriberName,
				subscriberMagic: this.iccEventConfig.subscriberMagic
			},
			lastRegisterResult: this.lastRegisterResult,
			lastRegisterAt: this.lastRegisterAt
		};
	}

	/**
	 * 处理 ICC 回调推送的通用事件：仅处理 category=alarm 且有 info 的事件。
	 *   - 按 info.nodeCode || info.deviceCode 匹配 electric_line.line_code，命中回填 lineId/lineCode/lineName/customerId
	 *   - uuid 相同（同一事件重复推送）跳过
	 *   - 按 alarm_code 幂等 upsert：存在则更新 alarmStat/isEvent/alarmTime/syncedAt 等，不存在则新增
	 */
	async handleCallback(body: any): Promise<IccEventHandleResult> {
		const result: IccEventHandleResult = { received: 0, processed: 0, inserted: 0, updated: 0, unmatched: 0 };

		const events = this.extractEvents(body);
		result.received = events.length;
		if (events.length === 0) return result;

		// 预取本系统已同步线路 → line_code → {lineId,lineName,customerId} 映射，避免 N+1
		const lines = await this.lineModel.findAll({
			attributes: ["id", "lineCode", "lineName", "customerId"]
		});
		const lineMap = new Map<string, LineMatchInfo>();
		for (const line of lines) {
			if (!line.lineCode) continue;
			lineMap.set(line.lineCode, {
				lineId: line.id,
				lineName: line.lineName || undefined,
				customerId: line.customerId
			});
		}

		for (const event of events) {
			// 通用事件格式：仅处理 category=alarm 且有 info 对象的事件
			if (!event || event.category !== "alarm") continue;
			const info = event.info;
			if (!info || typeof info !== "object") continue;
			result.processed++;

			const alarmCode = toOptionalString(info.alarmCode);
			if (!alarmCode) continue;

			// uuid 去重：同一 uuid 重复推送直接跳过（uuid 非唯一列，按存在性判断）
			const uuid = toOptionalString(event.uuid ?? info.uuid);
			if (uuid) {
				const byUuid = await this.alarmModel.findOne({ where: { uuid } });
				if (byUuid) continue;
			}

			// 按 nodeCode || deviceCode 匹配本系统线路（line_code）
			const lineFields: { lineId: number | null; lineCode: string | null; lineName: string | null; customerId: number | null } = {
				lineId: null,
				lineCode: null,
				lineName: null,
				customerId: null
			};
			const matchKey = toOptionalString(info.nodeCode) || toOptionalString(info.deviceCode);
			if (matchKey) {
				const lineInfo = lineMap.get(matchKey);
				if (lineInfo) {
					lineFields.lineId = lineInfo.lineId;
					lineFields.lineCode = matchKey;
					lineFields.lineName = lineInfo.lineName || null;
					lineFields.customerId = lineInfo.customerId;
				} else {
					result.unmatched++;
				}
			} else {
				result.unmatched++;
			}

			const base = {
				uuid,
				nodeCode: toOptionalString(info.nodeCode),
				deviceCode: toOptionalString(info.deviceCode),
				deviceName: toOptionalString(info.deviceName),
				orgCode: toOptionalString(info.orgCode),
				orgName: toOptionalString(info.orgName),
				alarmType: toInt(info.alarmType),
				alarmGrade: toInt(info.alarmGrade),
				alarmStat: toInt(info.alarmStat),
				isEvent: toBool(info.isEvent),
				alarmValue: toInt(info.alarmValue),
				alarmTime: secondsToDate(info.alarmDate)
			};

			// 按 alarm_code 幂等 upsert
			const existing = await this.alarmModel.findOne({ where: { alarmCode } });
			if (existing) {
				await existing.update({
					...base,
					...lineFields,
					syncedAt: new Date()
				} as any);
				result.updated++;
			} else {
				await this.alarmModel.create({
					alarmCode,
					...base,
					...lineFields,
					syncedAt: new Date()
				} as any);
				result.inserted++;
			}
		}

		return result;
	}

	/** 从回调 body 提取通用事件数组（兼容 data.events / events / data 数组 / body 数组 / body 单条事件 五种形态） */
	private extractEvents(body: any): any[] {
		try {
			if (!body || typeof body !== "object") return [];
			if (Array.isArray(body)) return body;
			if (Array.isArray(body.events)) return body.events;
			const data = body.data;
			if (data && Array.isArray(data.events)) return data.events;
			if (Array.isArray(data)) return data;
			// 兼容 ICC 直接推送单条通用事件对象（body 本身即事件）
			if (typeof body.category === "string" && body.info && typeof body.info === "object") return [body];
			return [];
		} catch {
			return [];
		}
	}

	/** 记录注册结果到内存并返回 */
	private remember(result: IccRegisterResult): IccRegisterResult {
		this.lastRegisterResult = result;
		this.lastRegisterAt = new Date();
		return result;
	}
}