Split gateway control and dataplane into separate services

src/server/config.js

@@ -16,6 +16,7 @@ export const settings = {
   proxyPort,
   tproxyPort: parsePort(process.env.TPROXY_PORT, 7895),
   tproxyChain: process.env.TPROXY_CHAIN || "VPN_PROXY_TPROXY",
+  dataplaneSocket: process.env.DATAPLANE_SOCKET || "/run/vpn-proxy/dataplane.sock",
   bindIp: process.env.PROXY_BIND_IP || "0.0.0.0",
   dataDir,
   distDir: process.env.DIST_DIR || "/app/dist",

src/server/dataplane.js

@@ -0,0 +1,70 @@
+import fs from 'node:fs';
+import http from 'node:http';
+import path from 'node:path';
+import { settings } from './config.js';
+import { createSingboxRuntime } from './singboxRuntime.js';
+
+const socketPath = settings.dataplaneSocket;
+const runtime = createSingboxRuntime({
+  configPath: settings.configPath,
+  gateway: true,
+  tproxyChain: settings.tproxyChain,
+});
+let ready = false;
+
+function sendJson(res, statusCode, payload) {
+  res.writeHead(statusCode, { 'content-type': 'application/json; charset=utf-8' });
+  res.end(JSON.stringify(payload));
+}
+
+const server = http.createServer(async (req, res) => {
+  try {
+    if (req.method === 'GET' && req.url === '/status') {
+      return sendJson(res, ready ? 200 : 503, {
+        ...await runtime.refresh(),
+        ready,
+      });
+    }
+    if (req.method === 'POST' && req.url === '/apply') {
+      return sendJson(res, 200, await runtime.apply());
+    }
+    if (req.method === 'POST' && req.url === '/restart') {
+      return sendJson(res, 200, await runtime.restart());
+    }
+    if (req.method === 'POST' && req.url === '/stop') {
+      return sendJson(res, 200, await runtime.stop());
+    }
+    return sendJson(res, 404, { error: 'Не найдено' });
+  } catch (error) {
+    return sendJson(res, 500, { error: error.message || String(error) });
+  }
+});
+
+fs.mkdirSync(path.dirname(socketPath), { recursive: true });
+fs.rmSync(socketPath, { force: true });
+server.listen(socketPath, async () => {
+  fs.chmodSync(socketPath, 0o660);
+  try {
+    await runtime.apply();
+  } catch (error) {
+    console.warn(`[dataplane] sing-box не запущен: ${error.message}`);
+  } finally {
+    ready = true;
+    console.log(`[dataplane] control socket: ${socketPath}`);
+  }
+});
+
+let shuttingDown = false;
+async function shutdown() {
+  if (shuttingDown) return;
+  shuttingDown = true;
+  ready = false;
+  await runtime.shutdown();
+  server.close(() => {
+    fs.rmSync(socketPath, { force: true });
+    process.exit(0);
+  });
+}
+
+process.on('SIGTERM', shutdown);
+process.on('SIGINT', shutdown);

src/server/dataplaneClient.js

@@ -0,0 +1,43 @@
+import http from 'node:http';
+
+function request(socketPath, pathname, method = 'GET') {
+  return new Promise((resolve, reject) => {
+    const req = http.request({ socketPath, path: pathname, method }, (res) => {
+      const chunks = [];
+      res.on('data', (chunk) => chunks.push(chunk));
+      res.on('end', () => {
+        let body = {};
+        try {
+          body = JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}');
+        } catch {
+          return reject(new Error('Dataplane вернул невалидный JSON'));
+        }
+        if ((res.statusCode || 500) >= 400) {
+          return reject(new Error(body.error || `Dataplane HTTP ${res.statusCode}`));
+        }
+        resolve(body);
+      });
+    });
+    req.on('error', reject);
+    req.setTimeout(6000, () => req.destroy(new Error('Dataplane не ответил за 6 секунд')));
+    req.end();
+  });
+}
+
+export function createDataplaneClient(socketPath, send = request) {
+  let current = { running: false, startedAt: null };
+  const update = async (pathname, method) => {
+    current = await send(socketPath, pathname, method);
+    return current;
+  };
+
+  return {
+    get running() { return Boolean(current.running); },
+    get startedAt() { return current.startedAt || null; },
+    refresh: () => update('/status', 'GET'),
+    apply: () => update('/apply', 'POST'),
+    restart: () => update('/restart', 'POST'),
+    stop: () => update('/stop', 'POST'),
+    shutdown: async () => current,
+  };
+}

src/server/index.js

@@ -1,8 +1,8 @@
 import fs from 'node:fs';
 import http from 'node:http';
 import path from 'node:path';
-import { spawn, spawnSync } from 'node:child_process';
 import { isDeepStrictEqual } from 'node:util';
