# バッチ設計書 12-fullSync

## 概要

本ドキュメントは、クライアントとサーバー間のデータ同期を行う「fullSync」サービスのバッチ設計書です。CRDT（Conflict-free Replicated Data Type）ベースのマージ処理により、複数デバイス間でのデータ整合性を保証します。

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

本バッチは、家計簿アプリケーションにおけるマルチデバイス対応とクラウド同期を実現するための中核的なデータ同期サービスです。ローカルデータベースの変更をサーバーに送信し、他のクライアントからの変更を受信・マージします。

**業務上の目的・背景**：ユーザーがスマートフォン、タブレット、デスクトップなど複数のデバイスから家計簿データにアクセスし、どのデバイスで入力した変更も他のデバイスに反映されることを保証します。オフライン時にもローカルで変更を行え、オンライン復帰時に自動的にサーバーと同期することで、シームレスなユーザー体験を提供します。CRDTアルゴリズムにより、同時編集によるコンフリクトを自動解決し、データの整合性を維持します。

**バッチの実行タイミング**：ローカルでデータ変更が発生した後、1秒のディレイを置いて自動実行されます（`FULL_SYNC_DELAY = 1000`）。また、アプリケーションがフォーカスを得た時（`app-focused`ハンドラー）や手動での同期要求時にも実行されます。テスト環境ではディレイなしで即時実行されます。

**主要な処理内容**：
1. ローカルのCRDTメッセージ（変更履歴）を収集し、エンコード
2. サーバーに対してPOSTリクエストでメッセージを送信
3. サーバーから受信したメッセージをデコードし、ローカルデータベースに適用
4. Merkleトライを用いた差分検出により、未同期データがあれば再帰的に同期
5. 同期完了後、スプレッドシートの予算変更をトリガー

**前後の処理との関連**：データ変更（INSERT/UPDATE/DELETE）発生時に`sendMessages`が呼ばれ、`scheduleFullSync`経由で本処理がスケジュールされます。同期完了後は`sync-event`を発行し、`advanceSchedulesService`などの後続処理をトリガーします。

**影響範囲**：全てのCRDT管理テーブル（transactions、accounts、payees、categories、category_groups、schedules、rules等）とメタデータテーブル（messages_crdt、messages_clock）に影響を与えます。また、プリファレンス（lastSyncedTimestamp等）も更新されます。

## バッチ種別

データ連携 / 双方向同期

## 実行スケジュール

| 項目 | 内容 |
|-----|------|
| 実行頻度 | 随時（イベント駆動） |
| 実行時刻 | ローカル変更発生後1秒 |
| 実行曜日 | - |
| 実行日 | - |
| トリガー | イベント（ローカルデータ変更後、アプリフォーカス時、手動実行） |

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| 同期モード | SYNCING_MODEが'enabled'であること（'disabled'や'offline'でないこと） |
| サーバー接続 | サーバーURLが設定されていること |
| 認証済み | ユーザートークンが有効であること |
| ファイルオープン | 予算ファイルがオープンされていること（prefs.getPrefs()がnullでないこと） |
| グループID | cloudFileIdとgroupIdが設定されていること |

### 実行可否判定

```typescript
function scheduleFullSync() {
  clearFullSyncTimeout();

  if (checkSyncingMode('enabled') && !checkSyncingMode('offline')) {
    if (process.env.NODE_ENV === 'test') {
      return fullSync().then(res => {
        if (isError(res)) {
          throw res.error;
        }
        return res;
      });
    } else {
      syncTimeout = setTimeout(fullSync, FULL_SYNC_DELAY);
    }
  }
}
```

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| sinceTimestamp | string | No | lastSyncedTimestamp or 5分前 | 同期開始タイムスタンプ |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| messages_crdt | DB | ローカルのCRDTメッセージ（変更履歴） |
| messages_clock | DB | Merkle clockの状態 |
| preferences | DB | lastSyncedTimestamp、groupId、cloudFileId |
| async-storage | Storage | user-token（認証トークン） |
| SYNC_SERVER | API | サーバーからの同期レスポンス |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| 各CRDTテーブル | DB | 受信したメッセージに基づくデータ更新 |
| messages_crdt | DB | 受信したメッセージの記録 |
| messages_clock | DB | 更新されたMerkle clockの保存 |
| preferences | DB | lastSyncedTimestampの更新 |
| async-storage | Storage | readOnlyフラグ（認証失敗時） |

### 出力ファイル仕様

ファイル出力はありません。

## 処理フロー

### 処理シーケンス

```
1. 同期開始イベント発行
   └─ app.events.emit('sync', { type: 'start' })
2. ローカルメッセージ収集
   └─ getMessagesSince(since)でsinceタイムスタンプ以降のメッセージを取得
3. メッセージエンコード
   └─ encoder.encode()でgroupId、cloudFileId、メッセージをバイナリ化
4. サーバー送信
   └─ postBinary()でSYNC_SERVERにPOSTリクエスト
5. レスポンスデコード
   └─ encoder.decode()でサーバーレスポンスをパース
6. メッセージ適用
   └─ receiveMessages()で受信メッセージをローカルDBに適用
7. Merkle差分チェック
   ├─ 差分あり：sinceTimestampを更新して再帰的に_fullSyncを呼び出し
   └─ 差分なし：lastSyncedTimestampを保存
8. 同期完了イベント発行
   └─ app.events.emit('sync', { type: 'success', tables, syncDisabled })
```

### フローチャート

