本シリーズについて

Amplify Gen 2 で開発していると、a.model() で足りない処理はすべて Lambda に回したくなります。しかし実際には、カスタムリゾルバーで書けるものが少なくありません。本シリーズでは、実務で遭遇した要件を題材に、リゾルバーで完結できるケースを紹介します。

• 第1回：複数テーブルを安全に更新する（TransactWriteItems）（本記事）
• 第2回：別テーブルの状態に依存する認可（パイプラインリゾルバー）

はじめに

こんにちは！マイリンです。

Amplify で開発していると、@model（Gen 2 では a.model()）を書くだけで、CRUD のリゾルバーが自動生成されます。認可も .authorization() で宣言できます。しばらくのあいだ、私は自分でリゾルバーを書く必要がありませんでした。

しかし、ある要件で行き詰まりました。

「2 つのテーブルを同時に更新したい。ただし、片方が失敗したら、もう片方も書き込まないでほしい」

具体的には、シナリオの完了記録を作成すると同時に、そのユーザーがそのシナリオを完了した回数を加算する、という処理です。片方だけが成功すると、集計値と実際の記録がずれてしまいます。

自動生成されるリゾルバーは、1 回の操作で 1 つの項目しか扱いません。

このとき私が最初に考えたのは、a.handler.function() で Lambda を追加することでした。整合性が絡む処理なのだから、コードで制御するべきだろう、と。

しかし、実際にやろうとしていたのは「2 件書き込む」だけです。外部ライブラリも、時間のかかる処理も必要ありません。この程度のために Lambda を 1 つ増やす必要が、本当にあるのでしょうか。

結論から言うと、必要ありませんでした。本記事では、パイプラインリゾルバーでは解決できない理由と、TransactWriteItems を使った実装、そして実装中にハマった 5 つの点を書きます。

対象読者

• Amplify で AppSync を使っている方
• 複数テーブルの整合性を保つ必要がある方
• a.model() で足りない処理を、すべて Lambda で書いている方

カスタムリゾルバーを一度も書いたことがなくても読めるように書いていますが、GraphQL と DynamoDB の基本的な用語は前提とします。

パイプラインリゾルバーは原子性を保証しない

パイプラインリゾルバーは、複数の AppSync 関数を順番に実行する仕組みです。関数ごとに異なるデータソースを指定できるため、「テーブル A に書いてから、テーブル B に書く」という処理は確かに書けます。

私はこれで十分だと考えていました。しかし、これは誤解でした。

パイプラインが保証するのは「順番」であって、「原子性」ではありません。

関数を 2 つ並べた場合、それは独立した 2 回の API 呼び出しです。次のような順序で処理が進みます。

関数1: ScenarioProgressing に PutItem  → 成功（コミット済み）
関数2: UserScenario に UpdateItem      → 失敗

この時点で、関数 1 の書き込みはすでに完了しています。DynamoDB 側には何も残っていない、という状態にはなりません。そして、それを取り消してくれる仕組みは AppSync にはありません。

「では、失敗したときに関数 3 で打ち消せばよいのでは」と考えるかもしれません。いわゆる補償処理です。しかし、この方法には次の問題が残ります。

• 補償処理自体が失敗する可能性がある（そのときは誰が直すのか）
• 補償が実行されるまでの短い時間、不整合な状態が外から見える
• 関数が増えるほど、失敗パターンの組み合わせが増える

補償処理は、そもそもトランザクションが使えない場面（複数サービスをまたぐ場合など）のための手法です。同じ DynamoDB の中で完結するなら、素直にトランザクションを使うべきです。

TransactWriteItems という答え

AppSync の DynamoDB リゾルバーには、TransactWriteItems という操作が用意されています。これは 1 回のリクエストで複数テーブルに書き込むもので、公式ドキュメントには「トランザクションは all-or-nothing で実行され、いずれかのリクエスト項目でエラーが発生した場合、トランザクション全体が実行されない」と明記されています。

サポートされている操作は 4 種類です。

操作	内容
PutItem	項目を作成・上書きする
UpdateItem	項目の一部を更新する
DeleteItem	項目を削除する
ConditionCheck	書き込まずに条件だけを検査する

4 つ目の ConditionCheck は見落とされがちですが、有用です。「別のテーブルのレコードが特定の状態であるときだけ書き込む」といった要件を、1 回のトランザクションで表現できます。

1 回のトランザクションに含められる項目は 100 件までです。

実装 1：スキーマの定義

まず、カスタムミューテーションを定義します。a.handler.custom にデータソースとハンドラーのパスを指定します。

// amplify/data/resource.ts
import { a } from '@aws-amplify/backend';

