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

from __future__ import annotations

import hashlib
import mimetypes
from datetime import datetime
from pathlib import PurePosixPath
from typing import Any, Iterable
from uuid import uuid4

from sqlalchemy import and_, exists, func, or_
from sqlalchemy.orm import Session

from govoplan_core.core.campaigns import CAPABILITY_CAMPAIGNS_ACCESS, CampaignAccessProvider
from govoplan_files.backend.db.models import CampaignAttachmentUse, FileAsset, FileBlob, FileShare, FileVersion
from govoplan_files.backend.runtime import get_registry, settings
from govoplan_files.backend.storage.access import ensure_owner_access, ensure_share_target_exists, user_group_ids
from govoplan_files.backend.storage.backends import (
    StorageBackendError,
    StorageObjectMissing,
    get_storage_backend,
)
from govoplan_files.backend.storage.common import FileConflictResolution, FileStorageError, UploadedStoredFile, utcnow
from govoplan_files.backend.storage.paths import filename_from_path, join_folder_filename, normalize_folder, normalize_logical_path
from govoplan_files.backend.storage.provenance import source_provenance_from_metadata
from govoplan_files.backend.storage.recovery import begin_blob_write_recovery
from govoplan_files.backend.storage.integrity import (
    QUARANTINED_BLOB_STATUSES,
    read_verified_blob_bytes,
)
from govoplan_files.backend.storage.share_state import effective_file_share_clause


def _campaign_access_provider() -> CampaignAccessProvider:
    registry = get_registry()
    if registry is None or not hasattr(registry, "has_capability") or not registry.has_capability(CAPABILITY_CAMPAIGNS_ACCESS):
        raise FileStorageError("Campaign module is not installed")
    capability = registry.require_capability(CAPABILITY_CAMPAIGNS_ACCESS)
    if not isinstance(capability, CampaignAccessProvider):
        raise FileStorageError("Campaign access capability is invalid")
    return capability


def _ensure_campaign_exists(session: Session, *, tenant_id: str, campaign_id: str) -> None:
    if not _campaign_access_provider().campaign_exists(session, tenant_id=tenant_id, campaign_id=campaign_id):
        raise FileStorageError("Campaign not found")


def _asset_query_for_owner(session: Session, *, tenant_id: str, owner_type: str, owner_id: str):
    query = session.query(FileAsset).filter(FileAsset.tenant_id == tenant_id, FileAsset.owner_type == owner_type)
    if owner_type == "user":
        return query.filter(FileAsset.owner_user_id == owner_id)
    if owner_type == "group":
        return query.filter(FileAsset.owner_group_id == owner_id)
    raise FileStorageError("Unsupported owner type")


def _storage_bucket_name() -> str:
    return settings.file_storage_s3_bucket or settings.s3_bucket


def _storage_backend_name() -> str:
    return settings.file_storage_backend.lower().strip()


def _storage_key(*, tenant_id: str, checksum: str) -> str:
    # Object locators remain opaque so recovery evidence never persists names.
    return f"tenants/{tenant_id}/files/{checksum[:2]}/{uuid4().hex}.blob"


