harbor-net/test/server/connection-service.test.js

import assert from 'node:assert/strict';
import { readFileSync } from 'node:fs';
import test from 'node:test';

import {
  captureRuntimeCommand,
  createConnectionService,
} from '../../dist/server/features/connection/index.js';
import { createConnectionRuntimeRoute } from '../../dist/server/http/routes/connectionRuntimeRoute.js';
import { createServerApplyRoute } from '../../dist/server/http/routes/serverApplyRoute.js';
import { HarborError } from '../../dist/shared/errors.js';

const serverA = { id: 'a', label: 'Alpha', host: 'a.example', port: 1, protocol: 'vless' };
const serverB = { id: 'b', label: 'Beta', host: 'b.example', port: 2, protocol: 'vless' };
const duplicateA = { id: 'a2', label: 'Alpha', host: 'a2.example', port: 3, protocol: 'vless' };
const rules = [{ type: 'domain_suffix', value: 'example', enabled: true }];

function createHarness(overrides = {}) {
  let state = structuredClone(overrides.state ?? {
    revision: 5,
    servers: [serverA, serverB],
    selectedServerId: serverA.id,
    appliedServerId: serverA.id,
    routeRules: rules,
    appliedRouteRules: rules,
    connectionDesired: 'running',
  });
  let config = Object.hasOwn(overrides, 'config') ? overrides.config : 'old-config';
  let running = overrides.running ?? true;
  let tail = Promise.resolve();
  let active = 0;
  let peak = 0;
  let stateUpdateIndex = 0;
  const events = [];
  const failures = {
    ...overrides.failures,
    stateUpdates: [...(overrides.failures?.stateUpdates || [])],
    runtimeStarts: [...(overrides.failures?.runtimeStarts || [])],
    runtimeStops: [...(overrides.failures?.runtimeStops || [])],
    runtimeRestarts: [...(overrides.failures?.runtimeRestarts || [])],
  };

  function applyState(mutator) {
    const failure = failures.stateUpdates[stateUpdateIndex++];
    if (failure instanceof Error) throw failure;
    const revision = state.revision + 1;
    state = { ...structuredClone(mutator(structuredClone(state))), revision };
    events.push(stateUpdateIndex === 1 ? 'state.desired' : stateUpdateIndex === 2 ? 'state.applied' : 'state.restore');
    if (failure?.after) throw failure.after;
    return structuredClone(state);
  }

  const serialize = (operation) => {
    const run = async () => {
      active += 1;
      peak = Math.max(peak, active);
      try {
        return await operation();
      } finally {
        active -= 1;
      }
    };
    const result = tail.then(run, run);
    tail = result.then(() => undefined, () => undefined);
    return result;
  };

  const stopRuntime = async () => {
    events.push('runtime.stop');
    const failure = failures.runtimeStops.shift();
    if (failure) {
      running = false;
      throw failure;
    }
    running = false;
  };

  const restartRuntime = async () => {
    events.push('runtime.restart');
    const failure = failures.runtimeRestarts.shift();
    if (failure) {
      if (failure.code !== 'CONFIG_INVALID') running = false;
      throw failure;
    }
    running = true;
  };

  const service = createConnectionService({
    state: {
      read: () => structuredClone(state),
      update: applyState,
    },
    subscription: {
      readConfig: () => overrides.missingSubscription ? null : { outbounds: [] },
    },
    config: {
      exists: () => config !== null,
      build: (_subscription, selectedServerId, routeRules) => ({ selectedServerId, routeRules }),
      read: () => config,
      write: (value) => {
        events.push('config.write');
        if (failures.configWrite) throw failures.configWrite;
        config = JSON.stringify(value);
        if (failures.configWriteAfter) throw failures.configWriteAfter;
      },
      restore: (value) => {
        events.push('config.restore');
        if (failures.configRestore) throw failures.configRestore;
        config = value;
      },
      remove: () => {
        events.push('config.remove');
        if (failures.configRemove) throw failures.configRemove;
        config = null;
      },
    },
    runtime: {
      isRunning: async () => {
        if (failures.runtimeStatus) throw failures.runtimeStatus;
        return running;
      },
      start: async () => {
        events.push('runtime.start');
        const failure = failures.runtimeStarts.shift();
        if (failure) {
          running = false;
          throw failure;
        }
        if (overrides.startGate) await overrides.startGate();
        running = true;
      },
      stop: stopRuntime,
      stopCommand: () => captureRuntimeCommand(stopRuntime),
      restartCommand: () => captureRuntimeCommand(
        restartRuntime,
        { preMutationErrorCodes: ['CONFIG_INVALID'] },
      ),
    },
    serialize,
    now: () => new Date('2026-08-08T12:00:00.000Z'),
  });

  return {
    service,
    events,
    enqueue: serialize,
    setState: (value) => { state = structuredClone(value); },
    snapshot: () => structuredClone({ state, config, running, peak }),
  };
}

