diff --git a/applications/bulk_document_download/handler.py b/applications/bulk_document_download/handler.py index c5af7014c..a0d90798c 100644 --- a/applications/bulk_document_download/handler.py +++ b/applications/bulk_document_download/handler.py @@ -29,7 +29,7 @@ from infrastructure.email.ses_smtp_email_sender import SesSmtpEmailSender from infrastructure.postgres.config import PostgresConfig from infrastructure.postgres.engine import make_engine, make_session from infrastructure.s3.s3_client import S3Client -from infrastructure.s3.s3_document_bytes_reader import S3DocumentBytesReader +from infrastructure.s3.s3_document_downloader import S3DocumentDownloader from orchestration.bulk_document_download_orchestrator import ( BulkDocumentDownloadOrchestrator, ) @@ -72,7 +72,7 @@ def handler(body: dict[str, Any], context: Any) -> dict[str, Any]: orchestrator = BulkDocumentDownloadOrchestrator( uploaded_files=UploadedFilePostgresRepository(session), addresses=LandlordAddressResolver(session, landlord_property_ids), - documents=S3DocumentBytesReader(boto_s3), + documents=S3DocumentDownloader(boto_s3), packages=S3Client(boto_s3, os.environ["DOCUMENT_EXPORTS_BUCKET"]), email=_email_sender(), url_ttl_seconds=_URL_TTL_SECONDS, diff --git a/backend/app/documents/download_tasks.py b/backend/app/documents/download_tasks.py index 9c4940407..272da5e47 100644 --- a/backend/app/documents/download_tasks.py +++ b/backend/app/documents/download_tasks.py @@ -87,16 +87,24 @@ class DocumentDownloadTasks: def create_download_subtask( self, task_id: UUID, subtask_id: UUID, recipe: dict[str, Any] - ) -> None: + ) -> bool: """Pre-create the single ``waiting`` sub_task under the app-owned task, pinning the recipe (landlord_property_ids, recipient_email, package_name) onto its inputs — the worker's reproducible instructions - (ADR-0055/0060).""" + (ADR-0055/0060). + + The insert is **conditional on the task having no sub_task yet**, so the + DB — not a prior read — arbitrates a double-submit race: two concurrent + requests can both pass the ``already_distributed`` check, but only one + insert lands. Returns whether this call created the sub_task.""" now = datetime.now(timezone.utc) - self._session.connection().execute( + result = self._session.connection().execute( text( "INSERT INTO sub_task (id, task_id, status, inputs, updated_at)" - " VALUES (:id, :task_id, 'waiting', :inputs, :updated_at)" + " SELECT :id, :task_id, 'waiting', :inputs, :updated_at" + " WHERE NOT EXISTS (" + " SELECT 1 FROM sub_task WHERE task_id = :task_id" + " )" ), { "id": subtask_id, @@ -106,3 +114,4 @@ class DocumentDownloadTasks: }, ) self._session.commit() + return result.rowcount == 1 diff --git a/backend/app/documents/router.py b/backend/app/documents/router.py index 06bc24723..52d57d06a 100644 --- a/backend/app/documents/router.py +++ b/backend/app/documents/router.py @@ -52,6 +52,11 @@ def get_requesting_user_email(user: Any = Depends(validate_jwt_token)) -> str: email: Any = getattr(user, "email", None) if email is None and isinstance(user, dict): email = cast(dict[str, Any], user).get("email") + if not email and get_settings().ENVIRONMENT == "local": + # Local `get_user` returns a dummy dict with no email; a placeholder lets + # the route be exercised end-to-end on a dev machine (email is a no-op + # locally anyway). + email = "local-dev@example.com" if not email: raise HTTPException( status_code=400, @@ -133,7 +138,17 @@ async def bulk_download( "recipient_email": recipient_email, "package_name": f"portfolio-{portfolio_id}-{body.task_id}", } - tasks.create_download_subtask(body.task_id, subtask_id, recipe) + created = tasks.create_download_subtask(body.task_id, subtask_id, recipe) + if not created: + # Lost a double-submit race — another request already created the + # sub_task. The DB arbitrated; don't enqueue a second job. + raise HTTPException( + status_code=409, + detail=( + f"Task {body.task_id} already has a sub_task — a download has " + "already been started for it." + ), + ) send_message( json.dumps({"task_id": str(body.task_id), "subtask_id": str(subtask_id)}) diff --git a/domain/bulk_document_download/package_plan.py b/domain/bulk_document_download/package_plan.py index 535928596..ba12e3560 100644 --- a/domain/bulk_document_download/package_plan.py +++ b/domain/bulk_document_download/package_plan.py @@ -33,11 +33,13 @@ class ResolvedDocument: @dataclass(frozen=True) class PackageEntry: """One file to place in the Download Package: where it lands in the ZIP and - the S3 object to stream in from.""" + the S3 object to stream in from. Carries the property identity so a + read-time failure can be reported as a SkippedDocument (ADR-0060).""" zip_path: str s3_bucket: str s3_key: str + landlord_property_id: str @dataclass(frozen=True) @@ -95,6 +97,7 @@ def plan_download_package( f"{document.document_type}{_extension_of(document.s3_key)}", s3_bucket=document.s3_bucket, s3_key=document.s3_key, + landlord_property_id=document.landlord_property_id, ) for document in latest_by_group.values() ] diff --git a/infrastructure/s3/s3_document_bytes_reader.py b/infrastructure/s3/s3_document_bytes_reader.py deleted file mode 100644 index 2e0739249..000000000 --- a/infrastructure/s3/s3_document_bytes_reader.py +++ /dev/null @@ -1,21 +0,0 @@ -"""Reads document bytes from arbitrary S3 buckets (ADR-0060). - -Uploaded files live across many source buckets (each `uploaded_files` row -carries its own `s3_file_bucket`), so — unlike the bucket-bound `S3Client` — -this reader takes the bucket per call. Used by the Bulk Document Download -orchestrator to stream each chosen file into the ZIP. -""" - -from __future__ import annotations - -from typing import Any - - -class S3DocumentBytesReader: - def __init__(self, boto_s3_client: Any) -> None: - self._client = boto_s3_client - - def read(self, bucket: str, key: str) -> bytes: - response: dict[str, Any] = self._client.get_object(Bucket=bucket, Key=key) - body: bytes = response["Body"].read() - return body diff --git a/infrastructure/s3/s3_document_downloader.py b/infrastructure/s3/s3_document_downloader.py new file mode 100644 index 000000000..e1ee7bd12 --- /dev/null +++ b/infrastructure/s3/s3_document_downloader.py @@ -0,0 +1,20 @@ +"""Streams uploaded documents from arbitrary S3 buckets to local disk (ADR-0060). + +Uploaded files live across many source buckets (each `uploaded_files` row +carries its own `s3_file_bucket`), so — unlike the bucket-bound `S3Client` — +this takes the bucket per call. Downloading to a file (rather than reading the +object into memory) keeps a single multi-GB member off the heap when it is added +to the on-disk ZIP. +""" + +from __future__ import annotations + +from typing import Any + + +class S3DocumentDownloader: + def __init__(self, boto_s3_client: Any) -> None: + self._client = boto_s3_client + + def download(self, bucket: str, key: str, dest_path: str) -> None: + self._client.download_file(Bucket=bucket, Key=key, Filename=dest_path) diff --git a/orchestration/bulk_document_download_orchestrator.py b/orchestration/bulk_document_download_orchestrator.py index ba979a1c6..c3c0db2cd 100644 --- a/orchestration/bulk_document_download_orchestrator.py +++ b/orchestration/bulk_document_download_orchestrator.py @@ -38,8 +38,8 @@ class AddressResolver(Protocol): def address_for(self, landlord_property_id: str) -> str: ... -class DocumentBytesReader(Protocol): - def read(self, bucket: str, key: str) -> bytes: ... +class DocumentDownloader(Protocol): + def download(self, bucket: str, key: str, dest_path: str) -> None: ... @dataclass(frozen=True) @@ -58,7 +58,7 @@ class BulkDocumentDownloadOrchestrator: *, uploaded_files: UploadedFileReader, addresses: AddressResolver, - documents: DocumentBytesReader, + documents: DocumentDownloader, packages: S3Client, email: EmailSender, url_ttl_seconds: int, @@ -83,36 +83,44 @@ class BulkDocumentDownloadOrchestrator: resolved = [self._resolve(file) for file in files] plan = plan_download_package([r for r in resolved if r is not None]) + skipped: list[SkippedDocument] = list(plan.skipped) + if not plan.entries: - # Nothing to package (ADR-0060): a recorded failure, not an empty - # ZIP emailed to the user. The skip detail rides on the failure. - raise SubTaskFailure( - "no documents resolved for the selection", - details={ - "skipped": [ - { - "landlord_property_id": s.landlord_property_id, - "s3_key": s.s3_key, - "reason": s.reason, - } - for s in plan.skipped - ] - }, - ) + raise self._empty_failure(skipped) # Stream to a /tmp file and multipart-upload it (ADR-0060): a Download - # Package can be several GB, so it is never held whole in memory — each - # source file is read, written into the on-disk ZIP, and released; the - # finished archive is streamed up from disk. + # Package can be several GB, so it is never held whole in memory. Each + # source file is streamed to its own temp file, added to the on-disk ZIP, + # then deleted; the finished archive is streamed up from disk. Best-effort + # (ADR-0060): a file that can't be read — deleted, or in a bucket the role + # can't reach — becomes a SkippedDocument, it does not fail the package. key = f"{self._package_key_prefix}/{package_name}.zip" + included = 0 with tempfile.TemporaryDirectory() as tmp_dir: zip_path = os.path.join(tmp_dir, "package.zip") with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as archive: - for entry in plan.entries: - archive.writestr( - entry.zip_path, - self._documents.read(entry.s3_bucket, entry.s3_key), - ) + for index, entry in enumerate(plan.entries): + member_path = os.path.join(tmp_dir, f"member-{index}") + try: + self._documents.download( + entry.s3_bucket, entry.s3_key, member_path + ) + except Exception: + skipped.append( + SkippedDocument( + landlord_property_id=entry.landlord_property_id, + s3_key=entry.s3_key, + reason="unreadable", + ) + ) + continue + archive.write(member_path, arcname=entry.zip_path) + os.remove(member_path) + included += 1 + + if included == 0: + # Every file that resolved failed to read — no package to send. + raise self._empty_failure(skipped) self._packages.upload_file(zip_path, key) url = self._packages.generate_presigned_url(key, self._url_ttl_seconds) @@ -121,7 +129,7 @@ class BulkDocumentDownloadOrchestrator: subject="Your document download is ready", body=( f"Your document download is ready. It contains " - f"{len(plan.entries)} document(s).\n\n" + f"{included} document(s).\n\n" f"Download it here (link valid for " f"{self._url_ttl_seconds // 60} minutes):\n{url}" ), @@ -129,8 +137,26 @@ class BulkDocumentDownloadOrchestrator: return DownloadPackageResult( presigned_url=url, package_s3_key=key, - included=len(plan.entries), - skipped=plan.skipped, + included=included, + skipped=tuple(skipped), + ) + + def _empty_failure(self, skipped: list[SkippedDocument]) -> SubTaskFailure: + """A recorded failure when the selection yields no readable documents + (ADR-0060) — never an empty ZIP emailed to the user. The skip detail + rides on the failure.""" + return SubTaskFailure( + "no documents could be packaged for the selection", + details={ + "skipped": [ + { + "landlord_property_id": s.landlord_property_id, + "s3_key": s.s3_key, + "reason": s.reason, + } + for s in skipped + ] + }, ) def _resolve(self, file: UploadedFile) -> Optional[ResolvedDocument]: @@ -142,5 +168,9 @@ class BulkDocumentDownloadOrchestrator: document_type=file.file_type, s3_bucket=file.s3_file_bucket, s3_key=file.s3_file_key, + # TODO: the cast is only needed because UploadedFile types + # s3_upload_timestamp as `object` (though it is NOT NULL datetime in + # the DB). Type that column `datetime` on the table so callers don't + # each cast — see PR #1498 review. uploaded_at=cast(datetime, file.s3_upload_timestamp), ) diff --git a/tests/infrastructure/test_s3_document_bytes_reader.py b/tests/infrastructure/test_s3_document_bytes_reader.py deleted file mode 100644 index 8bbcea895..000000000 --- a/tests/infrastructure/test_s3_document_bytes_reader.py +++ /dev/null @@ -1,29 +0,0 @@ -"""The multi-bucket document reader (ADR-0060) — reads object bytes from -whichever source bucket an uploaded file lives in.""" - -from collections.abc import Iterator - -import pytest -from moto import mock_aws - -from infrastructure.s3.s3_document_bytes_reader import S3DocumentBytesReader -from tests.infrastructure import make_boto_client - - -@pytest.fixture -def reader() -> Iterator[S3DocumentBytesReader]: - with mock_aws(): - boto_client = make_boto_client("s3") - boto_client.create_bucket(Bucket="src-a") - boto_client.create_bucket(Bucket="src-b") - boto_client.put_object(Bucket="src-a", Key="surveys/one.pdf", Body=b"AAA") - boto_client.put_object(Bucket="src-b", Key="photos/two.zip", Body=b"BBB") - yield S3DocumentBytesReader(boto_client) - - -def test_reads_bytes_from_the_named_source_bucket( - reader: S3DocumentBytesReader, -) -> None: - # act / assert — each object is read from its own bucket. - assert reader.read("src-a", "surveys/one.pdf") == b"AAA" - assert reader.read("src-b", "photos/two.zip") == b"BBB" diff --git a/tests/infrastructure/test_s3_document_downloader.py b/tests/infrastructure/test_s3_document_downloader.py new file mode 100644 index 000000000..58753d2b1 --- /dev/null +++ b/tests/infrastructure/test_s3_document_downloader.py @@ -0,0 +1,36 @@ +"""The multi-bucket document downloader (ADR-0060) — streams object bytes from +whichever source bucket an uploaded file lives in, to local disk.""" + +from collections.abc import Iterator +from pathlib import Path + +import pytest +from moto import mock_aws + +from infrastructure.s3.s3_document_downloader import S3DocumentDownloader +from tests.infrastructure import make_boto_client + + +@pytest.fixture +def downloader() -> Iterator[S3DocumentDownloader]: + with mock_aws(): + boto_client = make_boto_client("s3") + boto_client.create_bucket(Bucket="src-a") + boto_client.create_bucket(Bucket="src-b") + boto_client.put_object(Bucket="src-a", Key="surveys/one.pdf", Body=b"AAA") + boto_client.put_object(Bucket="src-b", Key="photos/two.zip", Body=b"BBB") + yield S3DocumentDownloader(boto_client) + + +def test_downloads_bytes_from_the_named_source_bucket( + downloader: S3DocumentDownloader, tmp_path: Path +) -> None: + # act — each object streams from its own bucket to a local file. + a = tmp_path / "a" + b = tmp_path / "b" + downloader.download("src-a", "surveys/one.pdf", str(a)) + downloader.download("src-b", "photos/two.zip", str(b)) + + # assert + assert a.read_bytes() == b"AAA" + assert b.read_bytes() == b"BBB" diff --git a/tests/orchestration/test_bulk_document_download_orchestrator.py b/tests/orchestration/test_bulk_document_download_orchestrator.py index 40c483d80..d9b3a0d23 100644 --- a/tests/orchestration/test_bulk_document_download_orchestrator.py +++ b/tests/orchestration/test_bulk_document_download_orchestrator.py @@ -45,8 +45,11 @@ class _FakeDocuments: def __init__(self, blobs: dict[tuple[str, str], bytes]) -> None: self._blobs = blobs - def read(self, bucket: str, key: str) -> bytes: - return self._blobs[(bucket, key)] + def download(self, bucket: str, key: str, dest_path: str) -> None: + # KeyError for an object we don't hold — stands in for a missing/ + # unreadable source object. + with open(dest_path, "wb") as handle: + handle.write(self._blobs[(bucket, key)]) class _RecordingEmail: @@ -176,3 +179,66 @@ def test_packages_the_good_documents_and_reports_the_skipped_ones( archive = packages.get_object(result.package_s3_key) with zipfile.ZipFile(io.BytesIO(archive)) as zf: assert zf.namelist() == ["12 Oak Street (LP1)/site_note.pdf"] + + +def test_a_file_that_cannot_be_read_is_skipped_not_fatal(packages: S3Client) -> None: + # Arrange — two documents for one property; the photo pack's object is + # missing (download raises), the site note reads fine. Best-effort (ADR-0060): + # the run must still deliver the readable file and report the other. + good = _uploaded_file("LP1", "src-bucket", "surveys/a.pdf") + missing = _uploaded_file( + "LP1", "src-bucket", "surveys/gone.pdf", file_type=FileTypeEnum.PHOTO_PACK + ) + email = _RecordingEmail() + orchestrator = BulkDocumentDownloadOrchestrator( + uploaded_files=_FakeUploadedFiles([good, missing]), + addresses=_FakeAddresses({"LP1": "12 Oak Street"}), + documents=_FakeDocuments({("src-bucket", "surveys/a.pdf"): b"PDF-BYTES"}), + packages=packages, + email=email, + url_ttl_seconds=3600, + ) + + # Act + result = orchestrator.run( + landlord_property_ids=["LP1"], + recipient_email="user@example.com", + package_name="partial", + ) + + # Assert — the readable file is delivered; the missing one is reported. + assert result.included == 1 + assert [(s.s3_key, s.reason) for s in result.skipped] == [ + ("surveys/gone.pdf", "unreadable") + ] + assert len(email.sent) == 1 + archive = packages.get_object(result.package_s3_key) + with zipfile.ZipFile(io.BytesIO(archive)) as zf: + assert zf.namelist() == ["12 Oak Street (LP1)/site_note.pdf"] + + +def test_the_address_fallback_string_becomes_the_folder_name( + packages: S3Client, +) -> None: + # Arrange — a property whose address resolved to the fallback string. + doc = _uploaded_file("LP9", "src-bucket", "surveys/a.pdf") + orchestrator = BulkDocumentDownloadOrchestrator( + uploaded_files=_FakeUploadedFiles([doc]), + addresses=_FakeAddresses({"LP9": "address unavailable"}), + documents=_FakeDocuments({("src-bucket", "surveys/a.pdf"): b"PDF-BYTES"}), + packages=packages, + email=_RecordingEmail(), + url_ttl_seconds=3600, + ) + + # Act + result = orchestrator.run( + landlord_property_ids=["LP9"], + recipient_email="user@example.com", + package_name="fallback", + ) + + # Assert — the fallback flows through to a usable folder name. + archive = packages.get_object(result.package_s3_key) + with zipfile.ZipFile(io.BytesIO(archive)) as zf: + assert zf.namelist() == ["address unavailable (LP9)/site_note.pdf"]