+import { createDataplaneClient } from './dataplaneClient.js';
 import { settings } from './config.js';
 import {
   applyGatewayPreference,
@@ -13,7 +13,7 @@ import {
   readHostNetworkState,
   sameGatewayRoute,
 } from './gatewayPresence.js';
-import { setGatewayInterception } from './gatewayRouting.js';
+import { createSingboxRuntime } from './singboxRuntime.js';
 import { tcpPing } from './ping.js';
 import { buildSharedProxyInfo } from './sharedProxy.js';
 import {
@@ -29,8 +29,14 @@ const GATEWAY_DISCOVERY_INTERVAL_MS = 5_000;
 
 fs.mkdirSync(settings.dataDir, { recursive: true });
 
-let singboxProcess = null;
-let singboxStartedAt = null;
+const remoteDataplane = settings.appMode === 'gateway' && Boolean(process.env.DATAPLANE_SOCKET);
+const singboxRuntime = remoteDataplane
+  ? createDataplaneClient(settings.dataplaneSocket)
+  : createSingboxRuntime({
+      configPath: settings.configPath,
+      gateway: settings.appMode === 'gateway',
+      tproxyChain: settings.tproxyChain,
+    });
 let subscriptionRefreshPromise = null;
 let subscriptionRefreshTimer = null;
 let gatewayDiscoveryPromise = null;
@@ -110,79 +116,11 @@ function buildActiveConfig(subscriptionConfig, selectedTag) {
   });
 }
 
-function checkSingboxConfig() {
-  const result = spawnSync('sing-box', ['check', '-c', settings.configPath], {
-    encoding: 'utf8',
-  });
-  if (result.status !== 0) {
-    throw new Error((result.stderr || result.stdout || 'sing-box check failed').trim());
-  }
-}
+const stopSingbox = () => singboxRuntime.stop();
+const startSingbox = () => singboxRuntime.apply();
 
