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

from __future__ import annotations

from datetime import datetime

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

from govoplan_core.core.change_sequence import record_change
from govoplan_core.core.sqlalchemy_change_tracking import (
    ensure_object_id,
    has_attr_changes,
    object_state,
    operation_for_soft_deletable,
    previous_value,
)
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 = object_state(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",
            "expires_at",
            "revoked_at",
            "revoked_by_user_id",
        ),
    ):
        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_expires_at": _isoformat(share.expires_at),
            "share_revoked_at": _isoformat(share.revoked_at),
            "share_revoked_by_user_id": share.revoked_by_user_id,
        },
    )


def _ensure_id(obj: object) -> str:
    return ensure_object_id(obj, new_uuid)


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",
]