dcrouter/test/suite/performance/test.concurrency.ts

import * as plugins from '@push.rocks/tapbundle';
import { expect, tap } from '@push.rocks/tapbundle';
import * as net from 'net';
import { startTestServer, stopTestServer } from '../server.loader.js';

const TEST_PORT = 2525;

tap.test('prepare server', async () => {
  await startTestServer();
  await new Promise(resolve => setTimeout(resolve, 100));
});

tap.test('PERF-02: Concurrency testing - Multiple simultaneous connections', async (tools) => {
  const done = tools.defer();
  const concurrentCount = 20;
  const connectionResults: Array<{
    connectionId: number;
    success: boolean;
    duration: number;
    error?: string;
  }> = [];

  const createConcurrentConnection = (connectionId: number): Promise<void> => {
    return new Promise((resolve) => {
      const startTime = Date.now();
      const socket = net.createConnection({
        host: 'localhost',
        port: TEST_PORT,
        timeout: 10000
      });

      let state = 'connecting';
      let receivedData = '';

      const timeoutHandle = setTimeout(() => {
        socket.destroy();
        connectionResults.push({
          connectionId,
          success: false,
          duration: Date.now() - startTime,
          error: 'Connection timeout'
        });
        resolve();
      }, 10000);

      socket.on('connect', () => {
        state = 'connected';
      });

      socket.on('data', (chunk) => {
        receivedData += chunk.toString();
        const lines = receivedData.split('\r\n');

        for (const line of lines) {
          if (!line.trim()) continue;

          if (state === 'connected' && line.startsWith('220')) {
            state = 'ehlo';
            socket.write(`EHLO testhost-${connectionId}\r\n`);
          } else if (state === 'ehlo' && line.includes('250 ') && !line.includes('250-')) {
            // Final 250 response received
            state = 'quit';
            socket.write('QUIT\r\n');
          } else if (state === 'quit' && line.startsWith('221')) {
            clearTimeout(timeoutHandle);
            socket.end();
            connectionResults.push({
              connectionId,
              success: true,
              duration: Date.now() - startTime
            });
            resolve();
          }
        }
      });

      socket.on('error', (error) => {
        clearTimeout(timeoutHandle);
        connectionResults.push({
          connectionId,
          success: false,
          duration: Date.now() - startTime,
          error: error.message
        });
        resolve();
      });

      socket.on('close', () => {
        clearTimeout(timeoutHandle);
        if (!connectionResults.find(r => r.connectionId === connectionId)) {
          connectionResults.push({
            connectionId,
            success: false,
            duration: Date.now() - startTime,
            error: 'Connection closed unexpectedly'
          });
        }
        resolve();
      });
    });
  };

  try {
    // Create all concurrent connections
    const promises: Promise<void>[] = [];
    console.log(`Creating ${concurrentCount} concurrent connections...`);

    for (let i = 0; i < concurrentCount; i++) {
      promises.push(createConcurrentConnection(i));
      // Small stagger to avoid overwhelming the system
      if (i % 5 === 0) {
        await new Promise(resolve => setTimeout(resolve, 10));
      }
    }

    // Wait for all connections to complete
    await Promise.all(promises);

    // Analyze results
    const successful = connectionResults.filter(r => r.success).length;
    const failed = connectionResults.filter(r => !r.success).length;
    const successRate = successful / concurrentCount;
    const avgDuration = connectionResults
      .filter(r => r.success)
      .reduce((sum, r) => sum + r.duration, 0) / successful || 0;

    console.log(`\nConcurrency Test Results:`);
    console.log(`Total connections: ${concurrentCount}`);
    console.log(`Successful: ${successful} (${(successRate * 100).toFixed(1)}%)`);
    console.log(`Failed: ${failed}`);
    console.log(`Average duration: ${avgDuration.toFixed(0)}ms`);

    if (failed > 0) {
      const errors = connectionResults
        .filter(r => !r.success)
        .map(r => r.error)
        .filter((v, i, a) => a.indexOf(v) === i); // unique errors
      console.log(`Unique errors: ${errors.join(', ')}`);
    }

    // Success if at least 80% of connections succeed
    expect(successRate).toBeGreaterThanOrEqual(0.8);
    done.resolve();
  } catch (error) {
    done.reject(error);
  }
});

