Google Drive 整合 Phase 3 — Review Fix Implementation Plan

const ALLOWED_ORIGIN = window.location.origin
let popup = null
let messageHandler = null

function connect() {
    cleanup()  // A3 也會用到，先抽出
    connecting.value = true
    popup = window.open(`${import.meta.env.VITE_API_BASE_URL}/api/1.0/integrations/google-drive/oauth/start`, ...)
    if (!popup) {
        toast.add({severity: 'warn', summary: t('lang.cloud_integration.popup_blocked')})
        connecting.value = false
        return
    }
    messageHandler = (event) => {
        if (event.origin !== ALLOWED_ORIGIN) return
        if (event.source !== popup) return
        const msg = event.data
        if (!msg || msg.source !== 'guidant-drive-oauth') return
        // ... 既有邏輯
    }
    window.addEventListener('message', messageHandler)
}

cd /Users/chouraymond/Projects/Billows/Audit-Manager/compliance-manager-fe
git add src/components/integrations/GoogleDriveIntegrationCard.vue \
        src/config/locales/i18n/zh-tw/cloud-integration.json \
        src/config/locales/i18n/en/cloud-integration.json
git commit -m "fix(cloud-integration): OAuth popup 加 origin/source 驗證避免訊息偽造"

let popupCheckTimer = null
let popupTimeoutTimer = null

function connect() {
    cleanup()
    connecting.value = true
    popup = window.open(...)
    if (!popup) {
        toast.add({severity: 'warn', summary: t('lang.cloud_integration.popup_blocked')})
        connecting.value = false
        return
    }
    messageHandler = (event) => { /* ...A1 */ }
    window.addEventListener('message', messageHandler)

    // 偵測 user 手動關掉 popup
    popupCheckTimer = setInterval(() => {
        if (popup.closed && connecting.value) {
            cleanup()
            connecting.value = false
            toast.add({severity: 'info', summary: t('lang.cloud_integration.connect_cancelled')})
        }
    }, 500)

    // 5 分鐘 timeout fallback
    popupTimeoutTimer = setTimeout(() => {
        if (connecting.value) {
            cleanup()
            connecting.value = false
            toast.add({severity: 'warn', summary: t('lang.cloud_integration.connect_timeout')})
        }
    }, 5 * 60 * 1000)
}

function cleanup() {
    if (messageHandler) {
        window.removeEventListener('message', messageHandler)
        messageHandler = null
    }
    if (popupCheckTimer) {
        clearInterval(popupCheckTimer)
        popupCheckTimer = null
    }
    if (popupTimeoutTimer) {
        clearTimeout(popupTimeoutTimer)
        popupTimeoutTimer = null
    }
    if (popup && !popup.closed) {
        try { popup.close() } catch (e) { /* ignore cross-origin */ }
    }
    popup = null
}

"connect_cancelled": "已取消 Google Drive 連線",
"connect_timeout": "連線逾時，請重新嘗試"

"connect_cancelled": "Google Drive connection cancelled",
"connect_timeout": "Connection timed out, please retry"

git add src/components/integrations/GoogleDriveIntegrationCard.vue \
        src/config/locales/i18n/zh-tw/cloud-integration.json \
        src/config/locales/i18n/en/cloud-integration.json
git commit -m "fix(cloud-integration): OAuth popup 關閉/逾時偵測避免按鈕卡 loading"

def handle(self, payload: dict) -> None:
    tenant_id = payload["tenant_id"]
    self_email = self._get_self_email(tenant_id)
    cursor = self._get_cursor(tenant_id)

    while True:
        resp = self._drive.list_changes(tenant_id, cursor, page_size=100)
        for change in resp.get("changes", []):
            try:
                self._process_one_change(tenant_id, change, self_email)
            except Exception as e:
                logger.warning(
                    "process_one_change failed (skipped): tenant=%s change=%s err=%s",
                    tenant_id, change.get("fileId"), e,
                )
                # 個別 change 失敗不能阻塞 cursor 推進，否則會卡死
        next_page = resp.get("nextPageToken")
        new_cursor = next_page if next_page else resp.get("newStartPageToken", cursor)
        if new_cursor != cursor:
            self._update_cursor(tenant_id, new_cursor)  # commit per page
            cursor = new_cursor
        if not next_page:
            break

@transaction
def _update_cursor(self, tenant_id: int, cursor: str) -> None:
    integration = self._tenant_integration.get_by_tenant_id(tenant_id)
    if integration:
        integration.drive_change_cursor = cursor
        self._tenant_integration.update(integration)