-function stopSingbox() {
-  return new Promise((resolve) => {
-    if (settings.appMode === 'gateway') {
-      setGatewayInterception(false, settings.tproxyChain);
-    }
-    if (!singboxProcess) {
-      singboxStartedAt = null;
-      return resolve();
-    }
-
-    const current = singboxProcess;
-    singboxProcess = null;
-    singboxStartedAt = null;
-    const timeout = setTimeout(() => {
-      current.kill('SIGKILL');
-      resolve();
-    }, 4000);
-
-    current.once('exit', () => {
-      clearTimeout(timeout);
-      resolve();
-    });
-    current.kill('SIGTERM');
-  });
-}
-
-async function startSingbox() {
-  if (!fs.existsSync(settings.configPath)) {
-    if (settings.appMode === 'gateway') {
-      setGatewayInterception(false, settings.tproxyChain);
-    }
-    return false;
-  }
-  checkSingboxConfig();
-  await stopSingbox();
-
-  const child = spawn('sing-box', ['run', '-c', settings.configPath], {
-    stdio: ['ignore', 'inherit', 'inherit'],
-  });
-  singboxProcess = child;
-  singboxStartedAt = new Date().toISOString();
-  try {
-    if (settings.appMode === 'gateway') {
-      setGatewayInterception(true, settings.tproxyChain);
-    }
-  } catch (error) {
-    child.kill('SIGTERM');
-    singboxProcess = null;
-    singboxStartedAt = null;
-    throw error;
-  }
-  child.once('exit', () => {
-    if (singboxProcess === child) {
-      singboxProcess = null;
-      singboxStartedAt = null;
-      if (settings.appMode === 'gateway') {
-        setGatewayInterception(false, settings.tproxyChain);
-      }
-    }
-  });
-  return true;
-}
-
-function publicState() {
+async function publicState() {
+  await singboxRuntime.refresh();
   const state = readJson(settings.statePath, {});
   const gatewayAutoEnabled = state.gatewayAutoEnabled !== false;
   return {
@@ -190,8 +128,8 @@ function publicState() {
     port: settings.port,
     proxyPort: settings.proxyPort,
     configExists: fs.existsSync(settings.configPath),
-    singboxRunning: Boolean(singboxProcess),
-    singboxStartedAt,
+    singboxRunning: singboxRuntime.running,
+    singboxStartedAt: singboxRuntime.startedAt,
     subscriptionHost: subscriptionHost(state.subscriptionUrl),
     hasSubscription: Boolean(state.subscriptionUrl),
     selectedTag: state.selectedTag || '',
@@ -230,7 +168,7 @@ async function applyGatewayAutoState(nextState, { reconfigure = true } = {}) {
   const previousConfig = fs.existsSync(settings.configPath)
     ? fs.readFileSync(settings.configPath, 'utf8')
     : null;
-  const wasRunning = Boolean(singboxProcess);
+  const wasRunning = singboxRuntime.running;
   try {
     const configured = writeCurrentConfig();
     if (reconfigure && configured && wasRunning) await startSingbox();
@@ -388,9 +326,9 @@ function refreshSavedSubscription() {
       writeJson(settings.subscriptionCachePath, { url: subscriptionUrl, ...parsed });
 
       try {
-        if (singboxProcess && activeConfigChanged) await applySelectedServer(selectedTag);
+        if (singboxRuntime.running && activeConfigChanged) await applySelectedServer(selectedTag);
         else if (selectedTag) {
-          if (!singboxProcess) writeSingboxConfig(buildActiveConfig(parsed.config, selectedTag));
+          if (!singboxRuntime.running) writeSingboxConfig(buildActiveConfig(parsed.config, selectedTag));
         } else {
           removeSingboxConfig();
         }
@@ -428,14 +366,14 @@ function refreshSavedSubscription() {
 
 async function handleApi(req, res) {
   if (req.method === 'GET' && req.url === '/api/state') {
-    return sendJson(res, 200, publicState());
+    return sendJson(res, 200, await publicState());
   }
 
   if (req.method === 'GET' && req.url === '/api/shared-proxy') {
     return sendJson(res, 200, buildSharedProxyInfo({
       appMode: settings.appMode,
       proxyPort: settings.proxyPort,
-      running: Boolean(singboxProcess),
+      running: (await singboxRuntime.refresh()).running,
       hostHeader: req.headers.host,
       sharedProxyHost: settings.sharedProxyHost,
     }));
@@ -508,7 +446,7 @@ async function handleApi(req, res) {
       });
       await applyGatewayAutoState(applyGatewayPreference(gatewayAutoState, enabled));
     });
-    return sendJson(res, 200, { success: true, gatewayAuto: publicState().gatewayAuto });
+    return sendJson(res, 200, { success: true, gatewayAuto: (await publicState()).gatewayAuto });
   }
 
   if (req.method === 'DELETE' && req.url === '/api/subscription') {
@@ -542,7 +480,7 @@ async function handleApi(req, res) {
         error.statusCode = 400;
         throw error;
       }
-      await startSingbox();
+      await singboxRuntime.restart();
     });
     return sendJson(res, 200, { success: true, singboxRunning: true });
   }
@@ -591,7 +529,7 @@ const server = http.createServer(async (req, res) => {
 async function shutdown() {
   clearInterval(subscriptionRefreshTimer);
   clearInterval(gatewayDiscoveryTimer);
-  await serializeControl(() => stopSingbox());
+  await serializeControl(() => singboxRuntime.shutdown());
   process.exit(0);
 }
 

src/server/singboxRuntime.js

@@ -0,0 +1,88 @@
+import crypto from 'node:crypto';
+import fs from 'node:fs';
+import { spawn, spawnSync } from 'node:child_process';
+import { setGatewayInterception } from './gatewayRouting.js';
+
+export function createSingboxRuntime({ configPath, gateway = false, tproxyChain = '' }) {
+  let child = null;
+  let configHash = '';
+  let startedAt = null;
+
+  const state = () => ({ running: Boolean(child), startedAt });
+
+  async function stop() {
+    if (gateway) setGatewayInterception(false, tproxyChain);
+    if (!child) {
+      configHash = '';
+      startedAt = null;
+      return state();
+    }
+
+    const current = child;
+    child = null;
+    configHash = '';
+    startedAt = null;
+    await new Promise((resolve) => {
+      const timeout = setTimeout(() => {
+        current.kill('SIGKILL');
+        resolve();
+      }, 4000);
+      current.once('exit', () => {
+        clearTimeout(timeout);
+        resolve();
+      });
+      current.kill('SIGTERM');
+    });
+    return state();
+  }
+
+  async function apply({ force = false } = {}) {
+    if (!fs.existsSync(configPath)) {
+      await stop();
+      return state();
+    }
+
+    const check = spawnSync('sing-box', ['check', '-c', configPath], { encoding: 'utf8' });
+    if (check.status !== 0) {
+      throw new Error((check.stderr || check.stdout || 'sing-box check failed').trim());
+    }
+
+    const nextHash = crypto.createHash('sha256').update(fs.readFileSync(configPath)).digest('hex');
+    if (!force && child && nextHash === configHash) return state();
+
+    await stop();
+    const current = spawn('sing-box', ['run', '-c', configPath], {
+      stdio: ['ignore', 'inherit', 'inherit'],
+    });
+    child = current;
+    configHash = nextHash;
+    startedAt = new Date().toISOString();
+    try {
+      if (gateway) setGatewayInterception(true, tproxyChain);
+    } catch (error) {
+      current.kill('SIGTERM');
+      child = null;
+      configHash = '';
+      startedAt = null;
+      throw error;
+    }
+    current.once('exit', () => {
+      if (child !== current) return;
+      child = null;
+      configHash = '';
+      startedAt = null;
+      if (gateway) setGatewayInterception(false, tproxyChain);
+    });
+    return state();
+  }
+
+  return {
+    get running() { return Boolean(child); },
+    get startedAt() { return startedAt; },
+    refresh: async () => state(),
+    apply,
+    restart: () => apply({ force: true }),
+    stop,
+    shutdown: stop,
+  };
+}