data-access-patterns

Instruction file imported from yonger0718/aws-cqrs-test (.cursor/rules/database/data-access-patterns.mdc). Copyright stays with the author.

數據存取模式指南

🔄 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 端點)

# 存取路徑: 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 端點)

# 存取路徑: 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 端點)

# 存取路徑: 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 處理流程

# 實現於: 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)

資料轉換規則

# 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 - QueryService 類別
• Stream 處理器: stream_processor_lambda/app.py
• Lambda 查詢處理: query_result_lambda/app.py

配置與設置

• LocalStack 表格設置: setup.sh
• Docker 環境: setup_docker.sh

測試與驗證

• 整合測試: test_integration.py
• Lambda 測試: test_lambdas/
• 測試事件: test-events/

⚡ 效能優化策略

查詢最佳化

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

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

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

資料傳輸最佳化

# 使用 ProjectionExpression 減少傳輸量
ProjectionExpression="transaction_id, #token, platform, notification_title"

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

監控與除錯

# 檢查 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 操作頻率
• 資料一致性: 兩側資料同步率