數據存取模式指南

---
description: when needs to query
globs:
alwaysApply: false
---
# 數據存取模式指南

## 🔄 CQRS 存取模式 (基於實際架構)

### 寫入模式 (Command Side)
- **表名**: `command-records`
- **主鍵**: `transaction_id` (HASH) + `created_at` (RANGE)
- **主要操作**: PUT_ITEM (新增推播記錄)
- **觸發**: 推播服務寫入命令
- **後續**: 自動觸發 DynamoDB Stream
- **Stream 配置**: `NEW_AND_OLD_IMAGES`

### 讀取模式 (Query Side)
- **表名**: `notification-records`
- **主鍵**: `transaction_id` (HASH 唯一鍵)
- **主要操作**: GET_ITEM、SCAN (現有實現)
- **索引使用**: 主表 + GSI (status-index 可選)
- **資料來源**: Stream 同步的最佳化資料

## 📊 實際查詢存取模式

### 模式1: 交易推播查詢 (`/tx` 端點)
```python
# 存取路徑: notification-records 主表 - GET_ITEM
# 實現於: QueryService.query_transaction_notifications()
if transaction_id:
    response = table.get_item(Key={"transaction_id": transaction_id})
else:
    # 查詢最近記錄 - SCAN
    response = table.scan(Limit=limit)
```

### 模式2: 失敗記錄查詢 (`/fail` 端點)
```python
# 存取路徑: notification-records 主表 - GET_ITEM + SCAN
# 實現於: QueryService.query_failed_notifications()
if transaction_id:
    response = table.get_item(Key={"transaction_id": transaction_id})
    # 檢查 status == "FAILED"
else:
    response = table.scan(FilterExpression=Attr("status").eq("FAILED"))
```

### 模式3: SNS 推播查詢 (`/sns` 端點)
```python
# 存取路徑: notification-records 主表 - SCAN + FILTER
# 實現於: QueryService.query_sns_notifications()
response = table.scan(
    FilterExpression=Attr("sns_id").eq(sns_id),
    ProjectionExpression="transaction_id, #token, platform, ..."
)
```

## 🔄 Stream 資料同步模式

### Stream 處理流程
```python
# 實現於: stream_processor_lambda/app.py
def process_stream_record(record):
    # 1. 解析 Command Record
    command_record = parse_command_record(new_image)

    # 2. 轉換為 Query Record
    query_record = transform_to_query_record(command_record)

    # 3. 儲存至 Query Side
    save_query_record(query_record)
```

### 資料轉換規則
```python
# CommandRecord → QueryRecord
# 移除敏感欄位: device_token, payload
# 保留查詢欄位: transaction_id, status, platform, etc.
def transform_to_query_record(command_record):
    return QueryRecord(
        transaction_id=command_record.transaction_id,
        created_at=command_record.created_at,
        status=command_record.status,
        platform=command_record.platform,
        # ... 其他非敏感欄位
    )
```

## 🛠️ 實作參考文件

### 核心實現
- **FastAPI 查詢服務**: [main.py](mdc:query-service/eks_handler/main.py) - QueryService 類別
- **Stream 處理器**: [stream_processor_lambda/app.py](mdc:query-service/lambdas/stream_processor_lambda/app.py)
- **Lambda 查詢處理**: [query_result_lambda/app.py](mdc:query-service/lambdas/query_result_lambda/app.py)

### 配置與設置
- **LocalStack 表格設置**: [setup.sh](mdc:query-service/infra/localstack/setup.sh)
- **Docker 環境**: [setup_docker.sh](mdc:query-service/infra/localstack/setup_docker.sh)

### 測試與驗證
- **整合測試**: [test_integration.py](mdc:query-service/tests/test_integration.py)
- **Lambda 測試**: [test_lambdas/](mdc:query-service/tests/test_lambdas)
- **測試事件**: [test-events/](mdc:query-service/test-events)

## ⚡ 效能優化策略

### 查詢最佳化
```python
# 推薦：點查詢 (最快)
table.get_item(Key={"transaction_id": "tx_001"})

# 謹慎：全表掃描 (較慢)
table.scan(FilterExpression=Attr("status").eq("FAILED"))

# 建議：增加 GSI 索引
# sns_id 查詢可能需要專用 GSI 來提升效能
```

### 資料傳輸最佳化
```python
# 使用 ProjectionExpression 減少傳輸量
ProjectionExpression="transaction_id, #token, platform, notification_title"

# 使用 ExpressionAttributeNames 處理保留字
ExpressionAttributeNames={"#token": "token", "#status": "status"}
```

### 監控與除錯
```bash
# 檢查 Stream 處理狀態
docker logs query-service-stream-processor-lambda-1

# 監控表格大小
awslocal dynamodb describe-table --table-name notification-records

# 驗證資料同步
./scripts/testing/test_full_flow.sh
```

## 🎯 開發建議

### 新增查詢模式
1. 評估是否需要新的 GSI
2. 考慮查詢頻率和效能需求
3. 更新 Stream 處理邏輯 (如需要)
4. 添加相應的測試案例

### 效能監控重點
- **Command Side**: 寫入延遲、Stream 處理成功率
- **Query Side**: 查詢延遲、scan 操作頻率
- **資料一致性**: 兩側資料同步率