diff --git a/src/govoplan_files/backend/router.py b/src/govoplan_files/backend/router.py index cf56016..9e35b42 100644 --- a/src/govoplan_files/backend/router.py +++ b/src/govoplan_files/backend/router.py @@ -991,85 +991,115 @@ def _asset_response(session: Session, asset: FileAsset, *, include_shares: bool ) -def _asset_list_response(session: Session, assets: list[FileAsset], *, include_shares: bool = False) -> list[FileAssetResponse]: - if not assets: - return [] +def _file_share_response(row: FileShare) -> FileShareResponse: + return FileShareResponse( + id=row.id, + target_type=row.target_type, + target_id=row.target_id, + permission=row.permission, + created_at=row.created_at.isoformat(), + revoked_at=row.revoked_at.isoformat() if row.revoked_at else None, + ) - asset_ids = [asset.id for asset in assets] + +def _asset_versions_and_blobs(session: Session, assets: list[FileAsset]) -> dict[str, tuple[FileVersion, FileBlob]]: version_ids = [asset.current_version_id for asset in assets if asset.current_version_id] if len(version_ids) != len(assets): raise FileStorageError("File has no current version") - - version_blob_rows: list[tuple[FileVersion, FileBlob]] = [] + rows: list[tuple[FileVersion, FileBlob]] = [] for chunk in _chunks(version_ids): - version_blob_rows.extend( + rows.extend( session.query(FileVersion, FileBlob) .join(FileBlob, FileBlob.id == FileVersion.blob_id) .filter(FileVersion.id.in_(chunk)) .all() ) - versions_by_id = {version.id: version for version, _blob in version_blob_rows} - blobs_by_version_id = {version.id: blob for version, blob in version_blob_rows} + return {version.id: (version, blob) for version, blob in rows} + +def _asset_shares_by_id(session: Session, asset_ids: list[str], *, include_shares: bool) -> dict[str, list[FileShareResponse]]: shares_by_asset_id: dict[str, list[FileShareResponse]] = {asset_id: [] for asset_id in asset_ids} - if include_shares: - for chunk in _chunks(asset_ids): - share_rows = ( - session.query(FileShare) - .filter(FileShare.file_asset_id.in_(chunk)) - .order_by(FileShare.created_at.desc()) - .all() - ) - for row in share_rows: - shares_by_asset_id.setdefault(row.file_asset_id, []).append( - FileShareResponse( - id=row.id, - target_type=row.target_type, - target_id=row.target_id, - permission=row.permission, - created_at=row.created_at.isoformat(), - revoked_at=row.revoked_at.isoformat() if row.revoked_at else None, - ) - ) + if not include_shares: + return shares_by_asset_id + for chunk in _chunks(asset_ids): + rows = ( + session.query(FileShare) + .filter(FileShare.file_asset_id.in_(chunk)) + .order_by(FileShare.created_at.desc()) + .all() + ) + for row in rows: + shares_by_asset_id.setdefault(row.file_asset_id, []).append(_file_share_response(row)) + return shares_by_asset_id + +def _sent_campaign_asset_ids(session: Session, asset_ids: list[str]) -> set[str]: sent_asset_ids: set[str] = set() for chunk in _chunks(asset_ids): - sent_rows = ( + rows = ( session.query(CampaignAttachmentUse.file_asset_id) .filter(CampaignAttachmentUse.file_asset_id.in_(chunk), CampaignAttachmentUse.use_stage == "sent") .distinct() .all() ) - sent_asset_ids.update(row[0] for row in sent_rows) + sent_asset_ids.update(row[0] for row in rows) + return sent_asset_ids + + +def _loaded_asset_response( + asset: FileAsset, + *, + version: FileVersion, + blob: FileBlob, + shares: list[FileShareResponse], + sent_asset_ids: set[str], +) -> FileAssetResponse: + metadata = asset.metadata_ or {} + return FileAssetResponse( + id=asset.id, + tenant_id=asset.tenant_id, + owner_type=asset.owner_type, + owner_id=_owner_id(asset), + display_path=asset.display_path, + filename=asset.filename, + description=asset.description, + size_bytes=blob.size_bytes, + content_type=blob.content_type, + checksum_sha256=blob.checksum_sha256, + version_id=version.id, + created_at=asset.created_at.isoformat(), + updated_at=asset.updated_at.isoformat(), + deleted_at=asset.deleted_at.isoformat() if asset.deleted_at else None, + audit_relevant=asset.id in sent_asset_ids, + metadata=metadata, + source_provenance=source_provenance_from_metadata(metadata), + source_revision=source_revision_from_metadata(metadata), + shares=shares, + ) + + +def _asset_list_response(session: Session, assets: list[FileAsset], *, include_shares: bool = False) -> list[FileAssetResponse]: + if not assets: + return [] + + asset_ids = [asset.id for asset in assets] + versions_and_blobs = _asset_versions_and_blobs(session, assets) + shares_by_asset_id = _asset_shares_by_id(session, asset_ids, include_shares=include_shares) + sent_asset_ids = _sent_campaign_asset_ids(session, asset_ids) responses: list[FileAssetResponse] = [] for asset in assets: - version = versions_by_id.get(asset.current_version_id or "") - blob = blobs_by_version_id.get(asset.current_version_id or "") - if not version or not blob: + version_and_blob = versions_and_blobs.get(asset.current_version_id or "") + if version_and_blob is None: raise FileStorageError("File version not found") - metadata = asset.metadata_ or {} + version, blob = version_and_blob responses.append( - FileAssetResponse( - id=asset.id, - tenant_id=asset.tenant_id, - owner_type=asset.owner_type, - owner_id=_owner_id(asset), - display_path=asset.display_path, - filename=asset.filename, - description=asset.description, - size_bytes=blob.size_bytes, - content_type=blob.content_type, - checksum_sha256=blob.checksum_sha256, - version_id=version.id, - created_at=asset.created_at.isoformat(), - updated_at=asset.updated_at.isoformat(), - deleted_at=asset.deleted_at.isoformat() if asset.deleted_at else None, - audit_relevant=asset.id in sent_asset_ids, - metadata=metadata, - source_provenance=source_provenance_from_metadata(metadata), - source_revision=source_revision_from_metadata(metadata), + _loaded_asset_response( + asset, + version=version, + blob=blob, shares=shares_by_asset_id.get(asset.id, []), + sent_asset_ids=sent_asset_ids, ) ) return responses @@ -1499,6 +1529,104 @@ def _entry_matches_delta_scope( ) +def _decode_files_delta_watermark(since: str) -> int: + try: + return decode_sequence_watermark(since) + except ValueError as exc: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc + + +def _changed_delta_resource_ids(entries: list[ChangeSequenceEntry]) -> tuple[list[str], list[str]]: + file_ids = list(dict.fromkeys(entry.resource_id for entry in entries if entry.resource_type == "file")) + folder_ids = list(dict.fromkeys(entry.resource_id for entry in entries if entry.resource_type == "folder")) + return file_ids, folder_ids + + +def _visible_assets_for_delta( + session: Session, + *, + principal: ApiPrincipal, + owner_type: Literal["user", "group"] | None, + owner_id: str | None, + campaign_id: str | None, + path_prefix: str | None, + changed_file_ids: list[str], +) -> dict[str, FileAsset]: + return { + asset.id: asset + for asset in list_assets_for_user( + session, + tenant_id=principal.tenant_id, + user_id=principal.user.id, + owner_type=owner_type, + owner_id=owner_id, + campaign_id=campaign_id, + path_prefix=path_prefix, + is_admin=_is_admin(principal), + ) + if asset.id in changed_file_ids + } + + +def _changed_visible_folders_for_delta( + session: Session, + *, + principal: ApiPrincipal, + owner_type: Literal["user", "group"] | None, + owner_id: str | None, + path_prefix: str | None, + changed_folder_ids: list[str], +) -> dict[str, FileFolder]: + return { + folder.id: folder + for folder in _visible_folders_for_delta( + session, + principal=principal, + owner_type=owner_type, + owner_id=owner_id, + path_prefix=path_prefix, + ) + if folder.id in changed_folder_ids + } + + +def _deleted_delta_items( + session: Session, + *, + principal: ApiPrincipal, + entries: list[ChangeSequenceEntry], + visible_assets: dict[str, FileAsset], + visible_folders: dict[str, FileFolder], + owner_type: Literal["user", "group"] | None, + owner_id: str | None, + campaign_id: str | None, + path_prefix: str | None, +) -> list[DeltaDeletedItem]: + deleted: dict[tuple[str, str], DeltaDeletedItem] = {} + for entry in entries: + if entry.resource_type == "file" and entry.resource_id in visible_assets: + continue + if entry.resource_type == "folder" and entry.resource_id in visible_folders: + continue + if not _entry_matches_delta_scope( + session, + principal, + entry, + owner_type=owner_type, + owner_id=owner_id, + campaign_id=campaign_id, + path_prefix=path_prefix, + ): + continue + deleted[(entry.resource_type, entry.resource_id)] = DeltaDeletedItem( + id=entry.resource_id, + resource_type=entry.resource_type, + revision=encode_sequence_watermark(entry.id), + deleted_at=entry.created_at if entry.operation == "deleted" else None, + ) + return list(deleted.values()) + + def _files_delta_response( session: Session, *, @@ -1510,10 +1638,7 @@ def _files_delta_response( since: str, limit: int, ) -> FileDeltaResponse: - try: - since_sequence = decode_sequence_watermark(since) - except ValueError as exc: - raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc + since_sequence = _decode_files_delta_watermark(since) if sequence_watermark_is_expired( session, since=since_sequence, @@ -1541,65 +1666,41 @@ def _files_delta_response( has_more = len(entries_plus_one) > limit entries = entries_plus_one[:limit] - changed_file_ids = list(dict.fromkeys(entry.resource_id for entry in entries if entry.resource_type == "file")) - changed_folder_ids = list(dict.fromkeys(entry.resource_id for entry in entries if entry.resource_type == "folder")) - - visible_assets = { - asset.id: asset - for asset in list_assets_for_user( - session, - tenant_id=principal.tenant_id, - user_id=principal.user.id, - owner_type=owner_type, - owner_id=owner_id, - campaign_id=campaign_id, - path_prefix=path_prefix, - is_admin=_is_admin(principal), - ) - if asset.id in changed_file_ids - } - visible_folders = { - folder.id: folder - for folder in _visible_folders_for_delta( - session, - principal=principal, - owner_type=owner_type, - owner_id=owner_id, - path_prefix=path_prefix, - ) - if folder.id in changed_folder_ids - } - - deleted: dict[tuple[str, str], DeltaDeletedItem] = {} - for entry in entries: - is_visible = ( - (entry.resource_type == "file" and entry.resource_id in visible_assets) - or (entry.resource_type == "folder" and entry.resource_id in visible_folders) - ) - if is_visible: - continue - if not _entry_matches_delta_scope( - session, - principal, - entry, - owner_type=owner_type, - owner_id=owner_id, - campaign_id=campaign_id, - path_prefix=path_prefix, - ): - continue - deleted[(entry.resource_type, entry.resource_id)] = DeltaDeletedItem( - id=entry.resource_id, - resource_type=entry.resource_type, - revision=encode_sequence_watermark(entry.id), - deleted_at=entry.created_at if entry.operation == "deleted" else None, - ) + changed_file_ids, changed_folder_ids = _changed_delta_resource_ids(entries) + visible_assets = _visible_assets_for_delta( + session, + principal=principal, + owner_type=owner_type, + owner_id=owner_id, + campaign_id=campaign_id, + path_prefix=path_prefix, + changed_file_ids=changed_file_ids, + ) + visible_folders = _changed_visible_folders_for_delta( + session, + principal=principal, + owner_type=owner_type, + owner_id=owner_id, + path_prefix=path_prefix, + changed_folder_ids=changed_folder_ids, + ) + deleted = _deleted_delta_items( + session, + principal=principal, + entries=entries, + visible_assets=visible_assets, + visible_folders=visible_folders, + owner_type=owner_type, + owner_id=owner_id, + campaign_id=campaign_id, + path_prefix=path_prefix, + ) watermark = encode_sequence_watermark(entries[-1].id) if has_more and entries else _files_delta_watermark(session, principal.tenant_id) return FileDeltaResponse( files=_asset_list_response(session, list(visible_assets.values()), include_shares=True), folders=[_folder_response(folder) for folder in visible_folders.values()], - deleted=list(deleted.values()), + deleted=deleted, watermark=watermark, has_more=has_more, full=False,