def _get_or_create_blob(
    session: Session,
    *,
    tenant_id: str,
    data: bytes,
    filename: str,
    content_type: str | None,
    actor_id: str,
    encryption_vault_id: str | None = None,
) -> FileBlob:
    checksum = hashlib.sha256(data).hexdigest()
    size = len(data)
    vault_id = str(encryption_vault_id or "").strip() or None
    protection_discriminator = f"vault:{vault_id}" if vault_id else "plaintext"
    blob = (
        session.query(FileBlob)
        .filter(FileBlob.tenant_id == tenant_id, FileBlob.checksum_sha256 == checksum, FileBlob.size_bytes == size, FileBlob.protection_discriminator == protection_discriminator)
        .with_for_update()
        .one_or_none()
    )
    if blob:
        backend = get_storage_backend()
        repair_required = (
            blob.integrity_status in QUARANTINED_BLOB_STATUSES
            or blob.quarantined_at is not None
        )
        try:
            backend.stat(blob.storage_key)
        except StorageObjectMissing:
            repair_required = True
        except StorageBackendError as exc:
            raise FileStorageError(str(exc)) from exc
        if repair_required:
            repair_token = hashlib.sha256(
                repr(
                    (
                        blob.integrity_status,
                        blob.integrity_checked_at,
                        blob.quarantined_at,
                        blob.storage_checksum_sha256,
                    )
                ).encode("utf-8")
            ).hexdigest()
            recovery = begin_blob_write_recovery(
                session,
                backend=backend,
                tenant_id=tenant_id,
                blob_id=blob.id,
                storage_key=blob.storage_key,
                semantic_checksum_sha256=checksum,
                semantic_size_bytes=size,
                protection_discriminator=protection_discriminator,
                created_new=False,
                repair_token=repair_token,
            )
            stored_data = data
            expected_envelope_id = blob.encryption_envelope_id
            if vault_id:
                from govoplan_files.backend.storage.content_protection import protect_blob_content

                protected = protect_blob_content(
                    session,
                    tenant_id=tenant_id,
                    blob_id=blob.id,
                    vault_id=vault_id,
                    ciphertext_ref=blob.storage_key,
                    plaintext=data,
                    actor_id=actor_id,
                    content_type=content_type,
                )
                if blob.encryption_envelope_id not in {None, protected.envelope.envelope_id}:
                    raise FileStorageError("The existing encrypted blob has another protection envelope.")
                stored_data = protected.ciphertext
                blob.encryption_envelope_id = protected.envelope.envelope_id
                expected_envelope_id = protected.envelope.envelope_id
                blob.storage_checksum_sha256 = hashlib.sha256(stored_data).hexdigest()
                blob.storage_size_bytes = len(stored_data)
            recovery.prepare_stored_bytes(
                stored_data,
                envelope_id=expected_envelope_id,
            )
            try:
                backend.put_bytes(blob.storage_key, stored_data, content_type="application/octet-stream" if vault_id else content_type)
            except StorageBackendError as exc:
                raise FileStorageError(str(exc)) from exc
            blob.integrity_status = "verified"
            blob.integrity_checked_at = utcnow()
            blob.integrity_failure = None
            blob.quarantined_at = None
        blob.ref_count += 1
        session.add(blob)
        return blob

    blob_id = str(uuid4())
    storage_key = _storage_key(tenant_id=tenant_id, checksum=checksum)
    backend = get_storage_backend()
    recovery = begin_blob_write_recovery(
        session,
        backend=backend,
        tenant_id=tenant_id,
        blob_id=blob_id,
        storage_key=storage_key,
        semantic_checksum_sha256=checksum,
        semantic_size_bytes=size,
        protection_discriminator=protection_discriminator,
        created_new=True,
    )
    stored_data = data
    envelope_id = None
    if vault_id:
        from govoplan_files.backend.storage.content_protection import protect_blob_content

        protected = protect_blob_content(
            session,
            tenant_id=tenant_id,
            blob_id=blob_id,
            vault_id=vault_id,
            ciphertext_ref=storage_key,
            plaintext=data,
            actor_id=actor_id,
            content_type=content_type,
        )
        stored_data = protected.ciphertext
        envelope_id = protected.envelope.envelope_id
    recovery.prepare_stored_bytes(stored_data, envelope_id=envelope_id)
    try:
        backend.put_bytes(storage_key, stored_data, content_type="application/octet-stream" if vault_id else content_type)
    except StorageBackendError as exc:
        raise FileStorageError(str(exc)) from exc
    blob = FileBlob(
        id=blob_id,
        tenant_id=tenant_id,
        storage_backend=_storage_backend_name(),
        storage_bucket=_storage_bucket_name(),
        storage_key=storage_key,
        checksum_sha256=checksum,
        size_bytes=size,
        protection_discriminator=protection_discriminator,
        encryption_envelope_id=envelope_id,
        storage_checksum_sha256=hashlib.sha256(stored_data).hexdigest() if vault_id else None,
        storage_size_bytes=len(stored_data) if vault_id else None,
        content_type=content_type,
        ref_count=1,
        integrity_status="verified",
        integrity_checked_at=utcnow(),
    )
    session.add(blob)
    session.flush()
    return blob


def create_file_asset(
    session: Session,
    *,
    tenant_id: str,
    owner_type: str,
    owner_id: str,
    user_id: str,
    filename: str,
    data: bytes,
    folder: str | None = None,
    display_path: str | None = None,
    content_type: str | None = None,
    description: str | None = None,
    metadata: dict[str, Any] | None = None,
    campaign_id: str | None = None,
    conflict_strategy: str = "reject",
    conflict_resolutions: Iterable[FileConflictResolution] | None = None,
    is_admin: bool = False,
    encryption_vault_id: str | None = None,
) -> UploadedStoredFile:
    owner_type = owner_type.lower().strip()
    ensure_owner_access(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, user_id=user_id, is_admin=is_admin)

    safe_filename = filename_from_path(normalize_logical_path(filename, fallback_filename="file"))
    logical_path = normalize_logical_path(display_path) if display_path else join_folder_filename(folder, safe_filename)
    if not content_type:
        content_type = mimetypes.guess_type(safe_filename)[0] or "application/octet-stream"

    conflict_strategy = _normalize_conflict_strategy(conflict_strategy)
    resolutions = _resolution_by_path(conflict_resolutions)
    resolution = resolutions.get(logical_path)
    action = resolution.action if resolution else conflict_strategy
    if resolution and resolution.new_path and action == "rename":
        logical_path = normalize_logical_path(resolution.new_path)
    elif _active_asset_exists(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, path=logical_path):
        if action == "reject":
            raise FileStorageError(f"Target file already exists: {logical_path}")
        if action == "overwrite":
            _soft_delete_conflicting_asset(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, path=logical_path)
        elif action == "rename":
            logical_path = _next_available_logical_path(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, desired_path=logical_path)
        elif action == "skip":
            raise FileStorageError(f"Skipped upload target: {logical_path}")
    elif action == "rename":
        logical_path = _next_available_logical_path(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, desired_path=logical_path)

    blob = _get_or_create_blob(session, tenant_id=tenant_id, data=data, filename=safe_filename, content_type=content_type, actor_id=user_id, encryption_vault_id=encryption_vault_id)
    asset = FileAsset(
        tenant_id=tenant_id,
        owner_type=owner_type,
        owner_user_id=owner_id if owner_type == "user" else None,
        owner_group_id=owner_id if owner_type == "group" else None,
        display_path=logical_path,
        filename=filename_from_path(logical_path),
        description=description,
        created_by_user_id=user_id,
        metadata_=metadata or {},
    )
    session.add(asset)
    session.flush()
    version = FileVersion(
        tenant_id=tenant_id,
        file_asset_id=asset.id,
        blob_id=blob.id,
        version_number=1,
        filename_at_upload=safe_filename,
        display_path_at_upload=logical_path,
        content_type=content_type,
        size_bytes=blob.size_bytes,
        checksum_sha256=blob.checksum_sha256,
        created_by_user_id=user_id,
    )
    session.add(version)
    session.flush()
    asset.current_version_id = version.id
    session.add(asset)
    if campaign_id:
        share_file(session, tenant_id=tenant_id, asset=asset, target_type="campaign", target_id=campaign_id, permission="read", user_id=user_id)
    return UploadedStoredFile(asset=asset, version=version, blob=blob)


