dcrouter/test/suite/performance/test.message-processing-time.ts

import * as plugins from '@git.zone/tstest/tapbundle';
import { expect, tap } from '@git.zone/tstest/tapbundle';
import * as net from 'net';
import { startTestServer, stopTestServer } from '../../helpers/server.loader.js';

const TEST_PORT = 2525;

let testServer;

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

tap.test('PERF-06: Message processing time - Various message sizes', async (tools) => {
  const done = tools.defer();
  const messageSizes = [1000, 5000, 10000, 25000, 50000]; // bytes
  const messageProcessingTimes: number[] = [];
  const processingRates: number[] = [];

  try {
    const socket = net.createConnection({
      host: 'localhost',
      port: TEST_PORT,
      timeout: 30000
    });

    await new Promise<void>((resolve, reject) => {
      socket.once('connect', resolve);
      socket.once('error', reject);
    });

    // Read greeting
    await new Promise<void>((resolve) => {
      socket.once('data', () => resolve());
    });

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

    console.log('Testing message processing times for various sizes...\n');

    for (let i = 0; i < messageSizes.length; i++) {
      const messageSize = messageSizes[i];
      const messageContent = 'A'.repeat(messageSize);
      
      const messageStart = Date.now();

      // Send MAIL FROM
      socket.write(`MAIL FROM:<sender${i}@example.com>\r\n`);
      
      await new Promise<void>((resolve) => {
        socket.once('data', (chunk) => {
          const response = chunk.toString();
          expect(response).toInclude('250');
          resolve();
        });
      });

      // Send RCPT TO
      socket.write(`RCPT TO:<recipient${i}@example.com>\r\n`);
      
      await new Promise<void>((resolve) => {
        socket.once('data', (chunk) => {
          const response = chunk.toString();
          expect(response).toInclude('250');
          resolve();
        });
      });

      // Send DATA
      socket.write('DATA\r\n');
      
      await new Promise<void>((resolve) => {
        socket.once('data', (chunk) => {
          const response = chunk.toString();
          expect(response).toInclude('354');
          resolve();
        });
      });

      // Send email content
      const emailContent = [
        `From: sender${i}@example.com`,
        `To: recipient${i}@example.com`,
        `Subject: Message Processing Test ${i} (${messageSize} bytes)`,
        '',
        messageContent,
        '.',
        ''
      ].join('\r\n');

      socket.write(emailContent);

      await new Promise<void>((resolve) => {
        socket.once('data', (chunk) => {
          const response = chunk.toString();
          expect(response).toInclude('250');
          resolve();
        });
      });

      const messageProcessingTime = Date.now() - messageStart;
      messageProcessingTimes.push(messageProcessingTime);
      
      const processingRateKBps = (messageSize / 1024) / (messageProcessingTime / 1000);
      processingRates.push(processingRateKBps);

      console.log(`${messageSize} bytes: ${messageProcessingTime}ms (${processingRateKBps.toFixed(1)} KB/s)`);

      // Send RSET
      socket.write('RSET\r\n');
      
      await new Promise<void>((resolve) => {
        socket.once('data', (chunk) => {
          const response = chunk.toString();
          expect(response).toInclude('250');
          resolve();
        });
      });

      // Small delay between tests
      await new Promise(resolve => setTimeout(resolve, 100));
    }

    // Calculate statistics
    const avgProcessingTime = messageProcessingTimes.reduce((a, b) => a + b, 0) / messageProcessingTimes.length;
    const avgProcessingRate = processingRates.reduce((a, b) => a + b, 0) / processingRates.length;
    const minProcessingTime = Math.min(...messageProcessingTimes);
    const maxProcessingTime = Math.max(...messageProcessingTimes);

    console.log(`\nMessage Processing Results:`);
    console.log(`Average processing time: ${avgProcessingTime.toFixed(0)}ms`);
    console.log(`Min/Max processing time: ${minProcessingTime}ms / ${maxProcessingTime}ms`);
    console.log(`Average processing rate: ${avgProcessingRate.toFixed(1)} KB/s`);

    socket.write('QUIT\r\n');
    socket.end();

    // Test passes if average processing time is less than 3000ms and rate > 10KB/s
    expect(avgProcessingTime).toBeLessThan(3000);
    expect(avgProcessingRate).toBeGreaterThan(10);
    done.resolve();
  } catch (error) {
    done.reject(error);
  }
});

