# Phase 3: Drive → 系統 同步 — Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use `superpowers:subagent-driven-development` (recommended) or `superpowers:executing-plans` to implement this plan task-by-task. **Spec:** [`design.md`](./design.md) (§4.2, §7, §8, §10, §11, §12.3, §13.3) **Master Index:** [`implementation-plan.md`](./implementation-plan.md) **Depends on:** Phase 1 + Phase 2（需要 OAuth、token manager、worker、folder mappings、APScheduler） **Goal:** User 在 Drive 上傳/更新/刪除 evidence 檔案 → 系統自動同步（檔案存進 Minio + DB 寫 evidence + UI 顯示）。Job COMPLETED 後停止接收新檔；revert 後可重抓 reconcile。前端 evidence 區塊區分來源、提供 Drive 連結、禁止刪除 DRIVE_SYNC evidence。 **Architecture:** 加 webhook receiver + channel renewer + 4 個新 job handler（PROCESS_DRIVE_CHANGES / IMPORT_DRIVE_FILE / SOFT_DELETE_EVIDENCE / RECONCILE_TASK_FOLDER）。Drive 檔案下載並快取到 Minio（沿用 jedi-file-upload）。`job_evidences` 表加 `source / drive_file_id / is_deleted` 等欄位。前端 evidence UI 改造（badge / Drive 連結 / disable delete）。Job revert 補一個 reconcile hook。 > **v0.2 Addendum（重要）**：本 plan v0.1 寫於「AO=task」假設下。實際上 1 AO 可能對應多個 task（job_execution），所以 v0.2 Phase 2 加入 Task layer（6 層結構）。Phase 3 隨之大幅簡化： > - **Task 9.5「補 `get_active_job_execution_by_ao_uid`」變成不必要** — 因為 Drive file 的 parent folder 直接是 task folder，TASK mapping 的 `scope_uid` 就是 `job_execution.uid`，不需要從 AO 反查 active job_execution > - **Task 9 ProcessDriveChangesHandler**：parent 必須是 `scope_type='TASK'` 的 mapping（不是 AO） > - **Task 10 ImportDriveFileHandler**：payload 改帶 `task_mapping_uid`（不是 `ao_mapping_uid`），handler 直接從 task mapping 取 `job_execution_id` > - **Task 12 ReconcileAoFolderHandler → ReconcileTaskFolderHandler**：以 task folder 為單位 reconcile（payload 帶 `task_uid`）。AO 級別的 reconcile 由「對該 AO 下所有 task 各跑一次」達成 > - **Edge case**：folder.parents 變動 → 需檢查 mapping 是 TASK 還是 AO/上層；TASK 改變 parent 通常代表 user 把整個 task folder 拖走（罕見）；上層改變則代表整層搬遷 > - **Task 軟刪 (系統內 `DELETE /grc/project//job/` 觸發)**：Phase 2 Task 20.6 hook 會直接 SQL UPDATE `drive_folder_mappings.is_unlinked=TRUE`，**不會**呼叫 Drive API 刪 folder。Phase 3 webhook 收到該 folder 內變更時應透過「`mapping.is_unlinked=TRUE` → SKIP」邏輯忽略（在 Task 9 ProcessDriveChangesHandler 的 `_find_task_mapping_for_file` 內已 implicit 處理；務必加 unit test 覆蓋） > > 詳見每個 Task 內 v0.2 註記。 > **v0.3 Addendum（Archive folder 機制）**： > - **Task 9 / Task 10**：`_process_one_change` 在處理任何 file change 之前，先 walk parents ancestors 檢查是否含 ARCHIVE mapping → 若有則 SKIP。理由：避免「系統 archive → Drive 變更 → webhook → 重新 import」迴圈。Task 10 ImportDriveFileHandler 也加 defensive check（即使 Task 9 已過濾，handler 為 per-job 也可能由其他流程獨立 enqueue）。 > - **Task 14 evidence delete guard**：移除 v0.1 / v0.2 的 `if evidence.source == "DRIVE_SYNC": raise ForbiddenError` 守衛，改為 `if DRIVE_SYNC → call orchestration.try_archive_drive_file(evidence) BEFORE soft-delete in DB`。`GRC_FORBIDDEN_DELETE_DRIVE_EVIDENCE` deprecated 但保留定義。 > - **新 Archive APIs**（位於 Task 14 之後新段落）： > - `DriveSyncOrchestrationService.try_archive_drive_file(evidence)` — best-effort 把 evidence 對應的 Drive 檔案 move 到所屬 AP 的 `_Archive/`；對應 single-evidence delete > - `DriveSyncOrchestrationService.try_archive_task_folder(task_uid)` — best-effort 把整個 task folder move 到 `_Archive/` + 標 mapping unlinked；對應 task delete（取代 Phase 2 Task 20.6 v0.2 的 `try_mark_task_folder_unlinked`） > - **Phase 2 Task 20.6 已同步改為 v0.3 archive 行為**（參見 phase-2 plan 內 v0.3 Addendum）。Phase 3 不需重複建 hook，但 orchestration method 實作放在 Phase 3（因為仰賴 Drive API client 的 `move_to_archive`，且需與 Phase 3 webhook ancestor-skip 邏輯一起測）。 > - **Test expectations**：`test/cloud_integration/test_job_evidence_drive_guard.py` 改名 / 改寫為 `test_job_evidence_drive_archive.py`，覆蓋「DELETE DRIVE_SYNC → orchestration.try_archive 被呼叫 → DB 軟刪 commit；archive 失敗仍允許 DB 軟刪」。Task 9 / 10 加 ancestor=ARCHIVE → SKIP 測試。 **Tech Stack:** Drive Changes API (`changes.watch`, `changes.list`) / Drive files.get_media + files.export / Minio via jedi-file-upload / Vue3 + PrimeVue --- ## File Structure (Phase 3) ### 後端新增 / 修改 ``` scripts/sql/ └── 2026-MM-DD-job-evidences-add-drive-source.sql # 新增：job_evidences ALTER infra/flow_engine/models/ └── job_evidence.py # 修改：加新欄位 (source/drive_file_id/...) domain/flow_engine/entity/ └── job_evidence_entity.py # 修改：加對應 attrs infra/flow_engine/mapper/ └── job_evidence_mapper.py # 修改：propagate 新欄位 infra/cloud_integration/google_drive/ ├── google_drive_api_client.py # 修改：加 changes.list / changes.watch / channels.stop / files.get_media / files.export └── google_drive_webhook_signer.py (optional) # 簽證 channel token app/cloud_integration/service/ ├── handlers/ │ ├── process_drive_changes_handler.py # 新增 │ ├── import_drive_file_handler.py # 新增 │ ├── soft_delete_evidence_handler.py # 新增 │ └── reconcile_ao_folder_handler.py # 新增 └── google_drive_webhook_service.py # 新增 (validate + enqueue) domain/cloud_integration/service/ └── webhook_channel_manager.py # 新增 (register / renew / stop) api/cloud_integration/routes/ ├── google_drive_webhook_route.py # 新增 webhook receiver └── google_drive_integration_route.py # 修改：connect callback 內呼叫 register webhook app/cloud_integration/service/ └── google_drive_integration_service.py # 修改：connect 完 register webhook + initial cursor # Evidence guard & response shape app/flow_engine/service/ └── job_evidence_service.py # 修改：delete_job_evidence 加 source 守衛 + add 預設 source / response 含新欄位 domain/flow_engine/service/ └── job_evidence_domain_service.py # 視需要加 by_drive_file_id 查詢 api/flow_engine/serializers/flow_engine/ └── job_evidence.py # 修改：response 加 source / drive_url / drive_file_id # Revert reconcile hook app/flow_engine/service/workflow_execution_service.py # 修改：revert_job 完成後 enqueue RECONCILE_TASK_FOLDER (v0.2) # Channel renewer core/scheduler.py # 修改：替換 placeholder domain/cloud_integration/service/webhook_channel_manager.py # 由 scheduler 呼叫 # DI wiring di_containers/cloud_integration/cloud_integration_containers.py # 修改：加 4 handler + webhook service + channel manager ``` ### 前端新增 / 修改 ``` src/components/grc/ ├── AuditControlRef.vue # 修改：evidence 列表加 source badge / Drive 連結 / 禁刪 DRIVE ├── JobExecutionDrawer.vue # 同上 └── EvidenceListItem.vue (optional 抽 component) # 視重構決定 ``` ### Tests ``` test/cloud_integration/ ├── test_process_drive_changes_handler.py ├── test_import_drive_file_handler.py ├── test_soft_delete_evidence_handler.py ├── test_reconcile_ao_folder_handler.py ├── test_webhook_channel_manager.py ├── test_google_drive_webhook_route.py └── test_job_evidence_drive_guard.py ``` --- ## Task List ### Task 1: SQL Migration — `job_evidences` 新欄位 **Files:** - Create: `scripts/sql/-job-evidences-add-drive-source.sql` > **v0.2 Addendum**：原 Phase 2 migration（`2026-04-24-google-drive-folder-mappings.sql`）的 `chk_drive_sync_jobs_type` CHECK constraint 含 `'RECONCILE_AO_FOLDER'`。Phase 3 改名為 `RECONCILE_TASK_FOLDER` 後，需在本 phase 的 migration 一併 ALTER： > ```sql > -- 5. 重新命名 RECONCILE job type () > ALTER TABLE compliance.drive_sync_jobs > DROP CONSTRAINT IF EXISTS chk_drive_sync_jobs_type; > ALTER TABLE compliance.drive_sync_jobs > ADD CONSTRAINT chk_drive_sync_jobs_type > CHECK (job_type IN ('INIT_PROJECT_FOLDERS','CREATE_FOLDER','RENAME_FOLDER', > 'PROCESS_DRIVE_CHANGES','IMPORT_DRIVE_FILE', > 'SOFT_DELETE_EVIDENCE','RECONCILE_TASK_FOLDER')); > ``` > 若 dev 環境已有 RECONCILE_AO_FOLDER 的歷史 row → 在 ALTER 之前 `UPDATE compliance.drive_sync_jobs SET job_type='RECONCILE_TASK_FOLDER' WHERE job_type='RECONCILE_AO_FOLDER';` - [ ] **Step 1**：建檔： ```sql -- Date: -- Purpose: job_evidences 加 Drive 同步相關欄位 + soft delete -- 1. 新增欄位 () ALTER TABLE compliance.job_evidences ADD COLUMN IF NOT EXISTS source VARCHAR(20) NOT NULL DEFAULT 'SYSTEM_UPLOAD', ADD COLUMN IF NOT EXISTS drive_file_id VARCHAR(100), ADD COLUMN IF NOT EXISTS drive_file_modified_at TIMESTAMPTZ, ADD COLUMN IF NOT EXISTS drive_last_modifying_user_email VARCHAR(255), ADD COLUMN IF NOT EXISTS is_deleted BOOLEAN NOT NULL DEFAULT FALSE, ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMPTZ, ADD COLUMN IF NOT EXISTS deleted_reason VARCHAR(40); -- 2. CHECK constraints () ALTER TABLE compliance.job_evidences ADD CONSTRAINT chk_job_evidences_source CHECK (source IN ('SYSTEM_UPLOAD','DRIVE_SYNC')); ALTER TABLE compliance.job_evidences ADD CONSTRAINT chk_job_evidences_deleted_reason CHECK (deleted_reason IS NULL OR deleted_reason IN ('USER_DELETED','DRIVE_DELETED','DRIVE_MOVED_OUT')); -- 3. 唯一索引：drive_file_id 唯一（NULL 允許多筆）() CREATE UNIQUE INDEX IF NOT EXISTS uq_job_evidences_drive_file_id ON compliance.job_evidences (drive_file_id) WHERE drive_file_id IS NOT NULL; -- 4. 一般查詢索引 () CREATE INDEX IF NOT EXISTS idx_job_evidences_source_active ON compliance.job_evidences (job_execution_id, source) WHERE is_deleted = FALSE; ``` - [ ] **Step 2**：跑 migration、`\d compliance.job_evidences` 確認新欄位。 - [ ] **Step 3**：commit ```bash git add scripts/sql/-job-evidences-add-drive-source.sql git commit -m "feat(job_evidence): add Drive sync columns + soft delete to job_evidences" ``` --- ### Task 2: ORM Model + Entity + Mapper 同步加新欄位 **Files:** - Modify: `infra/flow_engine/models/job_evidence.py` - Modify: `domain/flow_engine/entity/job_evidence_entity.py` - Modify: `infra/flow_engine/mapper/job_evidence_mapper.py` - [ ] **Step 1**：在 model 加： ```python source = Column(String(20), nullable=False, default="SYSTEM_UPLOAD") drive_file_id = Column(String(100)) drive_file_modified_at = Column(DateTime(timezone=True)) drive_last_modifying_user_email = Column(String(255)) is_deleted = Column(Boolean, nullable=False, default=False) deleted_at = Column(DateTime(timezone=True)) deleted_reason = Column(String(40)) ``` - [ ] **Step 2**：在 entity dataclass 加對應欄位。 - [ ] **Step 3**：在 mapper to_entity / apply_to_model 雙向加新欄位。 - [ ] **Step 4**：跑既有 evidence 相關測試：`pytest test/test_*evidence* -v`，確認沒 regression。 - [ ] **Step 5**：commit ```bash git add infra/flow_engine/models/job_evidence.py domain/flow_engine/entity/job_evidence_entity.py infra/flow_engine/mapper/job_evidence_mapper.py git commit -m "feat(job_evidence): add Drive sync attrs to model/entity/mapper" ``` --- ### Task 3: Repo 加 by_drive_file_id 查詢 + 軟刪除 helper **Files:** - Modify: `domain/flow_engine/repository/job_evidence_repository.py` (or i_*) - Modify: `infra/flow_engine/repository/job_evidence_repo_impl.py` - Create: `test/test_job_evidence_repo_drive.py` - [ ] **Step 1**：interface 加： ```python @abstractmethod def get_active_by_drive_file_id(self, drive_file_id: str) -> Optional[JobEvidenceEntity]: ... @abstractmethod def soft_delete_by_uid(self, uid: UUID, reason: str) -> bool: ... @abstractmethod def list_active_by_job_execution_and_source(self, job_execution_id: int, source: str) -> List[JobEvidenceEntity]: ... ``` - [ ] **Step 2**：寫測試（用 db_session）： - add 一筆 SYSTEM + 一筆 DRIVE_SYNC → list_active_by_source 各回 1 筆 - get_active_by_drive_file_id 找得到、軟刪後找不到 - soft_delete_by_uid 設 is_deleted=True、deleted_at、deleted_reason - [ ] **Step 3**：實作。 - [ ] **Step 4**：跑測試 → PASS。 - [ ] **Step 5**：commit ```bash git add domain/flow_engine/repository/ infra/flow_engine/repository/ test/test_job_evidence_repo_drive.py git commit -m "feat(job_evidence): add drive-aware repo queries + soft delete" ``` --- ### Task 4: GoogleDriveApiClient — 補 Changes / files.get_media / files.export **Files:** - Modify: `infra/cloud_integration/google_drive/google_drive_api_client.py` - Modify: `test/cloud_integration/test_google_drive_api_client.py` - [ ] **Step 1**：在 client 加方法： ```python # ── Changes API ──────────────────────────────── def get_start_page_token(self, tenant_id: int) -> str: resp = self._service(tenant_id).changes().getStartPageToken().execute() return resp["startPageToken"] def list_changes(self, tenant_id: int, page_token: str, page_size: int = 100) -> Dict[str, Any]: return self._service(tenant_id).changes().list( pageToken=page_token, pageSize=page_size, fields=("nextPageToken,newStartPageToken,changes(" "fileId,removed,time,file(" "id,name,mimeType,parents,trashed,modifiedTime,size," "webViewLink,lastModifyingUser(emailAddress,displayName)" "))"), spaces="drive", includeRemoved=True, ).execute() def watch_changes(self, tenant_id: int, channel_id: str, address: str, token: str, page_token: str, ttl_seconds: int) -> Dict[str, Any]: from datetime import datetime, timedelta, timezone expiration_ms = int((datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)).timestamp() * 1000) return self._service(tenant_id).changes().watch( pageToken=page_token, body={ "id": channel_id, "type": "web_hook", "address": address, "token": token, "expiration": expiration_ms, }, ).execute() def stop_channel(self, tenant_id: int, channel_id: str, resource_id: str) -> None: self._service(tenant_id).channels().stop(body={ "id": channel_id, "resourceId": resource_id, }).execute() # ── File download / export ───────────────────── def get_file_metadata(self, tenant_id: int, file_id: str) -> Optional[Dict[str, Any]]: try: return self._service(tenant_id).files().get( fileId=file_id, fields=("id,name,mimeType,parents,trashed,modifiedTime,size," "webViewLink,lastModifyingUser(emailAddress,displayName)"), ).execute() except Exception as e: if "404" in str(e) or "notFound" in str(e): return None raise def download_file(self, tenant_id: int, file_id: str, target_stream, max_bytes: int) -> int: """Download to target_stream. Aborts if cumulative bytes exceeds max_bytes. Returns total bytes downloaded. 用 target_stream.tell() 計實際寫入 bytes（穩定且不依賴 MediaDownloadProgress 跨版本行為）， 不要用 status.resumable_progress（不同 googleapiclient 版本可能 reset/None）。 """ from googleapiclient.http import MediaIoBaseDownload from jedi_common.handler.exception import BadRequestError from common.code.grc_error_code import GrcErrorCode svc = self._service(tenant_id) request = svc.files().get_media(fileId=file_id) downloader = MediaIoBaseDownload(target_stream, request) done = False while not done: _status, done = downloader.next_chunk() if target_stream.tell() > max_bytes: raise BadRequestError(GrcErrorCode.GRC_DRIVE_FILE_OVERSIZED) return target_stream.tell() def export_google_doc_link(self, web_view_link: str) -> str: """For Google native docs we don't download — we just store the webViewLink as LINK evidence.""" return web_view_link ``` - [ ] **Step 2**：補測試： - mock changes.list 結構回應 - mock channels.watch 帶 expiration - download_file 模擬超過 max_bytes 拋例外 - [ ] **Step 3**：跑 → PASS。 - [ ] **Step 4**：commit ```bash git add infra/cloud_integration/google_drive/google_drive_api_client.py test/cloud_integration/test_google_drive_api_client.py git commit -m "feat(cloud_integration): extend GoogleDriveApiClient with Changes/watch/download" ``` --- ### Task 5: WebhookChannelManager (Domain Service) **Files:** - Create: `domain/cloud_integration/service/webhook_channel_manager.py` - Create: `test/cloud_integration/test_webhook_channel_manager.py` - [ ] **Step 1**：實作（負責 register / renew / stop channel + cursor 初始化）： ```python import logging import secrets from datetime import datetime, timedelta, timezone from domain.cloud_integration.service.tenant_drive_integration_domain_service import TenantDriveIntegrationDomainService from infra.cloud_integration.google_drive.google_drive_api_client import GoogleDriveApiClient logger = logging.getLogger(__name__) CHANNEL_TTL_SECONDS = 7 * 24 * 3600 # Drive max RENEW_BEFORE_SECONDS = 24 * 3600 # renew if expires within 24h class WebhookChannelManager: def __init__( self, tenant_drive_integration_domain_service: TenantDriveIntegrationDomainService, drive_api_client: GoogleDriveApiClient, webhook_base_url: str, ): self._domain = tenant_drive_integration_domain_service self._drive = drive_api_client self._webhook_base_url = webhook_base_url def register_for_tenant(self, tenant_id: int) -> None: entity = self._domain.get_required(tenant_id) # 1. 取 startPageToken (cursor 初始) if not entity.drive_change_cursor: entity.drive_change_cursor = self._drive.get_start_page_token(tenant_id) # 2. 註冊 channel channel_id = secrets.token_urlsafe(16) token = secrets.token_urlsafe(32) address = f"{self._webhook_base_url.rstrip('/')}/api/webhooks/google-drive/{tenant_id}" resp = self._drive.watch_changes( tenant_id=tenant_id, channel_id=channel_id, address=address, token=token, page_token=entity.drive_change_cursor, ttl_seconds=CHANNEL_TTL_SECONDS, ) entity.webhook_channel_id = channel_id entity.webhook_resource_id = resp.get("resourceId") entity.webhook_token = token entity.webhook_expires_at = datetime.now(timezone.utc) + timedelta(seconds=CHANNEL_TTL_SECONDS) self._domain.update(entity) def renew_if_needed(self, tenant_id: int) -> bool: entity = self._domain.get_or_none(tenant_id) if entity is None or entity.status != "CONNECTED": return False now = datetime.now(timezone.utc) if entity.webhook_expires_at and entity.webhook_expires_at > now + timedelta(seconds=RENEW_BEFORE_SECONDS): return False # stop old channel best-effort if entity.webhook_channel_id and entity.webhook_resource_id: try: self._drive.stop_channel(tenant_id, entity.webhook_channel_id, entity.webhook_resource_id) except Exception: logger.warning("Failed to stop old channel for tenant %s", tenant_id) self.register_for_tenant(tenant_id) return True def renew_all(self) -> int: renewed = 0 # iterate all CONNECTED tenants — need a list method on domain service for tenant in self._domain.list_connected(): if self.renew_if_needed(tenant.tenant_id): renewed += 1 return renewed def stop_for_tenant(self, tenant_id: int) -> None: entity = self._domain.get_or_none(tenant_id) if entity and entity.webhook_channel_id and entity.webhook_resource_id: try: self._drive.stop_channel(tenant_id, entity.webhook_channel_id, entity.webhook_resource_id) except Exception: logger.warning("Failed to stop channel for tenant %s", tenant_id) entity.webhook_channel_id = None entity.webhook_resource_id = None entity.webhook_token = None entity.webhook_expires_at = None self._domain.update(entity) ``` > 需在 `TenantDriveIntegrationDomainService` 加 `list_connected()` 與對應 repo `list_by_status('CONNECTED')`。 - [ ] **Step 2**：在 repo / domain service 補 `list_by_status('CONNECTED')`。 - [ ] **Step 3**：寫測試覆蓋： - register_for_tenant 無 cursor → call get_start_page_token → 寫 entity - renew_if_needed 距到期 > 24h → 不 renew，回 False - renew_if_needed 距到期 < 24h → 停舊 channel + register 新 → 回 True - stop_for_tenant 清空欄位 - [ ] **Step 4**：跑 → PASS。 - [ ] **Step 5**：commit ```bash git add domain/cloud_integration/service/webhook_channel_manager.py \ domain/cloud_integration/repository/tenant_drive_integration_repository.py \ infra/cloud_integration/repository/tenant_drive_integration_repo_impl.py \ test/cloud_integration/test_webhook_channel_manager.py git commit -m "feat(cloud_integration): add WebhookChannelManager (register/renew/stop)" ``` --- ### Task 6: Connect & Disconnect 整合 Webhook Manager **Files:** - Modify: `app/cloud_integration/service/google_drive_integration_service.py` - Modify: `di_containers/cloud_integration/cloud_integration_containers.py` - [ ] **Step 1**：在 `__init__` 加注入 `webhook_channel_manager`。 - [ ] **Step 2**：`handle_callback` 在寫完 entity（status=CONNECTED）後，呼叫： ```python try: self._webhook_channel_manager.register_for_tenant(tenant_id) except Exception as e: logger.warning("Failed to register webhook channel for tenant %s: %s", tenant_id, e) # 不 raise — 連線本身已成功，channel 之後可由 renewer 補上 ``` - [ ] **Step 3**：`disconnect` 在標 disconnected 之前先呼叫： ```python try: self._webhook_channel_manager.stop_for_tenant(tenant_id) except Exception: pass ``` - [ ] **Step 4**：DI container：把 `webhook_channel_manager` 注入到 `google_drive_integration_service`。 - [ ] **Step 5**：commit ```bash git add app/cloud_integration/service/google_drive_integration_service.py di_containers/cloud_integration/cloud_integration_containers.py git commit -m "feat(cloud_integration): integrate WebhookChannelManager into connect/disconnect" ``` --- ### Task 7: Channel Renewer 排程 **Files:** - Modify: `core/scheduler.py` - [ ] **Step 1**：把 placeholder 換成實際呼叫： ```python def _renew_channels_job(): from di_containers.containers import Containers try: with app.app_context(): mgr = Containers.cloud_integration_container.webhook_channel_manager() count = mgr.renew_all() logger.info("WebhookChannelRenewer renewed %d channels", count) except Exception: logger.exception("WebhookChannelRenewer tick failed") _scheduler.add_job(_renew_channels_job, "interval", hours=24, id="webhook_channel_renewer", max_instances=1) ``` - [ ] **Step 2**：commit ```bash git add core/scheduler.py git commit -m "feat(cloud_integration): activate WebhookChannelRenewer scheduled job (daily)" ``` --- ### Task 8: Webhook Receiver Route **Files:** - Create: `api/cloud_integration/routes/google_drive_webhook_route.py` - Create: `app/cloud_integration/service/google_drive_webhook_service.py` - Modify: `api/cloud_integration/__init__.py` - Create: `test/cloud_integration/test_google_drive_webhook_route.py` - [ ] **Step 1**：app service 負責 validate + enqueue： ```python # app/cloud_integration/service/google_drive_webhook_service.py from jedi_common.handler.exception import PreconditionFailedError, NotFound from common.code.grc_error_code import GrcErrorCode from domain.cloud_integration.enums.sync_job_type import DriveSyncJobType from domain.cloud_integration.service.drive_sync_job_domain_service import DriveSyncJobDomainService from domain.cloud_integration.service.tenant_drive_integration_domain_service import TenantDriveIntegrationDomainService class GoogleDriveWebhookService: def __init__( self, tenant_drive_integration_domain_service: TenantDriveIntegrationDomainService, drive_sync_job_domain_service: DriveSyncJobDomainService, ): self._tenant = tenant_drive_integration_domain_service self._jobs = drive_sync_job_domain_service def handle_notification(self, tenant_id: int, channel_id: str, channel_token: str, resource_state: str) -> None: entity = self._tenant.get_or_none(tenant_id) if entity is None: raise NotFound(GrcErrorCode.GRC_DRIVE_INTEGRATION_NOT_FOUND) if entity.webhook_token != channel_token or entity.webhook_channel_id != channel_id: raise PreconditionFailedError(GrcErrorCode.GRC_DRIVE_WEBHOOK_TOKEN_INVALID) if resource_state == "sync": return # initial handshake # enqueue self._jobs.enqueue( tenant_id=tenant_id, job_type=DriveSyncJobType.PROCESS_DRIVE_CHANGES, payload={"trigger": "webhook"}, priority=10, ) ``` - [ ] **Step 2**：route： ```python # api/cloud_integration/routes/google_drive_webhook_route.py from dependency_injector.wiring import inject, Provide from flask import request from flask_apispec import MethodResource, doc from app.cloud_integration.service.google_drive_webhook_service import GoogleDriveWebhookService from common.util.response_util import return_response from di_containers.containers import Containers class GoogleDriveWebhookRoute(MethodResource): @doc(description="Google Drive change notification webhook", tags=["Cloud Integration"]) @inject def post( self, tenant_id: int, service: GoogleDriveWebhookService = Provide[ Containers.cloud_integration_container.google_drive_webhook_service ], ): channel_id = request.headers.get("X-Goog-Channel-ID", "") channel_token = request.headers.get("X-Goog-Channel-Token", "") resource_state = request.headers.get("X-Goog-Resource-State", "") # No JWT — auth via channel token only service.handle_notification(tenant_id, channel_id, channel_token, resource_state) return return_response(True, {"received": True}) ``` - [ ] **Step 3**：在 `api/cloud_integration/__init__.py` register webhook： ```python # Phase 1 已把 Blueprint url_prefix 改成 /api（見 Phase 1 Task 13 Step 2 設計注意） # 所以 webhook 直接用 /webhooks/... 相對路徑即可 api.add_resource(GoogleDriveWebhookRoute, "/webhooks/google-drive/") ``` > **不要嘗試 return list of Blueprints** — `main_app.py:22` 是 `app.register_blueprint(api_module.create_module())`，只接受單一 Blueprint，return list 會 crash。Phase 1 已經把 url_prefix 設成 `/api` 解決這個結構問題，這裡延用即可。 - [ ] **Step 4**：寫 E2E 測試（mock service）： - X-Goog header `tenant_id` 對應的 integration 不存在 → 404 - X-Goog-Channel-Token 不對 → 412 - X-Goog-Resource-State=`sync`（initial handshake）→ 200，不 enqueue - happy path（valid token + resource_state=`change`）→ 200，service.enqueue 被呼叫一次 - [ ] **Step 5**：commit ```bash git add app/cloud_integration/service/google_drive_webhook_service.py \ api/cloud_integration/routes/google_drive_webhook_route.py \ api/cloud_integration/__init__.py \ test/cloud_integration/test_google_drive_webhook_route.py git commit -m "feat(cloud_integration): add Drive webhook receiver + service" ``` --- ### Task 9: ProcessDriveChangesHandler **Files:** - Create: `app/cloud_integration/service/handlers/process_drive_changes_handler.py` - Create: `test/cloud_integration/test_process_drive_changes_handler.py` > **v0.2 Addendum**：以下 sample code 仍帶舊 `_find_ao_mapping_for_file()` 名稱與 `ao_mapping_uid` payload。實作時請改成： > - method 改名 `_find_task_mapping_for_file`，判斷 `mapping.scope_type == DriveScopeType.TASK` > - 進入 IMPORT enqueue 時 payload 用 `task_mapping_uid`（不是 ao_mapping_uid） > - 若檔案 parent 是 AO 容器層（非 task folder）→ 忽略並寫 admin alert（v0.2 規定 evidence 必須在 task folder） > **v0.3 Addendum (ARCHIVE ancestor SKIP)**：在 `_process_one_change` 對 file change（與 folder change）做 dispatch 之前，先呼叫新的 `_is_ancestor_archive(tenant_id, file)` helper： > ```python > def _is_ancestor_archive(self, tenant_id: int, file: dict) -> bool: > """檢查 file 的任一 parent 是否為 ARCHIVE mapping（直接 parent 即可， > 因為 archive 結構為 flat：_Archive/ 下沒有子目錄，只有 file 或 task folder）。 > 若需深度 walk，可用 cache 加速：預先載入 per-tenant ARCHIVE folder ID set。""" > parents = file.get("parents") or [] > for parent_id in parents: > mapping = self._folders.get_by_drive_folder_id(parent_id) > if mapping and mapping.scope_type == DriveScopeType.ARCHIVE: > return True > # task folder 整個被搬進 _Archive/ 的情境：file 自身為 folder，parent 是 ARCHIVE > # → 已被上面 loop 攔到。若 archive 結構未來改為樹狀，可在此擴充 recursive walk。 > return False > ``` > 在 `_process_one_change` 開頭： > ```python > if file := change.get("file"): > if self._is_ancestor_archive(tenant_id, file): > logger.debug("Skip change for file %s (ancestor=ARCHIVE)", file.get("id")) > return > ``` > 對 `removed`/`trashed` 事件也應檢查（若可推得 file metadata 含 parents），但 Drive API 對 removed change 通常不附 parents → 這時靠 `mapping.scope_type` 判斷：若 `_folders.get_by_drive_folder_id(file_id)` 對應 mapping 是 ARCHIVE 自身或位於 ARCHIVE 下，可選擇 SKIP；簡化版本可直接讓 SOFT_DELETE handler 自己處理（找不到 active evidence 自然 noop）。 > > 額外 unit test： > - 新檔被丟到 `_Archive/` 內 → SKIP，不 enqueue IMPORT > - 既有檔案被 move 到 `_Archive/`（webhook 收到 file change with parent=archive）→ 不再 enqueue IMPORT；軟刪由 DB-side 已完成（系統發起的刪除流程已在前面處理） > - 同 file 從 `_Archive/` 被拖回 active task folder → 不 SKIP，走正常 import 路徑 > **設計注意（idempotency by design）**：兩個 webhook 同時打進來會 enqueue 兩筆 PROCESS_DRIVE_CHANGES。 > `claim_next_pending FOR UPDATE SKIP LOCKED` 不會序列化「同 tenant」的 job，所以兩個 worker 可能同時跑。 > 兩個 worker 都會用同一個 cursor 拉到同一批變更，可能 enqueue 重複的 IMPORT job。 > **這是可接受的** — 因為： > 1. `job_evidences.drive_file_id` 有 UNIQUE 索引，重複 IMPORT 第二筆會在 DB 層被擋 > 2. modifiedTime 比對會把「沒變的檔案」視為 dedup 跳過 > 3. cursor 寫回是 last-write-wins，但寫回的值都一樣（同一個 newStartPageToken） > > 若日後發現重複 work 太浪費資源，可在 worker 加 per-tenant advisory lock： > `SELECT pg_try_advisory_xact_lock(hashtext('drive_changes_' || tenant_id))` > 暫不實作。 - [ ] **Step 1**：實作： ```python import logging from jedi_common.session.database.db import transaction from app.cloud_integration.service.handlers.base_job_handler import BaseJobHandler from domain.cloud_integration.enums.scope_type import DriveScopeType from domain.cloud_integration.enums.sync_job_type import DriveSyncJobType from domain.cloud_integration.service.drive_folder_mapping_domain_service import DriveFolderMappingDomainService from domain.cloud_integration.service.drive_sync_job_domain_service import DriveSyncJobDomainService from domain.cloud_integration.service.tenant_drive_integration_domain_service import TenantDriveIntegrationDomainService from infra.cloud_integration.google_drive.google_drive_api_client import GoogleDriveApiClient logger = logging.getLogger(__name__) GOOGLE_FOLDER_MIME = "application/vnd.google-apps.folder" GOOGLE_DOC_MIMES = { "application/vnd.google-apps.document", "application/vnd.google-apps.spreadsheet", "application/vnd.google-apps.presentation", "application/vnd.google-apps.form", "application/vnd.google-apps.drawing", } class ProcessDriveChangesHandler(BaseJobHandler): def __init__( self, tenant_drive_integration_domain_service: TenantDriveIntegrationDomainService, folder_mapping_domain_service: DriveFolderMappingDomainService, drive_sync_job_domain_service: DriveSyncJobDomainService, drive_api_client: GoogleDriveApiClient, ): self._tenant = tenant_drive_integration_domain_service self._folders = folder_mapping_domain_service self._jobs = drive_sync_job_domain_service self._drive = drive_api_client @property def job_type(self) -> str: return DriveSyncJobType.PROCESS_DRIVE_CHANGES def handle(self, job): tenant_id = job.tenant_id entity = self._tenant.get_required(tenant_id) cursor = entity.drive_change_cursor if not cursor: cursor = self._drive.get_start_page_token(tenant_id) while True: resp = self._drive.list_changes(tenant_id, cursor, page_size=100) for change in resp.get("changes", []): self._process_one_change(tenant_id, change) if "nextPageToken" in resp: cursor = resp["nextPageToken"] else: cursor = resp.get("newStartPageToken", cursor) break self._update_cursor(tenant_id, cursor) @transaction def _process_one_change(self, tenant_id: int, change: dict) -> None: file_id = change.get("fileId") if not file_id: return if change.get("removed") or (change.get("file") and change["file"].get("trashed")): # 刪除事件 — 排隊 SOFT_DELETE_EVIDENCE self._jobs.enqueue(tenant_id, DriveSyncJobType.SOFT_DELETE_EVIDENCE, {"drive_file_id": file_id}) # 也可能是 folder 被刪 — 標 unlinked（best-effort） self._folders.mark_unlinked_by_drive_id(file_id) return file = change["file"] # 是否 folder if file.get("mimeType") == GOOGLE_FOLDER_MIME: # 場景一：改名 — 我們忽略（系統是 source of truth） # 場景二：搬走 — 比對 parent 是否在我們管的下面 # 簡化：只在 parent 變動且我們有 mapping 時，標 unlinked + enqueue 重建 mapping = self._folders.get_by_drive_folder_id(file_id) # 需在 domain service 補 if mapping and file.get("parents"): if mapping.parent_drive_folder_id and mapping.parent_drive_folder_id not in file["parents"]: mapping.is_unlinked = True self._folders.upsert(mapping) # enqueue rebuild via CREATE_FOLDER self._jobs.enqueue( tenant_id, DriveSyncJobType.CREATE_FOLDER, { "scope_type": mapping.scope_type, "scope_uid": str(mapping.scope_uid) if mapping.scope_uid else None, "parent_drive_folder_id": mapping.parent_drive_folder_id, "name": mapping.display_name_snapshot, }, ) return # 是檔案 — 確認 parent 在我們管的 AO 資料夾裡 ao_mapping = self._find_ao_mapping_for_file(tenant_id, file) if ao_mapping is None: return # 不在我們關心的範圍 # enqueue IMPORT_DRIVE_FILE self._jobs.enqueue( tenant_id, DriveSyncJobType.IMPORT_DRIVE_FILE, { "drive_file_id": file_id, "ao_mapping_uid": str(ao_mapping.uid), "is_google_doc": file.get("mimeType") in GOOGLE_DOC_MIMES, }, ) def _find_ao_mapping_for_file(self, tenant_id, file): parents = file.get("parents", []) for parent_id in parents: mapping = self._folders.get_by_drive_folder_id(parent_id) if mapping and mapping.scope_type == DriveScopeType.AO: return mapping return None def _update_cursor(self, tenant_id: int, cursor: str) -> None: entity = self._tenant.get_required(tenant_id) entity.drive_change_cursor = cursor self._tenant.update(entity) ``` > 需在 `DriveFolderMappingDomainService` 補 `get_by_drive_folder_id` (delegate to repo). - [ ] **Step 2**：補 domain service method `get_by_drive_folder_id`。 - [ ] **Step 3**：寫測試（mock 4 個 dependencies）： - `removed` change → enqueue SOFT_DELETE - file in AO folder → enqueue IMPORT - file outside our mappings → 忽略 - folder moved → mark unlinked + enqueue CREATE_FOLDER - cursor 結尾用 newStartPageToken - [ ] **Step 4**：跑 → PASS。 - [ ] **Step 5**：commit ```bash git add app/cloud_integration/service/handlers/process_drive_changes_handler.py \ domain/cloud_integration/service/drive_folder_mapping_domain_service.py \ test/cloud_integration/test_process_drive_changes_handler.py git commit -m "feat(cloud_integration): add ProcessDriveChangesHandler (cursor-driven dispatch)" ``` --- ### Task 9.5: 補 `get_active_job_execution_by_ao_uid` Domain Method 〔v0.2 OBSOLETE — 不需執行〕 **Files:** - Modify: `domain/flow_engine/service/ext_workflow_execution_domain_service.py` (or 找出 AO ↔ job_execution 對應的合適位置) - Modify: `domain/flow_engine/repository/<相應 repo interface>` 補 query method - Modify: `infra/flow_engine/repository/<相應 repo impl>` 實作 - Create: `test/test_get_active_job_execution_by_ao_uid.py` > **⚠ v0.2 已不需要此 task**：加入 Task layer 後，Drive file 的 parent folder 即為 task folder，而 TASK mapping 的 `scope_uid` 直接 = `compliance.job_executions.uid`。Task 10 / 12 透過 `task_mapping_uid` 直接拿到 job_execution_id（一次 query），不需要從 AO 反查 active task。整個 task 9.5 跳過 — 但保留段落說明來歷。 > **背景（歷史）**：原 v0.1 假設「1 AO + 1 AP = 1 job_execution」，需從 AO uid 反查唯一 active job_execution。實際多 task 場景下這是錯的（會選錯 task）。v0.2 改為 task folder 方案後此 design assumption 失效，本 task 一併失效。 - [ ] **Step 1**：先 grep 確認是否已有類似 method（避免重複造輪子）： ```bash grep -rn "by_ao\|ao_uid\|assessment_object_uid" domain/flow_engine/ domain/grc/ grep -rn "active_job_execution\|current_job_execution" domain/ app/ ``` - **若已存在**（例如 `GrcJobDomainService.get_by_ao_uid`）→ 在 Phase 3 Task 10 / 12 直接注入該既有 service，本 Task 9.5 跳過。 - **若不存在**：繼續 Step 2。 - [ ] **Step 2**：在合適的 domain service 加 method（建議放 `ExtWorkflowExecutionDomainService`，因為跨 workflow + job_execution）： ```python def get_active_job_execution_by_ao_uid(self, ao_uid: UUID) -> Optional[JobExecutionEntity]: """根據 AO uid 找到當前 AP 週期下對應的 active job_execution。 回傳 None 表示該 AO 還沒被啟動到 workflow（或已 archive）。""" # 透過 repo 做 JOIN 查詢 return self._repo.get_active_job_execution_by_ao_uid(ao_uid) ``` - [ ] **Step 3**：在 repo interface + impl 加對應 query。impl 需 JOIN： ```sql SELECT je.* FROM compliance.job_executions je JOIN compliance.workflow_executions we ON je.workflow_execution_id = we.id JOIN public.assessment_plan_task_workflow_execution_mapping m ON m.workflow_execution_id = we.id JOIN oscal.assessment_plan_task apt ON apt.id = m.task_id WHERE apt.ao_uid = :ao_uid AND we.status NOT IN ('ARCHIVED', 'CANCELLED') -- adjust per actual enum ORDER BY we.created_at DESC LIMIT 1 ``` > **重要**：實際 table / column 名稱必須對照現有 schema 確認（grep 既有 query 找最近的範例）。 - [ ] **Step 4**：寫測試（用 db_session 建假資料，覆蓋：找得到 / 找不到 / 多版本取最新）。 - [ ] **Step 5**：跑測試 → PASS。 - [ ] **Step 6**：commit ```bash git add domain/flow_engine/ infra/flow_engine/ test/test_get_active_job_execution_by_ao_uid.py git commit -m "feat(flow_engine): add get_active_job_execution_by_ao_uid for Drive sync" ``` --- ### Task 10: ImportDriveFileHandler **Files:** - Create: `app/cloud_integration/service/handlers/import_drive_file_handler.py` - Create: `test/cloud_integration/test_import_drive_file_handler.py` > **v0.2 Addendum**：以下 sample code 是 v0.1 版本（參數 `ao_mapping_uid` + 透過 `ext_workflow_execution_domain_service.get_active_job_execution_by_ao_uid` 反查）。實作時改成： > - payload key 改成 `task_mapping_uid` > - constructor 移除 `ext_workflow_execution_domain_service` 注入 > - `handle()` 內： > ```python > task_mapping = self._folders.get_by_uid(UUID(payload["task_mapping_uid"])) > if task_mapping is None or task_mapping.scope_type != "TASK": > return # 防呆 > # task_mapping.scope_uid = job_execution.uid > job_execution = self._job_execution_domain_service.get_by_uid(task_mapping.scope_uid) > ``` > - 注入 `JobExecutionDomainService`（直接拿 task entity，無需多層 JOIN） > - 後續寫 evidence 時 `job_execution_id = job_execution.id` 不變 > **v0.3 Addendum (defensive ARCHIVE check)**：雖然 Task 9 ProcessDriveChangesHandler 已會在 dispatch 階段過濾 ancestor=ARCHIVE 的檔案，但 `ImportDriveFileHandler` 是 per-job 的 handler，可能由 RECONCILE / 手動 enqueue / 其他流程獨立觸發。為保險，handler 入口也檢查： > ```python > # 取最新 metadata 後，先檢查 archive ancestor > file = self._drive.get_file_metadata(job.tenant_id, drive_file_id) > if file is None or file.get("trashed"): > return > if self._is_ancestor_archive(job.tenant_id, file): # 同 Task 9 的 helper > logger.info("Skip import for file %s (ancestor=ARCHIVE)", drive_file_id) > return > ``` > 可把 `_is_ancestor_archive` 抽成共用 helper（例如放在 `DriveFolderMappingDomainService` 或一個 `archive_check.py` 工具模組）讓兩個 handler 共用。 > > 額外 unit test：對應檔案位於 `_Archive/` → handler.handle(job) 不寫任何 evidence、不下載檔案。 - [ ] **Step 1**：實作： ```python import io import logging from datetime import datetime from uuid import UUID from jedi_common.session.database.db import transaction from jedi_file_upload.app.service.file_upload_service import FileUploadService from werkzeug.datastructures import FileStorage from app.cloud_integration.service.handlers.base_job_handler import BaseJobHandler from common.code.grc_error_code import GrcErrorCode from domain.cloud_integration.enums.sync_job_type import DriveSyncJobType from domain.cloud_integration.service.drive_folder_mapping_domain_service import DriveFolderMappingDomainService from domain.flow_engine.entity.job_evidence_entity import JobEvidenceEntity from domain.flow_engine.service.job_evidence_domain_service import JobEvidenceDomainService from domain.flow_engine.service.ext_workflow_execution_domain_service import ExtWorkflowExecutionDomainService from infra.cloud_integration.google_drive.google_drive_api_client import GoogleDriveApiClient logger = logging.getLogger(__name__) DRIVE_SYNC_USER = "drive-sync@system" COMPLETED_JOB_STATUSES = {"COMPLETED"} # spec §10 / §11 def _parse_rfc3339(value: str): """Parse Drive API RFC3339 timestamp 為 tz-aware UTC datetime。 Drive 慣用 'Z' 結尾，Python fromisoformat 在 3.11+ 才直接支援 Z。""" from datetime import datetime return datetime.fromisoformat(value.replace("Z", "+00:00")) class ImportDriveFileHandler(BaseJobHandler): def __init__( self, folder_mapping_domain_service: DriveFolderMappingDomainService, job_evidence_domain_service: JobEvidenceDomainService, ext_workflow_execution_domain_service: ExtWorkflowExecutionDomainService, drive_api_client: GoogleDriveApiClient, file_upload_service: FileUploadService, file_size_limit_mb: int, ): self._folders = folder_mapping_domain_service self._evidence = job_evidence_domain_service self._ext_we = ext_workflow_execution_domain_service self._drive = drive_api_client self._file_upload = file_upload_service self._max_bytes = file_size_limit_mb * 1024 * 1024 @property def job_type(self) -> str: return DriveSyncJobType.IMPORT_DRIVE_FILE @transaction def handle(self, job): payload = job.payload ao_mapping = self._folders.get_by_uid(UUID(payload["ao_mapping_uid"])) drive_file_id = payload["drive_file_id"] is_google_doc = payload.get("is_google_doc", False) # 取目前最新 metadata（pageToken 可能落後） file = self._drive.get_file_metadata(job.tenant_id, drive_file_id) if file is None or file.get("trashed"): return # 已被刪，後續 SOFT_DELETE 會處理 # 找 job_execution（透過 AO uid → ext_workflow_execution → job_execution） job_execution = self._ext_we.get_active_job_execution_by_ao_uid(ao_mapping.scope_uid) if job_execution is None: logger.warning("No active job_execution for AO %s — skip file import", ao_mapping.scope_uid) return # COMPLETED 不接收 (spec §11) if job_execution.status in COMPLETED_JOB_STATUSES: logger.info("Job %s COMPLETED, skipping Drive file %s import", job_execution.uid, drive_file_id) return # 已存在 → check modifiedTime；同則跳過、新則 re-import existing = self._evidence.get_active_by_drive_file_id(drive_file_id) if existing and existing.drive_file_modified_at and file.get("modifiedTime"): # 用 datetime parse + UTC 比較，避免 isoformat string 比對因 timezone offset 格式差異而誤判 from datetime import timezone drive_modified = _parse_rfc3339(file["modifiedTime"]) # tz-aware UTC existing_modified = existing.drive_file_modified_at if existing_modified.tzinfo is None: existing_modified = existing_modified.replace(tzinfo=timezone.utc) if drive_modified == existing_modified: return # 沒變 if is_google_doc: # LINK evidence self._upsert_link_evidence(job_execution, ao_mapping, file, existing) return # Binary 檔案 — 檢查 size + 下載 size = int(file.get("size") or 0) if size > self._max_bytes: raise ValueError(f"File {drive_file_id} size {size} exceeds limit {self._max_bytes}") buf = io.BytesIO() try: total = self._drive.download_file(job.tenant_id, drive_file_id, buf, max_bytes=self._max_bytes) except ValueError: raise buf.seek(0) fs = FileStorage(stream=buf, filename=file["name"], content_type="application/octet-stream") save_dir = f"JOB_EVIDENCES/{job_execution.uid}" uploaded = self._file_upload.upload_files([fs], DRIVE_SYNC_USER, save_dir)[0] # 寫 evidence if existing: # update existing's file_id (re-import) existing.file_id = uploaded.id existing.content_hash = uploaded.checksum existing.drive_file_modified_at = _parse_rfc3339(file["modifiedTime"]) existing.drive_last_modifying_user_email = (file.get("lastModifyingUser") or {}).get("emailAddress") existing.updated_user = DRIVE_SYNC_USER self._evidence.update(existing) else: new_evidence = JobEvidenceEntity( main_workflow_execution_id=job_execution.main_workflow_execution_id, workflow_execution_id=job_execution.workflow_execution_id, job_execution_id=job_execution.id, evidence_type="FILE", source="DRIVE_SYNC", file_id=uploaded.id, drive_file_id=drive_file_id, drive_file_modified_at=_parse_rfc3339(file["modifiedTime"]), drive_last_modifying_user_email=(file.get("lastModifyingUser") or {}).get("emailAddress"), content_hash=uploaded.checksum, hash_algorithm="MD5", description=f"[Google Drive] {file['name']}", created_user=DRIVE_SYNC_USER, updated_user=DRIVE_SYNC_USER, ) self._evidence.add(new_evidence) def _upsert_link_evidence(self, job_execution, ao_mapping, file, existing): web_view_link = file.get("webViewLink") if existing: existing.reference_url = web_view_link existing.description = f"[Google Drive] {file['name']}" existing.drive_file_modified_at = _parse_rfc3339(file["modifiedTime"]) self._evidence.update(existing) else: entity = JobEvidenceEntity( main_workflow_execution_id=job_execution.main_workflow_execution_id, workflow_execution_id=job_execution.workflow_execution_id, job_execution_id=job_execution.id, evidence_type="LINK", source="DRIVE_SYNC", file_id=None, drive_file_id=file["id"], drive_file_modified_at=_parse_rfc3339(file["modifiedTime"]), drive_last_modifying_user_email=(file.get("lastModifyingUser") or {}).get("emailAddress"), reference_url=web_view_link, description=f"[Google Drive] {file['name']}", hash_algorithm="TEXT", created_user=DRIVE_SYNC_USER, updated_user=DRIVE_SYNC_USER, ) self._evidence.add(entity) ``` > 需在 `JobEvidenceDomainService` 加 `get_active_by_drive_file_id`、`update`；在 `ExtWorkflowExecutionDomainService` 加 `get_active_job_execution_by_ao_uid`（如果還沒有）。 - [ ] **Step 2**：補上述 domain service methods。 - [ ] **Step 3**：寫測試覆蓋： - 新檔 binary → 下載 + Minio + 寫 evidence - 已有 + modifiedTime 沒變 → 跳過 - 已有 + modifiedTime 變 → re-upload + update - is_google_doc → 走 LINK 路徑（不下載） - 超過 size limit → ValueError - job COMPLETED → skip 不寫 - file 已 trashed → 不做事 - [ ] **Step 4**：跑 → PASS。 - [ ] **Step 5**：commit ```bash git add app/cloud_integration/service/handlers/import_drive_file_handler.py \ domain/flow_engine/service/job_evidence_domain_service.py \ domain/flow_engine/service/ext_workflow_execution_domain_service.py \ test/cloud_integration/test_import_drive_file_handler.py git commit -m "feat(cloud_integration): add ImportDriveFileHandler (binary download + LINK for native)" ``` --- ### Task 11: SoftDeleteEvidenceHandler **Files:** - Create: `app/cloud_integration/service/handlers/soft_delete_evidence_handler.py` - Create: `test/cloud_integration/test_soft_delete_evidence_handler.py` - [ ] **Step 1**：實作： ```python import logging from jedi_common.session.database.db import transaction from app.cloud_integration.service.handlers.base_job_handler import BaseJobHandler from domain.cloud_integration.enums.sync_job_type import DriveSyncJobType from domain.flow_engine.service.job_evidence_domain_service import JobEvidenceDomainService logger = logging.getLogger(__name__) class SoftDeleteEvidenceHandler(BaseJobHandler): def __init__(self, job_evidence_domain_service: JobEvidenceDomainService): self._evidence = job_evidence_domain_service @property def job_type(self) -> str: return DriveSyncJobType.SOFT_DELETE_EVIDENCE @transaction def handle(self, job): drive_file_id = job.payload["drive_file_id"] existing = self._evidence.get_active_by_drive_file_id(drive_file_id) if existing is None: return # 沒對應 evidence，可能是無關檔案 existing.is_deleted = True existing.deleted_reason = "DRIVE_DELETED" from datetime import datetime, timezone existing.deleted_at = datetime.now(timezone.utc) self._evidence.update(existing) logger.info("Soft-deleted evidence %s (drive_file_id=%s)", existing.uid, drive_file_id) ``` - [ ] **Step 2**：測試： - 找不到 → 不報錯 - 找到 → 設 is_deleted/deleted_reason/deleted_at - [ ] **Step 3**：跑 → PASS。 - [ ] **Step 4**：commit ```bash git add app/cloud_integration/service/handlers/soft_delete_evidence_handler.py test/cloud_integration/test_soft_delete_evidence_handler.py git commit -m "feat(cloud_integration): add SoftDeleteEvidenceHandler" ``` --- ### Task 12: ReconcileTaskFolderHandler 〔v0.2 改名 — 原為 ReconcileAoFolderHandler〕 **Files:** - Create: `app/cloud_integration/service/handlers/reconcile_task_folder_handler.py` - Create: `test/cloud_integration/test_reconcile_task_folder_handler.py` > **v0.2 Addendum**：原 v0.1 為 AO-level reconcile（payload `ao_uid`），需要透過 `ExtWorkflowExecutionDomainService` 從 AO 反查 active task。v0.2 改為 task-level reconcile： > - 檔名 / class 名 / `job_type` enum 全部改為 `RECONCILE_TASK_FOLDER` > - Payload 改成 `{ tenant_id, task_uid }`（task_uid = job_execution.uid） > - handler 內直接 `self._folders.get_by_scope(tenant_id, "TASK", task_uid)` 拿 task folder mapping → list_children → 對帳 > - 不用注入 `ExtWorkflowExecutionDomainService`；改注入 `JobExecutionDomainService`（用 `task_uid` 拿 task entity 取 `id` 用於 evidence 查詢） > - AO 級別的 reconcile（如 user 在 admin 介面點「對帳整個 AO」）由「對該 AO 下所有 task 各 enqueue 一次 RECONCILE_TASK_FOLDER」實作，不需要新 handler > > 以下 sample code 是 v0.1 寫法，實作時請依上述指引重構。 - [ ] **Step 1**：實作（用於 job revert / 重建後對帳）： ```python import logging from uuid import UUID from jedi_common.session.database.db import transaction from app.cloud_integration.service.handlers.base_job_handler import BaseJobHandler from domain.cloud_integration.enums.scope_type import DriveScopeType from domain.cloud_integration.enums.sync_job_type import DriveSyncJobType from domain.cloud_integration.service.drive_folder_mapping_domain_service import DriveFolderMappingDomainService from domain.cloud_integration.service.drive_sync_job_domain_service import DriveSyncJobDomainService from domain.flow_engine.service.job_evidence_domain_service import JobEvidenceDomainService from infra.cloud_integration.google_drive.google_drive_api_client import GoogleDriveApiClient from domain.flow_engine.service.ext_workflow_execution_domain_service import ExtWorkflowExecutionDomainService logger = logging.getLogger(__name__) GOOGLE_FOLDER_MIME = "application/vnd.google-apps.folder" GOOGLE_DOC_MIMES = { "application/vnd.google-apps.document", "application/vnd.google-apps.spreadsheet", "application/vnd.google-apps.presentation", } class ReconcileAoFolderHandler(BaseJobHandler): def __init__( self, folder_mapping_domain_service: DriveFolderMappingDomainService, ext_workflow_execution_domain_service: ExtWorkflowExecutionDomainService, job_evidence_domain_service: JobEvidenceDomainService, drive_sync_job_domain_service: DriveSyncJobDomainService, drive_api_client: GoogleDriveApiClient, ): self._folders = folder_mapping_domain_service self._ext_we = ext_workflow_execution_domain_service self._evidence = job_evidence_domain_service self._jobs = drive_sync_job_domain_service self._drive = drive_api_client @property def job_type(self) -> str: return DriveSyncJobType.RECONCILE_AO_FOLDER def handle(self, job): ao_uid = UUID(job.payload["ao_uid"]) mapping = self._folders.get_by_scope(job.tenant_id, DriveScopeType.AO, ao_uid) if mapping is None or mapping.is_unlinked: logger.info("AO %s has no live mapping, skip reconcile", ao_uid) return job_execution = self._ext_we.get_active_job_execution_by_ao_uid(ao_uid) if job_execution is None: return # 1. 列 Drive 上目前所有檔案（in_parents=mapping.drive_folder_id） children = self._drive.list_children(job.tenant_id, mapping.drive_folder_id) drive_file_ids = {c["id"] for c in children if c.get("mimeType") != GOOGLE_FOLDER_MIME} # 2. 對每個 Drive file → 確認 evidence 存在；不存在 → enqueue IMPORT for child in children: if child.get("mimeType") == GOOGLE_FOLDER_MIME: continue ev = self._evidence.get_active_by_drive_file_id(child["id"]) if ev is None: self._jobs.enqueue( job.tenant_id, DriveSyncJobType.IMPORT_DRIVE_FILE, { "drive_file_id": child["id"], "ao_mapping_uid": str(mapping.uid), "is_google_doc": child.get("mimeType") in GOOGLE_DOC_MIMES, }, ) # 3. 對 DB 內每筆 DRIVE_SYNC active evidence → 若 drive_file_id 不在 Drive 列表 → 軟刪 db_evidences = self._evidence.list_active_by_job_execution_and_source(job_execution.id, "DRIVE_SYNC") for ev in db_evidences: if ev.drive_file_id and ev.drive_file_id not in drive_file_ids: self._jobs.enqueue( job.tenant_id, DriveSyncJobType.SOFT_DELETE_EVIDENCE, {"drive_file_id": ev.drive_file_id}, ) ``` - [ ] **Step 2**：測試覆蓋 - mapping 不存在 / unlinked → return - drive 有 / db 沒 → enqueue IMPORT - drive 沒 / db 有 → enqueue SOFT_DELETE - drive 有 folder → 不算檔案 - [ ] **Step 3**：跑 → PASS。 - [ ] **Step 4**：commit ```bash git add app/cloud_integration/service/handlers/reconcile_ao_folder_handler.py test/cloud_integration/test_reconcile_ao_folder_handler.py git commit -m "feat(cloud_integration): add ReconcileAoFolderHandler" ``` --- ### Task 13: DI 補齊 4 個 handler + webhook service **Files:** - Modify: `di_containers/cloud_integration/cloud_integration_containers.py` - [ ] **Step 1**：providers append： ```python webhook_channel_manager = providers.Singleton( WebhookChannelManager, tenant_drive_integration_domain_service=tenant_drive_integration_domain_service, drive_api_client=drive_api_client, webhook_base_url=config.DRIVE_WEBHOOK_PUBLIC_BASE_URL, # 與 config/config.py + spec env var 一致 ) google_drive_webhook_service = providers.Factory( GoogleDriveWebhookService, tenant_drive_integration_domain_service=tenant_drive_integration_domain_service, drive_sync_job_domain_service=drive_sync_job_domain_service, ) process_drive_changes_handler = providers.Factory( ProcessDriveChangesHandler, tenant_drive_integration_domain_service=tenant_drive_integration_domain_service, folder_mapping_domain_service=drive_folder_mapping_domain_service, drive_sync_job_domain_service=drive_sync_job_domain_service, drive_api_client=drive_api_client, ) import_drive_file_handler = providers.Factory( ImportDriveFileHandler, folder_mapping_domain_service=drive_folder_mapping_domain_service, job_evidence_domain_service=..., # 從 flow_engine container 引入 ext_workflow_execution_domain_service=..., drive_api_client=drive_api_client, file_upload_service=..., # 從 upload_file container 引入 file_size_limit_mb=config.DRIVE_FILE_SIZE_LIMIT_MB, ) soft_delete_evidence_handler = providers.Factory( SoftDeleteEvidenceHandler, job_evidence_domain_service=..., ) reconcile_task_folder_handler = providers.Factory( ReconcileTaskFolderHandler, # v0.2: 改名 folder_mapping_domain_service=drive_folder_mapping_domain_service, job_execution_domain_service=..., # v0.2: 不再用 ext_workflow_execution job_evidence_domain_service=..., drive_sync_job_domain_service=drive_sync_job_domain_service, drive_api_client=drive_api_client, ) # 更新 worker handlers list drive_sync_worker = providers.Singleton( DriveSyncWorker, drive_sync_job_domain_service=drive_sync_job_domain_service, handlers=providers.List( init_project_folders_handler, create_folder_handler, rename_folder_handler, process_drive_changes_handler, import_drive_file_handler, soft_delete_evidence_handler, reconcile_task_folder_handler, # v0.2: 改名 ), ) ``` - [ ] **Step 2**：在 `Containers` wire `flow_engine_container` / `upload_file_container` 進 `cloud_integration_container`。 - [ ] **Step 3**：**不需手動 from_dict**。`core/app_factory.py:141` 已自動 `container.config.from_dict(app.config)` 把全部 app.config 灌進 DI。 確認 `config/config.py` 的 BaseConfig 內有： ```python DRIVE_WEBHOOK_PUBLIC_BASE_URL = os.getenv("DRIVE_WEBHOOK_PUBLIC_BASE_URL", "") DRIVE_FILE_SIZE_LIMIT_MB = int(os.getenv("DRIVE_FILE_SIZE_LIMIT_MB", "20")) ``` （Phase 1 Pre-Implementation 已加；此處只是 sanity check） - [ ] **Step 4**：啟動後端確認 wire 無誤。 - [ ] **Step 5**：commit ```bash git add di_containers/ main_app.py git commit -m "feat(cloud_integration): wire all 7 handlers + webhook service into worker" ``` --- ### Task 14: Job Evidence Service — Delete (with Drive Archive) + Response Source **Files:** - Modify: `app/flow_engine/service/job_evidence_service.py` - Modify: `api/flow_engine/serializers/flow_engine/job_evidence.py` - Create: `test/test_job_evidence_drive_archive.py`（v0.3 改名，原 `test_job_evidence_drive_guard.py`） > **v0.3 Addendum (DRIVE_SYNC delete 改為 archive，不再禁刪)**：原 v0.1 / v0.2 在 delete service 內 raise `GRC_FORBIDDEN_DELETE_DRIVE_EVIDENCE` 阻擋 DRIVE_SYNC evidence 刪除。v0.3 改為：DRIVE_SYNC 也允許刪，刪之前先呼叫 orchestration `try_archive_drive_file(evidence)` 把 Drive 檔案搬到 `_Archive/`。Archive 失敗仍允許 DB 軟刪 commit（best-effort）。 - [ ] **Step 1**：在 `delete_job_evidence` 改為（v0.3 版本）： ```python @transaction def delete_job_evidence(self, uid: str) -> bool: job_evidence = self.job_evidence_domain_service.get_by_uid(uid) if not job_evidence: raise NotFound(ErrorCode.JOB_EVIDENCE_NOT_FOUND) # v0.3: 不再 raise GRC_FORBIDDEN_DELETE_DRIVE_EVIDENCE if job_evidence.source == "DRIVE_SYNC": # Best-effort: 把 Drive 檔案 move 到 AP 的 _Archive/ self._drive_sync_orchestration_service.try_archive_drive_file(job_evidence) # ...原 DB 軟刪邏輯不變（is_deleted=True, deleted_reason=USER_DELETED, ...） ``` - 在 `JobEvidenceService.__init__` 注入 `drive_sync_orchestration_service`（透過 DI container wiring） - [ ] **Step 2**：在 `add_job_evidence` 確保新增 evidence 預設 `source="SYSTEM_UPLOAD"`（只要 entity 預設值正確即可，service 程式碼不一定需要改）。 - [ ] **Step 3**：在 response serializer 加 `source` / `drive_file_id` / `drive_url`： ```python source = fields.String() drive_file_id = fields.String(allow_none=True) drive_url = fields.Method("get_drive_url", allow_none=True) def get_drive_url(self, obj): if obj.get("drive_file_id"): return f"https://drive.google.com/file/d/{obj['drive_file_id']}/view" return None ``` - [ ] **Step 4**：DTO 加 `source` / `drive_file_id` 欄位。 - [ ] **Step 5**：寫測試（v0.3 修改）： - DELETE 一筆 SYSTEM evidence → 200，orchestration 不被呼叫 - DELETE 一筆 DRIVE_SYNC evidence（Drive 連線正常）→ 200；mock `try_archive_drive_file` 被呼叫一次；DB 軟刪欄位被設定 - DELETE 一筆 DRIVE_SYNC evidence（orchestration raise）→ 仍 200（best-effort 吞掉）；DB 軟刪 commit - GET response 含 source / drive_url - [ ] **Step 6**：跑 → PASS。 - [ ] **Step 7**：commit ```bash git add app/flow_engine/service/job_evidence_service.py api/flow_engine/serializers/flow_engine/job_evidence.py test/test_job_evidence_drive_archive.py git commit -m "feat(job_evidence): allow delete of DRIVE_SYNC by archiving on Drive (v0.3)" ``` --- ### Task 14.5 (v0.3): Archive APIs in DriveSyncOrchestrationService **Files:** - Modify: `app/cloud_integration/service/drive_sync_orchestration_service.py` - Create: `test/cloud_integration/test_drive_sync_orchestration_archive.py` > v0.3 新增兩個 orchestration method 支援 archive 機制。實作放在 Phase 3（仰賴 Phase 2 v0.3 Addendum 加的 `GoogleDriveApiClient.move_to_archive`，且需與 Phase 3 ancestor-skip 邏輯一起測）。 - [ ] **Step 1**：實作 `try_archive_drive_file(evidence)`： ```python def try_archive_drive_file(self, evidence): """Best-effort 把 evidence 對應的 Drive 檔案 move 到所屬 AP 的 _Archive/。 用於 single-evidence delete (Task 14)。失敗只 log。""" try: if evidence.source != "DRIVE_SYNC" or not evidence.drive_file_id: return tenant_id = self._resolve_tenant_id_for_evidence(evidence) tenant_int = self._tenant_drive_integration_domain_service.get_or_none(tenant_id) if not tenant_int or tenant_int.status != "CONNECTED": return # 找 task mapping → 推 AP uid → 找 ARCHIVE mapping task_mapping = self._drive_folder_mapping_domain_service.get_by_scope( tenant_id=tenant_id, scope_type="TASK", scope_uid=self._job_execution_domain_service.get_by_id(evidence.job_execution_id).uid, ) if task_mapping is None: return ap_uid = self._resolve_ap_uid_for_task(task_mapping.scope_uid) archive_mapping = self._ensure_archive_folder(tenant_id, ap_uid) # Drive move self._drive_api_client.move_to_archive( tenant_id=tenant_id, file_id=evidence.drive_file_id, archive_folder_id=archive_mapping.drive_folder_id, current_parent_id=task_mapping.drive_folder_id, ) except Exception: logger.warning("Failed to archive Drive file for evidence %s", evidence.uid) ``` - [ ] **Step 2**：實作 `try_archive_task_folder(tenant_id, task_uid)`（已在 Phase 2 Task 20.6 v0.3 Addendum 給 sample code）— 把整個 task folder move 到 `_Archive/` + 標 mapping unlinked。 - [ ] **Step 3**：實作 `_ensure_archive_folder(tenant_id, ap_uid)` helper： ```python def _ensure_archive_folder(self, tenant_id, ap_uid): """取得對應 AP 的 ARCHIVE mapping；若不存在或 unlinked，重建 _Archive/ folder。""" mapping = self._drive_folder_mapping_domain_service.get_by_scope( tenant_id=tenant_id, scope_type="ARCHIVE", scope_uid=ap_uid, ) if mapping and not mapping.is_unlinked: return mapping # 找 AP folder mapping 當 parent ap_mapping = self._drive_folder_mapping_domain_service.get_by_scope( tenant_id=tenant_id, scope_type="AP", scope_uid=ap_uid, ) if ap_mapping is None: raise RuntimeError(f"AP mapping not found for ap_uid={ap_uid}") # Drive create + share drive = self._drive_api_client.create_folder(tenant_id, "_Archive", parent_id=ap_mapping.drive_folder_id) self._drive_api_client.share_anyone_writer(tenant_id, drive["id"]) # Upsert mapping new_mapping = DriveFolderMappingEntity( tenant_id=tenant_id, scope_type="ARCHIVE", scope_uid=ap_uid, parent_drive_folder_id=ap_mapping.drive_folder_id, drive_folder_id=drive["id"], display_name_snapshot="_Archive", ) return self._drive_folder_mapping_domain_service.upsert(new_mapping) ``` - [ ] **Step 4**：寫測試： - `try_archive_drive_file` happy path → `move_to_archive` 被呼叫一次 - `try_archive_drive_file` Drive API raise → silently log，不重新 raise - `try_archive_task_folder` happy path → folder move + mapping unlinked - `_ensure_archive_folder` mapping 已存在 → 直接回，不呼叫 Drive create - `_ensure_archive_folder` mapping 不存在 → create + upsert - `_ensure_archive_folder` mapping 存在但 is_unlinked=True → 重建 - [ ] **Step 5**：跑 → PASS。 - [ ] **Step 6**：commit ```bash git add app/cloud_integration/service/drive_sync_orchestration_service.py test/cloud_integration/test_drive_sync_orchestration_archive.py git commit -m "feat(cloud_integration): add archive APIs (try_archive_drive_file / try_archive_task_folder) (v0.3)" ``` --- ### Task 15: Job Revert Hook → enqueue RECONCILE **Files:** - Modify: `app/flow_engine/service/workflow_execution_service.py` - [ ] **Step 1**：grep 找 revert flow： ```bash grep -n "def revert" app/flow_engine/service/workflow_execution_service.py ``` - [ ] **Step 2**：在 revert 結束（job 從 COMPLETED → PROCESSING 後）enqueue（v0.2: 改用 task_uid）： ```python try: tenant_int = self._tenant_drive_integration_domain_service.get_or_none(tenant_id) if tenant_int and tenant_int.status == "CONNECTED": self._drive_sync_job_domain_service.enqueue( tenant_id=tenant_id, job_type=DriveSyncJobType.RECONCILE_TASK_FOLDER, payload={"task_uid": str(job_execution_uid)}, # v0.2: task_uid (= job_execution.uid) priority=20, ) except Exception: logger.warning("Failed to enqueue RECONCILE for revert job %s", job_execution_uid) ``` - [ ] **Step 3**：在 service `__init__` 注入兩個 domain services（必要時透過 container 補 dependency）。 - [ ] **Step 4**：commit ```bash git add app/flow_engine/service/workflow_execution_service.py di_containers/ git commit -m "feat(cloud_integration): enqueue RECONCILE_AO_FOLDER on job revert" ``` --- ### Task 16: 前端 — Evidence UI 加 source badge / Drive 連結 / 禁刪 **Files:** - Modify: `src/components/grc/AuditControlRef.vue` - Modify: `src/components/grc/JobExecutionDrawer.vue` - [ ] **Step 1**：在 evidence map / template 加 source 判斷： ```vue
{{ ev.file_name || ev.description }} {{ formatDate(ev.uploaded_at) }}
``` - [ ] **Step 2**：**決定走「在 evidence list response 內附帶 `task_drive_folder_url` + `ao_drive_folder_url`」這條路**（v0.2: 兩個 URL 都需要 — task 是 user 上傳檔案的入口；AO 是容器層連結）。 後端：在 `GET /job-evidences?job_execution_uid=...` 的 service 層回應 envelope 內加 `meta` 欄位： ```python # app/flow_engine/service/job_evidence_service.py @transaction def get_job_evidences_with_meta(self, job_execution_uid: str) -> dict: job_execution = self.job_execution_domain_service.get_job_execution(...) evidences = self.job_evidence_domain_service.get_by_job_execution_id(job_execution.id) task_url = self._resolve_drive_folder_url("TASK", job_execution.uid) # v0.2: 主要 URL ao_url = self._resolve_drive_folder_url("AO", job_execution.ao_uid) # 容器層 URL return { "data": JobEvidenceDTO.from_entity_list(evidences), "meta": { "task_drive_folder_url": task_url, "ao_drive_folder_url": ao_url, }, } def _resolve_drive_folder_url(self, scope_type, scope_uid): if scope_uid is None: return None mapping = self._drive_folder_mapping_domain_service.get_by_scope( tenant_id=current_tenant_id(), scope_type=scope_type, scope_uid=scope_uid, ) if mapping is None or mapping.is_unlinked: return None return f"https://drive.google.com/drive/folders/{mapping.drive_folder_id}" ``` Route 層相應改為呼叫新 method 並回 `{data, meta}` 結構。 前端： - `JobExecutionDrawer.vue`（單一 task 視角）：用 `meta.task_drive_folder_url` 顯示「📁 此任務的 Drive 資料夾」按鈕 - `AuditControlRef.vue`（AO 視角）：用 `meta.ao_drive_folder_url` 顯示容器層連結；evidence 若依 task 分組，每組用各自 task 的 URL（需另外查或從 evidence record 帶出） - [ ] **Step 3**：對應修改 i18n key（如 'evidence.from_drive', 'evidence.from_system', 'evidence.open_in_drive'）。 - [ ] **Step 4**：兩個 component 都要改（或抽出共用 `EvidenceListItem.vue`）。 - [ ] **Step 5**：commit (前端) ```bash git add src/components/grc/ src/lang/ git commit -m "feat(grc): show evidence source badge + Drive link + disable delete for Drive evidence" ``` --- ### Task 17: Playwright 驗收（end-to-end Drive sync） - [ ] **Step 1**：環境準備 - 後端 dev server，並確認 webhook public URL 可被 Google 打到（Cloudflare tunnel / ngrok） - 前端 dev server - 已連 Drive（前提：Phase 1） - 已啟動一個 project 並完成 INIT_PROJECT_FOLDERS（前提：Phase 2） - [ ] **Step 2**：Playwright 流程 1. 找一個 AO 的 Drive 資料夾 URL，在瀏覽器開 2. 上傳一個 PDF（< 20MB）到該資料夾 3. 等 30 秒（webhook + worker 處理時間） 4. 切回系統 → 進對應 AO → 截圖確認看到 Drive evidence + [📁 Drive] badge + 「在 Drive 開啟」按鈕 5. 點下載 → 確認檔案下載成功（從 Minio） 6. 在 Drive 上刪除該檔案 7. 等 30 秒 8. 系統重整 → 截圖確認 evidence 消失（軟刪除） - [ ] **Step 3**：另一輪測試 1. 上傳一個 Google Doc 連結到該 AO 資料夾（可以用 Drive 介面 New → Google Docs） 2. 等 sync 3. 系統內 evidence 區應出現 LINK type evidence + 點開連到 Drive Doc - [ ] **Step 4**：oversize 測試 1. 上傳 > 20MB 檔案到 AO 資料夾 2. 等 sync 3. 系統內**不應**出現該 evidence（worker mark FAILED） 4. admin 在「整合設定 → 失敗紀錄」看到對應 FAILED job - [ ] **Step 5**：revert 測試（若 revert UI 已存在） 1. complete 一個 job 2. 在 Drive 上加一個檔案到對應 AO 資料夾 3. 等 sync → 系統內**不應**出現新 evidence（job COMPLETED skip） 4. revert job → enqueue RECONCILE → 等執行 → 系統內出現之前略過的檔案 - [ ] **Step 6**：截圖整理放 changelog / PR description。 --- ### Task 18: Changelog **Files:** - Create: `docs/changelog/-google-drive-to-system-sync.md` - [ ] **Step 1**：寫 changelog（Phase 3 內容：webhook、changes pipeline、4 handler、job_evidences 加欄位、UI 改造、revert hook）。 - [ ] **Step 2**：commit。 --- ## Phase 3 驗收 Checklist - [ ] `job_evidences` 新欄位 migration 跑過 dev DB - [ ] Connect 流程會自動 register webhook channel - [ ] Disconnect 會 stop channel - [ ] APScheduler 每天執行 channel renewer - [ ] Webhook receiver 收到推送後會 enqueue PROCESS_DRIVE_CHANGES - [ ] PROCESS_DRIVE_CHANGES 正確分派 IMPORT / SOFT_DELETE / 重建 folder jobs（v0.2: 只接受 parent=TASK folder 的檔案） - [ ] IMPORT_DRIVE_FILE 對 binary 檔下載 + 寫 Minio + 寫 evidence (source=DRIVE_SYNC, job_execution_id 從 task mapping 直取) - [ ] IMPORT 對 Google Doc 走 LINK evidence - [ ] SOFT_DELETE_EVIDENCE 在 Drive 刪檔後正確軟刪 - [ ] RECONCILE_TASK_FOLDER（v0.2 改名）補抓新檔 + 軟刪缺檔 - [ ] DELETE evidence API 對 DRIVE_SYNC：呼叫 orchestration `try_archive_drive_file`（best-effort）→ DB 軟刪 commit；不再回 403（v0.3 deprecated `GRC_FORBIDDEN_DELETE_DRIVE_EVIDENCE`） - [ ] Archive folder（`_Archive/`）每 AP 一個，由 `INIT_PROJECT_FOLDERS` 建立並登錄為 `scope_type=ARCHIVE` - [ ] Sync pipeline 對 ancestor=ARCHIVE 的 file/folder change 正確 SKIP（不重新 import） - [ ] User 把 archived 檔案手動拖回 active task folder → 視為新 evidence 寫入（不嘗試 un-delete 原 row） - [ ] 前端 evidence 區有 source badge + Drive 連結 + 禁刪按鈕 - [ ] Task 區塊頂部有「此任務的 Drive 資料夾」連結；AO 區塊頂部有「此 AO 的 Drive 容器」連結 - [ ] 預覽功能對 DRIVE_SYNC 檔案正常運作（與系統上傳一致流程） - [ ] 超過 size limit 的 Drive 檔案不 import + admin 看得到 FAILED - [ ] Job COMPLETED 後 Drive 加檔不 sync - [ ] Revert 後 RECONCILE 補抓 - [ ] 所有 unit test + handler test PASS - [ ] Playwright 驗收完成（截圖） - [ ] Changelog 寫好 --- ## Phase 1 + 2 + 3 全部完成 → Final Demo - 用 Playwright 走一遍完整 user story： 1. Admin 連 Drive 2. PM 啟動 project → Drive 出現完整資料夾結構 3. 改一個 control 名稱 → Drive 同步改名 4. User 在 Drive 上傳 evidence 到 AO 資料夾 → 系統出現 evidence 5. User 在系統 UI 上看到 Drive badge + Drive 連結 6. User 在 Drive 刪檔 → 系統 evidence 消失 - 把每一步截圖收集到 PR description。