def sync_file_asset_from_source(
    session: Session,
    *,
    tenant_id: str,
    owner_type: str,
    owner_id: str,
    user_id: str,
    filename: str,
    data: bytes,
    metadata: dict[str, Any],
    folder: str | None = None,
    display_path: str | None = None,
    content_type: str | None = None,
    campaign_id: str | None = None,
    conflict_strategy: str = "rename",
    is_admin: bool = False,
) -> tuple[UploadedStoredFile, str, str | None]:
    owner_type = owner_type.lower().strip()
    ensure_owner_access(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, user_id=user_id, is_admin=is_admin)
    provenance = source_provenance_from_metadata(metadata)
    if not provenance:
        raise FileStorageError("Connector sync requires source provenance")
    existing = find_asset_by_source(
        session,
        tenant_id=tenant_id,
        owner_type=owner_type,
        owner_id=owner_id,
        source_provenance=provenance,
    )
    if existing is None:
        stored = create_file_asset(
            session,
            tenant_id=tenant_id,
            owner_type=owner_type,
            owner_id=owner_id,
            user_id=user_id,
            filename=filename,
            data=data,
            folder=folder,
            display_path=display_path,
            content_type=content_type,
            metadata=metadata,
            campaign_id=campaign_id,
            conflict_strategy=conflict_strategy,
            is_admin=is_admin,
        )
        return stored, "created", None

    previous_version_id = existing.current_version_id
    stored, action = update_file_asset_content(
        session,
        existing,
        tenant_id=tenant_id,
        user_id=user_id,
        filename=filename,
        data=data,
        content_type=content_type,
        metadata=metadata,
    )
    if campaign_id:
        share_file(session, tenant_id=tenant_id, asset=existing, target_type="campaign", target_id=campaign_id, permission="read", user_id=user_id)
    return stored, action, previous_version_id


def find_asset_by_source(
    session: Session,
    *,
    tenant_id: str,
    owner_type: str,
    owner_id: str,
    source_provenance: dict[str, Any],
) -> FileAsset | None:
    wanted = _source_identity(source_provenance)
    if wanted is None:
        return None
    assets = (
        _asset_query_for_owner(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id)
        .filter(FileAsset.deleted_at.is_(None))
        .order_by(FileAsset.updated_at.desc())
        .all()
    )
    for asset in assets:
        if _source_identity(source_provenance_from_metadata(asset.metadata_ or {})) == wanted:
            return asset
    return None


def update_file_asset_content(
    session: Session,
    asset: FileAsset,
    *,
    tenant_id: str,
    user_id: str,
    filename: str,
    data: bytes,
    content_type: str | None,
    metadata: dict[str, Any],
    encryption_vault_id: str | None = None,
) -> tuple[UploadedStoredFile, str]:
    if asset.tenant_id != tenant_id or asset.deleted_at is not None:
        raise FileStorageError("File not found")
    safe_filename = filename_from_path(normalize_logical_path(filename, fallback_filename="file"))
    if not content_type:
        content_type = mimetypes.guess_type(safe_filename)[0] or "application/octet-stream"
    current_version, current_blob = current_version_and_blob(session, asset)
    checksum = hashlib.sha256(data).hexdigest()
    asset.metadata_ = metadata
    session.add(asset)
    inherited_vault_id = encryption_vault_id
    if inherited_vault_id is None and current_blob.encryption_envelope_id:
        prefix = "vault:"
        if current_blob.protection_discriminator.startswith(prefix):
            inherited_vault_id = current_blob.protection_discriminator[len(prefix) :]
    inherited_vault_id = str(inherited_vault_id or "").strip() or None
    target_protection = (
        f"vault:{inherited_vault_id}" if inherited_vault_id else "plaintext"
    )
    if (
        current_blob.checksum_sha256 == checksum
        and current_blob.size_bytes == len(data)
        and current_blob.protection_discriminator == target_protection
    ):
        return UploadedStoredFile(asset=asset, version=current_version, blob=current_blob), "unchanged"

    blob = _get_or_create_blob(session, tenant_id=tenant_id, data=data, filename=safe_filename, content_type=content_type, actor_id=user_id, encryption_vault_id=inherited_vault_id)
    version = FileVersion(
        tenant_id=tenant_id,
        file_asset_id=asset.id,
        blob_id=blob.id,
        version_number=_next_version_number(session, asset.id),
        filename_at_upload=safe_filename,
        display_path_at_upload=asset.display_path,
        content_type=content_type,
        size_bytes=blob.size_bytes,
        checksum_sha256=blob.checksum_sha256,
        created_by_user_id=user_id,
    )
    session.add(version)
    session.flush()
    asset.current_version_id = version.id
    session.add(asset)
    return UploadedStoredFile(asset=asset, version=version, blob=blob), "updated"


