# バッチ設計書 11-createCollectionImport

## 概要

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

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

createCollectionImportは、CSV/JSONファイルからドキュメントをPayloadコレクションにバッチインポートするジョブキュータスクである。

**業務上の目的・背景**：大量のコンテンツを外部システムからPayload CMSに移行する際、手動での入力は非現実的である。本バッチは、CSVやJSONファイルから構造化されたデータを読み込み、指定したコレクションにドキュメントを一括登録・更新することで、データ移行作業を自動化する。既存システムからのマイグレーション、定期的なデータ同期、バックアップからのリストアなど、多様なユースケースに対応する。

**バッチの実行タイミング**：ユーザーが管理画面からインポートファイルをアップロードし、インポートを開始した時点でジョブキューに登録され、非同期で実行される。手動トリガーが基本だが、外部システムとの連携によりAPI経由での自動実行も可能。

**主要な処理内容**：
1. インポートドキュメントの取得とファイル読み込み（ローカル/クラウドストレージ対応）
2. ファイル形式（CSV/JSON）の判定とパース処理
3. CSVの場合はフラット構造からネスト構造への展開（unflatten）
4. インポートモード（create/update/upsert）に応じたドキュメント操作
5. マルチロケール対応（複数言語データの分離・順次更新）
6. バッチ単位での処理とエラー継続処理
7. 処理結果（成功件数、エラー件数、詳細）のインポートドキュメントへの記録

**前後の処理との関連**：importBatchProcessorを内部で使用してバッチ処理を実行する。エクスポートバッチ（exportBatchProcessor）と対をなし、エクスポートしたデータを別環境にインポートする用途で使用される。

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

## バッチ種別

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

## 実行スケジュール

| 項目 | 内容 |
|-----|------|
| 実行頻度 | 随時（ユーザー操作時） |
| 実行時刻 | 任意 |
| 実行曜日 | 任意 |
| 実行日 | 任意 |
| トリガー | イベント（ジョブキュー） |

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| インポートドキュメント存在 | importIdで指定されたインポートドキュメントが存在すること |
| ファイル存在 | インポートドキュメントに関連付けられたファイル（CSV/JSON）が存在すること |
| コレクション存在 | インポート先のコレクションがPayload設定に存在すること |
| ユーザー認証 | userCollectionとuserIDが指定され、有効なユーザーが存在すること |

### 実行可否判定

- インポートドキュメントが存在しない場合はエラー終了
- ファイルが存在しないまたは読み込めない場合はFileRetrievalErrorをスロー
- MIMEタイプが判定できない場合はFileRetrievalErrorをスロー
- maxLimitを超えるドキュメント数の場合はAPIErrorをスロー

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| importId | string | Yes | - | インポートドキュメントのID |
| importCollection | string | Yes | - | インポートコレクションのスラッグ |
| userID | string \| number | No | - | 実行ユーザーのID |
| userCollection | string | No | - | ユーザーコレクションのスラッグ |
| batchSize | number | No | 100 | 1バッチあたりの処理件数 |
| debug | boolean | No | false | デバッグログの出力有無 |
| defaultVersionStatus | 'draft' \| 'published' | No | 'published' | バージョン管理コレクションのデフォルトステータス |
| maxLimit | number | No | - | インポート可能な最大ドキュメント数 |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| インポートドキュメント | DB | インポート設定とファイル情報を保持 |
| インポートファイル | CSV/JSON | インポート対象データ（ローカルまたはクラウドストレージ） |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| 対象コレクション | DB | インポートされたドキュメント |
| インポートドキュメント | DB | 処理結果（ステータス、サマリー） |

### 出力ファイル仕様

ファイル出力なし（データベースへの出力のみ）

## 処理フロー

### 処理シーケンス

```
1. インポートドキュメント取得
   └─ importIdからインポートドキュメントをfindByIDで取得
2. コレクション設定取得
   └─ importCollectionからコレクション設定を取得
3. ファイル取得
   └─ getFileFromDocでローカル/クラウドストレージからファイルを取得
4. MIMEタイプ判定
   └─ ファイルのMIMEタイプからフォーマット（CSV/JSON）を決定
5. createImport呼び出し
   └─ インポート処理の本体を実行
   5.1. ユーザー認証
        └─ userCollectionとuserIDからユーザーを取得・検証
   5.2. ファイルパース
        └─ CSV: parseCSV → unflattenObject / JSON: parseJSON
   5.3. maxLimit検証
        └─ ドキュメント数がmaxLimitを超えていないか確認
   5.4. 無効フィールド除去
        └─ disabledFields設定に基づきフィールドを削除
   5.5. バッチプロセッサ実行
        └─ createImportBatchProcessor.processImportを呼び出し
6. 結果記録
   └─ インポートドキュメントにステータスとサマリーを更新
```

### フローチャート

```mermaid
flowchart TD
    A[タスク開始] --> B[インポートドキュメント取得]
    B --> C{ドキュメント存在?}
    C -->|No| D[Error: Document not found]
    C -->|Yes| E[コレクション設定取得]
    E --> F{コレクション存在?}
    F -->|No| G[Error: Collection not found]
    F -->|Yes| H[ファイル取得]
    H --> I{ファイル取得成功?}
    I -->|No| J[FileRetrievalError]
    I -->|Yes| K[MIMEタイプ判定]
    K --> L{MIMEタイプ判定可?}
    L -->|No| M[FileRetrievalError]
    L -->|Yes| N[createImport実行]
    N --> O{maxLimit超過?}
    O -->|Yes| P[APIError: Limit exceeded]
    O -->|No| Q[バッチ処理実行]
    Q --> R[結果集計]
    R --> S[インポートドキュメント更新]
    S --> T{エラー件数}
    T -->|0| U[status: completed]
    T -->|>0 and 成功>0| V[status: partial]
    T -->|全件失敗| W[status: failed]
    U --> X[タスク終了]
    V --> X
    W --> X
```

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

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

