Создаем микросервисы с NestJS, TCP и Typescript

// shared/src/lib/dto/user.dto.ts
export class User {
  id?: number;
  username: string;
  email: string;
  isActive: boolean;
}
 
// shared/src/lib/patterns/auth.patterns.ts
export const AUTH_SERVICE_PATTERNS = {
  CREATE_USER: 'create_user',
  GET_USER: 'get_user',
  VALIDATE_USER: 'validate_user'
}

@EventPattern('order.created')
handleOrderCreated(data: OrderCreatedEvent) {
  try {
    // Обработка заказа
    this.eventBus.emit('payment.process', { orderId: data.id });
  } catch (error) {
    // Компенсирующее действие
    this.eventBus.emit('order.cancel', { orderId: data.id });
  }
}

const user = await this.authClient.send('get_user', userDto).toPromise();

this.paymentClient.emit('payment.processed', { orderId: '123', status: 'success' });

this.httpService.get('http://service-a/resource')
  .pipe(
    retry(3),
    timeout(5000),
    catchError(error => of({ fallback: true }))
  )
  .subscribe(response => this.handleResponse(response));

@Injectable()
export class TraceIdInterceptor implements NestInterceptor {
  intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
    const request = context.switchToHttp().getRequest();
    const traceId = request.headers['x-trace-id'] || uuidv4();
    
    // Добавляем traceId в контекст запроса
    request.traceId = traceId;
    
    // Добавляем traceId в заголовки ответа
    const response = context.switchToHttp().getResponse();
    response.header('x-trace-id', traceId);
    
    return next.handle();
  }
}

// shared/src/lib/dto/create-user.dto.ts
import { IsNotEmpty, IsString } from 'class-validator';
 
export class CreateUserDto {
  @IsString()
  @IsNotEmpty()
  username: string;
 
  @IsNotEmpty()
  password: string;
}
 
// shared/src/lib/entities/user.entity.ts
export class User {
  id?: number;
  username: string;
  password: string;
}

// apps/api-gateway/src/auth/auth.module.ts
import { Module } from '@nestjs/common';
import { ClientsModule, Transport } from '@nestjs/microservices';
 
import { AuthController } from './auth.controller';
import { AuthService } from './auth.service';
 
@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'AUTH_MICROSERVICE',
        transport: Transport.TCP,
        options: {
          host: 'localhost',
          port: 3001,
        },
      },
    ]),
  ],
  providers: [AuthService],
  controllers: [AuthController],
})
export class AuthModule {}

// apps/auth-microservice/src/main.ts
import { Logger } from '@nestjs/common';
import { NestFactory } from '@nestjs/core';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';
 
import { AppModule } from './app/app.module';
 
async function bootstrap() {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(
    AppModule,
    {
      transport: Transport.TCP,
      options: {
        host: 'localhost',
        port: 3001,
      },
    }
  );
 
  await app.listen();
  Logger.log('Auth микросервис запущен');
}
bootstrap();

// shared/src/lib/constants/message-patterns.ts
export const AUTH_MESSAGE_PATTERNS = {
  GET_USER: 'get_user',
  CREATE_USER: 'create_user',
  VALIDATE_TOKEN: 'validate_token'
};

// main.ts (api-gateway)
import { ValidationPipe } from '@nestjs/common';
 
async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  app.useGlobalPipes(new ValidationPipe({
    whitelist: true, // Удаляет свойства, которые не определены в DTO
    forbidNonWhitelisted: true, // Выбрасывает ошибку при попытке передать неопределенные свойства
    transform: true, // Автоматически преобразует примитивные типы
  }));
  await app.listen(3000);
}

// apps/auth-microservice/src/app/user.repository.ts
import { Injectable } from '@nestjs/common';
 
import { CreateUserDto, User } from '@nestjs-microservices/shared';
 
@Injectable()
export class UserRepository {
  private users: User[] = [];
 
  save(user: CreateUserDto): User {
    const newUser = new User();
    newUser.id = this.users.length + 1;
    newUser.username = user.username;
    newUser.password = user.password;
    this.users.push(newUser);
    return newUser;
  }
 