```mermaid
flowchart TD
    A[同期開始] --> B{同期モード確認}
    B -->|disabled/offline| Z[終了]
    B -->|enabled| C[sync startイベント発行]
    C --> D[ローカルメッセージ取得]
    D --> E[メッセージエンコード]
    E --> F[サーバーへPOST]
    F --> G{レスポンス確認}
    G -->|エラー| H[エラー処理]
    H --> I[sync errorイベント発行]
    I --> Z
    G -->|成功| J[メッセージデコード]
    J --> K[受信メッセージ適用]
    K --> L{Merkle差分チェック}
    L -->|差分あり| M{リトライ上限?}
    M -->|Yes| N[out-of-sync エラー]
    N --> I
    M -->|No| O[再帰的に_fullSync]
    O --> D
    L -->|差分なし| P[lastSyncedTimestamp更新]
    P --> Q[sync successイベント発行]
    Q --> Z
```

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

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

| 処理 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| メッセージ取得 | messages_crdt | SELECT | sinceタイムスタンプ以降のメッセージを取得 |
| メッセージ適用 | 各CRDTテーブル | INSERT/UPDATE | 受信メッセージに基づくデータ変更 |
| メッセージ記録 | messages_crdt | INSERT | 受信メッセージをCRDTログに追加 |
| クロック更新 | messages_clock | UPDATE | Merkle trieの状態を保存 |
| 設定保存 | preferences | UPDATE | lastSyncedTimestampを更新 |

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

#### messages_crdt

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| SELECT | timestamp, dataset, row, column, value | timestamp > since | ローカル変更の取得 |
| INSERT | timestamp, dataset, row, column, value | 受信メッセージの内容 | サーバーからの変更記録 |

#### messages_clock

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| UPDATE | clock | serializeClock({...clock, merkle: currentMerkle}) | Merkle trie更新後 |

#### 各CRDTテーブル（transactions, accounts, payees等）

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| INSERT | id, column | row, value | 新規レコードの場合 |
| UPDATE | column | value WHERE id = row | 既存レコードの場合 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| out-of-sync | SyncError | Merkle差分解消のリトライが10回(同一diffTime)または100回(合計)を超えた | merkle hash再構築を試行、それでも解消しなければエラー通知 |
| invalid-schema | SyncError | 不正なテーブル名やカラム名でのクエリ実行 | エラーメタ情報と共にsync errorイベント発行 |
| decrypt-failure | SyncError | 暗号化データの復号に失敗 | sync errorイベント発行 |
| encrypt-failure | SyncError | データの暗号化に失敗 | sync errorイベント発行 |
| unauthorized | PostError | 認証トークンが無効または期限切れ | readOnlyモードに設定、sync unauthorizedイベント発行 |
| network-failure | PostError | サーバーへの接続に失敗 | sync error (subtype: network)イベント発行 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | 最大100回（同一diffTimeの場合は10回まで） |
| リトライ間隔 | 即時（再帰呼び出し） |
| リトライ対象エラー | Merkle差分が残っている場合 |

### 障害時対応

1. **out-of-sync**：ローカルのmessages_crdtからMerkle hashを再構築（rebuildMerkleHash）して再同期を試行
2. **unauthorized**：ユーザーをread-onlyモードに設定し、再認証を促す
3. **network-failure**：次回のローカル変更時またはアプリフォーカス時に自動リトライ
4. **invalid-schema**：バグの可能性が高いため、エラー詳細をcaptureExceptionで送信

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

| 項目 | 内容 |
|-----|------|
| トランザクション範囲 | 受信メッセージ適用処理全体（db.transaction内） |
| コミットタイミング | 全メッセージの適用完了時 |
| ロールバック条件 | トランザクション内で例外発生時（メッセージ適用失敗） |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | 数件〜数千件（変更メッセージ数に依存） |
| 目標処理時間 | 数秒〜数十秒（メッセージ量に依存） |
| メモリ使用量上限 | サーバー側ペイロード上限: 20MB |

## 排他制御

- `sequential`ラッパーにより`applyMessages`は逐次実行される
- `once`ラッパーにより`fullSync`は同時に1つだけ実行される
- `batchMessages`により複数のメッセージ送信をまとめて処理

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| 情報ログ | 同期開始時 | "Syncing since {timestamp}, {count} messages (attempt: {n})" |
| 情報ログ | サーバーレスポンス受信時 | "Got messages from server {count}" |
| デバッグログ | out-of-sync検出時 | 送受信メッセージ、クロック状態、Merkle hash等の詳細 |
| エラーログ | 例外発生時 | エラー内容（logger.log経由） |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| 同期エラー | 発生時 | クライアント（sync-eventイベント） |
| 認証失敗 | 発生時 | クライアント（sync unauthorizedイベント） |
| ネットワークエラー | 発生時 | クライアント（sync error subtypeイベント） |
| out-of-sync | 発生時 | Sentryへの例外送信（captureException） |

## 備考

- 本サービスは`once`関数でラップされており、同時に複数の同期処理が実行されることを防止します
- `FULL_SYNC_DELAY`（1000ms）のディレイにより、連続した変更操作での過剰な同期リクエストを抑制します
- インポートモード（`SYNCING_MODE === 'import'`）では、CRDTメッセージの記録やMerkle更新をスキップし、高速なデータ投入を実現します
- オフラインモード（`SYNCING_MODE === 'offline'`）では同期処理は実行されず、ローカル変更のみが記録されます
- サーバーからの受信メッセージは`Timestamp.recv`で論理クロックを更新してから適用されます
- 予算タイプ（`budgetType`）の変更は特別な処理が必要で、`setBudgetType`関数で即座に反映されます