def get_asset_for_user(session: Session, *, tenant_id: str, user_id: str, asset_id: str, require_write: bool = False, is_admin: bool = False) -> FileAsset:
    asset = session.get(FileAsset, asset_id)
    if not asset or asset.tenant_id != tenant_id or asset.deleted_at is not None:
        raise FileStorageError("File not found")
    if is_admin:
        return asset
    group_ids = user_group_ids(session, tenant_id=tenant_id, user_id=user_id)
    owns = (asset.owner_type == "user" and asset.owner_user_id == user_id) or (asset.owner_type == "group" and asset.owner_group_id in group_ids)
    if owns:
        return asset
    permission_values = ["read", "write", "manage"] if not require_write else ["write", "manage"]
    share = (
        session.query(FileShare)
        .filter(
            FileShare.tenant_id == tenant_id,
            FileShare.file_asset_id == asset.id,
            effective_file_share_clause(),
            FileShare.permission.in_(permission_values),
            or_(
                (FileShare.target_type == "user") & (FileShare.target_id == user_id),
                (FileShare.target_type == "group") & (FileShare.target_id.in_(group_ids)),
                (FileShare.target_type == "tenant") & (FileShare.target_id == tenant_id),
            ),
        )
        .first()
    )
    if not share:
        raise FileStorageError("No access to this file")
    return asset


def get_asset_for_share_management(
    session: Session,
    *,
    tenant_id: str,
    user_id: str,
    asset_id: str,
    is_admin: bool = False,
) -> FileAsset:
    asset = session.get(FileAsset, asset_id)
    if not asset or asset.tenant_id != tenant_id or asset.deleted_at is not None:
        raise FileStorageError("File not found")
    if is_admin:
        return asset
    owns_asset = (
        asset.owner_type == "user"
        and asset.owner_user_id == user_id
    ) or (
        asset.owner_type == "group"
        and asset.owner_group_id
        in user_group_ids(session, tenant_id=tenant_id, user_id=user_id)
    )
    if not owns_asset:
        raise FileStorageError("Only file owners and administrators can manage shares")
    return asset


def list_assets_for_user(
    session: Session,
    *,
    tenant_id: str,
    user_id: str,
    owner_type: str | None = None,
    owner_id: str | None = None,
    campaign_id: str | None = None,
    path_prefix: str | None = None,
    campaign_usage: str | None = None,
    audit_relevant: bool | None = None,
    include_deleted: bool = False,
    is_admin: bool = False,
) -> list[FileAsset]:
    query = _asset_visibility_query_for_user(
        session,
        tenant_id=tenant_id,
        user_id=user_id,
        owner_type=owner_type,
        owner_id=owner_id,
        campaign_id=campaign_id,
        path_prefix=path_prefix,
        campaign_usage=campaign_usage,
        audit_relevant=audit_relevant,
        include_deleted=include_deleted,
        is_admin=is_admin,
    )
    return query.order_by(FileAsset.display_path.asc(), FileAsset.updated_at.desc(), FileAsset.id.asc()).all()


def list_recent_assets_for_user(
    session: Session,
    *,
    tenant_id: str,
    user_id: str,
    limit: int,
    owner_type: str | None = None,
    owner_id: str | None = None,
    campaign_id: str | None = None,
    path_prefix: str | None = None,
    campaign_usage: str | None = None,
    audit_relevant: bool | None = None,
    is_admin: bool = False,
) -> list[FileAsset]:
    """Return a bounded recent projection through the normal access query."""

    query = _asset_visibility_query_for_user(
        session,
        tenant_id=tenant_id,
        user_id=user_id,
        owner_type=owner_type,
        owner_id=owner_id,
        campaign_id=campaign_id,
        path_prefix=path_prefix,
        campaign_usage=campaign_usage,
        audit_relevant=audit_relevant,
        is_admin=is_admin,
    )
    return (
        query.order_by(FileAsset.updated_at.desc(), FileAsset.id.asc())
        .limit(max(1, limit))
        .all()
    )