const schema = a.schema({
  ScenarioProgressing: a
    .model({
      progressId: a.id().required(),
      userId: a.id().required(),
      scenarioId: a.id().required(),
      completedAt: a.datetime(),
    })
    .identifier(['progressId'])
    .authorization((allow) => [allow.authenticated()]),

  UserScenario: a
    .model({
      userId: a.id().required(),
      completedCount: a.integer(),
      scenarioId: a.id().required(),
    })
    .identifier(['userId', 'scenarioId'])
    .authorization((allow) => [allow.authenticated()]),

  completeScenario: a
    .mutation()
    .arguments({ scenarioId: a.id().required() })
    .returns(a.boolean())
    .authorization((allow) => [allow.authenticated()])
    .handler(
      a.handler.custom({
        dataSource: a.ref('ScenarioProgressing'),
        entry: './completeScenario.js',
      })
    ),
});

.identifier() でキー名を明示しています。リゾルバー側の key と条件式も、この名前に合わせる必要があります。

UserScenario は (userId, scenarioId) の複合キーです。この記事の設計では、ユーザーが学習を開始した時点で該当の行がすでに作成されている前提になっています。この前提は後述の条件式に効いてきます。

ここで注意すべき点があります。dataSource には ScenarioProgressing を指定していますが、実際には UserScenario にも書き込みます。トランザクションは、宣言したデータソースの外にも手を伸ばします。 この点は後ほど権限の節で扱います。

また、戻り値を a.boolean() にしているのには理由があります。トランザクションの成功時に返ってくるのはキーだけで、書き込んだ内容そのものは返ってきません。この点も後述します。

実装 2：リクエストハンドラー

// amplify/data/completeScenario.js
import { util } from '@aws-appsync/utils';

export function request(ctx) {
  // 認証方式と sub の存在を先に確認する
  if (util.authType() !== 'User Pool Authorization') {
    util.unauthorized();
  }

  const userId = ctx.identity?.sub ?? '';
  if (!userId) {
    util.unauthorized();
  }
  
  const scenarioId = ctx.args.scenarioId
  const now = util.time.nowISO8601();

  // Amplify が生成する物理テーブル名を組み立てる
  const suffix = ctx.stash.awsAppsyncApiId + '-' + ctx.stash.amplifyApiEnvironmentName;

  return {
    operation: 'TransactWriteItems',
    transactItems: [
      {
        table: 'ScenarioProgressing-' + suffix,
        operation: 'PutItem',
        key: util.dynamodb.toMapValues({ progressId: util.autoId() }),
        attributeValues: util.dynamodb.toMapValues({
          completedAt: now,
          __typename: 'ScenarioProgressing',
          createdAt: now,
          updatedAt: now,
          userId, 
          scenarioId
        }),
        condition: {
          expression: 'attribute_not_exists(progressId)',
          returnValuesOnConditionCheckFailure: false,
        },
      },
      {
        table: 'UserScenario-' + suffix,
        operation: 'UpdateItem',
        key: util.dynamodb.toMapValues({ userId, scenarioId }),
        update: {
          expression: 'ADD completedCount :one SET updatedAt = :now',
          expressionValues: util.dynamodb.toMapValues({ ':one': 1, ':now': now }),
        },
        condition: {
          expression: 'attribute_exists(userId)',
          returnValuesOnConditionCheckFailure: false,
        },
      },
    ],
  };
}

それぞれの要素を順に見ていきます。

userId は引数から取らない、そして存在を確認する

userId は ctx.identity.sub から取得しています。クライアントから送られてきた値を使ってはいけません。他人の ID を送れば、他人のデータを書き換えられてしまいます。

ただし、それだけでは不十分です。ctx.identity.sub は、いつでも存在するとは限りません。

.authorization(allow => [allow.authenticated()]) が保証するのは「何らかの認証を通過していること」までです。API に複数の認証モードを設定している場合、ctx.identity の中身は認証方式によって変わります。IAM 認証やゲストアクセスを有効にしていると、期待する形の sub が入っていないことがあります。

ここで sub が空のまま処理が進むと、#scenarioId のようなキーでレコードが作られます。 書き込み系の処理では、これは実害のあるバグです。

そのため、util.authType() で認証方式を確認し、sub が取得できることを確認してから処理を進めます。

「今は User Pool しか設定していないから不要」と思える箇所ですが、後から認証モードを追加したときに挙動が変わります。認証モードの追加は、既存のリゾルバーを壊しうる変更です。

自動生成されたリゾルバーは owner を自動で埋めてくれていました。自分でリゾルバーを書くということは、こうした確認の責任も引き受けるということです。

