Merge pull request #1517 from Hestia-Homes/fix/bulk-download-worker-observability

Bulk download: readable email, deal-name folders, visible worker logs

applications/bulk_document_download/handler.py

@@ -16,6 +16,7 @@ Secrets Manager IAM-user credentials in).
 
 from __future__ import annotations
 
+import logging
 import os
 from typing import Any
 
@@ -40,6 +41,12 @@ from repositories.uploaded_file.uploaded_file_postgres_repository import (
 )
 from utilities.aws_lambda.task_handler import task_handler
 
+# Surface the worker's progress logs: AWS Lambda's root logger defaults to
+# WARNING, which drops the orchestrator's INFO stage/timing lines. Raise it so a
+# slow or failing run is observable in CloudWatch.
+logging.getLogger().setLevel(logging.INFO)
+logger = logging.getLogger(__name__)
+
 # The Download Package link is valid for 60 minutes (ADR-0060).
 _URL_TTL_SECONDS = 3600
 
@@ -68,6 +75,12 @@ def handler(body: dict[str, Any], context: Any) -> dict[str, Any]:
         recipient_email: str = str(recipe["recipient_email"])
         package_name: str = str(recipe["package_name"])
 
+        logger.info(
+            "bulk_document_download: starting run for %d properties (subtask=%s)",
+            len(landlord_property_ids),
+            trigger.subtask_id,
+        )
+
         boto_s3: Any = boto3.client("s3")  # pyright: ignore[reportUnknownMemberType]
         orchestrator = BulkDocumentDownloadOrchestrator(
             uploaded_files=UploadedFilePostgresRepository(session),

infrastructure/email/ses_smtp_email_sender.py

@@ -34,11 +34,16 @@ class SmtpTransport(Protocol):
     def send_message(self, message: EmailMessage) -> Any: ...
 
 
+# A short connect/IO timeout so an unreachable SES endpoint fails fast and is
+# logged, instead of blocking the Lambda until its 900s timeout (ADR-0059).
+_SMTP_TIMEOUT_SECONDS = 30
+
+
 def _default_smtp(host: str, port: int) -> SmtpTransport:
     # `smtplib.SMTP` satisfies SmtpTransport structurally; its stdlib stubs
     # declare wider signatures (extra optional args, an SMTP-typed context
     # manager), so the cast bridges the real transport to the port's surface.
-    return cast(SmtpTransport, smtplib.SMTP(host, port))
+    return cast(SmtpTransport, smtplib.SMTP(host, port, timeout=_SMTP_TIMEOUT_SECONDS))
 
 
 class SesSmtpEmailSender:
@@ -59,12 +64,16 @@ class SesSmtpEmailSender:
         self._from_address = from_address
         self._smtp_factory = smtp_factory
 
-    def send(self, *, to: str, subject: str, body: str) -> None:
+    def send(
+        self, *, to: str, subject: str, body: str, html_body: Optional[str] = None
+    ) -> None:
         message = EmailMessage()
         message["From"] = self._from_address
         message["To"] = to
         message["Subject"] = subject
         message.set_content(body)
+        if html_body is not None:
+            message.add_alternative(html_body, subtype="html")
         with self._smtp_factory(self._host, self._port) as smtp:
             smtp.starttls()
             smtp.login(self._username, self._password)

orchestration/bulk_document_download_orchestrator.py

@@ -10,9 +10,11 @@ rows are reported, not fatal; a wholly-empty selection is a recorded failure
 
 from __future__ import annotations
 
+import html
 import logging
 import os
 import tempfile
+import time
 import zipfile
 from dataclasses import dataclass
 from typing import Protocol, Sequence
@@ -81,20 +83,23 @@ class BulkDocumentDownloadOrchestrator:
         recipient_email: str,
         package_name: str,
     ) -> DownloadPackageResult:
+        started = time.monotonic()
         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)
+        null_type_skips = len(skipped)
 
         logger.info(
             "bulk_document_download: selected=%d matched_documents=%d "
-            "planned_entries=%d skipped=%d",
+            "planned_entries=%d null_type_skips=%d (gather+plan %.1fs)",
             len(landlord_property_ids),
             len(documents),
             len(plan.entries),
-            len(skipped),
+            null_type_skips,
+            time.monotonic() - started,
         )
 
         if not plan.entries:
@@ -112,6 +117,8 @@ class BulkDocumentDownloadOrchestrator:
         # can't reach — becomes a SkippedDocument, it does not fail the package.
         key = f"{self._package_key_prefix}/{package_name}.zip"
         included = 0