def list_assets_for_user_window(
    session: Session,
    *,
    tenant_id: str,
    user_id: str,
    owner_type: str | None = None,
    owner_id: str | None = None,
    campaign_id: str | None = None,
    path_prefix: str | None = None,
    campaign_usage: str | None = None,
    audit_relevant: bool | None = None,
    include_deleted: bool = False,
    is_admin: bool = False,
    page_size: int,
    after_display_path: str | None = None,
    after_updated_at=None,
    after_id: str | None = None,
) -> tuple[list[FileAsset], bool]:
    query = _asset_visibility_query_for_user(
        session,
        tenant_id=tenant_id,
        user_id=user_id,
        owner_type=owner_type,
        owner_id=owner_id,
        campaign_id=campaign_id,
        path_prefix=path_prefix,
        campaign_usage=campaign_usage,
        audit_relevant=audit_relevant,
        include_deleted=include_deleted,
        is_admin=is_admin,
    )
    if after_display_path is not None and after_updated_at is not None and after_id:
        query = query.filter(
            or_(
                FileAsset.display_path > after_display_path,
                and_(FileAsset.display_path == after_display_path, FileAsset.updated_at < after_updated_at),
                and_(FileAsset.display_path == after_display_path, FileAsset.updated_at == after_updated_at, FileAsset.id > after_id),
            )
        )
    rows = query.order_by(FileAsset.display_path.asc(), FileAsset.updated_at.desc(), FileAsset.id.asc()).limit(page_size + 1).all()
    return rows[:page_size], len(rows) > page_size


def _asset_visibility_query_for_user(
    session: Session,
    *,
    tenant_id: str,
    user_id: str,
    owner_type: str | None = None,
    owner_id: str | None = None,
    campaign_id: str | None = None,
    path_prefix: str | None = None,
    campaign_usage: str | None = None,
    audit_relevant: bool | None = None,
    include_deleted: bool = False,
    is_admin: bool = False,
):
    query = session.query(FileAsset).filter(FileAsset.tenant_id == tenant_id)
    if not include_deleted:
        query = query.filter(FileAsset.deleted_at.is_(None))
    if owner_type:
        query = query.filter(FileAsset.owner_type == owner_type)
        if owner_type == "user" and owner_id:
            query = query.filter(FileAsset.owner_user_id == owner_id)
        if owner_type == "group" and owner_id:
            query = query.filter(FileAsset.owner_group_id == owner_id)
    if campaign_id:
        campaign_share = exists().where(
            FileShare.tenant_id == tenant_id,
            FileShare.file_asset_id == FileAsset.id,
            FileShare.target_type == "campaign",
            FileShare.target_id == campaign_id,
            effective_file_share_clause(),
        )
        query = query.filter(campaign_share)
    elif not is_admin and not owner_type:
        group_ids = user_group_ids(session, tenant_id=tenant_id, user_id=user_id)
        active_share = exists().where(
            FileShare.tenant_id == tenant_id,
            FileShare.file_asset_id == FileAsset.id,
            effective_file_share_clause(),
            or_(
                (FileShare.target_type == "user") & (FileShare.target_id == user_id),
                (FileShare.target_type == "group") & (FileShare.target_id.in_(group_ids)),
                (FileShare.target_type == "tenant") & (FileShare.target_id == tenant_id),
            )
        )
        query = query.filter(
            or_(
                (FileAsset.owner_type == "user") & (FileAsset.owner_user_id == user_id),
                (FileAsset.owner_type == "group") & (FileAsset.owner_group_id.in_(group_ids)),
                active_share,
            )
        )
    if path_prefix:
        prefix = normalize_folder(path_prefix)
        if prefix:
            query = query.filter(FileAsset.display_path.like(f"{prefix}/%"))
    campaign_share_exists = exists().where(
        FileShare.tenant_id == tenant_id,
        FileShare.file_asset_id == FileAsset.id,
        FileShare.target_type == "campaign",
        effective_file_share_clause(),
    )
    campaign_use_exists = exists().where(
        CampaignAttachmentUse.tenant_id == tenant_id,
        CampaignAttachmentUse.file_asset_id == FileAsset.id,
    )
    if campaign_usage == "linked":
        query = query.filter(or_(campaign_share_exists, campaign_use_exists))
    elif campaign_usage == "unlinked":
        query = query.filter(~or_(campaign_share_exists, campaign_use_exists))

    if audit_relevant is not None:
        sent_use_exists = exists().where(
            CampaignAttachmentUse.tenant_id == tenant_id,
            CampaignAttachmentUse.file_asset_id == FileAsset.id,
            CampaignAttachmentUse.use_stage == "sent",
        )
        query = query.filter(sent_use_exists if audit_relevant else ~sent_use_exists)
    return query


def count_assets_for_user(
    session: Session,
    *,
    tenant_id: str,
    user_id: str,
    owner_type: str | None = None,
    owner_id: str | None = None,
    campaign_id: str | None = None,
    path_prefix: str | None = None,
    campaign_usage: str | None = None,
    audit_relevant: bool | None = None,
    include_deleted: bool = False,
    is_admin: bool = False,
) -> int:
    query = _asset_visibility_query_for_user(
        session,
        tenant_id=tenant_id,
        user_id=user_id,
        owner_type=owner_type,
        owner_id=owner_id,
        campaign_id=campaign_id,
        path_prefix=path_prefix,
        campaign_usage=campaign_usage,
        audit_relevant=audit_relevant,
        include_deleted=include_deleted,
        is_admin=is_admin,
    )
    return int(
        query.order_by(None)
        .with_entities(func.count(func.distinct(FileAsset.id)))
        .scalar()
        or 0
    )


