Web_BLS_Heartbeat_Server/src/stats/statsManager.js

class StatsCounters {
  constructor() {
    this._minuteBuf = new SharedArrayBuffer(BigInt64Array.BYTES_PER_ELEMENT * 4);
    this._minute = new BigInt64Array(this._minuteBuf);
  }

  incDbWritten(n = 1) {
    const v = BigInt(Math.max(0, Number(n) || 0));
    if (v === 0n) return;
    Atomics.add(this._minute, 0, v);
  }

  incFiltered(n = 1) {
    const v = BigInt(Math.max(0, Number(n) || 0));
    if (v === 0n) return;
    Atomics.add(this._minute, 1, v);
  }

  incKafkaPulled(n = 1) {
    const v = BigInt(Math.max(0, Number(n) || 0));
    if (v === 0n) return;
    Atomics.add(this._minute, 2, v);
  }

  incDbWriteFailed(n = 1) {
    const v = BigInt(Math.max(0, Number(n) || 0));
    if (v === 0n) return;
    Atomics.add(this._minute, 3, v);
  }

  snapshotAndResetMinute() {
    const dbWritten = Atomics.exchange(this._minute, 0, 0n);
    const filtered = Atomics.exchange(this._minute, 1, 0n);
    const kafkaPulled = Atomics.exchange(this._minute, 2, 0n);
    const dbWriteFailed = Atomics.exchange(this._minute, 3, 0n);
    return { dbWritten, filtered, kafkaPulled, dbWriteFailed };
  }
}

const pad2 = (n) => String(n).padStart(2, '0');
const pad3 = (n) => String(n).padStart(3, '0');

const formatTimestamp = (d) => {
  const year = d.getFullYear();
  const month = pad2(d.getMonth() + 1);
  const day = pad2(d.getDate());
  const hour = pad2(d.getHours());
  const minute = pad2(d.getMinutes());
  const second = pad2(d.getSeconds());
  const ms = pad3(d.getMilliseconds());
  return `${year}-${month}-${day} ${hour}:${minute}:${second}.${ms}`;
};

class StatsReporter {
  constructor({ redis, stats }) {
    this.redis = redis;
    this.stats = stats;
    this._timer = null;
    this._running = false;
    this._lastFlushMinute = null;
  }

  start() {
    if (this._running) return;
    this._running = true;
    this._scheduleNext();
  }

  stop() {
    this._running = false;
    if (this._timer) {
      clearTimeout(this._timer);
      this._timer = null;
    }
  }

  flushOnce() {
    if (!this.redis?.isEnabled?.()) return;
    const now = Date.now();
    const minuteKey = Math.floor(now / 60_000);
    if (this._lastFlushMinute === minuteKey) {
      return;
    }
    const { dbWritten, filtered, kafkaPulled, dbWriteFailed } = this.stats.snapshotAndResetMinute();
    this._lastFlushMinute = minuteKey;
    const ts = formatTimestamp(new Date());
    this.redis.pushConsoleLog?.({ level: 'info', message: `[STATS] ${ts} 数据库写入量: ${dbWritten}条`, metadata: { module: 'stats' } });
    this.redis.pushConsoleLog?.({ level: 'info', message: `[STATS] ${ts} 数据库写入失败量: ${dbWriteFailed}条`, metadata: { module: 'stats' } });
    this.redis.pushConsoleLog?.({ level: 'info', message: `[STATS] ${ts} 数据过滤量: ${filtered}条`, metadata: { module: 'stats' } });
    this.redis.pushConsoleLog?.({ level: 'info', message: `[STATS] ${ts} Kafka拉取量: ${kafkaPulled}条`, metadata: { module: 'stats' } });
  }

  _scheduleNext() {
    if (!this._running) return;
    if (this._timer) return;

    const now = Date.now();
    const delay = 60_000 - (now % 60_000);
    this._timer = setTimeout(() => {
      this._timer = null;
      try {
        this.flushOnce();
      } catch (err) {
        this.redis?.pushConsoleLog?.({
          level: 'warn',
          message: `[ERROR] ${formatTimestamp(new Date())} 统计任务异常: ${String(err?.message ?? err)}`,
          metadata: { module: 'stats' },
        });
      } finally {
        this._scheduleNext();
      }
    }, delay);
  }
}

export { StatsCounters, StatsReporter, formatTimestamp };