company-ai-platform/app/application/feishu/personal_data.py

from collections.abc import Sequence

from fastapi import HTTPException, status
from sqlalchemy import delete, select
from sqlalchemy.orm import Session

from app.core.config import get_settings
from app.core.utils.time import utc_now
from app.modules.audit.constants import AuditRiskLevel, AuditSource, AuditStatus
from app.modules.audit.models import AuditLog
from app.modules.feishu_users.constants import (
    FEISHU_USER_NOT_FOUND,
    LAST_ACTIVE_ADMIN_ERROR,
    FeishuCapability,
    FeishuUserRole,
    FeishuUserStatus,
    parse_admin_identities,
)
from app.modules.feishu.services import (
    FeishuInboundService,
    current_inbound_event_key,
)
from app.modules.feishu_users.bootstrap import admin_bootstrap_identity_hash
from app.modules.feishu_users.identifiers import feishu_audit_identity_hash
from app.modules.feishu_users.models import (
    FeishuAdminBootstrapTombstone,
    FeishuUser,
)
from app.modules.feishu_users.principal import FeishuPrincipal
from app.modules.personalization.schemas import ErasureConfirmation, ErasureResult
from app.modules.personalization.services.erasure import (
    ErasureHook,
    PersonalDataErasureService,
)
from app.modules.subscriptions.models import PushDelivery, PushSubscription

PERSONAL_DATA_ERASURE_ACTION = "feishu.user.personal_data_erased"