条件式で「作成」と「更新」を区別する

2 つの項目には、それぞれ逆の条件が付いています。

項目	操作	条件
ScenarioProgressing	PutItem	attribute_not_exists(progressId)
UserScenario	UpdateItem	attribute_exists(userId)

attribute_not_exists(progressId) — 上書きを防ぐ

PutItem は、同じキーの項目がすでにあれば丸ごと上書きします。 今回は progressId を util.autoId() で採番しているため衝突はまず起きませんが、条件を付けておけば「万が一キーが重複したときに既存の記録を消してしまう」という事故を防げます。

PutItem に対する保険だと考えてください。

attribute_exists(userId) — 意図しない作成を防ぐ

こちらのほうが実務では重要です。DynamoDB の UpdateItem は、対象の項目が存在しない場合に新規作成します。 いわゆる upsert であり、「更新」という名前から想像するのとは挙動が異なります。

条件式は、その操作が作成なのか更新なのかをコード上で宣言する役割を持っています。

ADD によるアトミックな加算

2 つ目の項目では ADD completedCount :one を使っています。現在値を読んでから 1 を足して書き戻す、という手順を踏みません。DynamoDB 側で加算されるため、同時実行しても値がずれません。

なお、属性名が DynamoDB の予約語と衝突する場合は、そのままでは式に書けません。たとえば count や status は予約語です。その場合は expressionNames で #count のような別名を付ける必要があります。今回の completedCount は予約語ではないため、そのまま記述しています。

実装 3：権限の付与

a.handler.custom で dataSource を指定していれば、そのテーブルへの権限は Amplify が自動で付与します。今回のスキーマでは ScenarioProgressing をデータソースとして宣言しているため、このテーブルについては何もする必要がありません。

問題はもう一方です。TransactWriteItems は 1 つの関数から複数のテーブルに書き込みますが、UserScenario はデータソースとして宣言されていません。足りないのはこの 1 つだけです。

// amplify/backend.ts
import { defineBackend } from '@aws-amplify/backend';
import { data } from './data/resource';

const backend = defineBackend({ data });

// ScenarioProgressing はデータソースなので付与不要
// データソース外の UserScenario にだけ権限を追加する
const userScenarioTable = backend.data.resources.tables['UserScenario'];
const role = backend.data.resources.roles['ScenarioProgressingIAMRole'];

userScenarioTable.grantWriteData(role);

つまり、付与が必要なのは「データソースとして宣言していないテーブル」だけです。すべてのテーブルに一律で grantWriteData を書くと、不要な権限を与えることになります。

権限が足りない場合、トランザクション全体が失敗します。しかもエラーメッセージからは、どのテーブルの権限が足りないのかが分かりにくいことがあります。トランザクションにテーブルを追加したときは、まず grantWriteData の書き忘れを疑ってください。

なお、この「自分で権限を書く」必要があるのは例外的なケースです。原則は 「宣言したデータソースの外に出るときだけ、権限を書く」 と覚えておくとよいと思います。

ロール名について

ロール名は <データソースに指定したモデル名>IAMRole の形式です。今回はデータソースが ScenarioProgressing なので ScenarioProgressingIAMRole になります。

ただし、この命名規則は Amplify Gen 2 の公式ドキュメントには記載されていません。 「データソース外のテーブルへの権限付与」という項目自体が存在しないため、バージョン更新時に変わる可能性があります。

本記事は @aws-amplify/backend ^1.16.1 でデプロイまで確認しています。

ハマりどころ ①：便利なユーティリティが使えない

通常の DynamoDB 操作であれば、@aws-appsync/utils/dynamodb の ddb.put() や ddb.update() が使えます。JavaScript のオブジェクトを渡すだけで、型変換は自動で行われます。

// 通常の書き込み（こう書ける）
import * as ddb from '@aws-appsync/utils/dynamodb';
return ddb.put({ key: { id: id }, item: { title: title } });

しかし、TransactWriteItems にはこれらのヘルパーがありません。リクエストオブジェクトを手で組み立て、util.dynamodb.toMapValues() で自分で型変換する必要があります。

toMapValues() は、{ userId: 'abc' } を { userId: { S: 'abc' } } のような DynamoDB の型表現に変換します。これを忘れると、実行時にエラーになります。

書き心地は一段階下がりますが、ここは割り切るしかありません。

ハマりどころ ②：テーブル名は物理名で指定する

table フィールドに渡すのは、DynamoDB の実際のテーブル名です。公式ドキュメントの例では 'posts' のような単純な文字列になっています。

