pipeline/backend-node/src/services/collector/edgeStatusReporter.ts

/**
 * Edge Status / Event Reporter
 *
 * IDC TimescaleDB 의 edge_status / edge_events 테이블에 직접 INSERT.
 * (기존: data-collector → Kafka → kafka-to-central-mqtt → MQTT → IDC consumer → edge_status)
 * (교체: pipeline-backend → IDC TimescaleDB 직접)
 *
 * 환경변수:
 *   PIPELINE_EDGE_REPORTER=true           기동 시 시작
 *   PIPELINE_EDGE_TARGET_DB_ID=3          external_db_connections.id
 *   PIPELINE_EDGE_ID=edge-0f4d04ed
 *   PIPELINE_EDGE_COMPANY_ID=spifox       edge_status.company_id (varchar)
 *   PIPELINE_EDGE_COMPANY_UUID=...        edge_events.company_id (uuid)
 *   PIPELINE_EDGE_STATUS_TABLE=edge_status_1
 *   PIPELINE_EDGE_EVENTS_TABLE=edge_events_1
 *   PIPELINE_EDGE_STATUS_INTERVAL_SEC=60
 */
import os from "os";
import { logger } from "../../utils/logger";
import { executeExternalQuery } from "../externalDbHelper";

interface ReporterConfig {
  targetDbId: number;
  edgeId: string;
  companyId: string;
  companyUuid: string | null;
  statusTable: string;
  eventsTable: string;
  intervalSec: number;
}

let cfg: ReporterConfig | null = null;
let statusTimer: NodeJS.Timeout | null = null;

function loadConfig(): ReporterConfig | null {
  if (process.env.PIPELINE_EDGE_REPORTER !== "true") return null;
  const targetDbId = Number(process.env.PIPELINE_EDGE_TARGET_DB_ID);
  const edgeId = process.env.PIPELINE_EDGE_ID;
  const companyId = process.env.PIPELINE_EDGE_COMPANY_ID;
  if (!Number.isFinite(targetDbId) || !edgeId || !companyId) {
    logger.warn("[EdgeReporter] 환경변수 누락 (TARGET_DB_ID/EDGE_ID/COMPANY_ID) — 비활성");
    return null;
  }
  return {
    targetDbId,
    edgeId,
    companyId,
    companyUuid: process.env.PIPELINE_EDGE_COMPANY_UUID || null,
    statusTable: process.env.PIPELINE_EDGE_STATUS_TABLE || "edge_status_1",
    eventsTable: process.env.PIPELINE_EDGE_EVENTS_TABLE || "edge_events_1",
    intervalSec: Number(process.env.PIPELINE_EDGE_STATUS_INTERVAL_SEC) || 60,
  };
}

function primaryIpAddress(): string | null {
  const ifaces = os.networkInterfaces();
  for (const name of Object.keys(ifaces)) {
    for (const it of ifaces[name] || []) {
      if (it.family === "IPv4" && !it.internal) return it.address;
    }
  }
  return null;
}

async function writeStatusRow(status: "online" | "offline"): Promise<void> {
  if (!cfg) return;
  try {
    await executeExternalQuery(
      cfg.targetDbId,
      `INSERT INTO ${cfg.statusTable} (time, company_id, edge_id, status, ip_address, firmware_version, metadata)
       VALUES (NOW(), $1, $2, $3, $4, $5, $6::jsonb)`,
      [
        cfg.companyId,
        cfg.edgeId,
        status,
        primaryIpAddress(),
        process.env.PIPELINE_FIRMWARE_VERSION || null,
        JSON.stringify({ source: "pipeline-backend", host: os.hostname() }),
      ],
    );
  } catch (err) {
    logger.warn(`[EdgeReporter] status INSERT 실패: ${(err as Error).message}`);
  }
}

export async function recordEdgeEvent(
  eventType: string,
  eventCode: string,
  message: string,
  severity = 3,
  metadata: Record<string, unknown> = {},
): Promise<void> {
  if (!cfg) return;
  if (!cfg.companyUuid) {
    // edge_events.company_id 는 UUID 타입이라 UUID 없으면 INSERT 불가 → 경고만
    logger.debug("[EdgeReporter] event 기록 스킵: PIPELINE_EDGE_COMPANY_UUID 미설정");
    return;
  }
  try {
    await executeExternalQuery(
      cfg.targetDbId,
      `INSERT INTO ${cfg.eventsTable} (time, company_id, edge_id, event_type, event_code, message, severity, acknowledged, metadata)
       VALUES (NOW(), $1::uuid, $2, $3, $4, $5, $6, false, $7::jsonb)`,
      [
        cfg.companyUuid,
        cfg.edgeId,
        eventType,
        eventCode,
        message,
        severity,
        JSON.stringify({ source: "pipeline-backend", ...metadata }),
      ],
    );
    logger.info(`[EdgeReporter] 이벤트 기록: ${eventType}/${eventCode} — ${message}`);
  } catch (err) {
    logger.warn(`[EdgeReporter] event INSERT 실패: ${(err as Error).message}`);
  }
}

export async function startEdgeReporter(): Promise<void> {
  cfg = loadConfig();
  if (!cfg) {
    logger.info("[EdgeReporter] 비활성 (PIPELINE_EDGE_REPORTER!=true)");
    return;
  }
  logger.info(
    `[EdgeReporter] 시작 target_db=${cfg.targetDbId} edge=${cfg.edgeId} ` +
      `status_tbl=${cfg.statusTable} events_tbl=${cfg.eventsTable} interval=${cfg.intervalSec}s`,
  );

  // 기동 이벤트 + 첫 하트비트
  await recordEdgeEvent(
    "pipeline",
    "startup",
    "pipeline-backend 기동 — Edge Reporter 활성",
    3,
  );
  await writeStatusRow("online");

  statusTimer = setInterval(() => {
    writeStatusRow("online").catch(() => undefined);
  }, cfg.intervalSec * 1000);
}

export async function stopEdgeReporter(): Promise<void> {
  if (statusTimer) {
    clearInterval(statusTimer);
    statusTimer = null;
  }
  if (cfg) {
    await writeStatusRow("offline").catch(() => undefined);
    await recordEdgeEvent("pipeline", "shutdown", "pipeline-backend 정지", 3).catch(
      () => undefined,
    );
  }
  cfg = null;
}

export function isEdgeReporterActive(): boolean {
  return cfg !== null;
}