  findOne(username: string): User | undefined {
    return this.users.find((user) => user.username === username);
  }
}

// apps/auth-microservice/src/app/app.service.ts
import { Injectable } from '@nestjs/common';
 
import { CreateUserDto, User } from '@nestjs-microservices/shared';
 
import { UserRepository } from './user.repository';
 
@Injectable()
export class AppService {
  constructor(private readonly userRepository: UserRepository) {}
 
  createUser(newUser: CreateUserDto): User {
    return this.userRepository.save(newUser);
  }
 
  getUser(username: string): User | undefined {
    return this.userRepository.findOne(username);
  }
}

// apps/auth-microservice/src/app/app.controller.ts
import { Controller } from '@nestjs/common';
import { MessagePattern } from '@nestjs/microservices';
 
import { CreateUserDto, User } from '@nestjs-microservices/shared';
 
import { AppService } from './app.service';
 
@Controller()
export class AppController {
  constructor(private readonly appService: AppService) {}
 
  @MessagePattern('get_user')
  handleGetUser(user: CreateUserDto) {
    return this.appService.getUser(user.username);
  }
 
  @MessagePattern('create_user')
  handleCreateUser(newUser: CreateUserDto) {
    return this.appService.createUser(newUser);
  }
}

// apps/api-gateway/src/auth/auth.service.ts
import { Inject, Injectable } from '@nestjs/common';
import { ClientProxy } from '@nestjs/microservices';
import { Observable } from 'rxjs';
 
import { CreateUserDto, User } from '@nestjs-microservices/shared';
 
@Injectable()
export class AuthService {
  constructor(
    @Inject('AUTH_MICROSERVICE') private readonly authClient: ClientProxy
  ) {}
 
  getUser(createUserDto: CreateUserDto): Observable<User> {
    return this.authClient.send<User, CreateUserDto>('get_user', createUserDto);
  }
 
  createUser(createUserDto: CreateUserDto): Observable<User> {
    return this.authClient.send<User, CreateUserDto>('create_user', createUserDto);
  }
}

// apps/api-gateway/src/auth/auth.controller.ts
import { Body, Controller, Post, BadRequestException } from '@nestjs/common';
import { lastValueFrom } from 'rxjs';
 
import { CreateUserDto, User } from '@nestjs-microservices/shared';
 
import { AuthService } from './auth.service';
 
@Controller('auth')
export class AuthController {
  constructor(private readonly authService: AuthService) {}
 
  @Post('login')
  async login(@Body() createUserDto: CreateUserDto) {
    const user: User = await lastValueFrom(
      this.authService.getUser(createUserDto),
      { defaultValue: undefined }
    );
    
    if (!user) {
      throw new BadRequestException('Неверные учетные данные');
    }
 
    const isMatch = user.password === createUserDto.password;
    if (!isMatch) {
      throw new BadRequestException('Неверный пароль');
    }
 
    console.log(`Пользователь ${user.username} успешно вошел в систему.`);
 
    return user;
  }
 
  @Post('signup')
  async signup(@Body() createUserDto: CreateUserDto) {
    const user: User = await lastValueFrom(
      this.authService.getUser(createUserDto),
      { defaultValue: undefined }
    );
    
    if (user) {
      throw new BadRequestException(
        [INLINE]Пользователь с именем ${createUserDto.username} уже существует![/INLINE]
      );
    }
 
    return this.authService.createUser(createUserDto);
  }
}

// Отправка события
this.authClient.emit<void, UserCreatedEvent>('user_created', { 
  userId: user.id, 
  username: user.username 
});
 
// Обработка события в другом сервисе
@EventPattern('user_created')
handleUserCreated(data: UserCreatedEvent) {
  // Обработка события
}

ClientsModule.register([
  {
    name: 'AUTH_MICROSERVICE',
    transport: Transport.TCP,
    options: {
      host: 'localhost',
      port: 3001,
      retryAttempts: 5, // Количество попыток переподключения
      retryDelay: 1000, // Задержка между попытками (мс)
    },
  },
])

// shared/src/lib/interfaces/message-patterns.interface.ts
export interface AuthMessagePatterns {
  'get_user': { username: string; password: string };
  'create_user': { username: string; password: string };
  'validate_token': { token: string };
}

