"""Team-management API — CRUD сотрудников, квоты, история (#2554, эпик #2549). Mounted at `/api/v1/team`; через Caddy `uri strip_prefix /trade-in` это `/trade-in/api/v1/team/*` снаружи. `app.services.auth_session.DB_ROLE_PATHS` уже закладывает `/api/v1/team/**` в scope роли `manager` (и `admin` через `/**`) для `rbac_guard` (см. `app.core.rbac`) — этот роутер добавляет ВТОРОЙ, более узкий барьер именно на identity: - `current_team_actor` резолвит юзера ТОЛЬКО из session-cookie (`app.services.auth_session.get_session_user`). Legacy `X-Authenticated-User` (Caddy trusted-header, dual-mode) НЕ принимается здесь — team-API новый, не участвует в переходном dual-mode auth. Без валидной cookie — 401, даже если `rbac_guard` пропустил запрос по legacy-заголовку (напр. admin через roles.yaml). - Роль должна быть `admin` или `manager` — иначе 403. Org-изоляция (главный инвариант фичи): manager видит/меняет ТОЛЬКО своих employee (`tradein_users.manager_id = actor.user_id`). Чужой/несуществующий employee_id → 404 (НЕ 403) — не подтверждаем/не опровергаем существование чужого сотрудника перед manager'ом. См. `_authorize_employee`. """ from __future__ import annotations import logging from dataclasses import dataclass from typing import Annotated, Any from fastapi import APIRouter, Depends, HTTPException, Query, Request from sqlalchemy import text from sqlalchemy.engine import RowMapping from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.core.config import settings from app.core.db import get_db from app.core.password import hash_password from app.schemas.team import ( EmployeeCreateRequest, EmployeeHistoryEntry, EmployeeOut, EmployeeUpdateRequest, QuotaStatusOut, ) from app.services import account_quota from app.services.auth_session import get_session_user, revoke_user_sessions from app.services.user_events import schedule_event logger = logging.getLogger(__name__) router = APIRouter() @dataclass class TeamActor: """Резолвленный из session-cookie актёр team-API — admin или manager.""" user_id: int username: str role: str # "admin" | "manager" async def current_team_actor( request: Request, db: Annotated[Session, Depends(get_db)], ) -> TeamActor: """Dependency: session-only identity, роль admin|manager, иначе 401/403. Намеренно НЕ читает `X-Authenticated-User` — см. модульный docstring. """ token = request.cookies.get(settings.session_cookie_name) if not token: raise HTTPException(status_code=401, detail="valid session required") try: session_user = get_session_user(db, token) except Exception: logger.exception("team: session lookup failed") raise HTTPException(status_code=401, detail="valid session required") from None if session_user is None: raise HTTPException(status_code=401, detail="valid session required") role = session_user["role"] if role not in ("admin", "manager"): raise HTTPException(status_code=403, detail="admin or manager role required") return TeamActor( user_id=session_user["user_id"], username=session_user["username"], role=role, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _fetch_employee_row(db: Session, employee_id: int) -> RowMapping | None: return ( db.execute( text( """ SELECT id, username, display_name, org_name, email, is_active, manager_id, created_at FROM tradein_users WHERE id = :id AND role = 'employee' """ ), {"id": employee_id}, ) .mappings() .fetchone() ) def _authorize_employee(actor: TeamActor, row: RowMapping | None) -> RowMapping: """404 (НЕ 403) если сотрудник не найден ИЛИ принадлежит другому manager'у. Org-изоляция: manager может видеть/менять только `manager_id == actor.user_id`. 404 вместо 403 — не палим существование чужого employee_id. """ if row is None: raise HTTPException(status_code=404, detail="employee not found") if actor.role == "manager" and row["manager_id"] != actor.user_id: raise HTTPException(status_code=404, detail="employee not found") return row def _upsert_quota_override( db: Session, username: str, monthly_limit: int, actor_username: str ) -> None: db.execute( text( """ INSERT INTO account_quota_overrides (username, monthly_limit, note) VALUES (:username, CAST(:monthly_limit AS integer), :note) ON CONFLICT (username) DO UPDATE SET monthly_limit = EXCLUDED.monthly_limit, note = EXCLUDED.note, updated_at = now() """ ), { "username": username, "monthly_limit": monthly_limit, "note": f"team-api: set by {actor_username}", }, ) def _employee_out(row: RowMapping, quota: dict[str, Any]) -> EmployeeOut: return EmployeeOut( id=row["id"], username=row["username"], display_name=row["display_name"], org_name=row["org_name"], email=row["email"], is_active=row["is_active"], manager_id=row["manager_id"], created_at=row["created_at"], quota=QuotaStatusOut(**quota), ) # --------------------------------------------------------------------------- # POST /employees # --------------------------------------------------------------------------- @router.post("/employees", response_model=EmployeeOut, status_code=201) async def create_employee( body: EmployeeCreateRequest, actor: Annotated[TeamActor, Depends(current_team_actor)], db: Annotated[Session, Depends(get_db)], ) -> EmployeeOut: """Создать сотрудника. Роль всегда `employee`. manager_id: для actor.role == manager — принудительно свой id (любое значение из тела ИГНОРИРУЕТСЯ, org-изоляция инвариант #2554). Для actor.role == admin — опционально из тела, валидируется что указанный id существует и role='manager' (иначе 422). """ existing = db.execute( text("SELECT id FROM tradein_users WHERE username = :u"), {"u": body.username}, ).fetchone() if existing is not None: raise HTTPException(status_code=409, detail="username already exists") try: password_hash = hash_password(body.password) except ValueError as e: raise HTTPException(status_code=422, detail=str(e)) from None manager_id: int | None if actor.role == "manager": # Инвариант org-изоляции: manager не может создать сотрудника под # чужим manager_id — любое значение из тела игнорируется молча. manager_id = actor.user_id else: manager_id = body.manager_id if manager_id is not None: mgr = db.execute( text("SELECT id FROM tradein_users WHERE id = :id AND role = 'manager'"), {"id": manager_id}, ).fetchone() if mgr is None: raise HTTPException( status_code=422, detail="manager_id does not reference an existing manager", ) try: row = ( db.execute( text( """ INSERT INTO tradein_users (username, password_hash, role, manager_id, display_name, org_name, email, is_active) VALUES (:username, :password_hash, 'employee', :manager_id, :display_name, :org_name, :email, true) RETURNING id, username, display_name, org_name, email, is_active, manager_id, created_at """ ), { "username": body.username, "password_hash": password_hash, "manager_id": manager_id, "display_name": body.display_name, "org_name": body.org_name, "email": body.email, }, ) .mappings() .fetchone() ) except IntegrityError: # TOCTOU: два конкурентных POST с одинаковым username между pre-check # выше и этим INSERT — UNIQUE-констрейнт на tradein_users.username ловит. db.rollback() raise HTTPException(status_code=409, detail="username already exists") from None assert row is not None # RETURNING на успешный INSERT всегда отдаёт строку if body.monthly_limit is not None: _upsert_quota_override(db, body.username, body.monthly_limit, actor.username) db.commit() schedule_event( event_type="employee_created", username=actor.username, payload={ "employee_id": row["id"], "employee_username": row["username"], "manager_id": manager_id, }, ) quota = account_quota.get_status(db, body.username) return _employee_out(row, quota) # --------------------------------------------------------------------------- # PATCH /employees/{id} # --------------------------------------------------------------------------- @router.patch("/employees/{employee_id}", response_model=EmployeeOut) async def update_employee( employee_id: int, body: EmployeeUpdateRequest, actor: Annotated[TeamActor, Depends(current_team_actor)], db: Annotated[Session, Depends(get_db)], ) -> EmployeeOut: """Частичное обновление сотрудника — block/unblock, лимит, профиль, пароль. manager может патчить ТОЛЬКО своих (manager_id == actor.user_id), иначе 404. При is_active=False — обязательно revoke всех сессий (иначе блокировка не подействует до истечения TTL текущей сессии сотрудника). """ row = _fetch_employee_row(db, employee_id) row = _authorize_employee(actor, row) new_password_hash: str | None = None if body.new_password is not None: try: new_password_hash = hash_password(body.new_password) except ValueError as e: raise HTTPException(status_code=422, detail=str(e)) from None db.execute( text( """ UPDATE tradein_users SET display_name = COALESCE(:display_name, display_name), org_name = COALESCE(:org_name, org_name), email = COALESCE(:email, email), is_active = COALESCE(CAST(:is_active AS boolean), is_active), password_hash = COALESCE(:password_hash, password_hash), updated_at = now() WHERE id = :id """ ), { "display_name": body.display_name, "org_name": body.org_name, "email": body.email, "is_active": body.is_active, "password_hash": new_password_hash, "id": employee_id, }, ) if body.monthly_limit is not None: _upsert_quota_override(db, row["username"], body.monthly_limit, actor.username) if body.is_active is False: # Обязательно ПОСЛЕ UPDATE, ДО финального commit — revoke_user_sessions # коммитит сам (см. app.services.auth_session), это флашит и наш # предшествующий UPDATE/quota-upsert в той же сессии. revoke_user_sessions(db, employee_id) db.commit() changed_profile_fields = [ f for f, v in ( ("display_name", body.display_name), ("org_name", body.org_name), ("email", body.email), ) if v is not None ] if changed_profile_fields: schedule_event( event_type="employee_updated", username=actor.username, payload={ "employee_id": employee_id, "employee_username": row["username"], "fields": changed_profile_fields, }, ) if body.new_password is not None: schedule_event( event_type="employee_password_reset", username=actor.username, payload={"employee_id": employee_id, "employee_username": row["username"]}, ) if body.is_active is not None: schedule_event( event_type="employee_blocked" if body.is_active is False else "employee_unblocked", username=actor.username, payload={"employee_id": employee_id, "employee_username": row["username"]}, ) if body.monthly_limit is not None: schedule_event( event_type="quota_changed", username=actor.username, payload={ "employee_id": employee_id, "employee_username": row["username"], "monthly_limit": body.monthly_limit, }, ) updated_row = _fetch_employee_row(db, employee_id) assert updated_row is not None # только что успешно обновили эту же строку quota = account_quota.get_status(db, updated_row["username"]) return _employee_out(updated_row, quota) # --------------------------------------------------------------------------- # GET /employees # --------------------------------------------------------------------------- @router.get("/employees", response_model=list[EmployeeOut]) async def list_employees( actor: Annotated[TeamActor, Depends(current_team_actor)], db: Annotated[Session, Depends(get_db)], manager_id: Annotated[int | None, Query()] = None, ) -> list[EmployeeOut]: """Список сотрудников. manager видит только своих; admin — всех, опц. ?manager_id=.""" params: dict[str, Any] = {} where = "WHERE role = 'employee'" if actor.role == "manager": where += " AND manager_id = :manager_id" params["manager_id"] = actor.user_id elif manager_id is not None: where += " AND manager_id = :manager_id" params["manager_id"] = manager_id rows = ( db.execute( text( f""" SELECT id, username, display_name, org_name, email, is_active, manager_id, created_at FROM tradein_users {where} ORDER BY created_at DESC """ ), params, ) .mappings() .all() ) result: list[EmployeeOut] = [] for row in rows: quota = account_quota.get_status(db, row["username"]) result.append(_employee_out(row, quota)) return result # --------------------------------------------------------------------------- # GET /employees/{id}/history # --------------------------------------------------------------------------- @router.get("/employees/{employee_id}/history", response_model=list[EmployeeHistoryEntry]) async def employee_history( employee_id: int, actor: Annotated[TeamActor, Depends(current_team_actor)], db: Annotated[Session, Depends(get_db)], limit: Annotated[int, Query(ge=1, le=200)] = 50, offset: Annotated[int, Query(ge=0)] = 0, ) -> list[EmployeeHistoryEntry]: """История оценок сотрудника (адрес/дата/результат) — из `user_events`, LEFT JOIN `trade_in_estimates` за фактическим результатом. Та же org-проверка что и в PATCH: чужой employee_id → 404. """ row = _fetch_employee_row(db, employee_id) row = _authorize_employee(actor, row) rows = ( db.execute( text( """ SELECT CAST(ue.estimate_id AS text) AS estimate_id, ue.payload ->> 'address' AS address, ue.payload ->> 'area_m2' AS area_m2, ue.payload ->> 'rooms' AS rooms, te.median_price, te.confidence, te.n_analogs, ue.created_at FROM user_events ue LEFT JOIN trade_in_estimates te ON te.id = ue.estimate_id WHERE ue.username = :username AND ue.event_type = 'estimate_request' ORDER BY ue.created_at DESC LIMIT :limit OFFSET :offset """ ), {"username": row["username"], "limit": limit, "offset": offset}, ) .mappings() .all() ) return [EmployeeHistoryEntry.model_validate(dict(r)) for r in rows]