// app.module.ts
import { Module } from '@nestjs/common';
import { TypeOrmModule } from '@nestjs/typeorm';
import { ConfigModule, ConfigService } from '@nestjs/config';
 
@Module({
  imports: [
    TypeOrmModule.forRootAsync({
      imports: [ConfigModule],
      useFactory: (config: ConfigService) => ({
        type: 'mysql',
        host: config.get('DB_HOST'),
        port: config.get('DB_PORT'),
        username: config.get('DB_USER'),
        password: config.get('DB_PASS'),
        database: config.get('DB_NAME'),
        entities: [__dirname + '//*.entity{.ts,.js}'],
        synchronize: false,
        logging: ['error'], // 避免日志影響性能
        extra: {
          connectionLimit: 100,
          queueLimit: 10,
          connectTimeout: 10000,
          acquireTimeout: 15000,
          timeout: 10000,
        },
        replication: {
          master: {
            host: config.get('DB_MASTER_HOST'),
            port: config.get('DB_MASTER_PORT'),
            username: config.get('DB_MASTER_USER'),
            password: config.get('DB_MASTER_PASS'),
          },
          slaves: [
            { host: 'slave1.example.com', port: 3306 },
            { host: 'slave2.example.com', port: 3306 },
            // ... 其他 slave
          ],
        },
      }),
      inject: [ConfigService],
    }),
  ],
})
export class AppModule {}

// redis-lock.service.ts
import { Injectable } from '@nestjs/common';
import { Redis } from 'ioredis';
 
@Injectable()
export class RedisLockService {
  private readonly LOCK_SCRIPT = `
    return redis.call('set', KEYS[1], ARGV[1], 'NX', 'PX', ARGV[2])
  `;
  private readonly UNLOCK_SCRIPT = `
    if redis.call("get",KEYS[1]) == ARGV[1] then
        return redis.call("del",KEYS[1])
    else
        return 0 
    end
  `;
 
  constructor(private readonly redis: Redis) {}
 
  async acquireLock(
    key: string,
    requestId: string,
    expireMs: number,
  ): Promise<boolean> {
    const result = await this.redis.eval(
      this.LOCK_SCRIPT,
      1,
      key,
      requestId,
      expireMs,
    );
    return result === 'OK';
  }
 
  async releaseLock(key: string, requestId: string): Promise<boolean> {
    const result = await this.redis.eval(this.UNLOCK_SCRIPT, 1, key, requestId);
    return result === 1;
  }
}

ALTER TABLE product ADD COLUMN version INT DEFAULT 0;
UPDATE product 
SET stock = stock - 1, version = version + 1 
WHERE id = ? AND version = ?

// product.service.ts
@Injectable()
export class ProductService {
  @InjectRepository(Product)
  private repo: Repository<Product>;
 
  async decreaseStock(productId: number, expectedVersion: number): Promise<boolean> {
    const result = await this.repo
      .createQueryBuilder()
      .update(Product)
      .set({ stock: () => 'stock - 1', version: () => 'version + 1' })
      .where('id = :id AND version = :version', { id: productId, version: expectedVersion })
      .execute();
 
    return result.affected > 0;
  }
}

// order.processor.ts
@Processor('order_queue')
export class OrderProcessor {
  @Process('create_order')
  async handleOrderCreation(job: Job) {
    const { orderId, userId, items } = job.data;
    try {
      await this.orderService.createOrder(orderId, userId, items);
      return { success: true };
    } catch (err) {
      await job.moveToFailed({ message: err.message }, true);
      throw err;
    }
  }
}

await this.queue.add('create_order', { orderId, userId, items }, {
  attempts: 3,
  backoff: { type: 'exponential', delay: 1000 },
});

// inventory.service.ts
@Injectable()
export class InventoryService {
  constructor(@InjectRepository(Product) private productRepo: Repository<Product>) {}
 
  async deductStockOptimistic(productId: string, quantity: number): Promise<boolean> {
    const queryRunner = this.productRepo.manager.connection.createQueryRunner();
    await queryRunner.startTransaction();
 
    try {
      const product = await queryRunner.manager.findOne(Product, {
        where: { id: productId },
        lock: { mode: 'pessimistic_read' }, // 可選：加悲觀讀鎖防止臟讀
      });
 
      if (!product || product.stock < quantity) {
        throw new Error('Insufficient stock');
      }
 
      const affected = await queryRunner.manager.update(
        Product,
        { id: productId, stock: MoreThanOrEqual(quantity), version: product.version },
        { stock: () => `stock - ${quantity}`, version: () => `version + 1` }
      );
 
      if (affected.affected === 0) {
        throw new Error('Concurrent update detected');
      }
 
      await queryRunner.commitTransaction();
      return true;
    } catch (err) {
      await queryRunner.rollbackTransaction();
      throw err;
    } finally {
      await queryRunner.release();
    }
  }
}

// redis.inventory.service.ts
@Injectable()
export class RedisInventoryService {
  private readonly STOCK_KEY = (id: string) => `inventory:${id}`;
 
  constructor(private readonly redisService: RedisService) {}
 
