govoplan-files/src/govoplan_files/backend/change_tracking.py

from __future__ import annotations

from datetime import datetime
from typing import Any

from sqlalchemy import event, inspect
from sqlalchemy.orm import Session as OrmSession

from govoplan_core.core.change_sequence import record_change
from govoplan_files.backend.db.models import FileAsset, FileFolder, FileShare, new_uuid

FILES_MODULE_ID = "files"
FILES_ASSETS_COLLECTION = "files.assets"
FILES_FOLDERS_COLLECTION = "files.folders"
FILES_CONNECTOR_PROFILES_COLLECTION = "files.connector_profiles"
FILES_CONNECTOR_CREDENTIALS_COLLECTION = "files.connector_credentials"
FILES_CONNECTOR_POLICIES_COLLECTION = "files.connector_policies"
FILES_CONNECTOR_SPACES_COLLECTION = "files.connector_spaces"

_REGISTERED = False


def register_files_change_tracking() -> None:
    global _REGISTERED
    if _REGISTERED:
        return
    event.listen(OrmSession, "before_flush", _record_files_changes)
    _REGISTERED = True


def _record_files_changes(session: OrmSession, _flush_context: object, _instances: object) -> None:
    for obj in tuple(session.new) + tuple(session.dirty):
        if isinstance(obj, FileAsset):
            _record_asset_change(session, obj)
        elif isinstance(obj, FileFolder):
            _record_folder_change(session, obj)
        elif isinstance(obj, FileShare):
            _record_share_visibility_change(session, obj)


def _record_asset_change(session: OrmSession, asset: FileAsset) -> None:
    operation = _operation_for_soft_deletable(
        asset,
        changed_attrs=(
            "owner_type",
            "owner_user_id",
            "owner_group_id",
            "current_version_id",
            "display_path",
            "filename",
            "description",
            "deleted_at",
            "metadata_",
        ),
    )
    if operation is None:
        return
    resource_id = _ensure_id(asset)
    record_change(
        session,
        module_id=FILES_MODULE_ID,
        collection=FILES_ASSETS_COLLECTION,
        resource_type="file",
        resource_id=resource_id,
        operation=operation,
        tenant_id=asset.tenant_id,
        actor_type="user" if asset.created_by_user_id else None,
        actor_id=asset.created_by_user_id,
        payload={
            "owner_type": asset.owner_type,
            "owner_id": _owner_id(asset.owner_type, asset.owner_user_id, asset.owner_group_id),
            "path": asset.display_path,
            "previous_path": _previous_value(asset, "display_path"),
            "filename": asset.filename,
            "deleted_at": _isoformat(asset.deleted_at),
        },
    )


def _record_folder_change(session: OrmSession, folder: FileFolder) -> None:
    operation = _operation_for_soft_deletable(
        folder,
        changed_attrs=("owner_type", "owner_user_id", "owner_group_id", "path", "deleted_at", "metadata_"),
    )
    if operation is None:
        return
    resource_id = _ensure_id(folder)
    record_change(
        session,
        module_id=FILES_MODULE_ID,
        collection=FILES_FOLDERS_COLLECTION,
        resource_type="folder",
        resource_id=resource_id,
        operation=operation,
        tenant_id=folder.tenant_id,
        actor_type="user" if folder.created_by_user_id else None,
        actor_id=folder.created_by_user_id,
        payload={
            "owner_type": folder.owner_type,
            "owner_id": _owner_id(folder.owner_type, folder.owner_user_id, folder.owner_group_id),
            "path": folder.path,
            "previous_path": _previous_value(folder, "path"),
            "deleted_at": _isoformat(folder.deleted_at),
        },
    )


def _record_share_visibility_change(session: OrmSession, share: FileShare) -> None:
    state = inspect(share)
    if not share.file_asset_id:
        return
    if not state.pending and not _has_attr_changes(
        state,
        ("file_asset_id", "target_type", "target_id", "permission", "revoked_at"),
    ):
        return
    _ensure_id(share)
    record_change(
        session,
        module_id=FILES_MODULE_ID,
        collection=FILES_ASSETS_COLLECTION,
        resource_type="file",
        resource_id=share.file_asset_id,
        operation="updated",
        tenant_id=share.tenant_id,
        actor_type="user" if share.created_by_user_id else None,
        actor_id=share.created_by_user_id,
        payload={
            "share_id": share.id,
            "share_target_type": share.target_type,
            "share_target_id": share.target_id,
            "share_permission": share.permission,
            "share_revoked_at": _isoformat(share.revoked_at),
        },
    )


def _operation_for_soft_deletable(obj: object, *, changed_attrs: tuple[str, ...]) -> str | None:
    state = inspect(obj)
    if state.pending:
        return "created"
    if not _has_attr_changes(state, changed_attrs):
        return None
    if "deleted_at" in state.attrs:
        history = state.attrs.deleted_at.history
        if history.has_changes():
            if any(value is not None for value in history.added):
                return "deleted"
            if any(value is not None for value in history.deleted):
                return "created"
    return "updated"


def _has_attr_changes(state: Any, attrs: tuple[str, ...]) -> bool:
    return any(name in state.attrs and state.attrs[name].history.has_changes() for name in attrs)


def _previous_value(obj: object, attr_name: str) -> str | None:
    state = inspect(obj)
    if attr_name not in state.attrs:
        return None
    history = state.attrs[attr_name].history
    if not history.has_changes() or not history.deleted:
        return None
    value = history.deleted[0]
    return str(value) if value is not None else None


def _ensure_id(obj: object) -> str:
    resource_id = getattr(obj, "id", None)
    if resource_id:
        return str(resource_id)
    resource_id = new_uuid()
    setattr(obj, "id", resource_id)
    return resource_id


def _owner_id(owner_type: str, owner_user_id: str | None, owner_group_id: str | None) -> str | None:
    if owner_type == "user":
        return owner_user_id
    if owner_type == "group":
        return owner_group_id
    return None


def _isoformat(value: datetime | None) -> str | None:
    return value.isoformat() if value else None


__all__ = [
    "FILES_ASSETS_COLLECTION",
    "FILES_CONNECTOR_CREDENTIALS_COLLECTION",
    "FILES_CONNECTOR_POLICIES_COLLECTION",
    "FILES_CONNECTOR_PROFILES_COLLECTION",
    "FILES_CONNECTOR_SPACES_COLLECTION",
    "FILES_FOLDERS_COLLECTION",
    "FILES_MODULE_ID",
    "register_files_change_tracking",
]