class FeishuPersonalDataService:
    """Coordinate Feishu identity erasure across personal-data domains."""

    def __init__(
        self,
        db: Session,
        *,
        extra_hooks: Sequence[ErasureHook] = (),
    ) -> None:
        self.db = db
        self.extra_hooks = tuple(extra_hooks)
        self.erasure = PersonalDataErasureService(db)

    def request_confirmation(
        self,
        principal: FeishuPrincipal,
    ) -> ErasureConfirmation:
        """Issue a short-lived confirmation code for the authenticated user."""

        principal.require_capability(FeishuCapability.PERSONAL_DATA)
        user = self._find_principal_user(principal)
        if user.status != FeishuUserStatus.ACTIVE:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Feishu user is disabled",
            )
        return self.erasure.request_confirmation(user.id)

    def confirm(
        self,
        principal: FeishuPrincipal,
        confirmation_code: str,
    ) -> ErasureResult:
        """Confirm and erase the authenticated user's personal data."""

        principal.require_capability(FeishuCapability.PERSONAL_DATA)
        user = self._lock_principal_user(principal)
        if user.status != FeishuUserStatus.ACTIVE:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Feishu user is disabled",
            )
        return self._confirm_and_erase(
            user,
            confirmation_code,
            audit_source=AuditSource.FEISHU,
        )

    def erase_by_user_code(self, code: str, *, actor: str) -> ErasureResult:
        """Erase a user selected by an authenticated internal service."""

        if not actor.strip():
            raise HTTPException(
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                detail="Authenticated service actor is required",
            )
        user = self._lock_user(code=code)
        confirmation = self.erasure.request_confirmation(user.id)

        # request_confirmation commits by design. Lock and re-check the last-admin
        # invariant in the deletion transaction before any personal row is removed.
        user = self._lock_user(code=code)
        return self._confirm_and_erase(
            user,
            confirmation.confirmation_code,
            audit_source=AuditSource.API,
        )

    def _find_principal_user(self, principal: FeishuPrincipal) -> FeishuUser:
        user = self.db.execute(
            select(FeishuUser).where(
                FeishuUser.id == principal.owner_id,
                FeishuUser.code == principal.user_code,
                FeishuUser.tenant_key == principal.tenant_key,
                FeishuUser.open_id == principal.open_id,
            )
        ).scalar_one_or_none()
        if user is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=FEISHU_USER_NOT_FOUND,
            )
        return user

    def _lock_principal_user(self, principal: FeishuPrincipal) -> FeishuUser:
        return self._lock_user(
            owner_id=principal.owner_id,
            code=principal.user_code,
            tenant_key=principal.tenant_key,
            open_id=principal.open_id,
        )

    def _lock_user(
        self,
        *,
        owner_id: int | None = None,
        code: str | None = None,
        tenant_key: str | None = None,
        open_id: str | None = None,
    ) -> FeishuUser:
        active_admin_ids = list(
            self.db.execute(
                select(FeishuUser.id)
                .where(
                    FeishuUser.role == FeishuUserRole.ADMIN,
                    FeishuUser.status == FeishuUserStatus.ACTIVE,
                )
                .order_by(FeishuUser.id.asc())
                .with_for_update()
            ).scalars()
        )
        filters = []
        if owner_id is not None:
            filters.append(FeishuUser.id == owner_id)
        if code is not None:
            filters.append(FeishuUser.code == code)
        if tenant_key is not None:
            filters.append(FeishuUser.tenant_key == tenant_key)
        if open_id is not None:
            filters.append(FeishuUser.open_id == open_id)
        user = self.db.execute(
            select(FeishuUser).where(*filters).with_for_update()
        ).scalar_one_or_none()
        if user is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=FEISHU_USER_NOT_FOUND,
            )
        if (
            user.role == FeishuUserRole.ADMIN
            and user.status == FeishuUserStatus.ACTIVE
            and active_admin_ids == [user.id]
        ):
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail=LAST_ACTIVE_ADMIN_ERROR,
            )
        return user

    def _confirm_and_erase(
        self,
        user: FeishuUser,
        confirmation_code: str,
        *,
        audit_source: str,
    ) -> ErasureResult:
        identifiers = tuple(
            sorted(
                {
                    value
                    for value in (
                        user.code,
                        user.open_id,
                        user.union_id,
                        user.user_id,
                        feishu_audit_identity_hash(
                            user.tenant_key,
                            user.open_id,
                        ),
                    )
                    if value
                },
                key=len,
                reverse=True,
            )
        )
        configured_admins = parse_admin_identities(
            getattr(get_settings(), "feishu_admin_identities", ())
        )
        bootstrap_identity_hash = (
            admin_bootstrap_identity_hash(user.tenant_key, user.open_id)
            if (user.tenant_key, user.open_id) in configured_admins
            else None
        )

        def delete_subscriptions(
            db: Session,
            owner_id: int,
            _anonymous_id: str,
        ) -> dict[str, int]:
            subscription_ids = select(PushSubscription.id).where(
                PushSubscription.owner_id == owner_id
            )
            deliveries = _row_count(
                db.execute(
                    delete(PushDelivery).where(
                        PushDelivery.subscription_id.in_(subscription_ids)
                    )
                ).rowcount
            )
            subscriptions = _row_count(
                db.execute(
                    delete(PushSubscription).where(
                        PushSubscription.owner_id == owner_id
                    )
                ).rowcount
            )
            return {
                "deliveries": deliveries,
                "subscriptions": subscriptions,
            }

        def clear_pending_inbound_events(
            db: Session,
            _owner_id: int,
            _anonymous_id: str,
        ) -> dict[str, int]:
            # The caller already holds the FeishuUser row. Inbound handlers
            # acquire that same identity fence before their own receipt row, so
            # erasure can safely fence related receipts in identity -> inbox order.
            cleared = FeishuInboundService(db).erase_identity_payloads(
                tenant_key=user.tenant_key,
                open_id=user.open_id,
                exclude_event_key=current_inbound_event_key(),
            )
            return {"inbound_events": cleared}

        def finalize_identity(
            db: Session,
            _owner_id: int,
            anonymous_id: str,
        ) -> dict[str, int]:
            # Flush the pending confirmation-request deletion before removing
            # the identity that owns it.
            db.flush()
            anonymized_logs = _anonymize_audit_logs(
                db,
                identifiers=identifiers,
                anonymous_id=anonymous_id,
            )
            if bootstrap_identity_hash is not None:
                existing_tombstone = db.scalar(
                    select(FeishuAdminBootstrapTombstone.id).where(
                        FeishuAdminBootstrapTombstone.identity_hash
                        == bootstrap_identity_hash
                    )
                )
                if existing_tombstone is None:
                    db.add(
                        FeishuAdminBootstrapTombstone(
                            identity_hash=bootstrap_identity_hash
                        )
                    )
            db.delete(user)
            db.flush()
            db.add(
                AuditLog(
                    actor=anonymous_id,
                    source=audit_source,
                    action=PERSONAL_DATA_ERASURE_ACTION,
                    target_type=None,
                    target_id=None,
                    risk_level=AuditRiskLevel.HIGH,
                    request_payload=None,
                    response_payload=None,
                    status=AuditStatus.SUCCESS,
                    request_id=None,
                    created_at=utc_now(),
                )
            )
            return {
                "audit_logs_anonymized": anonymized_logs,
                "identity": 1,
            }

        return self.erasure.confirm_and_erase(
            user.id,
            confirmation_code,
            before_hooks=(delete_subscriptions, clear_pending_inbound_events),
            extra_hooks=(*self.extra_hooks, finalize_identity),
        )


def _anonymize_audit_logs(
    db: Session,
    *,
    identifiers: Sequence[str],
    anonymous_id: str,
) -> int:
    if not identifiers:
        return 0
    changed_count = 0
    for record in db.execute(select(AuditLog)).scalars():
        searchable_values = (
            record.actor,
            record.target_id,
            record.request_payload,
            record.response_payload,
        )
        if not any(
            identifier in value
            for value in searchable_values
            if value is not None
            for identifier in identifiers
        ):
            continue
        record.actor = anonymous_id
        record.target_type = None
        record.target_id = None
        record.request_payload = None
        record.response_payload = None
        record.request_id = None
        changed_count += 1
    db.flush()
    return changed_count


def _row_count(value: int | None) -> int:
    return max(0, int(value or 0))