# バッチ設計書 13-importBatchProcessor

## 概要

本ドキュメントは、Payload CMSのインポート/エクスポートプラグインにおけるインポートバッチプロセッサ（importBatchProcessor）のバッチ処理設計を定義する。

### 本バッチの処理概要

importBatchProcessorは、パースされたドキュメント配列をバッチ単位で処理し、Payloadコレクションにインポートするための汎用バッチ処理ユーティリティである。

**業務上の目的・背景**：大量のドキュメントをインポートする際、1件ずつ処理すると時間がかかり、一方で全件を一度に処理するとメモリやトランザクションの問題が発生する。本バッチプロセッサは、ドキュメントを適切なサイズのバッチに分割して処理することで、パフォーマンスと安定性を両立する。また、個別のエラーを記録しながら処理を継続することで、部分的な成功を許容するロバストなインポート処理を実現する。

**バッチの実行タイミング**：createImport関数から内部的に呼び出される。ファイルのパースが完了した後、ドキュメント配列を受け取って処理を開始する。

**主要な処理内容**：
1. ドキュメント配列のバッチ分割（createBatches）
2. 各バッチの順次処理（processImportBatch）
3. インポートモード（create/update/upsert）に応じた操作
4. マルチロケールデータの分離と順次更新
5. バージョン管理対象コレクションのステータス処理
6. トランザクション分離による個別エラーの隔離
7. エラー分類と記録

**前後の処理との関連**：createImport関数から呼び出され、パース済みドキュメント配列を受け取る。createCollectionImportタスクの中核処理として機能する。

**影響範囲**：指定されたコレクションのドキュメント（作成/更新）、バージョン管理対象コレクションの場合はバージョン履歴にも影響。

## バッチ種別

データ連携 / データインポート / バッチ処理

## 実行スケジュール

| 項目 | 内容 |
|-----|------|
| 実行頻度 | 随時（createImportからの呼び出し時） |
| 実行時刻 | 任意 |
| 実行曜日 | 任意 |
| 実行日 | 任意 |
| トリガー | イベント（createImport経由） |

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| パース済みドキュメント | ドキュメント配列が提供されていること |
| コレクション存在 | インポート先コレクションがPayload設定に存在すること |
| 有効なリクエスト | PayloadRequestオブジェクトが有効であること |

### 実行可否判定

- ドキュメント配列が空の場合は即座に結果を返却（処理件数0）
- 各ドキュメントの処理エラーは記録して処理を継続

## 入力仕様

### 入力パラメータ

#### createImportBatchProcessor オプション

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| batchSize | number | No | 100 | 1バッチあたりの処理件数 |
| defaultVersionStatus | 'draft' \| 'published' | No | 'published' | バージョン管理コレクションのデフォルトステータス |

#### processImport オプション

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| collectionSlug | string | Yes | - | インポート先コレクションのスラッグ |
| documents | Record<string, unknown>[] | Yes | - | インポート対象ドキュメント配列 |
| importMode | 'create' \| 'update' \| 'upsert' | Yes | - | インポートモード |
| matchField | string | No | 'id' | update/upsert時のマッチングフィールド |
| req | PayloadRequest | Yes | - | Payloadリクエストオブジェクト |
| user | TypedUser | No | - | 実行ユーザー |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| パース済みドキュメント | メモリ | CSV/JSONからパースされたドキュメント配列 |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| ImportResult | メモリ | インポート結果 |
| 対象コレクション | DB | インポートされたドキュメント |

### ImportResult構造

| 項目名 | 型 | 説明 |
|--------|-----|------|
| errors | Array<{doc, error, index}> | エラー情報配列 |
| imported | number | 作成成功件数 |
| updated | number | 更新成功件数 |
| total | number | 処理対象総件数 |

### ImportError構造

| 項目名 | 型 | 説明 |
|--------|-----|------|
| type | string | エラー種別（validation/notFound/duplicate/database/unknown） |
| documentData | Record | エラー発生ドキュメント |
| error | string | エラーメッセージ |
| rowNumber | number | 1始まりの行番号 |
| field | string | (オプション) エラー発生フィールド |

## 処理フロー

### 処理シーケンス

```
1. バッチプロセッサ生成
   └─ createImportBatchProcessor(options)でインスタンス生成
2. processImport呼び出し
   └─ インポート処理のメイン実行
   2.1. バッチ分割
        └─ createBatches(documents, batchSize)
   2.2. 結果オブジェクト初期化
        └─ errors: [], imported: 0, updated: 0, total: documents.length
   2.3. バッチループ
        └─ 各バッチに対してprocessImportBatchを実行
   2.4. processImportBatch
        └─ リクエストのtransactionIDを分離
        2.4.1. ドキュメントループ
               └─ 各ドキュメントに対して処理
        2.4.2. インポートモード判定
               └─ create/update/upsertに応じた処理
        2.4.3. マルチロケール検出
               └─ extractMultiLocaleData
        2.4.4. ドキュメント操作
               └─ payload.create/payload.update
        2.4.5. ロケール更新
               └─ 追加ロケールの順次更新
        2.4.6. 結果記録
               └─ successful/failedに追加
   2.5. 結果集計
        └─ バッチ結果をマージ
3. 結果返却
   └─ ImportResultを返却
```

### フローチャート

