company-ai-platform/app/modules/feishu_users/services/identity.py

from collections.abc import Iterable
from typing import Any
from uuid import uuid4

from fastapi import HTTPException, status
from sqlalchemy import inspect, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.core.config import get_settings
from app.core.constants import ActorValue
from app.core.utils.time import utc_now
from app.modules.audit.constants import AuditRiskLevel, AuditSource
from app.modules.audit.schemas import AuditLogCreate
from app.modules.audit.service import AuditService
from app.modules.feishu_users.constants import (
    FEISHU_USER_CODE_PREFIX,
    FEISHU_USER_TARGET_TYPE,
    INVALID_FEISHU_IDENTITY,
    FeishuUserAuditAction,
    FeishuUserRole,
    FeishuUserStatus,
    parse_admin_identities,
)
from app.modules.feishu_users.bootstrap import admin_bootstrap_identity_hash
from app.modules.feishu_users.models import (
    FeishuAdminBootstrapTombstone,
    FeishuUser,
)
from app.modules.feishu_users.principal import FeishuPrincipal


class FeishuIdentityService:
    """Resolve or create identities only after the caller verifies the Feishu event."""

    def __init__(
        self,
        db: Session,
        admin_identities: str | Iterable[str] | None = None,
    ):
        self.db = db
        self.audit = AuditService(db)
        configured = (
            admin_identities
            if admin_identities is not None
            else getattr(get_settings(), "feishu_admin_identities", ())
        )
        self.admin_identities = parse_admin_identities(configured)

    def resolve_or_register(
        self,
        *,
        tenant_key: str,
        open_id: str,
        union_id: str | None = None,
        user_id: str | None = None,
        actor: str = ActorValue.FEISHU,
    ) -> FeishuPrincipal:
        """Return a principal for a previously verified Feishu sender."""

        tenant_key = _required_identity_part(tenant_key)
        open_id = _required_identity_part(open_id)
        union_id = _optional_identity_part(union_id)
        user_id = _optional_identity_part(user_id)
        record = self.get_by_identity(tenant_key=tenant_key, open_id=open_id)
        created = False
        if record is None:
            candidate = FeishuUser(
                code=f"{FEISHU_USER_CODE_PREFIX}-{uuid4().hex[:20].upper()}",
                tenant_key=tenant_key,
                open_id=open_id,
                union_id=union_id,
                user_id=user_id,
                role=self._initial_role(tenant_key, open_id),
                status=FeishuUserStatus.ACTIVE,
                last_active_at=utc_now(),
            )
            try:
                with self.db.begin_nested():
                    self.db.add(candidate)
                    self.db.flush()
                record = candidate
                created = True
            except IntegrityError:
                record = self.get_by_identity(tenant_key=tenant_key, open_id=open_id)
                if record is None:
                    raise

        record.last_active_at = utc_now()
        if union_id:
            record.union_id = union_id
        if user_id:
            record.user_id = user_id
        self.audit.record(
            AuditLogCreate(
                actor=actor,
                source=AuditSource.FEISHU,
                action=(
                    FeishuUserAuditAction.REGISTER
                    if created
                    else FeishuUserAuditAction.AUTHENTICATE
                ),
                target_type=FEISHU_USER_TARGET_TYPE,
                target_id=record.code,
                risk_level=AuditRiskLevel.LOW,
                response_payload={
                    "created": created,
                    "role": record.role,
                    "status": record.status,
                },
            )
        )
        self.db.commit()
        self.db.refresh(record)
        if created and inspect(self.db.get_bind()).has_table("market_watchlists"):
            # Import locally so the identity domain does not create a module cycle.
            from app.modules.market.service import MarketService

            MarketService(self.db).claim_legacy_watchlist(record.id, record.open_id)
        return FeishuPrincipal.from_user(record)

    def get_by_identity(self, *, tenant_key: str, open_id: str) -> FeishuUser | None:
        return self.db.execute(
            select(FeishuUser).where(
                FeishuUser.tenant_key == tenant_key,
                FeishuUser.open_id == open_id,
            )
        ).scalar_one_or_none()

    def get_by_code(self, code: str) -> FeishuUser | None:
        return self.db.execute(
            select(FeishuUser).where(FeishuUser.code == code)
        ).scalar_one_or_none()

    def _initial_role(self, tenant_key: str, open_id: str) -> str:
        if (tenant_key, open_id) not in self.admin_identities:
            return FeishuUserRole.USER
        identity_hash = admin_bootstrap_identity_hash(tenant_key, open_id)
        was_erased = self.db.scalar(
            select(FeishuAdminBootstrapTombstone.id)
            .where(
                FeishuAdminBootstrapTombstone.identity_hash == identity_hash
            )
            .limit(1)
        )
        return FeishuUserRole.USER if was_erased is not None else FeishuUserRole.ADMIN


def _required_identity_part(value: Any) -> str:
    text = str(value or "").strip()
    if not text:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=INVALID_FEISHU_IDENTITY,
        )
    return text


def _optional_identity_part(value: Any) -> str | None:
    text = str(value or "").strip()
    return text or None