cd /Users/chouraymond/Projects/Billows/Audit-Manager/compliance-manager-be
git add app/cloud_integration/service/handlers/process_drive_changes_handler.py
git commit -m "fix(cloud_integration): drive changes cursor 每頁 commit 避免 crash 重跑"

def register_for_tenant(self, tenant_id: int) -> None:
    integration = self._integration_repo.get_by_tenant_id(tenant_id)
    if not integration or integration.status != "CONNECTED":
        raise PreconditionFailedError(GrcErrorCode.GRC_DRIVE_NOT_CONNECTED)

    # Step A: stop 舊 channel（best-effort）+ 立刻清 DB 欄位（重點）
    if integration.webhook_channel_id and integration.webhook_resource_id:
        try:
            self._drive.stop_channel(tenant_id, integration.webhook_channel_id, integration.webhook_resource_id)
        except Exception as e:
            logger.warning("stop_channel failed (continuing): %s", e)
        integration.webhook_channel_id = None
        integration.webhook_resource_id = None
        integration.webhook_token = None
        integration.webhook_expires_at = None
        self._integration_repo.update(integration)  # commit 清空狀態

    # Step B: init cursor 若缺
    if not integration.drive_change_cursor:
        integration.drive_change_cursor = self._drive.get_start_page_token(tenant_id)

    # Step C: watch — 失敗時 DB 已是 clean state，下次 renew_if_needed 會重試
    channel_id = str(uuid.uuid4())
    token = secrets.token_urlsafe(16)
    callback_url = f"{config.DRIVE_WEBHOOK_PUBLIC_BASE_URL}/api/1.0/webhooks/google-drive/{tenant_id}"
    resp = self._drive.watch_changes(tenant_id, channel_id, token, callback_url)

    # Step D: 寫回新 channel 資訊
    integration.webhook_channel_id = channel_id
    integration.webhook_resource_id = resp["resourceId"]
    integration.webhook_token = token
    integration.webhook_expires_at = datetime.fromtimestamp(int(resp["expiration"]) / 1000, tz=timezone.utc)
    self._integration_repo.update(integration)

def renew_if_needed(self, integration) -> bool:
    # 既有條件：webhook_expires_at < now() + 1 day → renew
    needs_renew = (
        integration.webhook_expires_at is None  # 新增：channel 沒建起來
        or integration.webhook_expires_at < datetime.now(timezone.utc) + timedelta(days=1)
    )
    if not needs_renew:
        return False
    if integration.status != "CONNECTED":
        return False
    self.register_for_tenant(integration.tenant_id)
    return True

git add domain/cloud_integration/service/webhook_channel_manager.py core/scheduler.py
git commit -m "fix(cloud_integration): webhook register 失敗保證 DB clean state + renewer 自動 recovery"

# app/grc/service/project_service.py
class ProjectService:
    def __init__(
        self,
        ...,  # 既有依賴
        drive_folder_mapping_domain_service=None,  # 新加，Optional 因為 cloud_integration_container 是 sibling
    ):
        ...
        self._drive_folder_mapping_domain_service = drive_folder_mapping_domain_service

    @transaction
    def enrich_drive_synced(self, dtos: list, tenant_id: int) -> None:
        if not dtos or self._drive_folder_mapping_domain_service is None:
            return
        uids = [str(d.uid) for d in dtos]
        try:
            mappings = self._drive_folder_mapping_domain_service.list_active_by_scope_uids(
                tenant_id, "PROJECT", uids
            )
            synced = {str(m.scope_uid) for m in mappings}
        except Exception as e:
            logger.warning("enrich_drive_synced failed: %s", e)
            synced = set()
        for d in dtos:
            d.drive_synced = d.uid in synced

class GrcContainer(containers.DeclarativeContainer):
    config = providers.Configuration()
    cloud_integration_container = providers.DependenciesContainer()  # 新加

    project_service = providers.Singleton(
        ProjectService,
        ...,
        drive_folder_mapping_domain_service=cloud_integration_container.drive_folder_mapping_domain_service,
    )

# di_containers/containers.py
class Containers(containers.DeclarativeContainer):
    cloud_integration_container = providers.Container(CloudIntegrationContainer)
    grc_container = providers.Container(
        GrcContainer,
        cloud_integration_container=cloud_integration_container,  # 單向 OK，不能反向（記憶檔已記錄 deepcopy 陷阱）
    )

