// src/orders/orders.module.ts
import { Module } from "@nestjs/common";
import { ClientsModule, Transport } from "@nestjs/microservices";

@Module({
  imports: [
    ClientsModule.register([
      {
        name: "NOTIFICATION_SERVICE",
        transport: Transport.RMQ,
        options: {
          urls: ["amqp://localhost:5672"],
          queue: "notifications_queue",
          queueOptions: { durable: true },
        },
      },
    ]),
  ],
})
export class OrdersModule {}

// src/orders/orders.service.ts
import { Inject, Injectable } from "@nestjs/common";
import { ClientProxy } from "@nestjs/microservices";

@Injectable()
export class OrdersService {
  constructor(
    @Inject("NOTIFICATION_SERVICE")
    private readonly notificationClient: ClientProxy
  ) {}

  async createOrder(dto: CreateOrderDto): Promise<Order> {
    // 1. 주문 생성
    const order = await this.orderRepository.save({
      userId: dto.userId,
      items: dto.items,
      status: "PENDING",
    });

    // 2. 메시지 발행 (비동기)
    this.notificationClient.emit("order_created", {
      orderId: order.id,
      userId: order.userId,
      totalAmount: order.totalAmount,
    });

    // 3. 즉시 응답 (알림은 나중에 처리됨)
    return order;
  }

  async cancelOrder(orderId: string): Promise<void> {
    const order = await this.orderRepository.findById(orderId);
    order.status = "CANCELLED";
    await this.orderRepository.save(order);

    // 취소 이벤트 발행
    this.notificationClient.emit("order_cancelled", {
      orderId: order.id,
      userId: order.userId,
    });
  }
}

// src/notifications/notifications.controller.ts
import { Controller } from "@nestjs/common";
import { EventPattern, Payload } from "@nestjs/microservices";

@Controller()
export class NotificationsController {
  constructor(
    private readonly emailService: EmailService,
    private readonly pushService: PushService
  ) {}

  @EventPattern("order_created")
  async handleOrderCreated(
    @Payload() data: { orderId: string; userId: string; totalAmount: number }
  ): Promise<void> {
    console.log("주문 생성 이벤트 수신:", data);

    // 이메일 발송
    await this.emailService.sendOrderConfirmation(
      data.userId,
      data.orderId,
      data.totalAmount
    );

    // 푸시 알림
    await this.pushService.sendNotification(
      data.userId,
      `주문이 완료되었습니다. 주문번호: ${data.orderId}`
    );
  }

  @EventPattern("order_cancelled")
  async handleOrderCancelled(
    @Payload() data: { orderId: string; userId: string }
  ): Promise<void> {
    console.log("주문 취소 이벤트 수신:", data);

    await this.emailService.sendOrderCancellation(data.userId, data.orderId);
  }
}

// main.ts (Consumer 서비스)
import { NestFactory } from "@nestjs/core";
import { Transport, MicroserviceOptions } from "@nestjs/microservices";

async function bootstrap() {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(
    NotificationsModule,
    {
      transport: Transport.RMQ,
      options: {
        urls: ["amqp://localhost:5672"],
        queue: "notifications_queue",
        queueOptions: { durable: true },
        noAck: false, // 수동 ACK
      },
    }
  );

  await app.listen();
  console.log("Notification 서비스 시작");
}
bootstrap();

// src/orders/orders.module.ts
import { Module } from "@nestjs/common";
import { ClientsModule, Transport } from "@nestjs/microservices";

@Module({
  imports: [
    ClientsModule.register([
      {
        name: "KAFKA_SERVICE",
        transport: Transport.KAFKA,
        options: {
          client: {
            clientId: "orders",
            brokers: ["localhost:9092"],
          },
          producer: {
            allowAutoTopicCreation: true,
          },
        },
      },
    ]),
  ],
})
export class OrdersModule {}

// src/orders/orders.service.ts
@Injectable()
export class OrdersService {
  constructor(
    @Inject("KAFKA_SERVICE")
    private readonly kafkaClient: ClientKafka
  ) {}

  async onModuleInit() {
    // 토픽 구독 (응답 받을 경우)
    this.kafkaClient.subscribeToResponseOf("order.created");
    await this.kafkaClient.connect();
  }

  async createOrder(dto: CreateOrderDto): Promise<Order> {
    const order = await this.orderRepository.save({
      userId: dto.userId,
      items: dto.items,
      status: "PENDING",
    });

    // Kafka 메시지 발행
    this.kafkaClient.emit("order.created", {
      key: order.id, // 파티션 키
      value: {
        orderId: order.id,
        userId: order.userId,
        items: order.items,
        totalAmount: order.totalAmount,
        createdAt: new Date().toISOString(),
      },
    });

    return order;
  }
}

// src/inventory/inventory.controller.ts
import { Controller } from "@nestjs/common";
import { MessagePattern, Payload } from "@nestjs/microservices";

@Controller()
export class InventoryController {
  constructor(private readonly inventoryService: InventoryService) {}

  @MessagePattern("order.created")
  async handleOrderCreated(
    @Payload()
    message: {
      orderId: string;
      items: Array<{ productId: string; quantity: number }>;
    }
  ): Promise<void> {
    console.log("재고 차감 이벤트 수신:", message.orderId);

    // 재고 차감
    for (const item of message.items) {
      await this.inventoryService.decreaseStock(item.productId, item.quantity);
    }
  }
}

// main.ts (Consumer 서비스)
async function bootstrap() {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(
    InventoryModule,
    {
      transport: Transport.KAFKA,
      options: {
        client: {
          brokers: ["localhost:9092"],
          clientId: "inventory-consumer",
        },
        consumer: {
          groupId: "inventory-group",
        },
      },
    }
  );

  await app.listen();
}

// Dead Letter Queue (DLQ) 패턴
@Injectable()
export class OrderEventHandler {
  private readonly MAX_RETRIES = 3;

  @EventPattern("order.created")
  async handleOrderCreated(@Payload() data: OrderCreatedEvent): Promise<void> {
    try {
      await this.processOrder(data);
    } catch (error) {
      // 재시도 횟수 확인
      const retryCount = data.retryCount || 0;

      if (retryCount < this.MAX_RETRIES) {
        // 재시도
        this.kafkaClient.emit("order.created", {
          ...data,
          retryCount: retryCount + 1,
        });
      } else {
        // DLQ로 이동
        this.kafkaClient.emit("order.created.dlq", {
          ...data,
          error: error.message,
          failedAt: new Date().toISOString(),
        });
      }
    }
  }
}

// 멱등성 보장
@Injectable()
export class IdempotentHandler {
  constructor(private readonly redis: Redis) {}

  async handleMessage(messageId: string, handler: () => Promise<void>) {
    // 이미 처리된 메시지인지 확인
    const processed = await this.redis.get(`processed:${messageId}`);
    if (processed) {
      console.log("이미 처리된 메시지:", messageId);
      return;
    }

    // 처리
    await handler();

    // 처리 완료 표시 (24시간 유지)
    await this.redis.set(`processed:${messageId}`, "1", "EX", 86400);
  }
}

// 이벤트 기반 주문 처리 흐름
// 1. 주문 생성 → order.created 발행
// 2. 재고 서비스 → 재고 차감 → inventory.updated 발행
// 3. 결제 서비스 → 결제 처리 → payment.completed 발행
// 4. 알림 서비스 → 이메일/푸시 발송
// 5. 분석 서비스 → 데이터 저장