import assert from 'node:assert/strict';
import { execFileSync } from 'node:child_process';
import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { pathToFileURL } from 'node:url';
import test from 'node:test';

import { createLiveTrafficService } from '../../dist/server/services/liveTrafficService.js';

const subscriptionConfig = {
  outbounds: [{
    type: 'vless',
    tag: 'test-vpn',
    server: 'vpn.example.test',
    server_port: 443,
    uuid: '00000000-0000-4000-8000-000000000000',
    tls: { enabled: true },
  }],
};

function buildClientConfig(trafficSource) {
  const dataDir = fs.mkdtempSync(path.join(os.tmpdir(), `harbor-native-${trafficSource}-`));
  const singboxUrl = pathToFileURL(path.resolve('dist/server/singbox.js')).href;
  const script = `
    const { buildGatewayConfig } = await import(${JSON.stringify(singboxUrl)});
    const config = buildGatewayConfig(${JSON.stringify(subscriptionConfig)}, 'test-vpn');
    process.stdout.write(JSON.stringify(config));
  `;
  try {
    return JSON.parse(execFileSync(process.execPath, ['--input-type=module', '--eval', script], {
      cwd: path.resolve('.'),
      encoding: 'utf8',
      env: {
        ...process.env,
        APP_MODE: 'client',
        DATA_DIR: dataDir,
        SING_BOX_CACHE: path.join(dataDir, 'cache.db'),
        SING_BOX_TRAFFIC_SOURCE: trafficSource,
      },
    }));
  } finally {
    fs.rmSync(dataDir, { recursive: true, force: true });
  }
}

function abortableStream(value, signal) {
  return (async function* stream() {
    yield value;
    await new Promise((_, reject) => {
      const abort = () => reject(signal.reason || new Error('aborted'));
      if (signal.aborted) abort();
      else signal.addEventListener('abort', abort, { once: true });
    });
  }());
}

async function waitFor(check, timeout = 1_000) {
  const deadline = Date.now() + timeout;
  while (!check()) {
    if (Date.now() >= deadline) throw new Error('Timed out waiting for native traffic collector');
    await new Promise((resolve) => setTimeout(resolve, 5));
  }
}

test('client native mode adds the private API service and removes the retired DNS option', () => {
  const config = buildClientConfig('native');
  assert.deepEqual(config.services, [{
    type: 'api',
    listen: '127.0.0.1',
    listen_port: 19091,
    dashboard: false,
  }]);
  assert.deepEqual(config.dns, {});
  assert.equal(config.experimental.clash_api, undefined);
});

test('client disabled mode omits the API service and keeps the 1.13-compatible DNS option', () => {
  const config = buildClientConfig('disabled');
  assert.equal(config.services, undefined);
  assert.deepEqual(config.dns, { independent_cache: true });
  assert.equal(config.experimental.clash_api, undefined);
});

