test/test.perf-improvements.ts

import { tap, expect } from '@git.zone/tstest/tapbundle';
import { SmartProxy } from '../ts/index.js';
import * as http from 'http';
import * as https from 'https';
import * as http2 from 'http2';
import * as net from 'net';
import * as tls from 'tls';
import * as fs from 'fs';
import * as path from 'path';

// ---------------------------------------------------------------------------
// Port assignments (47600–47620 range to avoid conflicts)
// ---------------------------------------------------------------------------
const HTTP_ECHO_PORT  = 47600; // backend HTTP echo server
const PROXY_HTTP_PORT = 47601; // SmartProxy plain HTTP forwarding
const PROXY_HTTPS_PORT = 47602; // SmartProxy TLS-terminate HTTPS forwarding
const TCP_ECHO_PORT   = 47603; // backend TCP echo server
const PROXY_TCP_PORT  = 47604; // SmartProxy plain TCP forwarding

// ---------------------------------------------------------------------------
// Shared state
// ---------------------------------------------------------------------------
let httpEchoServer: http.Server;
let tcpEchoServer: net.Server;
let proxy: SmartProxy;

const certPem = fs.readFileSync(path.join(import.meta.dirname, '..', 'assets', 'certs', 'cert.pem'), 'utf8');
const keyPem  = fs.readFileSync(path.join(import.meta.dirname, '..', 'assets', 'certs', 'key.pem'), 'utf8');

// ---------------------------------------------------------------------------
// Helper: make an HTTP request and return { status, body }
// ---------------------------------------------------------------------------
function httpRequest(
  options: http.RequestOptions,
  body?: string,
): Promise<{ status: number; body: string }> {
  return new Promise((resolve, reject) => {
    const req = http.request(options, (res) => {
      let data = '';
      res.on('data', (chunk: string) => (data += chunk));
      res.on('end', () => resolve({ status: res.statusCode!, body: data }));
    });
    req.on('error', reject);
    req.setTimeout(5000, () => {
      req.destroy(new Error('timeout'));
    });
    if (body) req.end(body);
    else req.end();
  });
}

// Same but for HTTPS
function httpsRequest(
  options: https.RequestOptions,
  body?: string,
): Promise<{ status: number; body: string }> {
  return new Promise((resolve, reject) => {
    const req = https.request(options, (res) => {
      let data = '';
      res.on('data', (chunk: string) => (data += chunk));
      res.on('end', () => resolve({ status: res.statusCode!, body: data }));
    });
    req.on('error', reject);
    req.setTimeout(5000, () => {
      req.destroy(new Error('timeout'));
    });
    if (body) req.end(body);
    else req.end();
  });
}

// Helper: wait for metrics to settle on a condition
async function waitForMetrics(
  metrics: ReturnType<SmartProxy['getMetrics']>,
  condition: () => boolean,
  maxWaitMs = 3000,
): Promise<void> {
  const start = Date.now();
  while (Date.now() - start < maxWaitMs) {
    // Force a fresh poll
    await (proxy as any).metricsAdapter.poll();
    if (condition()) return;
    await new Promise((r) => setTimeout(r, 100));
  }
}

// ===========================================================================
// 1. Setup backend servers
// ===========================================================================
tap.test('setup - backend servers', async () => {
  // HTTP echo server: POST → echo:<body>, GET → ok
  httpEchoServer = http.createServer((req, res) => {
    if (req.method === 'POST') {
      let body = '';
      req.on('data', (chunk: string) => (body += chunk));
      req.on('end', () => {
        res.writeHead(200, { 'Content-Type': 'text/plain' });
        res.end(`echo:${body}`);
      });
    } else {
      res.writeHead(200, { 'Content-Type': 'text/plain' });
      res.end('ok');
    }
  });

  await new Promise<void>((resolve, reject) => {
    httpEchoServer.on('error', reject);
    httpEchoServer.listen(HTTP_ECHO_PORT, () => {
      console.log(`HTTP echo server on port ${HTTP_ECHO_PORT}`);
      resolve();
    });
  });

  // TCP echo server
  tcpEchoServer = net.createServer((socket) => {
    socket.on('data', (data) => socket.write(data));
  });

  await new Promise<void>((resolve, reject) => {
    tcpEchoServer.on('error', reject);
    tcpEchoServer.listen(TCP_ECHO_PORT, () => {
      console.log(`TCP echo server on port ${TCP_ECHO_PORT}`);
      resolve();
    });
  });
});