function assertDomainRestored(actual, expected) {
  const actualDomain = structuredClone(actual);
  const expectedDomain = structuredClone(expected);
  delete actualDomain.state.revision;
  delete expectedDomain.state.revision;
  delete actualDomain.peak;
  delete expectedDomain.peak;
  assert.deepEqual(actualDomain, expectedDomain);
  assert.ok(actual.state.revision >= expected.state.revision);
}

test('apply resolves ID inside the queue and commits desired, config, runtime, then applied state', async () => {
  const harness = createHarness({ state: {
    revision: 5,
    servers: [serverA, serverB, duplicateA],
    selectedServerId: serverA.id,
    appliedServerId: serverA.id,
    routeRules: rules,
    appliedRouteRules: rules,
  } });

  assert.deepEqual(await harness.service.apply('  b  ', 'Alpha'), { serverId: 'b', selectedTag: 'Beta' });
  const snapshot = harness.snapshot();
  assert.equal(snapshot.state.selectedServerId, 'b');
  assert.equal(snapshot.state.appliedServerId, 'b');
  assert.equal(snapshot.state.connectionDesired, 'running');
  assert.equal(snapshot.state.appliedAt, '2026-08-08T12:00:00.000Z');
  assert.deepEqual(snapshot.state.appliedRouteRules, rules);
  assert.deepEqual(harness.events, ['state.desired', 'config.write', 'runtime.start', 'state.applied']);

  await assert.rejects(harness.service.apply('', 'Alpha'), (error) => error.code === 'SERVER_NOT_FOUND');
  await assert.rejects(harness.service.apply('missing', 'Beta'), (error) => error.code === 'SERVER_NOT_FOUND');
});

test('apply accepts one legacy label and rejects missing config before mutation', async () => {
  const unique = createHarness();
  assert.deepEqual(await unique.service.apply('', '  Beta  '), { serverId: 'b', selectedTag: 'Beta' });

  const missing = createHarness({ missingSubscription: true });
  const before = missing.snapshot();
  await assert.rejects(missing.service.apply('b', ''), (error) => error.code === 'CONFIG_INVALID');
  assertDomainRestored(missing.snapshot(), before);
  assert.deepEqual(missing.events, []);
});

test('apply restores previous running domain/config/runtime after mutation failures', async () => {
  const failureCases = [
    { failures: { stateUpdates: [new Error('desired')] }, expected: 'desired' },
    { failures: { configWriteAfter: new Error('config') }, expected: 'config' },
    { failures: { runtimeStarts: [new Error('start')] }, expected: 'start' },
    { failures: { stateUpdates: [null, { after: new Error('applied') }] }, expected: 'applied' },
  ];

  for (const { failures, expected } of failureCases) {
    const harness = createHarness({ failures });
    const before = harness.snapshot();
    await assert.rejects(harness.service.apply('b', ''), new RegExp(expected));
    assertDomainRestored(harness.snapshot(), before);
  }
});