+        zip_bytes = 0
+        zip_started = time.monotonic()
         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:
@@ -141,18 +148,55 @@ class BulkDocumentDownloadOrchestrator:
                     matched=len(documents),
                     skipped=skipped,
                 )
+            zip_bytes = os.path.getsize(zip_path)
+            logger.info(
+                "bulk_document_download: packaged %d files (%d unreadable), "
+                "%.1f MB in %.1fs",
+                included,
+                len(skipped) - null_type_skips,
+                zip_bytes / 1_000_000,
+                time.monotonic() - zip_started,
+            )
+            upload_started = time.monotonic()
             self._packages.upload_file(zip_path, key)
+            logger.info(
+                "bulk_document_download: uploaded %.1f MB in %.1fs",
+                zip_bytes / 1_000_000,
+                time.monotonic() - upload_started,
+            )
         url = self._packages.generate_presigned_url(key, self._url_ttl_seconds)
 
-        self._email.send(
+        property_count = len({entry.landlord_property_id for entry in plan.entries})
-            to=recipient_email,
+        subject, body, html_body = self._compose_email(
-            subject="Your document download is ready",
+            url=url,
-            body=(
+            included=included,
-                f"Your document download is ready. It contains "
+            properties=property_count,
-                f"{included} document(s).\n\n"
+            skipped=len(skipped),
-                f"Download it here (link valid for "
+        )
-                f"{self._url_ttl_seconds // 60} minutes):\n{url}"
+        try:
-            ),
+            self._email.send(
+                to=recipient_email,
+                subject=subject,
+                body=body,
+                html_body=html_body,
+            )
+            logger.info("bulk_document_download: emailed %s", recipient_email)
+        except Exception:
+            # Best-effort delivery (ADR-0060): the link is also written to
+            # sub_task.outputs, so an email/transport failure must not lose a
+            # package that was already built and uploaded.
+            logger.warning(
+                "bulk_document_download: email delivery to %s failed; the link is "
+                "still recorded on sub_task.outputs",
+                recipient_email,
+                exc_info=True,
+            )
+
+        logger.info(
+            "bulk_document_download: done — %d documents, %.1f MB, total %.1fs",
+            included,
+            zip_bytes / 1_000_000,
+            time.monotonic() - started,
         )
         return DownloadPackageResult(
             presigned_url=url,
@@ -161,6 +205,40 @@ class BulkDocumentDownloadOrchestrator:
             skipped=tuple(skipped),
         )
 
+    def _compose_email(
+        self, *, url: str, included: int, properties: int, skipped: int
+    ) -> tuple[str, str, str]:
+        """The delivery email (ADR-0059): a subject, a plain-text body, and an
+        HTML alternative with a clean download button so the long presigned URL
+        isn't the visible content."""
+        minutes = self._url_ttl_seconds // 60
+        noun = "property" if properties == 1 else "properties"
+        skip_note = f" {skipped} item(s) were skipped." if skipped else ""
+        subject = f"Your document download is ready ({included} documents)"
+        body = (
+            "Your document package is ready.\n\n"
+            f"It contains {included} document(s) across {properties} {noun}."
+            f"{skip_note}\n\n"
+            f"Download it here (link valid for {minutes} minutes):\n{url}\n"
+        )
+        safe_url = html.escape(url, quote=True)
+        html_body = (
+            '<div style="font-family:Arial,Helvetica,sans-serif;font-size:15px;'
+            'color:#1a1a1a;line-height:1.5;">'
+            "<p>Your document package is ready.</p>"
+            f"<p>It contains <strong>{included} document(s)</strong> across "
+            f"<strong>{properties}</strong> {noun}.{skip_note}</p>"
+            '<p style="margin:24px 0;">'
+            f'<a href="{safe_url}" style="display:inline-block;padding:12px 22px;'
+            "background:#0b8457;color:#ffffff;text-decoration:none;border-radius:6px;"
+            'font-weight:bold;">Download documents</a></p>'
+            f'<p style="color:#666;font-size:13px;">This link expires in {minutes} '
+            "minutes. If the button doesn't work, paste this URL into your browser:"
+            f'<br><span style="word-break:break-all;">{safe_url}</span></p>'
+            "</div>"
+        )
+        return subject, body, html_body
+
     def _empty_failure(
         self, *, selected: int, matched: int, skipped: list[SkippedDocument]
     ) -> SubTaskFailure:

repositories/email/email_sender.py

@@ -6,10 +6,14 @@ transport (SES over SMTP today) stays swappable and testable behind it.
 
 from __future__ import annotations
 
-from typing import Protocol
+from typing import Optional, Protocol
 
 
 class EmailSender(Protocol):
-    def send(self, *, to: str, subject: str, body: str) -> None:
+    def send(
-        """Send a plain-text email to ``to``."""
+        self, *, to: str, subject: str, body: str, html_body: Optional[str] = None
+    ) -> None:
+        """Send an email to ``to``. ``body`` is the plain-text part; when
+        ``html_body`` is given it is added as an HTML alternative (clients that
+        render HTML show it, others fall back to the plain text)."""
         ...

repositories/property/landlord_address_resolver.py

@@ -1,20 +1,18 @@
-"""Resolves a property's display address from its `landlord_property_id`
+"""Resolves a property's display name for the Download Package folders (ADR-0060).
-(ADR-0060), for naming the folders in a Download Package.
 
-The address lives on the FE-owned `property` table (`uploaded_files` carries no
+The name comes from the HubSpot deal's ``dealname`` (``hubspot_deal_data``, keyed
-`property_id` — a known future gap). Addresses for the whole requested set are
+by ``landlord_property_id``) — these properties are HubSpot deals and usually have
-loaded once at construction; `address_for` is then a lookup with a safe
+no ``property`` row, so the deal name is the human-readable label. Names for the
-fallback for a `landlord_property_id` that has no property row or address.
+whole requested set are loaded once at construction; ``address_for`` is then a
+lookup with a safe fallback for a ``landlord_property_id`` that has no deal name.
 """
 
 from __future__ import annotations
 
 from typing import Sequence
 
-from sqlalchemy import select
+from sqlalchemy import text
-from sqlmodel import Session, col
+from sqlmodel import Session
-
-from infrastructure.postgres.property_table import PropertyRow
 
 
 class LandlordAddressResolver:
@@ -23,15 +21,18 @@ class LandlordAddressResolver:
     def __init__(
         self, session: Session, landlord_property_ids: Sequence[str]
     ) -> None:
-        stmt = select(PropertyRow).where(
+        rows = session.connection().execute(
-            col(PropertyRow.landlord_property_id).in_(list(landlord_property_ids))
+            text(
+                "SELECT landlord_property_id, dealname FROM hubspot_deal_data"
+                " WHERE landlord_property_id = ANY(:lpids)"
+                " AND dealname IS NOT NULL"
+            ),
+            {"lpids": list(landlord_property_ids)},
         )
-        rows = session.execute(stmt).scalars().all()  # pyright: ignore[reportDeprecated]
+        # First non-null deal name wins for a property with more than one deal.
-        self._by_id: dict[str, str] = {
+        self._by_id: dict[str, str] = {}
-            row.landlord_property_id: row.address
+        for landlord_property_id, dealname in rows.all():
-            for row in rows
+            self._by_id.setdefault(landlord_property_id, dealname)
-            if row.landlord_property_id is not None and row.address is not None
-        }
 
     def address_for(self, landlord_property_id: str) -> str:
         return self._by_id.get(landlord_property_id, self._FALLBACK)

tests/infrastructure/email/test_ses_smtp_email_sender.py

@@ -77,6 +77,41 @@ def test_send_delivers_a_plaintext_message_with_the_right_envelope() -> None:
     assert "https://signed-url" in message.get_content()
 
 
+def test_send_adds_an_html_alternative_when_given() -> None:
+    # Arrange
+    created: list[_FakeSmtp] = []
+
+    def factory(host: str, port: int) -> _FakeSmtp:
+        smtp = _FakeSmtp(host, port)
+        created.append(smtp)
+        return smtp
+
+    sender = SesSmtpEmailSender(
+        host="email-smtp.eu-west-2.amazonaws.com",
+        port=587,
+        username="AKIA_SMTP_USER",
+        password="smtp-password",
+        from_address="noreply@domna.homes",
+        smtp_factory=factory,
+    )
+
+    # Act
+    sender.send(
+        to="user@example.com",
+        subject="s",
+        body="plain text fallback",
+        html_body="<p>Download documents</p>",
+    )
+
+    # Assert — a multipart/alternative carrying both the plain text and the HTML.
+    message = created[0].messages[0]
+    assert message.get_content_type() == "multipart/alternative"
+    plain_part = message.get_body(preferencelist=("plain",))
+    html_part = message.get_body(preferencelist=("html",))
+    assert plain_part is not None and "plain text fallback" in plain_part.get_content()
+    assert html_part is not None and "Download documents" in html_part.get_content()
+
+
 def test_send_starts_tls_and_authenticates_before_sending() -> None:
     # Arrange — SES SMTP rejects unauthenticated, cleartext senders, so the
     # transport must STARTTLS and log in with the SMTP credentials.

tests/orchestration/test_bulk_document_download_orchestrator.py

@@ -56,10 +56,19 @@ class _FakeDocuments:
 
 class _RecordingEmail:
     def __init__(self) -> None:
-        self.sent: list[tuple[str, str, str]] = []
+        self.sent: list[tuple[str, str, str, Optional[str]]] = []
 
-    def send(self, *, to: str, subject: str, body: str) -> None:
+    def send(
-        self.sent.append((to, subject, body))
+        self, *, to: str, subject: str, body: str, html_body: Optional[str] = None
+    ) -> None:
+        self.sent.append((to, subject, body, html_body))
+
+
+class _FailingEmail:
+    def send(
+        self, *, to: str, subject: str, body: str, html_body: Optional[str] = None
+    ) -> None:
+        raise RuntimeError("smtp unreachable")
 
 
 def _document(
@@ -114,9 +123,38 @@ def test_builds_uploads_and_emails_a_download_package(packages: S3Client) -> Non
     # ...the presigned URL points at that object, and the requester is emailed it.
     assert result.package_s3_key in result.presigned_url
     assert len(email.sent) == 1
-    to, _subject, body = email.sent[0]
+    to, subject, body, html_body = email.sent[0]
     assert to == "user@example.com"
-    assert result.presigned_url in body
+    assert "1 documents" in subject
+    assert result.presigned_url in body  # raw URL in the plain-text fallback
+    assert html_body is not None
+    assert "Download documents" in html_body  # clean link, not the raw URL
+
+
+def test_an_email_failure_does_not_lose_the_built_package(packages: S3Client) -> None:
+    # Arrange — the package builds and uploads fine, but the email transport is
+    # down. Best-effort delivery (ADR-0060): the URL is still returned (→ outputs).
+    files = [_document("LP1", "src-bucket", "surveys/a.pdf")]
+    orchestrator = BulkDocumentDownloadOrchestrator(
+        uploaded_files=_FakeUploadedFiles(files),
+        addresses=_FakeAddresses({"LP1": "12 Oak Street"}),
+        documents=_FakeDocuments({("src-bucket", "surveys/a.pdf"): b"PDF-BYTES"}),
+        packages=packages,
+        email=_FailingEmail(),
+        url_ttl_seconds=3600,
+    )
+
+    # Act — must not raise despite the email failure.
+    result = orchestrator.run(
+        landlord_property_ids=["LP1"],
+        recipient_email="user@example.com",
+        package_name="email-down",
+    )
+
+    # Assert — the package was still uploaded and its URL returned.
+    assert result.included == 1
+    assert result.package_s3_key in result.presigned_url
+    assert packages.get_object(result.package_s3_key)  # uploaded despite email failure
 
 
 def test_a_selection_with_no_documents_is_a_recorded_failure(

tests/repositories/property/test_landlord_address_resolver.py

@@ -4,7 +4,7 @@ import pytest
 from sqlalchemy import Engine
 from sqlmodel import Session
 
-from infrastructure.postgres.property_table import PropertyRow
+from backend.app.db.models.hubspot_deal_data import HubspotDealData
 from repositories.property.landlord_address_resolver import LandlordAddressResolver
 
 
@@ -14,22 +14,26 @@ def session(db_engine: Engine) -> Iterator[Session]:
         yield s
 
 
-def test_resolves_known_addresses_and_falls_back_for_unknowns(
+def test_resolves_deal_names_and_falls_back_for_unknowns(
     session: Session,
 ) -> None:
-    # arrange — two properties with addresses; a third id has no property row.
+    # arrange — two properties with deal names; a third id has no deal.
     session.add(
-        PropertyRow(portfolio_id=1, landlord_property_id="LP1", address="12 Oak Street")
+        HubspotDealData(
+            deal_id="d1", landlord_property_id="LP1", dealname="12 Oak Street"
+        )
     )
     session.add(
-        PropertyRow(portfolio_id=1, landlord_property_id="LP2", address="9 Elm Road")
+        HubspotDealData(
+            deal_id="d2", landlord_property_id="LP2", dealname="9 Elm Road"
+        )
     )
     session.flush()
 
     # act
     resolver = LandlordAddressResolver(session, ["LP1", "LP2", "LP3"])
 
-    # assert
+    # assert — the deal name becomes the folder label; unknowns fall back.
     assert resolver.address_for("LP1") == "12 Oak Street"
     assert resolver.address_for("LP2") == "9 Elm Road"
     assert resolver.address_for("LP3") == "address unavailable"