ところが Amplify では、テーブル名は自動生成され、API ID と環境名がサフィックスとして付きます。開発環境と本番環境で名前が異なるため、ハードコードはできません。

ここで使えるのが、Amplify がスタッシュに入れてくれている値です。Amplify のドキュメントでは、テーブル名が <model-name>-<aws-appsync-api-id>-<amplify-api-environment-name> の形式になることが説明されています。

const suffix = ctx.stash.awsAppsyncApiId + '-' + ctx.stash.amplifyApiEnvironmentName;
// 例: ScenarioProgressing-abc123xyz-dev

ただし、この説明があるのはバッチ操作のページです。トランザクションについて調べているときには、まず辿り着かない場所にあります。私はこれを見つけるまでに時間を使いました。

ハマりどころ ③：Amplify のモデルが前提とする属性

attributeValues に __typename、createdAt、updatedAt を含めています。これは Amplify が生成するモデルが、これらの属性を前提としているためです。

自動生成されたリゾルバー経由で作成した項目には、これらが自動的に付与されます。手動で書き込む場合は、自分で設定しなければなりません。

これを忘れると、書き込み自体は成功しますが、後から list クエリで取得したときに型が解決できずエラーになる、といった形で問題が表面化します。原因が分かりにくい種類のバグです。

【要確認】必要な属性はモデルの定義によって異なります。自動生成のミューテーションで 1 件作成し、実際のテーブルの中身を確認してから記述することをおすすめします。

ハマりどころ ④：競合検出とは併用できない

公式ドキュメントには、この操作は競合検出（conflict detection）と併用した場合サポートされず、同時に使うとエラーになる可能性があると記載されています。

Amplify で DataStore を使っている場合、競合検出が有効になっています。この構成では TransactWriteItems を使えません。

これは実装の途中で気づくと手戻りが大きい制約です。設計の初期段階で、DataStore を使うかどうかを確定させておくべきです。

ハマりどころ ⑤：同じ項目を 2 回指定できない

ドキュメントには、2 つのリクエスト項目が同じ項目を対象にすることはできず、そうした場合は TransactionCanceledException が発生すると書かれています。

これは、入力の配列からループで transactItems を組み立てるときに起きやすい問題です。たとえば「複数のシナリオをまとめて完了させる」ように拡張したとき、入力配列に同じ scenarioId が重複して含まれていると、UserScenario の同じ項目を 2 回更新することになり TransactionCanceledException になります。

より厄介なのは、ユーザー単位の総計テーブルのようにキーが 1 つしかないテーブルを併せて更新する設計です。この場合、入力に重複がなくてもシナリオの数だけ同じ項目を指すため、必ず衝突します。入力の不備ではなく設計そのものが原因なので、テストデータでは再現せず、本番で初めて気づくことになりがちです。

対処はそれぞれ異なります。入力配列の重複は、transactItems を組み立てる前に除去します。総計テーブルのように必ず同じ項目を指すものは、直感的には「同じ項目に 2 回足せばよい」と思えますが DynamoDB は受け付けないため、加算値を先に合計してから 1 つの UpdateItem にまとめます。

配列から transactItems を組み立てる場合、APPSYNC_JS では while と C 形式の for が使えない点にも注意してください。for-of、for-in、Array.prototype.map() は使用できます。公式のサンプルも .map() を使っています。

エラーハンドリング：2 つの層を別々に見る

ここが通常のリゾルバーと最も違う点です。

トランザクションが失敗する経路は 2 つあります。

層	どこに出るか	例
インフラ層	ctx.error	権限不足、テーブル名の誤り、スロットリング
業務層	ctx.result.cancellationReasons	条件チェック不成立、項目の重複

この 2 つは別々に確認する必要があります。 後者を if (ctx.error) の中にネストさせると、条件チェックの失敗を取りこぼします。

export function response(ctx) {
  if (ctx.error) {
    util.error(ctx.error.message, ctx.error.type);
  }

  if (ctx.result.cancellationReasons) {
    util.error(
      'transaction cancelled',
      'TransactionCanceledException',
      null,
      ctx.result.cancellationReasons
    );
  }

  return true;
}

ctx.result.cancellationReasons は成功時には null になるため、そのまま真偽値として判定できます。ドキュメントによれば、keys と cancellationReasons は ctx.result に必ず存在することが保証されています。

util.error() の第 4 引数は errorInfo です。ここに cancellationReasons を渡すことで、何番目の項目がどの理由で失敗したのかをクライアントに伝えられます。配列の並び順はリクエスト項目の順序と一致し、失敗していない項目には type が None の要素が入ります。