# api/grc/routes/project_route.py
class ProjectListRoute(Resource):
    @inject
    def get(self, project_service: ProjectService = Provide[Containers.grc_container.project_service]):
        ...
        dtos = project_service.list(...)
        project_service.enrich_drive_synced(dtos, get_user_context().tenant_id)
        return {"data": [d.to_dict() for d in dtos], ...}

git add app/grc/service/project_service.py api/grc/routes/project_route.py \
        di_containers/grc/grc_containers.py di_containers/containers.py
git commit -m "refactor(grc): drive_synced enrich 從 route 搬到 app service (DDD)"

git add docs/changelog/2026-04-25-drive-review-fix-phase-a.md
git commit -m "docs: 新增 Drive 整合 Phase A review fix changelog"

# domain/cloud_integration/service/drive_api_client.py
from abc import ABC, abstractmethod

class IDriveApiClient(ABC):
    @abstractmethod
    def get_start_page_token(self, tenant_id: int) -> str: ...

    @abstractmethod
    def list_changes(self, tenant_id: int, cursor: str, page_size: int = 100) -> dict: ...

    @abstractmethod
    def watch_changes(self, tenant_id: int, channel_id: str, token: str, callback_url: str) -> dict: ...

    @abstractmethod
    def stop_channel(self, tenant_id: int, channel_id: str, resource_id: str) -> None: ...

    @abstractmethod
    def get_file_metadata(self, tenant_id: int, file_id: str, fields: str | None = None) -> dict | None: ...

    @abstractmethod
    def download_file(self, tenant_id: int, file_id: str, max_bytes: int) -> bytes: ...

    @abstractmethod
    def create_folder(self, tenant_id: int, name: str, parent_id: str) -> dict: ...

    @abstractmethod
    def rename(self, tenant_id: int, file_id: str, new_name: str) -> None: ...

    @abstractmethod
    def move_to_archive(self, tenant_id: int, file_id: str, archive_parent_id: str) -> None: ...

# infra/cloud_integration/google_drive/google_drive_api_client.py
from domain.cloud_integration.service.drive_api_client import IDriveApiClient

class GoogleDriveApiClient(IDriveApiClient):
    ...  # 既有實作

git mv domain/cloud_integration/service/webhook_channel_manager.py \
       app/cloud_integration/service/webhook_channel_manager.py

pytest test/cloud_integration/ -v

git add -A
git commit -m "refactor(cloud_integration): 抽 IDriveApiClient interface + WebhookChannelManager 改放 app 層"

def handle_notification(self, tenant_id: int, headers: dict) -> None:
    channel_id = headers.get("X-Goog-Channel-ID")
    token = headers.get("X-Goog-Channel-Token")

    if not channel_id or not token:
        logger.warning("webhook missing channel_id or token: tenant=%s", tenant_id)
        return  # silent 200

    integration = self._tenant_integration.get_by_tenant_id(tenant_id)
    if not integration:
        logger.warning("webhook for unknown tenant: %s", tenant_id)
        return

    if (integration.webhook_channel_id != channel_id
        or integration.webhook_token != token):
        logger.warning(
            "webhook channel/token mismatch: tenant=%s expected_channel=%s got=%s",
            tenant_id, integration.webhook_channel_id, channel_id,
        )
        return

    # enqueue PROCESS_DRIVE_CHANGES job
    self._drive_sync_jobs.enqueue(
        tenant_id=tenant_id,
        job_type="PROCESS_DRIVE_CHANGES",
        payload={"tenant_id": tenant_id},
    )

class GoogleDriveWebhookRoute(Resource):
    @inject
    def post(self, tenant_id: int,
             webhook_service: GoogleDriveWebhookService = Provide[...]):
        try:
            webhook_service.handle_notification(tenant_id, dict(request.headers))
        except Exception as e:
            logger.exception("webhook handle_notification crashed: %s", e)
        return {"received": True}, 200

# 缺 token：應該回 200
curl -X POST http://localhost:8000/api/1.0/webhooks/google-drive/102 \
     -H "X-Goog-Channel-ID: fake" \
     -i

# 錯 channel_id：應該回 200
curl -X POST http://localhost:8000/api/1.0/webhooks/google-drive/102 \
     -H "X-Goog-Channel-ID: wrong" \
     -H "X-Goog-Channel-Token: wrong" \
     -i

git add app/cloud_integration/service/google_drive_webhook_service.py \
        api/cloud_integration/routes/google_drive_webhook_route.py