import { catchError, timeout, retry } from 'rxjs/operators';
 
this.authService.getUser(createUserDto).pipe(
  retry(3), // Повторить запрос до 3 раз в случае ошибки
  timeout(5000), // Установить таймаут в 5 секунд
  catchError(error => {
    console.error('Ошибка при обращении к микросервису:', error);
    // Вернуть запасной вариант или перебросить ошибку
    throw new ServiceUnavailableException('Сервис аутентификации временно недоступен');
  })
)

@Catch()
export class AllExceptionsFilter implements RpcExceptionFilter {
  catch(exception: any, host: ArgumentsHost) {
    const ctx = host.switchToRpc();
    const data = ctx.getData();
    
    // Логируем ошибку вместе с контекстом
    console.error(`Ошибка при обработке ${JSON.stringify(data)}:`, exception);
    
    // Трансформируем ошибку в понятный формат
    return throwError(() => ({
      status: 'error',
      message: 'Произошла внутренняя ошибка',
      code: 'INTERNAL_ERROR',
      timestamp: new Date().toISOString(),
    }));
  }
}

@Post('login')
async login(@Body() createUserDto: CreateUserDto) {
  try {
    const user = await lastValueFrom(
      this.authService.getUser(createUserDto).pipe(
        timeout(5000),
        catchError(err => {
          if (err instanceof TimeoutError) {
            throw new ServiceUnavailableException('Сервис аутентификации не отвечает');
          }
          // Обрабатываем ошибки от микросервисов
          if (err?.status === 'error') {
            throw new InternalServerErrorException(err.message);
          }
          throw err;
        })
      )
    );
    
    // Остальной код...
  } catch (error) {
    // Обработка других ошибок
    throw error;
  }
}

// logger.service.ts
@Injectable()
export class LoggerService implements LoggerService {
  private context?: string;
  private correlationId?: string;
  
  setContext(context: string) {
    this.context = context;
    return this;
  }
  
  setCorrelationId(id: string) {
    this.correlationId = id;
    return this;
  }
  
  log(message: string, ...args: any[]) {
    this.writeLog('log', message, ...args);
  }
  
  error(message: string, trace?: string, ...args: any[]) {
    this.writeLog('error', message, ...(trace ? [trace] : []), ...args);
  }
  
  private writeLog(level: string, message: string, ...args: any[]) {
    const logEntry = {
      level,
      message,
      context: this.context,
      correlationId: this.correlationId,
      timestamp: new Date().toISOString(),
      // Добавляем метаданные, полезные для диагностики
      hostname: os.hostname(),
      pid: process.pid,
      data: args.length ? args : undefined
    };
    
    console.log(JSON.stringify(logEntry));
    // В реальном приложении отправляем логи в центральное хранилище
  }
}

@Injectable()
export class CorrelationIdInterceptor implements NestInterceptor {
  intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
    let correlationId: string;
    
    if (context.getType() === 'http') {
      const request = context.switchToHttp().getRequest();
      correlationId = request.headers['x-correlation-id'] || uuidv4();
      // Сохраняем ID в запросе для дальнейшего использования
      request.correlationId = correlationId;
    } else if (context.getType() === 'rpc') {
      const rpcContext = context.switchToRpc();
      const data = rpcContext.getData();
      correlationId = data.correlationId || uuidv4();
    }
    
    // При отправке запросов в другие микросервисы добавляйте correlationId
    // в метаданные или тело запроса
    
    return next.handle().pipe(
      tap(data => {
        // Можно добавить correlationId в ответ
        if (context.getType() === 'http') {
          const response = context.switchToHttp().getResponse();
          response.header('x-correlation-id', correlationId);
        }
      })
    );
  }
}

@Module({
  imports: [
    ClientsModule.registerAsync([{
      name: 'AUTH_SERVICE',
      useFactory: (configService: ConfigService) => ({
        transport: Transport.TCP,
        options: {
          host: configService.get('AUTH_SERVICE_HOST'),
          port: configService.get('AUTH_SERVICE_PORT'),
        },
      }),
      inject: [ConfigService],
    }]),
  ],
})
export class AuthModule {}

