diff --git a/CONTEXT.md b/CONTEXT.md index 1204aa9d9..34d380249 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -108,7 +108,7 @@ The single ZIP archive a **Bulk Document Download** produces: one folder per pro _Avoid_: export (that is the plan/scenario XLSX export), bundle, archive (ambiguous), zip (the format, not the concept) **Bulk Document Download**: -The feature/job that assembles a **Download Package** for a chosen set of properties (the union of one or more whole HubSpot projects, selected by `project_code`, and a hand-picked `landlord_property_id` list) and emails the requester a link. Runs on the app-owned-task + attach-mode lane (ADR-0055): the FastAPI route resolves the selection to a distinct `landlord_property_id` set — project codes are expanded via `hubspot_deal_data` (which holds the project↔property grain), hand-picked ids are taken as given — pins that set and the recipient email onto the `sub_task`, and the `applications/bulk_document_download` Lambda builds the package, writes the URL to `sub_task.outputs`, and emails it (ADR-0059/0060). `landlord_property_id` is the selection key throughout because `uploaded_files` is matched on it (it has no `property_id`). Files are matched by `landlord_property_id` (the missing `property_id` on `uploaded_files` is a known future gap). Selection is capped by property count at the trigger. +The feature/job that assembles a **Download Package** for a chosen set of properties (the union of one or more whole HubSpot projects, selected by `project_code`, and a hand-picked `landlord_property_id` list) and emails the requester a link. Runs on the app-owned-task + attach-mode lane (ADR-0055): the FastAPI route resolves the selection to a distinct `landlord_property_id` set — project codes are expanded via `hubspot_deal_data` (which holds the project↔property grain), hand-picked ids are taken as given — pins that set and the recipient email onto the `sub_task`, and the `applications/bulk_document_download` Lambda builds the package, writes the URL to `sub_task.outputs`, and emails it (ADR-0059/0060). The Lambda matches files by `hubspot_deal_id`, bridging each selected `landlord_property_id` through `hubspot_deal_data` (`uploaded_files.landlord_property_id` is unpopulated in practice, so the property identity comes from the deal bridge); coverage is limited to deal-id-linked sources. Selection is capped by property count at the trigger. _Avoid_: document export, bulk export, file dump ### Source data diff --git a/docs/adr/0060-bulk-document-download-package-model.md b/docs/adr/0060-bulk-document-download-package-model.md index 2eedb6d9f..886d7a4a1 100644 --- a/docs/adr/0060-bulk-document-download-package-model.md +++ b/docs/adr/0060-bulk-document-download-package-model.md @@ -55,11 +55,17 @@ lane (ADR-0055).** query — cheap, no S3 on the hot path). A `/tmp` size budget inside the Lambda is a backstop that fails the sub_task with a clear reason if a pathological selection still overflows. `N` starts at a conservative, tunable value. -- **Matching on `landlord_property_id`.** Files are gathered by - `landlord_property_id`; the missing `property_id` on `uploaded_files` is a - **known future gap**, out of scope here. Property display info (the folder - name) is enriched from the hubspot deals data - (`repositories/hubspot_deals/`). +- **Matching via `hubspot_deal_id`, bridged through `hubspot_deal_data`.** The + selection is a set of `landlord_property_id`s, but `uploaded_files` is matched + on **`hubspot_deal_id`**: in practice no upload source populates + `uploaded_files.landlord_property_id` (pashub, magic plan and the audit + generator set `hubspot_deal_id`; ECMK sets `hubspot_listing_id`), so the + worker joins `uploaded_files → hubspot_deal_data (on deal_id) → landlord_property_id` + and takes the property identity from the bridge, not the file row. **Coverage + limitation:** files linked *only* by `hubspot_listing_id` (ECMK) or `uprn` are + not reached; extending the bridge to those keys is a follow-up. Property + display info (the folder name) is enriched from the property record + (`repositories/property/`). - **Layout.** One folder per property named by human-readable **address** (hubspot-enriched), with `landlord_property_id` appended for uniqueness; inside, one file per Document Type = the newest by `s3_upload_timestamp`. @@ -94,16 +100,21 @@ lane (ADR-0055).** - **Strict "all-or-nothing" packaging.** Rejected: one unreadable row or a property with no documents would deny the entire package; best-effort + reporting matches "include documents and properties where they exist". -- **Match on `uprn`.** Deferred: `landlord_property_id` is the agreed key and - hubspot enrichment is keyed to the deal/property; revisit if/when the - `property_id` gap on `uploaded_files` is closed. +- **Match on `landlord_property_id` directly.** Rejected once tested against + real data: `uploaded_files.landlord_property_id` is unpopulated, so a direct + match returns nothing. `hubspot_deal_id` (bridged via `hubspot_deal_data`) is + the key that upload sources actually set. +- **Match on `uprn` / `hubspot_listing_id` as well.** Deferred: `hubspot_deal_id` + covers the bulk of sources; add `uprn`/`listing_id` to the bridge if a source + keyed only by those needs including. ## Consequences - New DDD pieces: `applications/bulk_document_download/` (thin handler + trigger body), `orchestration/bulk_document_download_orchestrator.py`, - packaging rules in `domain/`, a `landlord_property_id`-keyed - "latest-per-Document-Type" query on the uploaded-file repository, a + packaging rules in `domain/`, a `hubspot_deal_id`-bridged query on the + uploaded-file repository that returns a small `PropertyDocument` read-model + (so the orchestrator never names `infrastructure.postgres.*`), a `generate_presigned_url` + multipart upload on `S3Client`, and the email port/adapter from ADR-0059. The only `backend/` touch is the trigger route. - The stored, pinned `property_id` set makes a run reproducible even if the @@ -115,5 +126,8 @@ lane (ADR-0055).** - A **new S3 bucket** (`DOCUMENT_EXPORTS_BUCKET`) is provisioned for the packages, with its own IAM (single-writer, presign-read). Its retention policy is a deliberate follow-up, not set here. -- Closing the `uploaded_files.property_id` gap later would let matching move off - `landlord_property_id` without changing the package model. +- Because matching bridges through `hubspot_deal_data`, a file is only reachable + once its deal has a `hubspot_deal_data` row (and the file carries that + `hubspot_deal_id`). Populating `uploaded_files.landlord_property_id` at upload + time, or closing the `uploaded_files.property_id` gap, would let matching drop + the bridge without changing the package model. diff --git a/domain/bulk_document_download/property_document.py b/domain/bulk_document_download/property_document.py new file mode 100644 index 000000000..624a48504 --- /dev/null +++ b/domain/bulk_document_download/property_document.py @@ -0,0 +1,24 @@ +"""The repository's read-model for a property's uploaded document (ADR-0060). + +A small domain type the uploaded-file repository hands back, so the orchestrator +never names ``infrastructure.postgres.*``. Carries the property's +``landlord_property_id`` (resolved through the ``hubspot_deal_data`` bridge — see +the repository) alongside just the fields the Download Package needs. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from typing import Optional + + +@dataclass(frozen=True) +class PropertyDocument: + """One uploaded file matched to a property, ready to resolve into the plan.""" + + landlord_property_id: str + document_type: Optional[str] + s3_bucket: str + s3_key: str + uploaded_at: datetime diff --git a/orchestration/bulk_document_download_orchestrator.py b/orchestration/bulk_document_download_orchestrator.py index c3c0db2cd..c67ee0372 100644 --- a/orchestration/bulk_document_download_orchestrator.py +++ b/orchestration/bulk_document_download_orchestrator.py @@ -10,28 +10,30 @@ rows are reported, not fatal; a wholly-empty selection is a recorded failure from __future__ import annotations +import logging import os import tempfile import zipfile from dataclasses import dataclass -from datetime import datetime -from typing import Optional, Protocol, Sequence, cast +from typing import Protocol, Sequence from domain.bulk_document_download.package_plan import ( ResolvedDocument, SkippedDocument, plan_download_package, ) +from domain.bulk_document_download.property_document import PropertyDocument from domain.tasks.subtasks import SubTaskFailure -from infrastructure.postgres.uploaded_file_table import UploadedFile from infrastructure.s3.s3_client import S3Client from repositories.email.email_sender import EmailSender +logger = logging.getLogger(__name__) + class UploadedFileReader(Protocol): def by_landlord_property_ids( self, landlord_property_ids: Sequence[str] - ) -> list[UploadedFile]: ... + ) -> list[PropertyDocument]: ... class AddressResolver(Protocol): @@ -79,14 +81,28 @@ class BulkDocumentDownloadOrchestrator: recipient_email: str, package_name: str, ) -> DownloadPackageResult: - files = self._uploaded_files.by_landlord_property_ids(landlord_property_ids) - resolved = [self._resolve(file) for file in files] - plan = plan_download_package([r for r in resolved if r is not None]) + documents = self._uploaded_files.by_landlord_property_ids( + landlord_property_ids + ) + plan = plan_download_package([self._resolve(doc) for doc in documents]) skipped: list[SkippedDocument] = list(plan.skipped) + logger.info( + "bulk_document_download: selected=%d matched_documents=%d " + "planned_entries=%d skipped=%d", + len(landlord_property_ids), + len(documents), + len(plan.entries), + len(skipped), + ) + if not plan.entries: - raise self._empty_failure(skipped) + raise self._empty_failure( + selected=len(landlord_property_ids), + matched=len(documents), + skipped=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 @@ -120,7 +136,11 @@ class BulkDocumentDownloadOrchestrator: if included == 0: # Every file that resolved failed to read — no package to send. - raise self._empty_failure(skipped) + raise self._empty_failure( + selected=len(landlord_property_ids), + matched=len(documents), + skipped=skipped, + ) self._packages.upload_file(zip_path, key) url = self._packages.generate_presigned_url(key, self._url_ttl_seconds) @@ -141,13 +161,21 @@ class BulkDocumentDownloadOrchestrator: skipped=tuple(skipped), ) - def _empty_failure(self, skipped: list[SkippedDocument]) -> SubTaskFailure: + def _empty_failure( + self, *, selected: int, matched: int, 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.""" + (ADR-0060) — never an empty ZIP emailed to the user. The message carries + the stage counts (so they surface in the worker's WARNING log) and the + skip detail rides on the failure's ``details`` (→ ``sub_task.outputs``).""" return SubTaskFailure( - "no documents could be packaged for the selection", + f"no documents could be packaged for the selection " + f"(selected {selected} properties, matched {matched} documents, " + f"{len(skipped)} skipped)", details={ + "selected_properties": selected, + "matched_documents": matched, + "planned_entries": 0, "skipped": [ { "landlord_property_id": s.landlord_property_id, @@ -155,22 +183,16 @@ class BulkDocumentDownloadOrchestrator: "reason": s.reason, } for s in skipped - ] + ], }, ) - def _resolve(self, file: UploadedFile) -> Optional[ResolvedDocument]: - if file.landlord_property_id is None: - return None + def _resolve(self, document: PropertyDocument) -> ResolvedDocument: return ResolvedDocument( - landlord_property_id=file.landlord_property_id, - address=self._addresses.address_for(file.landlord_property_id), - 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), + landlord_property_id=document.landlord_property_id, + address=self._addresses.address_for(document.landlord_property_id), + document_type=document.document_type, + s3_bucket=document.s3_bucket, + s3_key=document.s3_key, + uploaded_at=document.uploaded_at, ) diff --git a/repositories/uploaded_file/uploaded_file_postgres_repository.py b/repositories/uploaded_file/uploaded_file_postgres_repository.py index 82c711231..26e2684c1 100644 --- a/repositories/uploaded_file/uploaded_file_postgres_repository.py +++ b/repositories/uploaded_file/uploaded_file_postgres_repository.py @@ -2,9 +2,10 @@ from __future__ import annotations from typing import Optional, Sequence -from sqlalchemy import select +from sqlalchemy import select, text from sqlmodel import Session, col +from domain.bulk_document_download.property_document import PropertyDocument from infrastructure.postgres.uploaded_file_table import FileTypeEnum, UploadedFile @@ -26,15 +27,40 @@ class UploadedFilePostgresRepository: def by_landlord_property_ids( self, landlord_property_ids: Sequence[str] - ) -> list[UploadedFile]: - """Every uploaded file held for the given properties, matched by - `landlord_property_id` (ADR-0060), across all Document Types. Returns - the raw rows — latest-per-Document-Type selection and null-type - skipping are the Download Package plan's job, not the query's.""" - stmt = select(UploadedFile).where( - col(UploadedFile.landlord_property_id).in_(list(landlord_property_ids)) + ) -> list[PropertyDocument]: + """Every uploaded file held for the given properties (ADR-0060), across + all Document Types, as a small domain read-model. + + Files are matched by **`hubspot_deal_id`**, bridged to the property + through `hubspot_deal_data` — `uploaded_files.landlord_property_id` is + unpopulated in practice (no upload source sets it), so the property + identity comes from the deal bridge, not the file row. Coverage is + therefore limited to deal-id-linked sources (pashub, magic plan, audit + generator); files keyed only by `hubspot_listing_id` or `uprn` are not + reached — see the ADR-0060 matching decision. + + Latest-per-Document-Type selection and null-type skipping are the + Download Package plan's job, not the query's.""" + rows = self._session.connection().execute( + text( + "SELECT h.landlord_property_id, u.file_type," + " u.s3_file_bucket, u.s3_file_key, u.s3_upload_timestamp" + " FROM uploaded_files u" + " JOIN hubspot_deal_data h ON h.deal_id = u.hubspot_deal_id" + " WHERE h.landlord_property_id = ANY(:lpids)" + ), + {"lpids": list(landlord_property_ids)}, ) - return list(self._session.execute(stmt).scalars().all()) # pyright: ignore[reportDeprecated] + return [ + PropertyDocument( + landlord_property_id=row[0], + document_type=row[1], + s3_bucket=row[2], + s3_key=row[3], + uploaded_at=row[4], + ) + for row in rows.all() + ] def insert(self, uploaded_file: UploadedFile) -> None: self._session.add(uploaded_file) diff --git a/tests/orchestration/test_bulk_document_download_orchestrator.py b/tests/orchestration/test_bulk_document_download_orchestrator.py index d9b3a0d23..47a071c38 100644 --- a/tests/orchestration/test_bulk_document_download_orchestrator.py +++ b/tests/orchestration/test_bulk_document_download_orchestrator.py @@ -7,12 +7,14 @@ import io import zipfile from collections.abc import Iterator from datetime import datetime, timezone +from typing import Optional import pytest from moto import mock_aws +from domain.bulk_document_download.property_document import PropertyDocument from domain.tasks.subtasks import SubTaskFailure -from infrastructure.postgres.uploaded_file_table import FileTypeEnum, UploadedFile +from infrastructure.postgres.uploaded_file_table import FileTypeEnum from infrastructure.s3.s3_client import S3Client from orchestration.bulk_document_download_orchestrator import ( BulkDocumentDownloadOrchestrator, @@ -23,14 +25,14 @@ DATA_BUCKET = "data-bucket" class _FakeUploadedFiles: - def __init__(self, files: list[UploadedFile]) -> None: - self._files = files + def __init__(self, documents: list[PropertyDocument]) -> None: + self._documents = documents def by_landlord_property_ids( self, landlord_property_ids: object - ) -> list[UploadedFile]: + ) -> list[PropertyDocument]: wanted = set(landlord_property_ids) # type: ignore[call-overload] - return [f for f in self._files if f.landlord_property_id in wanted] + return [d for d in self._documents if d.landlord_property_id in wanted] class _FakeAddresses: @@ -60,18 +62,18 @@ class _RecordingEmail: self.sent.append((to, subject, body)) -def _uploaded_file( +def _document( landlord_property_id: str, - s3_file_bucket: str, - s3_file_key: str, - file_type: FileTypeEnum = FileTypeEnum.SITE_NOTE, -) -> UploadedFile: - return UploadedFile( - s3_file_bucket=s3_file_bucket, - s3_file_key=s3_file_key, - s3_upload_timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc), + s3_bucket: str, + s3_key: str, + file_type: Optional[FileTypeEnum] = FileTypeEnum.SITE_NOTE, +) -> PropertyDocument: + return PropertyDocument( landlord_property_id=landlord_property_id, - file_type=file_type.value, + document_type=file_type.value if file_type is not None else None, + s3_bucket=s3_bucket, + s3_key=s3_key, + uploaded_at=datetime(2026, 1, 1, tzinfo=timezone.utc), ) @@ -85,7 +87,7 @@ def packages() -> Iterator[S3Client]: def test_builds_uploads_and_emails_a_download_package(packages: S3Client) -> None: # Arrange — one property with one site note. - files = [_uploaded_file("LP1", "src-bucket", "surveys/a.pdf")] + files = [_document("LP1", "src-bucket", "surveys/a.pdf")] email = _RecordingEmail() orchestrator = BulkDocumentDownloadOrchestrator( uploaded_files=_FakeUploadedFiles(files), @@ -133,27 +135,29 @@ def test_a_selection_with_no_documents_is_a_recorded_failure( # Act / Assert — a recorded SubTask failure, and nobody is emailed an empty # package. - with pytest.raises(SubTaskFailure): + with pytest.raises(SubTaskFailure) as exc_info: orchestrator.run( - landlord_property_ids=["LP1"], + landlord_property_ids=["LP1", "LP2"], recipient_email="user@example.com", package_name="empty", ) assert email.sent == [] + # ...and it carries the stage counts so the failure is debuggable: the + # message (→ the worker's WARNING log) and details (→ sub_task.outputs). + failure = exc_info.value + assert "matched 0 documents" in str(failure) + assert failure.details is not None + assert failure.details["selected_properties"] == 2 + assert failure.details["matched_documents"] == 0 + def test_packages_the_good_documents_and_reports_the_skipped_ones( packages: S3Client, ) -> None: # Arrange — one packable site note and one row with no Document Type. - good = _uploaded_file("LP1", "src-bucket", "surveys/a.pdf") - untyped = UploadedFile( - s3_file_bucket="src-bucket", - s3_file_key="mystery.bin", - s3_upload_timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc), - landlord_property_id="LP1", - file_type=None, - ) + good = _document("LP1", "src-bucket", "surveys/a.pdf") + untyped = _document("LP1", "src-bucket", "mystery.bin", file_type=None) email = _RecordingEmail() orchestrator = BulkDocumentDownloadOrchestrator( uploaded_files=_FakeUploadedFiles([good, untyped]), @@ -185,8 +189,8 @@ def test_a_file_that_cannot_be_read_is_skipped_not_fatal(packages: S3Client) -> # 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( + good = _document("LP1", "src-bucket", "surveys/a.pdf") + missing = _document( "LP1", "src-bucket", "surveys/gone.pdf", file_type=FileTypeEnum.PHOTO_PACK ) email = _RecordingEmail() @@ -221,7 +225,7 @@ 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") + doc = _document("LP9", "src-bucket", "surveys/a.pdf") orchestrator = BulkDocumentDownloadOrchestrator( uploaded_files=_FakeUploadedFiles([doc]), addresses=_FakeAddresses({"LP9": "address unavailable"}), diff --git a/tests/repositories/uploaded_file/test_uploaded_file_postgres_repository.py b/tests/repositories/uploaded_file/test_uploaded_file_postgres_repository.py index 40fc1773f..76ac9b758 100644 --- a/tests/repositories/uploaded_file/test_uploaded_file_postgres_repository.py +++ b/tests/repositories/uploaded_file/test_uploaded_file_postgres_repository.py @@ -5,6 +5,7 @@ from datetime import datetime, timedelta, timezone from sqlalchemy import Engine from sqlmodel import Session +from backend.app.db.models.hubspot_deal_data import HubspotDealData from infrastructure.postgres.uploaded_file_table import FileTypeEnum, UploadedFile from repositories.uploaded_file.uploaded_file_postgres_repository import ( UploadedFilePostgresRepository, @@ -80,28 +81,38 @@ def test_does_not_return_row_with_different_file_type(db_engine: Engine) -> None assert result is None -def _landlord_file( - landlord_property_id: str, +def _deal_file( + hubspot_deal_id: str, s3_file_key: str, file_type: FileTypeEnum = FileTypeEnum.SITE_NOTE, ) -> UploadedFile: + # A real uploaded file: keyed by hubspot_deal_id, landlord_property_id NULL + # (no upload source populates it — see ADR-0060 matching decision). return UploadedFile( s3_file_bucket=_BUCKET, s3_file_key=s3_file_key, s3_upload_timestamp=datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc), - landlord_property_id=landlord_property_id, + hubspot_deal_id=hubspot_deal_id, file_type=file_type.value, ) -def test_by_landlord_property_ids_returns_only_requested_properties( +def _deal(landlord_property_id: str, deal_id: str) -> HubspotDealData: + return HubspotDealData(deal_id=deal_id, landlord_property_id=landlord_property_id) + + +def test_by_landlord_property_ids_bridges_deal_id_to_the_property( db_engine: Engine, ) -> None: - # Arrange — files for three properties; only two are requested. + # Arrange — files are keyed by hubspot_deal_id; hubspot_deal_data bridges each + # deal to a landlord_property_id. Three properties exist; two are requested. with Session(db_engine) as session: - session.add(_landlord_file("LP1", "one.pdf")) - session.add(_landlord_file("LP2", "two.pdf")) - session.add(_landlord_file("LP3", "three.pdf")) + session.add(_deal("LP1", "deal-1")) + session.add(_deal("LP2", "deal-2")) + session.add(_deal("LP3", "deal-3")) + session.add(_deal_file("deal-1", "one.pdf")) + session.add(_deal_file("deal-2", "two.pdf")) + session.add(_deal_file("deal-3", "three.pdf")) session.commit() # Act @@ -109,7 +120,25 @@ def test_by_landlord_property_ids_returns_only_requested_properties( found = UploadedFilePostgresRepository(session).by_landlord_property_ids( ["LP1", "LP2"] ) - result = {(f.landlord_property_id, f.s3_file_key) for f in found} + result = {(d.landlord_property_id, d.s3_key) for d in found} + + # Assert — the file's property comes from the bridge, keyed by the deal. + assert result == {("LP1", "one.pdf"), ("LP2", "two.pdf")} + + +def test_by_landlord_property_ids_matches_nothing_without_a_deal_bridge( + db_engine: Engine, +) -> None: + # Arrange — a file whose deal has no hubspot_deal_data row can't be reached. + with Session(db_engine) as session: + session.add(_deal_file("orphan-deal", "orphan.pdf")) + session.commit() + + # Act + with Session(db_engine) as session: + found = UploadedFilePostgresRepository(session).by_landlord_property_ids( + ["LP1"] + ) # Assert - assert result == {("LP1", "one.pdf"), ("LP2", "two.pdf")} + assert found == []