Refactor VPN proxy components and update related behavior

src/server/index.ts

@@ -28,13 +28,19 @@ import {
   selectRefreshedServer,
 } from './subscription.js';
 import {
+  desiredProfile,
   normalizeStoredState,
   type OperationState,
   type RouteRule,
   type StoredState,
 } from '../shared/contracts/state.js';
+import { serverIdentityKey } from '../shared/serverIdentity.js';
 import { HarborError, normalizeHarborError } from '../shared/errors.js';
-import { createJsonStore, createStateStore } from './services/stateStore.js';
+import {
+  atomicWriteFile,
+  createJsonStore,
+  createStateStore,
+} from './services/stateStore.js';
 import { createDevicePolicyService } from './services/devicePolicyService.js';
 import {
   createDeviceInventoryService,
@@ -93,11 +99,49 @@ function errorMessage(error: unknown) {
 
 fs.mkdirSync(settings.dataDir, { recursive: true });
 
-const stateStore = createStateStore(settings.statePath);
+const stateFileExisted = fs.existsSync(settings.statePath);
+const legacyStateBytes = stateFileExisted
+  ? fs.readFileSync(settings.statePath, 'utf8')
+  : null;
+let legacyStateRecord: Record<string, unknown> = {};
+try {
+  legacyStateRecord = record(legacyStateBytes === null ? null : JSON.parse(legacyStateBytes));
+} catch {}
+const legacyStateVersion = Number.isSafeInteger(legacyStateRecord.schemaVersion)
+  ? Number(legacyStateRecord.schemaVersion)
+  : 0;
+const legacySubscriptionCacheBytes = fs.existsSync(settings.subscriptionCachePath)
+  ? fs.readFileSync(settings.subscriptionCachePath, 'utf8')
+  : null;
 const subscriptionCacheStore = createJsonStore({
   filePath: settings.subscriptionCachePath,
   defaultValue: null,
 });
+const rawLegacySubscriptionCache = subscriptionCacheStore.read();
+const legacyCacheRecord = record(rawLegacySubscriptionCache);
+const legacyStateSubscriptionUrl = String(legacyStateRecord.subscriptionUrl || '').trim();
+const legacyCacheSubscriptionUrl = String(legacyCacheRecord.url || '').trim();
+const legacyCacheOwnerMismatch = legacyStateVersion < 5
+  && Boolean(legacyCacheRecord.config)
+  && (legacyStateSubscriptionUrl
+    ? legacyCacheSubscriptionUrl !== legacyStateSubscriptionUrl
+    : !legacyCacheSubscriptionUrl);
+let legacySubscriptionCache = rawLegacySubscriptionCache;
+let legacySubscriptionCacheRejected = Boolean(subscriptionCacheStore.recovery);
+if (legacyCacheOwnerMismatch) {
+  legacySubscriptionCache = null;
+} else if (legacyCacheRecord.config) {
+  try {
+    legacySubscriptionCache = {
+      ...legacyCacheRecord,
+      ...normalizeSubscriptionConfig(legacyCacheRecord.config),
+    };
+  } catch {
+    legacySubscriptionCache = null;
+    legacySubscriptionCacheRejected = true;
+  }
+}
+const stateStore = createStateStore(settings.statePath, { legacySubscriptionCache });
 const deviceStore = createJsonStore<InventoryState>({
   filePath: settings.deviceStatePath,
   defaultValue: migrateDeviceInventoryState({}),
@@ -112,32 +156,77 @@ if (deviceStore.migration) {
 if (deviceStore.recovery) {
   console.warn(`[storage] corrupt devices recovered; backup: ${deviceStore.recovery.backupPath}`);
 }
-let cacheRecoveryLogged = false;
-
-function readRawSubscriptionCache() {
-  const cached = subscriptionCacheStore.read();
-  if (subscriptionCacheStore.recovery && !cacheRecoveryLogged) {
-    cacheRecoveryLogged = true;
-    console.warn(`[storage] corrupt subscription cache recovered; backup: ${subscriptionCacheStore.recovery.backupPath}`);
-  }
-  return cached;
-}
-
-function readSubscriptionCache() {
-  const raw = readRawSubscriptionCache();
-  const cached = record(raw);
-  return cached.config
-    ? { ...cached, ...normalizeSubscriptionConfig(cached.config), _persisted: raw }
-    : raw && typeof raw === 'object' && !Array.isArray(raw) ? cached : null;
-}
-
-const initialStoredState = stateStore.read();
+let initialStoredState = stateStore.read();
 if (stateStore.migration) {
   console.log(`[storage] state migrated to v${stateStore.migration.toVersion}; backup: ${stateStore.migration.backupPath}`);
 }
 if (stateStore.recovery) {
   console.warn(`[storage] corrupt state recovered; backup: ${stateStore.recovery.backupPath}`);
 }
+if (subscriptionCacheStore.recovery) {
+  console.warn(`[storage] corrupt subscription cache recovered; backup: ${subscriptionCacheStore.recovery.backupPath}`);
+}
+const rejectedLegacyMigration = legacySubscriptionCacheRejected
+  && (
+    !stateFileExisted
+    || Boolean(stateStore.recovery)
+    || Boolean(stateStore.migration && stateStore.migration.fromVersion < 5)
+  );
+const mismatchedLegacyMigration = legacyCacheOwnerMismatch
+  && (
+    !stateFileExisted
+    || Boolean(stateStore.recovery)
+    || Boolean(stateStore.migration && stateStore.migration.fromVersion < 5)
+  );
+if (rejectedLegacyMigration) {
+  initialStoredState = stateStore.update((state) => ({
+    ...state,
+    profiles: [],
+    desiredProfileId: '',
+    appliedProfileId: '',
+    appliedServerId: '',
+    appliedServerSnapshot: null,
+    connectionDesired: 'stopped',
+  }));
+  removeSingboxConfig();
+} else if (mismatchedLegacyMigration) {
+  initialStoredState = stateStore.update((state) => ({
+    ...state,
+    appliedProfileId: '',
+    appliedServerId: '',
+    appliedServerSnapshot: null,
+    connectionDesired: 'stopped',
+  }));
+  removeSingboxConfig();
+}
+if (
+  legacySubscriptionCacheBytes !== null
+  && (
+    Boolean(subscriptionCacheStore.recovery)
+    || (
+      Boolean(legacyCacheRecord.config)
+      && (
+        legacySubscriptionCacheRejected
+        || legacyCacheOwnerMismatch
+        || normalizeStoredState(initialStoredState).profiles.some((profile) => profile.subscriptionConfig)
+      )
+    )
+  )
+) {
+  const backupPath = subscriptionCacheStore.recovery?.backupPath
+    || `${settings.subscriptionCachePath}.backup-v1-${new Date().toISOString().replace(/[:.]/g, '-')}`;
+  if (!subscriptionCacheStore.recovery) atomicWriteFile(backupPath, legacySubscriptionCacheBytes);
+  subscriptionCacheStore.remove();
+  console.log(`[storage] legacy subscription cache migrated; backup: ${backupPath}`);
+}
+
+function readProfileConfig(profileId = '') {
+  const state = normalizeStoredState(stateStore.read());
+  const profile = profileId
+    ? state.profiles.find((candidate) => candidate.id === profileId)
+    : desiredProfile(state);
+  return profile?.subscriptionConfig || null;
+}
 
 const remoteDataplane = settings.appMode === 'gateway' && Boolean(process.env.DATAPLANE_SOCKET);
 const versionInfo = buildVersionInfo(settings.appMode);
@@ -215,7 +304,7 @@ const gatewayAutoService = createGatewayAutoService({
     update: updateStoredState,
   },
   subscription: {
-    readConfig: () => readSubscriptionCache()?.config || null,
+    readConfig: (profileId) => readProfileConfig(profileId),
   },
   config: {
     build: (subscriptionConfig, selectedServerId, routeRules, gatewayAuto) => (
@@ -238,6 +327,7 @@ const gatewayAutoService = createGatewayAutoService({
       { preMutationErrorCodes: remoteDataplane ? [] : ['CONFIG_INVALID'] },
     ),
     restoreRunning: () => startSingbox(),
+    stopCommand: () => captureRuntimeCommand(() => stopSingbox()),
   },
   discovery: {
     readHostNetwork: () => readHostNetworkState(settings.hostNetworkStatePath),
@@ -291,7 +381,7 @@ const deviceInventoryRoute = createDeviceInventoryRoute({
 });
 const prometheusMetricsRoute = createPrometheusMetricsRoute({ deviceInventory });
 const connectivityDiagnostics = createConnectivityDiagnosticsUseCase({
-  readState: () => stateStore.read(),
+  readState: () => normalizeStoredState(stateStore.read()),
   runDiagnostics: async (services, target) => remoteDataplane
     ? requireRemoteRuntime().runConnectivityDiagnostics(services, target)
     : requireLocalConnectivityDiagnostics().run({
@@ -306,7 +396,7 @@ const connectivityDiagnosticsRoute = createConnectivityDiagnosticsRoute({
 });
 const gatewayPresenceRoute = createGatewayPresenceRoute({
   appMode: settings.appMode,
-  readState: () => stateStore.read(),
+  readState: () => normalizeStoredState(stateStore.read()),
   getHwid,
 });
 const sharedProxyRoute = createSharedProxyRoute({
@@ -332,11 +422,6 @@ const subscriptionService = createSubscriptionService({
     read: () => normalizeStoredState(stateStore.read()),
     update: updateStoredState,
   },
-  cache: {
-    read: readRawSubscriptionCache,
-    write: (value) => { subscriptionCacheStore.write(value); },
-    remove: () => subscriptionCacheStore.remove(),
-  },
   config: {
     build: (subscriptionConfig, selectedServerId, routeRules) => (
       buildActiveConfig(subscriptionConfig, selectedServerId, routeRules)
@@ -364,16 +449,12 @@ const subscriptionService = createSubscriptionService({
     clearInterval: (timer) => clearInterval(timer),
   },
   onRefreshError: (error) => console.warn(`[control] подписка не обновлена: ${errorMessage(error)}`),
-});
-const subscriptionMutationRoute = createSubscriptionMutationRoute({
-  subscriptionService,
-  readBody,
-  withOperation,
-  sendState: (res, extra) => stateRoute.send(res, extra),
+  now: () => new Date(),
 });
 const serverHealthRoute = createServerHealthRoute({
   serverHealth: createServerHealthService({
-    readServers: () => normalizeStoredState(stateStore.read()).servers,
+    readProfiles: () => normalizeStoredState(stateStore.read()).profiles,
+    readDesiredProfileId: () => normalizeStoredState(stateStore.read()).desiredProfileId,
     ping: tcpPing,
   }),
   readBody,
@@ -384,11 +465,7 @@ const connectionService = createConnectionService({
     read: () => normalizeStoredState(stateStore.read()),
     update: updateStoredState,
   },
-  subscription: {
-    readConfig: () => readSubscriptionCache()?.config || null,
-  },
   config: {
-    exists: () => fs.existsSync(settings.configPath),
     build: (subscriptionConfig, selectedServerId, routeRules) => (
       buildActiveConfig(subscriptionConfig, selectedServerId, routeRules)
     ),
@@ -399,6 +476,10 @@ const connectionService = createConnectionService({
     restore: restoreSingboxConfig,
     remove: removeSingboxConfig,
   },
+  route: {
+    isGatewayDirect: () => settings.appMode === 'client'
+      && gatewayAutoService.read().mode === 'gateway-direct',
+  },
   runtime: {
     isRunning: async () => Boolean((await singboxRuntime.refresh()).running),
     start: () => startSingbox(),
@@ -412,6 +493,13 @@ const connectionService = createConnectionService({
   serialize: serializeControl,
   now: () => new Date(),
 });
+const subscriptionMutationRoute = createSubscriptionMutationRoute({
+  subscriptionService,
+  connection: connectionService,
+  readBody,
+  withOperation,
+  sendState: (res, extra) => stateRoute.send(res, extra),
+});
 const serverApplyRoute = createServerApplyRoute({
   connection: connectionService,
   readBody,
@@ -429,7 +517,7 @@ const routeRulesService = createRouteRulesService({
     update: updateStoredState,
   },
   subscription: {
-    readConfig: () => readSubscriptionCache()?.config || null,
+    readConfig: (profileId) => readProfileConfig(profileId),
   },
   config: {
     build: (subscriptionConfig, selectedServerId, routeRules) => (
@@ -460,27 +548,64 @@ const routeRulesRoute = createRouteRulesRoute({
 });
 
 function updateStoredState(update: (state: StoredState) => Record<string, unknown>) {
-  return stateStore.update((stored) => {
+  return normalizeStoredState(stateStore.update((stored) => {
     const current = normalizeStoredState(stored);
     const schemaVersion = stored.schemaVersion;
     const next = normalizeStoredState({ schemaVersion, ...update(current) });
     revision = Math.max(revision, current.revision) + 1;
     next.revision = revision;
     return { ...next, schemaVersion };
-  });
+  }));
 }
 
-async function withOperation<T>(kind: string, operation: () => Promise<T>): Promise<T> {
+async function withOperation<T>(
+  kind: string,
+  operation: (operationRevision: number) => Promise<T>,
+  {
+    expectedRevision,
+    profileId = null,
+    serverId = null,
+  }: { expectedRevision?: unknown; profileId?: unknown; serverId?: unknown } = {},
+): Promise<T> {
+  if (operationState.status === 'running') throw new HarborError('OPERATION_IN_PROGRESS');
+  const currentRevision = normalizeStoredState(stateStore.read()).revision;
+  if (expectedRevision !== undefined) {
+    if (!Number.isSafeInteger(expectedRevision) || Number(expectedRevision) !== currentRevision) {
+      throw new HarborError('STATE_CONFLICT');
+    }
+  }
   operationState = {
     kind,
     status: 'running',
     startedAt: new Date().toISOString(),
     error: null,
+    profileId: profileId == null ? null : String(profileId),
+    serverId: serverId == null ? null : String(serverId),
   };
-  updateStoredState((state) => state);
+  let operationRevision: number;
   try {
-    const result = await operation();
-    operationState = { kind: null, status: 'idle', startedAt: null, error: null };
+    operationRevision = updateStoredState((state) => state).revision;
+  } catch (error) {
+    operationState = {
+      kind: null,
+      status: 'idle',
+      startedAt: null,
+      error: null,
+      profileId: null,
+      serverId: null,
+    };
+    throw error;
+  }
+  try {
+    const result = await operation(operationRevision);
+    operationState = {
+      kind: null,
+      status: 'idle',
+      startedAt: null,
+      error: null,
+      profileId: null,
+      serverId: null,
+    };
     updateStoredState((state) => state);
     return result;
   } catch (error) {
@@ -536,7 +661,8 @@ function buildActiveConfig(
   selectedServerId: string,
   routeRules: RouteRule[] = stateStore.read().routeRules,
 ) {
-  return buildGatewayConfig(subscriptionConfig, selectedServerId, {
+  const normalizedConfig = normalizeSubscriptionConfig(subscriptionConfig).config;
+  return buildGatewayConfig(normalizedConfig, selectedServerId, {
     clientDirect: settings.appMode === 'client' && gatewayAutoService.read().mode === 'gateway-direct',
     routeRules,
   });
@@ -545,33 +671,22 @@ function buildActiveConfig(
 const stopSingbox = () => singboxRuntime.stop();
 const startSingbox = () => singboxRuntime.apply();
 
-function writeCurrentConfig(onlyIfSelectionChanged = false) {
+function writeCurrentConfig() {
   const state = normalizeStoredState(stateStore.read());
-  const cached = readSubscriptionCache();
-  if (!state.selectedServerId || !cached?.config) return false;
-  const cachedServers = cached.servers as StoredState['servers'];
-  const selectedServerId = selectRefreshedServer(
-    state.selectedServerId,
-    state.servers,
-    cachedServers,
-  );
-  if (onlyIfSelectionChanged && selectedServerId === state.selectedServerId) return false;
-  const activeConfig = selectedServerId
-    ? buildActiveConfig(cached.config, selectedServerId)
-    : null;
+  const hasAppliedTarget = Boolean(state.appliedProfileId && state.appliedServerId);
+  const profile = hasAppliedTarget
+    ? state.profiles.find((candidate) => candidate.id === state.appliedProfileId) || null
+    : desiredProfile(state);
+  const serverId = hasAppliedTarget ? state.appliedServerId : profile?.desiredServerId;
+  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);
   const previousConfig = fs.existsSync(settings.configPath)
     ? fs.readFileSync(settings.configPath, 'utf8')
     : null;
   try {
-    if (activeConfig) writeSingboxConfig(activeConfig);
-    else removeSingboxConfig();
-    updateStoredState((current) => ({
-      ...current,
-      servers: cachedServers,
-      selectedServerId,
-      appliedServerId: selectedServerId,
-      ...(!selectedServerId ? { connectionDesired: 'stopped' } : {}),
-    }));
+    writeSingboxConfig(activeConfig);
   } catch (error) {
     try {
       if (previousConfig === null) removeSingboxConfig();
@@ -581,7 +696,61 @@ function writeCurrentConfig(onlyIfSelectionChanged = false) {
     }
     throw error;
   }
-  return Boolean(selectedServerId);
+  return { profile, server };
+}
+
+const CONFIG_PROXY_TYPES = new Set(['vless', 'vmess', 'trojan', 'shadowsocks', 'hysteria2']);
+
+function currentConfigMatchesAppliedTarget(state: StoredState) {
+  if (!state.appliedProfileId || !state.appliedServerId || !state.appliedServerSnapshot) return false;
+  let config: Record<string, unknown>;
+  try {
+    config = record(JSON.parse(fs.readFileSync(settings.configPath, 'utf8')));
+  } catch {
+    return false;
+  }
+  const proxyOutbounds = (Array.isArray(config.outbounds) ? config.outbounds : [])
+    .map(record)
+    .filter((outbound) => CONFIG_PROXY_TYPES.has(String(outbound.type || '')));
+  const exactMatches = proxyOutbounds.filter((outbound) => (
+    String(outbound.tag || '') === state.appliedServerId
+  ));
+  const targetMatches = exactMatches.length
+    ? exactMatches
+    : proxyOutbounds.filter((outbound) => (
+        serverIdentityKey(outbound) === serverIdentityKey(state.appliedServerSnapshot)
+      ));
+  if (targetMatches.length !== 1) return false;
+  const outboundTag = String(targetMatches[0].tag || '');
+  const routeFinal = String(record(config.route).final || '');
+  const expectsGatewayDirect = settings.appMode === 'client'
+    && gatewayAutoService.read().mode === 'gateway-direct';
+  return expectsGatewayDirect ? routeFinal === 'direct' : routeFinal === outboundTag;
+}
+
+async function reconcileStoppedBoot({ removeConfig = false } = {}) {
+  try {
+    await stopSingbox();
+  } catch (error) {
+    console.warn(`[control] sing-box не остановлен при startup reconcile: ${errorMessage(error)}`);
+    return;
+  }
+  if (removeConfig) removeSingboxConfig();
+  const state = normalizeStoredState(stateStore.read());
+  if (
+    state.connectionDesired !== 'stopped'
+    || state.appliedProfileId
+    || state.appliedServerId
+    || state.appliedServerSnapshot
+  ) {
+    updateStoredState((current) => ({
+      ...current,
+      connectionDesired: 'stopped',
+      appliedProfileId: '',
+      appliedServerId: '',
+      appliedServerSnapshot: null,
+    }));
+  }
 }
 
 async function handleApi(req: IncomingMessage, res: ServerResponse) {
@@ -654,21 +823,45 @@ process.on('SIGINT', shutdown);
 
 await gatewayAutoService.refresh({ reconfigure: false })
   .catch((error: unknown) => console.warn(`[control] Gateway не определён: ${errorMessage(error)}`));
-try {
-  writeCurrentConfig(settings.appMode !== 'client' && fs.existsSync(settings.configPath));
-} catch (error) {
-  const candidate = record(error);
-  if (!String(candidate.code || '').startsWith('SUBSCRIPTION_')) throw error;
-  console.warn(`[storage] сохранённая подписка отклонена: ${errorMessage(error)}; возврат к первичной настройке`);
-  await subscriptionService.resetSavedSubscription({ stopRuntime: false });
+const bootState = normalizeStoredState(stateStore.read());
+const bootWantsRunning = bootState.connectionDesired === 'running'
+  || (bootState.connectionDesired === undefined && fs.existsSync(settings.configPath));
+if (bootWantsRunning) {
+  let target: ReturnType<typeof writeCurrentConfig> = null;
+  try {
+    target = writeCurrentConfig();
+  } catch (error) {
+    console.warn(`[storage] не удалось собрать сохранённую подписку: ${errorMessage(error)}`);
+  }
+  const canReuseCurrentConfig = target === null
+    && fs.existsSync(settings.configPath)
+    && currentConfigMatchesAppliedTarget(normalizeStoredState(stateStore.read()));
+  if (target || canReuseCurrentConfig) {
+    await startSingbox()
+      .then(() => {
+        const current = normalizeStoredState(stateStore.read());
+        const appliedProfile = target?.profile
+          || current.profiles.find((profile) => profile.id === current.appliedProfileId);
+        const appliedServer = target?.server
+          || current.appliedServerSnapshot;
+        if (appliedProfile && appliedServer) {
+          updateStoredState((state: StoredState) => ({
+            ...state,
+            connectionDesired: 'running',
+            appliedProfileId: appliedProfile.id,
+            appliedServerId: appliedServer.id,
+            appliedServerSnapshot: appliedServer,
+            ...(target ? { appliedRouteRules: state.routeRules } : {}),
+          }));
+        }
+      })
+      .catch((error: unknown) => console.warn(`[control] sing-box не запущен: ${errorMessage(error)}`));
+  } else {
+    await reconcileStoppedBoot({ removeConfig: true });
+  }
+} else {
+  await reconcileStoppedBoot();
 }
-await startSingbox()
-  .then(() => {
-    if (fs.existsSync(settings.configPath)) {
-      updateStoredState((state: StoredState) => ({ ...state, appliedRouteRules: state.routeRules }));
-    }
-  })
-  .catch((error: unknown) => console.warn(`[control] sing-box не запущен: ${errorMessage(error)}`));
 
 if (deviceInventory) {
   await deviceInventory.reconcilePolicies()