```mermaid
flowchart TD
    A[processImport開始] --> B[バッチ分割]
    B --> C[結果オブジェクト初期化]
    C --> D{次のバッチ?}
    D -->|No| E[結果返却]
    D -->|Yes| F[processImportBatch]
    F --> G[transactionID分離]
    G --> H{次のドキュメント?}
    H -->|No| I[バッチ結果マージ]
    I --> D
    H -->|Yes| J{importMode?}
    J -->|create| K[createモード処理]
    J -->|update| L[updateモード処理]
    J -->|upsert| M[upsertモード処理]
    K --> N[マルチロケール検出]
    L --> O[既存ドキュメント検索]
    M --> O
    O --> P{ドキュメント存在?}
    P -->|Yes| Q[更新処理]
    P -->|No & upsert| R[作成処理]
    P -->|No & update| S[エラー記録]
    N --> T{マルチロケール?}
    T -->|Yes| U[作成後ロケール更新]
    T -->|No| V[通常作成]
    Q --> W[結果記録]
    R --> W
    S --> W
    U --> W
    V --> W
    W --> H
```

## データベース操作仕様

### 操作別データベース影響一覧

| 処理 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| 既存ドキュメント検索 | {collectionSlug} | SELECT | update/upsert時のマッチング検索 |
| ドキュメント作成 | {collectionSlug} | INSERT | createモードまたはupsert新規作成 |
| ドキュメント更新 | {collectionSlug} | UPDATE | update/upsertモード |
| ロケール更新 | {collectionSlug} | UPDATE | マルチロケールの順次更新 |

### テーブル別操作詳細

#### 対象コレクション（{collectionSlug}）

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| SELECT | id, * | {matchField} = {matchValue} | depth: 0, limit: 1 |
| INSERT | * | インポートデータ | draft設定はバージョン管理有無で決定 |
| UPDATE | * | インポートデータ（id, _id, createdAt, updatedAt除外） | depth: 0 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| validation | バリデーションエラー | フィールド値の検証失敗 | エラー詳細を確認して修正 |
| notFound | 業務エラー | updateモードで既存ドキュメントなし | マッチフィールドの値を確認 |
| duplicate | 業務エラー | 一意制約違反 | 重複データを除外 |
| database | システムエラー | DB接続エラー、トランザクションエラー | システム状態を確認 |
| unknown | その他エラー | 上記以外のエラー | エラーメッセージを確認 |

### エラー分類ロジック

```typescript
if (message.includes('validation')) return 'validation'
if (message.includes('not found')) return 'notFound'
if (message.includes('duplicate') || message.includes('unique')) return 'duplicate'
if (message.includes('database') || message.includes('transaction')) return 'database'
return 'unknown'
```

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | なし（個別エラーは記録して継続） |
| リトライ間隔 | N/A |
| リトライ対象エラー | N/A |

### 障害時対応

- 個別ドキュメントのエラーはfailed配列に記録
- 他のドキュメントの処理は継続
- エラー詳細（rowNumber, documentData, error, type）を記録

## トランザクション仕様

| 項目 | 内容 |
|-----|------|
| トランザクション範囲 | ドキュメント単位 |
| コミットタイミング | 各ドキュメントの処理完了時 |
| ロールバック条件 | 個別ドキュメントのエラー発生時 |

**特記事項**:
- `isolateObjectProperty`でリクエストのtransactionIDを分離
- `req.transactionID = undefined`で親トランザクションから独立
- これにより、アクセス制御エラー（Forbidden）等で`killTransaction`が呼ばれても親トランザクションに影響しない

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | batchSizeに依存（デフォルト100件/バッチ） |
| 目標処理時間 | ドキュメント数とコレクション複雑さに依存 |
| メモリ使用量上限 | バッチサイズ分のドキュメントがメモリに保持 |

## 排他制御

- overrideAccessはfalse（ユーザー権限に基づくアクセス制御）
- 同一ドキュメントへの同時更新はPayloadの標準的なロック機構に依存

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| デバッグログ | ドキュメント作成時（debugモード時） | タイトル、ステータス |
| デバッグログ | ドキュメント更新時（debugモード時） | 既存ID、更新データ |
| エラーログ | ロケール更新失敗時 | ロケール、ドキュメントID、エラー詳細 |
| エラーログ | 更新失敗時 | ドキュメントID、エラー詳細 |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| エラー率 | 呼び出し元の設定に依存 | ImportResult.errors |

## 備考

### インポートモード詳細

1. **create**
   - 常に新規ドキュメントを作成
   - カスタムIDフィールドがない場合はインポートデータのidを削除
   - バージョン管理対象コレクションでは_statusまたはdefaultVersionStatusを使用

2. **update**
   - matchFieldでマッチするドキュメントを更新
   - マッチするドキュメントがない場合はエラー
   - id, _id, createdAt, updatedAtは更新データから除外

3. **upsert**
   - matchFieldでマッチすれば更新、なければ作成
   - MongoDB ObjectID形式でない場合のエラーハンドリングあり

### マルチロケール処理

```typescript
// 入力例：{ title: { en: 'Hello', ja: 'こんにちは' } }
// 処理：
// 1. 最初のロケール（en）でドキュメント作成
// 2. 残りのロケール（ja）で順次更新
```

### MongoDB ObjectID対応

- update/upsert時にIDフィールドでマッチングする場合
- 24文字の16進数文字列かどうかを正規表現でチェック
- 無効な形式の場合、upsertでは新規作成、updateではエラー