tap.test('PERF-06: Message processing time - Large message handling', async (tools) => {
  const done = tools.defer();
  const largeSizes = [100000, 250000, 500000]; // 100KB, 250KB, 500KB
  const results: Array<{ size: number; time: number; rate: number }> = [];

  try {
    const socket = net.createConnection({
      host: 'localhost',
      port: TEST_PORT,
      timeout: 60000 // Longer timeout for large messages
    });

    await new Promise<void>((resolve, reject) => {
      socket.once('connect', resolve);
      socket.once('error', reject);
    });

    // Read greeting
    await new Promise<void>((resolve) => {
      socket.once('data', () => resolve());
    });

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

    console.log('\nTesting large message processing...\n');

    for (let i = 0; i < largeSizes.length; i++) {
      const messageSize = largeSizes[i];
      
      const messageStart = Date.now();

      // Send MAIL FROM
      socket.write(`MAIL FROM:<largesender${i}@example.com>\r\n`);
      
      await new Promise<void>((resolve) => {
        socket.once('data', (chunk) => {
          const response = chunk.toString();
          expect(response).toInclude('250');
          resolve();
        });
      });

      // Send RCPT TO
      socket.write(`RCPT TO:<largerecipient${i}@example.com>\r\n`);
      
      await new Promise<void>((resolve) => {
        socket.once('data', (chunk) => {
          const response = chunk.toString();
          expect(response).toInclude('250');
          resolve();
        });
      });

      // Send DATA
      socket.write('DATA\r\n');
      
      await new Promise<void>((resolve) => {
        socket.once('data', (chunk) => {
          const response = chunk.toString();
          expect(response).toInclude('354');
          resolve();
        });
      });

      // Send large email content in chunks to avoid buffer issues
      socket.write(`From: largesender${i}@example.com\r\n`);
      socket.write(`To: largerecipient${i}@example.com\r\n`);
      socket.write(`Subject: Large Message Test ${i} (${messageSize} bytes)\r\n\r\n`);

      // Send content in 10KB chunks
      const chunkSize = 10000;
      let remaining = messageSize;
      while (remaining > 0) {
        const currentChunk = Math.min(remaining, chunkSize);
        socket.write('B'.repeat(currentChunk));
        remaining -= currentChunk;
        
        // Small delay to avoid overwhelming buffers
        if (remaining > 0) {
          await new Promise(resolve => setTimeout(resolve, 10));
        }
      }

      socket.write('\r\n.\r\n');

      const response = await new Promise<string>((resolve, reject) => {
        const timeout = setTimeout(() => {
          reject(new Error('Timeout waiting for message response'));
        }, 30000);

        socket.once('data', (chunk) => {
          clearTimeout(timeout);
          resolve(chunk.toString());
        });
      });

      expect(response).toInclude('250');

      const messageProcessingTime = Date.now() - messageStart;
      const processingRateMBps = (messageSize / (1024 * 1024)) / (messageProcessingTime / 1000);
      
      results.push({
        size: messageSize,
        time: messageProcessingTime,
        rate: processingRateMBps
      });

      console.log(`${(messageSize/1024).toFixed(0)}KB: ${messageProcessingTime}ms (${processingRateMBps.toFixed(2)} MB/s)`);

      // Send RSET
      socket.write('RSET\r\n');
      
      await new Promise<void>((resolve) => {
        socket.once('data', () => resolve());
      });

      // Delay between large tests
      await new Promise(resolve => setTimeout(resolve, 500));
    }

    const avgRate = results.reduce((sum, r) => sum + r.rate, 0) / results.length;
    console.log(`\nAverage large message rate: ${avgRate.toFixed(2)} MB/s`);

    socket.write('QUIT\r\n');
    socket.end();

    // Test passes if we can process at least 0.5 MB/s
    expect(avgRate).toBeGreaterThan(0.5);
    done.resolve();
  } catch (error) {
    done.reject(error);
  }
});

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

tap.start();