describe('AuthService', () => {
  let service: AuthService;
  let clientProxyMock: any;
 
  beforeEach(async () => {
    clientProxyMock = {
      send: jest.fn().mockImplementation(() => {
        return of({ id: 1, username: 'test' });
      }),
    };
 
    const module: TestingModule = await Test.createTestingModule({
      providers: [
        AuthService,
        {
          provide: 'AUTH_MICROSERVICE',
          useValue: clientProxyMock,
        },
      ],
    }).compile();
 
    service = module.get<AuthService>(AuthService);
  });
 
  it('должен получать пользователя по имени', async () => {
    const result = await lastValueFrom(service.getUser({ username: 'test', password: 'pass' }));
    expect(result).toEqual({ id: 1, username: 'test' });
    expect(clientProxyMock.send).toHaveBeenCalledWith('get_user', { username: 'test', password: 'pass' });
  });
});

describe('Интеграционные тесты аутентификации', () => {
  let app: INestApplication;
  let authMicroservice: INestMicroservice;
 
  beforeAll(async () => {
    // Запускаем микросервис аутентификации
    const authModule = await NestFactory.createMicroservice<MicroserviceOptions>(
      AuthModule,
      {
        transport: Transport.TCP,
        options: {
          host: 'localhost',
          port: 3001,
        },
      },
    );
    await authMicroservice.listen();
 
    // Запускаем API Gateway
    const moduleFixture = await Test.createTestingModule({
      imports: [AppModule],
    }).compile();
 
    app = moduleFixture.createNestApplication();
    await app.init();
  });
 
  afterAll(async () => {
    await app.close();
    await authMicroservice.close();
  });
 
  it('должен регистрировать нового пользователя', async () => {
    const response = await request(app.getHttpServer())
      .post('/auth/signup')
      .send({ username: 'testuser', password: 'password' })
      .expect(201);
 
    expect(response.body).toHaveProperty('id');
    expect(response.body.username).toBe('testuser');
  });
});

@Injectable()
export class OrderSagaService {
  constructor(
    private readonly productClient: ClientProxy,
    private readonly paymentClient: ClientProxy,
    private readonly shippingClient: ClientProxy,
    private readonly eventBus: EventBus,
  ) {}
 
  async createOrder(orderData: CreateOrderDto): Promise<Order> {
    // Шаг 1: Создаем заказ
    const order = await this.saveOrder(orderData);
    
    try {
      // Шаг 2: Резервируем товары
      await lastValueFrom(this.productClient.send('reserve_products', { 
        orderId: order.id, 
        items: orderData.items 
      }));
      
      // Шаг 3: Обрабатываем платеж
      await lastValueFrom(this.paymentClient.send('process_payment', { 
        orderId: order.id, 
        amount: orderData.totalAmount,
        paymentDetails: orderData.paymentDetails 
      }));
      
      // Шаг 4: Организуем доставку
      await lastValueFrom(this.shippingClient.send('arrange_shipping', { 
        orderId: order.id, 
        address: orderData.shippingAddress 
      }));
      
      // Отмечаем заказ как успешно созданный
      await this.finalizeOrder(order.id);
      
      return order;
    } catch (error) {
      // В случае ошибки запускаем компенсирующие действия
      this.eventBus.publish(new OrderFailedEvent(order.id, error));
      throw new InternalServerErrorException('Не удалось создать заказ');
    }
  }
  
  // Компенсирующие действия в случае сбоя
  @EventPattern('order.failed')
  async handleOrderFailed(data: { orderId: string, step: string }) {
    switch(data.step) {
      case 'payment':
        // Освобождаем зарезервированные товары
        await lastValueFrom(this.productClient.send('release_products', { orderId: data.orderId }));
        break;
      case 'shipping':
        // Отменяем платеж и освобождаем товары
        await lastValueFrom(this.paymentClient.send('refund_payment', { orderId: data.orderId }));
        await lastValueFrom(this.productClient.send('release_products', { orderId: data.orderId }));
        break;
    }
    
    // Отмечаем заказ как неудачный
    await this.cancelOrder(data.orderId);
  }
}