新零售实战 | 新零售O2O服务架构设计——从扫码购到即时配送的全链路实战

/**
 * 尝试创建订单并预占相关资源(Try阶段)
 * 
 * 1. 锁定订单商品对应的门店库存，防止超卖
 * 2. 冻结订单使用的优惠券，避免重复使用
 * 3. 创建状态为PENDING的订单记录
 * 
 * @param {Object} order - 订单对象，需包含商品、优惠券等订单信息
 * @returns {Promise<Object>} 返回新创建的订单记录Promise对象
 */
// Try阶段：资源预占
tryCreateOrder = async function (order) {
  // 执行库存锁定操作，预占商品库存资源
  await inventoryService.lock(order.items);
  
  // 执行优惠券冻结操作，标记优惠券为已使用状态
  await couponService.freeze(order.coupons);
  
  // 创建待处理状态的订单记录（此时订单尚未最终确认）
  return OrderModel.create({ ...order, status: 'PENDING' });
};

/**
 * 锁定库存项，从门店库存和中心仓库动态分配可用库存
 * 注：当前为示例代码，库存锁定逻辑待实现
 * 
 * @param {Array<Object>} items - 需要锁定的商品项数组，每个项需包含以下属性：
 *              - storeId {string} 门店ID，用于查询门店库存
 * @returns {Promise<Object>} 返回库存锁定结果（具体返回结构待实现）
 *                            可能包含成功锁定的库存数量及来源信息
 */
// 库存服务内部逻辑示例
async function lock(items) {
  // 并行获取门店级库存和中心仓库库存
  const storeStock = await getStoreInventory(item.storeId);
  const centerStock = await getCenterInventory();
  
  // 动态分配库存来源
}

/**
 * 条形码解码服务类，用于从图像缓冲区中解码条形码信息
 */
class BarcodeService {
  /**
   * 对输入的图像缓冲区进行条形码解码
   * @param {Buffer} imageBuffer - 包含图像数据的缓冲区对象，支持常见图片格式如JPEG/PNG
   * @returns {Promise<Array<{code: string, confidence: number}>>} - 返回包含解码结果的数组，
   * 每个元素包含识别到的条形码(code)及其置信度(confidence)
   */
  async decode(imageBuffer) {
    // 将图像缓冲区转换为TensorFlow张量，支持自动识别图像格式
    const tensor = tf.node.decodeImage(imageBuffer);

    // 执行模型推理获取检测结果，包含边界框和类别信息
    // 注意：此处假设model已预先加载且返回结构包含boxes/classes/scores
    const { boxes, classes } = await model.executeAsync(tensor);

    // 将检测结果映射为结构化数据
    // 注意：原代码中scores变量未定义，此处保留原始逻辑但可能存在运行时错误
    return boxes.map((box, i) => ({
      code: CLASS_NAMES[classes[i]],  // 从预设常量转换类别编号为可读文本
      confidence: scores[i],         // 置信度数值应来源于模型输出的scores字段
    }));
  }
}

class PaymentService {
  /**
   * 处理订单超时逻辑
   * 
   * 当订单超时时：
   * 1. 检查订单状态是否为未支付
   * 2. 若未支付则释放库存并取消订单
   * 
   * @param {string|number} orderId - 需要处理的订单ID
   * @returns {Promise<void>} 无直接返回值，但会修改订单状态和库存数据
   */
  async handleTimeout(orderId) {
    // 获取订单数据
    const order = await Order.findById(orderId);

    // 仅处理未支付状态的订单
    if (order.status === 'UNPAID') {
      // 并行执行库存释放和订单状态更新
      await Promise.all([
        // 释放被占用的库存
        Inventory.release(order.sku, order.quantity),
        // 将订单标记为已取消
        order.update({ status: 'CANCELED' })
      ]);
    }
  }
}

/**
 * 确认订单并完成后续业务操作
 * @param {string} orderId - 订单唯一标识符
 * @returns {Promise<void>} 异步操作Promise
 */