def current_version_and_blob(session: Session, asset: FileAsset) -> tuple[FileVersion, FileBlob]:
    if not asset.current_version_id:
        raise FileStorageError("File has no current version")
    version = session.get(FileVersion, asset.current_version_id)
    if not version:
        raise FileStorageError("File version not found")
    blob = session.get(FileBlob, version.blob_id)
    if not blob:
        raise FileStorageError("File blob not found")
    return version, blob


def current_versions_and_blobs(session: Session, assets: Iterable[FileAsset]) -> dict[str, tuple[FileVersion, FileBlob]]:
    asset_list = list(assets)
    if not asset_list:
        return {}
    version_ids = [asset.current_version_id for asset in asset_list if asset.current_version_id]
    if len(version_ids) != len(asset_list):
        raise FileStorageError("File has no current version")

    rows: list[tuple[FileVersion, FileBlob]] = []
    for chunk in _chunks(version_ids):
        rows.extend(
            session.query(FileVersion, FileBlob)
            .join(FileBlob, FileBlob.id == FileVersion.blob_id)
            .filter(FileVersion.id.in_(chunk))
            .all()
        )
    by_version_id = {version.id: (version, blob) for version, blob in rows}
    result: dict[str, tuple[FileVersion, FileBlob]] = {}
    for asset in asset_list:
        version_blob = by_version_id.get(asset.current_version_id or "")
        if not version_blob:
            raise FileStorageError("File version not found")
        result[asset.id] = version_blob
    return result


def read_asset_bytes(session: Session, asset: FileAsset) -> tuple[bytes, FileVersion, FileBlob]:
    version, blob = current_version_and_blob(session, asset)
    backend = get_storage_backend()
    return read_verified_blob_bytes(blob, backend=backend), version, blob


def read_asset_version_bytes(
    session: Session,
    asset: FileAsset,
    version_id: str,
) -> tuple[bytes, FileVersion, FileBlob]:
    version = session.get(FileVersion, version_id)
    if version is None or version.file_asset_id != asset.id:
        raise FileStorageError("File version not found")
    blob = session.get(FileBlob, version.blob_id)
    if blob is None or blob.tenant_id != asset.tenant_id:
        raise FileStorageError("File blob not found")
    backend = get_storage_backend()
    return read_verified_blob_bytes(blob, backend=backend), version, blob


def share_file(
    session: Session,
    *,
    tenant_id: str,
    asset: FileAsset,
    target_type: str,
    target_id: str,
    permission: str,
    user_id: str,
    expires_at: datetime | None = None,
) -> FileShare:
    target_type = target_type.lower().strip()
    permission = permission.lower().strip()
    if asset.tenant_id != tenant_id:
        raise FileStorageError("File not found")
    if target_type not in {"user", "group", "campaign", "tenant"}:
        raise FileStorageError("Unsupported share target")
    if permission not in {"read", "write", "manage"}:
        raise FileStorageError("Unsupported file permission")
    _validate_share_expiry(expires_at)
    if target_type in {"user", "group", "tenant"}:
        ensure_share_target_exists(tenant_id=tenant_id, target_type=target_type, target_id=target_id)
    if target_type == "campaign":
        _ensure_campaign_exists(session, tenant_id=tenant_id, campaign_id=target_id)
    existing = (
        session.query(FileShare)
        .filter(
            FileShare.tenant_id == tenant_id,
            FileShare.file_asset_id == asset.id,
            FileShare.target_type == target_type,
            FileShare.target_id == target_id,
            FileShare.revoked_at.is_(None),
        )
        .one_or_none()
    )
    if existing:
        existing.permission = permission
        existing.expires_at = expires_at
        session.add(existing)
        return existing
    share = FileShare(
        tenant_id=tenant_id,
        file_asset_id=asset.id,
        target_type=target_type,
        target_id=target_id,
        permission=permission,
        created_by_user_id=user_id,
        expires_at=expires_at,
    )
    session.add(share)
    return share


def share_files(
    session: Session,
    *,
    tenant_id: str,
    assets: Iterable[FileAsset],
    target_type: str,
    target_id: str,
    permission: str,
    user_id: str,
    expires_at: datetime | None = None,
) -> list[FileShare]:
    target_type = target_type.lower().strip()
    permission = permission.lower().strip()
    if target_type not in {"user", "group", "campaign", "tenant"}:
        raise FileStorageError("Unsupported share target")
    if permission not in {"read", "write", "manage"}:
        raise FileStorageError("Unsupported file permission")
    _validate_share_expiry(expires_at)
    if target_type in {"user", "group", "tenant"}:
        ensure_share_target_exists(tenant_id=tenant_id, target_type=target_type, target_id=target_id)
    if target_type == "campaign":
        _ensure_campaign_exists(session, tenant_id=tenant_id, campaign_id=target_id)

    asset_list = list(assets)
    if not asset_list:
        return []
    for asset in asset_list:
        if asset.tenant_id != tenant_id:
            raise FileStorageError("File not found")

    existing_by_asset: dict[str, FileShare] = {}
    for share in session.query(FileShare).filter(
        FileShare.tenant_id == tenant_id,
        FileShare.file_asset_id.in_([asset.id for asset in asset_list]),
        FileShare.target_type == target_type,
        FileShare.target_id == target_id,
        FileShare.revoked_at.is_(None),
    ).all():
        existing_by_asset.setdefault(share.file_asset_id, share)

    shares: list[FileShare] = []
    for asset in asset_list:
        existing = existing_by_asset.get(asset.id)
        if existing:
            existing.permission = permission
            existing.expires_at = expires_at
            session.add(existing)
            shares.append(existing)
            continue
        share = FileShare(
            tenant_id=tenant_id,
            file_asset_id=asset.id,
            target_type=target_type,
            target_id=target_id,
            permission=permission,
            created_by_user_id=user_id,
            expires_at=expires_at,
        )
        session.add(share)
        shares.append(share)
    return shares


