Redis 触发器和函数入门

#!js name=KeySpaceTriggers api_version=1.0
redis.registerKeySpaceTrigger(
  'updateProductStockQty',
  'orders:orderId:', // Keys starting with this prefix are monitored
  function (client, data) {
    const errors = [];

    try {
      if (
        client &&
        data?.event == 'json.set' &&
        data?.key != 'orders:orderId:index:hash'
      ) {
        const orderId = data.key;
        // get the order details from the orderId key
        let result = client.call('JSON.GET', orderId);
        result = result ? JSON.parse(result) : '';
        const order = Array.isArray(result) ? result[0] : result;

        if (order?.products?.length && !order.triggerProcessed) {
          try {
            //create a log stream to log the trigger events and errors
            client.call(
              'XGROUP',
              'CREATE',
              'TRIGGER_LOGS_STREAM',
              'TRIGGER_LOGS_GROUP',
              '$',
              'MKSTREAM',
            );
          } catch (streamConErr) {
            // if log stream already exists
          }

          // reduce stockQty for each product in the order
          for (const product of order.products) {
            let decreaseQtyBy = (-1 * product.qty).toString();
            client.call(
              'JSON.NUMINCRBY',
              `products:productId:${product.productId}`,
              '.stockQty',
              decreaseQtyBy,
            );

            // add log entry
            client.call(
              'XADD',
              'TRIGGER_LOGS_STREAM',
              '*',
              'message',
              `For productId ${product.productId}, stockQty ${decreaseQtyBy}`,
              'orderId',
              orderId,
              'function',
              'updateProductStockQty',
            );
          }

          // set triggerProcessed flag to avoid duplicate processing
          client.call('JSON.SET', orderId, '.triggerProcessed', '1');
        }
      }
    } catch (generalErr) {
      generalErr = JSON.stringify(
        generalErr,
        Object.getOwnPropertyNames(generalErr),
      );
      errors.push(generalErr);
    }

    if (errors.length) {
      //log error
      client.call(
        'XADD',
        'TRIGGER_LOGS_STREAM',
        '*',
        'message',
        JSON.stringify(errors),
        'orderId',
        data.key,
        'function',
        'updateProductStockQty',
      );
    }
  },
);