For agentic workers: REQUIRED SUB-SKILL: Use
superpowers:subagent-driven-development(recommended) orsuperpowers:executing-plansto implement this plan task-by-task.
Spec: design.md (§4.2, §7, §8, §10, §11, §12.3, §13.3) Master Index: 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/<pid>/job/<job_uid>觸發)：Phase 2 Task 20.6 hook 會直接 SQL UPDATEdrive_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_EVIDENCEdeprecated 但保留定義。- 新 Archive APIs（位於 Task 14 之後新段落）：
DriveSyncOrchestrationService.try_archive_drive_file(evidence)— best-effort 把 evidence 對應的 Drive 檔案 move 到所屬 AP 的_Archive/；對應 single-evidence deleteDriveSyncOrchestrationService.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
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 managersrc/components/grc/
├── AuditControlRef.vue # 修改：evidence 列表加 source badge / Drive 連結 / 禁刪 DRIVE
├── JobExecutionDrawer.vue # 同上
└── EvidenceListItem.vue (optional 抽 component) # 視重構決定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.pyjob_evidences 新欄位Files:
scripts/sql/<YYYY-MM-DD>-job-evidences-add-drive-source.sqlv0.2 Addendum：原 Phase 2 migration（
2026-04-24-google-drive-folder-mappings.sql）的chk_drive_sync_jobs_typeCHECK constraint 含'RECONCILE_AO_FOLDER'。Phase 3 改名為RECONCILE_TASK_FOLDER後，需在本 phase 的 migration 一併 ALTER：-- 5. 重新命名 RECONCILE job type (<YYYY-MM-DD>) 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';
-- Date: <YYYY-MM-DD>
-- Purpose: job_evidences 加 Drive 同步相關欄位 + soft delete
-- 1. 新增欄位 (<YYYY-MM-DD>)
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 (<YYYY-MM-DD>)
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 允許多筆）(<YYYY-MM-DD>)
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. 一般查詢索引 (<YYYY-MM-DD>)
CREATE INDEX IF NOT EXISTS idx_job_evidences_source_active
ON compliance.job_evidences (job_execution_id, source) WHERE is_deleted = FALSE;git add scripts/sql/<YYYY-MM-DD>-job-evidences-add-drive-source.sql
git commit -m "feat(job_evidence): add Drive sync columns + soft delete to job_evidences"Files:
infra/flow_engine/models/job_evidence.pydomain/flow_engine/entity/job_evidence_entity.pyinfra/flow_engine/mapper/job_evidence_mapper.pysource = 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))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"Files:
domain/flow_engine/repository/job_evidence_repository.py (or i_*)infra/flow_engine/repository/job_evidence_repo_impl.pytest/test_job_evidence_repo_drive.py@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]: ...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"Files:
infra/cloud_integration/google_drive/google_drive_api_client.pytest/cloud_integration/test_google_drive_api_client.py# ── 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_linkgit 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"Files:
domain/cloud_integration/service/webhook_channel_manager.pytest/cloud_integration/test_webhook_channel_manager.pyimport 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()與對應 repolist_by_status('CONNECTED')。
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)"Files:
app/cloud_integration/service/google_drive_integration_service.pydi_containers/cloud_integration/cloud_integration_containers.pytry:
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 補上try:
self._webhook_channel_manager.stop_for_tenant(tenant_id)
except Exception:
passgit 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"Files:
core/scheduler.pydef _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)git add core/scheduler.py
git commit -m "feat(cloud_integration): activate WebhookChannelRenewer scheduled job (daily)"Files:
api/cloud_integration/routes/google_drive_webhook_route.pyapp/cloud_integration/service/google_drive_webhook_service.pyapi/cloud_integration/__init__.pytest/cloud_integration/test_google_drive_webhook_route.py# 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,
)# 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})# Phase 1 已把 Blueprint url_prefix 改成 /api（見 Phase 1 Task 13 Step 2 設計注意）
# 所以 webhook 直接用 /webhooks/... 相對路徑即可
api.add_resource(GoogleDriveWebhookRoute, "/webhooks/google-drive/<int:tenant_id>")不要嘗試 return list of Blueprints —
main_app.py:22是app.register_blueprint(api_module.create_module())，只接受單一 Blueprint，return list 會 crash。Phase 1 已經把 url_prefix 設成/api解決這個結構問題，這裡延用即可。
tenant_id 對應的 integration 不存在 → 404sync（initial handshake）→ 200，不 enqueuechange）→ 200，service.enqueue 被呼叫一次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"Files:
app/cloud_integration/service/handlers/process_drive_changes_handler.pytest/cloud_integration/test_process_drive_changes_handler.pyv0.2 Addendum：以下 sample code 仍帶舊
_find_ao_mapping_for_file()名稱與ao_mapping_uidpayload。實作時請改成：
- 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：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開頭：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。 這是可接受的 — 因為：
job_evidences.drive_file_id有 UNIQUE 索引，重複 IMPORT 第二筆會在 DB 層被擋- modifiedTime 比對會把「沒變的檔案」視為 dedup 跳過
- cursor 寫回是 last-write-wins，但寫回的值都一樣（同一個 newStartPageToken）
若日後發現重複 work 太浪費資源，可在 worker 加 per-tenant advisory lock：
SELECT pg_try_advisory_xact_lock(hashtext('drive_changes_' || tenant_id))暫不實作。
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).
removed change → enqueue SOFT_DELETEgit 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)"get_active_job_execution_by_ao_uid Domain Method 〔v0.2 OBSOLETE — 不需執行〕Files:
domain/flow_engine/service/ext_workflow_execution_domain_service.py (or 找出 AO ↔︎ job_execution 對應的合適位置)domain/flow_engine/repository/<相應 repo interface> 補 query methodinfra/flow_engine/repository/<相應 repo impl> 實作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 一併失效。
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 跳過。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)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 找最近的範例）。
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"Files:
app/cloud_integration/service/handlers/import_drive_file_handler.pytest/cloud_integration/test_import_drive_file_handler.pyv0.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()內：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 入口也檢查：# 取最新 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、不下載檔案。
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（如果還沒有）。
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)"Files:
app/cloud_integration/service/handlers/soft_delete_evidence_handler.pytest/cloud_integration/test_soft_delete_evidence_handler.pyimport 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)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"Files:
app/cloud_integration/service/handlers/reconcile_task_folder_handler.pytest/cloud_integration/test_reconcile_task_folder_handler.pyv0.2 Addendum：原 v0.1 為 AO-level reconcile（payload
ao_uid），需要透過ExtWorkflowExecutionDomainService從 AO 反查 active task。v0.2 改為 task-level reconcile：
- 檔名 / class 名 /
job_typeenum 全部改為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 寫法，實作時請依上述指引重構。
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},
)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"Files:
di_containers/cloud_integration/cloud_integration_containers.pywebhook_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: 改名
),
)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）
git add di_containers/ main_app.py
git commit -m "feat(cloud_integration): wire all 7 handlers + webhook service into worker"Files:
app/flow_engine/service/job_evidence_service.pyapi/flow_engine/serializers/flow_engine/job_evidence.pytest/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 也允許刪，刪之前先呼叫 orchestrationtry_archive_drive_file(evidence)把 Drive 檔案搬到_Archive/。Archive 失敗仍允許 DB 軟刪 commit（best-effort）。
@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）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 Nonetry_archive_drive_file 被呼叫一次；DB 軟刪欄位被設定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)"Files:
app/cloud_integration/service/drive_sync_orchestration_service.pytest/cloud_integration/test_drive_sync_orchestration_archive.pyv0.3 新增兩個 orchestration method 支援 archive 機制。實作放在 Phase 3（仰賴 Phase 2 v0.3 Addendum 加的
GoogleDriveApiClient.move_to_archive，且需與 Phase 3 ancestor-skip 邏輯一起測）。
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)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)try_archive_drive_file happy path → move_to_archive 被呼叫一次try_archive_drive_file Drive API raise → silently log，不重新 raisetry_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 → 重建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)"Files:
app/flow_engine/service/workflow_execution_service.pygrep -n "def revert" app/flow_engine/service/workflow_execution_service.pytry:
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)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"Files:
src/components/grc/AuditControlRef.vuesrc/components/grc/JobExecutionDrawer.vue<div v-for="ev in evidenceList" :key="ev.uid" class="flex align-items-center gap-2 py-2">
<Tag v-if="ev.source === 'DRIVE_SYNC'" value="Drive" severity="info" icon="pi pi-cloud" />
<Tag v-else value="系統" severity="secondary" icon="pi pi-database" />
<span class="font-medium">{{ ev.file_name || ev.description }}</span>
<span class="text-xs text-color-secondary ml-2">{{ formatDate(ev.uploaded_at) }}</span>
<div class="ml-auto flex gap-1">
<Button v-if="ev.evidence_type === 'FILE'" icon="pi pi-download" text rounded
v-tooltip.top="'下載'" @click="downloadEvidence(ev)" />
<Button v-if="ev.evidence_type === 'FILE'" icon="pi pi-eye" text rounded
v-tooltip.top="'預覽'" @click="previewEvidence(ev)" />
<a v-if="ev.drive_url" :href="ev.drive_url" target="_blank" rel="noopener">
<Button icon="pi pi-external-link" text rounded v-tooltip.top="'在 Google Drive 開啟'" />
</a>
<Button v-if="ev.source !== 'DRIVE_SYNC' && canDelete"
icon="pi pi-trash" text rounded severity="danger"
v-tooltip.top="'刪除'" @click="deleteEvidence(ev)" />
</div>
</div>後端：在 GET /job-evidences?job_execution_uid=... 的 service 層回應 envelope 內加 meta 欄位：
# 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 帶出）git add src/components/grc/ src/lang/
git commit -m "feat(grc): show evidence source badge + Drive link + disable delete for Drive evidence"Files:
docs/changelog/<YYYY-MM-DD>-google-drive-to-system-sync.md