// ===========================================================================
// 2. Setup SmartProxy
// ===========================================================================
tap.test('setup - SmartProxy with 3 routes', async () => {
  proxy = new SmartProxy({
    routes: [
      // Plain HTTP forward: 47601 → 47600
      {
        name: 'http-forward',
        match: { ports: PROXY_HTTP_PORT },
        action: {
          type: 'forward',
          targets: [{ host: 'localhost', port: HTTP_ECHO_PORT }],
        },
      },
      // TLS-terminate HTTPS: 47602 → 47600
      {
        name: 'https-terminate',
        match: { ports: PROXY_HTTPS_PORT, domains: 'localhost' },
        action: {
          type: 'forward',
          targets: [{ host: 'localhost', port: HTTP_ECHO_PORT }],
          tls: {
            mode: 'terminate',
            certificate: {
              key: keyPem,
              cert: certPem,
            },
          },
        },
      },
      // Plain TCP forward: 47604 → 47603
      {
        name: 'tcp-forward',
        match: { ports: PROXY_TCP_PORT },
        action: {
          type: 'forward',
          targets: [{ host: 'localhost', port: TCP_ECHO_PORT }],
        },
      },
    ],
    metrics: {
      enabled: true,
      sampleIntervalMs: 100,
    },
    enableDetailedLogging: false,
  });

  await proxy.start();
  // Give the proxy a moment to fully bind
  await new Promise((r) => setTimeout(r, 500));
});

// ===========================================================================
// 3. HTTP/1.1 connection pooling: sequential requests reuse connections
// ===========================================================================
tap.test('HTTP/1.1 connection pooling: sequential requests reuse connections', async (tools) => {
  tools.timeout(30000);
  const metrics = proxy.getMetrics();
  const REQUEST_COUNT = 20;

  // Use a non-keepalive agent so each request closes the client→proxy socket
  // (Rust's backend connection pool still reuses proxy→backend connections)
  const agent = new http.Agent({ keepAlive: false });

  for (let i = 0; i < REQUEST_COUNT; i++) {
    const result = await httpRequest(
      {
        hostname: 'localhost',
        port: PROXY_HTTP_PORT,
        path: '/echo',
        method: 'POST',
        headers: { 'Content-Type': 'text/plain' },
        agent,
      },
      `msg-${i}`,
    );
    expect(result.status).toEqual(200);
    expect(result.body).toEqual(`echo:msg-${i}`);
  }

  agent.destroy();

  // Wait for all connections to settle and metrics to update
  await waitForMetrics(metrics, () => metrics.connections.active() === 0, 5000);
  expect(metrics.connections.active()).toEqual(0);

  // Bytes should have been transferred
  await waitForMetrics(metrics, () => metrics.totals.bytesIn() > 0);
  expect(metrics.totals.bytesIn()).toBeGreaterThan(0);
  expect(metrics.totals.bytesOut()).toBeGreaterThan(0);

  console.log(`HTTP pooling test: ${REQUEST_COUNT} requests completed. bytesIn=${metrics.totals.bytesIn()}, bytesOut=${metrics.totals.bytesOut()}`);
});