// Confirm阶段：最终提交
confirmOrder = async function (orderId) {
  // 获取订单实体
  const order = await OrderModel.findById(orderId);

  // 确认支付交易
  await paymentService.commit(order.paymentId);
  
  // 扣减商品库存
  await inventoryService.deduct(order.items);
  
  // 核销使用的优惠券
  await couponService.consume(order.coupons);

  // 触发配送调度
  if (order.deliveryType === 'INSTANT') {
    // 即时订单需要调度骑手
    await dispatchRider(order);
  }
};

/**
 * 选择最优的配送供应商
 * 静态方法，根据订单的预计到达时间(ETA)和成本选择最合适的配送供应商
 * @param {Object} order - 需要配送的订单对象，包含配送相关信息
 * @returns {Promise<Object>} 返回最优供应商的结果对象，包含eta和cost属性
 */
class DeliveryRouter {
  static async selectVendor(order) {
    // 并行获取三个配送渠道的估算结果：达达、蜂鸟和自配送
    const candidates = await Promise.all([
      DadaAPI.estimate(order),
      FengniaoAPI.estimate(order),
      SelfDelivery.calculate(order)
    ]);

    // 按ETA优先（升序）、成本其次（升序）的规则排序，并选择最优项
    return candidates.sort((a, b) => a.eta - b.eta || a.cost - b.cost)[0];
  }
}

/**
 * WebSocket 服务端路由 - 实时骑手位置订阅
 * @param {WebSocket} ws - WebSocket 连接实例 
 * @param {Request} req - HTTP 请求对象，包含路径参数
 * @sideeffect
 * 1. 根据订单ID创建 Redis 扫描流
 * 2. 监听数据流实时推送 GeoJSON 格式位置数据
 * 3. 维护 WebSocket 连接状态
 */
// 骑手位置订阅服务
app.ws('/tracking/:orderId', (ws, req) => {
  // 从路径参数获取订单标识
  const orderId = req.params.orderId;

  // 创建 Redis 位置数据扫描流（匹配 pattern: pos:<orderId>:*）
  const stream = redis.scanStream({
    match: `pos:${orderId}:*`, // 匹配当前订单的所有位置键
  });

  // 流式数据事件处理器：将 Redis 数据转换为 GeoJSON 格式
  stream.on('data', positions => {
    // 构建符合 GeoJSON 标准的特征集合
    ws.send(
      JSON.stringify({
        type: 'FeatureCollection',
        // 转换坐标点为地理要素点
        features: positions.map(p => ({
          geometry: {
            type: 'Point',
            // 坐标顺序遵循 [经度, 纬度] 标准
            coordinates: [p.lng, p.lat],
          },
        })),
      }),
    );
  });
});

/**
 * 核销订单状态管理函数
 * 
 * @param {string} code - 订单唯一核销码，用于检索对应订单
 * @returns {Promise<Object>} 包含核销商品信息和核销时间的对象
 *   - items {Array} 订单包含的商品清单
 *   - verifyTime {Date} 核销完成时间
 * @throws {Error} 当订单状态不符合核销条件时抛出异常
 */
// 核销状态管理
async function verifyPickup(code) {
  // 通过核销码查询关联订单
  const order = await Order.findByCode(code);
  
  // 验证订单当前状态是否允许核销
  if (order.status !== 'PICKUP_PENDING') {
    throw new Error('非法核销状态');
  }

  // 校验订单取货时间是否在允许的时间窗口内
  await validateTimeWindow(order.pickupTime); // 校验预约时段
  // 执行库存扣减操作（展示库存管理）
  await markInventoryAsPicked(order.items); // 扣减展示库存

  return {
    items: order.items,
    verifyTime: new Date(),
  };
}

/**
 * 门店商品库存预约调度类
 * 使用分布式锁保证库存操作的原子性，支持高频次库存操作
 */
class PickupScheduler {
  constructor() {
    this.storeCache = new LRU({ max: 1000 }); // 缓存热门门店数据
    this.lock = new Redlock([redisClients]); // 分布式锁
  }