git commit -m "fix(cloud_integration): webhook receiver 永遠回 200 避免 Drive 關閉 channel"

TERMINAL_ERROR_CODES = {"FILE_OVERSIZED", "FILE_NOT_FOUND_ON_DRIVE", "PERMISSION_DENIED"}

def fail_or_retry(self, job_id: int, error_msg: str, error_code: str | None = None) -> None:
    if error_code in TERMINAL_ERROR_CODES:
        self._repo.mark_failed(job_id, error_msg, retry=False)  # 不 retry
        return
    # 既有 retry 邏輯
    ...

try:
    if file_size > MAX_DOWNLOAD_BYTES:
        self._jobs.fail_or_retry(
            job.id,
            f"File oversized: {file_size} > {MAX_DOWNLOAD_BYTES}",
            error_code="FILE_OVERSIZED",
        )
        return
    ...
except DriveFileNotFoundError as e:
    self._jobs.fail_or_retry(job.id, str(e), error_code="FILE_NOT_FOUND_ON_DRIVE")
    return

git add domain/cloud_integration/service/drive_sync_job_domain_service.py \
        app/cloud_integration/service/handlers/import_drive_file_handler.py
git commit -m "fix(cloud_integration): deterministic 失敗（檔案過大/不存在）標 FAILED 不 retry"

try:
    if drive_orchestration is not None and tenant_id is not None:
        drive_orchestration.try_enqueue_reconcile_task_folder(tenant_id, job_uid)
except Exception as e:
    logger.warning(
        "revert_job: drive reconcile enqueue failed: tenant=%s job=%s err=%s",
        tenant_id, job_uid, e,
    )

git add app/flow_engine/service/workflow_execution_service.py
git commit -m "fix(workflow): revert_job drive reconcile 失敗補 log 避免吞錯"

# app/cloud_integration/service/handlers/archive_drive_file_handler.py
class ArchiveDriveFileHandler:
    def __init__(self, drive_api_client, tenant_integration_domain_service, ...):
        ...

    def handle(self, payload: dict) -> None:
        tenant_id = payload["tenant_id"]
        drive_file_id = payload["drive_file_id"]
        # 找 archive folder（同 orchestration._ensure_archive_folder 邏輯）
        # 呼叫 drive_api.move_to_archive
        # 失敗 → fail_or_retry（已有重試 schedule）

def try_archive_drive_file(self, tenant_id: int, drive_file_id: str | None) -> bool:
    if not drive_file_id:
        return False
    try:
        with session_scope():  # 獨立 transaction
            self._drive_sync_jobs.enqueue(
                tenant_id=tenant_id,
                job_type="ARCHIVE_DRIVE_FILE",
                payload={"tenant_id": tenant_id, "drive_file_id": drive_file_id},
            )
        return True
    except Exception as e:
        logger.warning("try_archive_drive_file enqueue failed: %s", e)
        return False

HANDLER_DISPATCH = {
    "INIT_PROJECT_FOLDERS": init_project_folders_handler,
    "CREATE_FOLDER": create_folder_handler,
    "RENAME_FOLDER": rename_folder_handler,
    "PROCESS_DRIVE_CHANGES": process_drive_changes_handler,
    "IMPORT_DRIVE_FILE": import_drive_file_handler,
    "SOFT_DELETE_EVIDENCE": soft_delete_evidence_handler,
    "RECONCILE_TASK_FOLDER": reconcile_task_folder_handler,
    "ARCHIVE_DRIVE_FILE": archive_drive_file_handler,  # 新加
}

-- scripts/sql/2026-04-25-add-archive-drive-file-job-type.sql
-- Date: 2026-04-25
ALTER TABLE compliance.drive_sync_jobs
DROP CONSTRAINT IF EXISTS chk_drive_sync_jobs_job_type;

ALTER TABLE compliance.drive_sync_jobs
ADD CONSTRAINT chk_drive_sync_jobs_job_type CHECK (
    job_type IN (
        'INIT_PROJECT_FOLDERS', 'CREATE_FOLDER', 'RENAME_FOLDER',
        'PROCESS_DRIVE_CHANGES', 'IMPORT_DRIVE_FILE',
        'SOFT_DELETE_EVIDENCE', 'RECONCILE_TASK_FOLDER',
        'ARCHIVE_DRIVE_FILE'  -- 2026-04-25
    )
);

PGPASSWORD='jedi@123!' psql -h 192.168.50.188 -p 25432 -U cm_app -d guidant_ai_stg \
    -f scripts/sql/2026-04-25-add-archive-drive-file-job-type.sql

