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)['"]/); });