test('apply compensates a desired state update that persists and then throws', async () => {
  const failure = new Error('desired persisted then failed');
  const harness = createHarness({ failures: { stateUpdates: [{ after: failure }] } });
  const before = harness.snapshot();

  await assert.rejects(harness.service.apply('b', ''), (error) => error === failure);
  const after = harness.snapshot();
  assertDomainRestored(after, before);
  assert.ok(after.state.revision > before.state.revision);
});

test('apply restores previous stopped state and prior config absence', async () => {
  const failure = new Error('start');
  const harness = createHarness({ config: null, running: false, failures: { runtimeStarts: [failure] } });
  const before = harness.snapshot();

  await assert.rejects(harness.service.apply('b', ''), (error) => error === failure);
  assertDomainRestored(harness.snapshot(), before);
  assert.ok(harness.events.includes('config.remove'));
  assert.ok(harness.events.includes('runtime.stop'));
});

test('apply rollback continues after restore failures and marks runtime rollback failure', async () => {
  const original = new Error('apply failed');
  const configFailure = new Error('config restore failed');
  const nonRuntime = createHarness({
    failures: { runtimeStarts: [original], configRestore: configFailure },
  });
  await assert.rejects(nonRuntime.service.apply('b', ''), (error) => {
    assert.ok(error instanceof AggregateError);
    assert.deepEqual(error.errors, [original, configFailure]);
    return true;
  });
  assert.ok(nonRuntime.events.filter((event) => event === 'runtime.start').length >= 2);
  assert.ok(nonRuntime.events.filter((event) => event.startsWith('state.')).length >= 2);

  const rollbackFailure = new Error('runtime rollback failed');
  const runtimeBroken = createHarness({ failures: { runtimeStarts: [original, rollbackFailure] } });
  await assert.rejects(runtimeBroken.service.apply('b', ''), (error) => {
    assert.equal(error.code, 'PROCESS_START_FAILED');
    assert.deepEqual(error.cause.errors, [original, rollbackFailure]);
    return true;
  });
  assert.ok(runtimeBroken.events.filter((event) => event.startsWith('state.')).length >= 2);
});

test('queued apply resolves latest state and concurrent valid applies remain serial', async () => {
  let release;
  const blocker = new Promise((resolve) => { release = resolve; });
  const stale = createHarness();
  const blocked = stale.enqueue(() => blocker);
  const apply = stale.service.apply('b', '');
  stale.setState({ ...stale.snapshot().state, servers: [serverA] });
  release();
  await blocked;
  await assert.rejects(apply, (error) => error.code === 'SERVER_NOT_FOUND');

  const serial = createHarness();
  const [first, second] = await Promise.all([
    serial.service.apply('a', ''),
    serial.service.apply('b', ''),
  ]);
  assert.equal(first.serverId, 'a');
  assert.equal(second.serverId, 'b');
  assert.equal(serial.snapshot().state.appliedServerId, 'b');
  assert.equal(serial.snapshot().peak, 1);
});