git add -A
git commit -m "feat(cloud_integration): archive 改非同步 job 增加韌性 + 自動 retry"

# Before
source = getattr(evidence, "source", "SYSTEM_UPLOAD")
if (drive_orchestration is not None
    and tenant_id is not None
    and source == "DRIVE_SYNC"):

# After
if evidence.source == "DRIVE_SYNC":
    drive_orchestration.try_archive_drive_file(tenant_id, evidence.drive_file_id)

pytest test/flow_engine/ -v

git add app/flow_engine/service/job_evidence_service.py api/flow_engine/routes/job_evidence_route.py
git commit -m "refactor(flow_engine): 移除 delete_job_evidence 過度防禦的 None check"

def _handle_folder_change(self, mapping, change):
    if mapping.scope_type == "ROOT":
        # ROOT folder 不應該被 mark unlinked + enqueue create
        # ROOT 健康檢查由 InitProjectFoldersHandler 處理
        logger.warning("ROOT folder change detected, skipping: tenant=%s", mapping.tenant_id)
        return
    # 既有邏輯

git add app/cloud_integration/service/handlers/process_drive_changes_handler.py
git commit -m "fix(cloud_integration): folder change handler 跳過 ROOT scope 避免 unique constraint 衝突"

def mark_failed(self, job_id: int, error_msg: str, retry: bool = True) -> None:
    job = self.session.get(DriveSyncJobModel, job_id)
    if not job:
        return
    if retry and job.retry_count < MAX_RETRIES:
        job.status = "PENDING"
        job.retry_count += 1
        job.last_error = error_msg
        job.started_at = None  # 新加：retry 重置
        job.next_run_at = self._compute_next_run(job.retry_count)
    else:
        job.status = "FAILED"
        job.last_error = error_msg
        job.completed_at = datetime.now(timezone.utc)

def test_mark_failed_retry_resets_started_at(repo, claimed_job):
    # claim 過後 started_at 會有值
    repo.mark_failed(claimed_job.id, "transient error", retry=True)
    refreshed = repo.get_by_id(claimed_job.id)
    assert refreshed.status == "PENDING"
    assert refreshed.started_at is None

git add infra/cloud_integration/repository/drive_sync_job_repo_impl.py \
        test/cloud_integration/test_drive_sync_job_repo.py
git commit -m "fix(cloud_integration): retry path 重置 started_at 避免 admin UI 顯示誤導"

# common/code/cloud_integration_error_code.py
from jedi_common.enums.base_code import BaseCode

class CloudIntegrationErrorCode(BaseCode):
    DRIVE_OAUTH_EXCHANGE_FAILED      = ("Google OAuth 授權交換失敗",  "CI_400001")
    DRIVE_TOKEN_REFRESH_FAILED       = ("Google Drive Token 更新失敗", "CI_400002")
    DRIVE_NOT_CONNECTED              = ("尚未連接 Google Drive",       "CI_412001")
    DRIVE_API_QUOTA_EXCEEDED         = ("Google Drive API 配額已滿",   "CI_429001")

# infra/cloud_integration/google_drive/exceptions.py
class OAuthExchangeError(Exception):
    pass

class TokenRefreshError(Exception):
    pass

raise OAuthExchangeError(f"Token endpoint returned {resp.status_code}: {resp.text}")

# app/cloud_integration/service/google_drive_integration_service.py
from common.code.cloud_integration_error_code import CloudIntegrationErrorCode
from infra.cloud_integration.google_drive.exceptions import OAuthExchangeError
from jedi_common.handler.exception import BadRequestError

try:
    tokens = self._oauth_client.exchange_code(code)
except OAuthExchangeError as e:
    logger.warning("OAuth exchange failed: %s", e)
    raise BadRequestError(CloudIntegrationErrorCode.DRIVE_OAUTH_EXCHANGE_FAILED)

git add common/code/cloud_integration_error_code.py \
        infra/cloud_integration/google_drive/exceptions.py \
        infra/cloud_integration/google_drive/google_oauth_client.py \
        app/cloud_integration/service/google_drive_integration_service.py
git commit -m "refactor(cloud_integration): OAuth/Token 錯誤改 ErrorCode pattern 取代 RuntimeError"

git add docs/changelog/2026-04-25-drive-review-fix-phase-b.md
git commit -m "docs: 新增 Drive 整合 Phase B review fix changelog"