| 処理 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| インポートドキュメント取得 | {importCollection} | SELECT | インポート設定の取得 |
| ユーザー取得 | {userCollection} | SELECT | 実行ユーザーの認証 |
| 既存ドキュメント検索 | {collectionSlug} | SELECT | update/upsertモード時の既存ドキュメント検索 |
| ドキュメント作成 | {collectionSlug} | INSERT | createモードまたはupsertで新規作成時 |
| ドキュメント更新 | {collectionSlug} | UPDATE | updateモードまたはupsert時 |
| インポートドキュメント更新 | {importCollection} | UPDATE | 処理結果の記録 |
| ロケール更新 | {collectionSlug} | UPDATE | マルチロケールデータの順次更新 |

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

#### インポートドキュメント（{importCollection}）

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| SELECT | id, filename, mimeType, url, collectionSlug, importMode, matchField | id = importId | インポート設定取得 |
| UPDATE | status | 'completed' / 'partial' / 'failed' | 処理結果に応じたステータス |
| UPDATE | summary.imported | 成功作成件数 | - |
| UPDATE | summary.updated | 成功更新件数 | - |
| UPDATE | summary.total | 処理対象総件数 | - |
| UPDATE | summary.issues | エラー件数 | - |
| UPDATE | summary.issueDetails | エラー詳細配列 | row, data, errorを含む |

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

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| SELECT | * | {matchField} = {matchValue} | update/upsert時の既存ドキュメント検索 |
| INSERT | * | インポートデータから | createモードまたはupsert新規作成 |
| UPDATE | * | インポートデータから（id, _id, createdAt, updatedAt除外） | update/upsertモード |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| NOT_FOUND | 業務エラー | インポートドキュメントが存在しない | importIdを確認して再実行 |
| COLLECTION_NOT_FOUND | 設定エラー | コレクションが存在しない | コレクション設定を確認 |
| FILE_RETRIEVAL | システムエラー | ファイル取得失敗 | ファイルパスとストレージ設定を確認 |
| MIMETYPE_ERROR | 業務エラー | MIMEタイプ判定不可 | ファイルのMIMEタイプを確認 |
| LIMIT_EXCEEDED | 業務エラー | maxLimit超過 | ファイルを分割するかlimit設定を確認 |
| USER_NOT_FOUND | 認証エラー | ユーザーが存在しない | ユーザー情報を確認 |
| VALIDATION_ERROR | 業務エラー | データ検証失敗 | インポートデータの形式を確認 |
| MATCH_FIELD_NOT_FOUND | 業務エラー | matchFieldの値がドキュメントにない | インポートデータのマッチフィールドを確認 |
| DOCUMENT_NOT_FOUND | 業務エラー | updateモードで既存ドキュメントが見つからない | マッチフィールドの値を確認 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | ジョブキュー設定に依存 |
| リトライ間隔 | ジョブキュー設定に依存 |
| リトライ対象エラー | システムエラー（ファイル取得失敗、DB接続エラー等） |

### 障害時対応

- 個別ドキュメントのインポートエラーは記録して処理を継続
- バッチ全体のエラーはインポートドキュメントのstatusに反映
- エラー詳細はsummary.issueDetailsに記録され、管理画面から確認可能
- トランザクションは各ドキュメントで独立しており、失敗したドキュメントのみロールバック

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

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

**特記事項**: バッチプロセッサは親リクエストのtransactionIDを分離（isolateObjectProperty）し、各ドキュメントを独立したトランザクションで処理する。これにより、1件の失敗が他のドキュメントに影響しない。

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | maxLimit設定に依存（デフォルト無制限） |
| 目標処理時間 | バッチサイズと件数に依存 |
| メモリ使用量上限 | ファイルサイズに依存（全データをメモリに読み込み） |

## 排他制御

- 同一インポートドキュメントへの同時実行は排他制御なし（ジョブキューで管理）
- 各ドキュメントの操作はPayloadの標準的なアクセス制御に従う
- overrideAccessはfalseに設定され、ユーザーの権限に基づいてアクセス制御

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| 開始ログ | タスク開始時 | インポート設定（collectionSlug, format, importMode, matchField） |
| 進捗ログ | バッチ処理中（debugモード時） | ファイル情報、パース結果、バッチ進捗 |
| 終了ログ | タスク終了時 | 処理結果（imported, updated, total, errors） |
| エラーログ | エラー発生時 | エラー詳細、対象ドキュメント情報 |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| 処理時間 | ジョブキュー設定に依存 | ジョブキュー監視 |
| エラー件数 | 全件失敗時 | インポートドキュメントのstatus='failed' |

## 備考

- CSV形式のインポートでは、フラット形式（dot notation）からネスト構造への変換が自動的に行われる
- マルチロケール対応: ロケールキー（en, ja等）を含むオブジェクトは、最初のロケールで作成後、他のロケールを順次更新
- バージョン管理対象コレクションでは、_statusフィールドまたはdefaultVersionStatusに基づいてdraft/publishedが決定
- MongoDB使用時、updateモードでIDフィールドの値がObjectID形式でない場合はエラーとなる（upsertでは新規作成）
- カスタムIDフィールドを持つコレクションでは、インポートデータのIDがそのまま使用される
