arm-core/api/src/services/consolidation_locks.py

"""Блокировка версии консолидации «взять в работу»: взятие, перехват, снятие (spec consolidations-api «Взять в работу,
перехват, снятие»; audit-trail `CONSOLIDATION_LOCK`, `CONSOLIDATION_LOCK_TAKEOVER`; design D-7, D-9).

Каждая операция начинается с `consolidation_version FOR UPDATE` (D-9): правка версии владельцем держит ту же строку,
поэтому перехват ждёт конца правки и видит её зафиксированной, а два одновременных взятия выстраиваются в очередь
(второе после commit первого видит чужую блокировку). Частичный UNIQUE `uq_consolidation_lock_version_active` —
последний рубеж: вставка идёт под SAVEPOINT, нарушение превращается в 409 `LOCKED_BY_OTHER` без отравления транзакции
запроса. ASSUMPTION(OQ-119): блокировка не истекает по времени.
"""

import datetime
import uuid

from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from src.api.dependencies.audit import RequestAudit
from src.common.clock import utc_now
from src.common.constraint_conflicts import constraint_name, resolve_unique_conflict
from src.common.exceptions import ConflictError
from src.domain.policy import VisibilityScope
from src.models.access import AppUser
from src.models.consolidation import ConsolidationLock
from src.models.enums import AuditAction, AuditSection, ConsolidationStatus
from src.repositories import consolidations as repo
from src.repositories.users import UserWithAccess
from src.schemas.consolidations import LockIn, LockOut
from src.services.consolidation_guard import lock_visible_version
from src.services.consolidations import OBJECT_TYPE, lock_out

ACTIVE_LOCK_INDEX = "uq_consolidation_lock_version_active"


async def release_active_lock(
    session: AsyncSession, version_id: uuid.UUID, now: datetime.datetime
) -> ConsolidationLock | None:
    """Снимает активную блокировку версии (`released_at = now`), если она есть; возвращает снятую или `None`.

    Аудит не пишет: вызывающий записывает своё событие (`CONSOLIDATION_LOCK`, перехват, `CANCELLED`). Версия должна быть
    под `FOR UPDATE` вызывающего.
    """
    lock = await repo.active_lock(session, version_id)
    if lock is None:
        return None
    lock.released_at = now
    await session.flush()
    return lock


async def _user_name(session: AsyncSession, user_id: uuid.UUID) -> str:
    user = await session.get(AppUser, user_id)
    return user.full_name if user is not None else ""


class ConsolidationLockService:
    """Взятие версии в работу от имени текущего пользователя в его области видимости."""

    def __init__(
        self, session: AsyncSession, audit: RequestAudit, scope: VisibilityScope, principal: UserWithAccess
    ) -> None:
        self._session = session
        self._audit = audit
        self._scope = scope
        self._principal = principal

    async def take(self, version_id: uuid.UUID, data: LockIn) -> tuple[LockOut, bool]:
        """Взять версию в работу; `(блокировка, создана ли новая)` — 201 при `True`, 200 при своей активной.

        Порядок: `FOR UPDATE` версии → видимость (404) → `CANCELLED` → 409 `VERSION_NOT_EDITABLE` → активная
        блокировка: нет — новая; своя — возвращается как есть; чужая без причины — 409 `LOCKED_BY_OTHER`; чужая с
        непустой причиной — прежняя снимается, новая с `taken_over_from`/`takeover_reason`, аудит перехвата.
        Причина при свободной версии не сохраняется (перехватывать не у кого).
        """
        version, _ = await lock_visible_version(self._session, self._scope, version_id)
        if version.status == ConsolidationStatus.CANCELLED:
            raise ConflictError("VERSION_NOT_EDITABLE", "Аннулированную версию нельзя взять в работу")
        user = self._principal.user
        current = await repo.active_lock(self._session, version.id)
        if current is not None and current.locked_by == user.id:
            return lock_out(current, user.full_name), False
        reason = (data.takeover_reason or "").strip() or None
        if current is not None and reason is None:
            raise await self._locked_by_other(current)

        now = utc_now()
        if current is None:
            lock = await self._insert(version.id, now)
            await self._audit.record(
                action=AuditAction.CONSOLIDATION_LOCK,
                section=AuditSection.CONSOLIDATION,
                object_type=OBJECT_TYPE,
                object_id=version.id,
                consolidation_version_id=version.id,
                after_value={"locked_by": user.id, "released": False},
            )
            return lock_out(lock, user.full_name), True

        previous_owner = current.locked_by
        await release_active_lock(self._session, version.id, now)
        lock = await self._insert(version.id, now, taken_over_from=previous_owner, takeover_reason=reason)
        await self._audit.record(
            action=AuditAction.CONSOLIDATION_LOCK_TAKEOVER,
            section=AuditSection.CONSOLIDATION,
            object_type=OBJECT_TYPE,
            object_id=version.id,
            consolidation_version_id=version.id,
            before_value={"locked_by": previous_owner},
            after_value={"locked_by": user.id},
            comment=reason,
        )
        return lock_out(lock, user.full_name), True

    async def release(self, version_id: uuid.UUID) -> None:
        """Снять свою активную блокировку (аудит `CONSOLIDATION_LOCK`, `released = true`); чужая — 409
        `NOT_LOCK_OWNER`; активной нет — ничего не делает (идемпотентно)."""
        version, _ = await lock_visible_version(self._session, self._scope, version_id)
        current = await repo.active_lock(self._session, version.id)
        if current is None:
            return
        user = self._principal.user
        if current.locked_by != user.id:
            raise ConflictError(
                "NOT_LOCK_OWNER",
                "Снять можно только свою блокировку",
                locked_by=current.locked_by,
                locked_at=current.locked_at,
            )
        await release_active_lock(self._session, version.id, utc_now())
        await self._audit.record(
            action=AuditAction.CONSOLIDATION_LOCK,
            section=AuditSection.CONSOLIDATION,
            object_type=OBJECT_TYPE,
            object_id=version.id,
            consolidation_version_id=version.id,
            after_value={"locked_by": user.id, "released": True},
        )

    async def _insert(
        self,
        version_id: uuid.UUID,
        now: datetime.datetime,
        *,
        taken_over_from: uuid.UUID | None = None,
        takeover_reason: str | None = None,
    ) -> ConsolidationLock:
        """Новая активная блокировка под SAVEPOINT: нарушение частичного UNIQUE → 409 `LOCKED_BY_OTHER` с владельцем."""
        lock = ConsolidationLock(
            consolidation_version_id=version_id,
            locked_by=self._principal.user.id,
            locked_at=now,
            taken_over_from=taken_over_from,
            takeover_reason=takeover_reason,
        )
        try:
            async with self._session.begin_nested():
                self._session.add(lock)
                await self._session.flush()
        except IntegrityError as exc:
            conflict = resolve_unique_conflict(exc)
            if conflict is None or constraint_name(exc) != ACTIVE_LOCK_INDEX:
                raise
            winner = await repo.active_lock(self._session, version_id)
            if winner is None:  # победитель уже снял блокировку — постоянная часть ответа без владельца
                raise ConflictError(conflict.code, conflict.message) from exc
            raise await self._locked_by_other(winner) from exc
        return lock

    async def _locked_by_other(self, lock: ConsolidationLock) -> ConflictError:
        return ConflictError(
            "LOCKED_BY_OTHER",
            "Версию держит в работе другой пользователь",
            locked_by=lock.locked_by,
            locked_by_name=await _user_name(self._session, lock.locked_by),
            locked_at=lock.locked_at,
        )