test('native collector authenticates all RC5 lifecycle calls and uses one-second Go duration intervals', async () => {
  const requests = [];
  const connection = {
    id: 'connection-1',
    inbound: 'mixed-in',
    inboundType: 'mixed',
    network: 'tcp',
    source: '127.0.0.1:54000',
    destination: '203.0.113.10:443',
    domain: 'example.test',
    protocol: 'tls',
    createdAt: 1_700_000_000_000n,
    closedAt: 0n,
    uplinkTotal: 12n,
    downlinkTotal: 34n,
    outbound: 'test-vpn',
    outboundType: 'vless',
    rule: 'final',
    chainList: ['test-vpn'],
  };
  const clientFactory = (port) => {
    requests.push(['factory', port]);
    return {
      async getVersion(_input, { signal, headers }) {
        requests.push(['version', signal instanceof AbortSignal, headers?.authorization]);
        return { version: '1.14.0-rc.5', apiVersion: 4 };
      },
      async getStartedAt(_input, { signal, headers }) {
        requests.push(['started-at', signal instanceof AbortSignal, headers?.authorization]);
        return { startedAt: 1_700_000_000_000n };
      },
      subscribeConnections({ interval }, { signal, headers }) {
        requests.push(['connections', interval, headers?.authorization]);
        return abortableStream({
          reset: true,
          events: [{ type: 0, id: connection.id, connection }],
        }, signal);
      },
      subscribeStatus({ interval }, { signal, headers }) {
        requests.push(['status', interval, headers?.authorization]);
        return abortableStream({
          connectionsIn: 1,
          uplinkTotal: 12n,
          downlinkTotal: 34n,
        }, signal);
      },
    };
  };
  const service = createLiveTrafficService({
    port: 19091,
    enabled: true,
    isRuntimeRunning: () => true,
    authorization: () => 'test-secret',
    clientFactory,
  });

  service.start();
  await waitFor(() => service.snapshot().source.state === 'live');
  const snapshot = service.snapshot();
  assert.equal(snapshot.epoch, 'sing-box-1700000000000');
  assert.equal(snapshot.source.singBoxVersion, '1.14.0-rc.5');
  assert.equal(snapshot.source.singBoxApiVersion, 4);
  assert.equal(snapshot.connections[0].destination.domain, 'example.test');
  assert.deepEqual(requests, [
    ['factory', 19091],
    ['version', true, 'Bearer test-secret'],
    ['started-at', true, 'Bearer test-secret'],
    ['connections', 1_000_000_000n, 'Bearer test-secret'],
    ['status', 1_000_000_000n, 'Bearer test-secret'],
  ]);
  await service.stop();
});

test('disabled collector stays cached and never constructs a native client', async () => {
  let factoryCalls = 0;
  const service = createLiveTrafficService({
    port: 19091,
    enabled: false,
    isRuntimeRunning: () => true,
    clientFactory: () => {
      factoryCalls += 1;
      throw new Error('must not connect');
    },
  });

  service.start();
  assert.equal(service.snapshot().source.state, 'disabled');
  assert.equal(factoryCalls, 0);
  await service.stop();
});

test('a native runtime materialization error is cached as incompatible without constructing a client', async () => {
  let factoryCalls = 0;
  const service = createLiveTrafficService({
    port: 19091,
    enabled: false,
    unavailableError: 'secret file unavailable at https://private.example/path',
    isRuntimeRunning: () => true,
    clientFactory: () => {
      factoryCalls += 1;
      throw new Error('must not connect');
    },
  });

  service.start();
  assert.equal(service.snapshot().source.state, 'incompatible');
  assert.equal(service.snapshot().source.error, 'secret file unavailable at [endpoint]');
  assert.equal(factoryCalls, 0);
  await service.stop();
});

for (const incompatible of [
  { version: '1.13.18', apiVersion: 4 },
  { version: '1.14.0-rc.5', apiVersion: 5 },
]) {
  test(`collector rejects unqualified sing-box ${incompatible.version} API ${incompatible.apiVersion}`, async () => {
    let startedAtCalls = 0;
    const service = createLiveTrafficService({
      port: 19091,
      enabled: true,
      isRuntimeRunning: () => true,
      clientFactory: () => ({
        async getVersion() { return incompatible; },
        async getStartedAt() {
          startedAtCalls += 1;
          return { startedAt: 1n };
        },
        subscribeConnections() { throw new Error('must not subscribe'); },
        subscribeStatus() { throw new Error('must not subscribe'); },
      }),
    });

    service.start();
    await waitFor(() => service.snapshot().source.state === 'incompatible');
    assert.equal(service.snapshot().source.singBoxVersion, incompatible.version);
    assert.equal(service.snapshot().source.singBoxApiVersion, incompatible.apiVersion);
    assert.equal(startedAtCalls, 0);
    await service.stop();
  });
}