def list_file_shares(
    session: Session,
    *,
    tenant_id: str,
    asset_id: str,
    include_inactive: bool = False,
) -> list[FileShare]:
    query = session.query(FileShare).filter(
        FileShare.tenant_id == tenant_id,
        FileShare.file_asset_id == asset_id,
    )
    if not include_inactive:
        query = query.filter(effective_file_share_clause())
    return query.order_by(FileShare.created_at.desc(), FileShare.id.desc()).all()


def revoke_file_share(
    session: Session,
    *,
    tenant_id: str,
    asset_id: str,
    share_id: str,
    user_id: str,
) -> tuple[FileShare, bool]:
    share = (
        session.query(FileShare)
        .filter(
            FileShare.id == share_id,
            FileShare.tenant_id == tenant_id,
            FileShare.file_asset_id == asset_id,
        )
        .one_or_none()
    )
    if share is None:
        raise FileStorageError("File share not found")
    if share.revoked_at is not None:
        return share, False
    share.revoked_at = utcnow()
    share.revoked_by_user_id = user_id
    session.add(share)
    return share, True


def current_file_share_for_target(
    session: Session,
    *,
    tenant_id: str,
    asset_id: str,
    target_type: str,
    target_id: str,
) -> FileShare | None:
    return (
        session.query(FileShare)
        .filter(
            FileShare.tenant_id == tenant_id,
            FileShare.file_asset_id == asset_id,
            FileShare.target_type == target_type.lower().strip(),
            FileShare.target_id == target_id,
            FileShare.revoked_at.is_(None),
        )
        .order_by(FileShare.created_at.desc(), FileShare.id.desc())
        .first()
    )


def _validate_share_expiry(expires_at: datetime | None) -> None:
    if expires_at is None:
        return
    from govoplan_files.backend.storage.share_state import file_share_is_active

    candidate = FileShare(expires_at=expires_at)
    if not file_share_is_active(candidate):
        raise FileStorageError("File share expiry must be in the future")



def soft_delete_assets(session: Session, assets: Iterable[FileAsset]) -> int:
    count = 0
    now = utcnow()
    for asset in assets:
        if asset.deleted_at is None:
            asset.deleted_at = now
            session.add(asset)
            count += 1
    return count


def asset_is_audit_relevant(session: Session, asset: FileAsset) -> bool:
    return (
        session.query(CampaignAttachmentUse)
        .filter(CampaignAttachmentUse.file_asset_id == asset.id, CampaignAttachmentUse.use_stage == "sent")
        .first()
        is not None
    )


def _asset_owner_id(asset: FileAsset) -> str:
    if asset.owner_type == "user" and asset.owner_user_id:
        return asset.owner_user_id
    if asset.owner_type == "group" and asset.owner_group_id:
        return asset.owner_group_id
    raise FileStorageError("File has no valid owner")


def _chunks(values: list[str], size: int = 900):
    for index in range(0, len(values), size):
        yield values[index:index + size]


def _active_asset_exists(session: Session, *, tenant_id: str, owner_type: str, owner_id: str, path: str, exclude_asset_id: str | None = None) -> bool:
    return _active_asset_at_path(
        session,
        tenant_id=tenant_id,
        owner_type=owner_type,
        owner_id=owner_id,
        path=path,
        exclude_asset_id=exclude_asset_id,
    ) is not None


def _active_asset_at_path(session: Session, *, tenant_id: str, owner_type: str, owner_id: str, path: str, exclude_asset_id: str | None = None) -> FileAsset | None:
    query = _asset_query_for_owner(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id).filter(
        FileAsset.deleted_at.is_(None),
        FileAsset.display_path == normalize_logical_path(path),
    )
    if exclude_asset_id:
        query = query.filter(FileAsset.id != exclude_asset_id)
    return query.first()


def _soft_delete_conflicting_asset(session: Session, *, tenant_id: str, owner_type: str, owner_id: str, path: str, exclude_asset_id: str | None = None) -> None:
    asset = _active_asset_at_path(
        session,
        tenant_id=tenant_id,
        owner_type=owner_type,
        owner_id=owner_id,
        path=path,
        exclude_asset_id=exclude_asset_id,
    )
    if asset is not None:
        asset.deleted_at = utcnow()
        session.add(asset)


