wwjcloud-nest-v1/wwjcloud/test/queue/queue-system.e2e-spec.ts

import { Test, TestingModule } from '@nestjs/testing';
import { INestApplication } from '@nestjs/common';
import request from 'supertest';
import { TestModule } from '../test.module';
import { TestService } from '../test.service';
import { UnifiedQueueService } from '../../src/core/queue/unifiedQueueService';

describe('Queue System (e2e)', () => {
  let app: INestApplication;
  let testService: TestService;
  let unifiedQueueService: UnifiedQueueService;

  beforeAll(async () => {
    const moduleFixture: TestingModule = await Test.createTestingModule({
      imports: [TestModule],
    }).compile();

    app = moduleFixture.createNestApplication();
    await app.init();

    testService = moduleFixture.get<TestService>(TestService);
    unifiedQueueService =
      moduleFixture.get<UnifiedQueueService>(UnifiedQueueService);
  });

  afterAll(async () => {
    if (app) {
      await app.close();
    }
  });

  describe('Test Controller Endpoints', () => {
    it('/test/status (GET) - should return service status', () => {
      return request(app.getHttpServer())
        .get('/test/status')
        .expect(200)
        .expect((res) => {
          expect(res.body).toHaveProperty('message');
          expect(res.body).toHaveProperty('timestamp');
          expect(res.body).toHaveProperty('services');
          expect(res.body.services).toHaveProperty('redis');
          expect(res.body.services).toHaveProperty('kafka');
        });
    });

    it('/test/kafka (POST) - should publish event to Kafka', () => {
      const testData = { test: 'kafka-event', value: 123 };
      return request(app.getHttpServer())
        .post('/test/kafka')
        .send(testData)
        .expect(201)
        .expect((res) => {
          expect(res.body).toHaveProperty('success', true);
          expect(res.body).toHaveProperty('message');
          expect(res.body).toHaveProperty('topic', 'test-topic');
          expect(res.body).toHaveProperty('data', testData);
        });
    });

    it('/test/redis (POST) - should enqueue job to Redis', () => {
      const testData = { test: 'redis-job', value: 456 };
      return request(app.getHttpServer())
        .post('/test/redis')
        .send(testData)
        .expect(201)
        .expect((res) => {
          expect(res.body).toHaveProperty('success', true);
          expect(res.body).toHaveProperty('message');
          expect(res.body).toHaveProperty('jobId');
          expect(res.body).toHaveProperty('data', testData);
        });
    });
  });

  describe('UnifiedQueueService', () => {
    it('should be defined', () => {
      expect(unifiedQueueService).toBeDefined();
    });

    it('should add task to queue', async () => {
      await expect(
        unifiedQueueService.addTask(
          'test-queue',
          'test-task',
          { test: 'data' },
          {
            priority: 1,
            delay: 0,
            attempts: 3,
          },
        ),
      ).resolves.toBeUndefined();
    });

    it('should process task from queue', async () => {
      let processedData: any = null;

      await unifiedQueueService.processTask('test-queue', async (job: any) => {
        processedData = job.data;
        return { success: true };
      });

      // Add a task to be processed
      await unifiedQueueService.addTask(
        'test-queue',
        'process-task',
        { test: 'process-data' },
        {
          priority: 1,
        },
      );

      // Wait a bit for processing
      await new Promise((resolve) => setTimeout(resolve, 300));

      expect(processedData).toBeDefined();
    });

    it('should publish event', async () => {
      const event = {
        eventType: 'test.event',
        aggregateId: 'test-123',
        aggregateType: 'Test',
        version: '1.0',
        occurredAt: new Date().toISOString(),
        tenantId: 'tenant-1',
        idempotencyKey: 'key-123',
        traceId: 'trace-123',
        data: { test: 'event-data' },
      };

      await expect(
        unifiedQueueService.publishEvent(event),
      ).resolves.not.toThrow();
    });
  });

  describe('Service Integration', () => {
    it('should have all required services available', () => {
      expect(testService).toBeDefined();
      expect(unifiedQueueService).toBeDefined();
    });
  });

  describe('Integration Tests', () => {
    it('should handle complete queue workflow', async () => {
      // Test the complete workflow: add task -> process task -> publish event
      const taskData = { workflow: 'test', step: 1 };

      // Add task
      await expect(
        unifiedQueueService.addTask(
          'workflow-queue',
          'workflow-task',
          taskData,
          {
            priority: 1,
          },
        ),
      ).resolves.toBeUndefined();

      // Process task and publish event
      await unifiedQueueService.processTask(
        'workflow-queue',
        async (job: any) => {
          const event = {
            eventType: 'workflow.completed',
            aggregateId: 'workflow-123',
            aggregateType: 'Workflow',
            version: '1.0',
            occurredAt: new Date().toISOString(),
            tenantId: 'tenant-1',
            idempotencyKey: 'workflow-key-123',
            traceId: 'workflow-trace-123',
            data: job.data,
          };

          await unifiedQueueService.publishEvent(event);
          return { success: true, processed: job.data };
        },
      );

      // Wait for processing
      await new Promise((resolve) => setTimeout(resolve, 300));
    });

    it('should handle error scenarios gracefully', async () => {
      // Test error handling in task processing
      await unifiedQueueService.processTask('error-queue', async (job: any) => {
        if (job.data.shouldFail) {
          throw new Error('Intentional test error');
        }
        return { success: true };
      });

      // Add a failing task
      await unifiedQueueService.addTask(
        'error-queue',
        'error',
        { shouldFail: true },
        {
          priority: 1,
          attempts: 1, // Only try once
        },
      );

      // Wait for processing attempt
      await new Promise((resolve) => setTimeout(resolve, 300));

      // The test passes if no unhandled errors are thrown
      expect(true).toBe(true);
    });
  });
});