guanghulab/backend/api-server/services/atomic-executor.js

/**
 * 原子执行器 · L3 执行原子性保障
 *
 * 包装所有写入操作，确保：
 * 1. 执行前保存检查点
 * 2. 执行中记录每一步
 * 3. 执行失败自动回滚
 * 4. 执行完成释放锁
 *
 * 使用方式：
 * var executor = new AtomicExecutor(devId);
 * var result = await executor.execute('创建工单', async function(ctx) {
 *   await ctx.step('写入Notion', async function() { ... });
 *   await ctx.step('触发Workflow', async function() { ... });
 * });
 *
 * 版权：国作登字-2026-A-00037559
 */

'use strict';

var executionLock = require('../middleware/execution-lock');
var checkpoint = require('./checkpoint');
var CheckpointManager = checkpoint.CheckpointManager;

// 锁释放延迟（给前端时间拉取最终状态）
var LOCK_RELEASE_DELAY_MS = 10000;

/**
 * 原子执行器
 * @param {string} devId - 开发者编号
 */
function AtomicExecutor(devId) {
  this.devId = devId;
}

/**
 * 执行操作（原子性保障）
 * @param {string} operationName - 操作描述
 * @param {Function} fn - 异步执行函数，接收 ctx 参数
 * @returns {Object} 执行结果
 */
AtomicExecutor.prototype.execute = async function(operationName, fn) {
  // 1. 获取执行锁
  var lockResult = executionLock.acquireLock(this.devId, operationName);
  if (!lockResult.success) {
    return lockResult;
  }

  var executionId = lockResult.executionId;
  var lock = lockResult.lock;
  var devId = this.devId;
  var cpManager = new CheckpointManager(devId, executionId);

  // 2. 保存初始检查点
  await cpManager.save('执行前状态', {
    type: 'initial',
    timestamp: new Date().toISOString()
  });

  // 3. 构建执行上下文
  var ctx = {
    executionId: executionId,

    /**
     * 执行单步操作
     * @param {string} stepName - 步骤名称
     * @param {Function} stepFn - 步骤执行函数
     * @returns {*} 步骤执行结果
     */
    step: async function(stepName, stepFn) {
      executionLock.logStep(devId, {
        message: '⏳ ' + stepName + '...',
        success: false
      });

      try {
        var result = await stepFn();

        executionLock.logStep(devId, {
          message: '✅ ' + stepName,
          success: true
        });

        return result;
      } catch (e) {
        executionLock.logStep(devId, {
          message: '❌ ' + stepName + ': ' + e.message,
          success: false
        });
        throw e;
      }
    },

    /**
     * 保存检查点
     * @param {string} label - 检查点标签
     * @param {Object} data - 快照数据
     * @returns {string} 检查点ID
     */
    checkpoint: async function(label, data) {
      return await cpManager.save(label, data);
    }
  };

  // 4. 执行主逻辑
  try {
    var result = await fn(ctx);

    // 5. 成功：更新锁状态
    lock.state = 'completed';
    lock.result = result;

    // 延迟释放锁（给前端时间拉取最终状态）
    setTimeout(function() { executionLock.releaseLock(devId); }, LOCK_RELEASE_DELAY_MS);

    return {
      success: true,
      executionId: executionId,
      result: result,
      reply: result && result.reply ? result.reply : '✅ 操作「' + operationName + '」执行完成。'
    };

  } catch (e) {
    // 6. 失败：自动回滚
    lock.state = 'rolling_back';
    executionLock.logStep(devId, {
      message: '🔄 执行失败，正在自动回滚...',
      success: false
    });

    try {
      await cpManager.rollbackTo('cp-1');
      executionLock.logStep(devId, {
        message: '✅ 已回滚到执行前状态',
        success: true
      });
    } catch (rollbackError) {
      executionLock.logStep(devId, {
        message: '⚠️ 回滚异常：' + rollbackError.message + '，已通知管理员',
        success: false
      });
    }

    lock.state = 'failed';
    lock.error = { message: e.message };

    setTimeout(function() { executionLock.releaseLock(devId); }, LOCK_RELEASE_DELAY_MS);

    return {
      success: false,
      executionId: executionId,
      error: e.message,
      reply: '❌ 操作「' + operationName + '」执行失败：' + e.message +
             '\n\n系统已自动回滚到执行前状态，没有数据被修改。'
    };
  }
};

module.exports = { AtomicExecutor: AtomicExecutor };