tap.test('PERF-02: Concurrency testing - Concurrent transactions', async (tools) => {
  const done = tools.defer();
  const transactionCount = 10;
  const transactionResults: Array<{
    transactionId: number;
    success: boolean;
    duration: number;
    error?: string;
  }> = [];

  const performConcurrentTransaction = (transactionId: number): Promise<void> => {
    return new Promise((resolve) => {
      const startTime = Date.now();
      const socket = net.createConnection({
        host: 'localhost',
        port: TEST_PORT,
        timeout: 15000
      });

      let state = 'connecting';

      const timeoutHandle = setTimeout(() => {
        socket.destroy();
        transactionResults.push({
          transactionId,
          success: false,
          duration: Date.now() - startTime,
          error: 'Transaction timeout'
        });
        resolve();
      }, 15000);

      const processResponse = async () => {
        try {
          // Read greeting
          await new Promise<void>((res) => {
            socket.once('data', () => res());
          });

          // Send EHLO
          socket.write(`EHLO testhost-tx-${transactionId}\r\n`);
          
          await new Promise<void>((res) => {
            let data = '';
            const handleData = (chunk: Buffer) => {
              data += chunk.toString();
              if (data.includes('250 ') && !data.includes('250-')) {
                socket.removeListener('data', handleData);
                res();
              }
            };
            socket.on('data', handleData);
          });

          // Complete email transaction
          socket.write(`MAIL FROM:<sender${transactionId}@example.com>\r\n`);
          
          await new Promise<void>((res) => {
            socket.once('data', (chunk) => {
              const response = chunk.toString();
              if (!response.includes('250')) {
                throw new Error('MAIL FROM failed');
              }
              res();
            });
          });

          socket.write(`RCPT TO:<recipient${transactionId}@example.com>\r\n`);
          
          await new Promise<void>((res) => {
            socket.once('data', (chunk) => {
              const response = chunk.toString();
              if (!response.includes('250')) {
                throw new Error('RCPT TO failed');
              }
              res();
            });
          });

          socket.write('DATA\r\n');
          
          await new Promise<void>((res) => {
            socket.once('data', (chunk) => {
              const response = chunk.toString();
              if (!response.includes('354')) {
                throw new Error('DATA command failed');
              }
              res();
            });
          });

          // Send email content
          const emailContent = [
            `From: sender${transactionId}@example.com`,
            `To: recipient${transactionId}@example.com`,
            `Subject: Concurrent test ${transactionId}`,
            '',
            `This is concurrent test message ${transactionId}`,
            '.',
            ''
          ].join('\r\n');

          socket.write(emailContent);

          await new Promise<void>((res) => {
            socket.once('data', (chunk) => {
              const response = chunk.toString();
              if (!response.includes('250')) {
                throw new Error('Message submission failed');
              }
              res();
            });
          });

          socket.write('QUIT\r\n');
          
          await new Promise<void>((res) => {
            socket.once('data', () => res());
          });

          clearTimeout(timeoutHandle);
          socket.end();

          transactionResults.push({
            transactionId,
            success: true,
            duration: Date.now() - startTime
          });
          resolve();

        } catch (error) {
          clearTimeout(timeoutHandle);
          socket.end();
          transactionResults.push({
            transactionId,
            success: false,
            duration: Date.now() - startTime,
            error: error instanceof Error ? error.message : 'Unknown error'
          });
          resolve();
        }
      };

      socket.on('connect', () => {
        state = 'connected';
        processResponse();
      });

      socket.on('error', (error) => {
        clearTimeout(timeoutHandle);
        if (!transactionResults.find(r => r.transactionId === transactionId)) {
          transactionResults.push({
            transactionId,
            success: false,
            duration: Date.now() - startTime,
            error: error.message
          });
        }
        resolve();
      });
    });
  };

  try {
    // Create concurrent transactions
    const promises: Promise<void>[] = [];
    console.log(`\nStarting ${transactionCount} concurrent email transactions...`);

    for (let i = 0; i < transactionCount; i++) {
      promises.push(performConcurrentTransaction(i));
      // Small stagger
      await new Promise(resolve => setTimeout(resolve, 50));
    }

    // Wait for all transactions
    await Promise.all(promises);

    // Analyze results
    const successful = transactionResults.filter(r => r.success).length;
    const failed = transactionResults.filter(r => !r.success).length;
    const successRate = successful / transactionCount;
    const avgDuration = transactionResults
      .filter(r => r.success)
      .reduce((sum, r) => sum + r.duration, 0) / successful || 0;

    console.log(`\nConcurrent Transaction Results:`);
    console.log(`Total transactions: ${transactionCount}`);
    console.log(`Successful: ${successful} (${(successRate * 100).toFixed(1)}%)`);
    console.log(`Failed: ${failed}`);
    console.log(`Average duration: ${avgDuration.toFixed(0)}ms`);

    // Success if at least 80% of transactions complete
    expect(successRate).toBeGreaterThanOrEqual(0.8);
    done.resolve();
  } catch (error) {
    done.reject(error);
  }
});

tap.test('cleanup server', async () => {
  await stopTestServer();
});

tap.start();