multi-simadmin/apps/api/src/application/connections/fleet-heartbeat.ts

import type { ConnectionProbe, ConnectionState } from './connection-probe.js';
import type { ConnectionSettingsService } from './connection-settings-service.js';
import type { InstanceService } from '../instances/instance-service.js';

/** Reachability class the console and the notification rules both speak in. */
export type FleetObservedState = 'online' | 'auth-required' | 'offline';

/**
 * Operator-facing copy shared by the notification templates so every channel reads alike.
 * `FLEET_STATE_NAMES` describes a steady state, `FLEET_STATE_EVENTS` describes arriving at it.
 */
export const FLEET_STATE_NAMES: Readonly<Record<FleetObservedState, string>> = Object.freeze({
  online: '在线',
  'auth-required': '需要认证',
  offline: '离线',
});

export const FLEET_STATE_EVENTS: Readonly<Record<FleetObservedState, string>> = Object.freeze({
  online: '恢复在线',
  'auth-required': '设备会话已失效',
  offline: '已离线',
});

export const FLEET_STATE_CODES: Readonly<Record<FleetObservedState, string>> = Object.freeze({
  online: 'online',
  'auth-required': 'auth_required',
  offline: 'offline',
});

export interface FleetStateTransition {
  readonly instanceId: string;
  readonly from: FleetObservedState;
  readonly to: FleetObservedState;
}

export interface FleetHeartbeatSummary {
  readonly probed: number;
  readonly failed: number;
  readonly transitions: readonly FleetStateTransition[];
  readonly startedAt: string;
  readonly finishedAt: string;
}

export interface FleetHeartbeatOptions {
  readonly instances: InstanceService;
  readonly probe: ConnectionProbe;
  readonly settings: ConnectionSettingsService;
  readonly now?: () => Date;
  /** Devices touched at the same time; the control plane talks to LAN hosts, not the internet. */
  readonly concurrency?: number;
  readonly pageSize?: number;
  /**
   * Called once per beat with every state change the beat revealed. Reachability is derived
   * from snapshot expiry, so a single lost probe never fires anything; a node has to stay
   * down for the whole offline window before it counts as a transition.
   */
  readonly onTransition?: (transitions: readonly FleetStateTransition[]) => Promise<void>;
}

const DEFAULT_CONCURRENCY = 6;
const DEFAULT_PAGE_SIZE = 100;
const MAX_PAGES = 100;

function observedState(state: ConnectionState | undefined): FleetObservedState {
  if (!state?.reachable) return 'offline';
  return state.authenticated ? 'online' : 'auth-required';
}

/**
 * Keeps device online state honest without waiting for an operator to open a page: every beat
 * probes each instance once and lets the snapshot journal carry the result until it expires.
 */
export class FleetHeartbeatCoordinator {
  readonly #instances: InstanceService;
  readonly #probe: ConnectionProbe;
  readonly #settings: ConnectionSettingsService;
  readonly #now: () => Date;
  readonly #concurrency: number;
  readonly #pageSize: number;
  readonly #onTransition:
    | ((transitions: readonly FleetStateTransition[]) => Promise<void>)
    | undefined;
  #timer: ReturnType<typeof setTimeout> | undefined;
  #running: Promise<FleetHeartbeatSummary> | undefined;
  #stopped = false;
  /** States seen by the previous beat; undefined until the first beat establishes a baseline. */
  #previous: ReadonlyMap<string, FleetObservedState> | undefined;

  constructor(options: FleetHeartbeatOptions) {
    this.#instances = options.instances;
    this.#probe = options.probe;
    this.#settings = options.settings;
    this.#now = options.now ?? (() => new Date());
    const concurrency = options.concurrency ?? DEFAULT_CONCURRENCY;
    if (!Number.isSafeInteger(concurrency) || concurrency < 1 || concurrency > 16)
      throw new RangeError('concurrency must be between 1 and 16');
    const pageSize = options.pageSize ?? DEFAULT_PAGE_SIZE;
    if (!Number.isSafeInteger(pageSize) || pageSize < 1 || pageSize > 100)
      throw new RangeError('pageSize must be between 1 and 100');
    this.#concurrency = concurrency;
    this.#pageSize = pageSize;
    this.#onTransition = options.onTransition;
  }

  start(): void {
    this.#stopped = false;
    if (this.#timer) return;
    this.#schedule();
  }

  stop(): void {
    this.#stopped = true;
    if (this.#timer) clearTimeout(this.#timer);
    this.#timer = undefined;
  }

  /** Runs a beat on demand; concurrent callers share the same pass. */
  runOnce(): Promise<FleetHeartbeatSummary> {
    if (this.#running) return this.#running;
    const pending = this.#beat().finally(() => {
      if (this.#running === pending) this.#running = undefined;
    });
    this.#running = pending;
    return pending;
  }

  #schedule(): void {
    if (this.#stopped) return;
    const delay = this.#settings.heartbeatMs;
    this.#timer = setTimeout(() => {
      this.#timer = undefined;
      void this.runOnce()
        .catch(() => undefined)
        .finally(() => this.#schedule());
    }, delay);
    this.#timer.unref?.();
  }

  async #beat(): Promise<FleetHeartbeatSummary> {
    const startedAt = this.#now().toISOString();
    const ids: string[] = [];
    for (let page = 1; page <= MAX_PAGES; page += 1) {
      const current = await this.#instances.list({ page, pageSize: this.#pageSize });
      for (const instance of current.items) if (!ids.includes(instance.id)) ids.push(instance.id);
      if (current.items.length < this.#pageSize) break;
    }
    const before = this.#previous;
    let probed = 0;
    let failed = 0;
    let cursor = 0;
    const workers = Array.from(
      { length: Math.min(this.#concurrency, ids.length) },
      async (): Promise<void> => {
        for (;;) {
          const index = cursor;
          cursor += 1;
          if (index >= ids.length) return;
          try {
            await this.#probe.test(ids[index] as string);
            probed += 1;
          } catch {
            failed += 1;
          }
        }
      },
    );
    await Promise.all(workers);
    const transitions = this.#diff(before, ids);
    if (transitions.length > 0) await this.#onTransition?.(transitions);
    return { probed, failed, transitions, startedAt, finishedAt: this.#now().toISOString() };
  }

  /** Snapshot of the given instances, including ones that have never been probed. */
  #states(ids: readonly string[]): ReadonlyMap<string, FleetObservedState> {
    const reachability = this.#probe.reachability();
    const states = new Map<string, FleetObservedState>();
    for (const id of ids) {
      const state = reachability.get(id);
      states.set(id, observedState(state));
    }
    return states;
  }

  /**
   * The first beat only records a baseline. Without one, every instance would start as
   * "offline" and then announce itself as recovered, so a restart would flood the queue.
   */
  #diff(
    before: ReadonlyMap<string, FleetObservedState> | undefined,
    ids: readonly string[],
  ): readonly FleetStateTransition[] {
    const after = this.#states(ids);
    this.#previous = after;
    if (before === undefined) return [];
    const transitions: FleetStateTransition[] = [];
    for (const [id, to] of after) {
      const from = before.get(id);
      if (from === undefined || from === to) continue;
      transitions.push({ instanceId: id, from, to });
    }
    return transitions;
  }
}