AWS Amplify多模型冲突处理：关联模型事务性解决方案问询

多数AWS Amplify/AppSync教程和示例仅覆盖单模型的自动冲突解决或自定义冲突处理，但面对跨模型的复杂关联约束（比如数据守恒、非负校验）且涉及离线同步场景时，常规的分步更新会导致数据不一致。

示例场景 #

我们以复活节彩蛋软心糖转移为例：孩子找到彩蛋后，将彩蛋中的所有软心糖转移到自己的篮子，要求软心糖总数守恒且数量不能为负，因此无法使用CRDT模型。

初始模型定义 #

type Child @model {
  id: ID!
  name: String!
  jellybeans: Int!
  version: Int! # 启用冲突检测的版本字段
}

type Egg @model {
  id: ID!
  jellybeans: Int!
  version: Int! # 启用冲突检测的版本字段
}

现有代码的问题 #

当前的转移逻辑是分步更新Child和Egg：

async function findEgg(childId, eggId) {
  const child = await DataStore.query(Child, childId);
  const egg = await DataStore.query(Egg, eggId);

  if (!child || !egg) return;

  const count = egg.jellybeans;

  // 先更新孩子的软心糖数量
  await DataStore.save(
    Child.copyOf(child, updated => {
      updated.jellybeans += count;
    })
  );

  // 再更新彩蛋的软心糖数量
  await DataStore.save(
    Egg.copyOf(egg, updated => {
      updated.jellybeans -= count;
    })
  );
}

当用户离线操作后同步时，会出现多种数据不一致问题：

• 若AppSync未检测到冲突，多个用户重复转移同一彩蛋，导致软心糖总数凭空增加，破坏守恒性；
• 冲突检测触发时，可能出现彩蛋软心糖数变为负数的情况；
• 分步更新中，前一个模型更新成功但后一个失败，导致数据部分生效，出现孤立错误。

解决方案：基于事务的跨模型原子操作 #

要实现要么全成功、要么全失败的原子操作，需要通过自定义GraphQL Mutation + AppSync Resolver事务来处理，同时引入转移记录作为事务的载体和审计依据。

1. 调整模型定义 #

新增JellybeanTransfer模型，用于记录转移操作的上下文，同时保留Child和Egg的版本字段用于冲突检测：

type Child @model {
  id: ID!
  name: String!
  jellybeans: Int!
  version: Int!
}

type Egg @model {
  id: ID!
  jellybeans: Int!
  version: Int!
}

# 转移记录模型：作为事务的载体，同时用于审计
type JellybeanTransfer @model {
  id: ID!
  childId: ID!
  eggId: ID!
  transferAmount: Int!
  timestamp: AWSDateTime!
  # 关联字段
  child: Child @connection(fields: ["childId"])
  egg: Egg @connection(fields: ["eggId"])
}

# 自定义Mutation：触发原子转移操作
type Mutation {
  transferJellybeans(childId: ID!, eggId: ID!): JellybeanTransfer
}

2. 实现事务性Resolver #

在AppSync中配置自定义Resolver，使用DynamoDB的TransactWriteItems API执行原子操作，同时完成三个动作：

• 检查彩蛋当前的软心糖数量是否大于0，且版本号与客户端查询时一致；
• 更新孩子的软心糖数量，同时递增版本号；
• 更新彩蛋的软心糖数量（清零），同时递增版本号；
• 创建转移记录。

Resolver的请求模板（VTL）核心逻辑如下：

#set($childKey = { "id": $ctx.args.childId })
#set($eggKey = { "id": $ctx.args.eggId })

# 获取当前彩蛋和孩子的状态（用于校验版本和数量）
#set($getEggRequest = {
  "operation": "GetItem",
  "key": $util.dynamodb.toMapValues($eggKey)
})
#set($getChildRequest = {
  "operation": "GetItem",
  "key": $util.dynamodb.toMapValues($childKey)
})

# 构造事务操作
#set($transactItems = [
  # 更新孩子的软心糖数量，校验版本号
  {
    "Update": {
      "TableName": "Child-${env}",
      "Key": $util.dynamodb.toMapValues($childKey),
      "UpdateExpression": "SET jellybeans = jellybeans + :amount, version = version + :inc",
      "ExpressionAttributeValues": {
        ":amount": { "N": "$ctx.result.egg.jellybeans" },
        ":inc": { "N": "1" },
        ":childVersion": { "N": "$ctx.result.child.version" }
      },
      "ConditionExpression": "version = :childVersion"
    }
  },
  # 更新彩蛋的软心糖数量为0，校验当前数量、版本号及非负
  {
    "Update": {
      "TableName": "Egg-${env}",
      "Key": $util.dynamodb.toMapValues($eggKey),
      "UpdateExpression": "SET jellybeans = :newAmount, version = version + :inc",
      "ExpressionAttributeValues": {
        ":newAmount": { "N": "0" },
        ":inc": { "N": "1" },
        ":currentAmount": { "N": "$ctx.result.egg.jellybeans" },
        ":eggVersion": { "N": "$ctx.result.egg.version" },
        ":zero": { "N": "0" }
      },
      "ConditionExpression": "jellybeans = :currentAmount AND version = :eggVersion AND jellybeans > :zero"
    }
  },
  # 创建转移记录
  {
    "Put": {
      "TableName": "JellybeanTransfer-${env}",
      "Item": {
        "id": { "S": $util.autoId() },
        "childId": { "S": $ctx.args.childId },
        "eggId": { "S": $ctx.args.eggId },
        "transferAmount": { "N": "$ctx.result.egg.jellybeans" },
        "timestamp": { "S": $util.time.nowISO8601() }
      }
    }
  }
])

# 执行事务
$util.toJson({
  "version": "2018-05-29",
  "operation": "TransactWriteItems",
  "transactItems": $transactItems
})

3. 前端代码调整 #

不再分步调用DataStore.save，而是调用自定义的transferJellybeans Mutation，无论是在线还是离线场景，DataStore都会将该操作加入同步队列，确保原子性：

import { API, graphqlOperation } from 'aws-amplify';
import { transferJellybeans } from './graphql/mutations';

async function findEgg(childId, eggId) {
  try {
    const result = await API.graphql(
      graphqlOperation(transferJellybeans, { childId, eggId })
    );
    console.log('转移成功:', result.data.transferJellybeans);
  } catch (error) {
    console.error('转移失败:', error);
    // 失败时无需手动回滚，事务已自动处理
  }
}

关键优势 #

• 原子性：所有操作要么全部成功，要么全部回滚，避免分步更新导致的不一致；
• 冲突检测：通过版本号校验，确保客户端操作基于最新的数据状态；
• 数据完整性：通过ConditionExpression校验软心糖数量非负，且彩蛋未被重复转移；
• 审计能力：JellybeanTransfer记录所有转移操作，方便后续排查问题。

内容的提问来源于stack exchange，提问作者jdowdell