def _split_logical_path(path: str) -> tuple[str, str, str, str]:
    normalized = normalize_logical_path(path)
    logical = PurePosixPath(normalized)
    folder = "" if str(logical.parent) == "." else str(logical.parent)
    filename = logical.name
    suffixes = "".join(PurePosixPath(filename).suffixes)
    stem = filename[: -len(suffixes)] if suffixes else filename
    return folder, filename, stem, suffixes


def _candidate_renamed_path(path: str, counter: int) -> str:
    folder, _filename, stem, suffixes = _split_logical_path(path)
    suffix = " copy" if counter == 1 else f" copy {counter}"
    next_name = f"{stem}{suffix}{suffixes}"
    return normalize_logical_path(f"{folder}/{next_name}" if folder else next_name)


def _next_available_logical_path(
    session: Session,
    *,
    tenant_id: str,
    owner_type: str,
    owner_id: str,
    desired_path: str,
    reserved_paths: set[str] | None = None,
    exclude_asset_id: str | None = None,
) -> str:
    reserved = reserved_paths or set()
    desired = normalize_logical_path(desired_path)
    if desired not in reserved and not _active_asset_exists(
        session,
        tenant_id=tenant_id,
        owner_type=owner_type,
        owner_id=owner_id,
        path=desired,
        exclude_asset_id=exclude_asset_id,
    ):
        return desired
    counter = 1
    while True:
        candidate = _candidate_renamed_path(desired, counter)
        if candidate not in reserved and not _active_asset_exists(
            session,
            tenant_id=tenant_id,
            owner_type=owner_type,
            owner_id=owner_id,
            path=candidate,
            exclude_asset_id=exclude_asset_id,
        ):
            return candidate
        counter += 1


def _resolution_by_path(conflict_resolutions: Iterable[FileConflictResolution] | None) -> dict[str, FileConflictResolution]:
    result: dict[str, FileConflictResolution] = {}
    for item in conflict_resolutions or []:
        target_path = normalize_logical_path(item.target_path)
        action = item.action.lower().strip()
        if action not in {"overwrite", "rename", "skip"}:
            raise FileStorageError("Unsupported conflict resolution")
        result[target_path] = FileConflictResolution(target_path=target_path, action=action, new_path=item.new_path)
    return result


def _normalize_conflict_strategy(strategy: str | None) -> str:
    normalized = (strategy or "reject").lower().strip()
    if normalized not in {"reject", "overwrite", "rename", "skip"}:
        raise FileStorageError("Unsupported conflict strategy")
    return normalized


def _next_version_number(session: Session, asset_id: str) -> int:
    row = (
        session.query(FileVersion.version_number)
        .filter(FileVersion.file_asset_id == asset_id)
        .order_by(FileVersion.version_number.desc())
        .first()
    )
    return (int(row[0]) if row else 0) + 1


def _source_identity(provenance: dict[str, Any] | None) -> tuple[object, ...] | None:
    if not provenance:
        return None
    connector_id = _clean_identity(provenance.get("connector_id"))
    provider = _clean_identity(provenance.get("provider"))
    external_id = _clean_identity(provenance.get("external_id"))
    if connector_id and external_id:
        return ("external_id", connector_id, provider, external_id)
    external_path = _clean_identity(provenance.get("external_path"))
    metadata = provenance.get("metadata") if isinstance(provenance.get("metadata"), dict) else {}
    library_id = _clean_identity(metadata.get("library_id") or metadata.get("profile_id") or metadata.get("share"))
    if connector_id and external_path:
        return ("external_path", connector_id, provider, library_id, external_path)
    return None


def _clean_identity(value: object) -> str | None:
    if value is None:
        return None
    text = str(value).strip()
    return text or None


def _copy_asset_to_path(
    session: Session,
    asset: FileAsset,
    *,
    tenant_id: str,
    target_owner_type: str,
    target_owner_id: str,
    target_path: str,
    user_id: str,
) -> FileAsset:
    version, blob = current_version_and_blob(session, asset)
    blob.ref_count += 1
    session.add(blob)
    normalized_path = normalize_logical_path(target_path)
    copied = FileAsset(
        tenant_id=tenant_id,
        owner_type=target_owner_type,
        owner_user_id=target_owner_id if target_owner_type == "user" else None,
        owner_group_id=target_owner_id if target_owner_type == "group" else None,
        display_path=normalized_path,
        filename=filename_from_path(normalized_path),
        description=asset.description,
        created_by_user_id=user_id,
        metadata_=dict(asset.metadata_ or {}),
    )
    session.add(copied)
    session.flush()
    copied_version = FileVersion(
        tenant_id=tenant_id,
        file_asset_id=copied.id,
        blob_id=blob.id,
        version_number=1,
        filename_at_upload=version.filename_at_upload,
        display_path_at_upload=normalized_path,
        content_type=version.content_type,
        size_bytes=blob.size_bytes,
        checksum_sha256=blob.checksum_sha256,
        created_by_user_id=user_id,
    )
    session.add(copied_version)
    session.flush()
    copied.current_version_id = copied_version.id
    session.add(copied)
    return copied


def rename_asset(asset: FileAsset, *, new_path: str) -> None:
    normalized = normalize_logical_path(new_path)
    asset.display_path = normalized
    asset.filename = filename_from_path(normalized)