  async deductWithPessimisticLock(
    productId: string,
    quantity: number,
    ttlMs = 5000
  ): Promise<boolean> {
    const lockKey = `lock:inventory:${productId}`;
    const client = this.redisService.getClient();
 
    // 獲取分布式鎖（SETNX + EXPIRE）
    const acquired = await client.set(lockKey, '1', 'PX', ttlMs, 'NX');
    if (!acquired) {
      throw new Error('Failed to acquire inventory lock');
    }
 
    try {
      const stockScript = `
        local current = tonumber(redis.call('GET', KEYS[1]))
        if not current or current < tonumber(ARGV[1]) then 
          return 0
        end
        redis.call('INCRBYFLOAT', KEYS[1], -ARGV[1])
        return 1
      `;
 
      const result = await client.eval(
        stockScript,
        1,
        this.STOCK_KEY(productId),
        quantity
      );
 
      return result === 1;
    } finally {
      // 釋放鎖
      await client.del(lockKey);
    }
  }
}

// order.processor.ts
@Processor('order_queue')
export class OrderProcessor {
  constructor(
    private readonly inventoryService: InventoryService,
    private readonly eventPublisher: EventPublisher
  ) {}
 
  @Process('create_order')
  async handleOrderCreation(job: Job<OrderPayload>): Promise<void> {
    const { orderId, productId, quantity } = job.data;
 
    try {
      const success = await this.inventoryService.deductStockOptimistic(productId, quantity);
      if (success) {
        await this.eventPublisher.emitAsync(new OrderCreatedEvent(orderId));
      } else {
        await this.eventPublisher.emitAsync(new OrderFailedEvent(orderId, 'Out of stock'));
      }
    } catch (error) {
      // 失敗重試機制
      if (job.attemptsMade < 3) {
        throw error; // 觸發(fā)重試
      } else {
        await this.eventPublisher.emitAsync(new OrderFailedEvent(orderId, error.message));
      }
    }
  }
}
 
// bootstrap with BullMQ 
const queue = new Queue('order_queue', { connection });
const worker = new Worker('order_queue', new OrderProcessor(...).handleOrderCreation, { connection });

// redis-lock.service.ts
import { Injectable } from '@nestjs/common';
import Redis from 'ioredis';
 
@Injectable()
export class RedisLockService {
  private readonly redis = new Redis({
    host: 'redis-cluster-host',
    port: 6379,
  });
 
  private readonly LOCK_PREFIX = 'lock:';
  private readonly DEFAULT_TTL = 5000; // 5s
 
  async acquireLock(key: string, ttlMs: number = this.DEFAULT_TTL): Promise<string | null> {
    const lockKey = this.LOCK_PREFIX + key;
    const token = Date.now().toString();
 
    const script = `
      if redis.call('GET', KEYS[1]) == false then
        return redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[2])
      else 
        return false 
      end
    `;
 
    const result = await this.redis.eval(script, 1, lockKey, token, ttlMs);
    return result ? token : null;
  }
 
  async releaseLock(key: string, token: string): Promise<boolean> {
    const lockKey = this.LOCK_PREFIX + key;
    const script = `
      if redis.call('GET', KEYS[1]) == ARGV[1] then
        return redis.call('DEL', KEYS[1])
      else 
        return 0
      end
    `;
    const result = await this.redis.eval(script, 1, lockKey, token);
    return result === 1;
  }
}

// inventory.service.ts
import { Injectable } from '@nestjs/common';
import { RedisLockService } from './redis-lock.service';
 
@Injectable()
export class InventoryService {
  constructor(private readonly lockService: RedisLockService) {}
 
  async deductStock(productId: number, quantity: number): Promise<boolean> {
    const lockKey = `stock:${productId}`;
    let token: string | null = null;
 
    try {
      token = await this.lockService.acquireLock(lockKey, 3000);
      if (!token) {
        throw new Error('Failed to acquire lock');
      }
 
      // 模擬數(shù)據(jù)庫檢查與扣減（應(yīng)在事務(wù)中）
      const current = await this.getCurrentStock(productId);
      if (current < quantity) {
        throw new Error('Insufficient stock');
      }
 
      await this.updateStockInDatabase(productId, current - quantity);
      return true;
 
    } catch (err) {
      console.error(err);
      return false;
    } finally {
      if (token) {
        await this.lockService.releaseLock(lockKey, token);
      }
    }
  }
 
  private async getCurrentStock(productId: number): Promise<number> {
    // 實際調(diào)用 TypeORM / Prisma 查詢
    return 100; // mock
  }
 
  private async updateStockInDatabase(productId: number, newStock: number): Promise<void> {
    // 更新數(shù)據(jù)庫邏輯 
    console.log(`Updating stock for ${productId} to ${newStock}`);
  }
}

// order-producer.service.ts
import { Injectable } from '@nestjs/common';
import { ClientProxy } from '@nestjs/microservices';
 
@Injectable()
export class OrderProducerService {
  constructor(private client: ClientProxy) {}
 
  emitOrderCreated(payload: { userId: number; productId: number; count: number }) {
    this.client.emit('order_created', payload);
  }
}
 
// order-consumer.service.ts
@EventPattern('order_created')
async handleOrderCreated try {
    await this.inventoryService.deductStock(data.productId, data.count);
    await this.orderRepository.save(data);
    console.log('Order processed:', data);
  } catch (err) {
    // 發(fā)送失敗則進入死信隊列或重試機制 
    console.error('Order processing failed:', err);
    throw err; // 觸發(fā)重試
  }
}