注意： 条件チェックが失敗したとき、returnValuesOnConditionCheckFailure を false にしていない項目については、テーブルに存在していた項目の中身が cancellationReasons に入ります。それをそのまま errorInfo に渡すと、既存データがクライアントに露出します。機密性のあるテーブルでは false を指定するか、必要な情報だけを抜き出してください。

成功しても書き込んだ内容は返ってこない

ドキュメントによれば、成功時にレスポンスとして返るのはリクエスト項目のキーのみで、その順序はリクエスト項目の順序と同じです。

そのため、ミューテーションの戻り値としてモデル全体を返したい場合は、次のどちらかが必要です。

• レスポンスハンドラーで、引数と生成した値からオブジェクトを組み立てる
• パイプラインの後段に読み取り用の関数を追加する

今回のように戻り値を Boolean にしておくのが最も単純です。スキーマ設計の段階で意識しておくとよい点です。

動作確認

デプロイ後、次の順序で確認しました。

1. 正常系：ミューテーションを実行し、両方のテーブルに反映されることを確認する
2. 連続実行：もう一度実行し、記録が 2 件になり完了数も 2 になることを確認する
3. 異常系：UserScenario が存在しないユーザーで実行する

3 が最も重要です。attribute_exists(userId) によってトランザクション全体が失敗し、ScenarioProgressing にも 1 件も書き込まれていないことを確認します。

ここで記録だけが増えていたら、トランザクションが効いていません。パイプラインで実装していた場合、まさにこの状態になります。

Lambda を選ぶべきライン

冒頭の問いに戻ります。「2 件書き込むために Lambda を増やす必要があるか」——答えは、ありませんでした。

TransactWriteItems はカスタムリゾルバーで完結します。Lambda を挟まないぶん、コールドスタートも、追加の IAM ロールも、デプロイパッケージの管理も発生しません。

ただし、トランザクション自体にはコストがあります。 公式ドキュメントには、DynamoDB がトランザクション内のすべての項目について 2 回の読み取りまたは書き込みを行う（準備用とコミット用）と記載されています。つまり 項目あたり 2 WCU です。Lambda の実行料金は消えますが、書き込みコストは倍になります。整合性のために払う対価だと理解しておくとよいと思います。

書けないものを先に知っておく

もう一つ、判断材料になるのがランタイムの制約です。APPSYNC_JS は ES6 に近い構文をサポートしており、オプショナルチェーン（?.）や null 合体演算子（??）も使えます。しかし、「普通の JavaScript がすべて書ける」わけではありません。

公式の「Supported runtime features」には、以下が未サポートであると記載されています。

• try / catch / finally / throw
• while ループ、C 形式の for (let i = 0; ...)
• インクリメント／デクリメント演算子（++ / --）
• 再帰呼び出し

特に try / catch が使えない点は重要です。エラーを捕捉して分岐する処理そのものが書けません。 失敗時にリトライする、代替処理に切り替える、といった要件が出てきた時点で、リゾルバーでは対応できません。

以上を踏まえて、以下の場合は Lambda を検討します。

状況	理由
100 件を超える	分割と部分失敗の設計が必要
集計サイズが 4MB を超える	件数に収まっていても上限に達する
エラーを捕捉して分岐したい	try / catch が使えない
リトライしたい	リゾルバーではバックオフを書けない
書き込み後に通知したい	外部システムへの連携が必要
競合検出を使っている	そもそも併用できない

まとめ

• パイプラインリゾルバーは順番を保証しますが、原子性は保証しません
• ctx.identity.sub は常に存在するとは限らないため、util.authType() で確認します
• 複数テーブルの整合性が必要なら TransactWriteItems を使います
• テーブル名は ctx.stash の値から組み立てます
• 権限を付与するのは、データソースとして宣言していないテーブルの分だけです
• ctx.error と ctx.result.cancellationReasons は別々に確認します
• 競合検出とは併用できません
• try / catch と while は使えません
• 成功時に返るのはキーだけです
• トランザクション書き込みは項目あたり 2 WCU かかります

この処理に Lambda は必要ありませんでした。

次回は、.authorization() では書けない認可要件を、パイプラインリゾルバーで実装する方法を扱います。こちらも、Lambda なしで完結するケースです。

誰かの参考になれば幸いです。

参考

• TransactWriteItems – AWS AppSync
• JavaScript resolvers overview – AWS AppSync
• Add custom queries and mutations – Amplify Gen 2
• Batch DynamoDB Operations – Amplify Gen 2
• Supported runtime features (APPSYNC_JS) – AWS AppSync
• Managing complex workflows with DynamoDB transactions
• Update expressions – Amazon DynamoDB