  /**
   * 执行商品库存预留操作
   * @param {string} storeId - 门店唯一标识符
   * @param {string} sku - 商品库存单位编码
   * @param {number} quantity - 需要预约的商品数量
   * @returns {Promise<Object>} 预约结果对象，包含状态码和预约码/错误信息
   * @desc 通过分布式锁保证单个门店库存操作的原子性，使用Redis哈希存储实时库存数据
   */
  async reserve(storeId, sku, quantity) {
    // 获取门店级别的分布式锁，5000毫秒自动释放防止死锁
    const lockKey = `store:${storeId}:lock`;
    const lock = await this.lock.acquire([lockKey], 5000);

    try {
      // 原子化库存操作区块：获取当前库存 -> 检查库存余量 -> 扣减库存
      const stock = await redis.hget(`store:${storeId}`, sku);
      
      if (stock >= quantity) {
        // 执行库存扣减并生成预约码（伪代码）
        await redis.hincrby(`store:${storeId}`, sku, -quantity);
        return { code: 200, data: generateReserveCode() };
      }

      // 库存不足时的错误返回
      return { code: 500, msg: '库存不足' };
    } finally {
      // 无论成功失败都释放分布式锁
      await lock.release();
    }
  }
}

/**
 * 核销服务类 - 处理核销码验证及相关业务操作
 */
class VerifyService {
  /**
   * 验证核销码并触发后续业务流程
   * @param {string} code - 需要验证的核销码字符串
   * @returns {Promise<boolean>} 验证成功返回true，失败抛出异常
   * @throws {Error} 当核销码失效时抛出异常
   */
  async verify(code) {
    // 构造Redis存储的键名（使用业务前缀避免键冲突）
    const key = `pickup:${code}`;
    
    /**
     * Redis事务操作：
     * 1. 原子化获取并删除核销码数据
     * 2. 使用事务保证操作原子性，防止并发问题
     */
    const session = await redis.multi().get(key).del(key).exec();

    // 检查核销码是否存在（事务结果数组结构：[ [err1, result1], [err2, result2] ]）
    if (!session[0][1]) throw new Error('核销码已失效');

    /**
     * 消息队列操作：
     * 1. 解析存储的会话数据
     * 2. 发送WMS出库指令到消息队列
     * 3. 使用独立消息通道保证系统间解耦
     */
    await mq.send('wms_outbound', JSON.parse(session[0][1]));
    return true;
  }
}

/**
 * 取消订单的补偿操作（Saga模式的Cancel阶段）
 * 
 * 该函数用于执行订单取消后的分布式事务回滚操作，依次进行：
 * 1. 支付回退
 * 2. 库存释放
 * 3. 优惠券恢复
 * 
 * @param {string} orderId - 需要取消的订单ID（MongoDB ObjectId格式）
 * @returns {Promise<void>} 无直接返回值，但包含异步操作的Promise链
 */
// Cancel阶段：补偿操作
cancelOrder = async function (orderId) {
  // 获取订单详细信息（包含支付、商品、优惠券等关联信息）
  const order = await OrderModel.findById(orderId);

  // 执行支付系统的金额回滚操作
  await paymentService.rollback(order.paymentId);

  // 释放订单中所有商品项占用的库存
  await inventoryService.release(order.items);

  // 恢复订单使用的优惠券额度/状态
  await couponService.restore(order.coupons);
};

/**
 * 库存同步类：负责管理多销售渠道的库存同步操作
 */
class InventorySync {
  constructor() {
    // 支持的库存同步渠道列表
    this.channels = ['store', 'warehouse', 'third_party'];
  }

  /**
   * 执行库存同步操作
   * @param {string} sku - 商品唯一标识符
   * @param {number} delta - 库存变动增量（可正可负）
   * @returns {Promise<void>} 异步操作Promise
   */
  async sync(sku, delta) {
    // 并行处理所有渠道的库存更新
    await Promise.all(
      this.channels.map(async channel => {
        // 生成Redis缓存键并执行原子增减操作
        const key = `stock:${channel}:${sku}`;
        await redis.incrby(key, delta);
        
        // 同步更新数据库中的库存记录
        await DB.updateStock(channel, sku, delta);
      }),
    );
  }
}