test('server apply route preserves defaults, operation kind, response, and single ownership', async () => {
  const calls = [];
  const sent = [];
  const route = createServerApplyRoute({
    connection: {
      apply: async (serverId, selectedTag) => {
        calls.push([serverId, selectedTag]);
        return { serverId: 'resolved', selectedTag: 'Resolved' };
      },
    },
    readBody: async () => ({}),
    withOperation: async (kind, operation) => {
      calls.push(kind);
      return operation();
    },
    sendState: async (_res, extra) => { sent.push(extra); },
  });
  const response = {};

  assert.equal(await route.handle({ method: 'GET', url: '/api/apply' }, response), false);
  assert.equal(await route.handle({ method: 'POST', url: '/api/apply' }, response), true);
  assert.deepEqual(calls, ['apply-server', ['', '']]);
  assert.deepEqual(sent, [{ serverId: 'resolved', selectedTag: 'Resolved' }]);

  const source = readFileSync(new URL('../../src/server/index.ts', import.meta.url), 'utf8');
  assert.match(source, /createServerApplyRoute\(\{/);
  assert.doesNotMatch(source, /function applySelectedServer|req\.url === ['"]\/api\/apply['"]/);
});

test('stop always cleans runtime and commits only desired stopped state', async () => {
  for (const running of [true, false]) {
    const harness = createHarness({ running });
    const before = harness.snapshot();
    await harness.service.stop();
    const after = harness.snapshot();
    assert.equal(after.running, false);
    assert.equal(after.state.connectionDesired, 'stopped');
    assert.equal(after.state.selectedServerId, before.state.selectedServerId);
    assert.equal(after.state.appliedServerId, before.state.appliedServerId);
    assert.deepEqual(after.state.appliedRouteRules, before.state.appliedRouteRules);
    assert.equal(harness.events.filter((event) => event === 'runtime.stop').length, 1);
  }
});

test('restart validates config inside the queue and commits applied runtime fields', async () => {
  const missing = createHarness({ config: null });
  const beforeMissing = missing.snapshot();
  await assert.rejects(missing.service.restart(), (error) => error.code === 'CONFIG_INVALID');
  assertDomainRestored(missing.snapshot(), beforeMissing);
  assert.deepEqual(missing.events, []);

  for (const running of [true, false]) {
    const harness = createHarness({ running, state: {
      revision: 1,
      servers: [serverA, serverB],
      selectedServerId: serverB.id,
      appliedServerId: serverA.id,
      routeRules: rules,
      appliedRouteRules: [],
      appliedAt: 'preserved',
      connectionDesired: 'stopped',
    } });
    await harness.service.restart();
    const after = harness.snapshot();
    assert.equal(after.running, true);
    assert.equal(after.state.appliedServerId, serverB.id);
    assert.equal(after.state.connectionDesired, 'running');
    assert.deepEqual(after.state.appliedRouteRules, rules);
    assert.equal(after.state.appliedAt, 'preserved');
  }
});

test('stop and restart compensate runtime and pre/post-write state failures', async () => {
  const statusFailure = new Error('status failed');
  for (const command of ['stop', 'restart']) {
    const status = createHarness({ failures: { runtimeStatus: statusFailure } });
    await status.service[command]();
    assert.ok(status.events.includes(`runtime.${command}`));
  }

  for (const stateFailure of [new Error('state before'), { after: new Error('state after') }]) {
    const stopHarness = createHarness({ failures: { stateUpdates: [stateFailure] } });
    const stopBefore = stopHarness.snapshot();
    await assert.rejects(stopHarness.service.stop(), /state/);
    assertDomainRestored(stopHarness.snapshot(), stopBefore);

    const restartHarness = createHarness({
      running: false,
      failures: { stateUpdates: [stateFailure] },
    });
    const restartBefore = restartHarness.snapshot();
    await assert.rejects(restartHarness.service.restart(), /state/);
    assertDomainRestored(restartHarness.snapshot(), restartBefore);
  }

  const stopFailure = new Error('stop failed');
  const stopped = createHarness({ failures: { runtimeStops: [stopFailure] } });
  const stoppedBefore = stopped.snapshot();
  await assert.rejects(stopped.service.stop(), (error) => error === stopFailure);
  assertDomainRestored(stopped.snapshot(), stoppedBefore);

  const restartFailure = new Error('restart failed');
  const restarted = createHarness({ running: false, failures: { runtimeRestarts: [restartFailure] } });
  const restartedBefore = restarted.snapshot();
  await assert.rejects(restarted.service.restart(), (error) => error === restartFailure);
  assertDomainRestored(restarted.snapshot(), restartedBefore);
});

test('restart preserves CONFIG_INVALID when validation fails before runtime mutation', async () => {
  const invalid = new HarborError('CONFIG_INVALID');
  const harness = createHarness({ failures: { runtimeRestarts: [invalid] } });
  const before = harness.snapshot();

  await assert.rejects(harness.service.restart(), (error) => error === invalid);
  assertDomainRestored(harness.snapshot(), before);
  assert.equal(harness.events.filter((event) => event === 'runtime.restart').length, 1);
  assert.equal(harness.events.includes('runtime.start'), false);
});

test('runtime command adapter makes local and remote mutation phase explicit', async () => {
  const localFailure = new HarborError('CONFIG_INVALID');
  assert.deepEqual(
    await captureRuntimeCommand(
      async () => { throw localFailure; },
      { preMutationErrorCodes: ['CONFIG_INVALID'] },
    ),
    { ok: false, mutationStarted: false, error: localFailure },
  );

  const remoteFailure = new HarborError('PROCESS_START_FAILED');
  assert.deepEqual(
    await captureRuntimeCommand(async () => { throw remoteFailure; }),
    { ok: false, mutationStarted: true, error: remoteFailure },
  );
});

test('runtime command rollback continues through state errors and maps runtime restore failure', async () => {
  const original = new Error('stop state failed');
  const stateRestore = new Error('state restore failed');
  const aggregate = createHarness({
    failures: { stateUpdates: [{ after: original }, stateRestore] },
  });
  await assert.rejects(aggregate.service.stop(), (error) => {
    assert.ok(error instanceof AggregateError);
    assert.deepEqual(error.errors, [original, stateRestore]);
    return true;
  });
  assert.equal(aggregate.snapshot().running, true);

  const stopFailure = new Error('stop failed');
  const runtimeRestore = new Error('start rollback failed');
  const broken = createHarness({
    failures: { runtimeStops: [stopFailure], runtimeStarts: [runtimeRestore] },
  });
  await assert.rejects(broken.service.stop(), (error) => {
    assert.equal(error.code, 'PROCESS_START_FAILED');
    assert.deepEqual(error.cause.errors, [stopFailure, runtimeRestore]);
    return true;
  });
});

test('apply, stop, and restart share one deterministic connection queue', async () => {
  const applyThenStop = createHarness();
  await Promise.all([
    applyThenStop.service.apply('b', ''),
    applyThenStop.service.stop(),
  ]);
  assert.equal(applyThenStop.snapshot().state.connectionDesired, 'stopped');
  assert.equal(applyThenStop.snapshot().peak, 1);

  const stopThenRestart = createHarness();
  await Promise.all([
    stopThenRestart.service.stop(),
    stopThenRestart.service.restart(),
  ]);
  assert.equal(stopThenRestart.snapshot().state.connectionDesired, 'running');
  assert.equal(stopThenRestart.snapshot().running, true);
  assert.equal(stopThenRestart.snapshot().peak, 1);
});

test('connection runtime route preserves operation kinds, response extras, and single ownership', async () => {
  const calls = [];
  const sent = [];
  const route = createConnectionRuntimeRoute({
    connection: {
      stop: async () => { calls.push('service.stop'); },
      restart: async () => { calls.push('service.restart'); },
    },
    withOperation: async (kind, operation) => {
      calls.push(kind);
      return operation();
    },
    sendState: async (_res, extra) => { sent.push(extra); },
  });
  const response = {};

  assert.equal(await route.handle({ method: 'GET', url: '/api/singbox/stop' }, response), false);
  assert.equal(await route.handle({ method: 'POST', url: '/api/singbox/stop' }, response), true);
  assert.equal(await route.handle({ method: 'POST', url: '/api/singbox/restart' }, response), true);
  assert.deepEqual(calls, ['stop', 'service.stop', 'start', 'service.restart']);
  assert.deepEqual(sent, [{ singboxRunning: false }, { singboxRunning: true }]);

  const source = readFileSync(new URL('../../src/server/index.ts', import.meta.url), 'utf8');
  assert.match(source, /createConnectionRuntimeRoute\(\{/);
  assert.doesNotMatch(source, /req\.url === ['"]\/api\/singbox\/(?:stop|restart)['"]/);
});