// ===========================================================================
// 4. HTTPS with TLS termination: multiple requests through TLS
// ===========================================================================
tap.test('HTTPS with TLS termination: multiple requests through TLS', async (tools) => {
  tools.timeout(30000);
  const REQUEST_COUNT = 10;

  const agent = new https.Agent({ keepAlive: false, rejectUnauthorized: false });

  for (let i = 0; i < REQUEST_COUNT; i++) {
    const result = await httpsRequest(
      {
        hostname: 'localhost',
        port: PROXY_HTTPS_PORT,
        path: '/echo',
        method: 'POST',
        headers: { 'Content-Type': 'text/plain' },
        rejectUnauthorized: false,
        servername: 'localhost',
        agent,
      },
      `tls-${i}`,
    );
    expect(result.status).toEqual(200);
    expect(result.body).toEqual(`echo:tls-${i}`);
  }

  agent.destroy();

  console.log(`HTTPS termination test: ${REQUEST_COUNT} requests completed successfully`);
});

// ===========================================================================
// 5. TLS ALPN negotiation verification
// ===========================================================================
tap.test('HTTP/2 end-to-end: ALPN h2 with multiplexed requests', async (tools) => {
  tools.timeout(15000);

  // Connect an HTTP/2 session over TLS
  const session = http2.connect(`https://localhost:${PROXY_HTTPS_PORT}`, {
    rejectUnauthorized: false,
  });

  await new Promise<void>((resolve, reject) => {
    session.on('connect', () => resolve());
    session.on('error', reject);
    setTimeout(() => reject(new Error('h2 connect timeout')), 5000);
  });

  // Verify ALPN negotiated h2
  const alpnProtocol = (session.socket as tls.TLSSocket).alpnProtocol;
  console.log(`TLS ALPN negotiated protocol: ${alpnProtocol}`);
  expect(alpnProtocol).toEqual('h2');

  // Send 5 multiplexed POST requests on the same h2 session
  const REQUEST_COUNT = 5;
  const promises: Promise<{ status: number; body: string }>[] = [];

  for (let i = 0; i < REQUEST_COUNT; i++) {
    promises.push(
      new Promise<{ status: number; body: string }>((resolve, reject) => {
        const reqStream = session.request({
          ':method': 'POST',
          ':path': '/echo',
          'content-type': 'text/plain',
        });

        let data = '';
        let status = 0;

        reqStream.on('response', (headers) => {
          status = headers[':status'] as number;
        });
        reqStream.on('data', (chunk: Buffer) => {
          data += chunk.toString();
        });
        reqStream.on('end', () => resolve({ status, body: data }));
        reqStream.on('error', reject);
        reqStream.end(`h2-msg-${i}`);
      }),
    );
  }

  const results = await Promise.all(promises);
  for (let i = 0; i < REQUEST_COUNT; i++) {
    expect(results[i].status).toEqual(200);
    expect(results[i].body).toEqual(`echo:h2-msg-${i}`);
  }

  await new Promise<void>((resolve) => session.close(() => resolve()));
  console.log(`HTTP/2 end-to-end: ${REQUEST_COUNT} multiplexed requests completed successfully`);
});