test('collector reconnects when either RC5 stream ends and cancels its sibling', async () => {
  let factoryCalls = 0;
  let authorizationCalls = 0;
  const attachedSecrets = [];
  let firstStatusSignal;
  const pending = (signal) => (async function* stream() {
    await new Promise((_, reject) => {
      const abort = () => reject(signal.reason || new Error('aborted'));
      if (signal.aborted) abort();
      else signal.addEventListener('abort', abort, { once: true });
    });
  }());
  const clientFactory = () => {
    factoryCalls += 1;
    const attempt = factoryCalls;
    return {
      async getVersion(_input, { headers }) {
        attachedSecrets.push(headers.authorization);
        return { version: '1.14.0-rc.5', apiVersion: 4 };
      },
      async getStartedAt() {
        return { startedAt: 1_700_000_000_000n };
      },
      subscribeConnections(_input, { signal }) {
        return attempt === 1 ? (async function* ended() {})() : pending(signal);
      },
      subscribeStatus(_input, { signal }) {
        if (attempt === 1) firstStatusSignal = signal;
        return pending(signal);
      },
    };
  };
  const service = createLiveTrafficService({
    port: 19091,
    enabled: true,
    isRuntimeRunning: () => true,
    authorization: () => `secret-${++authorizationCalls}`,
    clientFactory,
  });

  service.start();
  await waitFor(() => factoryCalls >= 2, 2_000);
  assert.equal(firstStatusSignal.aborted, true);
  assert.equal(factoryCalls, 2);
  assert.deepEqual(attachedSecrets, ['Bearer secret-1', 'Bearer secret-2']);
  await service.stop();
});

test('a failed lifecycle projection is retained across one collector reconnect', async () => {
  let factoryCalls = 0;
  const projected = [];
  const idleStream = (signal) => (async function* stream() {
    await new Promise((_, reject) => {
      const abort = () => reject(signal.reason || new Error('aborted'));
      if (signal.aborted) abort();
      else signal.addEventListener('abort', abort, { once: true });
    });
  }());
  const connection = {
    id: 'projection',
    inbound: 'tproxy-in',
    inboundType: 'tproxy',
    network: 'tcp',
    source: '192.168.50.7:54000',
    destination: '203.0.113.10:443',
    domain: 'example.test',
    protocol: 'tls',
    createdAt: 1_700_000_000_000n,
    closedAt: 0n,
    uplinkTotal: 1n,
    downlinkTotal: 0n,
    outbound: 'channel-selector',
    outboundType: 'selector',
    rule: 'final',
    chainList: ['channel-primary', 'channel-selector'],
  };
  const clientFactory = () => {
    factoryCalls += 1;
    const attempt = factoryCalls;
    return {
      async getVersion() {
        return { version: '1.14.0-rc.5', apiVersion: 4 };
      },
      async getStartedAt() {
        return { startedAt: 1_700_000_000_000n };
      },
      subscribeConnections(_input, { signal }) {
        return (async function* stream() {
          yield { reset: true, events: [{ type: 0, id: connection.id, connection }] };
          await new Promise((_, reject) => {
            signal.addEventListener('abort', () => reject(signal.reason || new Error('aborted')), { once: true });
          });
        }());
      },
      subscribeStatus(_input, { signal }) {
        return attempt === 1
          ? idleStream(signal)
          : (async function* stream() {
            while (!signal.aborted) {
              yield { connectionsIn: 1, uplinkTotal: 1n, downlinkTotal: 0n };
              await new Promise((resolve) => setTimeout(resolve, 20));
            }
          }());
      },
    };
  };
  const service = createLiveTrafficService({
    port: 19091,
    enabled: true,
    gateway: true,
    isRuntimeRunning: () => true,
    clientFactory,
    onProjection: async (batch) => {
      projected.push(batch.connections[0]?.traffic.uploadBytes || 'heartbeat');
      if (projected.length === 1) throw new Error('writer unavailable');
    },
  });

  service.start();
  await waitFor(() => factoryCalls >= 2
    && projected.includes('heartbeat')
    && service.snapshot().source.state === 'live', 2_000);
  assert.equal(projected[0], '1');
  assert.ok(projected.filter((value) => value === '1').length >= 2);
  assert.ok(projected.includes('heartbeat'));
  assert.equal(factoryCalls, 2);
  assert.equal(service.snapshot().source.error, null);
  assert.equal(service.snapshot().connections[0].route.kind, 'vpn');
  await service.stop();
});