Update Harbor client and gateway integration workflows

src/server/index.ts

@@ -16,7 +16,10 @@ import {
 import { createSingboxRuntime } from './singboxRuntime.js';
 import { tcpPing } from './ping.js';
 import {
+  buildDualChannelGatewayConfig,
   buildGatewayConfig,
+  dualChannelConfigMatchesApplied,
+  fingerprintSelectedOutbound,
   removeSingboxConfig,
   restoreSingboxConfig,
   writeSingboxConfig,
@@ -81,6 +84,13 @@ import { createConnectivityDiagnosticsRoute } from './http/routes/connectivityDi
 import { createGatewayPresenceRoute } from './http/routes/gatewayPresenceRoute.js';
 import { createSharedProxyRoute } from './http/routes/sharedProxyRoute.js';
 import { createVersionRoute } from './http/routes/versionRoute.js';
+import { createSingboxSelectorService } from './services/singboxSelectorService.js';
+import { createFailoverService } from './features/failover/failoverService.js';
+import { createFailoverRoute } from './http/routes/failoverRoute.js';
+import { createActivityJournalService } from './services/activityJournalService.js';
+import { createActivityJournalRoute } from './http/routes/activityJournalRoute.js';
+import { createDomainTrafficService, readSingboxConnections } from './services/domainTrafficService.js';
+import type { ActivityJournalEventInput } from '../shared/activityJournal.js';
 
 const MAX_BODY_BYTES = 1_000_000;
 const SUBSCRIPTION_REFRESH_INTERVAL_MS = 15 * 60 * 1000;
@@ -142,6 +152,14 @@ if (legacyCacheOwnerMismatch) {
   }
 }
 const stateStore = createStateStore(settings.statePath, { legacySubscriptionCache });
+const activityJournal = createActivityJournalService({ filePath: settings.activityJournalPath });
+const appendJournal = (event: ActivityJournalEventInput) => {
+  try {
+    activityJournal.append(event);
+  } catch (error) {
+    console.warn(`[journal] событие не сохранено: ${errorMessage(error)}`);
+  }
+};
 const deviceStore = createJsonStore<InventoryState>({
   filePath: settings.deviceStatePath,
   defaultValue: migrateDeviceInventoryState({}),
@@ -283,6 +301,37 @@ const deviceInventory = settings.appMode === 'gateway'
 const localConnectivityDiagnostics = !remoteDataplane
   ? createConnectivityDiagnosticsService({ proxyPort: settings.diagnosticsProxyPort })
   : null;
+const localFailoverDiagnostics = !remoteDataplane ? {
+  primary: createConnectivityDiagnosticsService({ proxyPort: settings.failoverPrimaryProxyPort }),
+  reserve: createConnectivityDiagnosticsService({ proxyPort: settings.failoverReserveProxyPort }),
+} : null;
+const localSelector = !remoteDataplane && settings.appMode === 'gateway'
+  ? createSingboxSelectorService({ port: settings.singboxApiPort })
+  : null;
+const localFailoverTraffic = !remoteDataplane && settings.appMode === 'gateway'
+  ? createDomainTrafficService({
+      observe: () => readSingboxConnections(settings.singboxApiPort),
+      devices: () => record(deviceInventory?.snapshot()).devices,
+    })
+  : null;
+let localFailoverTrafficTimer: NodeJS.Timeout | null = null;
+
+function setLocalFailoverActivityEnabled(enabled: boolean) {
+  if (!localFailoverTraffic) throw new Error('Failover activity недоступна');
+  if (!enabled) {
+    if (localFailoverTrafficTimer) clearInterval(localFailoverTrafficTimer);
+    localFailoverTrafficTimer = null;
+    localFailoverTraffic.disableActivity();
+    return;
+  }
+  localFailoverTraffic.enableActivity();
+  if (localFailoverTrafficTimer) return;
+  const refresh = () => localFailoverTraffic.refresh()
+    .catch((error: unknown) => console.warn(`[control] failover activity: ${errorMessage(error)}`));
+  void refresh();
+  localFailoverTrafficTimer = setInterval(refresh, 2_000);
+  localFailoverTrafficTimer.unref();
+}
 
 function requireLocalConnectivityDiagnostics() {
   if (!localConnectivityDiagnostics) throw new Error('Harbor local diagnostics are not configured');
@@ -355,6 +404,48 @@ const gatewayAutoService = createGatewayAutoService({
   onDiscoveryWarning: (reason) => console.warn(`[control] Gateway не используется: ${reason}`),
   onTimerError: (error) => console.warn(`[control] Gateway detection failed: ${errorMessage(error)}`),
 });
+const failoverDataplane = remoteDataplane ? {
+  checkConfig: (config: unknown) => requireRemoteRuntime().checkConfig(config),
+  runFailoverProbe: (role: 'primary' | 'reserve', services: unknown, target: unknown, timeoutMs: number) => (
+    requireRemoteRuntime().runFailoverProbe(role, services, target, timeoutMs)
+  ),
+  readFailoverSelector: () => requireRemoteRuntime().readFailoverSelector(),
+  selectFailoverRole: (role: 'primary' | 'reserve') => requireRemoteRuntime().selectFailoverRole(role),
+  setFailoverActivityEnabled: (enabled: boolean) => requireRemoteRuntime().setFailoverActivityEnabled(enabled),
+  readFailoverActivity: (threshold: number) => requireRemoteRuntime().readFailoverActivity(threshold),
+} : {
+  checkConfig: async (config: unknown) => singboxRuntime.checkConfig(config),
+  runFailoverProbe: async (role: 'primary' | 'reserve', services: unknown, target: unknown, timeoutMs: number) => {
+    if (!localFailoverDiagnostics) throw new Error('Failover diagnostics недоступна');
+    return localFailoverDiagnostics[role].runVpn({ services, target, timeoutMs });
+  },
+  readFailoverSelector: async () => {
+    if (!localSelector) throw new Error('Failover selector недоступен');
+    return localSelector.read();
+  },
+  selectFailoverRole: async (role: 'primary' | 'reserve') => {
+    if (!localSelector) throw new Error('Failover selector недоступен');
+    return localSelector.select(role);
+  },
+  setFailoverActivityEnabled: async (enabled: boolean) => setLocalFailoverActivityEnabled(enabled),
+  readFailoverActivity: async (threshold: number) => ({
+    activity: localFailoverTraffic?.activitySnapshot(threshold) || null,
+  }),
+};
+const failoverService = createFailoverService({
+  state: {
+    read: () => normalizeStoredState(stateStore.read()),
+    update: updateStoredState,
+  },
+  runtime: { isRunning: async () => Boolean((await singboxRuntime.refresh()).running) },
+  dataplane: failoverDataplane,
+  buildCandidate: buildFailoverCandidate,
+  serialize: serializeControl,
+  onWarning: (error) => console.warn(`[control] failover: ${errorMessage(error)}`),
+  onSwitch: (from, to, reason) => console.log(`[control] failover ${from} -> ${to}: ${reason}`),
+  onEvent: appendJournal,
+});
+const gatewayFailover = settings.appMode === 'gateway' ? failoverService : null;
 const stateService = createStateService({
   appMode: settings.appMode,
   readStoredState: () => stateStore.read(),
@@ -362,6 +453,7 @@ const stateService = createStateService({
   getGatewayAutoState: gatewayAutoService.read,
   getOperationState: () => operationState,
   configExists: () => fs.existsSync(settings.configPath),
+  getFailoverSnapshot: failoverService.snapshot,
 });
 const stateRoute = createStateRoute({
   stateService,
@@ -375,6 +467,14 @@ const gatewayAutoRoute = createGatewayAutoRoute({
   withOperation,
   readStatePayload: stateRoute.readPayload,
 });
+const failoverRoute = createFailoverRoute({
+  appMode: settings.appMode,
+  failover: failoverService,
+  readBody,
+  withOperation,
+  sendState: (res) => stateRoute.send(res),
+});
+const activityJournalRoute = createActivityJournalRoute({ journal: activityJournal });
 const deviceInventoryRoute = createDeviceInventoryRoute({
   deviceInventory,
   readBody,
@@ -427,9 +527,12 @@ const subscriptionService = createSubscriptionService({
     update: updateStoredState,
   },
   config: {
-    build: (subscriptionConfig, selectedServerId, routeRules) => (
-      buildActiveConfig(subscriptionConfig, selectedServerId, routeRules)
-    ),
+    build: (subscriptionConfig, selectedServerId, routeRules) => {
+      const state = normalizeStoredState(stateStore.read());
+      return state.appliedFailoverPolicy
+        ? buildFailoverCandidate({ ...state, routeRules }, 'applied').config
+        : buildActiveConfig(subscriptionConfig, selectedServerId, routeRules);
+    },
     read: () => fs.existsSync(settings.configPath)
       ? fs.readFileSync(settings.configPath, 'utf8')
       : null,
@@ -453,6 +556,12 @@ const subscriptionService = createSubscriptionService({
     clearInterval: (timer) => clearInterval(timer),
   },
   onRefreshError: (error) => console.warn(`[control] подписка не обновлена: ${errorMessage(error)}`),
+  onEvent: appendJournal,
+  failover: gatewayFailover ? {
+    reconcile: () => gatewayFailover.reconcile()
+      .catch((error) => console.warn(`[control] failover reconcile: ${errorMessage(error)}`)),
+    restoreAppliedActivation: gatewayFailover.restoreAppliedActivation,
+  } : undefined,
   now: () => new Date(),
 });
 const serverHealthRoute = createServerHealthRoute({
@@ -484,6 +593,14 @@ const connectionService = createConnectionService({
     isGatewayDirect: () => settings.appMode === 'client'
       && gatewayAutoService.read().mode === 'gateway-direct',
   },
+  failover: gatewayFailover ? {
+    build: buildFailoverCandidate,
+    prepareActivation: gatewayFailover.prepareActivation,
+    restoreAppliedActivation: gatewayFailover.restoreAppliedActivation,
+    reconcile: () => gatewayFailover.reconcile()
+      .catch((error) => console.warn(`[control] failover reconcile: ${errorMessage(error)}`)),
+  } : undefined,
+  onEvent: appendJournal,
   runtime: {
     isRunning: async () => Boolean((await singboxRuntime.refresh()).running),
     start: () => startSingbox(),
@@ -524,9 +641,12 @@ const routeRulesService = createRouteRulesService({
     readConfig: (profileId) => readProfileConfig(profileId),
   },
   config: {
-    build: (subscriptionConfig, selectedServerId, routeRules) => (
-      buildActiveConfig(subscriptionConfig, selectedServerId, routeRules)
-    ),
+    build: (subscriptionConfig, selectedServerId, routeRules) => {
+      const state = normalizeStoredState(stateStore.read());
+      return state.appliedFailoverPolicy
+        ? buildFailoverCandidate({ ...state, routeRules }, 'applied').config
+        : buildActiveConfig(subscriptionConfig, selectedServerId, routeRules);
+    },
     read: () => fs.existsSync(settings.configPath)
       ? fs.readFileSync(settings.configPath, 'utf8')
       : null,
@@ -548,6 +668,9 @@ const routeRulesService = createRouteRulesService({
   },
   serialize: serializeControl,
   runOperation: (operation) => withOperation('route-rules', operation),
+  afterApply: gatewayFailover ? () => gatewayFailover.reconcile()
+    .catch((error) => console.warn(`[control] failover reconcile: ${errorMessage(error)}`)) : undefined,
+  restoreAppliedActivation: gatewayFailover?.restoreAppliedActivation,
 });
 const routeRulesRoute = createRouteRulesRoute({
   routeRules: routeRulesService,
@@ -676,6 +799,47 @@ function buildActiveConfig(
   });
 }
 
+function buildFailoverCandidate(state: StoredState, source: 'desired' | 'applied' = 'desired') {
+  const policy = source === 'applied' ? state.appliedFailoverPolicy : state.failoverPolicy;
+  if (!policy) throw new HarborError('CONFIG_INVALID');
+  if (
+    policy.primary.profileId === policy.reserve.profileId
+    && policy.primary.serverId === policy.reserve.serverId
+  ) throw new HarborError('REQUEST_INVALID');
+  const channel = (role: 'primary' | 'reserve') => {
+    const target = policy[role];
+    const profile = state.profiles.find(({ id }) => id === target.profileId);
+    const server = profile?.servers.find(({ id }) => id === target.serverId);
+    if (!profile || !server || !profile.subscriptionConfig) throw new HarborError('SERVER_NOT_FOUND');
+    return { profile, server };
+  };
+  const primary = channel('primary');
+  const reserve = channel('reserve');
+  const applied = {
+    primary: policy.primary,
+    reserve: policy.reserve,
+    primaryConfigFingerprint: fingerprintSelectedOutbound(primary.profile.subscriptionConfig, primary.server.id),
+    reserveConfigFingerprint: fingerprintSelectedOutbound(reserve.profile.subscriptionConfig, reserve.server.id),
+  };
+  if (source === 'applied' && JSON.stringify(applied) !== JSON.stringify(state.appliedFailoverPolicy)) {
+    throw new HarborError('CONFIG_INVALID');
+  }
+  const defaultRole = source === 'applied'
+    && state.appliedProfileId === policy.reserve.profileId
+    && state.appliedServerId === policy.reserve.serverId
+    ? 'reserve'
+    : 'primary';
+  return {
+    config: buildDualChannelGatewayConfig({
+      primary: { subscriptionConfig: primary.profile.subscriptionConfig, selectedServerId: primary.server.id },
+      reserve: { subscriptionConfig: reserve.profile.subscriptionConfig, selectedServerId: reserve.server.id },
+    }, { routeRules: state.routeRules, defaultRole }),
+    applied,
+    primaryProfile: primary.profile,
+    primaryServer: primary.server,
+  };
+}
+
 const stopSingbox = () => singboxRuntime.stop();
 const startSingbox = () => singboxRuntime.apply();
 
@@ -689,7 +853,13 @@ function writeCurrentConfig() {
   const server = profile?.servers.find((candidate) => candidate.id === serverId);
   const subscriptionConfig = profile ? readProfileConfig(profile.id) : null;
   if (!profile || !server || !subscriptionConfig) return null;
-  const activeConfig = buildActiveConfig(subscriptionConfig, server.id);
+  let activeConfig: unknown;
+  if (state.appliedFailoverPolicy) {
+    const candidate = buildFailoverCandidate(state, 'applied');
+    activeConfig = candidate.config;
+  } else {
+    activeConfig = buildActiveConfig(subscriptionConfig, server.id);
+  }
   const previousConfig = fs.existsSync(settings.configPath)
     ? fs.readFileSync(settings.configPath, 'utf8')
     : null;
@@ -704,7 +874,7 @@ function writeCurrentConfig() {
     }
     throw error;
   }
-  return { profile, server };
+  return { profile, server, failoverApplied: state.appliedFailoverPolicy };
 }
 
 const CONFIG_PROXY_TYPES = new Set(['vless', 'vmess', 'trojan', 'shadowsocks', 'hysteria2']);
@@ -717,6 +887,13 @@ function currentConfigMatchesAppliedTarget(state: StoredState) {
   } catch {
     return false;
   }
+  if (state.appliedFailoverPolicy) {
+    const expectedRole = state.appliedProfileId === state.appliedFailoverPolicy.reserve.profileId
+      && state.appliedServerId === state.appliedFailoverPolicy.reserve.serverId
+      ? 'reserve'
+      : 'primary';
+    return dualChannelConfigMatchesApplied(config, state.appliedFailoverPolicy, expectedRole);
+  }
   const proxyOutbounds = (Array.isArray(config.outbounds) ? config.outbounds : [])
     .map(record)
     .filter((outbound) => CONFIG_PROXY_TYPES.has(String(outbound.type || '')));
@@ -750,6 +927,7 @@ async function reconcileStoppedBoot({ removeConfig = false } = {}) {
     || state.appliedProfileId
     || state.appliedServerId
     || state.appliedServerSnapshot
+    || state.appliedFailoverPolicy
   ) {
     updateStoredState((current) => ({
       ...current,
@@ -757,6 +935,7 @@ async function reconcileStoppedBoot({ removeConfig = false } = {}) {
       appliedProfileId: '',
       appliedServerId: '',
       appliedServerSnapshot: null,
+      appliedFailoverPolicy: null,
     }));
   }
 }
@@ -770,6 +949,8 @@ async function handleApi(req: IncomingMessage, res: ServerResponse) {
   if (await connectionRuntimeRoute.handle(req, res)) return;
   if (await routeRulesRoute.handle(req, res)) return;
   if (await gatewayAutoRoute.handle(req, res)) return;
+  if (await failoverRoute.handle(req, res)) return;
+  if (await activityJournalRoute.handle(req, res)) return;
   if (await connectivityDiagnosticsRoute.handle(req, res)) return;
 
   if (await versionRoute.handle(req, res)) return;
@@ -822,6 +1003,7 @@ async function shutdown() {
   subscriptionService.stopAutoRefresh();
   gatewayAutoService.stopDiscovery();
   if (deviceDiscoveryTimer) clearInterval(deviceDiscoveryTimer);
+  await gatewayFailover?.shutdown().catch((error) => console.warn(`[control] failover shutdown: ${errorMessage(error)}`));
   await serializeControl(() => singboxRuntime.shutdown());
   process.exit(0);
 }
@@ -846,8 +1028,14 @@ if (bootWantsRunning) {
     && currentConfigMatchesAppliedTarget(normalizeStoredState(stateStore.read()));
   if (target || canReuseCurrentConfig) {
     await startSingbox()
-      .then(() => {
+      .then(async () => {
         const current = normalizeStoredState(stateStore.read());
+        const bootRole = current.appliedFailoverPolicy
+          && current.appliedProfileId === current.appliedFailoverPolicy.reserve.profileId
+          && current.appliedServerId === current.appliedFailoverPolicy.reserve.serverId
+          ? 'reserve'
+          : current.appliedFailoverPolicy ? 'primary' : null;
+        if (bootRole) await failoverDataplane.selectFailoverRole(bootRole);
         const appliedProfile = target?.profile
           || current.profiles.find((profile) => profile.id === current.appliedProfileId);
         const appliedServer = target?.server
@@ -866,7 +1054,10 @@ if (bootWantsRunning) {
           }));
         }
       })
-      .catch((error: unknown) => console.warn(`[control] sing-box не запущен: ${errorMessage(error)}`));
+      .catch(async (error: unknown) => {
+        console.warn(`[control] sing-box не запущен: ${errorMessage(error)}`);
+        await reconcileStoppedBoot();
+      });
   } else {
     await reconcileStoppedBoot({ removeConfig: true });
   }
@@ -874,6 +1065,9 @@ if (bootWantsRunning) {
   await reconcileStoppedBoot();
 }
 
+await gatewayFailover?.reconcile()
+  .catch((error) => console.warn(`[control] failover reconcile: ${errorMessage(error)}`));
+
 if (deviceInventory) {
   await deviceInventory.reconcilePolicies()
     .catch((error: unknown) => console.warn(`[control] device policy не применена: ${errorMessage(error)}`));