// ===========================================================================
// 6. Connection stability: no leaked connections after repeated open/close
// ===========================================================================
tap.test('connection stability: no leaked connections after repeated open/close', async (tools) => {
  tools.timeout(60000);
  const metrics = proxy.getMetrics();
  const BATCH_SIZE = 50;

  // Ensure we start clean
  await waitForMetrics(metrics, () => metrics.connections.active() === 0);

  // Record total connections before
  await (proxy as any).metricsAdapter.poll();
  const totalBefore = metrics.connections.total();

  // --- Batch 1: 50 sequential TCP connections ---
  for (let i = 0; i < BATCH_SIZE; i++) {
    await new Promise<void>((resolve, reject) => {
      const client = new net.Socket();
      client.connect(PROXY_TCP_PORT, 'localhost', () => {
        const msg = `batch1-${i}`;
        client.write(msg);
        client.once('data', (data) => {
          expect(data.toString()).toEqual(msg);
          client.end();
        });
      });
      client.on('close', () => resolve());
      client.on('error', reject);
      client.setTimeout(5000, () => {
        client.destroy(new Error('timeout'));
      });
    });
  }

  // Wait for all connections to drain
  await waitForMetrics(metrics, () => metrics.connections.active() === 0, 5000);
  expect(metrics.connections.active()).toEqual(0);
  console.log(`Batch 1 done: active=${metrics.connections.active()}, total=${metrics.connections.total()}`);

  // --- Batch 2: another 50 ---
  for (let i = 0; i < BATCH_SIZE; i++) {
    await new Promise<void>((resolve, reject) => {
      const client = new net.Socket();
      client.connect(PROXY_TCP_PORT, 'localhost', () => {
        const msg = `batch2-${i}`;
        client.write(msg);
        client.once('data', (data) => {
          expect(data.toString()).toEqual(msg);
          client.end();
        });
      });
      client.on('close', () => resolve());
      client.on('error', reject);
      client.setTimeout(5000, () => {
        client.destroy(new Error('timeout'));
      });
    });
  }

  // Wait for all connections to drain again
  await waitForMetrics(metrics, () => metrics.connections.active() === 0, 5000);
  expect(metrics.connections.active()).toEqual(0);

  // Total should reflect ~100 new connections
  await (proxy as any).metricsAdapter.poll();
  const totalAfter = metrics.connections.total();
  const newConnections = totalAfter - totalBefore;
  console.log(`Batch 2 done: active=${metrics.connections.active()}, total=${totalAfter}, new=${newConnections}`);
  expect(newConnections).toBeGreaterThanOrEqual(BATCH_SIZE * 2);
});

// ===========================================================================
// 7. Concurrent connections: burst and drain
// ===========================================================================
tap.test('concurrent connections: burst and drain', async (tools) => {
  tools.timeout(30000);
  const metrics = proxy.getMetrics();
  const CONCURRENT = 20;

  // Ensure we start clean
  await waitForMetrics(metrics, () => metrics.connections.active() === 0, 5000);

  // Open 20 TCP connections simultaneously
  const clients: net.Socket[] = [];
  const connectPromises: Promise<void>[] = [];

  for (let i = 0; i < CONCURRENT; i++) {
    const client = new net.Socket();
    clients.push(client);
    connectPromises.push(
      new Promise<void>((resolve, reject) => {
        client.connect(PROXY_TCP_PORT, 'localhost', () => resolve());
        client.on('error', reject);
        client.setTimeout(5000, () => {
          client.destroy(new Error('timeout'));
        });
      }),
    );
  }

  await Promise.all(connectPromises);

  // Send data on all connections and wait for echo
  const echoPromises = clients.map((client, i) => {
    return new Promise<void>((resolve, reject) => {
      const msg = `concurrent-${i}`;
      client.once('data', (data) => {
        expect(data.toString()).toEqual(msg);
        resolve();
      });
      client.write(msg);
      client.on('error', reject);
    });
  });

  await Promise.all(echoPromises);

  // Poll metrics — active connections should be CONCURRENT
  await waitForMetrics(metrics, () => metrics.connections.active() >= CONCURRENT, 3000);
  const activeWhileOpen = metrics.connections.active();
  console.log(`Burst: active connections while open = ${activeWhileOpen}`);
  expect(activeWhileOpen).toBeGreaterThanOrEqual(CONCURRENT);

  // Close all connections
  for (const client of clients) {
    client.end();
  }

  // Wait for drain
  await waitForMetrics(metrics, () => metrics.connections.active() === 0, 5000);
  expect(metrics.connections.active()).toEqual(0);
  console.log('Drain: all connections closed, active=0');
});

// ===========================================================================
// 8. Cleanup
// ===========================================================================
tap.test('cleanup', async () => {
  await proxy.stop();

  await new Promise<void>((resolve) => {
    httpEchoServer.close(() => {
      console.log('HTTP echo server closed');
      resolve();
    });
  });

  await new Promise<void>((resolve) => {
    tcpEchoServer.close(() => {
      console.log('TCP echo server closed');
      resolve();
    });
  });
});

export default tap.start();