Aislo/common_util/common_util_project_delete.py

"""프로젝트 영구 삭제 — DB 행과 영구저장소 폴더를 실제로 지운다.

대시보드 지움은 늘 휴지통(`B01_Dashboard_Repository.soft_delete_project`)으로 간다. 여기로
오는 길은 둘 — 휴지통 [영구 삭제](시스템 관리자) · `TRASH_RETENTION_DAYS` 지난 휴지통을
정리 루프가 비움(`purge_expired_trash`).
"""

import logging
import shutil
from typing import Any

from common_util.common_util_audit import record_audit
from common_util.common_util_storage import resolve_project_root_for_delete
from config.config_db import get_db_pool
from config.config_system import TRASH_RETENTION_DAYS

logger = logging.getLogger(__name__)


async def hard_delete_project(
    project_id: str,
    actor_id: int,
    request: Any | None = None,
    action: str = "PROJECT_HARD_DELETE",
) -> bool:
    """휴지통의 프로젝트를 DB와 영구저장소에서 완전히 지운다. 되돌릴 수 없다.

    자식 테이블은 나열하지 않는다 — `projects.id`를 참조하는 테이블이 전부
    `ON DELETE CASCADE`라 행 하나를 지우면 연쇄로 사라진다(`db_management/001_create_schema.sql`).
    코드에 목록을 다시 적으면 스키마가 바뀔 때 조용히 어긋난다.

    순서는 **DB 먼저, 파일 나중**이다. 파일을 먼저 지우면 rmtree 성공 후 DB가 실패했을 때
    실체 없는 프로젝트가 목록에 남아 화면이 깨진다. 반대로 DB가 먼저 확정되면 rmtree가
    실패해도 고아 폴더(디스크 낭비)만 남고 시스템 자체는 정합하다.
    """
    pool = get_db_pool()
    async with pool.acquire() as connection, connection.cursor() as cursor:
        await connection.begin()
        # 휴지통에 든 것만 · FOR UPDATE — 두 PC 서버가 같은 행을 같이 지우거나 그 사이 되살리면
        # 뒤엣것은 기다렸다가 「없음」으로 빠짐.
        await cursor.execute(
            """SELECT storage_path FROM projects
               WHERE id = %s AND deleted_at IS NOT NULL FOR UPDATE""",
            (project_id,),
        )
        row = await cursor.fetchone()
        if not row:
            await connection.rollback()
            return False
        stored_path = row[0]

        # 경로 검증을 커밋 전에 끝낸다 — 폴더를 못 지울 경로면 DB도 건드리지 않는다.
        project_root: str | None = None
        if stored_path:
            try:
                project_root = resolve_project_root_for_delete(str(stored_path), project_id)
            except ValueError:
                await connection.rollback()
                logger.error(
                    "하드 삭제 중단 — 저장 경로가 안전하지 않다: project_id=%s storage_path=%r",
                    project_id,
                    stored_path,
                )
                raise
        else:
            logger.warning("하드 삭제 — storage_path가 비어 DB만 지운다: project_id=%s", project_id)

        await cursor.execute("DELETE FROM projects WHERE id = %s", (project_id,))
        if cursor.rowcount <= 0:
            await connection.rollback()
            return False
        # 감사 기록은 프로젝트가 사라진 뒤에도 남는다. resource_id는 FK가 없어 고아가 되지 않는다.
        await record_audit(
            cursor,
            actor_id=actor_id,
            action=action,
            resource_type="project",
            resource_ref=project_id,
            request=request,
        )
        await connection.commit()

    if project_root:
        try:
            shutil.rmtree(project_root)
        except FileNotFoundError:
            pass
        except OSError as error:
            # DB는 이미 확정됐다. 여기서 실패해도 삭제 자체는 성공이다 — 경로를 남겨 수동 정리한다.
            logger.error(
                "하드 삭제 — DB는 지웠으나 폴더가 남았다. 수동 정리 필요: %s (%s)",
                project_root,
                error,
            )
        else:
            logger.info("하드 삭제 완료: project_id=%s 폴더=%s", project_id, project_root)
    return True


# 지운 사람 = 마지막 「휴지통으로」 감사 기록 · 없으면 시스템 관리자 한 명(감사 기록 user_id 필수).
DELETED_BY_SQL = """COALESCE(
    (SELECT a.user_id FROM system_audit_logs a
     WHERE a.action = 'PROJECT_DELETE' AND a.resource_ref = p.id
     ORDER BY a.id DESC LIMIT 1),
    (SELECT MIN(su.id) FROM users su WHERE su.role = 'SYSTEM_ADMIN'))"""

_PURGE_LOCK = "aislo_trash_purge"


async def list_expired_trash(
    cursor: Any, days: int = TRASH_RETENTION_DAYS
) -> list[tuple[str, int]]:
    """휴지통에 `days` 일 넘게 있은 프로젝트 → (id, 지운 사람)."""
    await cursor.execute(
        f"""SELECT p.id, {DELETED_BY_SQL} FROM projects p
            WHERE p.deleted_at IS NOT NULL
              AND p.deleted_at <= CURRENT_TIMESTAMP - INTERVAL %s DAY
            ORDER BY p.deleted_at""",
        (days,),
    )
    return [(str(row[0]), int(row[1])) for row in await cursor.fetchall()]


async def purge_expired_trash() -> int:
    """기한 지난 휴지통을 영구 삭제 · 지운 수.

    두 PC 서버가 DB 하나를 같이 씀 → 이름 잠금(GET_LOCK)을 못 잡으면 이번 차례는 건너뜀 ·
    그 사이 남이 먼저 지운 행은 `hard_delete_project` 가 False 로 건너뜀.
    """
    removed = 0
    try:
        async with get_db_pool().acquire() as connection, connection.cursor() as cursor:
            await cursor.execute("SELECT GET_LOCK(%s, 0)", (_PURGE_LOCK,))
            if (await cursor.fetchone())[0] != 1:
                return 0
            try:
                for project_id, actor_id in await list_expired_trash(cursor):
                    try:
                        if await hard_delete_project(project_id, actor_id, action="PROJECT_PURGE"):
                            removed += 1
                    except ValueError:
                        pass  # 경로가 안전하지 않음 — hard_delete_project 가 로그를 남김 · 다음 것
            finally:
                await cursor.execute("SELECT RELEASE_LOCK(%s)", (_PURGE_LOCK,))
    except Exception:
        logger.exception("휴지통 정리 실패")
    if removed:
        logger.info("휴지통 정리: %d건 영구 삭제 (보관 %d일)", removed, TRASH_RETENTION_DAYS)
    return removed